carlos / platform Public

host, activator: review fixes — one cycle per tick, and an adoption bounce that spares a converged child

Astra's review of the converge leg (accept with fixes):

- The leg cycled every drifted instance of an app on one tick, where the
  sweep it stands in for cycles one per sweep. It now cycles one per
  tick, in host order.
- With the leg in place a slower adopt pass could reach its own restart
  after the instance had already been cycled and re-woken on the new
  build, and Bounce would stop the replacement. Adopt passes now use
  BounceAdopted, which leaves a child alone when it already runs the
  build live/<host> names. carlos restart keeps the unconditional Bounce.
- The regression test stood in for the activator. A second one runs the
  real pass, a real activator and the leg as fastConvergeLoop builds it.

Agent notes

Each new test was checked against a broken variant: a leg that cycles nothing, no per-tick limit, and a BounceAdopted that never spares.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 2 weeks ago dbf10da07a254b1e55e8fadef40e60dd7e78fcff parent d40e40a

4 files changed, +339 −10

internal/activator/activator.go +40 −0
internal/activator/converge_adopted_test.go +57 −0
internal/host/agent.go +59 −7
internal/host/converge_adopted_test.go +183 −3

CI passed — run details

diff --git a/internal/activator/activator.go b/internal/activator/activator.go

index 192f7b00b..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,40 @@ 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

diff --git a/internal/activator/converge_adopted_test.go b/internal/activator/converge_adopted_test.go

index 547841fe6..31f5781e9 100644

--- a/internal/activator/converge_adopted_test.go

+++ b/internal/activator/converge_adopted_test.go

@@ -144,3 +144,60 @@ func TestConvergeAdoptedLeavesWorkersToTheSweep(t *testing.T) {

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 4c6175c89..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.

@@ -1618,12 +1618,35 @@ func (d *fastDeps) observeRestart(restart func(string) error) func(string) error

// 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) int {

+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, live := range drifted() {

+ 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

@@ -1636,6 +1659,24 @@ func convergeAdopted(drifted func() map[string]string, get func(host string) (*r

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

@@ -1803,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))

@@ -1836,9 +1877,7 @@ 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) }

- }

+ 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

@@ -3102,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

index 9163a9a1a..f20a51967 100644

--- a/internal/host/converge_adopted_test.go

+++ b/internal/host/converge_adopted_test.go

@@ -2,11 +2,15 @@ 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"

)

@@ -31,7 +35,7 @@ func TestConvergeAdoptedCyclesOnlyWhenTheStampAgrees(t *testing.T) {

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 })

+ 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)

}

@@ -51,7 +55,7 @@ func TestConvergeAdoptedReadsNoRouteWhenNothingDrifted(t *testing.T) {

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)

}

@@ -153,7 +157,8 @@ func TestRootAdoptionOfAWarmExecInstanceIsFinishedByTheAgent(t *testing.T) {

return map[string]string{host: live}

env.reg.Get,

- func(h, ver string) bool { cycled = append(cycled, h+"@"+ver); spawned = ver; return true })

+ 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

@@ -173,3 +178,178 @@ func TestRootAdoptionOfAWarmExecInstanceIsFinishedByTheAgent(t *testing.T) {

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")

+ }

+ adopt("v1")

+ request() // wakes the instance onto v1

+ leg()

+ if !svc.Owns(host) {

+ t.Fatal("the leg stopped an instance whose build is the adopted one")

+ }

+ 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()

+ if !svc.Owns(host) {

+ t.Fatal("the leg cycled the instance again after it had converged")

+ }

+func TestConvergeLegForABoxWithNoActivator(t *testing.T) {

+ if convergeLegFor(nil, nil) != nil {

+ t.Fatal("convergeLegFor built a leg with no activator to ask")

+ }