amadan

carlos / platform Public

host: wire the agent socket, worker boot pass, and version write

Paul Campbell pushed by paul@keymail.dev 7e7896effa69233a15d2d553dc23b577075544db parent 87ba030
4 files changed, +474 −4
  • internal/activator/activator.go +37 −4
  • internal/activator/activator_test.go +57 −0
  • internal/host/agent.go +150 −0
  • internal/host/agent_test.go +230 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 8c3aa660d..22df6d142 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -189,9 +189,17 @@ type Backend interface {
// in production (routes.next_due), so a sleeping host's timer survives agent
// restarts instead of dying with the process. *registry.Registry satisfies it
// directly.
+//
+// SetVersions rides along here rather than on a separate interface (Task 8):
+// the registry handle Service already holds for the due-time index is the
+// same one that owns routes.version/prev_version, and in every deployment
+// (production and every test in this package) the two are the same
+// *registry.Registry — a second interface would only name a distinction
+// that doesn't exist anywhere it's used. NoteWorkerReady is its one caller.
type DueStore interface {
SetNextDue(host string, due int64) error
DueInstances(now int64) ([]registry.Route, error)
+ SetVersions(host, ver, prev string) error
}
// Lease is a held-but-not-yet-replicating lease (wake in progress; Eleven's
@@ -1339,26 +1347,51 @@ func (a *Service) NoteWorkerContact(host string) {
// NoteWorkerReady records a worker's own "I am up" — last-activity, the build
// it reports running, and the end of its settle window. Deliberately the
// child's claim, not ours: §9.2's honesty rule is that a sidecar which never
-// reports shows unknown, never "current". (Task 8 additionally persists the
-// version through the registry's version setter; this is the in-memory half
-// the agent socket needs today.)
+// reports shows unknown, never "current".
//
// It acts on a WAKING entry as well as an awake one — that is the whole point
// of the settle window: a child that boots fast says so and its wake resolves
// immediately instead of waiting out sidecarSettle.
+//
+// The registry write (Task 8) is the durable half of the same claim — the
+// in-memory e.workerVersion above is what the agent itself remembers between
+// restarts of nothing; the row is what `carlos sidecars ls` and everything
+// else off-process reads. It happens OUTSIDE a.mu: SetVersions is a SQL
+// write, and nothing under the entry lock should wait on disk.
+//
+// An EMPTY version is never written. §9.2's rule is that a sidecar which
+// never reports shows unknown — and the registry's own default for an
+// unset version IS the empty string (SetVersions is never called for a
+// fresh row), so "unknown" is represented by an ABSENT write, not by a
+// stamped "". Calling SetVersions with "" here would be a stamp: it would
+// overwrite whatever version the row already carried (from an earlier
+// report, or an adopt pass) with a value that reads as "verified empty"
+// rather than "this child said nothing this time" — the overclaim the rule
+// exists to forbid. So an empty report still ends the settle window and
+// stamps last-activity (the child IS up; it just isn't naming a build) and
+// simply skips the write.
func (a *Service) NoteWorkerReady(host, version string) {
a.mu.Lock()
- defer a.mu.Unlock()
e := a.entries[host]
if e == nil || !e.worker {
+ a.mu.Unlock()
return
}
e.lastActive = a.now()
e.workerVersion = version
+ prev := e.route.Version
if !e.readyNoted {
e.readyNoted = true
close(e.readyc)
}
+ a.mu.Unlock()
+
+ if version == "" {
+ return
+ }
+ if err := a.due.SetVersions(host, version, prev); err != nil {
+ log.Printf("activator: record worker version for %s: %v", host, err)
+ }
}
// SetWorkerNextDue records a sidecar's next scheduled work, straight to the
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index fc40a3ca9..417ce31a4 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -2623,6 +2623,63 @@ func TestWorkerReadyEndsSettleWindowEarly(t *testing.T) {
}
}
+// TestNoteWorkerReadyPersistsVersionToRegistry pins Task 8's registry half:
+// a worker's own claim over the agent socket must land on the row through
+// the registry's existing version setter, so `carlos sidecars ls` (and
+// anything else reading routes.version) can see it — not just the in-memory
+// entry the agent socket itself consults.
+func TestNoteWorkerReadyPersistsVersionToRegistry(t *testing.T) {
+ a, b, _, reg := testActivator(t, time.Hour)
+ host := "bot.x.internal"
+ if err := reg.Upsert(testWorkerRoute(host)); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ noteReadyAsChild(t, a, b, host, "v9")
+
+ rt, err := reg.Get(host)
+ if err != nil || rt == nil {
+ t.Fatalf("get %s: %+v, %v", host, rt, err)
+ }
+ if rt.Version != "v9" {
+ t.Fatalf("registry Version = %q, want %q", rt.Version, "v9")
+ }
+}
+
+// TestNoteWorkerReadyEmptyVersionNeverStampsUnknown pins §9.2's honesty rule:
+// a child that reports ready with no version claim must leave the registry
+// column exactly as it was (its default "" — displayed as unknown) rather
+// than write an empty string over whatever was there. Recording "" IS a
+// version stamp, and an empty stamp reads as "verified empty" rather than
+// "never reported" — the wrong claim.
+func TestNoteWorkerReadyEmptyVersionNeverStampsUnknown(t *testing.T) {
+ a, b, _, reg := testActivator(t, time.Hour)
+ host := "bot.x.internal"
+ // Seed a version on the row BEFORE the wake, the way a prior adopt pass
+ // would have — an empty report must not clobber it either.
+ if err := reg.Upsert(testWorkerRoute(host)); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := reg.SetVersions(host, "v8", ""); err != nil {
+ t.Fatalf("seed version: %v", err)
+ }
+
+ if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ noteReadyAsChild(t, a, b, host, "")
+
+ rt, err := reg.Get(host)
+ if err != nil || rt == nil {
+ t.Fatalf("get %s: %+v, %v", host, rt, err)
+ }
+ if rt.Version != "v8" {
+ t.Fatalf("registry Version = %q after an empty ready report, want the seeded %q untouched", rt.Version, "v8")
+ }
+}
+
// --- always-on workers: supervision ------------------------------------------
func alwaysOnWorkerRoute(host string) registry.Route {
diff --git a/internal/host/agent.go b/internal/host/agent.go
index b7498b778..863ec22e1 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -143,6 +143,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/agentsock"
"github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/caps"
@@ -276,6 +277,43 @@ func RunAgent(args []string) {
svc.Start()
act = svc
log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
+
+ // The agent socket: the one channel a sidecar dials back over
+ // (internal/agentsock, Task 7). Constructed only in this branch —
+ // the same condition that constructs the activator — because with
+ // hibernation off there are no workers on the box at all, and
+ // nothing would ever call in. cfg.AgentSocket is never empty here:
+ // activator.ConfigFromEnv defaults it to DefaultAgentSocket.
+ //
+ // The cancel is deferred right here, not called explicitly at the
+ // shutdown site below: RunAgent's defers run LIFO at return, so this
+ // one (registered first) fires AFTER the shutdown block's own
+ // `defer cancel()` for svc.Shutdown's context — i.e. once the
+ // children are already down — while staying the one place this
+ // context's lifetime is decided, which is what satisfies go vet's
+ // lostcancel check.
+ agentSockCtx, agentSockCancel := context.WithCancel(context.Background())
+ defer agentSockCancel()
+ agentSrv := agentsock.New(svc)
+ go serveAgentSocket(agentSockCtx, cfg.AgentSocket, agentSrv.Serve)
+ log.Printf("agent: agent socket armed at %s", cfg.AgentSocket)
+
+ // The boot pass (Task 8): a sidecar that does not hibernate must be
+ // running whenever the agent is — an agent restart otherwise leaves
+ // it down until a request or its due time happens to arrive, and an
+ // always-on worker has no request path at all to wake it. Run off
+ // the startup path (its own goroutine, not blocking RunAgent) since
+ // WakeWorker can block for the settle window per host and this must
+ // never delay the edge coming up; runWorkerBootPass itself wakes
+ // its candidates concurrently, bounded, rather than serially.
+ go func() {
+ routes, err := reg.List()
+ if err != nil {
+ log.Printf("agent: worker boot pass: list routes: %v", err)
+ return
+ }
+ runWorkerBootPass(routes, svc.WakeWorker)
+ }()
}
// Fleet box binding publish, startup half (customer-fleets plan, Task 6):
@@ -337,6 +375,118 @@ func RunAgent(args []string) {
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
svc.Shutdown(ctx)
+ // The agent socket's own defer (above, at its construction) stops it
+ // once this function returns — after Shutdown, per the LIFO ordering
+ // explained there. Service is single-use once Shutdown has run (its
+ // lifeCtx is cancelled), so there is nothing left to authorize a call
+ // against by the time the socket actually stops either way.
+ }
+}
+
+// --- Task 8: the agent socket's retry/backoff, and the worker boot pass ----
+
+// workersToStart is the boot pass's selection: a sidecar that does not
+// hibernate must be running whenever the agent is. A hibernating sidecar is
+// deliberately NOT started here — it waits for its due time, an operator
+// command, or (later) a delivery signal, which is the whole point of the
+// class. Status is checked the same way the due-time sweep checks it
+// (activator.go's sweepOnce): only "ready" (or the legacy "" some pre-status
+// rows carry) is eligible — a row still provisioning, or draining, must not
+// be spawned.
+func workersToStart(routes []registry.Route) []registry.Route {
+ var out []registry.Route
+ for _, rt := range routes {
+ if rt.Kind == "sidecar" && !rt.Hibernate && (rt.Status == "" || rt.Status == "ready") {
+ out = append(out, rt)
+ }
+ }
+ return out
+}
+
+// workerBootConcurrency bounds how many sidecars the boot pass wakes at
+// once. WakeWorker blocks for the settle window (activator's sidecarSettle,
+// 10s in production) whenever a child never calls /v1/ready — a slow or
+// silent sidecar, or simply one that only survives the window rather than
+// reporting early — so an unbounded fan-out could still spawn every child on
+// the box in the same instant (a thundering herd against S3 leases and
+// restore reads), while a serial pass would cost up to 10 seconds PER
+// sidecar of agent startup. This bound trades both off: bounded fan-out, not
+// one-at-a-time.
+const workerBootConcurrency = 8
+
+// runWorkerBootPass wakes every route workersToStart selects out of routes,
+// concurrently (bounded by workerBootConcurrency) rather than one at a time
+// — see workerBootConcurrency's doc comment for why serial is too slow and
+// unbounded is too bursty. wake is svc.WakeWorker in production; a seam here
+// so the concurrency shape is testable without a real activator.Service. A
+// wake that errors is logged and does not stop the rest — one sidecar that
+// can't come up must not strand every other one on the box at boot.
+func runWorkerBootPass(routes []registry.Route, wake func(registry.Route) error) {
+ toStart := workersToStart(routes)
+ if len(toStart) == 0 {
+ return
+ }
+ sem := make(chan struct{}, workerBootConcurrency)
+ var wg sync.WaitGroup
+ for _, rt := range toStart {
+ rt := rt
+ wg.Add(1)
+ sem <- struct{}{}
+ go func() {
+ defer wg.Done()
+ defer func() { <-sem }()
+ if err := wake(rt); err != nil {
+ log.Printf("agent: worker boot pass: wake %s: %v", rt.Host, err)
+ }
+ }()
+ }
+ wg.Wait()
+}
+
+// agentSocketRetryBase/Cap bound the agent socket's rebind backoff — vars,
+// not consts, so a test can shrink them (sidecarSettle's precedent in
+// internal/activator).
+var (
+ agentSocketRetryBase = time.Second
+ agentSocketRetryCap = 30 * time.Second
+)
+
+// serveAgentSocket runs serve (agentsock.Server.Serve in production) in a
+// retry loop with doubling backoff, forever, until ctx is cancelled. serve
+// is a seam so the retry/backoff shape is testable without a real unix
+// socket.
+//
+// A failed bind must NEVER be fatal to the agent: agentsock.Serve
+// deliberately REFUSES a socket another listener already owns (rather than
+// unlinking it) to prevent two agents both believing they own the box's
+// workers, and on an agent restart the old process's listener can take a
+// few seconds to release — that refusal is expected and transient. An agent
+// that exited because a socket was briefly busy would take down every route
+// on the box, so this loop only ever logs and retries; it has no path that
+// calls log.Fatal or returns an error to a caller who might.
+//
+// A clean return from serve (nil, once ctx is cancelled — agentsock.Serve's
+// own contract) ends the loop immediately with no retry: that is a normal
+// shutdown, not a failure.
+func serveAgentSocket(ctx context.Context, path string, serve func(context.Context, string) error) {
+ backoff := agentSocketRetryBase
+ for {
+ err := serve(ctx, path)
+ if err == nil || ctx.Err() != nil {
+ return
+ }
+ log.Printf("agent: agent socket %s: %v — retrying in %s", path, err, backoff)
+ timer := time.NewTimer(backoff)
+ select {
+ case <-timer.C:
+ case <-ctx.Done():
+ timer.Stop()
+ return
+ }
+ backoff *= 2
+ if backoff > agentSocketRetryCap {
+ backoff = agentSocketRetryCap
+ }
}
}
diff --git a/internal/host/agent_test.go b/internal/host/agent_test.go
index 07758c854..a21ecdabe 100644
--- a/internal/host/agent_test.go
+++ b/internal/host/agent_test.go
@@ -6,6 +6,7 @@ import (
"crypto/ed25519"
"encoding/base64"
"errors"
+ "fmt"
"log"
"net/http"
"net/http/httptest"
@@ -16,6 +17,7 @@ import (
"sort"
"strings"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -1790,3 +1792,231 @@ func TestGenPollPassRouteShapeCostsNoBucketReads(t *testing.T) {
t.Fatalf("the rebinding tick made %d bucket reads, want exactly 1 (the app's generation object, as before)", got)
}
}
+
+// ---- Task 8: the agent socket, the boot pass, and the version write -------
+
+// waitForCond polls until cond holds — the shared shape every wait-for-a-
+// goroutine test in this package needs (fastConvergeLoop's own tests have
+// no such helper because none of them race a goroutine against a channel).
+func waitForCond(t *testing.T, what string, cond func() bool) {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ if cond() {
+ return
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ t.Fatalf("timed out waiting for %s", what)
+}
+
+// TestWorkerBootPassStartsOnlyNonHibernatingSidecars is Task 8's own test
+// (brief step 1, verbatim): a sidecar that does not hibernate must be
+// running whenever the agent is; a hibernating one waits for its due time,
+// an operator command, or a delivery signal — the whole point of the class.
+// Non-sidecar routes and a non-ready sidecar are excluded too: the boot pass
+// exists to start THIS box's own always-on sidecars, not to touch instances
+// (which have their own convergence) or a row still mid-provisioning.
+func TestWorkerBootPassStartsOnlyNonHibernatingSidecars(t *testing.T) {
+ routes := []registry.Route{
+ {Host: "a.x.sidecar.internal", Kind: "sidecar", Hibernate: false, Status: "ready"},
+ {Host: "b.x.sidecar.internal", Kind: "sidecar", Hibernate: true, Status: "ready"},
+ {Host: "c.example.com", Kind: "instance", Hibernate: false, Status: "ready"},
+ {Host: "d.x.sidecar.internal", Kind: "sidecar", Hibernate: false, Status: "draining"},
+ }
+ got := workersToStart(routes)
+ if len(got) != 1 || got[0].Host != "a.x.sidecar.internal" {
+ t.Fatalf("workersToStart = %v — only a ready, non-hibernating sidecar starts at boot; a hibernating one waits for its timer or a command", got)
+ }
+}
+
+// TestWorkerBootPassWakesEveryNonHibernatingSidecar pins the end-to-end
+// wiring shape: given N candidate routes, the boot pass calls wake exactly
+// once per candidate, and a wake that errors does not stop the rest from
+// being tried (one bad sidecar must not strand every other one on the box
+// at boot).
+func TestWorkerBootPassWakesEveryNonHibernatingSidecar(t *testing.T) {
+ routes := []registry.Route{
+ {Host: "a.x.sidecar.internal", Kind: "sidecar", Status: "ready"},
+ {Host: "b.x.sidecar.internal", Kind: "sidecar", Status: "ready"},
+ {Host: "c.x.sidecar.internal", Kind: "sidecar", Hibernate: true, Status: "ready"}, // excluded
+ }
+ var mu sync.Mutex
+ woken := map[string]int{}
+ wake := func(rt registry.Route) error {
+ mu.Lock()
+ woken[rt.Host]++
+ mu.Unlock()
+ if rt.Host == "a.x.sidecar.internal" {
+ return errors.New("boom")
+ }
+ return nil
+ }
+ runWorkerBootPass(routes, wake)
+
+ mu.Lock()
+ defer mu.Unlock()
+ if woken["a.x.sidecar.internal"] != 1 || woken["b.x.sidecar.internal"] != 1 {
+ t.Fatalf("woken = %v, want exactly one call each for a and b", woken)
+ }
+ if _, ok := woken["c.x.sidecar.internal"]; ok {
+ t.Fatalf("a hibernating sidecar was woken at boot: %v", woken)
+ }
+}
+
+// TestWorkerBootPassWakesConcurrently pins the carried requirement that the
+// boot pass must NOT wake sidecars one at a time: WakeWorker blocks for the
+// settle window (10s) when nothing answers /v1/ready, so a serial pass over
+// N sidecars would cost 10N seconds of agent startup. This drives more
+// in-flight wakes than the concurrency bound and proves the bound is
+// actually reached (not stuck at 1, which a serial implementation would
+// show) and never exceeded (which an unbounded implementation would show).
+func TestWorkerBootPassWakesConcurrently(t *testing.T) {
+ n := workerBootConcurrency + 3 // strictly more than the bound
+ routes := make([]registry.Route, n)
+ for i := range routes {
+ routes[i] = registry.Route{
+ Host: fmt.Sprintf("s%d.x.sidecar.internal", i), Kind: "sidecar", Status: "ready",
+ }
+ }
+
+ var mu sync.Mutex
+ inFlight, maxInFlight := 0, 0
+ release := make(chan struct{})
+ wake := func(registry.Route) error {
+ mu.Lock()
+ inFlight++
+ if inFlight > maxInFlight {
+ maxInFlight = inFlight
+ }
+ mu.Unlock()
+ <-release
+ mu.Lock()
+ inFlight--
+ mu.Unlock()
+ return nil
+ }
+
+ done := make(chan struct{})
+ go func() {
+ runWorkerBootPass(routes, wake)
+ close(done)
+ }()
+
+ waitForCond(t, "the boot pass to saturate its concurrency bound", func() bool {
+ mu.Lock()
+ defer mu.Unlock()
+ return inFlight == workerBootConcurrency
+ })
+ // A beat with the bound already saturated: an unbounded implementation
+ // would keep climbing past it here rather than holding still.
+ time.Sleep(50 * time.Millisecond)
+ mu.Lock()
+ got := maxInFlight
+ mu.Unlock()
+ if got != workerBootConcurrency {
+ t.Fatalf("max concurrent wakes = %d, want exactly the bound %d", got, workerBootConcurrency)
+ }
+
+ close(release)
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("boot pass did not finish after its wakes were released")
+ }
+}
+
+// ---- the agent socket: retry/backoff on a bind that lost the race --------
+
+// TestServeAgentSocketReturnsOnCleanShutdown pins the non-error path: a nil
+// return (agentsock.Serve's contract once ctx is cancelled and the listener
+// shut down cleanly) must end the loop immediately, with no retry.
+func TestServeAgentSocketReturnsOnCleanShutdown(t *testing.T) {
+ var calls int32
+ serve := func(context.Context, string) error {
+ atomic.AddInt32(&calls, 1)
+ return nil
+ }
+ serveAgentSocket(context.Background(), "/tmp/does-not-matter.sock", serve)
+ if n := atomic.LoadInt32(&calls); n != 1 {
+ t.Fatalf("serve calls = %d, want exactly 1 — a clean nil return must never retry", n)
+ }
+}
+
+// TestServeAgentSocketRetriesWithBackoffOnBusySocket pins the carried
+// requirement that a failed bind — agentsock.Serve deliberately refusing a
+// socket another (likely still-shutting-down) incarnation still owns — must
+// NEVER be fatal: the loop retries with backoff instead of returning the
+// error up to a log.Fatal that would take the whole agent, and every route
+// on the box, down over a transient restart race.
+func TestServeAgentSocketRetriesWithBackoffOnBusySocket(t *testing.T) {
+ oldBase, oldCap := agentSocketRetryBase, agentSocketRetryCap
+ agentSocketRetryBase = time.Millisecond
+ agentSocketRetryCap = 4 * time.Millisecond
+ defer func() { agentSocketRetryBase, agentSocketRetryCap = oldBase, oldCap }()
+
+ var calls int32
+ succeeded := make(chan struct{})
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ serve := func(c context.Context, path string) error {
+ n := atomic.AddInt32(&calls, 1)
+ if n < 4 {
+ return errors.New("agentsock: /run/carlos/agent.sock is already served by a live listener")
+ }
+ close(succeeded)
+ <-c.Done() // the fourth "bind" holds until shutdown, like the real Serve
+ return nil
+ }
+
+ done := make(chan struct{})
+ go func() {
+ serveAgentSocket(ctx, "/run/carlos/agent.sock", serve)
+ close(done)
+ }()
+
+ select {
+ case <-succeeded:
+ case <-time.After(2 * time.Second):
+ t.Fatal("a busy socket was never retried to success")
+ }
+ cancel()
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("serveAgentSocket did not return after its context was cancelled")
+ }
+}
+
+// TestServeAgentSocketStopsRetryingOnContextCancel pins the other half of
+// "never fatal, never stuck": a permanently busy socket must not wedge the
+// goroutine forever — cancelling ctx ends the retry loop promptly even
+// mid-backoff.
+func TestServeAgentSocketStopsRetryingOnContextCancel(t *testing.T) {
+ oldBase, oldCap := agentSocketRetryBase, agentSocketRetryCap
+ agentSocketRetryBase = time.Millisecond
+ agentSocketRetryCap = 2 * time.Millisecond
+ defer func() { agentSocketRetryBase, agentSocketRetryCap = oldBase, oldCap }()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ var calls int32
+ serve := func(context.Context, string) error {
+ atomic.AddInt32(&calls, 1)
+ return errors.New("still busy")
+ }
+ done := make(chan struct{})
+ go func() {
+ serveAgentSocket(ctx, "/run/carlos/agent.sock", serve)
+ close(done)
+ }()
+
+ waitForCond(t, "at least one retry to have happened", func() bool {
+ return atomic.LoadInt32(&calls) >= 2
+ })
+ cancel()
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("serveAgentSocket did not stop retrying after context cancellation")
+ }
+}