9711789a2 · carlos/platform

diff --git a/internal/host/agent.go b/internal/host/agent.go

index 559c0a27c..c09c85b6d 100644

--- a/internal/host/agent.go

+++ b/internal/host/agent.go

@@ -237,6 +237,9 @@ func runAgent(args []string, buildClaims ClaimFactory) {

if err != nil {

log.Fatalf("agent: host claims: %v", err)

}

+ // The box's lifetime judge, built once and shared by every consumer of

+ // instance records below (newBoxJudge): nil with host claims on.

+ boxJudge := newBoxJudge(claimEntryEnv(claimCtx))

dd := dataDir()

regPath := peekRegistryPath(args, dd)

// A second handle on the same registry file the edge opens — the registry

@@ -596,8 +599,8 @@ func runAgent(args []string, buildClaims ClaimFactory) {

// comes before the latch).

wireSeedPlacement(eb, claims.Writes.Enabled, seedDataOpener(cfg.Bucket, cfg.Region),

func() (release.Store, error) { return openDeployStore(context.Background()) },

- seedBox{Fence: svc, Prefix: cfg.Prefix, Swept: seedSweptAtStart.Load, NextFenceGen: fenceGenerations(reg)})

- armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes, claims.Providers, rolloutSink, moveCarry(bs, bsErr))

+ seedBox{Fence: svc, Prefix: cfg.Prefix, Swept: seedSweptAtStart.Load, NextFenceGen: fenceGenerations(reg), Judge: boxJudge})

+ armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes, claims.Providers, rolloutSink, moveCarry(bs, bsErr), boxJudge)

// carlos run's worker, armed with the moves pass as it was

// before the startup moved to the background (#281). Its

// passes ride the converge and instances loops, which the

@@ -605,7 +608,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {

// latch: no run is picked up before the startup sweep has

// stopped a previous agent's leftover units, the cmd units

// among them.

- armRunsPass(svc, reg, dd, rolloutSink)

+ armRunsPass(svc, reg, dd, rolloutSink, boxJudge)

},

Ready: func() {

close(started)

@@ -789,7 +792,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {

// The relocation pass (stage B Task 10): it says itself what it started.

startRelocationPass(agentCtx, claimCtx, claims, reg, svc)

go adoptLoop(reg, svc, sd, claims.Writes)

- go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision)

+ go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, boxJudge, claims.Providers.Provision)

if logSink != nil {

go logshipLoop(logSink, status.Label(readMachineID()))

}

@@ -801,12 +804,12 @@ func runAgent(args []string, buildClaims ClaimFactory) {

// sees them (cfg.HostWrites.Enabled, set by configureWakeClaims).

buildStateClaims := cfg != nil && cfg.HostWrites.Enabled

collectBuildStates := func() (map[string]status.BuildStatePins, error) { return buildStatePins(reg, buildStateClaims) }

- go statusLoop(reg, svc, collectBuildStates)

+ go statusLoop(reg, svc, collectBuildStates, boxJudge)

// The memory early-warning (memory-pressure spec §2.4): per box, gated

// by CARLOS_HOST_METRICS — one log line and nothing else where unset.

go memwatchLoop()

go domainsLoop(reg, peekAcmeCache(args), claims.Writes.Enabled)

- go instancesLoop(reg, claims.Writes, bs, claims.Providers.Provision)

+ go instancesLoop(reg, claims.Writes, bs, boxJudge, claims.Providers.Provision)

// The deployment-global election, armed BEFORE the loops it gates so a

// single-box deployment is holding well before their first tick (a

// minute away) — see housekeeper.go. Not a `go` of its own: it publishes

@@ -815,7 +818,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {

if claims.Writes.Enabled {

stopHousekeeper = startHousekeeper()

}

- go retentionLoop(reg, collectBuildStates)

+ go retentionLoop(reg, collectBuildStates, boxJudge)

go registrySnapshotLoop(reg)

go edgesFileLoop(reg)

// Region HA enrollment evidence (spec 1 §5.1): once, at start, and only

@@ -832,7 +835,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {

// own loop for mintMeshKeyLoop's exact reason, and dormant on every box

// today: CARLOS_STEERING_ZONE_ID unset means one log line at arm and

// nothing else, ever. See steering_loop.go.

- go steeringLoop()

+ go steeringLoop(boxJudge)

// The sweep reads the registry only through this closure — the stamped

// bespoke units to sample beside the carlos- ones, re-read each tick.

go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) }, alerts)

