amadan

carlos / platform Public

host: lifetime gating witnesses: a fieldless record beside a retired slot is gone to every box consumer

TestHostConsumersIgnoreAFieldlessRecordBesideASlot drives the production
functions with an unjudged control beside each: reconcile retires the row and
mints nothing, the edges file stops listing it, the offline reassert, the
staging deletion gate, the runs boot pass, pickup, the detached execution's
done CAS and launchable ignore it, steering and retention omit it, the proof
binding source as the agent and root build it finds nothing, and the seed
callback wireSeedPlacement installs refuses and keeps the snapshot.

TestOfflineFinishingRemovesTheSourceRow runs an offline move on a judged box
through reconcileSerializedWith; TestOfflineMoveE2E gains a judged run of the
whole move. TestClaimsOnConsumersReadNoSlot, TestBoxJudgeIsSharedAndGated and
TestRootRestartReadersAreJudged pin the gate, the one shared judge and root's
restart readers with every production construction of their source.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude f5216bed704ca014c614d10bd5838d9c984f5843 parent cf6cbb3
4 files changed, +709 −2
  • internal/host/lifetime_gating_test.go +693 −0
  • internal/host/movesoffline_e2e_test.go +7 −0
  • internal/host/movesoffline_land_test.go +3 −0
  • internal/host/movesoffline_test.go +6 −2
