amadan

carlos / platform Public

zdd plan 3 task 10: claims-on generation types and refusals; a retiring claimed generation drains first

Reduced scope (controller ruling P2-9): every claims-on launch is closed on
this base, so this builds the launch-independent parts plan 5 compiles
against and makes every claims-on swap an explicit abandonment.

- StartPolicy.Gen/Fallback/HealthBy/Gate/PinCheck, ClaimStart.Gen,
  ClaimRuntime.Generations, actEntry.pinFailed/pinFailReason.
- startAdmitted keeps main's RoleLauncher.Start line, settle/producer,
  retry loop and s.Failed. Its before hook runs Gate and PinCheck and
  marks a pinned wake's exec (markPinExec); a refusal before the hook is
  claimAdmission's (stepStale); HealthBy caps Ready; pinFailed is recorded
  only for a stepFailed start that ran (cmd != nil).
- The pre-Restore generation check through genMatches (number, socket and
  control socket).
- doClaimWake: errWakeUnproven at its head, policy.Gate, and policy.Gen
  from claimGenAt when gens and Generations are both on.
- claimGenAt, ExecBackend/SplitBackend.GenPaths.
- claimProcess.Retire and stopMember's roleRetiring claims arm (index P6
  #4); a SIGTERM error other than ErrProcessDone is logged.
- errClaimsNoGenerations replaces errHoldNeedsPins; swapOverlap (before
  the admission and raise) and swapHold (before anything stops) refuse a
  claims-on runtime without Generations and Pins.

