amadan

carlos / platform Public

host, activator: the agent finishes an adoption the root pass performed

A new version on a real box is staged, pointed live and stamped by the
root pass, which cannot cycle the agent's exec children. The agent's own
pass then finds the host up to date and never reaches its restart
callback, so a warm instance served the old build until the activator
sweep's 90s grace ran out: 110-200s per deploy on the flagship (#231).

The fast tick now asks the activator which warm instances have drifted
from live/<host> and cycles each one whose route is stamped with the
version the symlink names. The stamp is written after the adopting pass
is finished with the host, which is what makes the grace safe to skip.
Workers stay with the sweep.
Human

Prompt

OK, go ahead and message it. And let's fix all the bugs.

From this commit's own message. The prompt ledger has no entries for it — which is what a squashed, rebased or amended commit looks like, and what every commit looks like in a repo without a ledger.

Summary

Adds activator Drifted/ConvergeAdopted and a converge leg on the agent's 2s tick, so a root-adopted release reaches a warm exec instance in seconds instead of after the 90s sweep grace.
Agent notes
Discussion #231. The regression test is TestRootAdoptionOfAWarmExecInstanceIsFinishedByTheAgent; it fails with the leg disabled. No new member-facing strings: the cycle emits the existing 'instance stopping' event. Bounce's '(carlos restart)' wording on the direct adopt path is untouched and still mislabels an adoption.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 11047d35372dee24548fd035ba5b9c464cb718bd parent 8d8e4ba
4 files changed, +457 −0
  • internal/activator/activator.go +79 −0
  • internal/activator/converge_adopted_test.go +146 −0
  • internal/host/agent.go +57 −0
  • internal/host/converge_adopted_test.go +175 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 16ef6357a..192f7b00b 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1390,6 +1390,85 @@ func (a *Service) Bounce(host string) bool {
return true
}
+// Drifted lists the awake INSTANCES whose child was exec'd from a build
+// live/<host> no longer names, each with the version the symlink names now.
+// It is the cheap half of ConvergeAdopted below: a caller polling at seconds
+// cadence asks this first and, on the ordinary tick where nothing has shipped,
+// gets nothing back and reads no registry row at all. One readlink per awake
+// instance, the same read the sweep makes.
+//
+// A read error, an unknown spawn version and an unknown symlink target are all
+// "not drifted" — the sweep's own rule (ConvergeDecision): with nothing to
+// compare there is nothing to act on. Workers are never listed; see
+// ConvergeAdopted for why they stay with the sweep.
+func (a *Service) Drifted() map[string]string {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ var out map[string]string
+ for _, e := range a.entries {
+ if e.worker || e.state != actAwake || e.spawnVersion == "" || a.fences[e.host] != 0 {
+ continue
+ }
+ cur, err := a.liveVer(e.host)
+ if err != nil || cur == "" || cur == e.spawnVersion {
+ continue
+ }
+ if out == nil {
+ out = map[string]string{}
+ }
+ out[e.host] = cur
+ }
+ return out
+}
+
+// ConvergeAdopted cycles host's awake instance onto ver, a version an adopt
+// pass has FINISHED adopting for it, without the sweep's ConvergeGrace. It
+// reports whether it started a cycle.
+//
+// It exists because the adoption a real box performs is delegated. The agent
+// runs sandboxed and cannot write the version store, so the root pass stages,
+// repoints live/<host> and stamps the route — and cycles no exec child,
+// because the children belong to this process. The agent's own pass then
+// finds the host up to date and never reaches its restart callback, which
+// left the sweep's grace as the only thing standing between a deploy and the
+// build it shipped (discussion #231). This is the restart half of that
+// adoption, performed by the one process that can.
+//
+// The grace is safe to skip HERE and nowhere else, because of what the caller
+// vouches for: ver is the version the route is STAMPED with, and the stamp is
+// written only after the adopting pass is done with the host — past its own
+// restart and any rollback. The sweep cannot know that; it sees a symlink
+// move and waits to learn whether it holds. ver must also still be what the
+// symlink names, checked here under the lock: a stamp the symlink has moved
+// on from is a newer adoption in flight, and its own stamp will bring the
+// caller back.
+//
+// Like Bounce it never starts anything — an instance's next request is the
+// wake half — and it is the same stop the sweep's convergence performs:
+// SIGTERM drain, final checkpoint, lease release.
+//
+// Instances only. A worker's registry version is what its child REPORTS
+// (cleared on every respawn), not what adoption stamped, so the agreement
+// this rests on does not exist for one; the sweep's convergeWorkerLocked,
+// behind the grace, remains a worker's path onto a new build.
+func (a *Service) ConvergeAdopted(host, ver string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.worker || e.state != actAwake || a.fences[host] != 0 {
+ return false
+ }
+ if ver == "" || e.spawnVersion == "" || e.spawnVersion == ver {
+ return false
+ }
+ if cur, err := a.liveVer(host); err != nil || cur != ver {
+ return false
+ }
+ log.Printf("activator: %s serving %s but %s is adopted — cycling onto the new build", host, e.spawnVersion, ver)
+ a.beginStopLocked(e)
+ return true
+}
+
// --- standalone sidecars (workers) -------------------------------------------
//
// A worker is a member of this same state machine with the request path
diff --git a/internal/activator/converge_adopted_test.go b/internal/activator/converge_adopted_test.go
new file mode 100644
index 000000000..547841fe6
--- /dev/null
+++ b/internal/activator/converge_adopted_test.go
@@ -0,0 +1,146 @@
+package activator
+
+import (
+ "context"
+ "reflect"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+// The adoption a real box performs is delegated: the sandboxed agent cannot
+// write the version store, so the ROOT pass stages, repoints live/<host> and
+// stamps the version — and cycles no exec child, because the children are the
+// agent's. Nothing then told the agent the symlink had moved, and the awake
+// child served the old build until the sweep's grace ran out (discussion #231:
+// 110-200s per deploy on the flagship). ConvergeAdopted is the agent finishing
+// the half it delegated: no grace, no sweep.
+func TestConvergeAdoptedCyclesAStaleInstanceWithoutTheGrace(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour) // idle far away — only convergence can stop it
+ var ver atomic.Value
+ ver.Store("v1")
+ a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
+ b.execVerFn = func() string { return ver.Load().(string) }
+ host := "one.example.com"
+
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ if got := a.Drifted(); len(got) != 0 {
+ t.Fatalf("Drifted = %v with the symlink unmoved, want none", got)
+ }
+
+ ver.Store("v2") // the root pass repoints under the running child
+ if got, want := a.Drifted(), map[string]string{host: "v2"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("Drifted = %v, want %v", got, want)
+ }
+ if !a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted refused an awake instance serving v1 under a symlink that names v2")
+ }
+ // No clock advance and no sweep: the cycle is the call's own doing.
+ waitFor(t, "the adopted-version cycle", func() bool { return a.entryCount() == 0 })
+ if !b.lastProc().isStopped() {
+ t.Fatal("stale instance's process was not stopped")
+ }
+ if !b.lastRep().isClosed() {
+ t.Fatal("the cycle skipped the final checkpoint / lease release")
+ }
+
+ // The next request wakes it onto v2, and a converged child is left alone.
+ release, err = a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("re-wake after the cycle: %v", err)
+ }
+ release()
+ if got := a.Drifted(); len(got) != 0 {
+ t.Fatalf("Drifted = %v after re-waking onto v2, want none", got)
+ }
+ if a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted cycled a child that already runs the adopted version")
+ }
+ if a.entryCount() != 1 {
+ t.Fatal("a converged instance was stopped")
+ }
+}
+
+// Every refusal leaves the instance exactly as it was. In particular it never
+// WAKES anything: a sleeper picks the new build up at its own next wake.
+func TestConvergeAdoptedRefusals(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ var ver atomic.Value
+ ver.Store("v1")
+ a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
+ b.execVerFn = func() string { return ver.Load().(string) }
+ host := "one.example.com"
+
+ if a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted acted on a host that is asleep")
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("ConvergeAdopted woke a sleeping host")
+ }
+
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ ver.Store("v2")
+
+ // The caller's version and the symlink must agree. A stamp that names
+ // something the symlink does not is an adoption still in flight (or
+ // already superseded), and cycling now would land on whichever build the
+ // symlink happens to name.
+ if a.ConvergeAdopted(host, "v3") {
+ t.Fatal("ConvergeAdopted cycled onto a version the live symlink does not name")
+ }
+ if a.ConvergeAdopted(host, "") {
+ t.Fatal("ConvergeAdopted cycled for an empty version")
+ }
+ if a.entryCount() != 1 {
+ t.Fatal("a refused ConvergeAdopted stopped the instance")
+ }
+}
+
+// An unknown spawn version is "don't converge", the same rule the sweep keeps:
+// with nothing to compare, a cycle could never be shown to have been needed.
+func TestConvergeAdoptedLeavesAnUnknownSpawnVersionAlone(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.liveVer = func(string) (string, error) { return "v2", nil }
+ b.execVerFn = func() string { return "" }
+ host := "one.example.com"
+
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ if got := a.Drifted(); len(got) != 0 {
+ t.Fatalf("Drifted = %v for a child whose spawn version is unknown, want none", got)
+ }
+ if a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted cycled a child whose spawn version is unknown")
+ }
+}
+
+// A worker is out of this path's reach on purpose: its registry version is
+// what the child REPORTS, not what adoption stamped, so "the stamp agrees with
+// the symlink" means something else there. The sweep's own worker converge
+// (one move, behind the grace) still covers it.
+func TestConvergeAdoptedLeavesWorkersToTheSweep(t *testing.T) {
+ a, _, _, ver := shippingActivator(t, time.Hour)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ waitFor(t, "the worker to be awake", func() bool { return a.entrySpawnVersion(host) == "v1" })
+ ver.Store("v2")
+ if got := a.Drifted(); len(got) != 0 {
+ t.Fatalf("Drifted = %v, want no workers in it", got)
+ }
+ if a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted cycled a worker")
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 255b399d9..4c6175c89 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -1376,6 +1376,7 @@ type fastDeps struct {
restartLeg func(ctx context.Context) // the `carlos restart` check (issue #113), or a test hook
wakeLeg func(ctx context.Context) // the sidecar wake check (design: sidecar members), or a test hook
scheduleLeg func(ctx context.Context) // the schedule converge (spec §6.2), or a test hook; nil = not armed
+ convergeLeg func() // finishes adoptions another process performed (convergeAdopted); nil = no activator armed
systemLeg *systemLeg // the platform self-update leg (spec §4); nil = not armed
lastSerial int64
seeded bool
@@ -1595,6 +1596,46 @@ func (d *fastDeps) observeRestart(restart func(string) error) func(string) error
}
}
+// convergeAdopted finishes the adoptions this process delegated or never saw.
+//
+// The agent is sandboxed (carlos-edge.service: ProtectSystem=strict, no /opt
+// write), so on a real box a new version is staged, pointed live and stamped
+// by a ROOT pass — carlos-stage.service on this loop's own request, or the
+// hourly carlos-adopt.service. That pass cannot cycle an exec child
+// (rootRestartRoute: the children are this process's), and once it has run,
+// this process's own pass finds the host up to date and never reaches the
+// restart callback RestartHibernating exists to hand it. The warm child was
+// left on the old build until the activator sweep's ConvergeGrace ran out:
+// 110-200s per deploy on the flagship, past `carlos deploy`'s own watch
+// (discussion #231).
+//
+// So, for each warm instance the activator reports drifted, cycle it once the
+// route is STAMPED with the version its symlink now names. The stamp is the
+// adopting pass saying it is finished with the host (adoptHost writes it last,
+// after its restart and any rollback); a drift the stamp does not yet agree
+// with is an adoption still in flight, and this leg waits a tick for it rather
+// than race it. execBounceable keeps the usual line between the two halves: a
+// route whose process the root pass's systemctl arms can reach is never this
+// leg's to cycle.
+//
+// Never wakes anything — ConvergeAdopted is a stop, and a sleeper execs the
+// new build at its own next wake. drifted returning nothing is the ordinary
+// tick and costs no registry read. Returns how many it cycled.
+func convergeAdopted(drifted func() map[string]string, get func(host string) (*registry.Route, error), converge func(host, ver string) bool) int {
+ n := 0
+ for host, live := range drifted() {
+ rt, err := get(host)
+ if err != nil || rt == nil || rt.Version != live || !execBounceable(*rt) {
+ continue
+ }
+ if converge(host, live) {
+ n++
+ log.Printf("agent: %s was adopted onto %s by another pass — cycled its warm instance onto it", host, live)
+ }
+ }
+ return n
+}
+
// fastTick is one fast-converge tick — two independent jobs (spec §3-§4):
//
// 1. Serial check: one ReadSerial. The FIRST successful read only seeds
@@ -1624,6 +1665,9 @@ func (d *fastDeps) observeRestart(restart func(string) error) func(string) error
// exactly those hosts — an empty sweep costs zero store reads. After
// the sweep, members still unstaged back off; staged members clear
// their backoff state.
+//
+// Between the two, every tick, the converge leg (convergeAdopted) cycles any
+// warm exec child that another process's adoption left on its old build.
func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
doc, _, err := release.ReadSerial(ctx, d.store)
switch {
@@ -1685,6 +1729,16 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
d.systemLeg.tick(ctx, now)
}
+ // The converge leg, on EVERY tick and before the sweep's early return. It
+ // cannot ride the serial switch above: the root pass that finishes a
+ // delegated adoption does so seconds AFTER the bump this loop acted on,
+ // and nothing bumps the serial again to mark it. Nor can it ride the
+ // sweep: an already-staged edge route is never a sweep member. It reads
+ // no store, and no registry row unless a warm child has actually drifted.
+ if d.convergeLeg != nil {
+ d.convergeLeg()
+ }
+
members := d.fastMembers(now)
if len(members) == 0 {
return // zero store reads: no PassFilter, no bake-clock write
@@ -1782,6 +1836,9 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
}
d.restart = restart
d.passOpts = opts
+ if svc != nil {
+ d.convergeLeg = func() { convergeAdopted(svc.Drifted, reg.Get, svc.ConvergeAdopted) }
+ }
// The serial-triggered full pass set, in spec order: mint routes
// (reconcile), then reconcile aliases, then stage them (adopt), then deliver
// their config. Each error is one log line, never fatal — the hourly floors
diff --git a/internal/host/converge_adopted_test.go b/internal/host/converge_adopted_test.go
new file mode 100644
index 000000000..9163a9a1a
--- /dev/null
+++ b/internal/host/converge_adopted_test.go
@@ -0,0 +1,175 @@
+package host
+
+import (
+ "context"
+ "reflect"
+ "runtime"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The stamp is the whole licence for skipping the sweep's grace: it is written
+// only after the adopting pass is finished with the host. A drift whose route
+// is still stamped with the old version is an adoption in flight, and one with
+// no row at all is not a route this box adopts.
+func TestConvergeAdoptedCyclesOnlyWhenTheStampAgrees(t *testing.T) {
+ rows := map[string]*registry.Route{
+ "done.test": {Host: "done.test", Version: "v2"},
+ "inflight.test": {Host: "inflight.test", Version: "v1"},
+ "named.test": {Host: "named.test", Version: "v2", Unit: "console.service"},
+ "unitback.test": {Host: "unitback.test", Version: "v2", Backing: registry.BackingUnit},
+ }
+ drifted := func() map[string]string {
+ return map[string]string{
+ "done.test": "v2", "inflight.test": "v2", "named.test": "v2",
+ "unitback.test": "v2", "norow.test": "v2",
+ }
+ }
+ var cycled []string
+ n := convergeAdopted(drifted,
+ func(h string) (*registry.Route, error) { return rows[h], nil },
+ func(h, ver string) bool { cycled = append(cycled, h+"@"+ver); return true })
+ if want := []string{"done.test@v2"}; !reflect.DeepEqual(cycled, want) {
+ t.Fatalf("cycled = %v, want %v", cycled, want)
+ }
+ if n != 1 {
+ t.Fatalf("convergeAdopted = %d, want 1", n)
+ }
+}
+
+// The ordinary tick, where nothing has shipped, must cost no registry read:
+// this leg runs every two seconds on every box.
+func TestConvergeAdoptedReadsNoRouteWhenNothingDrifted(t *testing.T) {
+ n := convergeAdopted(func() map[string]string { return nil },
+ func(h string) (*registry.Route, error) {
+ t.Fatalf("read the route for %s with nothing drifted", h)
+ return nil, nil
+ },
+ func(h, ver string) bool {
+ t.Fatalf("cycled %s with nothing drifted", h)
+ return false
+ })
+ if n != 0 {
+ t.Fatalf("convergeAdopted = %d, want 0", n)
+ }
+}
+
+// The leg runs on EVERY tick, not only on a serial move: the root pass that
+// finishes the adoption does so seconds after the bump the agent acted on, and
+// nothing bumps the serial again to mark it. It also runs on a tick with no
+// sweep members, which is the tick an already-staged edge route always gets.
+func TestFastTickRunsTheConvergeLegEveryTick(t *testing.T) {
+ env := newFastFixture(t)
+ d := env.deps(nil)
+ var legs int
+ d.convergeLeg = func() { legs++ }
+ now := time.Unix(1754640000, 0)
+ ctx := context.Background()
+ fastTick(ctx, d, now)
+ fastTick(ctx, d, now.Add(2*time.Second))
+ if legs != 2 {
+ t.Fatalf("the converge leg ran %d times over two quiet ticks, want 2", legs)
+ }
+ d.convergeLeg = nil // a box with no activator armed
+ fastTick(ctx, d, now.Add(4*time.Second))
+}
+
+// The production sequence behind discussion #231, with the real pass on both
+// sides. The agent cannot write the version store, so the ROOT pass adopts a
+// new release for a warm exec instance: it stages, repoints and stamps, and —
+// having no handle on the agent's children — cycles nothing. The agent's own
+// pass then finds the host up to date and never reaches its restart callback.
+// Before the converge leg, that left the child on the old build until the
+// activator sweep's 90s grace ran out.
+func TestRootAdoptionOfAWarmExecInstanceIsFinishedByTheAgent(t *testing.T) {
+ env := newFastFixture(t)
+ ctx := context.Background()
+ plat := runtime.GOOS + "-" + runtime.GOARCH
+ host := "warm.test"
+ ship := func(ver string) {
+ t.Helper()
+ m := release.Manifest{
+ App: "fastapp", Version: ver, Kind: "binary",
+ Artifacts: map[string]release.Artifact{plat: {}},
+ }
+ if err := release.Ship(ctx, env.store, "bab", m, map[string][]byte{plat: []byte("build " + ver)}); err != nil {
+ t.Fatalf("Ship %s: %v", ver, err)
+ }
+ if err := release.Promote(ctx, env.store, env.priv, "bab", "fastapp", "edge", ver); err != nil {
+ t.Fatalf("Promote %s: %v", ver, err)
+ }
+ }
+ // rootRestartRoute with systemctl stubbed out: same dispatch, bounce nil.
+ rootPass := func() {
+ t.Helper()
+ restart := makeRestartRoute(env.reg, func(string) {}, func(string) error { return nil }, nil)
+ if err := release.Pass(ctx, env.reg, env.store, env.pub, env.optDir, restart, release.BakeDir(env.dataDir)); err != nil {
+ t.Fatalf("root pass: %v", err)
+ }
+ }
+ if err := env.reg.Upsert(registry.Route{
+ Host: host, Kind: "instance", App: "fastapp", Channel: "edge",
+ Status: "ready", Hibernate: true,
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ ship("v1")
+ rootPass()
+ spawned := "v1" // the awake child's build: frozen at exec, as the activator records it
+
+ ship("v2")
+ rootPass()
+ if v, err := release.LiveVersion(env.optDir, host); err != nil || v != "v2" {
+ t.Fatalf("live = %q, %v after the root pass; want v2", v, err)
+ }
+ if rt, err := env.reg.Get(host); err != nil || rt == nil || rt.Version != "v2" {
+ t.Fatalf("route = %+v, %v after the root pass; want it stamped v2", rt, err)
+ }
+
+ // The agent's tick. Its pass has a bounce to offer and never gets to use
+ // it: the host is already up to date.
+ var bounced, cycled []string
+ restart := makeRestartRoute(env.reg, func(string) {}, func(string) error { return nil },
+ func(h string) bool { bounced = append(bounced, h); return true })
+ d := env.deps(nil)
+ d.restart = restart
+ d.passOpts = fastPassOpts(env.dataDir, func(string, string, error) {}, restart, nil)
+ d.fullPass = func(ctx context.Context) {
+ if err := release.Pass(ctx, d.reg, d.store, d.pub, d.optDir, d.restart, d.passOpts...); err != nil {
+ t.Fatalf("agent pass: %v", err)
+ }
+ }
+ d.convergeLeg = func() {
+ convergeAdopted(
+ func() map[string]string {
+ live, _ := release.LiveVersion(env.optDir, host)
+ if live == "" || live == spawned {
+ return nil
+ }
+ return map[string]string{host: live}
+ },
+ env.reg.Get,
+ func(h, ver string) bool { cycled = append(cycled, h+"@"+ver); spawned = ver; return true })
+ }
+ now := time.Unix(1754640000, 0)
+ fastTick(ctx, d, now) // seeds the serial
+ if err := release.BumpSerial(ctx, env.store, now); err != nil {
+ t.Fatalf("BumpSerial: %v", err)
+ }
+ fastTick(ctx, d, now.Add(10*time.Second)) // the serial move: full pass, then the leg
+
+ if len(bounced) != 0 {
+ t.Fatalf("the agent's pass bounced %v — it found the host already adopted, so this test no longer models the gap", bounced)
+ }
+ if want := []string{host + "@v2"}; !reflect.DeepEqual(cycled, want) {
+ t.Fatalf("cycled = %v, want %v: the warm child is still on v1 under a symlink and a stamp that both say v2", cycled, want)
+ }
+ fastTick(ctx, d, now.Add(12*time.Second))
+ if len(cycled) != 1 {
+ t.Fatalf("cycled again on the next tick: %v", cycled)
+ }
+}