amadan
host: reconcile, sidecars, moves and steering read through the record view; index-only changes run a scoped pass (#336)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7 files changed,
+1187
−58
internal/appconfig/syncexec.go+22 −1internal/appconfig/syncexec_test.go+89 −0internal/host/agent.go+131 −55internal/host/changeindex.go+295 −0internal/host/changeindex_test.go+626 −0internal/host/movespass.go+20 −1internal/host/steering_loop.go+4 −1
diff --git a/internal/appconfig/syncexec.go b/internal/appconfig/syncexec.go| index 493627a7a..1caa1b368 100644 |
| --- a/internal/appconfig/syncexec.go |
| +++ b/internal/appconfig/syncexec.go |
| @@ -124,6 +124,9 @@ func SyncExecWithCallbacks(ctx context.Context, reg *registry.Registry, s releas |
| if !execSyncable(rt) && !execSidecarCandidate(rt) && !execAddrCandidate(rt) { |
| continue |
| } |
| + if !o.kept(rt) { |
| + continue |
| + } |
| if o.fenced != nil && o.fenced(rt.Host) { |
| continue |
| } |
| @@ -321,6 +324,11 @@ func SyncExecWithCallbacks(ctx context.Context, reg *registry.Registry, s releas |
| // `carlos restart -app X` (host/restart.go), which pairs the stop |
| // with an explicit wake; this pass only ever writes the file. |
| inScope := func(rt registry.Route) (ok, addrShape, sidecarShape bool) { |
| + // A route WithRouteFilter leaves out is out of scope before |
| + // anything is read for it: no bundle, no opt-in, no overlay. |
| + if !o.kept(rt) { |
| + return false, false, false |
| + } |
| if execSyncable(rt) { |
| return true, false, false |
| } |
| @@ -697,7 +705,7 @@ func SyncExecWithCallbacks(ctx context.Context, reg *registry.Registry, s releas |
| // the app's own supervisor has stopped reading it) because carlos has |
| // no way to know when that is. |
| for _, rt := range routes { |
| - if !execAddrCandidate(rt) { |
| + if !execAddrCandidate(rt) || !o.kept(rt) { |
| continue |
| } |
| c := checkOptedIn(release.RouteAccount(rt), rt.App, rt.Environment) |
| @@ -752,12 +760,25 @@ type SyncOption func(*syncOptions) |
| type syncOptions struct { |
| fenced func(host string) bool |
| + keep func(rt registry.Route) bool |
| dirAbsent func(host, dir string) |
| outcome func(host string, done bool) |
| hostWrites hostwrite.Config |
| hostWriteCount int |
| } |
| +// WithRouteFilter narrows one SyncExec pass to the routes keep accepts. A |
| +// route it leaves out is out of scope for the pass: nothing is read for it, |
| +// and its file is neither written nor removed (reap keys on route existence, |
| +// not scope). The agent's scoped pass uses it to converge only the apps the |
| +// change index says moved (S3 change-index design §4.5). Nil keeps every |
| +// route. |
| +func WithRouteFilter(keep func(registry.Route) bool) SyncOption { |
| + return func(o *syncOptions) { o.keep = keep } |
| +} |
| + |
| +func (o *syncOptions) kept(rt registry.Route) bool { return o.keep == nil || o.keep(rt) } |
| + |
| // WhenInstanceDirAbsent hands SyncExec a callback for every nested row whose |
| // bundle it could not write because the row's instance directory does not |
| // exist yet: the activator makes that directory at the instance's first |
diff --git a/internal/appconfig/syncexec_test.go b/internal/appconfig/syncexec_test.go| index 37440129e..ef8cbc533 100644 |
| --- a/internal/appconfig/syncexec_test.go |
| +++ b/internal/appconfig/syncexec_test.go |
| @@ -10,6 +10,7 @@ import ( |
| "reflect" |
| "slices" |
| "strings" |
| + "sync" |
| "testing" |
| "time" |
| @@ -2646,3 +2647,91 @@ func TestSyncExecAcksDeliveredOnWake(t *testing.T) { |
| t.Fatalf("ack = %+v, want delivered on wake: Hibernating, not restarted, no error", a) |
| } |
| } |
| + |
| +// keyReadStore records every key and prefix a pass reads through it. |
| +type keyReadStore struct { |
| + release.Store |
| + mu sync.Mutex |
| + read []string |
| +} |
| + |
| +func (s *keyReadStore) note(k string) { s.mu.Lock(); s.read = append(s.read, k); s.mu.Unlock() } |
| + |
| +func (s *keyReadStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + s.note(key) |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +func (s *keyReadStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + s.note(prefix) |
| + return s.Store.List(ctx, prefix) |
| +} |
| + |
| +func (s *keyReadStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| + s.note(prefix) |
| + return s.Store.ListDirs(ctx, prefix) |
| +} |
| + |
| +// TestSyncExecWithRouteFilter pins the scoped config pass the agent runs |
| +// when the change index says one app moved (S3 change-index design §4.5): |
| +// a route the filter leaves out is not in scope, so nothing is read for it, |
| +// its file is left exactly as it is (neither rewritten nor reaped), and the |
| +// routes the filter keeps converge as on a full pass. |
| +func TestSyncExecWithRouteFilter(t *testing.T) { |
| + t.Parallel() |
| + ctx, base, reg, dd, priv, _ := execKit(t) |
| + pk, err := ParsePrivateKey(priv, "test") |
| + if err != nil { |
| + t.Fatalf("ParsePrivateKey: %v", err) |
| + } |
| + for _, app := range []string{"shop", "cafe"} { |
| + if err := SetEnv(ctx, base, "bab", app, map[string]string{"APP": app}); err != nil { |
| + t.Fatalf("SetEnv %s: %v", app, err) |
| + } |
| + } |
| + if err := SetExecDelivery(ctx, base, "bab", "cafe", ExecDeliveryFile); err != nil { |
| + t.Fatalf("SetExecDelivery: %v", err) |
| + } |
| + for _, rt := range []registry.Route{ |
| + {Host: "shop.example.com", Kind: "instance", App: "shop", Account: "bab", Status: "ready", Socket: "/run/carlos/shop.sock"}, |
| + {Host: "cafe.example.com", Kind: "instance", App: "cafe", Account: "bab", Status: "ready", Socket: "/run/carlos/cafe.sock"}, |
| + // The addr shape: its opt-in is read before the resolution loop. |
| + {Host: "cafe-addr.example.com", Kind: "instance", App: "cafe", Account: "bab", Status: "ready", Addr: "127.0.0.1:8444"}, |
| + } { |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatalf("Upsert %s: %v", rt.Host, err) |
| + } |
| + } |
| + stale := filepath.Join(dd, "cafe.example.com.config.env") |
| + if err := os.WriteFile(stale, []byte("OLD=1\n"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + s := &keyReadStore{Store: base} |
| + keep := WithRouteFilter(func(rt registry.Route) bool { return rt.App != "cafe" }) |
| + var bounced []string |
| + if err := SyncExec(ctx, reg, s, pk, dd, func(h string) error { bounced = append(bounced, h); return nil }, "", nil, keep); err != nil { |
| + t.Fatalf("SyncExec: %v", err) |
| + } |
| + var shopRead bool |
| + for _, k := range s.read { |
| + if strings.HasPrefix(k, "apps/bab/cafe/") { |
| + t.Errorf("the filtered pass read %s, under an app the filter left out", k) |
| + } |
| + shopRead = shopRead || strings.HasPrefix(k, "apps/bab/shop/") |
| + } |
| + if !shopRead { |
| + t.Fatalf("the pass read nothing for shop (%v): the witness pinned nothing", s.read) |
| + } |
| + if got, err := os.ReadFile(filepath.Join(dd, "shop.example.com.config.env")); err != nil || string(got) != "APP=shop\n" { |
| + t.Fatalf("shop's file = %q, %v; want APP=shop", got, err) |
| + } |
| + if got, err := os.ReadFile(stale); err != nil || string(got) != "OLD=1\n" { |
| + t.Fatalf("cafe's file = %q, %v; a route the filter leaves out keeps its file as it is", got, err) |
| + } |
| + if _, err := os.Stat(filepath.Join(dd, "cafe-addr.example.com.config.env")); !os.IsNotExist(err) { |
| + t.Fatalf("cafe's addr route got a file from a pass that left it out: %v", err) |
| + } |
| + if !slices.Equal(bounced, []string{"shop.example.com"}) { |
| + t.Fatalf("bounced %v, want only shop.example.com", bounced) |
| + } |
| +} |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 4526673f4..1d266f687 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -788,6 +788,12 @@ 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) |
| + // The change index (S3 change-index design §4): armed before the |
| + // loops that read through it, with its backstop running from here. |
| + // Unset CARLOS_CHANGE_INDEX, or a fleet box: nothing. |
| + if v := armChangeIndex(agentCtx, os.Getenv, openDeployStore); v != nil { |
| + go recordViewBackstopLoop(agentCtx, v) |
| + } |
| go adoptLoop(reg, svc, sd, claims.Writes) |
| go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision) |
| if logSink != nil { |
| @@ -1373,6 +1379,14 @@ var configSyncMu sync.Mutex |
| // mutex lives in a shared wrapper, not at the call sites, so a future third |
| // caller cannot forget it. |
| func syncExecSerialized(ctx context.Context, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes ...hostwrite.Config) error { |
| + return syncExecSerializedFiltered(ctx, nil, reg, store, priv, dd, svc, writes...) |
| +} |
| + |
| +// syncExecSerializedFiltered is syncExecSerialized over the routes keep |
| +// accepts (nil: every route): the scoped pass's config leg (S3 change-index |
| +// design §4.5). The filter is applied where SyncExec chooses its routes |
| +// (appconfig.WithRouteFilter), so a route it leaves out costs no read. |
| +func syncExecSerializedFiltered(ctx context.Context, keep func(registry.Route) bool, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes ...hostwrite.Config) error { |
| configSyncMu.Lock() |
| defer configSyncMu.Unlock() |
| pending := map[string]string{} |
| @@ -1382,6 +1396,9 @@ func syncExecSerialized(ctx context.Context, reg *registry.Registry, store relea |
| appconfig.WhenInstanceDirAbsent(func(host, dir string) { pending[host] = dir }), |
| appconfig.WithHostOutcome(func(host string, done bool) { outcomes[host] = done }), |
| } |
| + if keep != nil { |
| + opts = append(opts, appconfig.WithRouteFilter(keep)) |
| + } |
| for _, cfg := range writes { |
| opts = append(opts, appconfig.WithHostWrite(cfg)) |
| } |
| @@ -1394,7 +1411,9 @@ func syncExecSerialized(ctx context.Context, reg *registry.Registry, store relea |
| } else { |
| err = appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc), os.Getenv("PLATFORM_APPS_DOMAIN"), configAckFunc(ctx, store, "agent"), opts...) |
| } |
| - configDirsPending.record(pending, outcomes, err == nil) |
| + // A filtered pass names only the routes it kept, so it is never |
| + // complete: a host it left out keeps its wait (pendingDirs.record). |
| + configDirsPending.record(pending, outcomes, err == nil && keep == nil) |
| return err |
| } |
| @@ -1569,6 +1588,13 @@ 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() |
| + // With the change index armed, the record walks below read the record |
| + // view's snapshot, refreshed here under reconcileMu so a pass's snapshot |
| + // is never older than the previous pass's (S3 change-index design §4.3). |
| + // Writes still go to s, and a key written through the snapshot store is |
| + // read from s afterwards. Disarmed, or when the view cannot vouch, rs is |
| + // s and the pass walks the bucket as before. |
| + rs, _, _ := viewStore(ctx, s) |
| // 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 |
| @@ -1594,7 +1620,7 @@ func reconcileSerializedWith(ctx context.Context, s release.Store, reg *registry |
| // Phase 1's hooks are set. |
| if w != nil && w.cfg.OfflineCapable && w.cfg.Offline != nil { |
| if !w.cfg.Offline.built() || w.cfg.Offline.failed() { |
| - refreshOfflineView(ctx, s, w.cfg.Offline) |
| + refreshOfflineView(ctx, rs, w.cfg.Offline) |
| } |
| cfg = withOfflineHolds(cfg, offlineHold(w.cfg), w.cfg.Offline) |
| } |
| @@ -1602,8 +1628,8 @@ func reconcileSerializedWith(ctx context.Context, s release.Store, reg *registry |
| // every in-scope record with a run, and acts after the pass. |
| cfg = withRunsHook(cfg) |
| return errors.Join( |
| - provision.Reconcile(ctx, s, reg, cfg), |
| - provision.ReconcileSidecars(ctx, s, reg, cfg), |
| + provision.Reconcile(ctx, rs, reg, cfg), |
| + provision.ReconcileSidecars(ctx, rs, reg, cfg), |
| ) |
| } |
| @@ -2102,6 +2128,17 @@ type fastDeps struct { |
| restartHold map[string]*restartFailure // hosts held out of the sweep after a failed restart; see noteRestartFailed |
| serialBroken bool // last ReadSerial failed and was logged; reset on success |
| serialNotify chan<- struct{} |
| + |
| + // The change index (CARLOS_CHANGE_INDEX=read, S3 change-index design |
| + // §4.5); watch nil leaves the tick exactly as before. The watch is shared |
| + // with the record view, so Poll's moved list is advisory: the tick diffs |
| + // Observed against indexActed, the generations it last acted on, and |
| + // moves that map on only once a pass has covered them. |
| + watch *release.ChangeWatch |
| + scopedPass func(ctx context.Context, moved []string) // passLegs.scoped, or a test hook |
| + fullDue *atomic.Bool // the backstop's request for a full pass; nil = none |
| + indexActed map[string]uint64 // nil until the first good observation seeds it |
| + indexBroken bool // last Poll failed and was logged; reset on success |
| } |
| // restartFailure is one host's cross-pass memo of a version its restart |
| @@ -2534,6 +2571,10 @@ func logSlow(what string, start time.Time, detail string) { |
| // proof wait the sweep left standing; other staged members clear their |
| // backoff state. |
| func fastTick(ctx context.Context, d *fastDeps, now time.Time) { |
| + // The index is observed BEFORE any pass this tick runs, so a pass covers |
| + // at least what was observed and indexActed can take it afterwards. |
| + obs, indexOK := d.observeIndex(ctx) |
| + fullRan := false |
| doc, _, err := release.ReadSerial(ctx, d.store) |
| switch { |
| case err != nil: |
| @@ -2555,7 +2596,8 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) { |
| default: |
| d.serialBroken = false |
| if doc.Serial != d.lastSerial && now.Sub(d.lastFull) >= fastFullClamp { |
| - d.fullPass(ctx) |
| + d.runFull(ctx) |
| + fullRan = true |
| if d.restartLeg != nil { |
| d.restartLeg(ctx) |
| } |
| @@ -2601,6 +2643,16 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) { |
| } |
| } |
| + // The backstop's full pass (spec §4.4), on this goroutine: requested |
| + // after each rebuild, run once, under the same clamp. |
| + if !fullRan && d.fullDue != nil && d.fullDue.Load() && now.Sub(d.lastFull) >= fastFullClamp { |
| + d.runFull(ctx) |
| + fullRan = true |
| + d.lastFull = now |
| + } |
| + // An index-only change runs the scoped pass (spec §4.5). |
| + d.indexTick(ctx, now, obs, indexOK, fullRan) |
| + |
| // The self-update tick, outside the serial switch and before the sweep's |
| // early return: it is what runs a spread deadline down and re-touches an |
| // unanswered request, and neither has anything to do with routes. It |
| @@ -2655,6 +2707,67 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) { |
| } |
| } |
| +// runFull runs the full pass set's fullPass, taking any backstop request |
| +// first: a full pass that starts after the backstop's rebuild satisfies it. |
| +func (d *fastDeps) runFull(ctx context.Context) { |
| + if d.fullDue != nil { |
| + d.fullDue.Store(false) |
| + } |
| + d.fullPass(ctx) |
| +} |
| + |
| +// observeIndex polls the change index and returns its observed generations; |
| +// ok is false when no watch is armed or the index is unusable (then the |
| +// serial and the floors carry the tick, as before the index). A failure is |
| +// logged once per state change, never per tick. |
| +func (d *fastDeps) observeIndex(ctx context.Context) (map[string]uint64, bool) { |
| + if d.watch == nil { |
| + return nil, false |
| + } |
| + // Poll's moved list is advisory (shared watch): only Observed counts. |
| + if _, err := d.watch.Poll(ctx); err != nil { |
| + if !d.indexBroken { |
| + log.Printf("agent: fast converge change index: %v (scoped passes wait; the serial and the floors carry on)", err) |
| + d.indexBroken = true |
| + } |
| + return nil, false |
| + } |
| + obs, ok := d.watch.Observed() |
| + if ok && d.indexBroken { |
| + log.Printf("agent: fast converge change index readable again") |
| + d.indexBroken = false |
| + } |
| + return obs, ok |
| +} |
| + |
| +// indexTick acts on an index observation: the first one seeds indexActed |
| +// without a pass (boot must not replay history; the backstop's first full |
| +// pass and the loops' first passes cover boot), a full pass this tick |
| +// covers everything observed, and otherwise app and account scopes that |
| +// moved since indexActed run one scoped pass, clamped like the full pass: |
| +// a change inside the clamp leaves indexActed as it was, so the next tick |
| +// past the window acts on it. indexActed moves on only after a pass ran. |
| +func (d *fastDeps) indexTick(ctx context.Context, now time.Time, obs map[string]uint64, ok, fullRan bool) { |
| + if !ok { |
| + return |
| + } |
| + if d.indexActed == nil || fullRan { |
| + d.indexActed = obs |
| + return |
| + } |
| + moved := indexScopes(d.indexActed, obs) |
| + if len(moved) == 0 { |
| + d.indexActed = obs // control/ scopes only: not this loop's business |
| + return |
| + } |
| + if now.Sub(d.lastFull) < fastFullClamp || d.scopedPass == nil { |
| + return |
| + } |
| + d.scopedPass(ctx, moved) |
| + d.indexActed = obs |
| + d.lastFull = now |
| +} |
| + |
| // fastConvergeLoop replaces the canary-only fast adopt loop with the |
| // carlos-deploy fast set (spec §3-§4): a 2s serial poll that turns a console |
| // bump (promote, instance create, config write) into the ordinary verified |
| @@ -2784,56 +2897,19 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| // a serial bump is an explicit console action, bounded by whoever keeps |
| // making it — and it is what lets a recovery promote through the console |
| // converge immediately even mid-hold. |
| - d.fullPass = func(ctx context.Context) { |
| - // Each stage's time, logged together when the set is slow, so the |
| - // journal says WHICH stage held a promote back. |
| - var took []string |
| - t := time.Now() |
| - mark := func(name string) { |
| - took = append(took, fmt.Sprintf("%s %s", name, time.Since(t).Round(100*time.Millisecond))) |
| - t = time.Now() |
| - } |
| - 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 { |
| - log.Printf("agent: fast converge reconcile: %v", err) |
| - } |
| - mark("reconcile") |
| - // The registry's build-state copies, refreshed at every reconcile |
| - // (settings spec §4.2), before adopt reads them; refreshAll keeps |
| - // to its own budget, and the stage is timed apart from reconcile. |
| - if bs != nil { |
| - bs.refreshAll(ctx) |
| - mark("build states") |
| - } |
| - telemetryPass(ctx, store, reg) |
| - mark("telemetry") |
| - runMovesPass(ctx, store) |
| - mark("moves") |
| - runRunsPass(ctx) |
| - mark("runs") |
| - // Folded once per pass and handed in, never re-read inside the |
| - // loop: this leg runs on every serial bump (a seconds cadence), |
| - // and the fold is a LoadBoxes plus a List plus one Get per edges |
| - // file. On a fleet of under a dozen boxes that is a handful of |
| - // Gets per bump, which is accepted here rather than cached — a |
| - // cache would need its own staleness rule, and a STALE route set |
| - // is the one thing a security guard must not have. |
| - deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now()) |
| - if err := domains.ReconcileAliases(ctx, store, reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil { |
| - log.Printf("agent: fast converge alias pass: %v", err) |
| - } |
| - mark("aliases") |
| - if err := release.Pass(ctx, reg, store, pub, optDir, restart, opts...); err != nil { |
| - log.Printf("agent: fast converge adopt pass: %v", err) |
| - } |
| - mark("adopt") |
| - if priv != nil { |
| - if err := syncExecSerialized(ctx, reg, store, priv, dd, svc, writes); err != nil { |
| - log.Printf("agent: fast converge config sync: %v", err) |
| - } |
| - mark("config") |
| - } |
| + legs := &passLegs{store: store, reg: reg, pub: pub, optDir: optDir, restart: restart, opts: opts, |
| + priv: priv, dd: dd, svc: svc, writes: writes, bs: bs, |
| + cfg: func() provision.ReconcileConfig { |
| + return configuredClaimProvision(provisionTickConfig(os.Getenv), configure) |
| + }} |
| + d.fullPass = legs.full |
| + // The change index (S3 change-index design §4.5): an index-only change |
| + // runs the scoped pass, and the backstop's request runs the full one, |
| + // both on this loop's goroutine. Nil watch: the tick is as before. |
| + if watch := armedChangeWatch.Load(); watch != nil { |
| + d.watch = watch |
| + d.scopedPass = legs.scoped |
| + d.fullDue = &backstopFullDue |
| } |
| // The restart leg: this process's own children bounced here, everything |
| // unit-shaped relayed to a root pass. Deliberately NOT inside fullPass |
diff --git a/internal/host/changeindex.go b/internal/host/changeindex.go| new file mode 100644 |
| index 000000000..7e400f84a |
| --- /dev/null |
| +++ b/internal/host/changeindex.go |
| @@ -0,0 +1,295 @@ |
| +package host |
| + |
| +// The box agent's reads through the S3 change index (S3 change-index design |
| +// §4.3-§4.5, discussion #336). Armed only by CARLOS_CHANGE_INDEX=read; unset, |
| +// every pointer below stays nil and every loop reads the bucket exactly as |
| +// before. |
| + |
| +import ( |
| + "context" |
| + "crypto/ecdh" |
| + "crypto/ed25519" |
| + "fmt" |
| + "log" |
| + "math/rand/v2" |
| + "sort" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/domains" |
| + "github.com/carlosframework/platform/internal/edgesfile" |
| + "github.com/carlosframework/platform/internal/hostwrite" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// changeIndexArmed reports whether this process reads through the change |
| +// index. Only the exact value "read" arms it (spec §7). |
| +func changeIndexArmed(getenv func(string) string) bool { |
| + return getenv("CARLOS_CHANGE_INDEX") == "read" |
| +} |
| + |
| +// armedChangeWatch is the process's one watch on the change index, and |
| +// armedRecordView the record view that follows it. RunAgent sets both once |
| +// at start-up when armed; nil means today's behaviour everywhere. |
| +var ( |
| + armedChangeWatch atomic.Pointer[release.ChangeWatch] |
| + armedRecordView atomic.Pointer[provision.RecordView] |
| +) |
| + |
| +// backstopFullDue is the backstop's request for the unrestricted full pass: |
| +// set after each Rebuild, taken by fastTick on the fast converge loop's own |
| +// goroutine (fastDeps' maps are single-goroutine by construction). |
| +var backstopFullDue atomic.Bool |
| + |
| +// armChangeIndex arms the change index for this process: a watch on the |
| +// deployment store, and a record view over the same store. A fleet box |
| +// stays disarmed with the variable set: its store door has no conditional |
| +// read, so its watch would be unusable and every Refresh would walk anyway |
| +// (spec §3.4). The view's store must take no agent lock on any read path |
| +// (RecordView.mu is a leaf, spec §4.3): open hands back the plain deployment |
| +// store the other loops use. |
| +func armChangeIndex(ctx context.Context, getenv func(string) string, open func(context.Context) (release.Store, error)) *provision.RecordView { |
| + if !changeIndexArmed(getenv) { |
| + return nil |
| + } |
| + if shouldSkipPublisherElection(getenv) { |
| + log.Printf("agent: change index: CARLOS_CHANGE_INDEX=read is ignored on a fleet box; record reads stay on the walk") |
| + return nil |
| + } |
| + store, err := open(ctx) |
| + if err != nil { |
| + log.Printf("agent: change index: not armed: %v", err) |
| + return nil |
| + } |
| + watch := release.NewChangeWatch(store) |
| + v := provision.NewRecordView(store, watch) |
| + armedChangeWatch.Store(watch) |
| + armedRecordView.Store(v) |
| + log.Printf("agent: change index armed: record walks read the record view (backstop %s)", release.ChangeIndexBackstop) |
| + return v |
| +} |
| + |
| +// viewUse remembers whether the last viewStore call used the view, so the |
| +// switch is logged once per change of state, never per tick. |
| +var viewUse struct { |
| + mu sync.Mutex |
| + last int // 0 none yet, 1 using the view, 2 walking |
| +} |
| + |
| +func noteViewUse(ok bool, why string) { |
| + state := 2 |
| + if ok { |
| + state = 1 |
| + } |
| + viewUse.mu.Lock() |
| + changed := viewUse.last != state |
| + viewUse.last = state |
| + viewUse.mu.Unlock() |
| + if !changed { |
| + return |
| + } |
| + if ok { |
| + log.Printf("agent: record view: record walks read the view") |
| + return |
| + } |
| + log.Printf("agent: record view: %s; record walks read the bucket until it can vouch again", why) |
| +} |
| + |
| +// viewStore is the store a record-family walker reads: the armed view's |
| +// current snapshot over live, refreshed first, when the refresh returned nil |
| +// and the snapshot is Complete. Otherwise ok is false and s is live, so the |
| +// caller walks the bucket as before the index (spec §3.4). |
| +// |
| +// Only record-family walkers take it (Reconcile, ReconcileSidecars, the |
| +// offline keep view, the moves pass's listing, steering). Its ListDirs of |
| +// apps/<a>/ names only apps with record-family state, so app enumerators |
| +// (hostops.Apps, purge and activation sweeps, mail, economics) must never |
| +// get it. |
| +// |
| +// live must take no agent lock on any read path: RecordView.mu is a leaf |
| +// (lock order w.pass -> reconcileMu -> RecordView.mu). |
| +func viewStore(ctx context.Context, live release.Store) (s release.Store, epoch uint64, ok bool) { |
| + v := armedRecordView.Load() |
| + if v == nil || live == nil { |
| + return live, 0, false |
| + } |
| + err := v.Refresh(ctx) |
| + sn := v.Snapshot() |
| + switch { |
| + case err != nil: |
| + noteViewUse(false, fmt.Sprintf("refresh failed (%v)", err)) |
| + return live, 0, false |
| + case sn == nil || !sn.Complete: |
| + noteViewUse(false, "the snapshot is incomplete") |
| + return live, 0, false |
| + } |
| + noteViewUse(true, "") |
| + return sn.Store(live), sn.Epoch, true |
| +} |
| + |
| +// recordViewBackstopLoop is the backstop (spec §4.4): Rebuild at start and |
| +// every release.ChangeIndexBackstop, jittered by +-10%, each followed by the |
| +// unrestricted full pass, so every leg is re-run over everything at least |
| +// that often. The full pass runs on the fast converge loop's goroutine |
| +// (backstopFullDue); on a box where that loop is not armed only the |
| +// rebuild runs. |
| +func recordViewBackstopLoop(ctx context.Context, v *provision.RecordView) { |
| + for { |
| + recordViewBackstop(ctx, v, &backstopFullDue) |
| + select { |
| + case <-ctx.Done(): |
| + return |
| + case <-time.After(backstopEvery()): |
| + } |
| + } |
| +} |
| + |
| +// recordViewBackstop is one backstop round: the rebuild, then the request |
| +// for a full pass, whether or not the rebuild succeeded (a failed one |
| +// leaves the view unable to vouch, so that pass walks). |
| +func recordViewBackstop(ctx context.Context, v *provision.RecordView, due *atomic.Bool) { |
| + if err := v.Rebuild(ctx); err != nil { |
| + log.Printf("agent: record view backstop: %v", err) |
| + } |
| + due.Store(true) |
| +} |
| + |
| +// backstopEvery is release.ChangeIndexBackstop jittered by +-10%. |
| +func backstopEvery() time.Duration { |
| + b := release.ChangeIndexBackstop |
| + return b - b/10 + rand.N(b/5) |
| +} |
| + |
| +// indexScopes is the scopes of obs whose generation differs from acted's, |
| +// sorted, kept to the scopes a pass acts on (apps/<a>/<b> and accounts/<a>). |
| +// The control/ scopes belong to other readers. |
| +func indexScopes(acted, obs map[string]uint64) []string { |
| + var moved []string |
| + for scope, gen := range obs { |
| + if have, ok := acted[scope]; ok && have == gen { |
| + continue |
| + } |
| + if _, _, ok := provision.ScopeApp(scope); ok { |
| + moved = append(moved, scope) |
| + continue |
| + } |
| + if a, ok := strings.CutPrefix(scope, "accounts/"); ok && a != "" && !strings.Contains(a, "/") { |
| + moved = append(moved, scope) |
| + } |
| + } |
| + sort.Strings(moved) |
| + return moved |
| +} |
| + |
| +// movedRoutes keeps the routes whose (account, app) has an app scope in |
| +// moved. Account scopes keep nothing here: their effect on a box (an |
| +// account soft-deleted) is Reconcile's, which runs whole. |
| +func movedRoutes(moved []string) func(registry.Route) bool { |
| + apps := map[[2]string]bool{} |
| + for _, scope := range moved { |
| + if a, b, ok := provision.ScopeApp(scope); ok { |
| + apps[[2]string{a, b}] = true |
| + } |
| + } |
| + return func(rt registry.Route) bool { return apps[[2]string{release.RouteAccount(rt), rt.App}] } |
| +} |
| + |
| +// passLegs is the serial-triggered pass set's inputs (fastConvergeLoop), |
| +// shared by the full pass and the scoped one so the two run the same legs |
| +// in the same order and cannot drift. |
| +type passLegs struct { |
| + store release.Store |
| + reg *registry.Registry |
| + pub ed25519.PublicKey |
| + optDir string |
| + restart func(string) error |
| + opts []release.PassOption |
| + priv *ecdh.PrivateKey // nil: no config leg (no CARLOS_SECRETS_KEY) |
| + dd string |
| + svc *activator.Service |
| + writes hostwrite.Config |
| + bs *buildStates |
| + cfg func() provision.ReconcileConfig // read per pass, as provisionTickConfig always was |
| +} |
| + |
| +// passAfterReconcileHook, when set, runs in every pass right after its |
| +// reconcile leg. Tests only. |
| +var passAfterReconcileHook func() |
| + |
| +// full is the serial-triggered pass set, in spec order: mint routes |
| +// (reconcile), then reconcile aliases, then stage them (adopt), then deliver |
| +// their config, over every hosted route. |
| +func (l *passLegs) full(ctx context.Context) { l.run(ctx, "serial pass stages", nil) } |
| + |
| +// scoped is the pass an index-only change runs (spec §4.5, codex 10): the |
| +// same legs, with adopt and config sync restricted to the hosted routes |
| +// whose app scope moved, so changing one app re-reads that app only. |
| +// Reconcile and the moves pass run whole, over the record view. |
| +func (l *passLegs) scoped(ctx context.Context, moved []string) { |
| + l.run(ctx, "scoped pass stages", movedRoutes(moved)) |
| +} |
| + |
| +// run is one pass set; keep nil is the unrestricted full pass. Each error is |
| +// one log line, never fatal: the hourly floors stand underneath every leg. |
| +func (l *passLegs) run(ctx context.Context, what string, keep func(registry.Route) bool) { |
| + // Each stage's time, logged together when the set is slow, so the |
| + // journal says WHICH stage held a promote back. |
| + var took []string |
| + t := time.Now() |
| + mark := func(name string) { |
| + took = append(took, fmt.Sprintf("%s %s", name, time.Since(t).Round(100*time.Millisecond))) |
| + t = time.Now() |
| + } |
| + defer func(start time.Time) { logSlow(what, start, strings.Join(took, ", ")) }(time.Now()) |
| + |
| + if err := reconcileSerialized(ctx, l.store, l.reg, l.cfg()); err != nil { |
| + log.Printf("agent: fast converge reconcile: %v", err) |
| + } |
| + mark("reconcile") |
| + if hook := passAfterReconcileHook; hook != nil { |
| + hook() |
| + } |
| + // The registry's build-state copies, refreshed at every reconcile |
| + // (settings spec §4.2), before adopt reads them; refreshAll keeps |
| + // to its own budget, and the stage is timed apart from reconcile. |
| + if l.bs != nil { |
| + l.bs.refreshAll(ctx) |
| + mark("build states") |
| + } |
| + telemetryPass(ctx, l.store, l.reg) |
| + mark("telemetry") |
| + // The moves pass takes its own snapshot inside w.pass (spec §4.3): it |
| + // is handed the plain store, never this pass's reconcile snapshot. |
| + runMovesPass(ctx, l.store) |
| + mark("moves") |
| + runRunsPass(ctx) |
| + mark("runs") |
| + // Folded once per pass and handed in, never re-read inside the |
| + // loop: this leg runs on every serial bump (a seconds cadence), |
| + // and the fold is a LoadBoxes plus a List plus one Get per edges |
| + // file. On a fleet of under a dozen boxes that is a handful of |
| + // Gets per bump, which is accepted here rather than cached — a |
| + // cache would need its own staleness rule, and a STALE route set |
| + // is the one thing a security guard must not have. Aliases read the |
| + // domain claims, not the record families, so they read the bucket. |
| + deploymentRoutes := edgesfile.DeploymentRoutes(ctx, l.store, time.Now()) |
| + if err := domains.ReconcileAliases(ctx, l.store, l.reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil { |
| + log.Printf("agent: fast converge alias pass: %v", err) |
| + } |
| + mark("aliases") |
| + if err := release.PassFilter(ctx, l.reg, l.store, l.pub, l.optDir, l.restart, keep, l.opts...); err != nil { |
| + log.Printf("agent: fast converge adopt pass: %v", err) |
| + } |
| + mark("adopt") |
| + if l.priv != nil { |
| + if err := syncExecSerializedFiltered(ctx, keep, l.reg, l.store, l.priv, l.dd, l.svc, l.writes); err != nil { |
| + log.Printf("agent: fast converge config sync: %v", err) |
| + } |
| + mark("config") |
| + } |
| +} |
diff --git a/internal/host/changeindex_test.go b/internal/host/changeindex_test.go| new file mode 100644 |
| index 000000000..a37978144 |
| --- /dev/null |
| +++ b/internal/host/changeindex_test.go |
| @@ -0,0 +1,626 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "os" |
| + "slices" |
| + "sort" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/appconfig" |
| + "github.com/carlosframework/platform/internal/domains" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// ciCountingStore is a DirStore with the change index on that records every |
| +// read by method and key. Embedding the concrete store forwards every |
| +// optional extension (ConditionalGetter and ETagLister included), so the |
| +// view and the watch see the store production gives them. |
| +type ciCountingStore struct { |
| + *release.DirStore |
| + mu sync.Mutex |
| + calls []string |
| + failList string // List and ListETags of exactly this prefix fail |
| +} |
| + |
| +func (s *ciCountingStore) setFailList(p string) { s.mu.Lock(); s.failList = p; s.mu.Unlock() } |
| + |
| +func (s *ciCountingStore) listFails(prefix string) bool { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return s.failList != "" && s.failList == prefix |
| +} |
| + |
| +func newCICountingStore(t *testing.T) *ciCountingStore { |
| + t.Helper() |
| + d := release.NewDirStore(t.TempDir()) |
| + d.EnableChangeIndex() |
| + return &ciCountingStore{DirStore: d} |
| +} |
| + |
| +func (s *ciCountingStore) note(method, key string) { |
| + s.mu.Lock() |
| + s.calls = append(s.calls, method+" "+key) |
| + s.mu.Unlock() |
| +} |
| + |
| +func (s *ciCountingStore) reset() { s.mu.Lock(); s.calls = nil; s.mu.Unlock() } |
| + |
| +func (s *ciCountingStore) got() []string { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + out := slices.Clone(s.calls) |
| + sort.Strings(out) |
| + return out |
| +} |
| + |
| +// count is how many recorded calls start with method+" " and name a key |
| +// under prefix. |
| +func (s *ciCountingStore) count(method, prefix string) int { |
| + n := 0 |
| + for _, c := range s.got() { |
| + if strings.HasPrefix(c, method+" "+prefix) { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +func (s *ciCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + s.note("Get", key) |
| + return s.DirStore.Get(ctx, key) |
| +} |
| + |
| +func (s *ciCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) { |
| + s.note("GetIfChanged", key) |
| + return s.DirStore.GetIfChanged(ctx, key, etag) |
| +} |
| + |
| +func (s *ciCountingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + s.note("List", prefix) |
| + if s.listFails(prefix) { |
| + return nil, errors.New("list refused") |
| + } |
| + return s.DirStore.List(ctx, prefix) |
| +} |
| + |
| +func (s *ciCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) { |
| + s.note("ListETags", prefix) |
| + if s.listFails(prefix) { |
| + return nil, errors.New("list refused") |
| + } |
| + return s.DirStore.ListETags(ctx, prefix) |
| +} |
| + |
| +func (s *ciCountingStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| + s.note("ListDirs", prefix) |
| + return s.DirStore.ListDirs(ctx, prefix) |
| +} |
| + |
| +func (s *ciCountingStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) { |
| + s.note("ListMod", prefix) |
| + return s.DirStore.ListMod(ctx, prefix) |
| +} |
| + |
| +// ciWorld is a deployment of one account with two apps, shop and cafe, each |
| +// with one instance record and a config bundle, written through the indexed |
| +// store. |
| +type ciWorld struct { |
| + live *ciCountingStore |
| + acct string |
| + now time.Time |
| +} |
| + |
| +func newCIWorld(t *testing.T) *ciWorld { |
| + t.Helper() |
| + t.Setenv("CARLOS_HOST_CLAIM", "") // the appconfig and provision guards read the process env |
| + ctx := context.Background() |
| + w := &ciWorld{live: newCICountingStore(t), acct: domains.Sqid(2), now: time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)} |
| + manifest, err := json.Marshal(map[string]any{"id": 2, "sqid": w.acct, "name": "fixture", "created_at": w.now}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := w.live.Put(ctx, "accounts/"+w.acct+"/account.json", manifest, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, app := range []string{"shop", "cafe"} { |
| + if err := provision.PutManifest(ctx, w.live, w.acct, app, provision.Manifest{Schema: 1, App: app, Domain: "example.com"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + w.record(t, app, app+".example.com") |
| + if err := appconfig.SetEnv(ctx, w.live, w.acct, app, map[string]string{"APP": app}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + w.live.reset() |
| + return w |
| +} |
| + |
| +func (w *ciWorld) record(t *testing.T, app, host string) { |
| + t.Helper() |
| + rec := provision.Record{Schema: 1, Host: host, Slug: strings.SplitN(host, ".", 2)[0], Account: w.acct, App: app, Channel: "stable", CreatedAt: w.now} |
| + if err := provision.CreateRecord(context.Background(), w.live, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func (w *ciWorld) scope(app string) string { return "apps/" + w.acct + "/" + app } |
| + |
| +// armView arms the change index over live for one test, as RunAgent does, |
| +// and disarms it afterwards. Tests that call it stay serial. |
| +func armView(t *testing.T, live release.Store) *provision.RecordView { |
| + t.Helper() |
| + watch := release.NewChangeWatch(live) |
| + v := provision.NewRecordView(live, watch) |
| + armedChangeWatch.Store(watch) |
| + armedRecordView.Store(v) |
| + t.Cleanup(func() { |
| + armedChangeWatch.Store(nil) |
| + armedRecordView.Store(nil) |
| + }) |
| + return v |
| +} |
| + |
| +// disarmView makes sure no earlier test left the view armed. |
| +func disarmView(t *testing.T) { |
| + t.Helper() |
| + armedChangeWatch.Store(nil) |
| + armedRecordView.Store(nil) |
| +} |
| + |
| +func ciReconcileConfig(t *testing.T) provision.ReconcileConfig { |
| + return provision.ReconcileConfig{DataDir: t.TempDir(), Layout: flatLayout} |
| +} |
| + |
| +// recordFamilyLists counts listings under the record families. |
| +func recordFamilyLists(s *ciCountingStore) int { |
| + n := 0 |
| + for _, m := range []string{"List", "ListETags", "ListDirs"} { |
| + n += s.count(m, "apps/") + s.count(m, "accounts/") |
| + } |
| + return n |
| +} |
| + |
| +func TestChangeIndexArmedOnlyForRead(t *testing.T) { |
| + t.Parallel() |
| + for v, want := range map[string]bool{"": false, "read": true, "READ": false, "write": false, "1": false, " read": false} { |
| + if got := changeIndexArmed(func(k string) string { |
| + if k == "CARLOS_CHANGE_INDEX" { |
| + return v |
| + } |
| + return "" |
| + }); got != want { |
| + t.Errorf("CARLOS_CHANGE_INDEX=%q armed = %v, want %v", v, got, want) |
| + } |
| + } |
| +} |
| + |
| +// A fleet box stays disarmed with the variable set: its store door has no |
| +// conditional read, so every Refresh would walk anyway (spec §3.4). |
| +func TestChangeIndexStaysDisarmedOnFleetBox(t *testing.T) { |
| + disarmView(t) |
| + t.Cleanup(func() { disarmView(t) }) |
| + opened := 0 |
| + open := func(context.Context) (release.Store, error) { opened++; return newCICountingStore(t), nil } |
| + env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } } |
| + |
| + if v := armChangeIndex(context.Background(), env(map[string]string{"CARLOS_CHANGE_INDEX": "read", "CARLOS_FLEET": "pi"}), open); v != nil || armedRecordView.Load() != nil || armedChangeWatch.Load() != nil { |
| + t.Fatal("a fleet box armed the change index") |
| + } |
| + if v := armChangeIndex(context.Background(), env(nil), open); v != nil || armedRecordView.Load() != nil { |
| + t.Fatal("an unset CARLOS_CHANGE_INDEX armed the change index") |
| + } |
| + if opened != 0 { |
| + t.Fatalf("a disarmed box opened the deployment store %d time(s)", opened) |
| + } |
| + v := armChangeIndex(context.Background(), env(map[string]string{"CARLOS_CHANGE_INDEX": "read"}), open) |
| + if v == nil || armedRecordView.Load() != v || armedChangeWatch.Load() == nil { |
| + t.Fatal("a platform box with CARLOS_CHANGE_INDEX=read did not arm") |
| + } |
| +} |
| + |
| +// Armed, the record walks of a reconcile pass (the offline keep view's |
| +// listing, Reconcile and ReconcileSidecars) read the view: once it is built, |
| +// two passes with no writes list nothing on the bucket. |
| +func TestReconcileUsesViewWhenArmed(t *testing.T) { |
| + w := newCIWorld(t) |
| + v := armView(t, w.live) |
| + if err := v.Rebuild(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := restartTestRegistry(t) |
| + // A capable worker whose keep view has read nothing yet, so the pass |
| + // lists records for it too (refreshOfflineView). |
| + mw := newMovesWorker(moveCfg{OfflineCapable: true}) |
| + mw.cfg.Offline = newOfflineView(mw.cfg, nil) |
| + cfg := ciReconcileConfig(t) |
| + w.live.reset() |
| + for range 2 { |
| + if err := reconcileSerializedWith(context.Background(), w.live, reg, cfg, mw); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if n := w.live.count("List", ""); n != 0 { |
| + t.Fatalf("two armed passes with no writes listed the bucket %d time(s): %v", n, w.live.got()) |
| + } |
| + if n := recordFamilyLists(w.live); n != 0 { |
| + t.Fatalf("two armed passes listed the record families %d time(s): %v", n, w.live.got()) |
| + } |
| + if !mw.cfg.Offline.built() || mw.cfg.Offline.failed() { |
| + t.Fatal("the offline keep view was not read from the view") |
| + } |
| + for _, h := range []string{"shop.example.com", "cafe.example.com"} { |
| + if rt, err := reg.Get(h); err != nil || rt == nil { |
| + t.Fatalf("no route for %s from the view's records: %v", h, err) |
| + } |
| + } |
| +} |
| + |
| +// Disarmed, the same passes walk the bucket exactly as before the index. |
| +func TestReconcileWalksWhenDisarmed(t *testing.T) { |
| + disarmView(t) |
| + w := newCIWorld(t) |
| + reg := restartTestRegistry(t) |
| + cfg := ciReconcileConfig(t) |
| + for range 2 { |
| + if err := reconcileSerializedWith(context.Background(), w.live, reg, cfg, nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if n := w.live.count("List", "accounts/"); n < 2 { |
| + t.Fatalf("two disarmed passes listed accounts/ %d time(s), want one walk each: %v", n, w.live.got()) |
| + } |
| + if n := w.live.count("GetIfChanged", ""); n != 0 { |
| + t.Fatalf("a disarmed pass read the change index: %v", w.live.got()) |
| + } |
| +} |
| + |
| +// instancesLoop's floor is reconcileSerialized over its own store: armed, |
| +// with nothing changed, it costs one conditional GET of the index and no |
| +// listing. |
| +func TestInstancesLoopFloorIsRefreshWhenArmed(t *testing.T) { |
| + w := newCIWorld(t) |
| + v := armView(t, w.live) |
| + if err := v.Rebuild(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := restartTestRegistry(t) |
| + cfg := ciReconcileConfig(t) |
| + if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + w.live.reset() |
| + if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := w.live.count("GetIfChanged", release.ChangeIndexKey); n != 1 { |
| + t.Fatalf("a quiet floor read the index %d time(s), want 1: %v", n, w.live.got()) |
| + } |
| + if n := w.live.count("List", "") + w.live.count("ListETags", "") + w.live.count("ListDirs", ""); n != 0 { |
| + t.Fatalf("a quiet floor listed the bucket: %v", w.live.got()) |
| + } |
| +} |
| + |
| +// A view that cannot vouch (its Refresh failed, so there is no Complete |
| +// snapshot) is never used: the caller walks live, as before the index. |
| +func TestViewStoreFallsBackWhenTheViewCannotVouch(t *testing.T) { |
| + disarmView(t) |
| + w := newCIWorld(t) |
| + if s, _, ok := viewStore(context.Background(), w.live); ok || s != release.Store(w.live) { |
| + t.Fatal("a disarmed viewStore did not hand back live") |
| + } |
| + armView(t, w.live) |
| + w.live.setFailList("accounts/") |
| + if s, _, ok := viewStore(context.Background(), w.live); ok || s != release.Store(w.live) { |
| + t.Fatal("viewStore used the view after a failed Refresh") |
| + } |
| + w.live.setFailList("") |
| + s, epoch, ok := viewStore(context.Background(), w.live) |
| + if !ok || s == release.Store(w.live) || epoch == 0 { |
| + t.Fatalf("viewStore after a good Refresh: ok %v epoch %d", ok, epoch) |
| + } |
| +} |
| + |
| +// ---- moves ----------------------------------------------------------------- |
| + |
| +// The moves pass takes its own snapshot inside w.pass, never the one the |
| +// reconcile leg of its pass set read (spec §4.3, codex 1): pass A is paused |
| +// after its reconcile; pass B runs whole; a write lands; A's moves leg then |
| +// consumes the newest epoch, never A's own older one. |
| +func TestMovesPassTakesSnapshotUnderPass(t *testing.T) { |
| + w := newCIWorld(t) |
| + v := armView(t, w.live) |
| + mw := newMovesWorker(moveCfg{}) |
| + armedMovesWorker.Store(mw) |
| + t.Cleanup(func() { armedMovesWorker.Store(nil) }) |
| + reg := restartTestRegistry(t) |
| + rcfg := ciReconcileConfig(t) |
| + legs := &passLegs{store: w.live, reg: reg, optDir: t.TempDir(), dd: t.TempDir(), |
| + cfg: func() provision.ReconcileConfig { return rcfg }} |
| + |
| + paused, resume := make(chan uint64), make(chan struct{}) |
| + // Only the first pass (A) pauses; a sync.Once would block pass B on A. |
| + var first atomic.Bool |
| + passAfterReconcileHook = func() { |
| + if first.CompareAndSwap(false, true) { |
| + paused <- v.Snapshot().Epoch |
| + <-resume |
| + } |
| + } |
| + t.Cleanup(func() { passAfterReconcileHook = nil }) |
| + |
| + done := make(chan struct{}) |
| + go func() { defer close(done); legs.full(context.Background()) }() // pass A |
| + epochA := <-paused |
| + |
| + w.record(t, "shop", "two.shop.example.com") |
| + legs.full(context.Background()) // pass B, whole |
| + epochB := mw.lastEpoch |
| + if epochB <= epochA { |
| + t.Fatalf("pass B's moves consumed epoch %d, not past A's reconcile at %d", epochB, epochA) |
| + } |
| + w.record(t, "cafe", "two.cafe.example.com") |
| + close(resume) |
| + <-done |
| + if got := mw.lastEpoch; got <= epochB { |
| + t.Fatalf("pass A's moves consumed epoch %d after B consumed %d: it did not refresh under w.pass", got, epochB) |
| + } |
| + if got, want := mw.lastEpoch, v.Snapshot().Epoch; got != want { |
| + t.Fatalf("pass A's moves consumed epoch %d, the newest is %d", got, want) |
| + } |
| +} |
| + |
| +// A snapshot older than one the worker already consumed is refused: the |
| +// pass logs it and takes no step (here, the offline keep view stays unread). |
| +func TestMovesPassRefusesOlderEpoch(t *testing.T) { |
| + w := newCIWorld(t) |
| + v := armView(t, w.live) |
| + if err := v.Rebuild(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + logs := captureLog(t) |
| + mw := newMovesWorker(moveCfg{}) |
| + mw.cfg.Offline = newOfflineView(mw.cfg, nil) |
| + newest := v.Snapshot().Epoch |
| + mw.lastEpoch = newest + 5 // as if a newer snapshot had been consumed |
| + if err := movesPass(context.Background(), w.live, mw); err != nil { |
| + t.Fatal(err) |
| + } |
| + if mw.cfg.Offline.built() { |
| + t.Fatal("a pass over an older snapshot read its records") |
| + } |
| + if mw.lastEpoch != newest+5 { |
| + t.Fatalf("lastEpoch went from %d to %d", newest+5, mw.lastEpoch) |
| + } |
| + want := "agent: moves pass: snapshot epoch" |
| + if !strings.Contains(logs(), want) || !strings.Contains(logs(), "older than") { |
| + t.Fatalf("no refusal line in %q", logs()) |
| + } |
| + // The same worker at the current epoch runs. |
| + mw.lastEpoch = 0 |
| + if err := movesPass(context.Background(), w.live, mw); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !mw.cfg.Offline.built() || mw.lastEpoch != v.Snapshot().Epoch { |
| + t.Fatalf("the pass at the current epoch did not run (built %v, epoch %d)", mw.cfg.Offline.built(), mw.lastEpoch) |
| + } |
| +} |
| + |
| +// ---- fastTick -------------------------------------------------------------- |
| + |
| +// ciFastDeps is a fastDeps over an indexed store with counting hooks for |
| +// both pass shapes. |
| +func ciFastDeps(t *testing.T) (*fastDeps, *release.DirStore, *[]string, *[][]string) { |
| + t.Helper() |
| + env := newFastFixture(t) |
| + ds := release.NewDirStore(t.TempDir()) |
| + ds.EnableChangeIndex() |
| + env.store = ds |
| + var full []string |
| + var scoped [][]string |
| + d := env.deps(func() { full = append(full, "full") }) |
| + d.watch = release.NewChangeWatch(ds) |
| + d.scopedPass = func(_ context.Context, moved []string) { scoped = append(scoped, moved) } |
| + return d, ds, &full, &scoped |
| +} |
| + |
| +func TestFastTickIndexOnlyRunsScopedPass(t *testing.T) { |
| + t.Parallel() |
| + ctx := context.Background() |
| + d, ds, full, scoped := ciFastDeps(t) |
| + now := time.Unix(1754640000, 0) |
| + fastTick(ctx, d, now) // seeds the serial and the index: no pass |
| + if len(*full)+len(*scoped) != 0 { |
| + t.Fatalf("the seeding tick ran a pass: full %v scoped %v", *full, *scoped) |
| + } |
| + if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + fastTick(ctx, d, now.Add(10*time.Second)) |
| + if len(*full) != 0 { |
| + t.Fatalf("an index-only change ran the full pass: %v", *full) |
| + } |
| + if want := [][]string{{"apps/bab/shop"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) { |
| + t.Fatalf("scoped passes %v, want %v", *scoped, want) |
| + } |
| + // Inside the clamp the change waits, and is acted on at the next tick |
| + // past it. |
| + if _, err := ds.Put(ctx, "apps/bab/cafe/instances/one.cafe.example.com.json", []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + fastTick(ctx, d, now.Add(12*time.Second)) |
| + if len(*scoped) != 1 { |
| + t.Fatalf("a change inside the clamp ran at once: %v", *scoped) |
| + } |
| + fastTick(ctx, d, now.Add(16*time.Second)) |
| + if want := [][]string{{"apps/bab/shop"}, {"apps/bab/cafe"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) { |
| + t.Fatalf("scoped passes %v, want %v", *scoped, want) |
| + } |
| + fastTick(ctx, d, now.Add(30*time.Second)) |
| + if len(*scoped) != 2 { |
| + t.Fatalf("an unchanged index ran the scoped pass again: %v", *scoped) |
| + } |
| + // Scopes outside the record view's (control/edges) run no pass. |
| + if _, err := ds.Put(ctx, "control/edges/one.json", []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + fastTick(ctx, d, now.Add(40*time.Second)) |
| + if len(*scoped) != 2 || len(*full) != 0 { |
| + t.Fatalf("a control/ scope ran a pass: full %v scoped %v", *full, *scoped) |
| + } |
| +} |
| + |
| +// A shared watch: another poller (the record view) taking Poll's moved list |
| +// first does not hide the movement from fastTick. |
| +func TestFastTickActsOnObservedNotPollReturn(t *testing.T) { |
| + t.Parallel() |
| + ctx := context.Background() |
| + d, ds, _, scoped := ciFastDeps(t) |
| + now := time.Unix(1754640000, 0) |
| + fastTick(ctx, d, now) |
| + if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if moved, err := d.watch.Poll(ctx); err != nil || len(moved) == 0 { |
| + t.Fatalf("the sibling poll saw %v, %v", moved, err) |
| + } |
| + fastTick(ctx, d, now.Add(10*time.Second)) |
| + if want := [][]string{{"apps/bab/shop"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) { |
| + t.Fatalf("scoped passes %v, want %v: a sibling's Poll swallowed the movement", *scoped, want) |
| + } |
| +} |
| + |
| +func TestSerialChangeStillRunsFullPass(t *testing.T) { |
| + t.Parallel() |
| + ctx := context.Background() |
| + d, ds, full, scoped := ciFastDeps(t) |
| + now := time.Unix(1754640000, 0) |
| + fastTick(ctx, d, now) |
| + if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := release.BumpSerial(ctx, ds, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + fastTick(ctx, d, now.Add(10*time.Second)) |
| + if len(*full) != 1 || len(*scoped) != 0 { |
| + t.Fatalf("a serial change with the index moved: full %v scoped %v, want one full pass only", *full, *scoped) |
| + } |
| + // The full pass covered the movement: the next tick runs nothing. |
| + fastTick(ctx, d, now.Add(20*time.Second)) |
| + if len(*full) != 1 || len(*scoped) != 0 { |
| + t.Fatalf("the movement the full pass covered ran again: full %v scoped %v", *full, *scoped) |
| + } |
| +} |
| + |
| +// The backstop asks for the unrestricted full pass after its rebuild; the |
| +// loop's own tick runs it, once. |
| +func TestFastTickRunsTheBackstopsFullPass(t *testing.T) { |
| + t.Parallel() |
| + ctx := context.Background() |
| + d, ds, full, scoped := ciFastDeps(t) |
| + var due atomic.Bool |
| + d.fullDue = &due |
| + now := time.Unix(1754640000, 0) |
| + fastTick(ctx, d, now) |
| + v := provision.NewRecordView(ds, d.watch) |
| + recordViewBackstop(ctx, v, &due) |
| + if v.Snapshot() == nil { |
| + t.Fatal("the backstop did not rebuild the view") |
| + } |
| + fastTick(ctx, d, now.Add(10*time.Second)) |
| + if len(*full) != 1 || len(*scoped) != 0 || due.Load() { |
| + t.Fatalf("after the backstop: full %v scoped %v due %v, want one full pass", *full, *scoped, due.Load()) |
| + } |
| + fastTick(ctx, d, now.Add(20*time.Second)) |
| + if len(*full) != 1 { |
| + t.Fatalf("the backstop's full pass ran again: %v", *full) |
| + } |
| +} |
| + |
| +func TestBackstopEveryIsJitteredAroundTheBackstop(t *testing.T) { |
| + t.Parallel() |
| + lo, hi := release.ChangeIndexBackstop*9/10, release.ChangeIndexBackstop*11/10 |
| + for range 200 { |
| + if d := backstopEvery(); d < lo || d >= hi { |
| + t.Fatalf("backstopEvery() = %v, want in [%v, %v)", d, lo, hi) |
| + } |
| + } |
| +} |
| + |
| +// ---- the scoped pass ------------------------------------------------------- |
| + |
| +// Codex 10's witness: on a box hosting shop and cafe, an index bump for shop |
| +// runs a scoped pass that reads nothing under cafe's prefix, while shop's |
| +// new config is delivered. |
| +func TestScopedPassTouchesOnlyMovedApp(t *testing.T) { |
| + w := newCIWorld(t) |
| + armView(t, w.live) |
| + armedMovesWorker.Store(newMovesWorker(moveCfg{})) |
| + t.Cleanup(func() { armedMovesWorker.Store(nil) }) |
| + raw, _, err := appconfig.GenerateKey() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + pk, err := appconfig.ParsePrivateKey(raw, "test") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := restartTestRegistry(t) |
| + dd := t.TempDir() |
| + legs := &passLegs{store: w.live, reg: reg, optDir: t.TempDir(), priv: pk, dd: dd, |
| + cfg: func() provision.ReconcileConfig { return provision.ReconcileConfig{DataDir: dd, Layout: flatLayout} }} |
| + legs.full(context.Background()) |
| + for _, h := range []string{"shop.example.com", "cafe.example.com"} { |
| + rt, err := reg.Get(h) |
| + if err != nil || rt == nil { |
| + t.Fatalf("fixture: no route for %s: %v", h, err) |
| + } |
| + if !strings.Contains(readConfigEnv(t, dd, *rt), "APP=") { |
| + t.Fatalf("fixture: %s got no config from the full pass", h) |
| + } |
| + } |
| + |
| + if err := appconfig.SetEnv(context.Background(), w.live, w.acct, "shop", map[string]string{"APP": "shop2"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + w.live.reset() |
| + legs.scoped(context.Background(), []string{w.scope("shop")}) |
| + |
| + cafe := "apps/" + w.acct + "/cafe/" |
| + shop := "apps/" + w.acct + "/shop/" |
| + var shopReads int |
| + for _, c := range w.live.got() { |
| + _, key, _ := strings.Cut(c, " ") |
| + if strings.HasPrefix(key, cafe) { |
| + t.Errorf("the scoped pass for shop read %s", c) |
| + } |
| + if strings.HasPrefix(key, shop) { |
| + shopReads++ |
| + } |
| + } |
| + if shopReads == 0 { |
| + t.Fatalf("the scoped pass read nothing under shop: %v", w.live.got()) |
| + } |
| + rt, _ := reg.Get("shop.example.com") |
| + if got := readConfigEnv(t, dd, *rt); !strings.Contains(got, "APP=shop2") { |
| + t.Fatalf("shop's config after the scoped pass = %q, want APP=shop2", got) |
| + } |
| +} |
| + |
| +func readConfigEnv(t *testing.T, dd string, rt registry.Route) string { |
| + t.Helper() |
| + b, err := os.ReadFile(appconfig.ConfigEnvPath(dd, rt)) |
| + if err != nil && !os.IsNotExist(err) { |
| + t.Fatal(err) |
| + } |
| + return string(b) |
| +} |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index 16efd306e..b7d03b0e9 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -110,6 +110,10 @@ type movesWorker struct { |
| // lost names the hosts whose claim lists this box in Lost, as this pass |
| // read them (lostHere); rebuilt every pass under pass. |
| lost map[string]bool |
| + // lastEpoch is the record view snapshot epoch the last pass consumed |
| + // (S3 change-index design §4.3); guarded by pass. A pass refuses an |
| + // older one: a guard, since publication is monotonic. |
| + lastEpoch uint64 |
| } |
| // staging is the memoized staging writer for a target bucket region. |
| @@ -1060,6 +1064,21 @@ func moveReadyDeltaFromEnv(getenv func(string) string) (delta int64, ok bool) { |
| func movesPass(ctx context.Context, s release.Store, w *movesWorker) error { |
| w.pass.Lock() |
| defer w.pass.Unlock() |
| + // With the change index armed, the record listing below reads the record |
| + // view, refreshed here inside w.pass, never a snapshot handed over from |
| + // a reconcile leg (S3 change-index design §4.3). Only the listing does: |
| + // every step after it reads and writes s, so a step's own re-read and |
| + // CAS see the bucket as they always have. Lock order: w.pass -> |
| + // RecordView.mu, which is a leaf. |
| + records := s |
| + if vs, epoch, ok := viewStore(ctx, s); ok { |
| + if epoch < w.lastEpoch { |
| + log.Printf("agent: moves pass: snapshot epoch %d older than %d consumed", epoch, w.lastEpoch) |
| + return nil |
| + } |
| + w.lastEpoch = epoch |
| + records = vs |
| + } |
| cfg := w.cfg |
| if cfg.claimRecords { |
| ctx = provision.WithRecordSequence(ctx) |
| @@ -1076,7 +1095,7 @@ 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) |
| + recs, unreadable, err := provision.ListAllRecordsStrict(ctx, records) |
| if err != nil { |
| return fmt.Errorf("moves: list records: %w", err) |
| } |
diff --git a/internal/host/steering_loop.go b/internal/host/steering_loop.go| index 6b2379d99..a13a55399 100644 |
| --- a/internal/host/steering_loop.go |
| +++ b/internal/host/steering_loop.go |
| @@ -126,7 +126,10 @@ func steeringPass(ctx context.Context, s release.Store, converge func(context.Co |
| if err != nil { |
| return err |
| } |
| - routes, err := steeredRoutesFromRecords(ctx, s) |
| + // The records through the record view when the change index is armed |
| + // and the view can vouch; the bucket otherwise (viewStore). |
| + rs, _, _ := viewStore(ctx, s) |
| + routes, err := steeredRoutesFromRecords(ctx, rs) |
| if err != nil { |
| return err |
| } |