amadan

carlos / platform Public

host: config generation loop — a set converges like a canary adopts (issue #44)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013WVANvB83eHrRANRpEL8mn
Paul Campbell 5d72385a6feea6a03b6d4878ff460fab258441a9 parent 1ce7c64
2 files changed, +439 −8
  • internal/host/agent.go +182 −8
  • internal/host/agent_test.go +257 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 49d4d016d..b2a30e6d8 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -67,6 +67,7 @@ import (
"crypto/ed25519"
"errors"
"flag"
+ "fmt"
"io"
"log"
"math/rand/v2"
@@ -149,6 +150,7 @@ func RunAgent(args []string) {
go domainsLoop(reg, peekAcmeCache(args))
go metricsLoop()
go configSyncLoop(reg, svc, dd)
+ go configGenLoop(reg, svc, dd)
go socketAdvisoryLoop(reg, svc)
// The same serving loop as `carlos edge`, with the activator injected.
@@ -247,22 +249,194 @@ func configSyncLoop(reg *registry.Registry, svc *activator.Service, dd string) {
log.Printf("agent: config sync loop disabled: %v", err)
return
}
- bounce := func(host string) error {
+ log.Printf("agent: config sync loop armed (hourly + jitter)")
+ delay := time.Minute
+ for {
+ time.Sleep(delay)
+ delay = time.Hour + rand.N(10*time.Minute)
+ if err := appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc)); err != nil {
+ log.Printf("agent: config sync: %v", err)
+ }
+ }
+}
+
+// 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
+// loop, unlike the adopt pair: those two differ from each other (filter,
+// nag) where these two must not — the delivery must be identical whichever
+// loop happened to notice a change first, and a divergence would mean an
+// app converges one way at 3s and another way on the hour. A nil svc (no
+// activator armed) still returns a working no-op: the files converge for
+// future spawns, there is simply nothing running to restart.
+func bounce(svc *activator.Service) func(string) error {
+ return func(host string) error {
if svc == nil {
- return nil // no activator: files still converge for future spawns
+ return nil
}
if svc.Bounce(host) {
log.Printf("agent: config changed, bounced %s", host)
}
return nil
}
- log.Printf("agent: config sync loop armed (hourly + jitter)")
- delay := time.Minute
+}
+
+// genAbsent marks an app whose generation object was READ and found
+// missing, distinguishing it from an app never seen at all — the two are
+// different in the only way this loop cares about: a never-seen app must
+// record silently (first sight), while a known-absent app that suddenly
+// has a generation object is an app whose config was just written for the
+// first time, which must sync. A real etag can never collide with it: no
+// store produces one containing a NUL byte (S3 etags are hex/quoted,
+// DirStore's is hex).
+const genAbsent = "\x00absent"
+
+// genPollPass is one tick of configGenLoop, split out for tests: read every
+// exec-config app's generation etag and report whether ANY moved since the
+// cache was built.
+//
+// Semantics, each of which is a deliberate choice:
+//
+// - First sight of an app (no cache entry) records silently and does NOT
+// sync. Boot must not re-bounce the world — every app would look
+// "changed" on the first tick after a restart. configSyncLoop's ~1min
+// first pass owns boot convergence.
+// - An unchanged etag does nothing.
+// - Absent that stays absent does nothing, forever: an app last written
+// by a pre-generation CLI has no generation object and never will
+// until someone writes its config again. That is legal, not a fault.
+// - Any movement syncs: the etag changed, or absent -> present.
+// present -> absent shouldn't happen (generation objects are only ever
+// created and overwritten, never deleted), but it's treated as a change
+// too — one extra SyncExec pass is cheap, and pretending we didn't see
+// it would be a lie about the only thing this function reports.
+// - ONE bool for the whole pass, however many apps moved: the caller runs
+// one SyncExec, which resolves every route itself. Per-app syncs would
+// multiply bucket reads to deliver exactly the same files.
+// - Apps that vanish from the route set lose their cache entries, so a
+// re-added app is a first sight again rather than a comparison against
+// an etag remembered from before it left (a route being re-added gets
+// fresh config from the spawn path anyway).
+// - A read error that isn't "not found" leaves that app's cache entry
+// exactly as it was and keeps evaluating the rest: the errors come back
+// for the caller to aggregate into ONE log line per pass, never one per
+// app per tick. Clearing the entry instead would turn a recovered blip
+// into a bogus first sight or a bogus change.
+//
+// The cache is mutated in place — it lives across ticks in the loop, which
+// is the whole point (there is nowhere else to remember what "changed"
+// means).
+func genPollPass(ctx context.Context, routes []registry.Route, store release.Store, cache map[[2]string]string) (bool, []error) {
+ apps := appconfig.ExecConfigApps(routes)
+ live := make(map[[2]string]bool, len(apps))
+ changed := false
+ var readErrs []error
+ for _, k := range apps {
+ live[k] = true
+ _, etag, err := store.Get(ctx, appconfig.GenKey(k[0], k[1]))
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ etag = genAbsent
+ case err != nil:
+ readErrs = append(readErrs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
+ continue // cache entry untouched: this app is unknown this tick, not changed
+ }
+ prev, seen := cache[k]
+ cache[k] = etag
+ if seen && prev != etag {
+ changed = true
+ }
+ }
+ for k := range cache {
+ if !live[k] {
+ delete(cache, k)
+ }
+ }
+ return changed, readErrs
+}
+
+// configGenLoop is canaryAdoptLoop's shape applied to issue #44: a config
+// set should converge like a canary adopts — seconds, not the next hour.
+// Every ~3s it reads one small object per exec-config app (the generation
+// object every config write touches) and, on any movement, runs ONE
+// SyncExec pass for the whole tick. Spurious movement costs a pass that
+// finds nothing to write: SyncExec bounces only routes whose rendered file
+// actually changed, so a redundant touch can never cost a restart.
+//
+// Gated on CARLOS_SECRETS_KEY exactly as configSyncLoop is, for a reason
+// beyond convention: without the key SyncExec cannot render an app's
+// environment at all, so this loop would poll the bucket every 3 seconds
+// forever to signal no one. openDeployStore failing is the same story with
+// nothing to read.
+//
+// Cardinality: unlike canaryAdoptLoop, which is bounded to the deliberately
+// rare canary/* routes, this watches EVERY exec-config app on the box. That
+// is accepted knowingly — one GET of the smallest object in the bucket per
+// app per 3s, against the store canaryAdoptLoop already hits every 2s. What
+// keeps it bounded is that the pass reads generation objects only: no
+// manifests, no pointers, no per-environment resolution. All of that
+// happens inside the SyncExec a real change triggers.
+//
+// Deliberately NOT done here:
+//
+// - No fast first pass. The loop sleeps BEFORE its first tick and the
+// first sight of every app is silent (see genPollPass) — boot
+// convergence belongs to configSyncLoop's ~1min first pass, and a box
+// coming up must not bounce every instance it has because it has never
+// seen their etags before.
+// - No unit-backed routes. This process cannot restart units (see
+// RunAdopt); the root adopt path keeps that half, at its own cadence.
+// - No hourly floor replaced. configSyncLoop stays exactly as it was: it
+// catches apps written by older CLIs (no generation object at all),
+// ticks this loop missed, and the boot-retry gap of issue #27.
+//
+// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
+// whole decision lives in genPollPass, which is tested.
+func configGenLoop(reg *registry.Registry, svc *activator.Service, dd string) {
+ raw := os.Getenv("CARLOS_SECRETS_KEY")
+ if raw == "" {
+ log.Printf("agent: config generation loop disabled: CARLOS_SECRETS_KEY not set")
+ return
+ }
+ priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
+ if err != nil {
+ log.Printf("agent: config generation loop disabled: %v", err)
+ return
+ }
+ ctx := context.Background()
+ store, err := openDeployStore(ctx)
+ if err != nil {
+ log.Printf("agent: config generation loop disabled: %v", err)
+ return
+ }
+ // One cache for the life of the process: what "changed" means is
+ // exactly "differs from the last tick that could read it".
+ cache := map[[2]string]string{}
+ log.Printf("agent: config generation loop armed (3s + jitter, exec-config apps)")
for {
- time.Sleep(delay)
- delay = time.Hour + rand.N(10*time.Minute)
- if err := appconfig.SyncExec(ctx, reg, store, priv, dd, bounce); err != nil {
- log.Printf("agent: config sync: %v", err)
+ time.Sleep(3*time.Second + rand.N(time.Second))
+ routes, err := reg.List()
+ if err != nil {
+ // The registry is a local file this process already holds open;
+ // a failing List is a broken box, not a passing blip, so it is
+ // worth a line even at this cadence (the same call is logged per
+ // tick by statusLoop).
+ log.Printf("agent: config generation poll: list routes: %v", err)
+ continue
+ }
+ changed, readErrs := genPollPass(ctx, routes, store, cache)
+ if len(readErrs) > 0 {
+ // ONE line per pass, never one per app: a bucket having a bad
+ // minute would otherwise flood the journal at 20 lines a minute
+ // per app. The count carries the scale, the first error carries
+ // the cause; they are the same cause in practice.
+ log.Printf("agent: config generation poll: %d apps unreadable: %v", len(readErrs), readErrs[0])
+ }
+ if !changed {
+ continue
+ }
+ if err := appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc)); err != nil {
+ log.Printf("agent: config generation sync: %v", err)
}
}
}
diff --git a/internal/host/agent_test.go b/internal/host/agent_test.go
new file mode 100644
index 000000000..d76f88694
--- /dev/null
+++ b/internal/host/agent_test.go
@@ -0,0 +1,257 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// genKit is the shared setup for the generation-poll tests: a directory
+// store standing in for the deployment bucket, and a context. No registry
+// — genPollPass takes routes, not a *registry.Registry, precisely so these
+// tests need no bolt database (the loop does its own reg.List()).
+func genKit(t *testing.T) (context.Context, release.Store) {
+ t.Helper()
+ return context.Background(), release.NewDirStore(t.TempDir())
+}
+
+// execRoute is the socket-exec instance shape appconfig.ExecConfigApps
+// accepts (execSyncable): a non-platform, ready instance with a socket
+// upstream. Field-for-field the fixture shape syncexec_test.go uses, so a
+// change to the predicates fails these tests for the same reason it fails
+// appconfig's.
+func execRoute(host, acct, app string) registry.Route {
+ return registry.Route{
+ Host: host, Kind: "instance", App: app, Account: acct,
+ Status: "ready", Socket: "/run/carlos/" + host + ".sock",
+ }
+}
+
+// genEtag reads the app's generation etag straight from the store, so a
+// test can assert the cache holds what the bucket actually reports rather
+// than a value the test made up.
+func genEtag(t *testing.T, ctx context.Context, s release.Store, acct, app string) string {
+ t.Helper()
+ _, etag, err := s.Get(ctx, appconfig.GenKey(acct, app))
+ if err != nil {
+ t.Fatalf("read generation %s/%s: %v", acct, app, err)
+ }
+ return etag
+}
+
+func touch(t *testing.T, ctx context.Context, s release.Store, acct, app string) {
+ t.Helper()
+ if err := appconfig.TouchGeneration(ctx, s, acct, app); err != nil {
+ t.Fatalf("TouchGeneration %s/%s: %v", acct, app, err)
+ }
+}
+
+// failingKeyStore fails Get for exactly one key — a transient bucket blip
+// against a single app's generation object while every other key in the
+// same pass resolves normally (the shape an S3 503 against one object
+// takes). Same wrapper appconfig's syncexec_test.go uses, for the same
+// reason.
+type failingKeyStore struct {
+ release.Store
+ failKey string
+ err error
+}
+
+func (f *failingKeyStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if key == f.failKey {
+ return nil, "", f.err
+ }
+ return f.Store.Get(ctx, key)
+}
+
+func TestGenPollPassFirstSightIsSilent(t *testing.T) {
+ ctx, s := genKit(t)
+ touch(t, ctx, s, "bab", "eleven")
+ routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")}
+ cache := map[[2]string]string{}
+
+ changed, errs := genPollPass(ctx, routes, s, cache)
+ if changed {
+ t.Fatal("first sight of an app must not trigger a sync (boot would re-bounce the world)")
+ }
+ if len(errs) != 0 {
+ t.Fatalf("unexpected read errors: %v", errs)
+ }
+ want := genEtag(t, ctx, s, "bab", "eleven")
+ if got := cache[[2]string{"bab", "eleven"}]; got != want {
+ t.Fatalf("cache etag = %q, want the store's %q", got, want)
+ }
+}
+
+func TestGenPollPassUnchangedIsSilent(t *testing.T) {
+ ctx, s := genKit(t)
+ touch(t, ctx, s, "bab", "eleven")
+ routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")}
+ cache := map[[2]string]string{}
+
+ genPollPass(ctx, routes, s, cache)
+ changed, errs := genPollPass(ctx, routes, s, cache)
+ if changed {
+ t.Fatal("an unmoved generation etag must not trigger a sync")
+ }
+ if len(errs) != 0 {
+ t.Fatalf("unexpected read errors: %v", errs)
+ }
+}
+
+func TestGenPollPassTouchedGenerationTriggersSync(t *testing.T) {
+ ctx, s := genKit(t)
+ touch(t, ctx, s, "bab", "eleven")
+ routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")}
+ cache := map[[2]string]string{}
+ genPollPass(ctx, routes, s, cache)
+
+ touch(t, ctx, s, "bab", "eleven")
+ changed, errs := genPollPass(ctx, routes, s, cache)
+ if !changed {
+ t.Fatal("a moved generation etag must trigger a sync")
+ }
+ if len(errs) != 0 {
+ t.Fatalf("unexpected read errors: %v", errs)
+ }
+ want := genEtag(t, ctx, s, "bab", "eleven")
+ if got := cache[[2]string{"bab", "eleven"}]; got != want {
+ t.Fatalf("cache etag = %q, want the new %q (a stale cache would re-sync forever)", got, want)
+ }
+ if changed, _ := genPollPass(ctx, routes, s, cache); changed {
+ t.Fatal("the pass after a change must be silent again")
+ }
+}
+
+func TestGenPollPassAbsentGenerationStaysSilent(t *testing.T) {
+ ctx, s := genKit(t)
+ // No touch at all: an app last written by a pre-generation CLI. Legal
+ // forever — the hourly configSyncLoop is its floor.
+ routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")}
+ cache := map[[2]string]string{}
+
+ for i := range 2 {
+ changed, errs := genPollPass(ctx, routes, s, cache)
+ if changed {
+ t.Fatalf("pass %d: an app with no generation object must never trigger a sync", i)
+ }
+ if len(errs) != 0 {
+ t.Fatalf("pass %d: absent is not a read error: %v", i, errs)
+ }
+ }
+ if got := cache[[2]string{"bab", "eleven"}]; got != genAbsent {
+ t.Fatalf("cache = %q, want the known-absent sentinel %q", got, genAbsent)
+ }
+}
+
+func TestGenPollPassAbsentToPresentTriggersSync(t *testing.T) {
+ ctx, s := genKit(t)
+ routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")}
+ cache := map[[2]string]string{}
+ genPollPass(ctx, routes, s, cache) // records known-absent
+
+ touch(t, ctx, s, "bab", "eleven") // the app's first ever config write
+ changed, errs := genPollPass(ctx, routes, s, cache)
+ if !changed {
+ t.Fatal("absent -> present is the first write to an old app: it must sync")
+ }
+ if len(errs) != 0 {
+ t.Fatalf("unexpected read errors: %v", errs)
+ }
+}
+
+func TestGenPollPassOneChangedAppSyncsOnce(t *testing.T) {
+ ctx, s := genKit(t)
+ touch(t, ctx, s, "bab", "eleven")
+ touch(t, ctx, s, "bab", "titogo")
+ routes := []registry.Route{
+ execRoute("a.example.com", "bab", "eleven"),
+ execRoute("b.example.com", "bab", "titogo"),
+ }
+ cache := map[[2]string]string{}
+ genPollPass(ctx, routes, s, cache)
+
+ touch(t, ctx, s, "bab", "titogo")
+ changed, errs := genPollPass(ctx, routes, s, cache)
+ if !changed {
+ t.Fatal("one moved app in the set must report the pass as changed")
+ }
+ if len(errs) != 0 {
+ t.Fatalf("unexpected read errors: %v", errs)
+ }
+ // One flag for the whole pass: the caller runs ONE SyncExec, which
+ // resolves every app itself. The unchanged app's cache entry must be
+ // carried forward untouched so it doesn't re-fire next tick.
+ if got, want := cache[[2]string{"bab", "eleven"}], genEtag(t, ctx, s, "bab", "eleven"); got != want {
+ t.Fatalf("unchanged app's cache = %q, want %q", got, want)
+ }
+ if changed, _ := genPollPass(ctx, routes, s, cache); changed {
+ t.Fatal("the next pass must be silent for both apps")
+ }
+}
+
+func TestGenPollPassDropsVanishedApps(t *testing.T) {
+ ctx, s := genKit(t)
+ touch(t, ctx, s, "bab", "eleven")
+ touch(t, ctx, s, "bab", "titogo")
+ both := []registry.Route{
+ execRoute("a.example.com", "bab", "eleven"),
+ execRoute("b.example.com", "bab", "titogo"),
+ }
+ one := both[:1]
+ cache := map[[2]string]string{}
+ genPollPass(ctx, both, s, cache)
+
+ genPollPass(ctx, one, s, cache) // titogo's route is gone
+ if _, ok := cache[[2]string{"bab", "titogo"}]; ok {
+ t.Fatal("a route that left the registry must leave no cache entry behind")
+ }
+ // The proof that dropping matters: while the app was unrouted its
+ // config moved. Re-adding the route must record silently (first sight
+ // again), not fire a sync off a remembered etag — a route being
+ // re-added already gets fresh config from the spawn path.
+ touch(t, ctx, s, "bab", "titogo")
+ changed, errs := genPollPass(ctx, both, s, cache)
+ if changed {
+ t.Fatal("a re-added app is a first sight, not a change")
+ }
+ if len(errs) != 0 {
+ t.Fatalf("unexpected read errors: %v", errs)
+ }
+}
+
+func TestGenPollPassReadErrorLeavesCacheAndKeepsGoing(t *testing.T) {
+ ctx, dir := genKit(t)
+ touch(t, ctx, dir, "bab", "eleven")
+ touch(t, ctx, dir, "bab", "titogo")
+ routes := []registry.Route{
+ execRoute("a.example.com", "bab", "eleven"),
+ execRoute("b.example.com", "bab", "titogo"),
+ }
+ cache := map[[2]string]string{}
+ genPollPass(ctx, routes, dir, cache)
+ elevenEtag := cache[[2]string{"bab", "eleven"}]
+
+ // eleven's generation object goes unreadable; titogo's moves.
+ boom := errors.New("bucket said no")
+ s := &failingKeyStore{Store: dir, failKey: appconfig.GenKey("bab", "eleven"), err: boom}
+ touch(t, ctx, dir, "bab", "titogo")
+
+ changed, errs := genPollPass(ctx, routes, s, cache)
+ if !changed {
+ t.Fatal("one app failing must not stop the others being evaluated")
+ }
+ if len(errs) != 1 {
+ t.Fatalf("read errors = %v, want exactly one (aggregated into one log line per pass)", errs)
+ }
+ if !errors.Is(errs[0], boom) {
+ t.Fatalf("read error = %v, want it to wrap %v", errs[0], boom)
+ }
+ if got := cache[[2]string{"bab", "eleven"}]; got != elevenEtag {
+ t.Fatalf("unreadable app's cache = %q, want it untouched at %q (a cleared entry would re-fire on recovery)", got, elevenEtag)
+ }
+}