amadan

carlos / platform Public

zdd plan 1 task 3: the activator holds generation handles and answers AppSocket

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b98eed71bf1b043da547982385b0f9675643ee70 parent a744e0c
4 files changed, +428 −2
  • internal/activator/activator.go +38 −2
  • internal/activator/genservice.go +115 −0
  • internal/activator/genservice_test.go +236 −0
  • internal/edge/gensource.go +39 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index f71e93bfd..d84b208b4 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -382,6 +382,12 @@ type actEntry struct {
spawnVersion string
staleSince time.Time
+ // gen is the generation this entry's process is (zero-downtime spec
+ // §7.1): its number, socket, control socket and build. Set under a.mu at
+ // the awake transition and never changed after it in this plan (plan 3's
+ // switch replaces it). Zero for a worker and before the wake resolves.
+ gen Gen
+
// Sidecar co-spawn: the sidecar exists exactly while its instance is
// awake. All sidecar fields are written under a.mu in the awake state only
// (wake's attach, the sweeper's restart); a stop flips the state under the
@@ -661,6 +667,18 @@ type Service struct {
sweepStop chan struct{}
sweepDone chan struct{}
+ // gens is the generations switch (SetGenerations) as read at New. Off:
+ // every spawn binds the canonical socket and no retire hook fires.
+ gens bool
+ // cacheGens is host -> *atomic.Uint64, the newest generation that has
+ // been current for that host (CacheGen). A sync.Map, never under a.mu:
+ // the edge cache reads it while holding the cache's own lock.
+ cacheGens sync.Map
+ // retiredMu guards retiredHooks (OnGenRetired). A leaf lock: nothing is
+ // called while it is held.
+ retiredMu sync.Mutex
+ retiredHooks []func(host, socket string)
+
// OnWake, when set, is called with the ROUTE after EVERY successful wake
// — fleet boxes use it to publish a binding object to the deployment
// bucket (customer-fleets plan, Task 6) naming this box as the current
@@ -691,6 +709,7 @@ func New(cfg Config, backend Backend, due DueStore) *Service {
due: due,
now: time.Now,
entries: make(map[string]*actEntry),
+ gens: GenerationsEnabled(),
}
// The service lifetime, cancelled by Shutdown. Only work that a shutdown
// should ABANDON hangs off it — today that is exactly the supervised
@@ -993,6 +1012,16 @@ func (a *Service) wake(e *actEntry) {
// cycle. Fresh wake, so it can't be stale yet.
e.spawnVersion = w.spawnVersion
e.staleSince = time.Time{}
+ // The generation is current from here (spec §7.1). A wake the
+ // claim path ran (doClaimWake) spawned through its provider, not
+ // startGen, so it gets the canonical handle and a number of its own.
+ e.gen = w.gen
+ if e.gen.N == 0 && !e.worker {
+ e.gen = canonicalGen(e.route, nextGen(), w.spawnVersion)
+ }
+ if e.gen.N != 0 {
+ a.noteCurrent(e.host, e.gen.N)
+ }
if a.fences[e.host] != 0 {
a.beginStopLocked(e)
} else if e.bounceAfterWake {
@@ -1075,6 +1104,7 @@ type wokenInstance struct {
sidecar Proc
sidecarWanted bool
spawnVersion string // live/<host> target read at exec — for build convergence
+ gen Gen // the spawned generation's handle (instances only; zero for a worker)
// spawnTook is how long Backend.Start took: exec to the socket accepting,
// which is the instance's OWN boot and nothing of the platform's. wakeTook
// is the whole wake around it, filled in by wake() rather than here
@@ -1279,7 +1309,9 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
}
proc, spawnVer, err = a.backend.StartWorker(ctx, rt, workerEnv(a.agentSocket(), host, token))
} else {
- proc, spawnVer, err = a.backend.Start(ctx, rt)
+ var g Gen
+ proc, g, err = a.startGen(ctx, rt)
+ spawnVer, w.gen = g.Build, g
}
spawnTook := a.now().Sub(spawnStart)
w.proc = proc
@@ -2214,7 +2246,7 @@ func (a *Service) stop(e *actEntry) {
// sidecar's ONLY way back up, since nothing can dial one.
if !e.worker {
dueCtx, cancelDue := context.WithTimeout(context.Background(), 5*time.Second)
- due, err := a.backend.NextDue(dueCtx, e.route)
+ due, err := a.nextDue(dueCtx, e)
cancelDue()
if err != nil {
log.Printf("activator: next-due probe %s: %v — recording none (it sleeps until a request)", e.host, err)
@@ -2371,7 +2403,11 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
e.stopOutcome = outcome
a.rememberStopLocked(e.host, outcome)
delete(a.entries, e.host)
+ retired := e.gen
a.mu.Unlock()
+ // The generation stopped being current when its entry went (spec §7.2:
+ // its idle connections close now, not at the transport's idle timeout).
+ a.genRetired(e.host, retired)
close(e.stopped)
log.Printf("activator: %s hibernated", e.host)
}
diff --git a/internal/activator/genservice.go b/internal/activator/genservice.go
new file mode 100644
index 000000000..5b6cbeac6
--- /dev/null
+++ b/internal/activator/genservice.go
@@ -0,0 +1,115 @@
+package activator
+
+import (
+ "context"
+ "sync/atomic"
+
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// The Service's generation surface (zero-downtime spec §7.1). AppSocket is
+// for traffic; the handle (Gen) is for whoever started the generation.
+var _ edge.GenSource = (*Service)(nil)
+
+// GenerationsOn is the dormant switch as this Service read it at New.
+func (a *Service) GenerationsOn() bool { return a.gens }
+
+// startGen spawns the next generation of rt's instance: the process-wide
+// counter's next number, and the backend's generation spawn when it has
+// one. A backend without one is spawned with Start and gets the canonical
+// handle.
+func (a *Service) startGen(ctx context.Context, rt registry.Route) (Proc, Gen, error) {
+ n := nextGen()
+ if g, ok := a.backend.(GenBackend); ok {
+ return g.StartGen(ctx, rt, n, a.gens)
+ }
+ proc, ver, err := a.backend.Start(ctx, rt)
+ if err != nil {
+ return proc, Gen{}, err
+ }
+ return proc, canonicalGen(rt, n, ver), nil
+}
+
+// AppSocket names host's CURRENT generation's socket, for traffic: the edge
+// proxy, scheduled delivery, the next-due probe at a stop, and the status
+// probe (spec §7.1). It answers for an awake or stopping entry (a stopping
+// generation is still the one in-flight requests reach). Asleep, waking or
+// a worker: false.
+func (a *Service) AppSocket(host string) (string, bool) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.state == actWaking || e.gen.Socket == "" {
+ return "", false
+ }
+ return e.gen.Socket, true
+}
+
+// ServingGen is AppSocket with the generation's number and build: the edge
+// picks a connection pool, stamps X-Carlos-Version and tags cache entries
+// from it (spec §7.2).
+func (a *Service) ServingGen(host string) (edge.GenInfo, bool) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.state == actWaking || e.gen.Socket == "" {
+ return edge.GenInfo{}, false
+ }
+ return edge.GenInfo{Socket: e.gen.Socket, N: e.gen.N, Build: e.gen.Build}, true
+}
+
+// CacheGen is the newest generation that has been current for host in this
+// process. Lock-free on purpose: the edge cache calls it under its own lock,
+// and plan 3's switch purges the cache while holding a.mu.
+func (a *Service) CacheGen(host string) (uint64, bool) {
+ v, ok := a.cacheGens.Load(host)
+ if !ok {
+ return 0, false
+ }
+ return v.(*atomic.Uint64).Load(), true
+}
+
+// noteCurrent records that generation n of host has become current.
+func (a *Service) noteCurrent(host string, n uint64) {
+ v, _ := a.cacheGens.LoadOrStore(host, new(atomic.Uint64))
+ v.(*atomic.Uint64).Store(n)
+}
+
+// OnGenRetired registers fn for every generation that stops being current
+// (edge.GenSource). Hooks fire only with generations on.
+func (a *Service) OnGenRetired(fn func(host, socket string)) {
+ a.retiredMu.Lock()
+ defer a.retiredMu.Unlock()
+ a.retiredHooks = append(a.retiredHooks, fn)
+}
+
+// genRetired tells every hook that g has stopped being host's current
+// generation. The caller holds no lock. A no-op with generations off: the
+// canonical path is shared by every generation of a host then, so closing
+// "its" idle connections could close the next process's.
+func (a *Service) genRetired(host string, g Gen) {
+ if !a.gens || g.Socket == "" {
+ return
+ }
+ a.retiredMu.Lock()
+ hooks := append([]func(string, string){}, a.retiredHooks...)
+ a.retiredMu.Unlock()
+ for _, fn := range hooks {
+ fn(host, g.Socket)
+ }
+}
+
+// nextDue is the sleep-time next-due probe, dialled at the instance's
+// current generation (spec §7.1, `backend_exec.go`'s NextDue). Hibernation
+// joins any successor first (plan 3), so at a stop the current generation
+// is the only one. With generations off AppSocket is rt.Socket, and this is
+// the old call.
+func (a *Service) nextDue(ctx context.Context, e *actEntry) (int64, error) {
+ if g, ok := a.backend.(GenBackend); ok {
+ if sock, ok := a.AppSocket(e.host); ok {
+ return g.NextDueVia(ctx, e.route, sock)
+ }
+ }
+ return a.backend.NextDue(ctx, e.route)
+}
diff --git a/internal/activator/genservice_test.go b/internal/activator/genservice_test.go
new file mode 100644
index 000000000..51d678540
--- /dev/null
+++ b/internal/activator/genservice_test.go
@@ -0,0 +1,236 @@
+package activator
+
+import (
+ "context"
+ "fmt"
+ "path/filepath"
+ "slices"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// genFakeBackend is fakeBackend with the generation capability. A spawn
+// with perGen "binds" /gen/<n>.sock (nothing is bound: the fake proc stands
+// in for the process) and NextDueVia records the socket it was asked to
+// dial, so a test can see which generation a probe reached.
+type genFakeBackend struct {
+ *fakeBackend
+ gmu sync.Mutex
+ perGens []bool
+ dueSocks []string
+}
+
+func (b *genFakeBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
+ b.gmu.Lock()
+ b.perGens = append(b.perGens, perGen)
+ b.gmu.Unlock()
+ proc, ver, err := b.fakeBackend.Start(ctx, rt)
+ if err != nil {
+ return proc, Gen{}, err
+ }
+ sock := rt.Socket
+ if perGen {
+ sock = fmt.Sprintf("/gen/%d.sock", n)
+ }
+ return proc, Gen{N: n, Socket: sock, Control: sock + ".control", Build: ver}, nil
+}
+
+func (b *genFakeBackend) NextDueVia(ctx context.Context, rt registry.Route, socket string) (int64, error) {
+ b.gmu.Lock()
+ b.dueSocks = append(b.dueSocks, socket)
+ b.gmu.Unlock()
+ return b.fakeBackend.NextDue(ctx, rt)
+}
+
+func (b *genFakeBackend) seen() (perGens []bool, dueSocks []string) {
+ b.gmu.Lock()
+ defer b.gmu.Unlock()
+ return slices.Clone(b.perGens), slices.Clone(b.dueSocks)
+}
+
+// genActivator is testActivator over genFakeBackend, with the switch set
+// for the test's duration BEFORE New reads it.
+func genActivator(t *testing.T, on bool) (*Service, *genFakeBackend) {
+ t.Helper()
+ withGenerations(t, on)
+ gb := &genFakeBackend{fakeBackend: &fakeBackend{execVerFn: func() string { return "v7" }}}
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return New(Config{Bucket: "test-bucket", Idle: time.Hour}, gb, reg), gb
+}
+
+// genTestRoute is testRoute with its canonical socket in the test's own
+// temp dir. With generations on the activator links a route's canonical
+// path to its current generation (Task 4), and testRoute's shared
+// /tmp/nope.sock must never become a real link.
+func genTestRoute(t *testing.T, host string) registry.Route {
+ t.Helper()
+ rt := testRoute(host)
+ rt.Socket = filepath.Join(t.TempDir(), "c.sock")
+ return rt
+}
+
+// stopAndWait bounces host's awake instance and waits for the stop to finish.
+func stopAndWait(t *testing.T, a *Service, host string) {
+ t.Helper()
+ if !a.Bounce(host) {
+ t.Fatalf("Bounce(%s) started no stop", host)
+ }
+ waitFor(t, host+" to stop", func() bool { return !a.Owns(host) })
+}
+
+func TestAppSocketNamesTheCurrentGeneration(t *testing.T) {
+ a, gb := genActivator(t, true)
+ rt := genTestRoute(t, "cur.example.com")
+ if _, ok := a.AppSocket(rt.Host); ok {
+ t.Fatal("AppSocket answered for an instance that never woke")
+ }
+ arrangeAwake(t, a, rt)
+ g, ok := a.ServingGen(rt.Host)
+ if !ok || g.N == 0 {
+ t.Fatalf("ServingGen = %+v, %v; want the awake generation", g, ok)
+ }
+ sock, ok := a.AppSocket(rt.Host)
+ if !ok || sock != fmt.Sprintf("/gen/%d.sock", g.N) || g.Socket != sock || g.Build != "v7" {
+ t.Fatalf("AppSocket = %q (%v), ServingGen = %+v; want /gen/%d.sock and build v7", sock, ok, g, g.N)
+ }
+ if n, ok := a.CacheGen(rt.Host); !ok || n != g.N {
+ t.Fatalf("CacheGen = %d, %v; want %d", n, ok, g.N)
+ }
+ if perGens, _ := gb.seen(); !slices.Equal(perGens, []bool{true}) {
+ t.Fatalf("StartGen perGen = %v, want [true]", perGens)
+ }
+}
+
+// Flag off: one generation per entry, on the canonical socket.
+func TestAppSocketIsTheCanonicalSocketWithGenerationsOff(t *testing.T) {
+ a, gb := genActivator(t, false)
+ rt := genTestRoute(t, "off.example.com")
+ arrangeAwake(t, a, rt)
+ if sock, ok := a.AppSocket(rt.Host); !ok || sock != rt.Socket {
+ t.Fatalf("AppSocket = %q, %v; want the canonical %q", sock, ok, rt.Socket)
+ }
+ if perGens, _ := gb.seen(); !slices.Equal(perGens, []bool{false}) {
+ t.Fatalf("StartGen perGen = %v, want [false]", perGens)
+ }
+}
+
+// A backend without the capability (UnitBackend, older doubles) still gets a
+// numbered, canonical handle.
+func TestPlainBackendGetsTheCanonicalHandle(t *testing.T) {
+ withGenerations(t, true)
+ a, _, _, _ := testActivator(t, time.Hour)
+ rt := genTestRoute(t, "plain.example.com")
+ arrangeAwake(t, a, rt)
+ g, ok := a.ServingGen(rt.Host)
+ if !ok || g.Socket != rt.Socket || g.N == 0 {
+ t.Fatalf("ServingGen = %+v, %v; want the canonical socket and a number", g, ok)
+ }
+}
+
+// Spec §12 "Generation identity": a stop, then a crash, then a wake never
+// reuses a generation number. CacheGen keeps the last number while the
+// instance sleeps.
+func TestGenerationNumbersNeverRepeatAcrossStopCrashAndWake(t *testing.T) {
+ a, gb := genActivator(t, true)
+ rt := genTestRoute(t, "ident.example.com")
+ arrangeAwake(t, a, rt)
+ g1, _ := a.ServingGen(rt.Host)
+
+ stopAndWait(t, a, rt.Host)
+ if _, ok := a.AppSocket(rt.Host); ok {
+ t.Fatal("AppSocket answered for a sleeping instance")
+ }
+ if n, ok := a.CacheGen(rt.Host); !ok || n != g1.N {
+ t.Fatalf("CacheGen while asleep = %d, %v; want the last current generation %d", n, ok, g1.N)
+ }
+
+ arrangeAwake(t, a, rt)
+ g2, _ := a.ServingGen(rt.Host)
+ if g2.N <= g1.N {
+ t.Fatalf("wake after a stop took generation %d after %d", g2.N, g1.N)
+ }
+
+ gb.fakeBackend.mu.Lock()
+ crashed := gb.fakeBackend.procs[len(gb.fakeBackend.procs)-1]
+ gb.fakeBackend.mu.Unlock()
+ crashed.exit()
+ arrangeAwake(t, a, rt) // finds the dead child, stops it, wakes a fresh one
+ g3, _ := a.ServingGen(rt.Host)
+ if g3.N <= g2.N {
+ t.Fatalf("wake after a crash took generation %d after %d", g3.N, g2.N)
+ }
+ if n, _ := a.CacheGen(rt.Host); n != g3.N {
+ t.Fatalf("CacheGen = %d, want %d", n, g3.N)
+ }
+}
+
+// The sleep-time next-due probe dials AppSocket: the generation socket with
+// the switch on, the canonical one with it off.
+func TestNextDueAtStopDialsTheCurrentGeneration(t *testing.T) {
+ for _, on := range []bool{true, false} {
+ t.Run(fmt.Sprintf("generations=%v", on), func(t *testing.T) {
+ a, gb := genActivator(t, on)
+ rt := genTestRoute(t, "due.example.com")
+ arrangeAwake(t, a, rt)
+ g, _ := a.ServingGen(rt.Host)
+ stopAndWait(t, a, rt.Host)
+ want := rt.Socket
+ if on {
+ want = fmt.Sprintf("/gen/%d.sock", g.N)
+ }
+ if _, dueSocks := gb.seen(); !slices.Equal(dueSocks, []string{want}) {
+ t.Fatalf("next-due probe dialled %v, want [%s]", dueSocks, want)
+ }
+ })
+ }
+}
+
+// A retire hook hears about a generation that stopped being current, with
+// the switch on, and never with it off (the canonical path is shared by every
+// generation then).
+func TestGenRetiredFiresAtStopOnlyWithGenerationsOn(t *testing.T) {
+ for _, on := range []bool{true, false} {
+ t.Run(fmt.Sprintf("generations=%v", on), func(t *testing.T) {
+ a, _ := genActivator(t, on)
+ var mu sync.Mutex
+ var got []string
+ a.OnGenRetired(func(host, socket string) {
+ mu.Lock()
+ got = append(got, host+" "+socket)
+ mu.Unlock()
+ })
+ rt := genTestRoute(t, "ret.example.com")
+ arrangeAwake(t, a, rt)
+ g, _ := a.ServingGen(rt.Host)
+ stopAndWait(t, a, rt.Host)
+ mu.Lock()
+ defer mu.Unlock()
+ var want []string
+ if on {
+ want = []string{rt.Host + " " + g.Socket}
+ }
+ if !slices.Equal(got, want) {
+ t.Fatalf("retire hooks = %v, want %v", got, want)
+ }
+ })
+ }
+}
+
+// A worker binds nothing, so it has no app socket for traffic.
+func TestAppSocketIsFalseForAWorker(t *testing.T) {
+ a, _ := genActivator(t, true)
+ rt := testWorkerRoute("bot.x.internal")
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ waitFor(t, "the worker to wake", func() bool { return a.Owns(rt.Host) })
+ if sock, ok := a.AppSocket(rt.Host); ok {
+ t.Fatalf("AppSocket(worker) = %q, want false", sock)
+ }
+}
diff --git a/internal/edge/gensource.go b/internal/edge/gensource.go
new file mode 100644
index 000000000..02890878a
--- /dev/null
+++ b/internal/edge/gensource.go
@@ -0,0 +1,39 @@
+package edge
+
+// GenInfo is what the edge needs to know about an instance's CURRENT
+// generation (zero-downtime spec §7.1, §7.2): the socket that takes new
+// connections, the process-wide generation number the cache tags entries
+// with, and the build X-Carlos-Version reports.
+type GenInfo struct {
+ Socket string
+ N uint64
+ Build string
+}
+
+// GenSource is the activator's generation view, as the edge sees it. It is
+// declared here for the reason Activator is: internal/activator imports
+// this package, never the reverse. *activator.Service implements it.
+//
+// The edge uses it only when GenerationsOn reports true. With it false
+// (CARLOS_SWAP unset, the default) every path below is skipped and the edge
+// proxies, caches and stamps versions exactly as before.
+type GenSource interface {
+ // GenerationsOn is the activator's dormant switch, read once when the
+ // activator was built.
+ GenerationsOn() bool
+ // AppSocket names the current generation's socket for traffic. False
+ // when the instance has no current generation (asleep, or waking).
+ AppSocket(host string) (socket string, ok bool)
+ // ServingGen is AppSocket with the generation's number and build.
+ ServingGen(host string) (GenInfo, bool)
+ // CacheGen is the newest generation that has been current for host in
+ // this process, kept after the instance sleeps so a sleeping instance's
+ // cached pages still answer. False when no generation of host has ever
+ // been current here. It must not take any lock the activator holds
+ // while calling into the cache.
+ CacheGen(host string) (uint64, bool)
+ // OnGenRetired registers fn to be called, synchronously and outside
+ // every activator lock, each time a generation stops being current.
+ // fn must not block.
+ OnGenRetired(fn func(host, socket string))
+}