amadan

carlos / platform Public

Move the whole bounded wake fan-out off the tick goroutine, not just each wake

Paul Campbell pushed by paul@keymail.dev a562ec34ca7b471b43471227aa6a1f506b5de82a parent cb1d128
4 files changed, +196 −55
  • internal/host/restart.go +70 −20
  • internal/host/restart_test.go +52 −0
  • internal/host/sidecarwake.go +33 −35
  • internal/host/sidecarwake_test.go +41 −0
diff --git a/internal/host/restart.go b/internal/host/restart.go
index 5f2646328..7bfa50fd5 100644
--- a/internal/host/restart.go
+++ b/internal/host/restart.go
@@ -228,6 +228,57 @@ func bounceExecRoutes(routes []registry.Route, moved [][2]string, bounce func(ho
return n
}
+// dispatchBoundedWakes wakes every route in toWake, bounded by
+// workerBootConcurrency wakes running at once, entirely inside ONE
+// goroutine this function launches — so dispatchBoundedWakes ITSELF
+// returns to its caller immediately, no matter how many routes are queued
+// behind it.
+//
+// This is fix round 2's own correction of round 1's own fix (sidecar-
+// members Task 13): the round-1 shape acquired the semaphore in the
+// CALLER's loop, one route at a time — which dispatches each wake off the
+// blocking path individually, but still leaves the loop ITSELF, and
+// therefore the caller, blocked on the sem acquisition once
+// workerBootConcurrency wakes are already in flight and a
+// (workerBootConcurrency+1)th route needs dispatching. For an app shaped
+// like Eleven's — a sidecar PER ENROLLED MEMBER, so routinely more than 8
+// sidecars for one app on one box — that reintroduces exactly the bug this
+// idiom exists to fix: a restart or wake burst on such an app would stall
+// the box's single per-tick convergence goroutine for up to wakeWaitBound
+// (30s) per sidecar past the 8th. Moving the WHOLE bounded fan-out inside
+// one dispatched goroutine fixes it: the bound still governs how many
+// wakes run concurrently, and now ONLY that — never when the caller gets
+// to continue.
+//
+// Shared by cycleWorkers (this file) and sidecarwake.go's
+// dispatchSidecarWakes, the two callers of `carlos restart` and the
+// sidecar wake feature respectively that both need "wake N sidecars
+// without blocking the caller, bounded" — one idiom, not two copies that
+// would each need this fix applied separately.
+//
+// wake is called once per route in toWake; report, if non-nil, is called
+// once per wake with its result (nil error on success) so each caller can
+// log in its own voice. A nil wake or an empty toWake is a working no-op.
+func dispatchBoundedWakes(toWake []registry.Route, wake func(rt registry.Route) error, report func(rt registry.Route, err error)) {
+ if wake == nil || len(toWake) == 0 {
+ return
+ }
+ go func() {
+ sem := make(chan struct{}, workerBootConcurrency)
+ for _, rt := range toWake {
+ rt := rt
+ sem <- struct{}{}
+ go func() {
+ defer func() { <-sem }()
+ err := wake(rt)
+ if report != nil {
+ report(rt, err)
+ }
+ }()
+ }
+ }()
+}
+
// cycleWorkers is bounceExecRoutes' sidecar counterpart, and deliberately a
// separate function rather than a widened bounceExecRoutes: a sidecar is
// activator-managed exactly like a socket-exec instance, but Bounce alone
@@ -244,17 +295,18 @@ func bounceExecRoutes(routes []registry.Route, moved [][2]string, bounce func(ho
//
// The BOUNCE half is synchronous, in-process and effectively instant — it
// is counted in the return value, exactly like bounceExecRoutes' own bounce
-// call. The WAKE half is NOT: WakeWorker can block up to wakeWaitBound
-// (30s, activator.go) per sidecar waiting out the settle window, and this
-// function's only caller (restartRequestLeg) runs synchronously on the
-// box's single per-tick convergence goroutine (fastTick) — waiting here
-// would delay every other app's serial-driven convergence, and the
-// root-restart marker touch that relays unit-backed apps' own restarts,
-// sequentially, by up to that bound PER SIDECAR (fix round 2 review
-// finding). Wakes are therefore dispatched off this call's return path,
-// bounded by the same workerBootConcurrency Task 8's worker boot pass uses
-// (internal/host/agent.go) — one idiom for "wake several workers without
-// blocking a caller", not a second one. Safe because WakeWorker is
+// call, and it stays in THIS function's own loop: only the eligible,
+// actually-stopped routes are collected into toWake. The WAKE half is NOT
+// instant: WakeWorker can block up to wakeWaitBound (30s, activator.go) per
+// sidecar waiting out the settle window, and this function's only caller
+// (restartRequestLeg) runs synchronously on the box's single per-tick
+// convergence goroutine (fastTick) — waiting on it, even indirectly via a
+// per-route semaphore acquisition in this function's own loop, would delay
+// every other app's serial-driven convergence behind it (see
+// dispatchBoundedWakes' own doc comment for the exact shape of that bug,
+// caught and fixed in review round 2). The whole bounded fan-out is
+// therefore handed to dispatchBoundedWakes, which returns immediately
+// regardless of how many routes it was given. Safe because WakeWorker is
// idempotent and `carlos restart`'s contract is "cycle the processes", not
// "return once they are up" — a sidecar that comes up a moment after this
// tick has still been restarted. A wake that errors is logged and does not
@@ -272,8 +324,8 @@ func cycleWorkers(routes []registry.Route, moved [][2]string, bounce func(host s
for _, k := range moved {
want[k] = true
}
- sem := make(chan struct{}, workerBootConcurrency)
n := 0
+ var toWake []registry.Route
for _, rt := range routes {
if !rt.IsSidecar() || !want[[2]string{release.RouteAccount(rt), rt.App}] {
continue
@@ -283,15 +335,13 @@ func cycleWorkers(routes []registry.Route, moved [][2]string, bounce func(host s
}
n++
log.Printf("agent: restart requested: bounced sidecar %s", rt.Host)
- rt := rt
- sem <- struct{}{}
- go func() {
- defer func() { <-sem }()
- if err := wake(rt); err != nil {
- log.Printf("agent: restart requested: re-wake sidecar %s: %v", rt.Host, err)
- }
- }()
+ toWake = append(toWake, rt)
}
+ dispatchBoundedWakes(toWake, wake, func(rt registry.Route, err error) {
+ if err != nil {
+ log.Printf("agent: restart requested: re-wake sidecar %s: %v", rt.Host, err)
+ }
+ })
return n
}
diff --git a/internal/host/restart_test.go b/internal/host/restart_test.go
index dbf649ac6..2991818ea 100644
--- a/internal/host/restart_test.go
+++ b/internal/host/restart_test.go
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
+ "fmt"
"os"
"path/filepath"
"slices"
@@ -933,3 +934,54 @@ func TestCycleWorkersWithoutAnActivatorOrWakeIsANoop(t *testing.T) {
t.Fatalf("cycleWorkers = %d with no activator armed, want 0 and no panic", n)
}
}
+
+// TestCycleWorkersReturnsPromptlyWithManySlowWakesQueued is review round
+// 2's own regression pin: a NARROW earlier version of the fan-out
+// (fix round 1) acquired the workerBootConcurrency semaphore in cycleWorkers'
+// OWN loop, one route at a time — so once workerBootConcurrency (8) wakes
+// were already in flight, the (workerBootConcurrency+1)th route's
+// acquisition blocked cycleWorkers itself, and therefore restartRequestLeg
+// and fastTick's single per-box convergence goroutine, for up to
+// wakeWaitBound (30s) per sidecar past the 8th — a narrow reintroduction of
+// exactly the bug TestCycleWorkersReturnsBeforeASlowWakeCompletes already
+// pins for a SINGLE sidecar. This drives strictly more than
+// workerBootConcurrency sidecars, every wake deliberately parked, and
+// asserts cycleWorkers still returns immediately: the shape an app like
+// Eleven's (a sidecar per enrolled member) hits routinely, not on rare
+// unlucky timing.
+func TestCycleWorkersReturnsPromptlyWithManySlowWakesQueued(t *testing.T) {
+ n := workerBootConcurrency + 5 // strictly more than the bound
+ routes := make([]registry.Route, n)
+ for i := range routes {
+ routes[i] = sidecarRoute(fmt.Sprintf("s%d.x.sidecar.internal", i), "bdm", "eleven")
+ }
+ moved := [][2]string{{"bdm", "eleven"}}
+ bounce := func(h string) bool { return true }
+ slow := make(chan struct{})
+ var mu sync.Mutex
+ woken := map[string]bool{}
+ wake := func(rt registry.Route) error {
+ <-slow
+ mu.Lock()
+ woken[rt.Host] = true
+ mu.Unlock()
+ return nil
+ }
+
+ start := time.Now()
+ got := cycleWorkers(routes, moved, bounce, wake)
+ elapsed := time.Since(start)
+ if got != n {
+ t.Fatalf("cycleWorkers = %d, want %d (every route bounced)", got, n)
+ }
+ if elapsed > time.Second {
+ t.Fatalf("cycleWorkers took %s to return with %d wakes blocked (bound %d) — the whole bounded fan-out must run off the caller's goroutine, not just each individual wake", elapsed, n, workerBootConcurrency)
+ }
+
+ close(slow)
+ waitForCond(t, "every queued sidecar to be woken eventually", func() bool {
+ mu.Lock()
+ defer mu.Unlock()
+ return len(woken) == n
+ })
+}
diff --git a/internal/host/sidecarwake.go b/internal/host/sidecarwake.go
index f300abad0..b3c4246b2 100644
--- a/internal/host/sidecarwake.go
+++ b/internal/host/sidecarwake.go
@@ -11,14 +11,17 @@
//
// Split into the same two shapes restart.go uses for exactly the same
// reason: sidecarWakeChanges is the cheap, synchronous etag comparison
-// (restartGenPass's counterpart), while dispatchSidecarWakes is the
-// WakeWorker fan-out — which can block up to wakeWaitBound (30s,
-// activator.go) per sidecar waiting out the settle window — dispatched off
-// the caller's return path, bounded by workerBootConcurrency (agent.go),
-// the SAME idiom Task 8's boot pass and cycleWorkers both use. sidecarWakeLeg
-// is the two glued together and wired into fastConvergeLoop beside the
-// restart leg (agent.go's fastDeps.wakeLeg), so it rides the same 2s
-// serial-driven convergence `carlos restart` already gets.
+// (restartGenPass's counterpart), while dispatchSidecarWakes hands the
+// WakeWorker fan-out to restart.go's shared dispatchBoundedWakes — which
+// can block up to wakeWaitBound (30s, activator.go) per sidecar waiting
+// out the settle window, and runs the WHOLE bounded fan-out inside one
+// goroutine it launches itself, so the caller returns immediately
+// regardless of how many sidecars are queued (review round 2: an app
+// shaped like Eleven's, a sidecar per enrolled member, is routinely more
+// than workerBootConcurrency). sidecarWakeLeg is the two glued together and
+// wired into fastConvergeLoop beside the restart leg (agent.go's
+// fastDeps.wakeLeg), so it rides the same 2s serial-driven convergence
+// `carlos restart` already gets.
package host
import (
@@ -113,36 +116,31 @@ func sidecarWakeChanges(ctx context.Context, routes []registry.Route, store rele
return moved, readErrs
}
-// dispatchSidecarWakes wakes every route in moved, concurrently, bounded by
-// workerBootConcurrency — the SAME idiom Task 8's runWorkerBootPass and
-// cycleWorkers' own wake half use, deliberately not a third shape.
-// WakeWorker can block up to wakeWaitBound (30s) per sidecar waiting out
-// the settle window, and this function's only caller (sidecarWakeLeg) runs
-// synchronously on the box's single per-tick convergence goroutine
-// (fastTick) — waiting here would delay every other app's serial-driven
-// convergence behind it. Safe because WakeWorker is idempotent and a wake
-// request's contract is "wake the process", not "return once it's up" — a
-// sidecar that comes up a moment after this tick has still been woken. A
-// wake that errors is logged and does not stop, delay, or fail the others.
+// dispatchSidecarWakes wakes every route in moved through the shared
+// dispatchBoundedWakes helper (restart.go) — the SAME idiom Task 8's
+// runWorkerBootPass and cycleWorkers' own wake half use, deliberately not a
+// third shape. WakeWorker can block up to wakeWaitBound (30s) per sidecar
+// waiting out the settle window, and this function's only caller
+// (sidecarWakeLeg) runs synchronously on the box's single per-tick
+// convergence goroutine (fastTick) — dispatchBoundedWakes' own doc comment
+// covers why the WHOLE bounded fan-out, not just each individual wake,
+// must be off that goroutine (review round 2: an app shaped like Eleven's,
+// a sidecar per enrolled member, routinely queues more than
+// workerBootConcurrency sidecars behind one wake burst). Safe because
+// WakeWorker is idempotent and a wake request's contract is "wake the
+// process", not "return once it's up" — a sidecar that comes up a moment
+// after this tick has still been woken. A wake that errors is logged and
+// does not stop, delay, or fail the others.
//
// wake may be nil (no activator armed on this box): a working no-op.
func dispatchSidecarWakes(moved []registry.Route, wake func(rt registry.Route) error) {
- if wake == nil || len(moved) == 0 {
- return
- }
- sem := make(chan struct{}, workerBootConcurrency)
- for _, rt := range moved {
- rt := rt
- sem <- struct{}{}
- go func() {
- defer func() { <-sem }()
- if err := wake(rt); err != nil {
- log.Printf("agent: sidecar wake requested: wake %s: %v", rt.Host, err)
- return
- }
- log.Printf("agent: sidecar wake requested: woke %s", rt.Host)
- }()
- }
+ dispatchBoundedWakes(moved, wake, func(rt registry.Route, err error) {
+ if err != nil {
+ log.Printf("agent: sidecar wake requested: wake %s: %v", rt.Host, err)
+ return
+ }
+ log.Printf("agent: sidecar wake requested: woke %s", rt.Host)
+ })
}
// sidecarWakeLeg is sidecarWakeChanges and dispatchSidecarWakes glued
diff --git a/internal/host/sidecarwake_test.go b/internal/host/sidecarwake_test.go
index d00a6186d..bbedae5d5 100644
--- a/internal/host/sidecarwake_test.go
+++ b/internal/host/sidecarwake_test.go
@@ -3,6 +3,7 @@ package host
import (
"context"
"errors"
+ "fmt"
"slices"
"sync"
"testing"
@@ -430,3 +431,43 @@ func TestFastTickToleratesANilWakeLeg(t *testing.T) {
}
fastTick(context.Background(), d, now) // must not panic
}
+
+// TestDispatchSidecarWakesReturnsPromptlyWithManySlowWakesQueued is
+// dispatchSidecarWakes' half of review round 2's regression pin (see
+// restart_test.go's TestCycleWorkersReturnsPromptlyWithManySlowWakesQueued
+// for the full shape of the bug this catches): strictly more than
+// workerBootConcurrency sidecars, every wake deliberately parked, must
+// still return this function immediately — the whole bounded fan-out runs
+// inside dispatchBoundedWakes' own goroutine now, not just each individual
+// wake dispatched from a loop that can itself stall once the bound fills.
+func TestDispatchSidecarWakesReturnsPromptlyWithManySlowWakesQueued(t *testing.T) {
+ n := workerBootConcurrency + 5
+ moved := make([]registry.Route, n)
+ for i := range moved {
+ moved[i] = wakeSidecarRoute(fmt.Sprintf("s%d.x.sidecar.internal", i), "bdm", "eleven", fmt.Sprintf("s%d", i))
+ }
+ slow := make(chan struct{})
+ var mu sync.Mutex
+ woken := map[string]bool{}
+ wake := func(rt registry.Route) error {
+ <-slow
+ mu.Lock()
+ woken[rt.Host] = true
+ mu.Unlock()
+ return nil
+ }
+
+ start := time.Now()
+ dispatchSidecarWakes(moved, wake)
+ elapsed := time.Since(start)
+ if elapsed > time.Second {
+ t.Fatalf("dispatchSidecarWakes took %s to return with %d wakes blocked (bound %d) — the whole bounded fan-out must run off the caller's goroutine", elapsed, n, workerBootConcurrency)
+ }
+
+ close(slow)
+ waitForCond(t, "every queued sidecar to be woken eventually", func() bool {
+ mu.Lock()
+ defer mu.Unlock()
+ return len(woken) == n
+ })
+}