amadan

carlos / platform Public

zdd-readiness: retention pins the good build; a hold build is published before its incumbent stops

Bucket retention pins rt.GoodVersion beside Version and PrevVersion, so a planned wake's fallback keeps its artifact (authorized flag-off exception). swapHold registers its build as a pending generation under a.mu at admission, writes the generations file off a.mu, then re-takes a.mu and re-runs the full admission (same entry, ownership, newest, kind, layout, busy) before beginStopLocked. A failed write or a failed re-check abandons with Executed false and the incumbent serving; the pending record is dropped when swapHold returns. Astra R1 T12, R3 section 5 item 9, plan review #10.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude d3d464cdb3aef1ebbaf7eba5f48dc43e82cd9cae parent 5ad5548
6 files changed, +283 −17
  • internal/activator/activator_test.go +7 −1
  • internal/activator/generations_file.go +16 −0
  • internal/activator/swap_hold_test.go +113 −0
  • internal/activator/swaphold.go +83 −15
  • internal/host/retention.go +11 −1
  • internal/host/retention_test.go +53 −0
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 5fa25f5fa..3adef1dd0 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -134,13 +134,19 @@ type fakeProc struct {
// window) by the observation's own polling, with no sleep.
polls int
exitAt int
+ // onStopEnter, when set, runs as Stop is entered, before the gate: a
+ // test records what was already true when the stop began.
+ onStopEnter func()
}
func (p *fakeProc) Stop(context.Context) error {
p.mu.Lock()
p.stopEnters++
- gate := p.stopGate
+ gate, enter := p.stopGate, p.onStopEnter
p.mu.Unlock()
+ if enter != nil {
+ enter()
+ }
if gate != nil {
<-gate
}
diff --git a/internal/activator/generations_file.go b/internal/activator/generations_file.go
index 5d8080466..0f4bc9d12 100644
--- a/internal/activator/generations_file.go
+++ b/internal/activator/generations_file.go
@@ -41,6 +41,15 @@ import (
// recorded as a pending build: that state already lives as long as the
// spawn's process may, so nothing has to be dropped when the spawn installs.
//
+// A hold is the exception (zdd-readiness Task 13, Astra R1 T12, R3 §5 item
+// 9). It stops its incumbent before its pinned entry exists, and from that
+// stop until the pinned wake publishes nothing serves and nothing names the
+// hold's build, so root could sweep or restage it under the wake about to
+// run it. swapHold records the build as a pending generation and writes the
+// file BEFORE the stop, and drops the record when it returns: by then the
+// pinned entry, or a retained failed wake, names the build for as long as a
+// process of it may run.
+//
// The file is written at Start, and before the first worker spawn (so it exists, even
// empty, before any worker runs), before every staged spawn's exec, naming
// the build about to run (startWorker's before-exec hook, ruling R19 item 3),
@@ -67,6 +76,13 @@ type pendingBuild struct {
func (a *Service) addPendingGeneration(host string, id release.StagedIdentity) pendingGeneration {
a.mu.Lock()
defer a.mu.Unlock()
+ return a.addPendingGenerationLocked(host, id)
+}
+
+// addPendingGenerationLocked is addPendingGeneration for a caller that holds
+// a.mu: a hold registers its build in the same critical section as its
+// admission (swapHold).
+func (a *Service) addPendingGenerationLocked(host string, id release.StagedIdentity) pendingGeneration {
if a.genPending == nil {
a.genPending = map[pendingGeneration]pendingBuild{}
}
diff --git a/internal/activator/swap_hold_test.go b/internal/activator/swap_hold_test.go
index 299edd80e..b518623d2 100644
--- a/internal/activator/swap_hold_test.go
+++ b/internal/activator/swap_hold_test.go
@@ -3,6 +3,7 @@ package activator
import (
"context"
"errors"
+ "fmt"
"os"
"slices"
"strings"
@@ -13,6 +14,7 @@ import (
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
// A hold is stop-then-start through the barrier (spec §7.8). A request that
@@ -735,3 +737,114 @@ func TestSwapHoldAcceptsANestedRow(t *testing.T) {
t.Fatalf("Swap = %+v, want a done hold on the staged build", res)
}
}
+
+// zdd-readiness Task 13 (Astra R1 T12, R3 §5 item 9): a hold stops its
+// incumbent before its own build execs, so the build has to be in the
+// generations file before that stop begins. Published only at the pinned
+// wake's exec, root could sweep or restage the staged build in the window
+// where nothing serves at all.
+
+// The generations file names the hold's build when the incumbent's stop
+// begins. The fake records what the file said at that moment, before the
+// stop's own gate.
+func TestSwapHoldPublishesItsBuildBeforeTheIncumbentStops(t *testing.T) {
+ f := newSwapFixture(t)
+ f.a.GenerationsFile = release.GenerationsPath(t.TempDir())
+ incumbent := f.b.lastProc()
+ incumbent.mu.Lock()
+ incumbent.onStopEnter = func() {
+ f.b.record(fmt.Sprintf("stop-begins:file=%s", strings.Join(hostsOf(f.a.GenerationsFile)[f.host], ",")))
+ }
+ incumbent.mu.Unlock()
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapDone || res.Gen.StagedPath != f.staged {
+ t.Fatalf("Swap = %+v, want a done hold on the staged build", res)
+ }
+ evs := f.b.events()
+ began := slices.IndexFunc(evs, func(ev string) bool { return strings.HasPrefix(ev, "stop-begins:") })
+ if began < 0 {
+ t.Fatalf("events %q: the incumbent's stop never began", evs)
+ }
+ if evs[began] != "stop-begins:file=v1,v2" {
+ t.Fatalf("when the incumbent's stop began the file named %q, want the hold's build v2 beside v1", evs[began])
+ }
+ if i := slices.Index(evs, "startGen:"+f.staged); i < began {
+ t.Fatalf("events %q: the hold's exec at %d, before its incumbent's stop at %d", evs, i, began)
+ }
+ // The pending registration went with the swap: the file names what runs.
+ waitForGenerations(t, f.a.GenerationsFile, map[string][]string{f.host: {"v2"}})
+ f.a.mu.Lock()
+ pending := len(f.a.genPending)
+ f.a.mu.Unlock()
+ if pending != 0 {
+ t.Fatalf("%d pending generations left after the hold finished", pending)
+ }
+}
+
+// A hold whose build cannot be published stops nothing: the incumbent keeps
+// serving and the attempt is abandoned with nothing executed.
+func TestSwapHoldUnpublishedLeavesTheIncumbentServing(t *testing.T) {
+ f := newSwapFixture(t)
+ f.a.GenerationsFile = unwritableGenerationsFile(t)
+ incumbent := f.b.lastProc()
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapAbandoned || res.Executed || !strings.Contains(res.Reason, errStagedBuildNotPublished.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned, not executed: unpublished", res)
+ }
+ if n := incumbent.stopEntered(); n != 0 || incumbent.isStopped() {
+ t.Fatalf("the incumbent's stop was entered %d times for a hold whose build was never published", n)
+ }
+ f.a.mu.Lock()
+ e := f.a.entries[f.host]
+ awake := e != nil && e.state == actAwake && e.gen.N == f.incumbent.N && e.holdNext == nil
+ pending := len(f.a.genPending)
+ f.a.mu.Unlock()
+ if !awake {
+ t.Fatal("want the incumbent's entry still awake on its generation, with no hold on it")
+ }
+ if pending != 0 {
+ t.Fatalf("%d pending generations left by an abandoned hold", pending)
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatalf("StartGen calls = %d, want 0", gens)
+ }
+}
+
+// The publication runs off a.mu, so the admission is asked again once it is
+// written. An attempt superseded while the file was written withdraws its
+// pending build and stops nothing.
+func TestSwapHoldReadmitsAfterThePublication(t *testing.T) {
+ f := newSwapFixture(t)
+ f.a.GenerationsFile = release.GenerationsPath(t.TempDir())
+ incumbent := f.b.lastProc()
+ at := f.attempt(ModeHold)
+ // newest is asked under a.mu at each admission. The newer attempt
+ // arrives exactly when the hold's build is in the file: after the
+ // first admission, before the second.
+ newest := func(host string) uint64 {
+ if slices.Contains(hostsOf(f.a.GenerationsFile)[host], "v2") {
+ return at.SchedRev + 1
+ }
+ return at.SchedRev
+ }
+ res := f.a.Swap(context.Background(), at, newest)
+ if res.Outcome != SwapAbandoned || res.Executed || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned, not executed: superseded", res)
+ }
+ if n := incumbent.stopEntered(); n != 0 {
+ t.Fatalf("the incumbent's stop was entered %d times for a superseded hold", n)
+ }
+ f.a.mu.Lock()
+ e := f.a.entries[f.host]
+ awake := e != nil && e.state == actAwake && e.holdNext == nil
+ pending := len(f.a.genPending)
+ f.a.mu.Unlock()
+ if !awake || pending != 0 {
+ t.Fatalf("awake with no hold = %v, pending generations = %d", awake, pending)
+ }
+ // The withdrawn build is no longer published.
+ waitForGenerations(t, f.a.GenerationsFile, map[string][]string{f.host: {"v1"}})
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatalf("StartGen calls = %d, want 0", gens)
+ }
+}
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go
index 8da573642..66b186770 100644
--- a/internal/activator/swaphold.go
+++ b/internal/activator/swaphold.go
@@ -62,7 +62,10 @@ func (a *Service) lifeDone() <-chan struct{} {
//
// A row that cannot run the staged build at all is refused before
// anything stops: a stop-then-start whose start cannot happen would only
-// cost the instance its running process.
+// cost the instance its running process. So is a hold whose build cannot be
+// published in the generations file first (zdd-readiness Task 13): the
+// admission registers the build as a pending generation, the file is written
+// off a.mu, and the admission runs again before the stop begins.
//
// The caller waits for the stop for at most wakeWaitBound, and no longer
// than its own context or the agent's life. Any of the three withdraws the
@@ -100,29 +103,69 @@ func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string)
h.settleCancel = scancel
}
var (
- held *actEntry // the entry whose stop the hold waits on
- stopAt time.Time // wall clock: the old generation's stop, for the drain time (spec §9.4)
+ held *actEntry // the entry whose stop the hold waits on
+ stopAt time.Time // wall clock: the old generation's stop, for the drain time (spec §9.4)
+ admitted *actEntry // the entry the first admission registered the build against
+ pend pendingGeneration
)
+ // The hold's build is published BEFORE the incumbent stops
+ // (zdd-readiness Task 13, Astra R1 T12, R3 §5 item 9). From the stop
+ // until the pinned wake's own publication nothing serves and, without
+ // this, nothing names the build, so root could sweep or restage it under
+ // the wake about to run it. The record is dropped when swapHold returns,
+ // whatever the outcome: a withdrawn hold ran nothing, and a taken one is
+ // named by its pinned entry (or a retained failed wake) from then on.
+ defer func() {
+ if pend == 0 {
+ return
+ }
+ a.mu.Lock()
+ a.dropPendingGenerationLocked(pend)
+ a.mu.Unlock()
+ _ = a.persistGenerations()
+ }()
+ // Admission, then the registration in the same critical section.
err = a.admit(ctx, work, func() error {
a.mu.Lock()
defer a.mu.Unlock()
e := a.entries[at.Host]
- if err := a.swapChecksLocked(e, at, newest); err != nil {
+ if err := a.holdAdmitLocked(e, at, newest); err != nil {
return err
}
- if e.worker || e.route.Kind != "instance" || e.route.Backing == registry.BackingUnit {
- return errSwapKind
+ admitted = e
+ if id, ok := a.genIdentity(h.gen); ok {
+ pend = a.addPendingGenerationLocked(at.Host, id)
}
- // The layouts that can run a staged generation: flat, foreign on its
- // canonical socket, and nested (root mounts the staged build,
- // Task 9).
- if c, ok := a.backend.(Classifier); ok {
- if l, _ := c.Layout(e.route); l != LayoutFlat && l != LayoutForeign && l != LayoutNested {
- return errSwapLayout
- }
+ return nil
+ })
+ if err != nil {
+ return abandon(err)
+ }
+ // persistGenerations takes genMu and then a.mu, so it runs here, between
+ // the two critical sections, never inside the admission. A write that
+ // fails stops nothing: the incumbent keeps serving and the attempt is
+ // abandoned with nothing of its build run (errStagedBuildNotPublished,
+ // which is never a failure of the build).
+ if err := a.publishBeforeExec(); err != nil {
+ return abandon(err)
+ }
+ // The admission once more, with a.mu re-taken: anything the first one
+ // checked may have changed while the file was written (plan review #10).
+ // The same entry, still awake and owned, the attempt still the newest,
+ // and every gate of the first admission. A refusal withdraws the pending
+ // build (the deferred drop) and stops nothing.
+ err = a.admit(ctx, work, func() error {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[at.Host]
+ if err := a.holdAdmitLocked(e, at, newest); err != nil {
+ return err
}
- if len(e.ops.members) > 0 {
- return errSwapBusy
+ if e != admitted {
+ // The entry the build was admitted against has gone (a sleep and
+ // a fresh wake in the window). The checks pass on its successor,
+ // but this attempt was never admitted there.
+ return errSwapNotAwake
}
e.holdNext = h
held, stopAt = e, time.Now()
@@ -157,6 +200,31 @@ func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string)
return a.holdResult(sctx, at, newest, hw, time.Since(stopAt)) // finishStop answers once the old generation is proven gone
}
+// holdAdmitLocked is a hold's admission (spec §7.4, §7.8): the attempt's
+// checks on e, then the gates that refuse a row before anything stops (the
+// kind, the layout, an operation already under way). swapHold asks it twice,
+// before the build's publication and again after it, with a.mu re-taken.
+// The caller holds a.mu.
+func (a *Service) holdAdmitLocked(e *actEntry, at Attempt, newest func(string) uint64) error {
+ if err := a.swapChecksLocked(e, at, newest); err != nil {
+ return err
+ }
+ if e.worker || e.route.Kind != "instance" || e.route.Backing == registry.BackingUnit {
+ return errSwapKind
+ }
+ // The layouts that can run a staged generation: flat, foreign on its
+ // canonical socket, and nested (root mounts the staged build, Task 9).
+ if c, ok := a.backend.(Classifier); ok {
+ if l, _ := c.Layout(e.route); l != LayoutFlat && l != LayoutForeign && l != LayoutNested {
+ return errSwapLayout
+ }
+ }
+ if len(e.ops.members) > 0 {
+ return errSwapBusy
+ }
+ return nil
+}
+
// withdrawHold clears h from e while the stop has not taken it, and reports
// whether it did. False means finishStop already took the hold and answered
// on h.woke in the same critical section.
diff --git a/internal/host/retention.go b/internal/host/retention.go
index 6b31240ba..c22136abb 100644
--- a/internal/host/retention.go
+++ b/internal/host/retention.go
@@ -3,7 +3,7 @@
// routes for, read the app's retention record from the bucket and delete
// what release.PruneEligible says is safe — protecting, on top of the
// bucket-derived safety list, the versions THIS box's registry still
-// records as adopted or previous. Boxes race each other harmlessly: every
+// records as adopted, previous or good (the fallback). Boxes race each other harmlessly: every
// delete is idempotent by store contract.
//
// Hosted-apps-only is a ruling, not an accident: an app with routes on no
@@ -44,6 +44,16 @@ func retentionPass(ctx context.Context, store release.Store, pub ed25519.PublicK
if rt.PrevVersion != "" {
pins[k][rt.PrevVersion] = true
}
+ // The good build is a planned wake's fallback (zero-downtime spec
+ // §8.5), and it can be neither Version nor PrevVersion once two
+ // releases have failed in a row. Pruned, the fallback has no artifact
+ // exactly when a failed release needs it (zdd-readiness Task 13,
+ // Astra R1 T12). Pinned whatever CARLOS_SWAP says, the plan's
+ // authorized flag-off exception: keeping one more version is the
+ // cheap side of the trade.
+ if rt.GoodVersion != "" {
+ pins[k][rt.GoodVersion] = true
+ }
}
for k, protected := range pins {
ret, err := release.GetRetention(ctx, store, k.acct, k.app)
diff --git a/internal/host/retention_test.go b/internal/host/retention_test.go
index 174049ca8..9c27643eb 100644
--- a/internal/host/retention_test.go
+++ b/internal/host/retention_test.go
@@ -144,6 +144,59 @@ func TestRetentionPassResolvesEmptyAccountRoutesViaRouteAccount(t *testing.T) {
}
}
+// The good build is the fallback a planned wake boots when the new build
+// fails (zero-downtime spec §8.5), so retention keeps it even when no
+// route installs it as Version or PrevVersion any more (zdd-readiness
+// Task 13, Astra R1 T12). Pruned, the fallback's artifact is gone exactly
+// when a failed release needs it.
+func TestRetentionPassKeepsTheGoodBuild(t *testing.T) {
+ ctx := context.Background()
+ dir := t.TempDir()
+ s := release.NewDirStore(dir)
+ now := time.Now()
+ base := now.Add(-30 * 24 * time.Hour)
+ vname := func(i int) string { return "v" + string(rune('0'+i/10)) + string(rune('0'+i%10)) }
+ for i := 1; i <= 25; i++ {
+ stampKey(t, s, dir, release.ManifestKey("bab", "kass", vname(i)), base.Add(time.Duration(i)*time.Hour))
+ stampKey(t, s, dir, release.ArtifactKey("bab", "kass", vname(i), "linux-arm64"), base.Add(time.Duration(i)*time.Hour))
+ }
+ _, pub := retentionKeys(t)
+ if err := release.SetRetention(ctx, s, "bab", "kass", 20, "test", now); err != nil {
+ t.Fatal(err)
+ }
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ // Installed v03 over v02, but the last build that served healthily is
+ // v01: all three rank outside keep-20.
+ if err := reg.Upsert(registry.Route{
+ Host: "kass.example.com", Kind: "instance", App: "kass", Account: "bab",
+ Version: "v03", PrevVersion: "v02", GoodVersion: "v01", Status: "ready",
+ }); err != nil {
+ t.Fatal(err)
+ }
+
+ if err := retentionPass(ctx, s, pub, reg); err != nil {
+ t.Fatal(err)
+ }
+
+ for _, kept := range []string{"v01", "v02", "v03"} {
+ if _, _, err := s.Get(ctx, release.ManifestKey("bab", "kass", kept)); err != nil {
+ t.Fatalf("%s pruned, want kept as a pin: %v", kept, err)
+ }
+ }
+ if _, _, err := s.Get(ctx, release.ArtifactKey("bab", "kass", "v01", "linux-arm64")); err != nil {
+ t.Fatalf("the good build's artifact was pruned: %v", err)
+ }
+ for _, gone := range []string{"v04", "v05"} {
+ if _, _, err := s.Get(ctx, release.ManifestKey("bab", "kass", gone)); err == nil {
+ t.Fatalf("%s still present, want pruned", gone)
+ }
+ }
+}
+
func TestRetentionPassSkipsAppsWithoutAPolicy(t *testing.T) {
ctx := context.Background()
dir := t.TempDir()