diff --git a/internal/host/lifetime_gating_test.go b/internal/host/lifetime_gating_test.go
new file mode 100644
index 000000000..7924d1e02
--- /dev/null
+++ b/internal/host/lifetime_gating_test.go
@@ -0,0 +1,693 @@
+//go:build linux
+
+package host
+
+import (
+ "context"
+ "errors"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/runs"
+ "github.com/carlosframework/platform/internal/seeds"
+)
+
+// The box's consumers of instance records under the lifetime judge
+// (self-serve moves plan 1, Task 6): a claims-off box judges every record it
+// acts on, the record then its host incarnation slot, and a claims-on box
+// reads no slot. Each scene puts a fieldless record beside a retired slot
+// (an old console's late write after the lifetime ended) and drives the
+// production function the agent runs, with an unjudged control beside it.
+
+// retireLifetime writes host's incarnation slot at 1, retired, for acct/app,
+// through the production slot writer: the record under that key is now
+// behind its slot.
+func retireLifetime(t *testing.T, s release.Store, acct, app, host string) {
+ t.Helper()
+ if _, err := provision.WriteHostIncarnation(context.Background(), s, host,
+ provision.HostIncarnation{Account: acct, App: app, Incarnation: 1, State: provision.SlotRetired}, ""); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// slotReads counts the reads of host incarnation slots made through it.
+type slotReads struct {
+ release.Store
+ mu sync.Mutex
+ n int
+}
+
+func (s *slotReads) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.HasPrefix(key, "control/host-incarnations/") {
+ s.mu.Lock()
+ s.n++
+ s.mu.Unlock()
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func (s *slotReads) reads() int {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.n
+}
+
+// claimsOffEnv and claimsOnEnv are the two host-claim switches newBoxJudge
+// reads.
+func claimsOffEnv(string) string { return "" }
+func claimsOnEnv(k string) string {
+ if k == "CARLOS_HOST_CLAIM" {
+ return "1"
+ }
+ return ""
+}
+
+func TestOfflineFinishingRemovesTheSourceRow(t *testing.T) {
+ wd := newOffWorld(t)
+ reads := &slotReads{Store: wd.s}
+ wd.s = reads
+ wd.judge = newBoxJudge(claimsOffEnv)
+ src := wd.box("ie-a", "", nil)
+ src.tick()
+ if rt := src.row(offHost); rt == nil || !src.lists(offHost) {
+ t.Fatalf("fixture: the source does not serve the host: %+v", rt)
+ }
+
+ // Through the move the record names another region from the flip on:
+ // a valid record out of this box's scope keeps today's rules, so the
+ // keep view holds the source row.
+ wd.start()
+ src.tick()
+ wd.seal()
+ wd.sealed()
+ wd.flip()
+ for _, step := range []func(){func() {}, wd.landed, wd.open} {
+ step()
+ wd.advance(time.Minute)
+ src.tick()
+ if rt := src.row(offHost); rt == nil {
+ t.Fatalf("the source row went before finishing (record %s)", wd.record().Move.State)
+ }
+ }
+
+ // At finishing the keep view lets go and the row is removed, exactly
+ // as without a judge; then the source reports gone.
+ wd.finishing()
+ for range 2 {
+ wd.advance(time.Minute)
+ src.tick()
+ }
+ if rt := src.row(offHost); rt != nil {
+ t.Fatalf("the source row survived finishing: %+v", rt)
+ }
+ if r := wd.report(moves.OfflineRoleSource, "ie-a"); r.Phase != moves.OfflinePhaseGone {
+ t.Fatalf("the source's report is %q, want gone", r.Phase)
+ }
+ if reads.reads() == 0 {
+ t.Fatal("fixture: the box judged nothing (no slot was read)")
+ }
+}
+
+func TestHostConsumersIgnoreAFieldlessRecordBesideASlot(t *testing.T) {
+ ctx := context.Background()
+
+ t.Run("reconcile, the edges file and the offline reassert", func(t *testing.T) {
+ wd := newOffWorld(t)
+ wd.judge = newBoxJudge(claimsOffEnv)
+ a := wd.box("ie-a", "", nil)
+ a.tick()
+ if !a.lists(offHost) {
+ t.Fatal("fixture: the edges file does not list the host")
+ }
+ wd.start()
+ a.tick() // held moving; the view and the report know the move
+ before := wd.report(moves.OfflineRoleSource, "ie-a")
+ retireLifetime(t, wd.s, wd.acct, "app", offHost)
+ wd.advance(time.Minute)
+
+ // The re-assert's own read of the record finds it gone: nothing is
+ // acted on and the report is not rewritten.
+ if n := a.w.offlineReassert(ctx, wd.s); n != 0 {
+ t.Fatalf("the re-assert acted on %d records", n)
+ }
+ if r := wd.report(moves.OfflineRoleSource, "ie-a"); !r.At.Equal(before.At) {
+ t.Fatalf("the source's report was rewritten at %v", r.At)
+ }
+
+ // Reconcile retires the row (held by the move, whose record is
+ // gone: its keep is no answer) and mints nothing; the edges file
+ // stops listing the host.
+ a.reconcile()
+ if rt := a.row(offHost); rt != nil {
+ t.Fatalf("the row of a stale record survived: %+v", rt)
+ }
+ a.edges()
+ if a.lists(offHost) {
+ t.Fatal("the edges file still lists the host")
+ }
+ a.tick()
+ if rt := a.row(offHost); rt != nil {
+ t.Fatalf("a stale record was minted: %+v", rt)
+ }
+ })
+
+ t.Run("the staging deletion gate", func(t *testing.T) {
+ for _, retire := range []bool{false, true} {
+ sw := newSealWorld(t)
+ sw.judge = newBoxJudge(claimsOffEnv)
+ ie := sw.ie
+ // Armed with the box's judge as armMovesPass arms it: the worker
+ // and its offline view (the reconcile pass reads sw.judge).
+ ie.w.cfg.Judge, ie.w.cfg.Offline.judge = sw.judge, sw.judge
+ sw.instance(ie)
+ sw.stage()
+ sw.mustStagePass(ie)
+ extra := activator.MovesPrefix(offHost) + "0000/only-in-staging.ltx"
+ sw.fake.Put(sealUK, extra, []byte("staged"))
+ sw.fake.Put(sealIE, sealPrefix+"/"+offHost+"/scratch/new.bin", []byte("new"))
+ sw.restage()
+ var once sync.Once
+ sw.fake.FailOn = func(op, key string) error {
+ if op == "copy" && retire {
+ // The lifetime ends while the round's copy is in flight.
+ once.Do(func() { retireLifetime(t, sw.s, sw.acct, "app", offHost) })
+ }
+ return nil
+ }
+ _ = ie.w.offlineRound(ctx, sw.s, sw.record())
+ sw.fake.FailOn = nil
+ _, kept := sw.fake.Current(sealUK, extra)
+ if kept != retire {
+ t.Fatalf("retired %v: the staging-only object kept = %v", retire, kept)
+ }
+ }
+ })
+
+ t.Run("the runs boot pass, pickup, the detached execution and launchable", func(t *testing.T) {
+ type scene struct {
+ wd *offWorld
+ rec provision.Record
+ req runs.Request
+ id string
+ }
+ newScene := func(t *testing.T) scene {
+ wd := newOffWorld(t)
+ rec := wd.record()
+ id := "run_abcdefghijklmnopqrstuvwxyz"
+ inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req := runs.Request{Schema: 1, ID: id, Account: rec.Account, App: rec.App, Host: rec.Host, Incarnation: inc, Argv: []string{"import"}, TimeoutS: 60, RequestedAt: time.Now()}
+ putJSONObject(t, wd.s, runs.RequestKey(rec.Account, rec.App, id), req)
+ got, err := provision.SetRecordRun(ctx, wd.s, rec.Account, rec.App, rec.Host,
+ provision.RunAdmit(provision.RunAdmission{ID: id, By: "op@example.com", Deadline: time.Now().Add(time.Minute)}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return scene{wd: wd, rec: *got, req: req, id: id}
+ }
+ owned := func(acct string) func(string) (*registry.Route, error) {
+ return func(h string) (*registry.Route, error) {
+ if h != offHost {
+ return nil, nil
+ }
+ return &registry.Route{Host: offHost, Kind: "instance", Hibernate: true, Socket: "/run/x.sock", Provisioned: true, App: "app", Account: acct}, nil
+ }
+ }
+ runState := func(t *testing.T, s release.Store, acct string) string {
+ rec, err := provision.GetRecord(ctx, s, acct, "app", offHost)
+ if err != nil || rec.ActiveRun == nil {
+ t.Fatalf("record %+v, %v", rec, err)
+ }
+ return rec.ActiveRun.State
+ }
+ worker := func(t *testing.T, x scene, judge *provision.LifetimeJudge, log *execLog) *runsWorker {
+ w := newRunsWorker(x.wd.s, owned(x.wd.acct), nil, t.TempDir(), "box-1", "boot-1")
+ w.judge = judge
+ w.exec = log.exec
+ w.start = func(f func()) { f() }
+ w.booted.Store(true)
+ return w
+ }
+ for _, judged := range []bool{false, true} {
+ var judge *provision.LifetimeJudge
+ if judged {
+ judge = newBoxJudge(claimsOffEnv)
+ }
+ want := 1
+ if judged {
+ want = 0
+ }
+ // The boot pass lists the records.
+ x := newScene(t)
+ retireLifetime(t, x.wd.s, x.wd.acct, "app", offHost)
+ var boot execLog
+ w := worker(t, x, judge, &boot)
+ w.booted.Store(false)
+ w.bootPass(ctx)
+ if boot.count() != want {
+ t.Fatalf("judged %v: the boot pass launched %d runs", judged, boot.count())
+ }
+ // A pass acts on what reconcile noted: the pickup's CAS.
+ x = newScene(t)
+ retireLifetime(t, x.wd.s, x.wd.acct, "app", offHost)
+ var picked execLog
+ w = worker(t, x, judge, &picked)
+ w.note(x.rec)
+ w.pass(ctx)
+ if picked.count() != want {
+ t.Fatalf("judged %v: the pass picked up %d runs", judged, picked.count())
+ }
+ if judged && runState(t, x.wd.s, x.wd.acct) != provision.RunQueued {
+ t.Fatal("a stale record's run was picked")
+ }
+ }
+
+ // The real pickup-to-execution path: picked while the record was
+ // valid, executed (detached, as the agent runs it) after its
+ // lifetime ended. The launch fails, so execute's own SetRecordRun
+ // is the done CAS; judged, it is refused.
+ for _, judged := range []bool{false, true} {
+ var judge *provision.LifetimeJudge
+ if judged {
+ judge = newBoxJudge(claimsOffEnv)
+ }
+ x := newScene(t)
+ w := newRunsWorker(x.wd.s, owned(x.wd.acct), failingLauncher{}, t.TempDir(), "box-1", "boot-1")
+ w.judge = judge
+ w.booted.Store(true)
+ var detached func()
+ w.start = func(f func()) { detached = f }
+ w.consider(provision.WithLifetimes(ctx, judge), x.rec)
+ if detached == nil || runState(t, x.wd.s, x.wd.acct) != provision.RunPicked {
+ t.Fatalf("judged %v: fixture: the run was not picked", judged)
+ }
+ retireLifetime(t, x.wd.s, x.wd.acct, "app", offHost)
+ detached()
+ got := runState(t, x.wd.s, x.wd.acct)
+ if judged && got != provision.RunPicked {
+ t.Fatalf("the detached execution wrote a stale record's run: %s", got)
+ }
+ if !judged && got == provision.RunPicked {
+ t.Fatal("fixture: unjudged, the execution did not finish the run")
+ }
+ }
+
+ // launchable reads the record before the launch.
+ x := newScene(t)
+ if _, err := provision.SetRecordRun(ctx, x.wd.s, x.wd.acct, "app", offHost, provision.RunPick(x.id, "box-1", "boot-1", time.Now().Add(time.Minute))); err != nil {
+ t.Fatal(err)
+ }
+ retireLifetime(t, x.wd.s, x.wd.acct, "app", offHost)
+ var none execLog
+ for _, judged := range []bool{false, true} {
+ var judge *provision.LifetimeJudge
+ if judged {
+ judge = newBoxJudge(claimsOffEnv)
+ }
+ err := worker(t, x, judge, &none).launchable(ctx, x.req, newRunHandle())
+ if judged != errors.Is(err, errRunNotOurs) {
+ t.Fatalf("judged %v: launchable = %v", judged, err)
+ }
+ }
+ })
+
+ t.Run("the steering loop's desired state and retention's app records", func(t *testing.T) {
+ wd := newOffWorld(t)
+ steered := provision.Record{Schema: 1, Host: "steer.example.com", Slug: "steer", Account: wd.acct, App: "app", Channel: "stable", Steering: "latency", CreatedAt: wd.now}
+ if err := provision.CreateRecord(ctx, wd.s, steered); err != nil {
+ t.Fatal(err)
+ }
+ retireLifetime(t, wd.s, wd.acct, "app", steered.Host)
+ for _, judged := range []bool{false, true} {
+ var judge *provision.LifetimeJudge
+ if judged {
+ judge = newBoxJudge(claimsOffEnv)
+ }
+ pctx := provision.WithLifetimes(ctx, judge)
+ var desired []registry.Route
+ if err := steeringPass(pctx, wd.s, func(_ context.Context, routes []registry.Route, _ [][]byte) error {
+ desired = routes
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ if got := len(desired) == 1; got == judged {
+ t.Fatalf("judged %v: the steering desired state %+v", judged, desired)
+ }
+ recs, err := appRecords(pctx, wd.s, wd.acct, "app")
+ if err != nil {
+ t.Fatal(err)
+ }
+ // offHost's record has no slot and stays; the steered one is
+ // stale.
+ if want := map[bool]int{false: 2, true: 1}[judged]; len(recs) != want {
+ t.Fatalf("judged %v: retention read %d app records, want %d", judged, len(recs), want)
+ }
+ }
+ })
+
+ t.Run("the proof binding source as the agent and root construct it", func(t *testing.T) {
+ wd := newOffWorld(t)
+ retireLifetime(t, wd.s, wd.acct, "app", offHost)
+ builds := map[string]func() *proofBindingSource{
+ // The agent's and the root history's, with its cached index.
+ "agent": func() *proofBindingSource { return productionProofBindingSource(wd.s, systemdUnits{}, t.TempDir()) },
+ // productionRestartRequestPass's.
+ "root restart requests": func() *proofBindingSource {
+ s := productionProofBindingSource(wd.s, systemdUnits{}, t.TempDir())
+ s.cacheIndex = false
+ return s
+ },
+ // productionRootUpdatesHeld's, under the retained census.
+ "root updates under the census": func() *proofBindingSource {
+ s := productionProofBindingSource(wd.s, systemdUnits{}, t.TempDir())
+ s.cacheIndex = false
+ s.claim = func(ctx context.Context, host string) (hostclaim.Observed, error) {
+ return hostclaim.Read(ctx, wd.s, host)
+ }
+ s.box = func(context.Context) (string, error) { return "box", nil }
+ return s
+ },
+ }
+ for name, build := range builds {
+ s := build()
+ if s.judge == nil {
+ t.Fatalf("%s: host claims off, the source has no judge", name)
+ }
+ // No record (the named-unit fallback after it has nothing for
+ // this host either), and the lookup itself found it gone.
+ if got, err := s.record(context.Background(), offHost); err == nil || got.record != nil {
+ t.Fatalf("%s: proof evidence from a stale record: %+v, %v", name, got, err)
+ }
+ if _, _, err := s.findRecord(context.Background(), offHost); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("%s: the record lookup answered %v, want not found", name, err)
+ }
+ s = build()
+ s.judge = nil
+ if got, err := s.record(context.Background(), offHost); err != nil || got.record == nil {
+ t.Fatalf("%s: fixture: unjudged, the record is not found: %v", name, err)
+ }
+ }
+ })
+
+ t.Run("the seed callback wireSeedPlacement installs", func(t *testing.T) {
+ for _, judged := range []bool{false, true} {
+ deploy, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ rec := seedRecordFixture(t, deploy, seeds.KindReplica, seeds.StateReady)
+ snap := putSnapshot(t, data, rec.Host, rec.Seed.ID)
+ retireLifetime(t, deploy, rec.Account, rec.App, rec.Host)
+ var judge *provision.LifetimeJudge
+ if judged {
+ judge = newBoxJudge(claimsOffEnv)
+ }
+ eb := &activator.ExecBackend{}
+ wireSeedPlacement(eb, false, storeOpener(data), storeOpener(deploy), seedBox{Judge: judge})
+ err := eb.OnSeedPlaced(context.Background(), rec.Account, rec.App, rec.Host, rec.Seed.ID)
+ if judged {
+ if !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the callback answered %v for a stale record", err)
+ }
+ if !snapshotPresent(t, data, snap) || seedStateOf(t, deploy, rec) != seeds.StateReady {
+ t.Fatal("the callback placed a stale record's seed or deleted its snapshot")
+ }
+ continue
+ }
+ if err != nil || snapshotPresent(t, data, snap) {
+ t.Fatalf("fixture: unjudged, the callback did not place the seed: %v", err)
+ }
+ }
+ })
+}
+
+// failingLauncher refuses every launch.
+type failingLauncher struct{}
+
+func (failingLauncher) RunCmd(context.Context, registry.Route, activator.CmdSpec) (activator.CmdHandle, error) {
+ return nil, errors.New("the unit could not start")
+}
+
+// A claims-on box reads no slot: newBoxJudge gives it no judge, and every
+// consumer reads the record as it always has, a record beside a retired
+// slot included, without a slot read.
+func TestClaimsOnConsumersReadNoSlot(t *testing.T) {
+ if j := newBoxJudge(claimsOnEnv); j != nil {
+ t.Fatal("a claims-on box has a lifetime judge")
+ }
+ ctx := context.Background()
+ wd := newOffWorld(t)
+ retireLifetime(t, wd.s, wd.acct, "app", offHost)
+ reads := &slotReads{Store: wd.s}
+ wd.s = reads
+ judge := newBoxJudge(claimsOnEnv)
+ wd.judge = judge
+
+ a := wd.box("ie-a", "", nil)
+ a.tick()
+ if rt := a.row(offHost); rt == nil {
+ t.Fatal("claims on, reconcile did not mint the record's row")
+ }
+ refreshOfflineView(ctx, reads, a.w.cfg.Offline)
+ if recs, err := appRecords(provision.WithLifetimes(ctx, judge), reads, wd.acct, "app"); err != nil || len(recs) != 1 {
+ t.Fatalf("retention: %d records, %v", len(recs), err)
+ }
+ if _, err := steeredRoutesFromRecords(provision.WithLifetimes(ctx, judge), reads); err != nil {
+ t.Fatal(err)
+ }
+ if rec, err := judgedRecord(ctx, reads, a.w.cfg.Judge, wd.acct, "app", offHost); err != nil || rec == nil {
+ t.Fatalf("the moves reader: %v", err)
+ }
+ if rec, err := MovesRecordsFrom(reads)(ctx, wd.acct, "app", offHost); err != nil || rec == nil {
+ t.Fatalf("MovesRecordsFrom: %v", err)
+ }
+ // ClaimRootHTTP's claims-on reader: its context carries no judge.
+ rec := wd.record()
+ inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := &hostclaim.Claim{Host: offHost, Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey(rec.Account, rec.App, rec.Host), Account: rec.Account, App: rec.App, Incarnation: inc}}
+ if _, _, err := rootClaimRecord(ctx, reads, c, t.TempDir()); err != nil {
+ t.Fatalf("rootClaimRecord: %v", err)
+ }
+ if n := reads.reads(); n != 0 {
+ t.Fatalf("claims-on consumers read %d slots", n)
+ }
+}
+
+// One judge per box process, from the host-claim switch, carried by every
+// consumer the agent arms.
+func TestBoxJudgeIsSharedAndGated(t *testing.T) {
+ if newBoxJudge(claimsOnEnv) != nil {
+ t.Fatal("claims on: a judge")
+ }
+ j := newBoxJudge(claimsOffEnv)
+ if j == nil {
+ t.Fatal("claims off: no judge")
+ }
+
+ // The moves worker and its offline view, as armMovesPass arms them.
+ prev := armedMovesWorker.Load()
+ t.Cleanup(func() { armedMovesWorker.Store(prev) })
+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil, j)
+ w := armedMovesWorker.Load()
+ if w.cfg.Judge != j || w.cfg.Offline == nil || w.cfg.Offline.judge != j {
+ t.Fatalf("the moves worker's judge %p, its view's %p, want %p", w.cfg.Judge, w.cfg.Offline.judge, j)
+ }
+ // The reconcile pass's config.
+ if cfg := boxProvisionConfig(j, nil); cfg.Judge != j {
+ t.Fatal("the reconcile config does not carry the box's judge")
+ }
+ // The proof binding source sets its own from the same switch.
+ t.Setenv("CARLOS_HOST_CLAIM", "")
+ if productionProofBindingSource(release.NewDirStore(t.TempDir()), systemdUnits{}, t.TempDir()).judge == nil {
+ t.Fatal("claims off: the proof binding source has no judge")
+ }
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
+ if productionProofBindingSource(release.NewDirStore(t.TempDir()), systemdUnits{}, t.TempDir()).judge != nil {
+ t.Fatal("claims on: the proof binding source has a judge")
+ }
+
+ // The agent builds it once and hands that one value to every
+ // consumer it arms (runAgent, in source).
+ calls := agentCalls(t, "runAgent")
+ if n := calls.assigns["boxJudge"]; n != 1 {
+ t.Fatalf("runAgent assigns boxJudge %d times, want once", n)
+ }
+ for _, callee := range []string{"armMovesPass", "armRunsPass", "fastConvergeLoop", "instancesLoop", "statusLoop", "retentionLoop", "steeringLoop"} {
+ if !calls.passes(callee, "boxJudge") {
+ t.Errorf("runAgent's %s call does not pass boxJudge", callee)
+ }
+ }
+ if !calls.seedBoxJudge {
+ t.Error("runAgent's seedBox does not carry boxJudge")
+ }
+}
+
+// agentFacts is what one function's source says about boxJudge.
+type agentFacts struct {
+ assigns map[string]int
+ args map[string]bool // callee + "/" + an identifier it is passed
+ seedBoxJudge bool
+}
+
+// passes reports whether callee is handed ident as an argument.
+func (a agentFacts) passes(callee, ident string) bool { return a.args[callee+"/"+ident] }
+
+func agentCalls(t *testing.T, fn string) agentFacts {
+ t.Helper()
+ f, err := parser.ParseFile(token.NewFileSet(), "agent.go", nil, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ out := agentFacts{assigns: map[string]int{}, args: map[string]bool{}}
+ for _, d := range f.Decls {
+ fd, ok := d.(*ast.FuncDecl)
+ if !ok || fd.Name.Name != fn {
+ continue
+ }
+ ast.Inspect(fd.Body, func(n ast.Node) bool {
+ switch n := n.(type) {
+ case *ast.AssignStmt:
+ for _, l := range n.Lhs {
+ if id, ok := l.(*ast.Ident); ok {
+ out.assigns[id.Name]++
+ }
+ }
+ case *ast.CallExpr:
+ if id, ok := n.Fun.(*ast.Ident); ok {
+ for _, a := range n.Args {
+ if ai, ok := a.(*ast.Ident); ok {
+ out.args[id.Name+"/"+ai.Name] = true
+ }
+ }
+ }
+ case *ast.CompositeLit:
+ if id, ok := n.Type.(*ast.Ident); ok && id.Name == "seedBox" {
+ for _, e := range n.Elts {
+ if kv, ok := e.(*ast.KeyValueExpr); ok {
+ if k, ok := kv.Key.(*ast.Ident); ok && k.Name == "Judge" {
+ if v, ok := kv.Value.(*ast.Ident); ok && v.Name == "boxJudge" {
+ out.seedBoxJudge = true
+ }
+ }
+ }
+ }
+ }
+ }
+ return true
+ })
+ }
+ return out
+}
+
+// Root's restart readers: the target, the history creation and the held
+// update all read the record under root's judge. The three production
+// constructions open root-owned state (/var/lib/carlos-claims), so the
+// readers are driven over the restart fixture's root state, and every
+// production construction of the source is pinned to carry a judge given
+// by its caller.
+func TestRootRestartReadersAreJudged(t *testing.T) {
+ f := newRestartConsumptionFixture(t)
+ ctx := t.Context()
+ unjudged := f.p.source
+ in, err := unjudged.installation(ctx, observedHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rt := registry.Route{Host: observedHost, Account: in.Account, App: in.App}
+ if _, err := unjudged.target(ctx, in); err != nil {
+ t.Fatalf("fixture: unjudged target: %v", err)
+ }
+ if _, err := unjudged.creation(ctx, rt); err != nil {
+ t.Fatalf("fixture: unjudged creation: %v", err)
+ }
+ retireLifetime(t, f.e.store, in.Account, in.App, observedHost)
+ judged := unjudged
+ judged.judge = newBoxJudge(claimsOffEnv)
+ if target, err := judged.target(ctx, in); !errors.Is(err, os.ErrNotExist) || target.Host != "" {
+ t.Fatalf("a stale record's restart target: %+v, %v", target, err)
+ }
+ if creation, err := judged.creation(ctx, rt); !errors.Is(err, os.ErrNotExist) || creation != "" {
+ t.Fatalf("a stale record's history creation: %q, %v", creation, err)
+ }
+ held := rootUpdateSource{source: judged, store: f.e.store, binding: &proofBindingSource{}}
+ if input, err := held.resolve(ctx, observedHost); !errors.Is(err, os.ErrNotExist) || input.Route.Host != "" {
+ t.Fatalf("a stale record's held update: %+v, %v", input, err)
+ }
+
+ // Every production construction of the source carries a judge, and
+ // each of the three production functions takes it from its caller.
+ files, err := filepath.Glob("*.go")
+ if err != nil {
+ t.Fatal(err)
+ }
+ fset := token.NewFileSet()
+ takes := map[string]bool{}
+ literals := 0
+ for _, name := range files {
+ if strings.HasSuffix(name, "_test.go") {
+ continue
+ }
+ file, err := parser.ParseFile(fset, name, nil, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, d := range file.Decls {
+ if fd, ok := d.(*ast.FuncDecl); ok {
+ for _, p := range fd.Type.Params.List {
+ if star, ok := p.Type.(*ast.StarExpr); ok {
+ if sel, ok := star.X.(*ast.SelectorExpr); ok && sel.Sel.Name == "LifetimeJudge" {
+ takes[fd.Name.Name] = true
+ }
+ }
+ }
+ }
+ }
+ ast.Inspect(file, func(n ast.Node) bool {
+ lit, ok := n.(*ast.CompositeLit)
+ if !ok {
+ return true
+ }
+ if id, ok := lit.Type.(*ast.Ident); !ok || id.Name != "rootRestartRequestSource" {
+ return true
+ }
+ literals++
+ for _, e := range lit.Elts {
+ if kv, ok := e.(*ast.KeyValueExpr); ok {
+ if k, ok := kv.Key.(*ast.Ident); ok && k.Name == "judge" {
+ return true
+ }
+ }
+ }
+ t.Errorf("%s: a rootRestartRequestSource with no judge", fset.Position(lit.Pos()))
+ return true
+ })
+ }
+ if literals < 3 {
+ t.Fatalf("found %d production constructions of the source, want at least 3", literals)
+ }
+ for _, fn := range []string{"productionRestartRequestPass", "rootHistoryCreation", "productionRootUpdatesHeld"} {
+ if !takes[fn] {
+ t.Errorf("%s does not take the caller's judge", fn)
+ }
+ }
+ if calls := agentCalls(t, "runAdoptUnderRootRelease"); !calls.passes("productionRestartRequestPass", "rootJudge") || calls.assigns["rootJudge"] != 1 {
+ t.Error("runAdoptUnderRootRelease does not hand productionRestartRequestPass root's judge")
+ }
+}
diff --git a/internal/host/movesoffline_e2e_test.go b/internal/host/movesoffline_e2e_test.go
index ee87816cb..c93c1ae79 100644
--- a/internal/host/movesoffline_e2e_test.go
+++ b/internal/host/movesoffline_e2e_test.go
@@ -302,6 +302,9 @@ func newE2EWorld(t *testing.T, opts landOpts) *e2eWorld {
// The deployment: the ops account, app's manifest and the instance's
// record at home.
wd := &offWorld{t: t, s: rs, acct: release.OpsAccount, now: now}
+ if opts.judged {
+ wd.judge = provision.NewLifetimeJudge()
+ }
if _, _, err := rs.Get(ctx, domains.AccountManifestKey(wd.acct)); errors.Is(err, os.ErrNotExist) {
manifest, _ := json.Marshal(map[string]any{"id": console.OpsAccountID, "sqid": wd.acct, "name": "ops", "created_at": now})
if _, err := rs.Put(ctx, domains.AccountManifestKey(wd.acct), manifest, ""); err != nil {
@@ -1064,6 +1067,10 @@ func (wd *offWorld) reportFor(op, role, box string) moves.OfflineReport {
// the console's doors and both boxes' production passes.
func TestOfflineMoveE2E(t *testing.T) {
t.Run("done", func(t *testing.T) { offlineMoveDone(t, landOpts{}) })
+ // The same move with every box judging lifetimes, as a claims-off agent
+ // does (newBoxJudge): a record with no slot is valid, so every step and
+ // every finishing assertion is the unjudged run's.
+ t.Run("done, judged", func(t *testing.T) { offlineMoveDone(t, landOpts{judged: true}) })
t.Run("nested source with launch files", func(t *testing.T) { offlineMoveDone(t, landOpts{nested: true}) })
t.Run("no files directory", func(t *testing.T) { offlineMoveDone(t, landOpts{noFiles: true}) })
t.Run("rollback after a tampered archive", offlineMoveRollback)
diff --git a/internal/host/movesoffline_land_test.go b/internal/host/movesoffline_land_test.go
index 011d743fa..c261ffd7c 100644
--- a/internal/host/movesoffline_land_test.go
+++ b/internal/host/movesoffline_land_test.go
@@ -83,6 +83,9 @@ type landOpts struct {
configKey string
noFiles bool
nested bool
+ // judged arms every box with a lifetime judge, as a claims-off agent
+ // arms newBoxJudge's (the end-to-end test's claims-off variant).
+ judged bool
}
// landWorld is offHost at home (ie-a, data-ie) moving to uk (uk-box,
diff --git a/internal/host/movesoffline_test.go b/internal/host/movesoffline_test.go
index 1eb64817d..b965d8e38 100644
--- a/internal/host/movesoffline_test.go
+++ b/internal/host/movesoffline_test.go
@@ -45,6 +45,10 @@ type offWorld struct {
acct string
mu sync.Mutex
now time.Time
+ // judge, when set, is every box's lifetime judge, wired as the agent
+ // wires newBoxJudge's (the moves worker, its offline view and the
+ // reconcile pass). Nil: the boxes judge nothing, as before.
+ judge *provision.LifetimeJudge
}
func (wd *offWorld) clock() time.Time {
@@ -198,7 +202,7 @@ func (b *offBox) getenv(k string) string {
// worker over the same registry and data directory.
func (b *offBox) restart() {
b.svc = activator.New(activator.Config{Bucket: "data", Idle: time.Hour}, &fakeWakeBackend{}, b.reg)
- cfg := moveCfg{Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock}
+ cfg := moveCfg{Judge: b.wd.judge, Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock}
armOffline(&cfg, b.dd, b.reg, b.svc, nil, hostwrite.Config{}, b.getenv)
// The stage-requested marker is root's business, and its throttled
// timer would outlive the test's directories.
@@ -210,7 +214,7 @@ func (b *offBox) restart() {
func (b *offBox) reconcile() {
b.wd.t.Helper()
- cfg := provision.ReconcileConfig{Region: b.region, DataDir: b.dd}
+ cfg := provision.ReconcileConfig{Region: b.region, DataDir: b.dd, Judge: b.wd.judge}
if b.nested {
cfg.Layout = func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }
} else {