Not built, owed to the plan that opens the isolated launch: entryParent
and the strict parent's life, setParent's claims arm, the Fallback
re-Prepare, the claimWakeSpawn seam. The retry's reuse of policy.Gen.N
(PrepareGen accepting an existing empty g<n>) is unchecked for the same
reason: nothing reaches it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 86e307018e721d80e3aeb7b1d54665a0662285a9 parent b7bd320
10 files changed, +626 −35
  • internal/activator/activator.go +11 −1
  • internal/activator/claimgen.go +24 −0
  • internal/activator/claimgen_test.go +356 −0
  • internal/activator/claimwake.go +163 −7
  • internal/activator/genbackend.go +34 −0
  • internal/activator/opset.go +14 −3
  • internal/activator/swap.go +8 −1
  • internal/activator/swap_hold_test.go +6 −20
  • internal/activator/swap_overlap_test.go +5 −0
  • internal/activator/swaphold.go +5 −3
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 9d6af7eb6..39a404c18 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -454,6 +454,13 @@ type actEntry struct {
// holdNext is a hold swap waiting for this entry's stop to finish.
// finishStop installs its pinned wake. Written under a.mu.
holdNext *holdSpec
+ // pinFailed is the pinned build a claims-on start ran and saw fail
+ // before it was ready, and pinFailReason why (zero-downtime spec §8.2).
+ // startAdmitted writes them under a.mu, only for a failure classifyStep
+ // calls the build's; plan 5 reports them. Nil when the pinned build
+ // served, or never ran.
+ pinFailed *Gen
+ pinFailReason string
// settleCancel, on a hold canary's pinned entry, ends its settle
// observation. beginStopLocked calls it while the process has not
// exited, so an intentional stop is never read as a crash (spec §6.4).
@@ -1239,7 +1246,10 @@ func (a *Service) wake(e *actEntry) {
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.
+ // startGen. With generations on and a runtime that honours
+ // StartPolicy.Gen it carries the handle the provider bound (Task
+ // 10); otherwise 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)
diff --git a/internal/activator/claimgen.go b/internal/activator/claimgen.go
new file mode 100644
index 000000000..fe8fe390f
--- /dev/null
+++ b/internal/activator/claimgen.go
@@ -0,0 +1,24 @@
+package activator
+
+import (
+ "errors"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// errClaimsNoGenerations refuses a claims-on swap whose claim runtime cannot
+// start a generation of its own (ClaimRuntime.Generations and Pins). It is
+// returned before anything is raised, stopped or spawned.
+var errClaimsNoGenerations = errors.New("swap: claims are on and the claim runtime cannot start a generation of its own")
+
+// claimGenAt is generation n's handle for a claims-on spawn: the generation
+// sockets where the row has them, and the canonical socket (with a real
+// number) where it does not.
+func (a *Service) claimGenAt(rt registry.Route, n uint64) Gen {
+ if gp, ok := a.backend.(genPather); ok {
+ if g, ok := gp.GenPaths(rt, n); ok {
+ return g
+ }
+ }
+ return Gen{N: n, Socket: rt.Socket, Control: rt.Socket + ".control"}
+}
diff --git a/internal/activator/claimgen_test.go b/internal/activator/claimgen_test.go
new file mode 100644
index 000000000..0ab304ca0
--- /dev/null
+++ b/internal/activator/claimgen_test.go
@@ -0,0 +1,356 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os/exec"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// claimedStartFor is a ClaimStart for obs with every hook a no-op, version
+// v. It is the one ClaimStart these tests build: main's claims-on launch
+// needs a RoleLauncher and CurrentAttempt for valid() (claimwindow_test.go),
+// and on this base RoleLaunch.Start refuses every shape before its before
+// hook runs (index P2-9).
+func claimedStartFor(obs hostclaim.Observed, work hostwrite.Work, v string) ClaimStart {
+ return ClaimStart{RoleLauncher: &RoleLaunch{}, CurrentAttempt: func() hostclaim.WakeRetryAttempt { return hostclaim.WakeRetryAttempt{} },
+ Host: obs.Host, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"}, Version: v,
+ Restore: func(context.Context) error { return nil },
+ Materialize: func(context.Context) error { return nil }, Build: func() *exec.Cmd { return nil },
+ Ready: func(context.Context, *exec.Cmd) error { return nil }, Reserve: func(context.Context, int) error { return nil },
+ Failed: func(context.Context, int) error { return nil }, Healthy: func(context.Context, int, *exec.Cmd) error { return nil },
+ Replicate: func(context.Context, string) (Rep, error) { return nil, nil },
+ Publish: func(context.Context, hostclaim.Observed) (hostclaim.Observed, error) { return obs, nil },
+ Retain: func(Proc, Rep, error) {}}
+}
+
+func claimedObs(host string) hostclaim.Observed {
+ return hostclaim.Observed{Host: host, ETag: "one", Claim: &hostclaim.Claim{
+ Holder: &hostclaim.Holder{Location: "instances/" + host + "/inc", Session: 1, Generation: "one"},
+ Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"},
+ }}
+}
+
+// Without a runtime that builds per-generation starts, a claims-on swap is
+// refused before anything is raised, stopped or spawned.
+func TestClaimsOnSwapNeedsGenerations(t *testing.T) {
+ for _, mode := range []SwapMode{ModeOverlap, ModeHold} {
+ f := newSwapFixture(t)
+ f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: &swapWork{}}}
+ f.a.cfg.Claims = &ClaimRuntime{Pins: true} // Generations false
+ res := f.a.Swap(context.Background(), f.attempt(mode), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errClaimsNoGenerations.Error()) {
+ t.Fatalf("%s: Swap = %+v", mode, res)
+ }
+ if gens, raises, _ := f.b.calls(); gens != 0 || raises != 0 {
+ t.Fatalf("%s: spawned %d, raised %d", mode, gens, raises)
+ }
+ if f.b.lastProc().isStopped() {
+ t.Fatalf("%s: the incumbent was stopped", mode)
+ }
+ }
+}
+
+// A provider that binds anything but the generation the activator chose is
+// refused before Restore touches the database. With generations off (a
+// zero Gen) nothing is checked, as today.
+func TestStartAdmittedRefusesAGenerationMismatch(t *testing.T) {
+ obs := claimedObs("x.example")
+ work, _ := (hostwrite.Config{}).Select(context.Background(), obs.Host, "spawn")
+ stop := errors.New("stop at restore")
+ want := Gen{N: 9, Socket: "/run/carlos/g/abc.9.sock", Control: "/run/carlos/g/abc.9.sock.control"}
+ for _, tc := range []struct {
+ name string
+ policy Gen
+ bound Gen
+ restored bool
+ }{
+ {"matching", want, want, true},
+ {"another socket", want, Gen{N: 9, Socket: "/run/carlos/x.example.sock", Control: want.Control}, false},
+ {"another control socket", want, Gen{N: 9, Socket: want.Socket, Control: "/run/carlos/x.example.sock.control"}, false},
+ {"another number", want, Gen{N: 10, Socket: want.Socket, Control: want.Control}, false},
+ {"generations off", Gen{}, Gen{}, true},
+ } {
+ restored := false
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box",
+ Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil },
+ Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
+ s := claimedStartFor(obs, work, "v1")
+ s.Gen = tc.bound
+ s.Restore = func(context.Context) error { restored = true; return stop }
+ return s, nil
+ }}}}
+ err := a.startAdmitted(context.Background(), &actEntry{host: obs.Host}, obs, startPolicy{MaxStarts: 1, Window: time.Second, Gen: tc.policy})
+ if restored != tc.restored {
+ t.Fatalf("%s: restored = %v, want %v (err %v)", tc.name, restored, tc.restored, err)
+ }
+ if !tc.restored && !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("%s: %v, want hostwrite.ErrUnavailable", tc.name, err)
+ }
+ if tc.restored && !errors.Is(err, stop) {
+ t.Fatalf("%s: %v, want the restore's own error", tc.name, err)
+ }
+ }
+}
+
+// The activator chooses claims-on generation sockets the way plan 1 lays
+// them out, and a canonical handle with a real number where there is no
+// generation path.
+func TestClaimGenAt(t *testing.T) {
+ b := testExecBackend(t)
+ b.RunDirOverride = t.TempDir()
+ a := New(Config{}, b, nil)
+ flat := registry.Route{Host: "c.example.com", Kind: "instance", DBPath: b.DataDir + "/c.example.com.db", Socket: b.RunDirOverride + "/c.example.com.sock"}
+ if g := a.claimGenAt(flat, 4); g.N != 4 || g.Socket != b.flatGenSocket(flat.Host, 4) || g.Control != g.Socket+".control" {
+ t.Fatalf("flat: %+v", g)
+ }
+ foreign := registry.Route{Host: "f.example.com", Kind: "instance", Socket: t.TempDir() + "/odd.sock"}
+ if g := a.claimGenAt(foreign, 5); g.N != 5 || g.Socket != foreign.Socket || g.Control != foreign.Socket+".control" {
+ t.Fatalf("foreign: %+v", g)
+ }
+ // The split backend asks its exec half; a unit-backed row has no
+ // generation path of its own.
+ split := New(Config{}, &SplitBackend{Exec: b}, nil)
+ if g := split.claimGenAt(flat, 6); g.N != 6 || g.Socket != b.flatGenSocket(flat.Host, 6) {
+ t.Fatalf("split, flat: %+v", g)
+ }
+ unit := flat
+ unit.Backing = registry.BackingUnit
+ if g := split.claimGenAt(unit, 7); g.N != 7 || g.Socket != unit.Socket {
+ t.Fatalf("split, unit: %+v", g)
+ }
+}
+
+// Spec §7.4 (index P2-9): every claims-on launch is closed on this base, and
+// the work's own validation refuses before the exec. Nothing ran, so the
+// attempt is abandoned, never SwapFailed: a failure record would halt the
+// cohort for a build that never executed.
+func TestAClaimsRefusalAtTheExecIsNoFailure(t *testing.T) {
+ for _, refusal := range []error{hostguard.ErrStampMismatch, hostguard.ErrFenced, hostguard.ErrNoClaim} {
+ f := newSwapFixture(t)
+ f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: &swapWork{}}}
+ f.a.cfg.Claims = &ClaimRuntime{Pins: true, Generations: true}
+ f.a.claimSpawn = func(context.Context, *actEntry, registry.Route, hostwrite.Work, Gen) (Gen, Proc, error) {
+ return Gen{}, nil, fmt.Errorf("work: %w", refusal)
+ }
+ if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapAbandoned {
+ t.Fatalf("%v at the exec: Swap = %+v, want abandoned", refusal, res)
+ }
+ }
+ // No seam: the successor's spawn is the closed launch itself. Its
+ // refusal comes before anything runs, so it is the admission's
+ // (claimAdmission), and the attempt is abandoned. The runtime is never
+ // read or prepared: there is no launch to prepare for.
+ f := newSwapFixture(t)
+ f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: &swapWork{}}}
+ f.a.cfg.Claims = &ClaimRuntime{Pins: true, Generations: true,
+ Read: func(context.Context, string) (hostclaim.Observed, error) {
+ return hostclaim.Observed{}, errors.New("s3: GetObject: request timeout")
+ },
+ Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
+ t.Fatal("prepared a successor on a closed launch")
+ return ClaimStart{}, nil
+ }}
+ if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapAbandoned {
+ t.Fatalf("a closed launch: Swap = %+v, want abandoned", res)
+ }
+}
+
+// Index P2-9: a claims-on overlap whose runtime has Generations and Pins,
+// with no test seam standing in for the launch, passes every check, raises
+// the parent, and is refused at the spawn by the closed launch. It is
+// abandoned with nothing spawned, the raise undone, the incumbent still
+// serving and no failure recorded (SwapFailed would halt the cohort).
+func TestAClaimsOnOverlapOnAClosedLaunchIsAbandoned(t *testing.T) {
+ f := newSwapFixture(t)
+ w := &swapWork{}
+ f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: w}}
+ f.a.cfg.Claims = &ClaimRuntime{Pins: true, Generations: true}
+ res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, ErrClaimIsolationUnavailable.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned on the closed launch", res)
+ }
+ if !strings.Contains(res.Reason, errClaimsAdmission.Error()) {
+ t.Fatalf("Reason %q: the closed launch is the admission's refusal", res.Reason)
+ }
+ gens, raises, restores := f.b.calls()
+ if gens != 0 || raises != 1 || restores != 1 {
+ t.Fatalf("StartStaged %d, raises %d, restores %d: want nothing spawned and the parent restored", gens, raises, restores)
+ }
+ if f.b.lastProc().isStopped() {
+ t.Fatal("the incumbent was stopped")
+ }
+ if got := f.appSocket(t); got != f.incumbent.Socket {
+ t.Fatalf("AppSocket = %q, want the incumbent", got)
+ }
+ f.a.mu.Lock()
+ e := f.a.entries[f.host]
+ state, pinFailed := e.state, e.pinFailed
+ f.a.mu.Unlock()
+ if state != actAwake || pinFailed != nil {
+ t.Fatalf("entry state %v, pinFailed %+v: the incumbent should serve on, with nothing recorded", state, pinFailed)
+ }
+}
+
+// Index P2-9: a claims-on hold whose runtime cannot start a generation of
+// its own is abandoned before the incumbent stops. A hold is stop-then-start
+// (spec §7.8): refused after the stop, it would only cost the instance its
+// running process, since the closed launch could never start the pin.
+func TestAClaimsOnHoldWithoutGenerationsIsAbandonedBeforeTheStop(t *testing.T) {
+ f := newSwapFixture(t)
+ f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: &swapWork{}}}
+ touched := false
+ f.a.cfg.Claims = &ClaimRuntime{Box: "box", Pins: true, // Generations false
+ Read: func(context.Context, string) (hostclaim.Observed, error) {
+ touched = true
+ return claimedObs(f.host), nil
+ },
+ Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
+ touched = true
+ return ClaimStart{}, nil
+ }}
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errClaimsNoGenerations.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned: %v", res, errClaimsNoGenerations)
+ }
+ if f.b.lastProc().isStopped() {
+ t.Fatal("the incumbent was stopped by a hold that could never start its pin")
+ }
+ if ev := f.b.events(); indexPrefix(ev, "stop:") >= 0 {
+ t.Fatalf("events %v: the hold began a stop", ev)
+ }
+ if touched {
+ t.Fatal("the claim runtime was read or prepared for a refused hold")
+ }
+ f.a.mu.Lock()
+ e := f.a.entries[f.host]
+ state, held := e.state, e.holdNext
+ f.a.mu.Unlock()
+ if state != actAwake || held != nil {
+ t.Fatalf("entry state %v, holdNext %v: want awake, no hold", state, held)
+ }
+}
+
+// Spec §6.4, §7.4 (index P2-9): a claimed planned wake on this base meets
+// the closed launch, which refuses before its exec hook. That refusal is the
+// admission's (claimAdmission), so it is stale, never a failed build: no
+// pinFailed, no fallback. Prepare and Restore ran once.
+func TestAClaimedPlannedWakeOnAClosedLaunchRecordsNoFailure(t *testing.T) {
+ obs := claimedObs("x.example")
+ work, _ := (hostwrite.Config{}).Select(context.Background(), obs.Host, "spawn")
+ prepares, restores, failed := 0, 0, 0
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box",
+ Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil },
+ Prepare: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (ClaimStart, error) {
+ prepares++
+ s := claimedStartFor(obs, work, p.Pin.Build)
+ s.Restore = func(context.Context) error { restores++; return nil }
+ s.Failed = func(context.Context, int) error { failed++; return nil }
+ return s, nil
+ }}}}
+ e := &actEntry{host: obs.Host}
+ gateCalls := 0
+ policy := startPolicy{MaxStarts: 1, Window: time.Second, Pin: &Gen{Build: "v2"}, Fallback: &Gen{Build: "v1"},
+ HealthBy: time.Now().Add(time.Second),
+ Gate: func() error { gateCalls++; return nil },
+ PinCheck: func(Gen) error { gateCalls++; return nil }}
+ err := a.startAdmitted(context.Background(), e, obs, policy)
+ if !errors.Is(err, ErrClaimIsolationUnavailable) || !errors.Is(err, errClaimsAdmission) {
+ t.Fatalf("startAdmitted = %v, want the closed launch as the admission's refusal", err)
+ }
+ if got := classifyStep(nil, err); got != stepStale {
+ t.Fatalf("classifyStep = %v, want stepStale", got)
+ }
+ if prepares != 1 || restores != 1 {
+ t.Fatalf("Prepare ran %d times, Restore %d: want once each, and no fallback", prepares, restores)
+ }
+ if failed != 1 {
+ t.Fatalf("Failed ran %d times: main's retry accounting is unchanged", failed)
+ }
+ if e.pinFailed != nil || e.pinFailReason != "" {
+ t.Fatalf("pinFailed = %+v (%q): a refused launch was recorded as a failed build", e.pinFailed, e.pinFailReason)
+ }
+ if gateCalls != 0 {
+ t.Fatalf("Gate/PinCheck ran %d times: the closed launch never reaches its exec hook", gateCalls)
+ }
+ if e.pinExec {
+ t.Fatal("pinExec set: nothing reached the exec")
+ }
+}
+
+// Spec §7.4 (Task 2): a failed wake whose generation is not proven stopped
+// still holds the host's lease. A claims-on wake is refused at its head, as
+// doWake refuses one, before the claim is even read.
+func TestAClaimedWakeRefusesWhileTheLastFailedWakeIsUnproven(t *testing.T) {
+ read := false
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box",
+ Read: func(context.Context, string) (hostclaim.Observed, error) {
+ read = true
+ return claimedObs("x.example"), nil
+ },
+ StopForeign: func(context.Context, string) error { return nil }}}}
+ a.unprovenWakes = map[string]*unprovenWake{"x.example": {}}
+ e := &actEntry{host: "x.example", route: registry.Route{Host: "x.example", Kind: "instance"}}
+ if _, err := a.doClaimWake(context.Background(), e); !errors.Is(err, errWakeUnproven) {
+ t.Fatalf("doClaimWake = %v, want errWakeUnproven", err)
+ }
+ if read {
+ t.Fatal("the claim was read for a host whose last failed wake is unproven")
+ }
+}
+
+// orderContainment records whether the leader had already exited when
+// KillAndEmpty ran, then kills and waits like the real one.
+type orderContainment struct {
+ p *claimProcess
+ exitedAtKill bool
+}
+
+func (c *orderContainment) KillAndEmpty(context.Context) error {
+ c.exitedAtKill = c.p.Exited()
+ if c.p.cmd.Process != nil {
+ _ = c.p.cmd.Process.Kill()
+ }
+ <-c.p.done
+ return nil
+}
+
+func (c *orderContainment) Close() error { return nil }
+
+// Spec §7.5 step 6 (index P6 #4): a retiring claims-on generation gets
+// SIGTERM and its drain before KillAndEmpty, so its in-flight requests
+// finish. An unready candidate is still killed at once. Stop alone
+// (cgroup.kill first) killed the retiring generation's requests.
+func TestARetiringClaimedGenerationDrainsBeforeItsKill(t *testing.T) {
+ if _, err := exec.LookPath("sleep"); err != nil {
+ t.Skip("no sleep")
+ }
+ for _, tc := range []struct {
+ role genRole
+ drained bool
+ }{{roleRetiring, true}, {roleSuccessor, false}} {
+ cmd := exec.Command("sleep", "30")
+ if err := cmd.Start(); err != nil {
+ t.Fatal(err)
+ }
+ p := newClaimProcess(cmd, nil)
+ c := &orderContainment{p: p}
+ p.prepared = c
+ a := &Service{}
+ e := &actEntry{host: "claims.example.com"}
+ if !a.stopMember(e, &genMember{gen: Gen{N: 2}, role: tc.role, proc: p}) {
+ t.Fatalf("%s: stop not proven", tc.role)
+ }
+ if c.exitedAtKill != tc.drained {
+ t.Fatalf("%s: leader exited before KillAndEmpty = %v, want %v", tc.role, c.exitedAtKill, tc.drained)
+ }
+ }
+}
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
index 80434b07f..a2f1dece7 100644
--- a/internal/activator/claimwake.go
+++ b/internal/activator/claimwake.go
@@ -3,8 +3,12 @@ package activator
import (
"context"
"errors"
+ "fmt"
+ "log"
+ "os"
"os/exec"
"sync"
+ "syscall"
"time"
"github.com/carlosframework/platform/internal/cgroup"
@@ -30,6 +34,29 @@ type StartPolicy struct {
// Pin.Digest. startAdmitted refuses a ClaimStart whose Version is not
// Pin.Build.
Pin *Gen
+ // Gen is the generation the activator chose for this start: its number
+ // and the sockets the provider must bind (zero-downtime plan 3, Task
+ // 10). Zero is today's start on the canonical socket. startAdmitted
+ // refuses a ClaimStart that bound anything else (genMatches).
+ Gen Gen
+ // Fallback is a first wake's good build (zero-downtime spec §8.2), set
+ // only beside Pin, by plan 5's planner. It is honoured once the isolated
+ // launch opens; on this base the launch refuses before anything runs
+ // (stepStale), so there is nothing to fall back from.
+ Fallback *Gen
+ // HealthBy, when set, caps the pinned start's Ready window: a first
+ // wake's health check ends at the earlier of Window and HealthBy (spec
+ // §8.2: the waiters' deadline minus 25 s).
+ HealthBy time.Time
+ // Gate, when set, is the attempt gate (taking a.mu itself): a claims-on
+ // hold's pinCheck, or a planned wake's ownership check (plan 5).
+ // startAdmitted runs it in the launch's before hook, immediately before
+ // every exec, after the admission and Restore (spec §6.3, §7.4).
+ Gate func() error
+ // PinCheck, when set, is the planner's selection gate (plan 5's
+ // WakePlan.Check) for the pin about to exec. startAdmitted runs it
+ // beside Gate, at every exec.
+ PinCheck func(pin Gen) error
}
type startPolicy = StartPolicy
@@ -48,6 +75,12 @@ type ClaimRuntime struct {
// refused on a claims-on box whose runtime does not, before anything
// stops.
Pins bool
+ // Generations reports that Prepare honours StartPolicy.Gen (binds its
+ // sockets and reports them in ClaimStart.Gen). Without it a claims-on
+ // box keeps one generation per entry, on the canonical socket, and
+ // every swap is refused before anything is raised, stopped or spawned
+ // (errClaimsNoGenerations).
+ Generations bool
}
// ClaimStart is one admitted session. Its closures retain the admission tuple.
@@ -79,6 +112,16 @@ type ClaimStart struct {
// Retain owns uncertain process or replica cleanup until it can be proved.
Retain func(Proc, Rep, error)
Version string
+ // Gen is what the provider bound. startAdmitted refuses a start whose
+ // number, socket or control socket is not StartPolicy.Gen's.
+ Gen Gen
+}
+
+// genMatches reports whether a provider bound exactly the generation the
+// activator chose: its number, its socket and its control socket. A handle
+// that names a socket nobody serves would route requests nowhere.
+func genMatches(want, got Gen) bool {
+ return want.N == got.N && want.Socket == got.Socket && want.Control == got.Control
}
func (s ClaimStart) valid() bool {
@@ -92,6 +135,18 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance,
if r == nil || r.Box == "" || r.Read == nil || r.StopForeign == nil {
return nil, hostwrite.ErrUnavailable
}
+ // A failed wake whose generation is not proven stopped still holds the
+ // host, and its process may still run against the database
+ // (zero-downtime spec §7.4). As doWake does, no new wake starts until
+ // the sweep proves it, and a stray legacy sidecar refuses it the same
+ // way.
+ a.mu.Lock()
+ _, unproven := a.unprovenWakes[e.host]
+ unproven = unproven || len(a.straySidecars[e.host]) > 0
+ a.mu.Unlock()
+ if unproven {
+ return nil, fmt.Errorf("%s: refusing to wake: %w", e.host, errWakeUnproven)
+ }
observed, err := r.Read(ctx, e.host)
if err != nil {
return nil, err
@@ -131,15 +186,21 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance,
policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Pin: e.pin}
if e.pin != nil && e.pinCheck != nil {
// A claims-on hold's pinned wake: the attempt gate before anything
- // is admitted. Task 10 runs it again inside the work at every exec,
+ // is admitted, and again in the launch's before hook at every exec,
// after the admission and Restore (spec §6.3, §7.4), as
// startPinnedChecked does for a claims-off one.
- a.mu.Lock()
- err := e.pinCheck()
- a.mu.Unlock()
- if err != nil {
+ gate := func() error {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return e.pinCheck()
+ }
+ if err := gate(); err != nil {
return nil, err
}
+ policy.Gate = gate
+ }
+ if a.gens && r.Generations {
+ policy.Gen = a.claimGenAt(rt, nextGen())
}
if err := r.Prerequisites(ctx, rt, observed, policy); err != nil {
return nil, err
@@ -198,6 +259,12 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
// refused before Restore touches the database.
return hostwrite.ErrUnavailable
}
+ if policy.Gen.N != 0 && !genMatches(policy.Gen, s.Gen) {
+ // The provider bound something other than the generation the
+ // activator chose: its handle would name a socket nobody serves.
+ // Refused before Restore touches the database.
+ return hostwrite.ErrUnavailable
+ }
if s.Live != nil {
if err := s.Live.Err(); err != nil {
return err
@@ -250,8 +317,38 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
}
// This hook runs after strict preparation and immediately before Start.
// Restore and materialization consume only the separate cut deadline.
- beforeStart := func() error { return budget.start(ctx, s.OpenProbe) }
+ reached := false // the exec hook ran: an error before it is the admission's (claimAdmission)
+ beforeStart := func() error {
+ reached = true
+ if policy.Gate != nil {
+ // The attempt gate at the exec: a claims-on hold's pinCheck,
+ // or a planned wake's ownership check (plan 5).
+ if err := policy.Gate(); err != nil {
+ return err
+ }
+ }
+ if policy.PinCheck != nil && policy.Pin != nil {
+ // The planner's selection gate for the build about to exec
+ // (plan 5): a failure recorded since planning holds it.
+ if err := policy.PinCheck(*policy.Pin); err != nil {
+ return err
+ }
+ }
+ if err := budget.start(ctx, s.OpenProbe); err != nil {
+ return err
+ }
+ if e.pin != nil {
+ // A hold's pinned wake has reached its build's exec: from
+ // here a failure is the build's (Task 6, markPinExec).
+ a.markPinExec(e)
+ }
+ return nil
+ }
cmd, prepared, settled, producer, startErr := s.RoleLauncher.Start(ctx, "main", s.CurrentAttempt(), s.Build, beforeStart)
+ // A refusal that came back before the exec hook (the closed isolated
+ // launch, a claim read, an absent claim, a changed stamp) ran nothing
+ // of the build: it is the admission's, never the build's failure.
+ startErr = claimAdmission(startErr, reached)
p := newClaimProcess(cmd, prepared)
p.settle = settled
@@ -261,6 +358,11 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
if policy.Probe {
deadline = budget.opened.Add(policy.Window)
}
+ if !policy.HealthBy.IsZero() && policy.HealthBy.Before(deadline) {
+ // A first wake's pinned build: health until the waiters'
+ // deadline minus 25 s (zero-downtime spec §8.2).
+ deadline = policy.HealthBy
+ }
readyCtx, cancelReady := context.WithDeadline(ctx, deadline)
startErr = s.Ready(readyCtx, cmd)
cancelReady()
@@ -288,7 +390,22 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
startErr = s.Healthy(ctx, attempt, cmd)
}
if startErr == nil {
- e.claimResult = &wokenInstance{proc: p, rep: rep, spawnVersion: s.Version, stopOutcome: &stopObservation{}}
+ a.mu.Lock()
+ if e.pinFailed != nil && policy.Pin != nil && e.pinFailed.Build == policy.Pin.Build {
+ e.pinFailed, e.pinFailReason = nil, "" // a retry of the same pin served
+ }
+ a.mu.Unlock()
+ // The generation's handle: the sockets the provider bound
+ // (genMatches, above). Zero with generations off, and wake()
+ // falls back to the canonical handle as before.
+ g := policy.Gen
+ if g.N != 0 {
+ g.Build = s.Version
+ if policy.Pin != nil {
+ g.StagedPath, g.Digest, g.Demand = policy.Pin.StagedPath, policy.Pin.Digest, policy.Pin.Demand
+ }
+ }
+ e.claimResult = &wokenInstance{proc: p, rep: rep, spawnVersion: s.Version, gen: g, stopOutcome: &stopObservation{}}
return nil
}
cleanup, cancelCleanup := context.WithTimeout(context.Background(), 10*time.Second)
@@ -303,6 +420,20 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
return errors.Join(startErr, err)
}
last = startErr
+ // THE outcome classifier (Task 4): a start the wake's lifecycle ended
+ // (maintenance, Shutdown, a stop), that the attempt gate refused, or
+ // that the launch refused before its exec hook is no failure of the
+ // build and records nothing (spec §6.4, §7.4). Only a pinned build
+ // that ran (cmd) and failed before it was ready is recorded, for plan
+ // 5's events (a failed first wake, spec §8.2).
+ if policy.Pin != nil && !ready && cmd != nil && classifyStep(ctx, startErr) == stepFailed {
+ a.mu.Lock()
+ if e.pinFailed == nil {
+ failed := *policy.Pin
+ e.pinFailed, e.pinFailReason = &failed, startErr.Error()
+ }
+ a.mu.Unlock()
+ }
// A replication/publication failure has an uncertain remote outcome.
// The next process start cannot reuse that opening as another attempt.
if ready {
@@ -408,6 +539,31 @@ func (p *claimProcess) Stop(ctx context.Context) error {
return nil
}
+// Retire is a serving claims-on generation's retirement (zero-downtime spec
+// §7.5 step 6, index P6 #4): SIGTERM to its leader, the app's own drain for
+// up to grace, then Stop, whose KillAndEmpty on the generation's own child
+// and the emptiness proof are unchanged. A candidate that never became
+// ready is not retired: its owner Stops it, which kills at once.
+func (p *claimProcess) Retire(ctx context.Context, grace time.Duration) error {
+ if p.cmd != nil && p.cmd.Process != nil && !p.Exited() {
+ err := p.cmd.Process.Signal(syscall.SIGTERM)
+ switch {
+ case err == nil, errors.Is(err, os.ErrProcessDone):
+ t := time.NewTimer(grace)
+ select {
+ case <-p.done:
+ case <-t.C:
+ case <-ctx.Done():
+ }
+ t.Stop()
+ default:
+ // No drain is possible; Stop still kills and proves it.
+ log.Printf("activator: SIGTERM to a retiring claims-on generation (pid %d): %v; stopping it without a drain", p.cmd.Process.Pid, err)
+ }
+ }
+ return p.Stop(ctx)
+}
+
// provenStopped is a claims-on generation's termination proof (zero-downtime
// spec §7.4): its Stop has completed, which takes KillAndEmpty on its own
// child, the exit, the settle and the close. Stop is idempotent once
diff --git a/internal/activator/genbackend.go b/internal/activator/genbackend.go
index 8e656ea8e..96d4a55a3 100644
--- a/internal/activator/genbackend.go
+++ b/internal/activator/genbackend.go
@@ -324,3 +324,37 @@ func (b *ExecBackend) retiredGate(host, control string) func() bool {
}
return func() bool { return b.Retired(host, control) }
}
+
+// genPather is a backend that can name a generation's sockets.
+type genPather interface {
+ GenPaths(rt registry.Route, n uint64) (Gen, bool)
+}
+
+// GenPaths is the generation socket pair a spawn of generation n of rt
+// binds, as plan 1 lays them out. It reports false for a row with no
+// generation path (foreign, fenced, conflict).
+func (b *ExecBackend) GenPaths(rt registry.Route, n uint64) (Gen, bool) {
+ p := b.pathsFor(rt)
+ var sock string
+ switch p.layout {
+ case LayoutFlat:
+ sock = b.flatGenSocket(rt.Host, n)
+ case LayoutNested:
+ sock = nestedGenSocketIn(p.runDir, n)
+ default:
+ return Gen{}, false
+ }
+ return Gen{N: n, Socket: sock, Control: sock + ".control"}, true
+}
+
+// GenPaths forwards to the exec half for exec rows. A unit row has no
+// generation path of its own.
+func (b *SplitBackend) GenPaths(rt registry.Route, n uint64) (Gen, bool) {
+ if rt.Backing == registry.BackingUnit {
+ return Gen{}, false
+ }
+ if gp, ok := b.Exec.(genPather); ok {
+ return gp.GenPaths(rt, n)
+ }
+ return Gen{}, false
+}
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index 096d69f94..7eda235a5 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -145,9 +145,20 @@ func (a *Service) stopMember(e *actEntry, m *genMember) bool {
if m.proc == nil {
return a.handlelessGenEmpty(rt, m.gen)
}
- ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
- defer cancel()
- err := m.proc.Stop(ctx)
+ var err error
+ if cp, ok := m.proc.(*claimProcess); ok && m.role == roleRetiring {
+ // A serving claims-on generation drains like a flat one (spec §7.5
+ // step 6, index P6 #4): SIGTERM and actStopTimeout first, then
+ // KillAndEmpty and the proof, with their own bound. An unready
+ // candidate takes the arm below: Stop kills it at once.
+ ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout+instancectl.DeadlineControl)
+ err = cp.Retire(ctx, actStopTimeout)
+ cancel()
+ } else {
+ ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ err = m.proc.Stop(ctx)
+ cancel()
+ }
if errors.Is(err, ErrStillRunning) {
return false
}
diff --git a/internal/activator/swap.go b/internal/activator/swap.go
index bd46a8f6d..5dba4d72f 100644
--- a/internal/activator/swap.go
+++ b/internal/activator/swap.go
@@ -294,7 +294,9 @@ func (a *Service) genHealthy(ctx context.Context, g Gen) error {
// root launcher cannot carry claim reservations. So a claims-on spawn is
// refused before anything runs, as a claims admission (NoOwnership, the
// attempt abandoned), and is never sent down the claims-off StartStaged.
-// claimSpawn is the test seam that stands in for the open launch.
+// swapOverlap has already refused a runtime without Generations and Pins
+// (errClaimsNoGenerations); one with both reaches this refusal. claimSpawn
+// is the test seam that stands in for the open launch.
func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen, gate func() error) (Gen, Proc, error) {
if hostwrite.Legacy(work) {
if err := gate(); err != nil {
@@ -327,6 +329,11 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n
if err != nil {
return abandon(fmt.Errorf("admission: %w", err))
}
+ if r := a.cfg.Claims; !hostwrite.Legacy(work) && (r == nil || !r.Generations || !r.Pins) {
+ // Claims on, and the runtime cannot start a generation of its own:
+ // refused before the admission and the raise (Task 10).
+ return abandon(errClaimsNoGenerations)
+ }
mctx, cancel := a.attemptCtx(ctx)
defer cancel()
m := &genMember{
diff --git a/internal/activator/swap_hold_test.go b/internal/activator/swap_hold_test.go
index 740987f0a..b6d41343e 100644
--- a/internal/activator/swap_hold_test.go
+++ b/internal/activator/swap_hold_test.go
@@ -4,7 +4,6 @@ import (
"context"
"errors"
"os"
- "os/exec"
"slices"
"strings"
"sync"
@@ -453,13 +452,13 @@ func TestSwapHoldRefusedAtTheStop(t *testing.T) {
}
}
-// Claims on: a hold needs a claim runtime that can pin a wake to a staged
-// build. Without one it is refused BEFORE anything stops.
+// Claims on: a hold needs a claim runtime that can start a pinned
+// generation of its own. Without one it is refused BEFORE anything stops.
func TestSwapHoldClaimsOnNeedsPins(t *testing.T) {
f := newSwapFixture(t)
f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: &swapWork{}}}
res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
- if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errHoldNeedsPins.Error()) {
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errClaimsNoGenerations.Error()) {
t.Fatalf("Swap = %+v", res)
}
if f.b.lastProc().isStopped() {
@@ -625,21 +624,6 @@ func TestDoClaimWakeRunsThePinGateFirst(t *testing.T) {
}
}
-// pinTestClaimStart is a ClaimStart that passes valid(): main's claims-on
-// launch needs a RoleLauncher and CurrentAttempt (claimwindow_test.go). On
-// this base RoleLaunch.Start always refuses, so no test here reaches it.
-func pinTestClaimStart(obs hostclaim.Observed, work hostwrite.Work, version string, restore func(context.Context) error) ClaimStart {
- return ClaimStart{RoleLauncher: &RoleLaunch{}, CurrentAttempt: func() hostclaim.WakeRetryAttempt { return hostclaim.WakeRetryAttempt{} },
- Host: obs.Host, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"}, Version: version,
- Restore: restore,
- Materialize: func(context.Context) error { return nil }, Build: func() *exec.Cmd { return nil },
- Ready: func(context.Context, *exec.Cmd) error { return nil }, Reserve: func(context.Context, int) error { return nil },
- Failed: func(context.Context, int) error { return nil }, Healthy: func(context.Context, int, *exec.Cmd) error { return nil },
- Replicate: func(context.Context, string) (Rep, error) { return nil, nil },
- Publish: func(context.Context, hostclaim.Observed) (hostclaim.Observed, error) { return obs, nil },
- Retain: func(Proc, Rep, error) {}}
-}
-
// A claim runtime that prepared a different build from the pinned one is
// refused before Restore touches the database.
func TestStartAdmittedRefusesAPinMismatch(t *testing.T) {
@@ -657,7 +641,9 @@ func TestStartAdmittedRefusesAPinMismatch(t *testing.T) {
a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box",
Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil },
Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
- return pinTestClaimStart(obs, work, tc.version, func(context.Context) error { restored = true; return stop }), nil
+ s := claimedStartFor(obs, work, tc.version)
+ s.Restore = func(context.Context) error { restored = true; return stop }
+ return s, nil
}}}}
policy := startPolicy{MaxStarts: 1, Window: time.Second, Pin: &Gen{Build: "v2"}}
err := a.startAdmitted(context.Background(), &actEntry{host: obs.Host}, obs, policy)
diff --git a/internal/activator/swap_overlap_test.go b/internal/activator/swap_overlap_test.go
index ec3f91a6b..8830b2c7b 100644
--- a/internal/activator/swap_overlap_test.go
+++ b/internal/activator/swap_overlap_test.go
@@ -396,6 +396,9 @@ func TestSwapClaimsOn(t *testing.T) {
t.Run(tc.name, func(t *testing.T) {
f := newSwapFixture(t)
f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: tc.sel()}
+ // A runtime that can start a generation of its own, so each row
+ // is refused where it names, not by errClaimsNoGenerations.
+ f.a.cfg.Claims = &ClaimRuntime{Pins: true, Generations: true}
if tc.wantGens > 0 {
useClaimSpawn(f)
}
@@ -419,6 +422,7 @@ func TestSwapClaimsOn(t *testing.T) {
f := newSwapFixture(t)
w := &swapWork{}
f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: w}}
+ f.a.cfg.Claims = &ClaimRuntime{Pins: true, Generations: true}
useClaimSpawn(f)
if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
t.Fatalf("Swap = %+v", res)
@@ -436,6 +440,7 @@ func TestSwapClaimsOn(t *testing.T) {
f := newSwapFixture(t)
w := &swapWork{}
f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: w}}
+ f.a.cfg.Claims = &ClaimRuntime{Pins: true, Generations: true}
res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, ErrClaimIsolationUnavailable.Error()) {
t.Fatalf("Swap = %+v, want abandoned: %v", res, ErrClaimIsolationUnavailable)
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go
index 340980cba..40877c47c 100644
--- a/internal/activator/swaphold.go
+++ b/internal/activator/swaphold.go
@@ -13,7 +13,6 @@ import (
)
var (
- errHoldNeedsPins = errors.New("swap: claims are on and the claim runtime cannot pin a wake to a staged build")
// errHoldStopSlow: the old generation's stop outlasted wakeWaitBound, the
// bound a hold holds requests for (spec §7.8). The hold is withdrawn
// before its stop finishes, so nothing of the attempt's build runs.
@@ -80,8 +79,11 @@ func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string)
if err != nil {
return abandon(fmt.Errorf("admission: %w", err))
}
- if !hostwrite.Legacy(work) && (a.cfg.Claims == nil || !a.cfg.Claims.Pins) {
- return abandon(errHoldNeedsPins)
+ if r := a.cfg.Claims; !hostwrite.Legacy(work) && (r == nil || !r.Generations || !r.Pins) {
+ // Claims on, and the runtime cannot start a pinned generation of its
+ // own: refused before anything stops (Task 10). A stop whose start
+ // cannot happen would only cost the instance its running process.
+ return abandon(errClaimsNoGenerations)
}
sctx, scancel := context.WithCancel(ctx)
defer scancel()