@@ -1569,6 +1572,14 @@ func reconcileSerialized(ctx context.Context, s release.Store, reg *registry.Reg

func reconcileSerializedWith(ctx context.Context, s release.Store, reg *registry.Registry, cfg provision.ReconcileConfig, w *movesWorker) error {

reconcileMu.Lock()

defer reconcileMu.Unlock()

+ // Claims off, the pass judges every record by the box's judge and its

+ // record writes run judged and sequenced (nil claims on: unchanged). A

+ // held row whose own record is gone is deleted only while its op holds

+ // it, with the move's timers.

+ ctx = provision.WithLifetimes(ctx, cfg.Judge)

+ if cfg.DeleteIfHold == nil && reg != nil {

+ cfg.DeleteIfHold = reg.DeleteIfHoldWithTimers

+ }

// A fresh row is minted at the layout this box's intent names

// (ReconcileConfig.Layout): nested on a nested box, so a new tenant

// never runs flat there. Set here, the one place both callers pass

@@ -2674,8 +2685,9 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {

//

// Thin and untested by convention, like adoptLoop/statusLoop — the tick's

// whole decision lives in fastTick/fastMembers, which are tested.

-func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config, serialNotify chan<- struct{}, bs *buildStates, configure ...func(*provision.ReconcileConfig)) {

- ctx := context.Background()

+func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config, serialNotify chan<- struct{}, bs *buildStates, judge *provision.LifetimeJudge, configure ...func(*provision.ReconcileConfig)) {

+ // Every pass this loop runs carries the box's judge (newBoxJudge).

+ ctx := provision.WithLifetimes(context.Background(), judge)

store, err := openDeployStore(ctx)

if err != nil {

declinedHoldsStay("no deploy store")

@@ -2795,7 +2807,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu

}

defer func(start time.Time) { logSlow("serial pass stages", start, strings.Join(took, ", ")) }(time.Now())

- if err := reconcileSerialized(ctx, store, reg, configuredClaimProvision(provisionTickConfig(os.Getenv), configure)); err != nil {

+ if err := reconcileSerialized(ctx, store, reg, boxProvisionConfig(judge, configure)); err != nil {

log.Printf("agent: fast converge reconcile: %v", err)

}

mark("reconcile")

@@ -3110,6 +3122,15 @@ func provisionTickConfig(getenv func(string) string) provision.ReconcileConfig {

return cfg

}

+// boxProvisionConfig is one reconcile pass's Config on this box:

+// provisionTickConfig from the environment, the claim configuration, and the

+// box's lifetime judge (newBoxJudge; nil with host claims on).

+func boxProvisionConfig(judge *provision.LifetimeJudge, configure []func(*provision.ReconcileConfig)) provision.ReconcileConfig {

+ cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure)

+ cfg.Judge = judge

+ return cfg

+}

+

// resolveHost is the production DNS seam: the default resolver with a hard

// 3s timeout, so one dead nameserver can't stall a whole reconcile pass.

func resolveHost(host string) ([]string, error) {

@@ -3355,14 +3376,15 @@ func domainsDuty(pass func(context.Context) error) housekeeperDuty {

// already had a store+reg in scope) — that coupled binding materialization

// to PLATFORM_APPS_DOMAIN being set, which has nothing to do with fleets;

// this loop has no such gate and shares the same store already opened above.

-func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildStates, configure ...func(*provision.ReconcileConfig)) {

- ctx := context.Background()

+func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildStates, judge *provision.LifetimeJudge, configure ...func(*provision.ReconcileConfig)) {

+ // Every pass this loop runs carries the box's judge (newBoxJudge).

+ ctx := provision.WithLifetimes(context.Background(), judge)

store, err := openDeployStore(ctx)

if err != nil {

log.Printf("agent: instances loop disabled: %v", err)

return

}

- cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure)

+ cfg := boxProvisionConfig(judge, configure)

backfill := newClaimBackfill(writes, store, cfg.ReleasePub, os.Getenv("CARLOS_DEPLOYMENT_BUCKET"))

log.Printf("agent: instances loop armed (5min + jitter), region %q pools %v", cfg.Region, cfg.Pools)

// Platform-only: a fleet box (CARLOS_FLEET set) never reads other

@@ -3443,7 +3465,7 @@ func offlineReassertLoop(ctx context.Context, s release.Store) {

// silently, until the day it wins the lease (housekeeper.go). On a

// single-box deployment the holder is always this box, so the behaviour is

// unchanged.

-func retentionLoop(reg *registry.Registry, buildStates func() (map[string]status.BuildStatePins, error)) {

+func retentionLoop(reg *registry.Registry, buildStates func() (map[string]status.BuildStatePins, error), judge *provision.LifetimeJudge) {

ctx := context.Background()

// Platform boxes only (spec F6). Retention prunes the PLATFORM's release

// objects — a customer's box scheduling deletes of them was already wrong

@@ -3491,7 +3513,9 @@ func retentionLoop(reg *registry.Registry, buildStates func() (map[string]status

// that is halfway through its deletes when renewal fails must stop, not

// finish against a lock another box now owns (housekeeper.go).

prune := housekeeperDuty{"retention pass", func(ctx context.Context) error {

- return retentionPass(ctx, store, reports, pub, reg)

+ // The box's judge: a record its slot no longer vouches for pins

+ // no build (appRecords).

+ return retentionPass(provision.WithLifetimes(ctx, judge), store, reports, pub, reg)

}}

delay := time.Minute

for {

@@ -3677,8 +3701,9 @@ func upstreamStates() map[string][]status.UpstreamState {

// logging that every 5 minutes forever would bury the log in noise. Any

// OTHER acquire error (a real S3/credential problem) is logged, same as any

// other tick failure.

-func statusLoop(reg *registry.Registry, svc *activator.Service, buildStates func() (map[string]status.BuildStatePins, error)) {

- ctx := context.Background()

+func statusLoop(reg *registry.Registry, svc *activator.Service, buildStates func() (map[string]status.BuildStatePins, error), judge *provision.LifetimeJudge) {

+ // The status tick runs under the box's judge like every other pass.

+ ctx := provision.WithLifetimes(context.Background(), judge)

// bucket is read for the lease key alone (lease.Acquire below), which a

// fleet box never reaches: shouldSkipPublisherElection short-circuits the

// election before it, and a fleet box is the only shape that gets a store

@@ -4547,6 +4572,8 @@ func runAdoptUnderRootRelease(args []string, buildClaims ClaimFactory, ensureUni

log.Fatalf("adopt: host claims: %v", err)

}

claimsEnabled := hostwrite.Enabled(claimEntryEnv(claimCtx))

+ // Root's lifetime judge (newBoxJudge), for its restart readers.

+ rootJudge := newBoxJudge(claimEntryEnv(claimCtx))

var claims *ClaimWiring

if !claimsEnabled {

// Keep the flag-off startup effects in their old order:

@@ -4701,7 +4728,7 @@ func runAdoptUnderRootRelease(args []string, buildClaims ClaimFactory, ensureUni

// parsed below and is fatal when unusable. A restart request must not fail

// on a key it never needs.

if *restartOnly {

- logRestartRequestError(productionRestartRequestPass(ctx, store, claims.Writes, func() error {

+ logRestartRequestError(productionRestartRequestPass(ctx, store, claims.Writes, rootJudge, func() error {

routes, err := reg.List()

if err != nil {

log.Fatalf("adopt: list routes: %v", err)

@@ -4754,7 +4781,7 @@ func runAdoptUnderRootRelease(args []string, buildClaims ClaimFactory, ensureUni

// with an unstageable release would otherwise never reach this line — and

// a wedged instance is most worth cycling precisely on a box where

// something else is already wrong.

- logRestartRequestError(productionRestartRequestPass(ctx, store, claims.Writes, func() error {

+ logRestartRequestError(productionRestartRequestPass(ctx, store, claims.Writes, rootJudge, func() error {

if routes, err := reg.List(); err != nil {

log.Printf("adopt: list routes for the restart pass: %v", err)

} else {

diff --git a/internal/host/agent_test.go b/internal/host/agent_test.go

index 67b614b40..bb956e1b8 100644

--- a/internal/host/agent_test.go

+++ b/internal/host/agent_test.go

@@ -1862,7 +1862,7 @@ func TestRetentionLoopIsAPlatformBoxJob(t *testing.T) {

t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")

t.Setenv("CARLOS_DEPLOYMENT_DIR", "")

- retentionLoop(testRegistry(t), noBuildStates) // must return; a fleet box never reaches the loop

+ retentionLoop(testRegistry(t), noBuildStates, nil) // must return; a fleet box never reaches the loop

if got := logs(); !strings.Contains(got, skipLine) {

t.Fatalf("log = %q, want the F6 skip line verbatim", got)

@@ -1905,7 +1905,7 @@ func TestRetentionLoopIsAPlatformBoxJob(t *testing.T) {

// runs retentionPass a minute from now, which lists routes — nil

// would panic the whole test binary long after the test that started

// it has passed.

- go retentionLoop(testRegistry(t), noBuildStates)

+ go retentionLoop(testRegistry(t), noBuildStates, nil)

deadline := time.Now().Add(5 * time.Second)

for {

diff --git a/internal/host/boxjudge.go b/internal/host/boxjudge.go

new file mode 100644

index 000000000..f40ef0f26

--- /dev/null

+++ b/internal/host/boxjudge.go

@@ -0,0 +1,35 @@

+package host

+

+import (

+ "context"

+

+ "github.com/carlosframework/platform/internal/hostwrite"

+ "github.com/carlosframework/platform/internal/provision"

+ "github.com/carlosframework/platform/internal/release"

+)

+

+// newBoxJudge is the box's lifetime judge (self-serve moves plan 1, Task 6),

+// built once per process from the host-claim switch and handed to every

+// consumer of instance records on the box: the reconcile pass, the moves

+// worker and its offline view, the runs worker, the steering loop,

+// retention, the status tick, the seed callback and root's proof and

+// restart readers. A claims-off box gets a judge: every record it acts on

+// is observed, the record then its host incarnation slot, and every pass

+// context carries it (provision.WithLifetimes), so record updates refuse a

+// record its slot no longer vouches for and record writes are sequenced. A

+// claims-on box gets nil: it reads no slot and behaves exactly as before.

+func newBoxJudge(getenv func(string) string) *provision.LifetimeJudge {

+ if hostwrite.Enabled(getenv) {

+ return nil

+ }

+ return provision.NewLifetimeJudge()

+}

+

+// judgedRecord is provision.GetRecord under j: a record its host incarnation

+// slot no longer vouches for, or a tombstone, reads as os.ErrNotExist; a

+// lifetime that cannot be judged is an error. A nil j is provision.GetRecord

+// exactly.

+func judgedRecord(ctx context.Context, s release.Store, j *provision.LifetimeJudge, acct, app, host string) (*provision.Record, error) {

+ rec, _, err := provision.ReadRecordJudged(ctx, s, j, acct, app, host)

+ return rec, err

+}

diff --git a/internal/host/moveshold.go b/internal/host/moveshold.go

index 3b9680953..a3652ea4b 100644

--- a/internal/host/moveshold.go

+++ b/internal/host/moveshold.go

@@ -263,7 +263,7 @@ func refreshMovesHolds(ctx context.Context, s release.Store) {

if v == nil || s == nil {

return

}

- recs, err := provision.ListAllRecords(ctx, s)

+ recs, _, err := provision.ListAllRecordsStrictJudged(ctx, s, provision.JudgeFrom(ctx))

if err != nil {

log.Printf("moves: hold view: list records: %v; rows are kept and timers frozen until a moves pass lists them", err)

return

diff --git a/internal/host/movesoffline.go b/internal/host/movesoffline.go

index a78e946af..d14c807a3 100644

--- a/internal/host/movesoffline.go

+++ b/internal/host/movesoffline.go

@@ -129,6 +129,10 @@ func offlineHold(box moveCfg) func(rec provision.Record) (status, op string, ok

// always have.

type offlineView struct {

home, region string

+ // judge is the box's lifetime judge (moveCfg.Judge): a listing read

+ // outside a moves pass (refreshOfflineView) is judged by it, so a record

+ // its slot no longer vouches for keeps no row. Nil: unjudged.

+ judge *provision.LifetimeJudge

// written is reconcile's HoldWritten for the offline hooks: the

// stage-requested marker after a hold write (registry.SetHold's

// contract, G10). Nil writes nothing.

@@ -150,7 +154,7 @@ type offlineSnapshot struct {

// newOfflineView builds the view for the box cfg describes. Unbuilt until

// its first refresh.

func newOfflineView(cfg moveCfg, written func(host string)) *offlineView {

- return &offlineView{home: cfg.ParkHomeRegion, region: cfg.parkRegion(), written: written}

+ return &offlineView{home: cfg.ParkHomeRegion, region: cfg.parkRegion(), written: written, judge: cfg.Judge}

}

// refresh rebuilds the view from a record list that was read, with the

@@ -242,7 +246,7 @@ func refreshOfflineView(ctx context.Context, s release.Store, v *offlineView) {

if v == nil || s == nil {

return

}

- recs, unreadable, err := provision.ListAllRecordsStrict(ctx, s)

+ recs, unreadable, err := provision.ListAllRecordsStrictJudged(provision.WithLifetimes(ctx, v.judge), s, v.judge)

if err != nil {

log.Printf("moves: offline view: list records: %v; every row is kept until a listing reads", err)

v.fail()

@@ -414,10 +418,11 @@ func (w *movesWorker) offlineReassert(ctx context.Context, s release.Store) int

sort.Strings(hosts)

w.pass.Lock()

defer w.pass.Unlock()

+ ctx = provision.WithLifetimes(ctx, cfg.Judge)

acted := 0

for _, host := range hosts {

owner := snap.owners[host]

- rec, err := provision.GetRecord(ctx, s, owner[0], owner[1], host)

+ rec, err := judgedRecord(ctx, s, cfg.Judge, owner[0], owner[1], host)

if err != nil {

w.logTrouble(host, fmt.Errorf("offline move: read the record to report again: %w", err))

continue

diff --git a/internal/host/movespass.go b/internal/host/movespass.go

index 16efd306e..725e17975 100644

--- a/internal/host/movespass.go

+++ b/internal/host/movespass.go

@@ -268,13 +268,14 @@ func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.Mirro

// this runs, in the same agent start. events is the agent's

// rollout sink, which the move log tells the app's members through (Plan 6

// Task 3).

-func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, providers ClaimProviders, events activator.EventSink, carry func(ctx context.Context, host, bucket, prefix string) error) {

+func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, providers ClaimProviders, events activator.EventSink, carry func(ctx context.Context, host, bucket, prefix string) error, judge *provision.LifetimeJudge) {

dataDir = movesDataDir(dataDir)

readyDelta, ok := moveReadyDeltaFromEnv(os.Getenv)

if !ok {

log.Printf("agent: %s=%q ignored: want a whole number of bytes, 0 or more — keeping the built-in default of %d", moveReadyDeltaEnv, os.Getenv(moveReadyDeltaEnv), defaultMoveReadyDelta)

}

cfg := moveCfg{

+ Judge: judge,

registerLanding: providers.registerLanding,

Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")),

OwnBucket: hib.Bucket,

@@ -697,6 +698,10 @@ func runMovesPass(ctx context.Context, s release.Store) {

// moveCfg carries what the worker needs, injectable for tests.

type moveCfg struct {

+ // Judge is the box's lifetime judge (newBoxJudge): nil on a claims-on

+ // box. Every pass of the worker runs under provision.WithLifetimes of

+ // it, and every record the worker reads is judged by it.

+ Judge *provision.LifetimeJudge

registerLanding func(context.Context, hostclaim.Observed) error

claimRecords bool // Captured CARLOS_HOST_CLAIM, independent of the moves switch.

Region string // this box's placement code, "" included (the home)

@@ -1064,6 +1069,9 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {

if cfg.claimRecords {

ctx = provision.WithRecordSequence(ctx)

}

+ // Claims off, the pass reads and updates records under the box's

+ // judge, sequenced (nil on a claims-on box: unchanged).

+ ctx = provision.WithLifetimes(ctx, cfg.Judge)

if cfg.Enabled {

// The pages the outbox and the index keep undelivered are sent again

// first, before any record or fleet read (movespaging.go, fix round

@@ -1076,7 +1084,8 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {

}

// Strict returns exactly ListAllRecords' records and errors, plus the

// hosts whose record did not decode, which the offline keep view keeps.

- recs, unreadable, err := provision.ListAllRecordsStrict(ctx, s)

+ // Judged, a record its slot no longer vouches for is not listed.

+ recs, unreadable, err := provision.ListAllRecordsStrictJudged(ctx, s, cfg.Judge)

if err != nil {

return fmt.Errorf("moves: list records: %w", err)

}

diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go

index ab3cd74c4..114af4a97 100644

--- a/internal/host/movespass_test.go

+++ b/internal/host/movespass_test.go

@@ -962,7 +962,7 @@ func TestArmMovesPassReadsTheReadyDelta(t *testing.T) {

{"lots", defaultMoveReadyDelta},

} {

t.Setenv("CARLOS_MOVE_READY_DELTA_BYTES", tc.raw)

- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil)

+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil, nil)

if got := armedMovesWorker.Load().cfg.ReadyDelta; got != tc.want {

t.Errorf("CARLOS_MOVE_READY_DELTA_BYTES=%q armed a bar of %d, want %d", tc.raw, got, tc.want)

}

@@ -981,7 +981,7 @@ func TestArmMovesPassReadsTheSwitch(t *testing.T) {

{"", false}, {"on", true}, {"1", true}, {"true", true}, {"off", false}, {"yes", false},

} {

t.Setenv(moves.SwitchEnv, tc.raw)

- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil)

+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil, nil)

if got := armedMovesWorker.Load().cfg.Enabled; got != tc.want {

t.Errorf("%s=%q armed Enabled=%v, want %v", moves.SwitchEnv, tc.raw, got, tc.want)

}

diff --git a/internal/host/movesphase1_test.go b/internal/host/movesphase1_test.go

index 60c53ddd0..a5559b936 100644

--- a/internal/host/movesphase1_test.go

+++ b/internal/host/movesphase1_test.go

@@ -144,7 +144,7 @@ func TestPhase1IsInertWithTheSwitchOff(t *testing.T) {

}

t.Setenv(moves.SwitchEnv, v)

eb := &activator.ExecBackend{}

- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, eb, hostwrite.Config{}, ClaimProviders{}, nil, nil)

+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, eb, hostwrite.Config{}, ClaimProviders{}, nil, nil, nil)

if (eb.Strict != nil) != on {

t.Fatalf("switch %v: eb.Strict set = %v", on, eb.Strict != nil)

}

diff --git a/internal/host/proofbinding.go b/internal/host/proofbinding.go

index f5bfbeeae..8d63cc7e8 100644

--- a/internal/host/proofbinding.go

+++ b/internal/host/proofbinding.go

@@ -66,6 +66,11 @@ type proofBindingSource struct {

openFile func(string) (*os.File, error)

unitDirs []string

healthy func(string, string, string, string) *release.HealthyBuild

+ // judge is the process's lifetime judge, set by the production

+ // constructor from the host-claim switch: findRecord arms whatever

+ // context it is handed with it, so a record its slot no longer vouches

+ // for yields no proof evidence. Nil on a claims-on box.

+ judge *provision.LifetimeJudge

}

func proofRouteReady(rt registry.Route, timeout time.Duration) func() error {

@@ -113,6 +118,9 @@ func productionProofBindingSource(store release.Store, sys unitSystem, optDir st

source.persistent = rootstate.PersistentIdentity

source.signedRecovery = true

}

+ // The agent and root construct the source here, so both judge the

+ // records they read for proof (the retained census included).

+ source.judge = newBoxJudge(os.Getenv)

if hostwrite.Enabled(os.Getenv) {

source.claim = func(ctx context.Context, host string) (hostclaim.Observed, error) {

return hostclaim.Read(ctx, store, host)

diff --git a/internal/host/proofnamed.go b/internal/host/proofnamed.go

index f916bde76..e75789ca7 100644

--- a/internal/host/proofnamed.go

+++ b/internal/host/proofnamed.go

@@ -162,6 +162,9 @@ func proofExecPath(b proofBinding, opt string) string {

}

func (s *proofBindingSource) findRecord(ctx context.Context, host string) (*provision.Record, string, error) {

+ // FindRecordByHost and RecordIndex.Find judge through the context's

+ // judge; proofObserver.observe hands this context.Background().

+ ctx = provision.WithLifetimes(ctx, s.judge)

if !s.cacheIndex {

return provision.FindRecordByHost(ctx, s.store, host)

}

diff --git a/internal/host/retention.go b/internal/host/retention.go

index 549578fa0..e6b1c7248 100644

--- a/internal/host/retention.go

+++ b/internal/host/retention.go

@@ -16,6 +16,7 @@ import (

"errors"

"fmt"

"log"

+ "os"

"strings"

"time"

@@ -305,11 +306,26 @@ func appRecords(ctx context.Context, store release.Store, acct, app string) ([]p

if !ok || host == "" {

continue

}

- rec, err := provision.GetRecord(ctx, store, acct, app, host)

+ // Observed under the pass context's judge (retentionLoop): a record

+ // its slot no longer vouches for pins nothing and is omitted; one

+ // that cannot be judged is an error, as every read failure is. A

+ // nil judge reads as GetRecord did.

+ obs := provision.JudgeFrom(ctx).Observe(ctx, store, acct, app, host)

+ var err error

+ switch {

+ case obs.Present && obs.Verdict == provision.LifetimeStale:

+ continue

+ case obs.Verdict == provision.LifetimeUnreadable || obs.Verdict == provision.LifetimeCorrupt:

+ err = obs.Err

+ case !obs.Present:

+ err = os.ErrNotExist

+ case obs.Rec.Retired:

+ err = fmt.Errorf("retired: %w", os.ErrNotExist)

+ }

if err != nil {

return nil, fmt.Errorf("reading the record of %s: %w", host, err)

}

- recs = append(recs, *rec)

+ recs = append(recs, *obs.Rec)

}

return recs, nil

}

diff --git a/internal/host/roothistory.go b/internal/host/roothistory.go

index 8e889a821..5fdecba14 100644

--- a/internal/host/roothistory.go

+++ b/internal/host/roothistory.go

@@ -22,7 +22,7 @@ func openRootHistory(store release.Store, optDir string) (*release.RootHistoryPa

}

source := productionProofBindingSource(store, systemdUnits{}, optDir)

h := boundRootHistory(&release.RootHistory{Store: s}, source.resolve, productionProofObserver(optDir))

- h.Creation = rootHistoryCreation(store)

+ h.Creation = rootHistoryCreation(store, newBoxJudge(os.Getenv))

m := rootHistoryMaintenance{history: h.History, store: store, source: source,

checks: productionRootRestartChecks(systemdUnits{}, optDir), now: time.Now,

readback: func(ctx context.Context, b release.HistoryBinding, snapshot func(context.Context) (release.HistoryMigrationStatus, error)) error {

diff --git a/internal/host/roothistory_creation.go b/internal/host/roothistory_creation.go

index c2515a0b7..78d049c83 100644

--- a/internal/host/roothistory_creation.go

+++ b/internal/host/roothistory_creation.go

@@ -47,7 +47,7 @@ func rootHistoryInstallation(in hostguard.RootInstallation) string {

return hex.EncodeToString(sum[:])

}

-func rootHistoryCreation(store release.Store) func(context.Context, registry.Route) (string, error) {

+func rootHistoryCreation(store release.Store, judge *provision.LifetimeJudge) func(context.Context, registry.Route) (string, error) {

return func(ctx context.Context, rt registry.Route) (string, error) {

census, err := rootparticipation.Open()

if err != nil {

@@ -59,7 +59,7 @@ func rootHistoryCreation(store release.Store) func(context.Context, registry.Rou

return "", err

}

defer state.Close()

- return (rootRestartRequestSource{state: state, store: store, census: census.Census}).creation(ctx, rt)

+ return (rootRestartRequestSource{state: state, store: store, census: census.Census, judge: judge}).creation(ctx, rt)

}

@@ -71,7 +71,8 @@ func (s rootRestartRequestSource) creation(ctx context.Context, rt registry.Rout

if in.Host != rt.Host || in.Account != rt.Account || in.App != rt.App || in.Record != provision.RecordKey(in.Account, in.App, in.Host) {

return "", os.ErrPermission

}

- rec, _, err := provision.GetRecordETag(ctx, s.store, in.Account, in.App, in.Host)

+ ctx = provision.WithLifetimes(ctx, s.judge)

+ rec, _, err := provision.ReadRecordJudged(ctx, s.store, s.judge, in.Account, in.App, in.Host)

if err != nil {

return "", err

}

diff --git a/internal/host/rootrestart_requests.go b/internal/host/rootrestart_requests.go

index 338bcbb53..a4f6189ad 100644

--- a/internal/host/rootrestart_requests.go

+++ b/internal/host/rootrestart_requests.go

@@ -11,6 +11,7 @@ import (

"github.com/carlosframework/platform/internal/hostbox"

"github.com/carlosframework/platform/internal/hostguard"

"github.com/carlosframework/platform/internal/hostwrite"

+ "github.com/carlosframework/platform/internal/provision"

"github.com/carlosframework/platform/internal/release"

"github.com/carlosframework/platform/internal/rootparticipation"

"github.com/carlosframework/platform/internal/rootstate"

@@ -27,7 +28,7 @@ type rootRestartRequests struct {

}

// runAdoptUnderRootRelease owns release exclusion for both call sites.

-func productionRestartRequestPass(ctx context.Context, store release.Store, writes hostwrite.Config, oldPass func() error) error {

+func productionRestartRequestPass(ctx context.Context, store release.Store, writes hostwrite.Config, judge *provision.LifetimeJudge, oldPass func() error) error {

return restartRequestPass(legacyRootRestart, oldPass, func() error {

census, err := rootparticipation.Open()

if err != nil {

@@ -49,7 +50,7 @@ func productionRestartRequestPass(ctx context.Context, store release.Store, writ

source.cacheIndex = false

checks := productionRootRestartChecks(sys, release.OptDir())

checks.host = source.forHost

- p := rootRestartRequests{journal: &release.RootRestarts{Store: journal}, source: rootRestartRequestSource{state: state, store: store, census: census.Census}, writes: writes, sys: sys, checks: checks, getenv: os.Getenv, optDir: release.OptDir()}

+ p := rootRestartRequests{journal: &release.RootRestarts{Store: journal}, source: rootRestartRequestSource{state: state, store: store, census: census.Census, judge: judge}, writes: writes, sys: sys, checks: checks, getenv: os.Getenv, optDir: release.OptDir()}

return p.run(ctx)

})

}

diff --git a/internal/host/rootrestart_requests_source.go b/internal/host/rootrestart_requests_source.go

index 52c62e09f..9775ff68f 100644

--- a/internal/host/rootrestart_requests_source.go

+++ b/internal/host/rootrestart_requests_source.go

@@ -20,6 +20,11 @@ type rootRestartRequestSource struct {

state rootstate.Store

store release.Store

census func() (rootparticipation.Census, error)

+ // judge is the root process's lifetime judge (newBoxJudge), given by

+ // every production construction: the target, creation and update

+ // readers judge the record and run under provision.WithLifetimes of it.

+ // Nil on a claims-on box.

+ judge *provision.LifetimeJudge

}

func (s rootRestartRequestSource) installation(ctx context.Context, host string) (hostguard.RootInstallation, error) {

@@ -67,7 +72,8 @@ func (s rootRestartRequestSource) target(ctx context.Context, in hostguard.RootI

if in.Record != provision.RecordKey(in.Account, in.App, in.Host) {

return release.RestartTarget{}, os.ErrPermission

}

- rec, revision, err := provision.GetRecordETag(ctx, s.store, in.Account, in.App, in.Host)

+ ctx = provision.WithLifetimes(ctx, s.judge)

+ rec, revision, err := provision.ReadRecordJudged(ctx, s.store, s.judge, in.Account, in.App, in.Host)

if err != nil {

return release.RestartTarget{}, err

}

diff --git a/internal/host/rootupdates.go b/internal/host/rootupdates.go

index 63494428a..5c375d745 100644

--- a/internal/host/rootupdates.go

+++ b/internal/host/rootupdates.go

@@ -77,10 +77,10 @@ func rootFailureLegacy(data string) ([]byte, error) {

}

func productionRootUpdates(dd string, probes ...func() (rootparticipation.Ledger, bool, error)) release.RootUpdateOpener {

- return productionRootUpdatesHeld(dd, nil, probes...)

+ return productionRootUpdatesHeld(dd, nil, newBoxJudge(os.Getenv), probes...)

}

-func productionRootUpdatesHeld(dd string, held *release.GenerationBinding, probes ...func() (rootparticipation.Ledger, bool, error)) release.RootUpdateOpener {

+func productionRootUpdatesHeld(dd string, held *release.GenerationBinding, judge *provision.LifetimeJudge, probes ...func() (rootparticipation.Ledger, bool, error)) release.RootUpdateOpener {

return func(ctx context.Context, store release.Store, pub ed25519.PublicKey) (*release.RootUpdateInputs, error) {

probe := rootparticipation.OpenForUpdate

if len(probes) > 1 || len(probes) == 1 && probes[0] == nil {

@@ -128,7 +128,7 @@ func productionRootUpdatesHeld(dd string, held *release.GenerationBinding, probe

binding.box = func(context.Context) (string, error) { return c.Box, nil }

observer := productionProofObserver(release.OptDir())

observer.resolve = binding.resolve

- p := rootUpdateSource{legacy: legacy, ready: func(rt registry.Route) error { return rootUpdateReady(ctx, rt) }, publish: publishRootRetry, source: rootRestartRequestSource{state: fences, store: store, census: func() (rootparticipation.Census, error) { return c, nil }}, binding: binding, state: state, book: book, store: store, pub: pub, data: dd, observe: func(rt registry.Route) (liveBuild, error) { return observer.observe(rt, 0) }}

+ p := rootUpdateSource{legacy: legacy, ready: func(rt registry.Route) error { return rootUpdateReady(ctx, rt) }, publish: publishRootRetry, source: rootRestartRequestSource{state: fences, store: store, census: func() (rootparticipation.Census, error) { return c, nil }, judge: judge}, binding: binding, state: state, book: book, store: store, pub: pub, data: dd, observe: func(rt registry.Route) (liveBuild, error) { return observer.observe(rt, 0) }}

p.heldGeneration = held

return &release.RootUpdateInputs{Resolve: p.resolve, Close: func() error { return errors.Join(fences.Close(), state.Close()) }}, nil

}

@@ -149,7 +149,8 @@ func (p rootUpdateSource) resolve(ctx context.Context, host string) (release.Roo

if in.Empty || in.Record != provision.RecordKey(in.Account, in.App, host) {

return out, os.ErrPermission

}

- rec, etag, err := provision.GetRecordETag(ctx, p.store, in.Account, in.App, host)

+ ctx = provision.WithLifetimes(ctx, p.source.judge)

+ rec, etag, err := provision.ReadRecordJudged(ctx, p.store, p.source.judge, in.Account, in.App, host)

if err != nil {

return out, err

}

diff --git a/internal/host/rootupdates_runtime.go b/internal/host/rootupdates_runtime.go

index 903ed3d11..62f45f346 100644

--- a/internal/host/rootupdates_runtime.go

+++ b/internal/host/rootupdates_runtime.go

@@ -196,7 +196,7 @@ func beginRootGenerationUpdate(ctx context.Context, s *rootGenerationSource, b r

if err != nil {

return nil, nil, err

}

- in, err := productionRootUpdatesHeld(dataDir(), &b)(ctx, s.store, pub)

+ in, err := productionRootUpdatesHeld(dataDir(), &b, newBoxJudge(os.Getenv))(ctx, s.store, pub)

if err != nil {

return nil, nil, err

}

diff --git a/internal/host/runs.go b/internal/host/runs.go

index f85eccefd..a2b44d7ba 100644

--- a/internal/host/runs.go

+++ b/internal/host/runs.go

@@ -59,6 +59,11 @@ type runsWorker struct {

// it launched would have its directory swept from under it.

booted atomic.Bool

+ // judge is the box's lifetime judge (newBoxJudge): every pass and every

+ // run's execution runs under provision.WithLifetimes of it, and every

+ // record read is judged. Nil on a claims-on box.

+ judge *provision.LifetimeJudge

+

mu sync.Mutex

noted map[string]provision.Record // host -> the record reconcile read

active map[string]*runHandle // run id -> the run this agent is executing

@@ -99,6 +104,7 @@ func (w *runsWorker) pass(ctx context.Context) {

if !w.booted.Load() {

return

}

+ ctx = provision.WithLifetimes(ctx, w.judge)

w.mu.Lock()

noted := w.noted

w.noted = map[string]provision.Record{}

@@ -119,7 +125,8 @@ func (w *runsWorker) bootPass(ctx context.Context) {

if err := runs.SweepRunDirs(w.dataDir); err != nil {

log.Printf("agent: runs: boot sweep of .carlos-run: %v", err)

}

- recs, err := provision.ListAllRecords(ctx, w.store)

+ ctx = provision.WithLifetimes(ctx, w.judge)

+ recs, _, err := provision.ListAllRecordsStrictJudged(ctx, w.store, w.judge)

if err != nil {

log.Printf("agent: runs: boot pass: %v", err)

return

@@ -236,7 +243,9 @@ func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt regist

log.Printf("agent: runs: %s %s: pickup answered %v, but the record names the run as this box's", rec.Host, id, err)

}

log.Printf("agent: runs: %s: picked run %s", rec.Host, id)

- w.start(func() { w.exec(context.Background(), *got, *req, rt, h) })

+ // Detached from the pass, and judged like it: execute's SetRecordRun

+ // calls refuse a record its slot no longer vouches for.

+ w.start(func() { w.exec(provision.WithLifetimes(context.Background(), w.judge), *got, *req, rt, h) })

}

// unfitLogMax caps the ids firstUnfit remembers; past it the set starts

@@ -275,7 +284,7 @@ func requestFits(rec provision.Record, id string, req runs.Request) error {

func (w *runsWorker) pickedAnyway(ctx context.Context, rec provision.Record, id string) *provision.Record {

for range 3 {

rctx, cancel := context.WithTimeout(ctx, runReadBound)

- cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host)

+ cur, err := judgedRecord(rctx, w.store, w.judge, rec.Account, rec.App, rec.Host)

cancel()

if errors.Is(err, os.ErrNotExist) {

return nil

@@ -356,7 +365,7 @@ func (w *runsWorker) closeInterrupted(ctx context.Context, rec provision.Record)

// armMovesPass: the behaviour is the worker's, tested there. events must be

// a genuinely nil interface when there is no log store, never a nil

// *applog.Sink (the trap agent.go documents).

-func armRunsPass(svc *activator.Service, reg *registry.Registry, dataDir string, events activator.EventSink) {

+func armRunsPass(svc *activator.Service, reg *registry.Registry, dataDir string, events activator.EventSink, judge *provision.LifetimeJudge) {

if svc == nil || reg == nil {

return

}

@@ -367,6 +376,7 @@ func armRunsPass(svc *activator.Service, reg *registry.Registry, dataDir string,

}

w := newRunsWorker(store, reg.Get, svc, dataDir, BoxLabel(), newBootID())

w.events = events

+ w.judge = judge

armedRuns.Store(w)

log.Printf("agent: runs armed (box %s, boot %s)", w.box, w.boot)

}

diff --git a/internal/host/seedwiring.go b/internal/host/seedwiring.go

index 62e71bdf9..760f8413a 100644

--- a/internal/host/seedwiring.go

+++ b/internal/host/seedwiring.go

@@ -47,8 +47,14 @@ func (o storeSeedObjects) Open(ctx context.Context, host, id string) (io.ReadClo

// placed (or committed) with this id is that case: a success, and the

// delete still runs, because the earlier call's delete may be the one that

// failed. A CAS refused for any other reason deletes nothing.

-func seedPlacedHook(deploy, data release.Store) func(ctx context.Context, acct, app, host, id string) error {

+//

+// judge is the box's lifetime judge: the callback's context is armed with it

+// first, whatever context the activator hands it, so on a claims-off box

+// SetRecordSeed refuses a record its slot no longer vouches for

+// (os.ErrNotExist, returned) and the snapshot is not deleted.

+func seedPlacedHook(deploy, data release.Store, judge *provision.LifetimeJudge) func(ctx context.Context, acct, app, host, id string) error {

return func(ctx context.Context, acct, app, host, id string) error {

+ ctx = provision.WithLifetimes(ctx, judge)

if deploy == nil || data == nil {

return errors.New("seed recording is not wired on this box")

}

@@ -57,7 +63,7 @@ func seedPlacedHook(deploy, data release.Store) func(ctx context.Context, acct,

}

_, err := provision.SetRecordSeed(ctx, deploy, acct, app, host, provision.SeedPlace(id))

if errors.Is(err, provision.ErrSeedConflict) {

- rec, gerr := provision.GetRecord(ctx, deploy, acct, app, host)

+ rec, gerr := judgedRecord(ctx, deploy, judge, acct, app, host)

if gerr != nil {

return gerr

}

@@ -116,6 +122,9 @@ type seedBox struct {

Prefix string

Swept func() bool

NextFenceGen func() (uint64, error)

+ // Judge is the box's lifetime judge, for the placed callback

+ // (seedPlacedHook). Nil on a claims-on box.

+ Judge *provision.LifetimeJudge

}

// wireSeedPlacement arms seeded instances on this box (seeded-instances

@@ -150,7 +159,7 @@ func wireSeedPlacement(eb *activator.ExecBackend, claimArmed bool, openData, ope

return

}

eb.Seeds = storeSeedObjects{data}

- eb.OnSeedPlaced = seedPlacedHook(deploy, data)

+ eb.OnSeedPlaced = seedPlacedHook(deploy, data, box.Judge)

h := &provision.SeedHooks{Data: data}

if !claimArmed {

h.NestedOK = eb.NestedOK

diff --git a/internal/host/seedwiring_test.go b/internal/host/seedwiring_test.go

index 7863222b2..008fd99b9 100644

--- a/internal/host/seedwiring_test.go

+++ b/internal/host/seedwiring_test.go

@@ -84,7 +84,7 @@ func TestSeedPlacedHookCASesThenDeletesOnlyThisSeed(t *testing.T) {

other, _ := seeds.NewID()

mine, theirs := putSnapshot(t, data, rec.Host, id), putSnapshot(t, data, rec.Host, other)

- hook := seedPlacedHook(deploy, data)

+ hook := seedPlacedHook(deploy, data, nil)

if err := hook(ctx, rec.Account, rec.App, rec.Host, id); err != nil {

t.Fatal(err)

}

@@ -123,7 +123,7 @@ func TestSeedPlacedHookIsIdempotent(t *testing.T) {

rec := seedRecordFixture(t, deploy, seeds.KindReplica, seeds.StateReady)

key := putSnapshot(t, data, rec.Host, rec.Seed.ID)

- boxA, boxB := seedPlacedHook(deploy, data), seedPlacedHook(deploy, data)

+ boxA, boxB := seedPlacedHook(deploy, data, nil), seedPlacedHook(deploy, data, nil)

if err := boxA(ctx, rec.Account, rec.App, rec.Host, rec.Seed.ID); err != nil {

t.Fatal(err)

}

@@ -154,7 +154,7 @@ func TestSeedPlacedHookIsIdempotent(t *testing.T) {

wg.Add(1)

go func() {

defer wg.Done()

- errs[i] = seedPlacedHook(deploy2, data2)(ctx, rec2.Account, rec2.App, rec2.Host, rec2.Seed.ID)

+ errs[i] = seedPlacedHook(deploy2, data2, nil)(ctx, rec2.Account, rec2.App, rec2.Host, rec2.Seed.ID)

}()

}

wg.Wait()

@@ -185,7 +185,7 @@ func TestSeedPlacedHookRefusedDeletesNothing(t *testing.T) {

deploy, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())

rec := seedRecordFixture(t, deploy, tc.kind, tc.state)

key := putSnapshot(t, data, rec.Host, rec.Seed.ID)

- if err := seedPlacedHook(deploy, data)(ctx, rec.Account, rec.App, rec.Host, rec.Seed.ID); !errors.Is(err, provision.ErrSeedConflict) {

+ if err := seedPlacedHook(deploy, data, nil)(ctx, rec.Account, rec.App, rec.Host, rec.Seed.ID); !errors.Is(err, provision.ErrSeedConflict) {

t.Fatalf("err = %v, want ErrSeedConflict", err)

}

if !snapshotPresent(t, data, key) {

@@ -200,14 +200,14 @@ func TestSeedPlacedHookRefusedDeletesNothing(t *testing.T) {

deploy, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())

id, _ := seeds.NewID()

key := putSnapshot(t, data, "shop-9.example.com", id)

- if err := seedPlacedHook(deploy, data)(ctx, domains.Sqid(2), "shop", "shop-9.example.com", id); err == nil {

+ if err := seedPlacedHook(deploy, data, nil)(ctx, domains.Sqid(2), "shop", "shop-9.example.com", id); err == nil {

t.Fatal("placing a seed with no record succeeded")

}

if !snapshotPresent(t, data, key) {

t.Fatal("a placement with no record deleted the snapshot")

}

// A box that has not wired the stores records nothing.

- if err := seedPlacedHook(nil, data)(ctx, domains.Sqid(2), "shop", "shop-9.example.com", id); err == nil {

+ if err := seedPlacedHook(nil, data, nil)(ctx, domains.Sqid(2), "shop", "shop-9.example.com", id); err == nil {

t.Fatal("an unwired hook reported the seed placed")

}

diff --git a/internal/host/steering_loop.go b/internal/host/steering_loop.go

index 6b2379d99..32d218663 100644

--- a/internal/host/steering_loop.go

+++ b/internal/host/steering_loop.go

@@ -97,7 +97,9 @@ func loadEdgesFileBodies(ctx context.Context, s release.Store) ([][]byte, error)

// untouched so the fold's own loud, documented refusals still fire. This

// loader must not become a second, quieter policy gate.

func steeredRoutesFromRecords(ctx context.Context, s release.Store) ([]registry.Route, error) {

- recs, err := provision.ListAllRecords(ctx, s)

+ // Judged by the pass context's judge (steeringLoop): a record its slot

+ // no longer vouches for steers nothing. Nil on a claims-on box.

+ recs, _, err := provision.ListAllRecordsStrictJudged(ctx, s, provision.JudgeFrom(ctx))

if err != nil {

return nil, err

}

@@ -201,8 +203,9 @@ func steeringDuty(zoneID string, pass func(context.Context) error) housekeeperDu

// plus one Route53 health check per EIP, which is what adds rung-2

// withdrawal for steered hosts even single-box. Nothing about it needs a

// second region to be correct.

-func steeringLoop() {

- ctx := context.Background()

+func steeringLoop(judge *provision.LifetimeJudge) {

+ // The box's judge rides every pass's context (steeredRoutesFromRecords).

+ ctx := provision.WithLifetimes(context.Background(), judge)

if shouldSkipPublisherElection(os.Getenv) {

log.Printf("agent: steering loop is a platform-box job — skipped on a fleet box")

return

@@ -229,7 +232,7 @@ func steeringLoop() {

}

appsDomain := strings.TrimSpace(os.Getenv("PLATFORM_APPS_DOMAIN"))

pass = func(ctx context.Context) error {

- return steeringPass(ctx, store, func(ctx context.Context, routes []registry.Route, edges [][]byte) error {

+ return steeringPass(provision.WithLifetimes(ctx, judge), store, func(ctx context.Context, routes []registry.Route, edges [][]byte) error {

return steering.ConvergeDeployment(ctx, client, zoneID, routes, edges, appsDomain)

})

}

diff --git a/internal/host/steering_loop_test.go b/internal/host/steering_loop_test.go

index 6e2ba9add..e9ad82d99 100644

--- a/internal/host/steering_loop_test.go

+++ b/internal/host/steering_loop_test.go

@@ -466,7 +466,7 @@ func TestSteeringLoopIsAPlatformBoxJob(t *testing.T) {

t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")

t.Setenv("CARLOS_DEPLOYMENT_DIR", "")

- steeringLoop() // must return; a fleet box never reaches the loop

+ steeringLoop(nil) // must return; a fleet box never reaches the loop

if got := logs(); !strings.Contains(got, skipLine) {

t.Fatalf("log = %q, want the skip line verbatim", got)

@@ -488,7 +488,7 @@ func TestSteeringLoopArmsWithoutReleasePubkey(t *testing.T) {

t.Setenv("CARLOS_STEERING_ZONE_ID", "")

buf := captureRacyLog(t)

- go steeringLoop()

+ go steeringLoop(nil)

waitFor(t, "the steering loop to arm despite no CARLOS_RELEASE_PUBKEY", func() bool {

return strings.Contains(buf.String(), "agent: steering loop armed")

@@ -505,7 +505,7 @@ func TestSteeringLoopDormantWhenZoneUnset(t *testing.T) {

t.Setenv("CARLOS_STEERING_ZONE_ID", "")

buf := captureRacyLog(t)

- go steeringLoop()

+ go steeringLoop(nil)

waitFor(t, "the dormant line", func() bool {

return strings.Contains(buf.String(), "steering: zone unset — dormant")

@@ -524,7 +524,7 @@ func TestSteeringLoopDormantWhenZoneMalformed(t *testing.T) {

t.Setenv("CARLOS_STEERING_ZONE_ID", "/hostedzone/Z0ABCDEF1234")

buf := captureRacyLog(t)

- go steeringLoop()

+ go steeringLoop(nil)

waitFor(t, "the dormant line for a malformed zone id", func() bool {

return strings.Contains(buf.String(), "steering: zone unset — dormant")

carlos / platform Public

host: one lifetime judge per box, on every pass context and every record consumer