amadan

carlos / platform Public

fix-adopted-exec-restart

Squashed from fix-adopted-exec-restart into main.
claude 3a86f34855599a2ac30bff977491b8998f39e6ba parent 8d8e4ba
5 files changed, +855 −3
  • AGENTS.md +6 −1
  • internal/activator/activator.go +119 −0
  • internal/activator/converge_adopted_test.go +203 −0
  • internal/host/agent.go +111 −2
  • internal/host/converge_adopted_test.go +416 −0

CI cancelled — superseded by a newer push — run details

diff --git a/AGENTS.md b/AGENTS.md
index cdca5b6a0..455dd458c 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -344,7 +344,12 @@ anything.
poll it at 2s and run the full pass set. Direct-bucket promotes (pinfra
scripts) don't bump but canary/* channels are name-scoped fast anyway.
Blocked stages relay to a root pass (`carlos-stage.path` — must be armed;
- update-platform-binary.sh does it). `carlos deploy` is the member one-command
+ update-platform-binary.sh does it). That root pass stages, repoints and
+ stamps in ~8s but cannot cycle the agent's exec children, so the agent's
+ 2s tick finishes the job: a warm instance whose route is stamped with the
+ version its symlink names is cycled at once (`convergeAdopted`, #231). The
+ activator sweep's 90s grace is only the backstop. `carlos deploy` is the
+ member one-command
form: ship → promote → wait until `X-Carlos-Version` on the URL reports the
shipped build over 2xx. Verify deploys with that header (`curl -sI
--resolve`), never a cached answer. `-I` is fine for the VERSION header, but
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 16ef6357a..15e9b11f5 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1381,6 +1381,12 @@ func (a *Service) Bounce(host string) bool {
if e == nil || e.state != actAwake {
return false
}
+ return a.bounceLocked(e)
+}
+
+// bounceLocked is Bounce's stop for an entry the caller has found awake,
+// under a.mu.
+func (a *Service) bounceLocked(e *actEntry) bool {
msg := "instance restarting (carlos restart)"
if e.worker {
msg = "sidecar restarting (carlos restart)"
@@ -1390,6 +1396,119 @@ func (a *Service) Bounce(host string) bool {
return true
}
+// BounceAdopted is Bounce for an ADOPT pass's restart callback, with one
+// thing Bounce cannot know taken into account: the callback fires after the
+// pass has repointed live/<host>, so a child already exec'd from the build the
+// symlink names has nothing to pick up, and stopping it buys a drain and a
+// wake for no change of binary.
+//
+// That child exists whenever something else finished the adoption first. A
+// root pass adopts, ConvergeAdopted cycles the instance, a request wakes it
+// onto the new build — and a slower pass that read the old symlink before all
+// of that reaches its own restart afterwards. Bounce would stop the
+// replacement; this leaves it alone.
+//
+// Everything else is Bounce: same no-op on a host that is not awake, same
+// event, same stop, and it applies to a worker exactly as Bounce does. An
+// unknown spawn version or an unreadable symlink bounces — with nothing to
+// compare, the safe reading of "adopted" is "the child may be stale".
+//
+// `carlos restart` and config delivery keep the unconditional Bounce: there
+// the request IS the reason, whatever build is running.
+func (a *Service) BounceAdopted(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.state != actAwake {
+ return false
+ }
+ if e.spawnVersion != "" {
+ if cur, err := a.liveVer(host); err == nil && cur == e.spawnVersion {
+ return false
+ }
+ }
+ return a.bounceLocked(e)
+}
+
+// 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..31f5781e9
--- /dev/null
+++ b/internal/activator/converge_adopted_test.go
@@ -0,0 +1,203 @@
+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")
+ }
+}
+
+// An adopt pass's restart callback fires after it has repointed live/<host>,
+// so by then the symlink names the version being adopted. A child ALREADY
+// exec'd from that version has nothing to pick up: stopping it is one more
+// drain and wake for no change of build. That child exists whenever something
+// else got there first — ConvergeAdopted finishing a root pass's adoption and
+// a request re-waking the instance, while a slower pass was still mid-host.
+func TestBounceAdoptedSparesAChildAlreadyOnTheLiveBuild(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ var ver atomic.Value
+ ver.Store("v2")
+ 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 a.BounceAdopted(host) {
+ t.Fatal("BounceAdopted stopped a child that already runs the build the symlink names")
+ }
+ if a.entryCount() != 1 {
+ t.Fatal("the converged child was stopped")
+ }
+
+ // The ordinary case is unchanged: the symlink has moved under the child.
+ ver.Store("v3")
+ if !a.BounceAdopted(host) {
+ t.Fatal("BounceAdopted spared a child serving v2 under a symlink that names v3")
+ }
+ waitFor(t, "the adoption bounce", func() bool { return a.entryCount() == 0 })
+ if a.BounceAdopted(host) {
+ t.Fatal("BounceAdopted acted on a host that is asleep")
+ }
+}
+
+// With nothing to compare it bounces, exactly as Bounce always has: a backend
+// that cannot name its build, or a symlink that cannot be read, must not turn
+// an adoption into a silent no-op over a child still on the old binary.
+func TestBounceAdoptedBouncesWhenTheBuildIsUnknown(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 !a.BounceAdopted(host) {
+ t.Fatal("BounceAdopted refused a child whose spawn version is unknown")
+ }
+ waitFor(t, "the bounce", func() bool { return a.entryCount() == 0 })
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 255b399d9..f4eafc19f 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -805,7 +805,7 @@ func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps)
return
}
optDir := release.OptDir()
- restart := makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, bounceFn(svc))
+ restart := makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, adoptBounceFn(svc))
log.Printf("agent: adopt loop armed (hourly + jitter)")
// First pass shortly after boot (Eleven's OnBootSec=5min, compressed —
// the edge is already up by the time this fires), then hourly + jitter.
@@ -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,87 @@ 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.
+//
+// At most limit instances are cycled per call, in host order. One pass can
+// adopt every warm instance of an app at once — a root pass repoints them all
+// inside a few seconds — and stopping them together would drain, checkpoint
+// and later restore them together too: every host of the app blips at the
+// same moment. The activator sweep spends the same care (one convergence
+// cycle per sweep); this leg keeps it at its own cadence, so N instances
+// converge over N ticks rather than in one. Only a cycle that actually
+// started counts against the limit — a drifted host this leg must wait on
+// costs the others nothing.
+//
+// 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, limit int) int {
+ stale := drifted()
+ if len(stale) == 0 {
+ return 0
+ }
+ hosts := make([]string, 0, len(stale))
+ for host := range stale {
+ hosts = append(hosts, host)
+ }
+ sort.Strings(hosts)
+ n := 0
+ for _, host := range hosts {
+ if n >= limit {
+ break
+ }
+ live := stale[host]
+ 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
+}
+
+// adoptedCyclesPerTick is convergeAdopted's limit on the fast loop: one warm
+// instance per 2s tick. A single-instance app — the common case — converges
+// on the first tick after its adoption is stamped; an app with a dozen warm
+// instances takes under half a minute, one blip at a time.
+const adoptedCyclesPerTick = 1
+
+// convergeLegFor builds the fast loop's converge leg over the box's
+// activator. nil for a box with none armed: there are no exec children to
+// cycle, and fastTick skips a nil leg.
+func convergeLegFor(svc *activator.Service, reg *registry.Registry) func() {
+ if svc == nil {
+ return nil
+ }
+ return func() {
+ convergeAdopted(svc.Drifted, reg.Get, svc.ConvergeAdopted, adoptedCyclesPerTick)
+ }
+}
+
// 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 +1706,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 +1770,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
@@ -1749,7 +1844,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
// sweep instead of the sweep re-cycling the service every tick. The
// remaining fields are filled in right below.
d := &fastDeps{reg: reg, store: store, pub: pub, optDir: optDir}
- restart := d.observeRestart(makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, bounceFn(svc)))
+ restart := d.observeRestart(makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, adoptBounceFn(svc)))
// Built once, used by both pass shapes below — see fastPassOpts for the
// DO-NOT list and why the stage requester is per-loop, not per-tick.
opts := fastPassOpts(dd, stageRequester(dd), restart, makeStartRoute(reg, systemctlStartUnit))
@@ -1782,6 +1877,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
}
d.restart = restart
d.passOpts = opts
+ d.convergeLeg = convergeLegFor(svc, reg)
// 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
@@ -3045,6 +3141,19 @@ func bounceFn(svc *activator.Service) func(host string) bool {
return svc.Bounce
}
+// adoptBounceFn is bounceFn for an ADOPT pass's restart dispatch
+// (makeRestartRoute on adoptLoop and fastConvergeLoop): the same bounce,
+// except that a child already exec'd from the build live/<host> names is left
+// running — see activator.BounceAdopted for the interleaving that produces
+// one. `carlos restart` keeps bounceFn: a restart request cycles whatever is
+// running.
+func adoptBounceFn(svc *activator.Service) func(host string) bool {
+ if svc == nil {
+ return nil
+ }
+ return svc.BounceAdopted
+}
+
// wakeFn adapts an activator to cycleWorkers' wake parameter, nil-safe like
// bounceFn: a box with no activator armed gets a nil callback, which
// cycleWorkers reads as "nothing to wake" rather than panicking on a nil
diff --git a/internal/host/converge_adopted_test.go b/internal/host/converge_adopted_test.go
new file mode 100644
index 000000000..6934b1c88
--- /dev/null
+++ b/internal/host/converge_adopted_test.go
@@ -0,0 +1,416 @@
+package host
+
+import (
+ "context"
+ "net/http"
+ "net/http/httptest"
+ "reflect"
+ "runtime"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "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 }, 10)
+ 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
+ }, 10)
+ 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 },
+ adoptedCyclesPerTick)
+ }
+ 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)
+ }
+}
+
+// One pass can adopt every warm instance of an app inside a few seconds.
+// Stopping them all on one tick would blip every host of the app at the same
+// moment; the leg spends its budget in host order and leaves the rest for the
+// ticks that follow. A drifted host it has to wait on costs the others nothing.
+func TestConvergeAdoptedCyclesOnePerTickInHostOrder(t *testing.T) {
+ spawned := map[string]string{"a.test": "v1", "b.test": "v1", "c.test": "v1", "wait.test": "v1"}
+ stamp := map[string]string{"a.test": "v2", "b.test": "v2", "c.test": "v2", "wait.test": "v1"}
+ drifted := func() map[string]string {
+ out := map[string]string{}
+ for h, v := range spawned {
+ if v != "v2" {
+ out[h] = "v2"
+ }
+ }
+ return out
+ }
+ get := func(h string) (*registry.Route, error) { return &registry.Route{Host: h, Version: stamp[h]}, nil }
+ var cycled []string
+ converge := func(h, ver string) bool { cycled = append(cycled, h); spawned[h] = ver; return true }
+
+ for tick := 1; tick <= 4; tick++ {
+ if n := convergeAdopted(drifted, get, converge, adoptedCyclesPerTick); tick <= 3 && n != 1 {
+ t.Fatalf("tick %d cycled %d, want 1", tick, n)
+ }
+ }
+ if want := []string{"a.test", "b.test", "c.test"}; !reflect.DeepEqual(cycled, want) {
+ t.Fatalf("cycled = %v, want %v: one per tick, in host order, never the host whose stamp does not agree", cycled, want)
+ }
+}
+
+// versionedBackend is activator.Backend with the one thing convergence needs
+// and fakeWakeBackend does not model: a child that knows which build it was
+// exec'd from, read off the SAME live symlink the release pass writes. It
+// binds no socket — the caller supplies the proxy — so it runs anywhere.
+type versionedBackend struct {
+ fakeWakeBackend
+ optDir string
+
+ mu sync.Mutex
+ starts []string // the build each spawn exec'd, in order
+}
+
+type flagProc struct {
+ mu sync.Mutex
+ stopped bool
+}
+
+func (p *flagProc) Stop(context.Context) error {
+ p.mu.Lock()
+ p.stopped = true
+ p.mu.Unlock()
+ return nil
+}
+
+func (p *flagProc) Exited() bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.stopped
+}
+
+func (b *versionedBackend) LiveVersion(host string) (string, error) {
+ return release.LiveVersion(b.optDir, host)
+}
+
+func (b *versionedBackend) Start(_ context.Context, rt registry.Route) (activator.Proc, string, error) {
+ ver, err := release.LiveVersion(b.optDir, rt.Host)
+ if err != nil {
+ return nil, "", err
+ }
+ b.mu.Lock()
+ b.starts = append(b.starts, ver)
+ b.mu.Unlock()
+ return &flagProc{}, ver, nil
+}
+
+func (b *versionedBackend) started() []string {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return append([]string(nil), b.starts...)
+}
+
+// The whole of discussion #231 with nothing stood in for: the real release
+// pass as root runs it, a real activator owning the child, and the leg exactly
+// as fastConvergeLoop builds it (convergeLegFor). The sweeper is never
+// started, so the ONLY thing that can cycle the instance is the leg — a 90s
+// grace cannot rescue this test.
+func TestConvergeLegCyclesARootAdoptedInstanceOnARealActivator(t *testing.T) {
+ env := newFastFixture(t)
+ ctx := context.Background()
+ plat := runtime.GOOS + "-" + runtime.GOARCH
+ host := "warm.test"
+ rt := registry.Route{
+ Host: host, Kind: "instance", App: "fastapp", Channel: "edge",
+ Status: "ready", Hibernate: true,
+ }
+ if err := env.reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ adopt := 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)
+ }
+ // The root pass: rootRestartRoute's dispatch (bounce nil), systemctl stubbed.
+ 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 for %s: %v", ver, err)
+ }
+ }
+ backend := &versionedBackend{optDir: env.optDir}
+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, backend, env.reg)
+ t.Cleanup(func() {
+ sctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ svc.Shutdown(sctx)
+ })
+ request := func() {
+ t.Helper()
+ rec := httptest.NewRecorder()
+ proxy := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })
+ if !svc.ServeHTTP(rec, httptest.NewRequest("GET", "http://"+host+"/", nil), rt, proxy) {
+ t.Fatal("the activator did not take the request")
+ }
+ if rec.Code != http.StatusNoContent {
+ t.Fatalf("request = %d, want 204", rec.Code)
+ }
+ }
+ leg := convergeLegFor(svc, env.reg)
+ if leg == nil {
+ t.Fatal("convergeLegFor returned no leg for a box with an activator")
+ }
+
+ // "Was it stopped?" is asked of the spawn count, after a request: Owns
+ // stays true while an entry drains, so it cannot tell a spared child from
+ // one being stopped. A request against a stopped instance spawns again.
+ adopt("v1")
+ request() // wakes the instance onto v1
+ leg()
+ request()
+ if got, want := backend.started(), []string{"v1"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v: the leg stopped an instance whose build is the adopted one", got, want)
+ }
+
+ adopt("v2") // root: staged, repointed, stamped — and nothing cycled
+ if !svc.Owns(host) {
+ t.Fatal("the root pass stopped the agent's child — this test no longer models the gap")
+ }
+ leg()
+ deadline := time.Now().Add(10 * time.Second)
+ for svc.Owns(host) {
+ if time.Now().After(deadline) {
+ t.Fatal("the warm instance is still running v1 after the leg: it was adopted onto v2 and nothing cycled it")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+
+ request() // the next request wakes it onto v2
+ if got, want := backend.started(), []string{"v1", "v2"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v", got, want)
+ }
+ leg()
+ request()
+ if got, want := backend.started(), []string{"v1", "v2"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v: the leg cycled the instance again after it had converged", got, want)
+ }
+}
+
+func TestConvergeLegForABoxWithNoActivator(t *testing.T) {
+ if convergeLegFor(nil, nil) != nil {
+ t.Fatal("convergeLegFor built a leg with no activator to ask")
+ }
+}
+
+// The adopt passes' restart dispatch must be built on adoptBounceFn, not
+// bounceFn. With the converge leg in place an instance can already be on the
+// new build when a slower pass reaches its own restart; the plain Bounce would
+// stop that replacement for nothing. A stale child still gets its cycle.
+func TestAdoptRestartDispatchSparesAConvergedChild(t *testing.T) {
+ env := newFastFixture(t)
+ host := "warm.test"
+ rt := registry.Route{
+ Host: host, Kind: "instance", App: "fastapp", Channel: "edge",
+ Status: "ready", Hibernate: true,
+ }
+ if err := env.reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ env.stageLive(t, host, "v1")
+ backend := &versionedBackend{optDir: env.optDir}
+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, backend, env.reg)
+ t.Cleanup(func() {
+ sctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ svc.Shutdown(sctx)
+ })
+ wake := func() {
+ t.Helper()
+ rec := httptest.NewRecorder()
+ proxy := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })
+ if !svc.ServeHTTP(rec, httptest.NewRequest("GET", "http://"+host+"/", nil), rt, proxy) || rec.Code != http.StatusNoContent {
+ t.Fatalf("wake request = %d", rec.Code)
+ }
+ }
+ restart := makeRestartRoute(env.reg, func(string) {}, func(u string) error {
+ t.Fatalf("named-unit arm fired for %q on a route that names no unit", u)
+ return nil
+ }, adoptBounceFn(svc))
+
+ // Asked of the spawn count, after a request — see the note in
+ // TestConvergeLegCyclesARootAdoptedInstanceOnARealActivator.
+ wake() // exec'd from v1, and live/<host> names v1
+ if err := restart(host); err != nil {
+ t.Fatalf("restart: %v", err)
+ }
+ wake()
+ if got, want := backend.started(), []string{"v1"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v: the adopt dispatch stopped a child already running the build live/<host> names", got, want)
+ }
+
+ env.stageLive(t, host, "v2") // the symlink moves under the running child
+ if err := restart(host); err != nil {
+ t.Fatalf("restart: %v", err)
+ }
+ wake()
+ if got, want := backend.started(), []string{"v1", "v2"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v: the adopt dispatch left a stale child running", got, want)
+ }
+}