amadan

carlos / platform Public

settings-4 task 5a: the build-state hook and its record before every staged spawn

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c243a4f9ba90d9c74471113a343717211e1cf0e1 parent 1cd62de
10 files changed, +707 −11
  • internal/activator/activator.go +8 −0
  • internal/activator/beforeexec.go +71 −0
  • internal/activator/beforeexec_test.go +403 −0
  • internal/activator/logsink.go +7 −0
  • internal/activator/swap.go +21 −3
  • internal/activator/swaphold.go +10 −0
  • internal/activator/workerbuild.go +14 −2
  • internal/host/agent.go +9 −4
  • internal/host/buildstate.go +46 −2
  • internal/host/buildstate_test.go +118 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index b8725520c..c9d690abd 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1034,6 +1034,14 @@ type Service struct {
// Service. Set before Start, like OnWake. Plan 5's scheduler fills it.
CoalesceRecycle func(host string) bool
+ // BeforeExec records, under the instance's lease, that build is about
+ // to run against host's database (settings spec §4.2), before every
+ // spawn (recordBuild). The context names the lease's location
+ // (BuildStateLocationFrom). A non-nil error refuses the spawn, and is
+ // never a failure of the build. nil records nothing: a box with no
+ // replica bucket. Set before Start, like OnWake.
+ BeforeExec func(ctx context.Context, host, build string) error
+
// OnWorkerServing, when set, hears a worker's captured serving identity
// each time a wake or a respawn installs a child (zero-downtime spec
// §7.9), an empty version included. It runs with no lock held, through
diff --git a/internal/activator/beforeexec.go b/internal/activator/beforeexec.go
new file mode 100644
index 000000000..cc842b5c5
--- /dev/null
+++ b/internal/activator/beforeexec.go
@@ -0,0 +1,71 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+)
+
+// errBuildStateUnwritten refuses a spawn whose build could not be recorded
+// as the one about to touch the instance's database (settings spec §4.2).
+// Nothing of the build ran, so it is never a failure of the build: the
+// classifiers read it as a refusal (classifyStep, verdictOf), and a member
+// sees the maintenance fence's wording (memberSafeWakeReason).
+var errBuildStateUnwritten = errors.New("the build state could not be written, so this build was not started")
+
+// recordBuild runs Service.BeforeExec, if set, for a spawn about to run
+// build against host's database (settings spec §4.2): the build state names
+// the build before the process opens the data, and a write that fails
+// refuses the spawn with errBuildStateUnwritten. Callers hold the
+// instance's lease and call it after their own gate, immediately before the
+// exec. A nil BeforeExec is today's behaviour.
+//
+// Not to be confused with withBeforeExec / runBeforeExec (workerbuild.go):
+// those carry startWorker's exec-time hook on a worker spawn's context, and
+// that hook is one of recordBuild's callers.
+//
+// A wake that already recorded this build (withRecordedBuild) is not
+// written twice. A build with no name (an executable outside the version
+// store) is refused while the hook is set: no build state could name it
+// (ruling R-EXEC-NOVERSION).
+//
+// With host claims off the lease is the hibernate lease, so a context that
+// names no location gets that one (Config.Bucket, HibernatePrefix), and the
+// agent writes beside its lock.json. A claimed wake attaches its own
+// (Task 5c); with claims on and none attached, the agent locates the host
+// through its claim.
+func (a *Service) recordBuild(ctx context.Context, host, build string) error {
+ if a.BeforeExec == nil {
+ return nil
+ }
+ if build != "" && recordedBuild(ctx) == build {
+ return nil
+ }
+ if build == "" {
+ log.Printf("activator: %s not started: the executable it would run is outside the version store, so no build state can name it (settings spec 4.2); move the instance onto a promoted build", host)
+ return fmt.Errorf("%w: the executable is outside the version store", errBuildStateUnwritten)
+ }
+ if _, ok := BuildStateLocationFrom(ctx); !ok && !a.cfg.HostWrites.Enabled {
+ ctx = WithBuildStateLocation(ctx, BuildStateLocation{Bucket: a.cfg.Bucket, Prefix: a.cfg.hibernatePrefix(host)})
+ }
+ if err := a.BeforeExec(ctx, host, build); err != nil {
+ log.Printf("activator: %s: %s not started: its build state could not be written: %v", host, build, err)
+ return fmt.Errorf("%w: %v", errBuildStateUnwritten, err)
+ }
+ return nil
+}
+
+type recordedBuildKey struct{}
+
+// withRecordedBuild marks ctx as carrying a wake that has already recorded
+// build (Task 5b's wake records once, before its seed and its spawn).
+func withRecordedBuild(ctx context.Context, build string) context.Context {
+ return context.WithValue(ctx, recordedBuildKey{}, build)
+}
+
+// recordedBuild is the build ctx's wake already recorded, or "".
+func recordedBuild(ctx context.Context) string {
+ s, _ := ctx.Value(recordedBuildKey{}).(string)
+ return s
+}
diff --git a/internal/activator/beforeexec_test.go b/internal/activator/beforeexec_test.go
new file mode 100644
index 000000000..6c21d0627
--- /dev/null
+++ b/internal/activator/beforeexec_test.go
@@ -0,0 +1,403 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "slices"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+// execLog is a BeforeExec hook that records host@build, the backend's spawn
+// count when it was asked (so a test can tell the record came before the
+// spawn), and the build-state location the context carried. fail, when
+// set, says which builds it refuses.
+type execLog struct {
+ mu sync.Mutex
+ calls []string
+ at []int
+ locs []BuildStateLocation
+ spawns func() int
+ fail func(build string) bool
+}
+
+func (l *execLog) hook(ctx context.Context, host, build string) error {
+ n := -1
+ if l.spawns != nil {
+ n = l.spawns()
+ }
+ loc, _ := BuildStateLocationFrom(ctx)
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ l.calls = append(l.calls, host+"@"+build)
+ l.at = append(l.at, n)
+ l.locs = append(l.locs, loc)
+ if l.fail != nil && l.fail(build) {
+ return errors.New("bucket refused the build state")
+ }
+ return nil
+}
+
+func (l *execLog) seen() ([]string, []int, []BuildStateLocation) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return append([]string(nil), l.calls...), append([]int(nil), l.at...), append([]BuildStateLocation(nil), l.locs...)
+}
+
+func failAll(string) bool { return true }
+
+// --- the hook itself ---
+
+// Ruling R-EXEC-NOVERSION: an identity outside the version store has no
+// build a build state could name. With the hook set it is refused; with no
+// hook (a box with no replica bucket) it runs as today.
+func TestAnUnrecordableBuildIsRefused(t *testing.T) {
+ a := &Service{cfg: Config{Bucket: "b"}}
+ if err := a.recordBuild(context.Background(), "one.example.com", ""); err != nil {
+ t.Fatalf("a nil hook refused: %v", err)
+ }
+ var l execLog
+ a.BeforeExec = l.hook
+ err := a.recordBuild(context.Background(), "one.example.com", "")
+ if !errors.Is(err, errBuildStateUnwritten) {
+ t.Fatalf("err = %v, want errBuildStateUnwritten", err)
+ }
+ if calls, _, _ := l.seen(); len(calls) != 0 {
+ t.Fatalf("the hook was asked to record no build: %v", calls)
+ }
+}
+
+// The unclaimed lease's location rides to the hook, so the agent writes
+// beside this lease's lock.json; a location the caller already attached
+// (the claimed path, Task 5c) is kept. A hook error is wrapped so every
+// classifier reads it as a refusal.
+func TestTheHookIsToldWhereTheLeaseIs(t *testing.T) {
+ a := &Service{cfg: Config{Bucket: "hib", Prefix: "inst"}}
+ var l execLog
+ a.BeforeExec = l.hook
+ if err := a.recordBuild(context.Background(), "one.example.com", "v1"); err != nil {
+ t.Fatal(err)
+ }
+ claimed := BuildStateLocation{Bucket: "data", Prefix: "instances/one.example.com/inc1"}
+ if err := a.recordBuild(WithBuildStateLocation(context.Background(), claimed), "one.example.com", "v1"); err != nil {
+ t.Fatal(err)
+ }
+ calls, _, locs := l.seen()
+ if len(calls) != 2 || calls[0] != "one.example.com@v1" {
+ t.Fatalf("calls = %v", calls)
+ }
+ if want := (BuildStateLocation{Bucket: "hib", Prefix: HibernatePrefix("inst", "one.example.com")}); locs[0] != want {
+ t.Fatalf("location = %+v, want the hibernate lease's %+v", locs[0], want)
+ }
+ if locs[1] != claimed {
+ t.Fatalf("location = %+v, want the caller's %+v kept", locs[1], claimed)
+ }
+ l.fail = failAll
+ err := a.recordBuild(context.Background(), "one.example.com", "v1")
+ if !errors.Is(err, errBuildStateUnwritten) || !strings.Contains(err.Error(), "bucket refused") {
+ t.Fatalf("err = %v, want errBuildStateUnwritten carrying the cause", err)
+ }
+}
+
+// A wake that already recorded this build (Task 5b) is not written twice;
+// another build is.
+func TestARecordedBuildIsNotWrittenTwice(t *testing.T) {
+ a := &Service{cfg: Config{Bucket: "b"}}
+ var l execLog
+ a.BeforeExec = l.hook
+ ctx := withRecordedBuild(context.Background(), "v1")
+ if err := a.recordBuild(ctx, "one.example.com", "v1"); err != nil {
+ t.Fatal(err)
+ }
+ if err := a.recordBuild(ctx, "one.example.com", "v2"); err != nil {
+ t.Fatal(err)
+ }
+ if calls, _, _ := l.seen(); len(calls) != 1 || calls[0] != "one.example.com@v2" {
+ t.Fatalf("calls = %v, want only the fallback's v2", calls)
+ }
+}
+
+// Ruling R-WAKEWORDS: a member never sees the internal text.
+func TestMemberSeesFencedForAnUnwrittenBuildState(t *testing.T) {
+ err := fmt.Errorf("start one.example.com: %w", fmt.Errorf("%w: s3: access denied", errBuildStateUnwritten))
+ if got, want := memberSafeWakeReason(err), memberSafeWakeReason(ErrFenced); got != want {
+ t.Fatalf("member reads %q, want the fence's %q", got, want)
+ }
+}
+
+// A refused record is never a failure of the build, whichever classifier
+// reads it.
+func TestAnUnwrittenBuildStateIsARefusal(t *testing.T) {
+ err := fmt.Errorf("spawn: %w", fmt.Errorf("%w: s3: access denied", errBuildStateUnwritten))
+ if got := classifyStep(nil, err); got != stepStale {
+ t.Fatalf("classifyStep = %v, want stepStale", got)
+ }
+ if got := verdictOf(err); got != gateNoOwnership {
+ t.Fatalf("verdictOf = %v, want a refusal", got)
+ }
+}
+
+// --- every chokepoint that execs a captured identity (anchors-exec §1) ---
+
+// #11: a hold swap's pinned wake records the attempt's build before its
+// exec; a refused record abandons the attempt with nothing exec'd.
+func TestBeforeExecRunsBeforeAHoldSwap(t *testing.T) {
+ f := newSwapFixture(t)
+ l := &execLog{spawns: func() int { n, _, _ := f.b.calls(); return n }}
+ f.a.BeforeExec = l.hook
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ calls, at, locs := l.seen()
+ if len(calls) != 1 || calls[0] != f.host+"@v2" || at[0] != 0 {
+ t.Fatalf("records %v at spawn counts %v, want %s@v2 before the exec", calls, at, f.host)
+ }
+ if locs[0] != (BuildStateLocation{Bucket: "test-bucket", Prefix: HibernatePrefix("", f.host)}) {
+ t.Fatalf("location = %+v", locs[0])
+ }
+ if gens, _, _ := f.b.calls(); gens != 1 {
+ t.Fatalf("StartStaged calls = %d, want 1", gens)
+ }
+
+ f = newSwapFixture(t)
+ f.a.BeforeExec = (&execLog{fail: failAll}).hook
+ res = f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errBuildStateUnwritten.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned for the unwritten build state", res)
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatalf("StartStaged calls = %d: a build ran unrecorded", gens)
+ }
+}
+
+// #12: an overlap successor records its build before its exec; a refused
+// record is SwapAbandoned, never SwapFailed (which the scheduler writes as
+// the build's failure), and the incumbent keeps serving.
+func TestBeforeExecRunsBeforeAnOverlapSuccessor(t *testing.T) {
+ f := newSwapFixture(t)
+ l := &execLog{spawns: func() int { n, _, _ := f.b.calls(); return n }}
+ f.a.BeforeExec = l.hook
+ res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
+ if res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if calls, at, _ := l.seen(); len(calls) != 1 || calls[0] != f.host+"@v2" || at[0] != 0 {
+ t.Fatalf("records %v at spawn counts %v, want %s@v2 before the exec", calls, at, f.host)
+ }
+
+ f = newSwapFixture(t)
+ f.a.BeforeExec = (&execLog{fail: failAll}).hook
+ res = f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errBuildStateUnwritten.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned (never failed) for the unwritten build state", res)
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatalf("StartStaged calls = %d: a build ran unrecorded", gens)
+ }
+ if got := f.appSocket(t); got != f.incumbent.Socket {
+ t.Fatalf("AppSocket = %q, want the incumbent's", got)
+ }
+ if _, raises, restores := f.b.calls(); raises != restores {
+ t.Fatalf("raises %d restores %d: the parent limit was not restored", raises, restores)
+ }
+}
+
+// #13: a planned boot records B before its exec; a refused record execs
+// nothing and records nothing against B.
+func TestBeforeExecRunsBeforeAPlannedBoot(t *testing.T) {
+ dir := t.TempDir()
+ next := stagedID(t, dir, "v2")
+ a, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap})
+ l := &execLog{spawns: func() int { return len(b.started()) }}
+ a.BeforeExec = l.hook
+ done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("the wake: %v", err)
+ }
+ done()
+ if calls, at, _ := l.seen(); len(calls) != 1 || calls[0] != "one.example.com@v2" || at[0] != 0 {
+ t.Fatalf("records %v at spawn counts %v, want one.example.com@v2 before the exec", calls, at)
+ }
+
+ dir = t.TempDir()
+ good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
+ a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
+ a.BeforeExec = (&execLog{fail: failAll}).hook
+ _, err = a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if !errors.Is(err, errBuildStateUnwritten) {
+ t.Fatalf("err = %v, want errBuildStateUnwritten", err)
+ }
+ if gens := b.started(); len(gens) != 0 {
+ t.Fatalf("spawned %+v: a build ran unrecorded", gens)
+ }
+ time.Sleep(50 * time.Millisecond)
+ if k := evs.kinds(); len(k) != 0 {
+ t.Fatalf("events %v: a build that never ran was recorded", k)
+ }
+}
+
+// #15: the fallback is a different build touching the data, so it records
+// again before its own exec. A refused record starts no good build; B's own
+// failure (it did fail) stays its own, and nothing falls back.
+func TestBeforeExecRunsBeforeAPlannedFallback(t *testing.T) {
+ dir := t.TempDir()
+ good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
+ a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
+ a.genHealth = func(_ context.Context, g Gen) error {
+ if g.Build == "v2" {
+ return errors.New("GET / answered 500")
+ }
+ return nil
+ }
+ l := &execLog{spawns: func() int { return len(b.started()) }}
+ a.BeforeExec = l.hook
+ done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("the wake: %v", err)
+ }
+ done()
+ calls, at, _ := l.seen()
+ if len(calls) != 2 || calls[0] != "one.example.com@v2" || calls[1] != "one.example.com@v1" || at[0] != 0 || at[1] != 1 {
+ t.Fatalf("records %v at spawn counts %v, want v2 before B's exec, then v1 before the fallback's", calls, at)
+ }
+ evs.wait(t, GenFellBack)
+
+ a, b, evs = planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
+ a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") }
+ a.BeforeExec = (&execLog{fail: func(build string) bool { return build == "v1" }}).hook
+ // The wake's error is B's failure (spawnPlanned wraps it); the
+ // fallback's refusal is its detail.
+ if _, err := a.beginRequest(context.Background(), testRoute("one.example.com")); err == nil || !strings.Contains(err.Error(), errBuildStateUnwritten.Error()) {
+ t.Fatalf("err = %v, want B's failure with the fallback's unwritten build state", err)
+ }
+ if gens := b.started(); len(gens) != 1 || gens[0].Build != "v2" {
+ t.Fatalf("spawned %+v, want B alone", gens)
+ }
+ if ev := evs.wait(t, GenFailed); ev.Failed.Version != "v2" {
+ t.Fatalf("failure event = %+v, want B's own", ev)
+ }
+ time.Sleep(50 * time.Millisecond)
+ for _, k := range evs.kinds() {
+ if k == GenFellBack {
+ t.Fatalf("events %v: a fallback that never ran was reported", evs.kinds())
+ }
+ }
+}
+
+// #5: a worker wake records the identity it is about to exec, after its
+// publication; a refused record execs nothing and fails the wake.
+func TestBeforeExecRunsBeforeAWorkerWake(t *testing.T) {
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ l := &execLog{spawns: b.workerStarts}
+ a.BeforeExec = l.hook
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ calls, at, locs := l.seen()
+ if len(calls) != 1 || calls[0] != host+"@v1" || at[0] != 0 {
+ t.Fatalf("records %v at spawn counts %v, want %s@v1 before the exec", calls, at, host)
+ }
+ if locs[0] != (BuildStateLocation{Bucket: "test-bucket", Prefix: HibernatePrefix("", host)}) {
+ t.Fatalf("location = %+v, want the worker's own lease", locs[0])
+ }
+
+ a, b, _, _, _ = stagedActivator(t)
+ a.BeforeExec = (&execLog{fail: failAll}).hook
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); !errors.Is(err, errBuildStateUnwritten) {
+ t.Fatalf("WakeWorker = %v, want errBuildStateUnwritten", err)
+ }
+ if n := b.workerStarts(); n != 0 {
+ t.Fatalf("%d worker children: a build ran unrecorded", n)
+ }
+}
+
+// #7: a crash restart records the captured build before it respawns it. A
+// refused record is a refusal: nothing exec'd, no "failed to respawn"
+// member error event, and the sweep asks again.
+func TestBeforeExecRunsBeforeAWorkerRecycle(t *testing.T) {
+ a, b, clk, _, _ := stagedActivator(t)
+ sink := &lockedSink{}
+ a.Logs = sink
+ host := "bot.x.internal"
+ l := &execLog{spawns: b.workerStarts}
+ a.BeforeExec = l.hook
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ b.lastWorker().kill()
+ a.sweepOnce()
+ waitForRestart(t, a, b.fakeBackend, host, 2)
+ calls, at, _ := l.seen()
+ if len(calls) != 2 || calls[1] != host+"@v1" || at[1] != 1 {
+ t.Fatalf("records %v at spawn counts %v, want the recycle's %s@v1 before its exec", calls, at, host)
+ }
+
+ l.mu.Lock()
+ l.fail = failAll
+ l.mu.Unlock()
+ b.lastWorker().kill()
+ clk.advance(sidecarRestartMax + time.Second)
+ a.sweepOnce()
+ waitForSpawnCalls(t, a, b, host, 3)
+ if n := b.workerStarts(); n != 2 {
+ t.Fatalf("%d worker children: a build ran unrecorded", n)
+ }
+ sink.mu.Lock()
+ defer sink.mu.Unlock()
+ for _, ev := range sink.events {
+ if ev.level == "error" {
+ t.Fatalf("member event %+v: a refused record read as a failed respawn", ev)
+ }
+ }
+}
+
+// #10: ReplaceWorker's attempt records its build before the exec; a refused
+// record is abandonment, never a failure of the build.
+func TestBeforeExecRunsBeforeReplaceWorker(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ l := &execLog{spawns: b.workerStarts}
+ a.BeforeExec = l.hook
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
+ if res.Outcome != SwapDone {
+ t.Fatalf("ReplaceWorker = %+v", res)
+ }
+ if calls, at, _ := l.seen(); len(calls) != 2 || calls[1] != host+"@v2" || at[1] != 1 {
+ t.Fatalf("records %v at spawn counts %v, want %s@v2 before the attempt's exec", calls, at, host)
+ }
+
+ a, b, _, _, _ = stagedActivator(t)
+ l = &execLog{fail: func(build string) bool { return build == "v2" }}
+ a.BeforeExec = l.hook
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ res = a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errBuildStateUnwritten.Error()) {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned: an unrecorded build never ran", res)
+ }
+ waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
+ // Every child had a recorded build before it, and v2 never served.
+ calls, _, _ := l.seen()
+ recorded := 0
+ for _, c := range calls {
+ if c != host+"@v2" {
+ recorded++
+ }
+ }
+ if !slices.Contains(calls, host+"@v2") || b.workerStarts() != recorded {
+ t.Fatalf("records %v, %d worker children: a child ran without its build recorded", calls, b.workerStarts())
+ }
+ if got := a.entryServing(host); got.Version == "v2" {
+ t.Fatalf("serving %+v: v2 ran unrecorded", got)
+ }
+}
diff --git a/internal/activator/logsink.go b/internal/activator/logsink.go
index 9caf80813..77b21ea8c 100644
--- a/internal/activator/logsink.go
+++ b/internal/activator/logsink.go
@@ -57,6 +57,13 @@ func memberSafeWakeReason(err error) string {
// operator's (persistGenerations logs it).
err = ErrFenced
}
+ if errors.Is(err, errBuildStateUnwritten) {
+ // A spawn refused because its build state could not be written
+ // first (settings spec §4.2; ruling R-WAKEWORDS): the platform holds
+ // the host until its own write succeeds, the fence's existing
+ // wording. The write error is the operator's (recordBuild logs it).
+ err = ErrFenced
+ }
if errors.Is(err, errWakeUnproven) || genStartUnproven(err) || errors.Is(err, errOutcomeUndelivered) {
// A wake refused while the host's last failed wake is not proven
// stopped (zero-downtime spec §7.4), or that failed wake itself: a
diff --git a/internal/activator/swap.go b/internal/activator/swap.go
index 0142fc7e8..10b8a86f7 100644
--- a/internal/activator/swap.go
+++ b/internal/activator/swap.go
@@ -135,13 +135,17 @@ func classifyStep(lifecycle context.Context, err error) stepOutcome {
// or as a claims validation inside the work at the exec. Nothing of
// the build ran because of it (spec §7.4).
return stepStale
- case errors.Is(err, ErrBuildNotPublished), errors.Is(err, errStagedBuildNotPublished):
+ case errors.Is(err, ErrBuildNotPublished), errors.Is(err, errStagedBuildNotPublished),
+ errors.Is(err, errBuildStateUnwritten):
// A spawn refused because its build could not be published before
// the exec (ruling R19 item 3; the worker and the staged instance
// sentinels, final review F9): nothing of the build ran, so it is
// abandonment, never a failure of the build. Classified here and not
// in verdictOf, which main's flag-off respawn reads; every caller of
// classifyStep is flag-on (swaps, holds, planned wakes).
+ // errBuildStateUnwritten (settings spec §4.2) is the same kind of
+ // refusal; verdictOf also names it, so it is stale through either
+ // arm, and named here beside its kin.
return stepStale
}
return stepFailed
@@ -178,7 +182,14 @@ const (
// - a row that cannot run a staged generation at all: errUnitNoGens,
// errSwapNoGens;
// - root refusing the nested staged executable (instancectl.ErrStagedExe,
-// Task 9): it validates before it mounts or starts anything.
+// Task 9): it validates before it mounts or starts anything;
+// - a build state that could not be written before the exec
+// (errBuildStateUnwritten, settings spec §4.2). Unlike
+// ErrBuildNotPublished it is a new sentinel, so naming it here changes
+// nothing main did: a worker respawn it refuses logs a refusal, not
+// "failed to respawn" and a member error event, and the sweep asks
+// again; a ReplaceWorker attempt it refuses is abandoned without a
+// restore, and the sweep's crash restart respawns the serving build.
//
// errSwapStopped (the barrier's cancel, from memberCurrentLocked) is
// NoOwnership too: the entry is stopping, so nothing may spawn for it.
@@ -200,7 +211,8 @@ func verdictOf(err error) gateVerdict {
errors.Is(err, hostbox.ErrChanged), errors.Is(err, errClaimsAdmission),
errors.Is(err, ErrStrictRequired), errors.Is(err, ErrStrictGroupInUse), errors.Is(err, ErrStrictCleanupPending),
errors.Is(err, ErrClaimIsolationUnavailable),
- errors.Is(err, errUnitNoGens), errors.Is(err, errSwapNoGens), errors.Is(err, instancectl.ErrStagedExe):
+ errors.Is(err, errUnitNoGens), errors.Is(err, errSwapNoGens), errors.Is(err, instancectl.ErrStagedExe),
+ errors.Is(err, errBuildStateUnwritten):
return gateNoOwnership
}
return gateProceed
@@ -330,6 +342,12 @@ func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntr
if err := gate(); err != nil {
return g, nil, err
}
+ // Settings spec §4.2: the successor's build is recorded before it
+ // runs against the data, under the incumbent's lease. A refusal is
+ // abandonment (classifyStep), never the build's failure.
+ if err := a.recordBuild(ctx, rt.Host, g.Build); err != nil {
+ return g, nil, err
+ }
return gb.StartStaged(withExecGate(ctx, gate), rt, g)
}
if a.claimSpawn != nil {
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go
index e569eeb16..014514d7b 100644
--- a/internal/activator/swaphold.go
+++ b/internal/activator/swaphold.go
@@ -374,6 +374,16 @@ func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen,
return nil, "", pin, err
}
}
+ // Settings spec §4.2: the build state names pin.Build before it runs
+ // against the data. Every staged instance exec funnels through here: a
+ // hold's pinned wake, a planned boot, the good build booted untried and
+ // a planned fallback (a different build touching the data, so it
+ // records again). A refusal returns no process: nothing ran. If the
+ // backend's own exec gate then refuses, the state names a build about to
+ // run that did not, which the spec allows ("or one about to").
+ if err := a.recordBuild(ctx, rt.Host, pin.Build); err != nil {
+ return nil, "", pin, err
+ }
g, proc, err := gb.StartStaged(withExecGate(ctx, gate), rt, pin)
if err != nil {
return proc, "", pin, err
diff --git a/internal/activator/workerbuild.go b/internal/activator/workerbuild.go
index df2758e4e..a59e0d0c2 100644
--- a/internal/activator/workerbuild.go
+++ b/internal/activator/workerbuild.go
@@ -92,6 +92,12 @@ func runBeforeExec(ctx context.Context, id release.StagedIdentity) error {
func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity, gate func() error) (Proc, release.StagedIdentity, pendingGeneration, error) {
s, ok := a.backend.(stagedWorkerStarter)
if !ok {
+ // No staged capability: no identity is captured, so no build state
+ // is recorded. Production's SplitBackend always has the capability;
+ // it sends a unit-backed row (which UnitBackend refuses as a worker,
+ // and which holds no real lease) to StartWorker without the hook.
+ // So this arm, and that one, record nothing: settings spec §4.2
+ // covers the rows that run under a lease in the replica bucket.
if gate != nil {
if err := gate(); err != nil {
return nil, release.StagedIdentity{}, 0, err
@@ -113,9 +119,15 @@ func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[st
return fmt.Errorf("%w: %v", ErrBuildNotPublished, err)
}
if gate != nil {
- return gate()
+ if err := gate(); err != nil {
+ return err
+ }
}
- return nil
+ // Settings spec §4.2: the build state names the exact identity
+ // about to be exec'd, under the worker's lease, last. Every worker
+ // spawn passes here: a wake (planned or not, boot and fallback), a
+ // crash restart, a max_run recycle, a converge and ReplaceWorker.
+ return a.recordBuild(ctx, rt.Host, id.Version)
}
p, id, err := s.StartWorkerStaged(withBeforeExec(ctx, hook), rt, env, onto)
if err != nil && pend != 0 {
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 0ffa1623a..4a656abaf 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -296,12 +296,12 @@ func runAgent(args []string, buildClaims ClaimFactory) {
}
// The build-state store (settings spec §4.2), built once after the claim
// wiring so it locates claimed hosts through cfg.Claims. nil, nil on a box
- // with no replica bucket. Settings plan 4 Task 5 installs the BeforeExec
- // hook from bs, and a refusing hook when bsErr is set; until then a
- // failure here is one operator line and the agent keeps running.
+ // with no replica bucket. The activator's BeforeExec hook comes from bs
+ // (buildStateHook, below): a failure here is one operator line, the
+ // agent keeps running, and every spawn is refused until it is fixed.
bs, bsErr := newBuildStates(cfg, reg, release.OptDir())
if bsErr != nil {
- log.Printf("agent: build state store unavailable: %v. this box does not keep build states", bsErr)
+ log.Printf("agent: build state store unavailable: %v. every instance and worker spawn is refused until it is fixed and the agent restarted", bsErr)
}
// Fleet box identity (customer-fleets plan, Task 6 fix round — controller
@@ -431,6 +431,11 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// (design §6.8). The registry satisfies RouteLookup with its own
// Get, exactly as DueStore above.
svc.Routes = reg
+ // Settings spec §4.2: before any process runs a build against an
+ // instance's database, the build state names it, under the lease. A
+ // store that failed to start refuses every spawn (never "run
+ // unrecorded"). Set before Start, like Routes.
+ svc.BeforeExec = buildStateHook(bs, bsErr)
// Every worker's captured serving build, published for root's pin
// set (zero-downtime deploys spec §4.2, ruling R17), whatever
// CARLOS_SWAP says. dd is dataDir(), the same data dir root's
diff --git a/internal/host/buildstate.go b/internal/host/buildstate.go
index 86caab1e8..f81ff3453 100644
--- a/internal/host/buildstate.go
+++ b/internal/host/buildstate.go
@@ -153,7 +153,9 @@ func (b *buildStates) hostLock(host string) *sync.Mutex {
}
// getState reads the object at key: nil and "" when there is none, an error
-// when it does not parse or names another host.
+// when it does not parse or names another host. That error names the key:
+// update never overwrites such an object (ruling R-BS-CORRUPT), so every
+// spawn on the host is refused until an operator repairs or deletes it.
func getState(ctx context.Context, store release.Store, key, host string) (*release.BuildState, string, error) {
raw, etag, err := store.Get(ctx, key)
switch {
@@ -164,7 +166,7 @@ func getState(ctx context.Context, store release.Store, key, host string) (*rele
}
var s release.BuildState
if err := json.Unmarshal(raw, &s); err != nil || s.Host != host {
- return nil, "", fmt.Errorf("build state for %s does not parse or names another host", host)
+ return nil, "", fmt.Errorf("build state %s for %s does not parse or names another host; repair or delete that object", key, host)
}
return &s, etag, nil
}
@@ -342,3 +344,45 @@ func (b *buildStates) update(ctx context.Context, host string, merge func(cur *r
}
return release.BuildState{}, fmt.Errorf("build state of %s: %d tries lost the race: %w", host, updateTries, release.ErrCASConflict)
}
+
+// beforeExec is Service.BeforeExec (settings spec §4.2): before build runs
+// against host's database, under its lease, the object names build as
+// last_executed, carries root's resolution from the consumed file, and
+// completes the request if build is its target (release.NextBuildState).
+// That write is what completes a request: at first run, not at install.
+//
+// It fails closed. A consumed file the agent cannot read refuses the exec:
+// a write without it could miss a resolution and leave R uncompleted after
+// its T ran. update sets the registry copy itself, and refuses an object it
+// cannot parse rather than overwrite it (ruling R-BS-CORRUPT).
+func (b *buildStates) beforeExec(ctx context.Context, host, build string) error {
+ if build == "" {
+ return fmt.Errorf("no build to record for %s", host)
+ }
+ consumed, err := release.ReadConsumed(b.optDir, host)
+ if err != nil {
+ return fmt.Errorf("read root's consumed file for %s: %w", host, err)
+ }
+ _, err = b.update(ctx, host, func(cur *release.BuildState) release.BuildState {
+ return release.NextBuildState(cur, consumed, build)
+ })
+ return err
+}
+
+// buildStateHook is Service.BeforeExec for this box (settings spec §4.2,
+// Astra I3 #1). Only a box with no replica bucket configured (bs nil, no
+// error) keeps the nil hook: it runs no hibernating instance, so it has no
+// build state. A box whose bucket is configured but whose build-state store
+// failed to start refuses every exec: the write before execution is
+// mandatory, and "could not record" never means "run unrecorded".
+func buildStateHook(bs *buildStates, initErr error) func(ctx context.Context, host, build string) error {
+ if initErr != nil {
+ return func(context.Context, string, string) error {
+ return fmt.Errorf("the build state store did not start on this box (%v); restart the agent once it is fixed", initErr)
+ }
+ }
+ if bs == nil {
+ return nil
+ }
+ return bs.beforeExec
+}
diff --git a/internal/host/buildstate_test.go b/internal/host/buildstate_test.go
index 2e7c0f1f6..53e52edc6 100644
--- a/internal/host/buildstate_test.go
+++ b/internal/host/buildstate_test.go
@@ -6,6 +6,7 @@ import (
"fmt"
"os"
"path/filepath"
+ "strings"
"testing"
"time"
@@ -607,3 +608,120 @@ func TestClaimsOnRegistersOnlyTheClaimBucket(t *testing.T) {
t.Fatal("claims on with no claim runtime was accepted")
}
}
+
+// --- the pre-exec write (Service.BeforeExec, settings spec §4.2) ---
+
+func writeConsumed(t *testing.T, b *buildStates, c release.Consumed) {
+ t.Helper()
+ c.Host = "a.example.com"
+ if err := release.WriteConsumed(b.optDir, c); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// The write before the exec names the build, carries root's resolution
+// from the consumed file, and completes the request when the build is its
+// target. The registry copy follows.
+func TestBeforeExecCompletesAResolvedRequestWhenItsTargetRuns(t *testing.T) {
+ b, _, reg := testBuildStates(t)
+ ctx := context.Background()
+ writeConsumed(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 7, Target: "v2"}, Completed: 3})
+ if err := b.beforeExec(ctx, "a.example.com", "v1"); err != nil {
+ t.Fatal(err)
+ }
+ st, _, err := b.read(ctx, "a.example.com")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.LastExecuted != "v1" || st.Resolved == nil || st.Resolved.ID != 7 || st.Resolved.Target != "v2" || st.Completed != 3 {
+ t.Fatalf("state after v1 = %+v, want the resolution carried and R pending", st)
+ }
+ if err := b.beforeExec(ctx, "a.example.com", "v2"); err != nil {
+ t.Fatal(err)
+ }
+ st, _, _ = b.read(ctx, "a.example.com")
+ if st.LastExecuted != "v2" || st.Completed != 7 || st.CompletedFailed {
+ t.Fatalf("state after T = %+v, want R completed", st)
+ }
+ if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v2" || c.State.Completed != 7 || !c.Known {
+ t.Fatalf("registry copy = %+v", c)
+ }
+}
+
+// Fail closed: a consumed file the agent cannot read refuses the exec, and
+// nothing is written. So does a build with no name.
+func TestBeforeExecRefusesAnUnreadableConsumedFile(t *testing.T) {
+ b, s, _ := testBuildStates(t)
+ ctx := context.Background()
+ if err := os.MkdirAll(filepath.Join(b.optDir, "consumed"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(release.ConsumedPath(b.optDir, "a.example.com"), []byte("{not json"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.beforeExec(ctx, "a.example.com", "v1"); err == nil {
+ t.Fatal("an unreadable consumed file let the exec through")
+ }
+ if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a build state was written without root's resolution: %v", err)
+ }
+ if err := os.Remove(release.ConsumedPath(b.optDir, "a.example.com")); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.beforeExec(ctx, "a.example.com", ""); err == nil {
+ t.Fatal("a build with no name was recorded")
+ }
+}
+
+// The write lands at the location the wake's context names (its lease).
+func TestBeforeExecWritesAtTheContextLocation(t *testing.T) {
+ b, hib, _ := testBuildStates(t)
+ other := release.NewDirStore(t.TempDir())
+ b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": other})
+ ctx := activator.WithBuildStateLocation(context.Background(), activator.BuildStateLocation{Bucket: "data", Prefix: "instances/a.example.com/inc1"})
+ if err := b.beforeExec(ctx, "a.example.com", "v1"); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := other.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
+ t.Fatalf("the build state is not at the context's location: %v", err)
+ }
+ if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the hibernate prefix was written too: %v", err)
+ }
+}
+
+// Ruling R-BS-CORRUPT: an object that does not parse is never overwritten;
+// the refusal names the key and says what to do.
+func TestBeforeExecRefusesACorruptStateAndNamesIt(t *testing.T) {
+ b, s, _ := testBuildStates(t)
+ ctx := context.Background()
+ key := "inst/a.example.com/build-state.json"
+ if _, err := s.Put(ctx, key, []byte("{torn"), ""); err != nil {
+ t.Fatal(err)
+ }
+ err := b.beforeExec(ctx, "a.example.com", "v1")
+ if err == nil || !strings.Contains(err.Error(), key) || !strings.Contains(err.Error(), "repair or delete") {
+ t.Fatalf("err = %v, want a refusal naming %s and saying to repair or delete it", err, key)
+ }
+ if raw, _, _ := s.Get(ctx, key); string(raw) != "{torn" {
+ t.Fatalf("the corrupt object was overwritten: %q", raw)
+ }
+}
+
+// Astra I3 #1: an initialization failure never leaves the nil hook.
+func TestAFailedBuildStateStartRefusesEveryExec(t *testing.T) {
+ hook := buildStateHook(nil, errors.New("s3: no credentials"))
+ if hook == nil {
+ t.Fatal("an initialization failure installed the nil hook: execs would run unrecorded")
+ }
+ if err := hook(context.Background(), "a.example.com", "v1"); err == nil {
+ t.Fatal("the refusing hook let an exec through")
+ }
+ if buildStateHook(nil, nil) != nil {
+ t.Fatal("a box with no replica bucket must keep today's nil hook")
+ }
+ b, _, _ := testBuildStates(t)
+ if hook := buildStateHook(b, nil); hook == nil || hook(context.Background(), "a.example.com", "v1") != nil {
+ t.Fatal("a started store's hook does not record")
+ }
+}