amadan
host+release: fast-converge review fixes — serialize Reconcile, seed absent serial at zero, export Adoptable (#101)
Review findings on 99ddfde: - Critical: Reconcile now runs from two goroutines and its removal phase deletes routes minted after its own record snapshot — reconcileSerialized (reconcileMu, syncExecSerialized's shape) is now the only way either loop runs it; pinned by TestReconcileSerializedHoldsTheMutex. - Important: two adopt.go premise comments claimed a narrowed pass never resolves held channels/production flags — no longer true with unstaged stable/beta routes in the fast set; rewritten with the real cost/safety reasoning (backoff bounds the flag reads; a PARTIAL Production map still must not repair a tainted bake clock). - Important: fastSweepable was a byte-copy of release.adoptable — exported as release.Adoptable (CapabilityRoute precedent), agent.go calls it. - Minor/rollout: an absent control/serial object now seeds at zero so the first-ever console bump on a fresh bucket fires instead of reading as history; TestFastTickActsOnSerialChangeNotBoot re-pins both halves.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
4 files changed,
+165
−57
internal/host/agent.go+46 −21internal/host/agent_test.go+80 −13internal/release/adopt.go+38 −22internal/release/adopt_test.go+1 −1
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 96c1881ae..04e1d6986 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -368,6 +368,31 @@ func syncExecSerialized(ctx context.Context, reg *registry.Registry, store relea |
| return appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc)) |
| } |
| +// reconcileMu serializes the two provisioning loops' Reconcile passes |
| +// against each other, the same shape as configSyncMu above. They can |
| +// genuinely overlap — instancesLoop's 5-minute pass mid-flight when a |
| +// serial bump fires fastConvergeLoop's full pass set — and Reconcile's |
| +// removal phase deletes any Provisioned, in-scope route that is absent from |
| +// ITS OWN record snapshot, which it lists at the top of the pass |
| +// (provision/reconcile.go): a route the newer pass minted between the older |
| +// pass's apps/ listing and its registry listing looks exactly like a |
| +// retired record to the older pass, which would delete it. The re-mint |
| +// minutes later comes back with its capability grants revoked |
| +// (registry.Delete purges grants — that is Delete's contract, not a bug). |
| +// Serializing in-process closes the race outright; the root adopt path runs |
| +// no Reconcile and is not this mutex's problem. |
| +var reconcileMu sync.Mutex |
| + |
| +// reconcileSerialized is the one way any agent loop runs provision.Reconcile |
| +// — the mutex lives in a shared wrapper, not at the call sites, so a future |
| +// third caller cannot forget it. Pinned by |
| +// TestReconcileSerializedHoldsTheMutex. |
| +func reconcileSerialized(ctx context.Context, s release.Store, reg *registry.Registry, cfg provision.ReconcileConfig) error { |
| + reconcileMu.Lock() |
| + defer reconcileMu.Unlock() |
| + return provision.Reconcile(ctx, s, reg, cfg) |
| +} |
| + |
| // bounce builds the restart callback SyncExec calls for a route whose |
| // rendered config actually changed. Shared by BOTH config loops (hourly |
| // configSyncLoop and the fast configGenLoop) rather than duplicated per |
| @@ -746,15 +771,6 @@ func fastPassOpts(dd string, stageReq func(host, ver string, err error), restart |
| } |
| } |
| -// fastSweepable mirrors release's adoptable predicate (adopt.go): the routes |
| -// a pass would consider at all. Anything else — platform aliases, non-ready |
| -// routes, routes with no app — must not count as outstanding work, or a |
| -// route the pass refuses on shape would make every tick sweep forever. |
| -func fastSweepable(rt registry.Route) bool { |
| - return rt.Kind == "instance" && rt.App != "" && !rt.Platform && |
| - (rt.Status == "" || rt.Status == "ready") |
| -} |
| - |
| // fastMembers computes the fast set for one tick — the routes with a reason |
| // to be looked at FASTER than the hourly floor: |
| // |
| @@ -778,7 +794,11 @@ func (d *fastDeps) fastMembers(now time.Time) []string { |
| var members []string |
| seen := map[string]bool{} |
| for _, rt := range routes { |
| - if !fastSweepable(rt) || seen[rt.Host] { |
| + // release.Adoptable, not a local copy: the sweep must count as |
| + // outstanding work exactly the routes a pass would consider — |
| + // anything else (platform aliases, non-ready routes, routes with no |
| + // app) would be swept forever, since the pass refuses it on shape. |
| + if !release.Adoptable(rt) || seen[rt.Host] { |
| continue |
| } |
| if isCanaryRoute(rt) { |
| @@ -825,9 +845,11 @@ func (d *fastDeps) backoffFor(host string) time.Duration { return d.step[host] } |
| // fastTick is one fast-converge tick — two independent jobs (spec §3-§4): |
| // |
| -// 1. Serial check: one ReadSerial. The FIRST successful sight of the |
| -// object only seeds lastSerial — boot must not replay history |
| -// (adoptLoop's 1-minute first pass covers boot). Afterward any CHANGE |
| +// 1. Serial check: one ReadSerial. The FIRST successful read only seeds |
| +// lastSerial — a present object at its value (boot must not replay |
| +// history; adoptLoop's 1-minute first pass covers boot), an absent one |
| +// at zero (absence is not history, and the first-ever bump on a fresh |
| +// bucket must fire, not read as replay). Afterward any CHANGE |
| // (not just increments — a reset must not wedge the box) runs the full |
| // pass set, clamped to one per fastFullClamp: a bump inside the window |
| // is deferred by NOT recording the acted serial, so the next tick past |
| @@ -838,19 +860,22 @@ func (d *fastDeps) backoffFor(host string) time.Duration { return d.step[host] } |
| // the sweep, members still unstaged back off; staged members clear |
| // their backoff state. |
| func fastTick(ctx context.Context, d *fastDeps, now time.Time) { |
| - doc, etag, err := release.ReadSerial(ctx, d.store) |
| + doc, _, err := release.ReadSerial(ctx, d.store) |
| switch { |
| case err != nil: |
| if !d.serialBroken { |
| log.Printf("agent: fast converge serial read: %v", err) |
| d.serialBroken = true |
| } |
| - case etag == "": |
| - // No serial object yet (pre-first-bump): nothing to seed — seeding |
| - // the zero doc here would make the first real bump replay as a |
| - // "change" that predates this box's interest. |
| - d.serialBroken = false |
| case !d.seeded: |
| + // The first successful read seeds — the ABSENT object included, |
| + // which ReadSerial reports as the zero doc. Absence is not history: |
| + // there is nothing to replay, and every deployment's bucket starts |
| + // without the object, so seeding zero is what lets the first-ever |
| + // console bump (0 → 1) fire instead of being swallowed as boot |
| + // noise. A PRESENT object seeds at its current value, and THAT is |
| + // the no-replay rule: the bumps behind it are history a booting box |
| + // must not act on (adoptLoop's 1-minute first pass covers boot). |
| d.lastSerial, d.seeded = doc.Serial, true |
| d.serialBroken = false |
| default: |
| @@ -933,7 +958,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service) { |
| // Each error is one log line, never fatal — the hourly floors stand |
| // underneath every leg. |
| fullPass: func(ctx context.Context) { |
| - if err := provision.Reconcile(ctx, store, reg, provisionTickConfig(os.Getenv)); err != nil { |
| + if err := reconcileSerialized(ctx, store, reg, provisionTickConfig(os.Getenv)); err != nil { |
| log.Printf("agent: fast converge reconcile: %v", err) |
| } |
| if err := release.Pass(ctx, reg, store, pub, optDir, restart, opts...); err != nil { |
| @@ -1249,7 +1274,7 @@ func instancesLoop(reg *registry.Registry) { |
| for { |
| time.Sleep(delay) |
| delay = 5*time.Minute + rand.N(time.Minute) |
| - if err := provision.Reconcile(ctx, store, reg, cfg); err != nil { |
| + if err := reconcileSerialized(ctx, store, reg, cfg); err != nil { |
| log.Printf("agent: instances pass: %v", err) |
| } |
| } |
diff --git a/internal/host/agent_test.go b/internal/host/agent_test.go| index aaa97093f..96f6c8c8a 100644 |
| --- a/internal/host/agent_test.go |
| +++ b/internal/host/agent_test.go |
| @@ -13,6 +13,7 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/appconfig" |
| + "github.com/carlosframework/platform/internal/provision" |
| "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| @@ -734,34 +735,100 @@ func (f *fastFixture) stageLive(t *testing.T, host, ver string) { |
| } |
| } |
| -// TestFastTickActsOnSerialChangeNotBoot pins seed-don't-replay: the first |
| -// sight of the serial object only records it (boot must not replay history — |
| -// adoptLoop's 1-minute first pass covers boot), a subsequent CHANGE runs the |
| -// full pass set exactly once, and an unchanged serial runs nothing. |
| +// mutexProbeStore fails the test if any store call arrives while |
| +// reconcileMu is NOT held — the deterministic pin that reconcileSerialized |
| +// really runs provision.Reconcile under the package mutex (TryLock succeeds |
| +// only when nobody holds it). No goroutines, no sleeps: holding the mutex |
| +// for the duration of the store work IS the serialization property. |
| +type mutexProbeStore struct { |
| + release.Store |
| + t *testing.T |
| + hits *int |
| +} |
| + |
| +func (s *mutexProbeStore) probe() { |
| + *s.hits++ |
| + if reconcileMu.TryLock() { |
| + reconcileMu.Unlock() |
| + s.t.Error("store consulted without reconcileMu held — a concurrent Reconcile pass could interleave and mass-retire fresh routes") |
| + } |
| +} |
| + |
| +func (s *mutexProbeStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + s.probe() |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +func (s *mutexProbeStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + s.probe() |
| + return s.Store.List(ctx, prefix) |
| +} |
| + |
| +// TestReconcileSerializedHoldsTheMutex pins finding-1's fix: Reconcile's |
| +// removal phase deletes any Provisioned in-scope route absent from its own |
| +// top-of-pass record snapshot, so two overlapping passes (instancesLoop's |
| +// 5-minute tick vs fastConvergeLoop's serial-triggered full set) could |
| +// delete a route the other pass had just minted — and the re-mint comes |
| +// back with capability grants revoked. Both loops therefore run Reconcile |
| +// only through reconcileSerialized, which must hold reconcileMu across the |
| +// whole pass. |
| +func TestReconcileSerializedHoldsTheMutex(t *testing.T) { |
| + var hits int |
| + s := &mutexProbeStore{Store: release.NewDirStore(t.TempDir()), t: t, hits: &hits} |
| + reg := restartTestRegistry(t) |
| + cfg := provision.ReconcileConfig{DataDir: t.TempDir()} |
| + if err := reconcileSerialized(context.Background(), s, reg, cfg); err != nil { |
| + t.Fatalf("reconcileSerialized: %v", err) |
| + } |
| + if hits == 0 { |
| + t.Fatal("the probe store was never consulted — this test pinned nothing") |
| + } |
| +} |
| + |
| +// TestFastTickActsOnSerialChangeNotBoot pins seed-don't-replay, in both of |
| +// its halves. An ABSENT serial object seeds at zero — absence is not |
| +// history, there is nothing to replay, and every production bucket starts |
| +// without the object, so a deployment's FIRST-EVER bump (0 → 1) must fire |
| +// rather than being swallowed as boot noise. A PRESENT object seeds at its |
| +// current value without replaying it (boot must not replay history — |
| +// adoptLoop's 1-minute first pass covers boot). After either seed, a CHANGE |
| +// runs the full pass set exactly once and an unchanged serial runs nothing. |
| func TestFastTickActsOnSerialChangeNotBoot(t *testing.T) { |
| env := newFastFixture(t) |
| var passes int |
| d := env.deps(func() { passes++ }) |
| now := time.Unix(1754640000, 0) |
| - fastTick(context.Background(), d, now) // no serial object at all: seeds nothing, runs nothing |
| - if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| - t.Fatalf("BumpSerial: %v", err) |
| - } |
| - fastTick(context.Background(), d, now.Add(2*time.Second)) // first sight: SEED only |
| + fastTick(context.Background(), d, now) // no serial object at all: seeds at zero, runs nothing |
| if passes != 0 { |
| - t.Fatalf("boot seed must not run a pass, got %d", passes) |
| + t.Fatalf("an absent serial object must not run a pass, got %d", passes) |
| } |
| if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| t.Fatalf("BumpSerial: %v", err) |
| } |
| - fastTick(context.Background(), d, now.Add(4*time.Second)) |
| + fastTick(context.Background(), d, now.Add(2*time.Second)) // the deployment's first-ever bump |
| if passes != 1 { |
| - t.Fatalf("serial change must run the pass set once, got %d", passes) |
| + t.Fatalf("the first-ever bump must fire (absence seeds at zero), got %d", passes) |
| } |
| - fastTick(context.Background(), d, now.Add(6*time.Second)) |
| + fastTick(context.Background(), d, now.Add(10*time.Second)) // past clamp, unchanged serial |
| if passes != 1 { |
| t.Fatalf("unchanged serial must not re-run, got %d", passes) |
| } |
| + |
| + // The boot rule for a PRESENT object: a restarted box (fresh state) |
| + // seeds at the current value and must not replay it as a change. |
| + var reboots int |
| + d2 := env.deps(func() { reboots++ }) |
| + fastTick(context.Background(), d2, now.Add(12*time.Second)) // first sight of serial=1: SEED only |
| + if reboots != 0 { |
| + t.Fatalf("boot seed of a present object must not run a pass, got %d", reboots) |
| + } |
| + if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| + t.Fatalf("BumpSerial: %v", err) |
| + } |
| + fastTick(context.Background(), d2, now.Add(14*time.Second)) |
| + if reboots != 1 { |
| + t.Fatalf("post-boot serial change must run the pass set once, got %d", reboots) |
| + } |
| } |
| // TestFastTickClampsFullPasses pins the 5s clamp: a bump landing inside the |
diff --git a/internal/release/adopt.go b/internal/release/adopt.go| index 7235d3d7a..ae00eb8fc 100644 |
| --- a/internal/release/adopt.go |
| +++ b/internal/release/adopt.go |
| @@ -250,7 +250,7 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551 |
| return err |
| } |
| adopts := func(rt registry.Route) bool { |
| - return adoptable(rt) && (extra == nil || extra(rt)) |
| + return Adoptable(rt) && (extra == nil || extra(rt)) |
| } |
| // Resolve every distinct (Account, App, Channel) once. Verification |
| @@ -331,11 +331,16 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551 |
| // Resolve each app's production flag once per pass — but ONLY for apps |
| // with a HELD head resolved this pass (spec §6). That filter is a cost |
| - // property, not a nicety: the canary fast loop (PassFilter with a |
| - // canary/*-only extra, every 2-3 SECONDS) resolves nothing but channels |
| - // whose HoldFor is 0, so it never pays a production-flag bucket read at |
| - // all; the hourly pass pays exactly one read per (account, app) with a |
| - // held head, not one per channel and not one per route. |
| + // property, not a nicety: any pass pays at most one read per (account, |
| + // app) with a held head, never one per channel or per route. For the |
| + // fast-converge sweep (PassFilter every 2-3 SECONDS) that usually means |
| + // zero flag reads — canary/* has HoldFor 0, and a fully-staged box's |
| + // sweep carries nothing else — but an UNSTAGED stable/beta route in the |
| + // sweep's set does resolve its held channel and pays this read too. |
| + // What keeps that bounded is the sweep's own membership, not this |
| + // filter: an unstaged host backs off 2s→5min (internal/host's |
| + // fastMembers), so a held, never-staging route costs one flag read per |
| + // backoff window at worst, not one per tick. |
| type prodState struct { |
| flag *ProductionFlag |
| err error // non-nil: the object exists but would not read or verify |
| @@ -411,17 +416,21 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551 |
| // changed-gated pass and the box would bake forever. |
| // |
| // But ONLY a full pass may perform that repair. A NARROWED pass |
| - // (extra != nil — the canary fast loop, every 2-3 seconds) resolves only |
| - // channels with no hold, so it fetched no production flags at all: its |
| - // in-memory Production map is EMPTY, not repaired. Letting it write would |
| - // overwrite the corrupt file with that emptiness AND clear the taint that |
| - // is currently failing closed, all before anything re-observed the flags |
| - // — and if the bucket flag had also been deleted, the latch would be gone |
| - // for good. So a tainted clock on a narrowed pass skips the Save entirely, |
| - // forced or ordinary: a pass that fetched no flags cannot repair what it |
| - // would overwrite. Nothing is lost — the canary observations it would |
| - // have recorded re-occur within seconds, and the hourly pass does the |
| - // repair with the flags actually in hand. |
| + // (extra != nil — the fast-converge sweep, every 2-3 seconds) fetches |
| + // production flags only for the held heads inside its narrowed set — |
| + // often none (canary/* has no hold), sometimes a few (an unstaged |
| + // stable/beta member), never reliably ALL of the apps whose latches the |
| + // clock file carries: its in-memory Production map is PARTIAL, not |
| + // repaired. Letting it write would overwrite the corrupt file with that |
| + // partial view AND clear the taint that is currently failing closed, |
| + // dropping the latch of every app outside its set before anything |
| + // re-observed those flags — and if a bucket flag had also been deleted, |
| + // that latch would be gone for good. So a tainted clock on a narrowed |
| + // pass skips the Save entirely, forced or ordinary: a pass that observed |
| + // only a slice of the flags cannot repair a file that carries them all. |
| + // Nothing is lost — the observations it would have recorded re-occur |
| + // within seconds, and the hourly pass does the repair with the full flag |
| + // set actually in hand. |
| // |
| // Residual, accepted: even the full pass only re-observes flags for apps |
| // with a currently-resolved HELD head, so an app whose channel pointer is |
| @@ -487,7 +496,7 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551 |
| return nil |
| } |
| -// adoptable filters the routes a Pass moves: lifecycle-managed instances of |
| +// Adoptable filters the routes a Pass moves: lifecycle-managed instances of |
| // a known app that are ready (an app-less or mid-provision route has nothing |
| // to adopt or must not be disturbed), excluding platform-minted routes. |
| // |
| @@ -499,20 +508,27 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551 |
| // the console then renders as though it were a real instance. The canonical |
| // route it aliases is adopted normally — that IS the deploy, and the alias |
| // serves the new version the moment it lands. |
| -func adoptable(rt registry.Route) bool { |
| +// |
| +// Exported for host.fastMembers (the fast-converge sweep's membership scan), |
| +// the same reason CapabilityRoute below is: the sweep must count as |
| +// "outstanding work" exactly the routes a pass would consider, or a route |
| +// the pass refuses on shape would be swept forever. A second copy of this |
| +// predicate in another package is exactly the drift that would let the two |
| +// disagree. |
| +func Adoptable(rt registry.Route) bool { |
| return rt.Kind == "instance" && rt.App != "" && !rt.Platform && |
| (rt.Status == "" || rt.Status == "ready") |
| } |
| // CapabilityRoute reports whether this route may materialize capabilities: |
| // a lifecycle-managed instance of a known app, unit-backed, non-platform, |
| -// ready (spec §4e). It is adoptable() plus the unit backing, deliberately — |
| +// ready (spec §4e). It is Adoptable() plus the unit backing, deliberately — |
| // a route this pass would not adopt is not one whose unit shape this pass |
| // has any business rewriting. |
| // |
| // A platform ALIAS never materialises anything: an alias is a second name |
| // for one process, and a second carlos-app@ unit claiming :25 would break |
| -// the real one. adoptable() already excludes them and this depends on that. |
| +// the real one. Adoptable() already excludes them and this depends on that. |
| // |
| // An EXEC-backed route (Backing == "") whose release declares capabilities |
| // materialises nothing here and is told so once per pass by RunAdopt's |
| @@ -525,7 +541,7 @@ func adoptable(rt registry.Route) bool { |
| // predicate in another package is exactly the drift that would let one pass |
| // materialize a shape the other refuses. |
| func CapabilityRoute(rt registry.Route) bool { |
| - return adoptable(rt) && rt.Backing == registry.BackingUnit |
| + return Adoptable(rt) && rt.Backing == registry.BackingUnit |
| } |
| // routeChannel is the channel a route follows; "" means stable. |
diff --git a/internal/release/adopt_test.go b/internal/release/adopt_test.go| index 9795b4a35..032852440 100644 |
| --- a/internal/release/adopt_test.go |
| +++ b/internal/release/adopt_test.go |
| @@ -1589,7 +1589,7 @@ func TestExecBackedRouteWithCapabilitiesMaterializesNothing(t *testing.T) { |
| // TestPlatformAliasRouteMaterializesNothing: a platform alias is a second |
| // name for one process. A second carlos-app@ unit claiming :25 would break |
| // the real one, so an alias materializes nothing — and, being outside |
| -// adoptable(), is not adopted at all. |
| +// Adoptable(), is not adopted at all. |
| func TestPlatformAliasRouteMaterializesNothing(t *testing.T) { |
| s := NewDirStore(t.TempDir()) |
| ctx := context.Background() |