amadan
fix-converge-own-loop
Squashed from fix-converge-own-loop into main.
4 files changed,
+245
−47
internal/activator/activator.go+28 −5internal/activator/converge_adopted_test.go+43 −0internal/host/agent.go+72 −22internal/host/converge_adopted_test.go+102 −20
CI passed — run details
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 0a719ed03..a1afa63f4 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -341,9 +341,14 @@ const ( |
| ) |
| type actEntry struct { |
| - host string |
| - route registry.Route |
| - state actState |
| + // bounceAfterWake is a Bounce that arrived while this INSTANCE was still |
| + // waking. The wake had already read its config and execed its build, so |
| + // the bounce's reason has not reached the child; wake() spends it the |
| + // moment the wake finishes, before any waiting request is admitted. |
| + bounceAfterWake bool |
| + host string |
| + route registry.Route |
| + state actState |
| ready chan struct{} // waking: closed when the wake resolves; then check err |
| err error // waking: the wake's outcome |
| @@ -986,8 +991,16 @@ func (a *Service) wake(e *actEntry) { |
| e.staleSince = time.Time{} |
| if a.fences[e.host] != 0 { |
| a.beginStopLocked(e) |
| - } |
| - if a.OnWake != nil && a.fences[e.host] == 0 { |
| + } else if e.bounceAfterWake { |
| + // A config delivery or `carlos restart` asked for while this wake |
| + // was under way (Bounce). Same spot and same reason as the fence |
| + // just above: the state flips to stopping before e.ready closes, |
| + // so a request waiting on the wake is never proxied to this child |
| + // — it waits out the stop and wakes a fresh one. |
| + e.bounceAfterWake = false |
| + a.bounceLocked(e) |
| + } |
| + if a.OnWake != nil && e.state == actAwake { |
| // Fire-and-forget, off this goroutine: e.route is read once here, |
| // before this goroutine can be reused for anything else — the |
| // hook must never share e.host's own goroutine's fate (a wedged |
| @@ -1402,6 +1415,16 @@ func (a *Service) Bounce(host string) bool { |
| a.mu.Lock() |
| defer a.mu.Unlock() |
| e := a.entries[host] |
| + if e != nil && e.state == actWaking && !e.worker { |
| + // Not dropped: held for the end of the wake (bounceAfterWake). Still |
| + // reported as no stop, because none has happened — the callers that |
| + // count stops, or fall through to systemctl on false, see what they |
| + // always saw. Workers keep the old rule: their restart pairs a stop |
| + // with its own WakeWorker only on true (host/restart.go cycleWorkers), |
| + // and a stop held for later would park one with nothing to start it. |
| + e.bounceAfterWake = true |
| + return false |
| + } |
| if e == nil || e.state != actAwake { |
| return false |
| } |
diff --git a/internal/activator/converge_adopted_test.go b/internal/activator/converge_adopted_test.go| index 31f5781e9..be7ed7f55 100644 |
| --- a/internal/activator/converge_adopted_test.go |
| +++ b/internal/activator/converge_adopted_test.go |
| @@ -201,3 +201,46 @@ func TestBounceAdoptedBouncesWhenTheBuildIsUnknown(t *testing.T) { |
| } |
| waitFor(t, "the bounce", func() bool { return a.entryCount() == 0 }) |
| } |
| + |
| +// A bounce that lands while the instance is still waking is not dropped. The |
| +// wake already read its config and execed its build, so whatever the bounce |
| +// was for — new config, a `carlos restart` — has not reached the child, and |
| +// nothing will ask again: config delivery bounces only when the rendered |
| +// file changes. The bounce is held and spent the moment the wake finishes, |
| +// before any waiting request is let through, so every request lands on a |
| +// child started after the bounce was asked for. |
| +func TestBounceDuringAWakeIsAppliedWhenTheWakeFinishes(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + b.startGate = make(chan struct{}) |
| + host := "one.example.com" |
| + |
| + served := make(chan error, 1) |
| + go func() { |
| + release, err := a.beginRequest(context.Background(), testRoute(host)) |
| + if err == nil { |
| + release() |
| + } |
| + served <- err |
| + }() |
| + waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 1 }) |
| + |
| + if a.Bounce(host) { |
| + t.Fatal("Bounce reported a stop on an instance that is still waking") |
| + } |
| + close(b.startGate) |
| + |
| + if err := <-served; err != nil { |
| + t.Fatalf("the request waiting on the wake: %v", err) |
| + } |
| + // It was served by a SECOND child: the first was stopped as soon as its |
| + // wake finished, and the waiting request woke a fresh one. |
| + if got := b.startCount(); got != 2 { |
| + t.Fatalf("children started = %d, want 2: the bounce asked for during the wake was dropped", got) |
| + } |
| + if !b.procs[0].isStopped() { |
| + t.Fatal("the child from the wake the bounce landed in is still running") |
| + } |
| + if a.Bounce(host) != true { |
| + t.Fatal("an ordinary bounce of the awake replacement was refused") |
| + } |
| +} |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 9862b90c0..b38bf0ae4 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -1388,7 +1388,6 @@ 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 |
| @@ -1671,15 +1670,18 @@ 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. |
| +// adoptedCyclesPerTick is convergeAdopted's limit per run of its loop: one |
| +// warm instance every adoptedConvergeEvery. A single-instance app — the common |
| +// case — converges on the first run 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. |
| +// adoptedConvergeEvery is how often the converge leg looks for a finished |
| +// adoption to act on. |
| +const adoptedConvergeEvery = 2 * time.Second |
| + |
| +// convergeLegFor builds the converge leg over the box's activator. nil for a |
| +// box with none armed: there are no exec children to cycle. |
| func convergeLegFor(svc *activator.Service, reg *registry.Registry) func() { |
| if svc == nil { |
| return nil |
| @@ -1689,6 +1691,48 @@ func convergeLegFor(svc *activator.Service, reg *registry.Registry) func() { |
| } |
| } |
| +// runConvergeLoop runs the converge leg every `every` until stop closes (nil: |
| +// for the life of the process). |
| +// |
| +// Its own loop, not a leg of fastTick, and the reason is measured: on the |
| +// flagship's Ireland box a fast tick ran 25 to 50 seconds whenever the serial |
| +// moved (a full reconcile and adopt pass over ~180 routes), and console |
| +// activity across every tenant moves it about once a minute. Riding the tick, |
| +// a stamped adoption waited ~48s for its cycle (discussion #231, 2026-10-05). |
| +// The leg needs none of the tick's state: it reads the activator and the |
| +// registry, both safe to read beside the tick, and its one write is a stop |
| +// under the activator's own lock — where the tick's adopt bounce is |
| +// version-aware (BounceAdopted), so the two cannot stop one child twice for |
| +// one build. |
| +func runConvergeLoop(leg func(), every time.Duration, stop <-chan struct{}) { |
| + t := time.NewTicker(every) |
| + defer t.Stop() |
| + for { |
| + select { |
| + case <-stop: |
| + return |
| + case <-t.C: |
| + leg() |
| + } |
| + } |
| +} |
| + |
| +// slowLegAfter is when logSlow speaks: a fast-loop leg that takes longer than |
| +// this is worth a line, because the loop's promise is seconds. |
| +const slowLegAfter = 5 * time.Second |
| + |
| +// logSlow logs that what took longer than slowLegAfter, with detail. Meant |
| +// for a defer: the loop's latency was invisible in the journal until someone |
| +// reconstructed it from timestamps by hand (#231). |
| +func logSlow(what string, start time.Time, detail string) { |
| + if d := time.Since(start); d > slowLegAfter { |
| + if detail != "" { |
| + detail = " (" + detail + ")" |
| + } |
| + log.Printf("agent: fast converge: %s took %s%s", what, d.Round(100*time.Millisecond), detail) |
| + } |
| +} |
| + |
| // fastTick is one fast-converge tick — two independent jobs (spec §3-§4): |
| // |
| // 1. Serial check: one ReadSerial. The FIRST successful read only seeds |
| @@ -1718,9 +1762,6 @@ func convergeLegFor(svc *activator.Service, reg *registry.Registry) func() { |
| // 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 { |
| @@ -1782,20 +1823,11 @@ 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 |
| } |
| + defer logSlow("fast sweep", time.Now(), fmt.Sprintf("%d routes", len(members))) |
| member := map[string]bool{} |
| for _, h := range members { |
| member[h] = true |
| @@ -1889,7 +1921,9 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| } |
| d.restart = restart |
| d.passOpts = opts |
| - d.convergeLeg = convergeLegFor(svc, reg) |
| + if leg := convergeLegFor(svc, reg); leg != nil { |
| + go runConvergeLoop(leg, adoptedConvergeEvery, nil) |
| + } |
| // 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 |
| @@ -1898,11 +1932,24 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| // making it — and it is what lets a recovery promote through the console |
| // converge immediately even mid-hold. |
| d.fullPass = func(ctx context.Context) { |
| + // Each stage's time, logged together when the set is slow, so the |
| + // journal says WHICH stage held a promote back. |
| + var took []string |
| + t := time.Now() |
| + mark := func(name string) { |
| + took = append(took, fmt.Sprintf("%s %s", name, time.Since(t).Round(100*time.Millisecond))) |
| + t = time.Now() |
| + } |
| + defer func(start time.Time) { logSlow("serial pass stages", start, strings.Join(took, ", ")) }(time.Now()) |
| + |
| if err := reconcileSerialized(ctx, store, reg, provisionTickConfig(os.Getenv)); err != nil { |
| log.Printf("agent: fast converge reconcile: %v", err) |
| } |
| + mark("reconcile") |
| telemetryPass(ctx, store, reg) |
| + mark("telemetry") |
| runMovesPass(ctx, store) |
| + mark("moves") |
| // Folded once per pass and handed in, never re-read inside the |
| // loop: this leg runs on every serial bump (a seconds cadence), |
| // and the fold is a LoadBoxes plus a List plus one Get per edges |
| @@ -1914,13 +1961,16 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| if err := domains.ReconcileAliases(ctx, store, reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil { |
| log.Printf("agent: fast converge alias pass: %v", err) |
| } |
| + mark("aliases") |
| if err := release.Pass(ctx, reg, store, pub, optDir, restart, opts...); err != nil { |
| log.Printf("agent: fast converge adopt pass: %v", err) |
| } |
| + mark("adopt") |
| if priv != nil { |
| if err := syncExecSerialized(ctx, reg, store, priv, dd, svc); err != nil { |
| log.Printf("agent: fast converge config sync: %v", err) |
| } |
| + mark("config") |
| } |
| } |
| // The restart leg: this process's own children bounced here, everything |
diff --git a/internal/host/converge_adopted_test.go b/internal/host/converge_adopted_test.go| index 6934b1c88..ea527cd27 100644 |
| --- a/internal/host/converge_adopted_test.go |
| +++ b/internal/host/converge_adopted_test.go |
| @@ -1,11 +1,14 @@ |
| package host |
| import ( |
| + "bytes" |
| "context" |
| + "log" |
| "net/http" |
| "net/http/httptest" |
| "reflect" |
| "runtime" |
| + "strings" |
| "sync" |
| "testing" |
| "time" |
| @@ -61,24 +64,31 @@ func TestConvergeAdoptedReadsNoRouteWhenNothingDrifted(t *testing.T) { |
| } |
| } |
| -// 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) |
| +// The leg runs on its own loop, never waiting on the fast tick: a tick that |
| +// takes a full reconcile and adopt pass can run tens of seconds, and a |
| +// stamped adoption must not wait behind it. Here the loop keeps calling the |
| +// leg on its own cadence, and stops when told. |
| +func TestConvergeLoopRunsTheLegOnItsOwnCadence(t *testing.T) { |
| + calls := make(chan struct{}, 100) |
| + stop := make(chan struct{}) |
| + done := make(chan struct{}) |
| + go func() { |
| + runConvergeLoop(func() { calls <- struct{}{} }, 5*time.Millisecond, stop) |
| + close(done) |
| + }() |
| + for i := 0; i < 3; i++ { |
| + select { |
| + case <-calls: |
| + case <-time.After(10 * time.Second): |
| + t.Fatalf("the loop ran the leg %d times, want it to keep running", i) |
| + } |
| + } |
| + close(stop) |
| + select { |
| + case <-done: |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("the loop did not stop") |
| } |
| - 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 |
| @@ -147,7 +157,7 @@ func TestRootAdoptionOfAWarmExecInstanceIsFinishedByTheAgent(t *testing.T) { |
| t.Fatalf("agent pass: %v", err) |
| } |
| } |
| - d.convergeLeg = func() { |
| + leg := func() { |
| convergeAdopted( |
| func() map[string]string { |
| live, _ := release.LiveVersion(env.optDir, host) |
| @@ -165,7 +175,8 @@ func TestRootAdoptionOfAWarmExecInstanceIsFinishedByTheAgent(t *testing.T) { |
| 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 |
| + fastTick(ctx, d, now.Add(10*time.Second)) // the serial move: the agent's full pass |
| + leg() // the converge loop's next run |
| 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) |
| @@ -173,7 +184,7 @@ func TestRootAdoptionOfAWarmExecInstanceIsFinishedByTheAgent(t *testing.T) { |
| 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)) |
| + leg() |
| if len(cycled) != 1 { |
| t.Fatalf("cycled again on the next tick: %v", cycled) |
| } |
| @@ -414,3 +425,74 @@ func TestAdoptRestartDispatchSparesAConvergedChild(t *testing.T) { |
| t.Fatalf("spawned builds = %v, want %v: the adopt dispatch left a stale child running", got, want) |
| } |
| } |
| + |
| +// A leg past slowLegAfter says so in the journal, with what it was doing; a |
| +// quick one says nothing, because the fast loop runs every two seconds. |
| +func TestLogSlowSpeaksOnlyWhenSlow(t *testing.T) { |
| + var buf bytes.Buffer |
| + prev, flags := log.Writer(), log.Flags() |
| + log.SetOutput(&buf) |
| + log.SetFlags(0) |
| + t.Cleanup(func() { log.SetOutput(prev); log.SetFlags(flags) }) |
| + |
| + logSlow("fast sweep", time.Now(), "3 routes") |
| + if buf.Len() != 0 { |
| + t.Fatalf("a quick leg logged %q", buf.String()) |
| + } |
| + logSlow("serial pass stages", time.Now().Add(-2*slowLegAfter), "reconcile 9.0s, adopt 1.0s") |
| + got := buf.String() |
| + for _, want := range []string{"serial pass stages took 10", "(reconcile 9.0s, adopt 1.0s)"} { |
| + if !strings.Contains(got, want) { |
| + t.Fatalf("slow-leg line %q lacks %q", got, want) |
| + } |
| + } |
| +} |
| + |
| +// The point of the leg's own loop: it acts while the fast tick is stuck in a |
| +// long full pass. The pass is held open here until the leg has cycled the |
| +// drifted instance, which it could never do if it still rode the tick. |
| +func TestConvergeLoopActsWhileTheFullPassIsBusy(t *testing.T) { |
| + env := newFastFixture(t) |
| + d := env.deps(nil) |
| + inPass, unblock := make(chan struct{}), make(chan struct{}) |
| + d.fullPass = func(context.Context) { close(inPass); <-unblock } |
| + |
| + cycled := make(chan string, 1) |
| + leg := func() { |
| + convergeAdopted( |
| + func() map[string]string { return map[string]string{"warm.test": "v2"} }, |
| + func(h string) (*registry.Route, error) { return ®istry.Route{Host: h, Version: "v2"}, nil }, |
| + func(h, ver string) bool { |
| + select { |
| + case cycled <- h + "@" + ver: |
| + default: |
| + } |
| + return true |
| + }, |
| + adoptedCyclesPerTick) |
| + } |
| + ctx := context.Background() |
| + 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) |
| + } |
| + tickDone := make(chan struct{}) |
| + go func() { fastTick(ctx, d, now.Add(10*time.Second)); close(tickDone) }() |
| + <-inPass // the tick is now stuck in its full pass |
| + // Started only now, so anything it cycles was cycled DURING the pass. |
| + stop := make(chan struct{}) |
| + defer close(stop) |
| + go runConvergeLoop(leg, 5*time.Millisecond, stop) |
| + |
| + select { |
| + case got := <-cycled: |
| + if got != "warm.test@v2" { |
| + t.Fatalf("cycled %q", got) |
| + } |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("the converge leg waited for the full pass") |
| + } |
| + close(unblock) |
| + <-tickDone |
| +} |