amadan
zdd-5 task 12: instance generations in the generations file; publication gates staged spawns; GoodVersion in root's pins
Extends plan 4's generationsSnapshotLocked (no second writer): every entry's current generation, a hold's pin, a planned wake's build and fallback, every operation-set member, retained failed wakes' generations and a socketless attempt's builds. With generations off an instance adds nothing, so the flag-off file is unchanged. A planned or pinned wake (doWake, doClaimWake) publishes before the lease, a replan publishes before its exec (checkPlan), and an overlap successor publishes before Step 3a. A failed write refuses the spawn with ErrBuildNotPublished: abandonment, never a failure of the build. release.BoxPins and the host version sweep pin GoodVersion.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
14 files changed,
+660
−42
internal/activator/activator.go+35 −9internal/activator/claimwake.go+12 −0internal/activator/generations_file.go+116 −21internal/activator/generations_instances_test.go+423 −0internal/activator/opset.go+2 −2internal/activator/swap.go+10 −1internal/activator/wakeplan.go+6 −0internal/activator/wakeplan_test.go+4 −2internal/activator/workerbuild.go+7 −5internal/activator/workerswap.go+3 −0internal/host/versionsweep.go+5 −0internal/host/versionsweep_test.go+17 −0internal/release/pins.go+3 −2internal/release/pins_test.go+17 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 1b3370b6e..efc47a9fe 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -1015,10 +1015,12 @@ type Service struct { |
| servingMu sync.Mutex |
| // GenerationsFile, when set, is where the activator publishes every |
| - // worker's captured serving build for root's pin set |
| - // (release.GenerationsPath of the data dir root's passes use as BakeDir; |
| - // see generations.go). Empty publishes nothing. Set before Start and |
| - // before any wake, like OnWake. |
| + // worker's captured serving build and every live or pending instance |
| + // generation's build for root's pin set (release.GenerationsPath of the |
| + // data dir root's passes use as BakeDir; see generations_file.go). A |
| + // staged spawn whose build cannot be written there first is refused. |
| + // Empty publishes nothing. Set before Start and before any wake, like |
| + // OnWake. |
| GenerationsFile string |
| // schedMu guards schedFences, each host's schedule write fence |
| @@ -1371,9 +1373,18 @@ func (a *Service) wake(e *actEntry) { |
| servingID release.StagedIdentity |
| servingInc uint64 |
| publishServing bool |
| + republish bool |
| ) |
| a.mu.Lock() |
| if err != nil { |
| + // The failed wake's entry goes, and with it everything it published |
| + // (a worker's captured build, a plan's builds, a pinned build); a |
| + // generation retained as an unproven wake is listed from there. |
| + republish = e.worker || a.publishesLocked(e) |
| + if u := a.unprovenWakes[e.host]; u != nil { |
| + _, listed := a.genIdentity(u.gen) |
| + republish = republish || listed |
| + } |
| e.err = err |
| outcome := &stopObservation{uncertain: true} |
| if w != nil && w.stopOutcome != nil { |
| @@ -1429,6 +1440,9 @@ func (a *Service) wake(e *actEntry) { |
| if e.gen.N != 0 { |
| a.noteCurrent(e.host, e.gen.N) |
| } |
| + // An instance's current generation is in the generations file from |
| + // here (spec §8.5). A worker's build was published at its install. |
| + republish = !e.worker && a.publishesLocked(e) |
| if e.sidecarGen == 0 { |
| // A wake that carried no generation number (the claim path) ran |
| // its sidecar on the live link, which is that wake's own build. |
| @@ -1482,9 +1496,9 @@ func (a *Service) wake(e *actEntry) { |
| // publication that blocks cannot hold that record up. |
| a.publishWorkerServing(e, servingID, servingInc) |
| } |
| - if err != nil && e.worker { |
| - // The failed wake's entry is gone, and with it any build its spawn |
| - // published (generations.go). |
| + if republish { |
| + // After a change, the write only logs: the file it failed to |
| + // replace pins at least as much, and the sweep retries it. |
| _ = a.persistGenerations() |
| } |
| } |
| @@ -1663,6 +1677,16 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| e.plan, e.firstWake = plan, plan.FirstWake |
| a.mu.Unlock() |
| } |
| + // 0.8. A staged build, planned or pinned, is published before anything |
| + // execs it (spec §4.2, §8.5): the entry now names it (and a plan's |
| + // fallback), so the write puts both in root's pin set. A failed write |
| + // refuses the wake here, before the lease: nothing is taken, nothing |
| + // runs, and no build is blamed (ErrBuildNotPublished). |
| + if plan.planned() || e.pin != nil { |
| + if err := a.publishBeforeExec(); err != nil { |
| + return w, fmt.Errorf("%s: refusing to wake: %w", host, err) |
| + } |
| + } |
| // 1. The lease first: everything after this line is done as the prefix's |
| // single writer. Contention is not retried here — on one box it means a |
| @@ -3417,6 +3441,7 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) { |
| outcome.uncertain = !outcome.complete() |
| e.stopOutcome = outcome |
| a.rememberStopLocked(e.host, outcome) |
| + republish := e.worker || a.publishesLocked(e) |
| delete(a.entries, e.host) |
| retired := e.gen |
| // The canonical link goes with the last generation (spec §7.1). Under |
| @@ -3439,8 +3464,9 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) { |
| h.woke <- holdWoke{next: next, err: herr} |
| } |
| a.mu.Unlock() |
| - if e.worker { |
| - // Its child is gone, so its build is no longer referenced by it. |
| + if republish { |
| + // Its processes are gone, so its builds are no longer referenced by |
| + // them. |
| _ = a.persistGenerations() |
| } |
| // The generation stopped being current when its entry went (spec §7.2: |
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go| index b0188ba60..b368192a0 100644 |
| --- a/internal/activator/claimwake.go |
| +++ b/internal/activator/claimwake.go |
| @@ -201,6 +201,18 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance, |
| a.mu.Unlock() |
| } |
| } |
| + // A pinned or planned build is published before the admission, as |
| + // doWake does before its lease (spec §4.2, §8.5): claimed wakes never |
| + // reach doWake's publication. A failed write refuses the wake with |
| + // nothing admitted and no build blamed. |
| + a.mu.Lock() |
| + staged := e.pin != nil || e.plan.planned() |
| + a.mu.Unlock() |
| + if staged { |
| + if err := a.publishBeforeExec(); err != nil { |
| + return nil, fmt.Errorf("%s: refusing to wake: %w", rt.Host, err) |
| + } |
| + } |
| 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 |
diff --git a/internal/activator/generations_file.go b/internal/activator/generations_file.go| index 9029ca3e3..2cbcd5848 100644 |
| --- a/internal/activator/generations_file.go |
| +++ b/internal/activator/generations_file.go |
| @@ -23,6 +23,24 @@ import ( |
| // (when it has one). It does so whatever CARLOS_SWAP says; this |
| // flag-independent writer is the one plan 5 extends. |
| // |
| +// Plan 5 (Task 12) adds every live or pending INSTANCE generation (spec |
| +// §8.5: current, successor, retiring, spawning): the current generation, a |
| +// hold's pinned one, a planned wake's build and its fallback, every |
| +// operation-set member, a failed wake retained until its end is proven, and |
| +// a socketless attempt's builds. With generations off an instance runs from |
| +// the live link, which root pins already, so an instance generation is |
| +// listed only with generations on or when it was exec'd from a staged path |
| +// (genIdentity); a flag-off file says exactly what plan 4's said. |
| +// |
| +// Every staged instance spawn (a planned wake and its fallback, a hold's |
| +// pinned wake, an overlap successor) publishes the entry state that names |
| +// its build BEFORE the exec, and does not exec when that write fails |
| +// (ErrBuildNotPublished): doWake and doClaimWake before the lease, checkPlan |
| +// for a replanned build, swapOverlap before its Step 3a. The entry state is |
| +// set first (e.plan, e.pin, the operation set) and then written, rather than |
| +// 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. |
| +// |
| // The file is written 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), |
| @@ -68,11 +86,9 @@ func (a *Service) dropPendingGenerationLocked(k pendingGeneration) { |
| } |
| // generationsSnapshotLocked is what the generations file should say now: for |
| -// every worker entry (waking, awake, mid-respawn or stopping, since its child |
| -// may still run) its captured serving build and the build of any shed child |
| -// its operation set (or, with generations off, the entry) retains, the build |
| -// of every retained failed worker wake (unprovenWakes), and every pending |
| -// build, each |
| +// every entry (waking, awake, mid-swap, mid-respawn or stopping, since its |
| +// processes may still run) what entryGenerationsLocked names, the build of |
| +// every retained failed wake (unprovenWakes), and every pending build, each |
| // listed under its host by version when the version is a valid version name, |
| // and by staged path (made absolute) when it has one. A build with neither |
| // names nothing root could restage or sweep. Each host's lists are sorted and |
| @@ -101,24 +117,16 @@ func (a *Service) generationsSnapshotLocked() release.Generations { |
| } |
| } |
| for host, e := range a.entries { |
| - if e.worker { |
| - add(host, e.workerServing) |
| - } |
| - // A shed respawn child retained in the operation set until its |
| - // termination is proven (final review F12). |
| - for _, m := range e.ops.members { |
| - add(host, m.build) |
| - } |
| - // With generations off, the entry keeps such a child itself |
| - // (final review R2-1). |
| - if e.shed != nil { |
| - add(host, e.shed.build) |
| - } |
| + a.entryGenerationsLocked(e, func(id release.StagedIdentity) { add(host, id) }) |
| } |
| - // A failed worker wake retained until its termination is proven: its |
| - // entry is gone, but its process may still run the build (F12). |
| + // A failed wake retained until its termination is proven: its entry is |
| + // gone, but its process may still run the build (F12; spec §4.2, §8.5). |
| + // A worker's is its captured identity, an instance's its generation. |
| for host, u := range a.unprovenWakes { |
| add(host, u.build) |
| + if id, ok := a.genIdentity(u.gen); ok { |
| + add(host, id) |
| + } |
| } |
| for _, pb := range a.genPending { |
| add(pb.host, pb.id) |
| @@ -126,6 +134,93 @@ func (a *Service) generationsSnapshotLocked() release.Generations { |
| return release.Generations{Hosts: sortedLists(hosts), Paths: sortedLists(paths)} |
| } |
| +// entryGenerationsLocked calls add with every build e's processes run or may |
| +// be about to run (spec §8.5): |
| +// - a worker's captured serving build; |
| +// - the current generation, and every operation-set member's (successor, |
| +// retiring, spawning), and the build of a shed worker child the set (or, |
| +// with generations off, the entry) retains (final review F12, R2-1); |
| +// - a hold's pinned generation, from the moment its waking entry exists; |
| +// - a planned wake's build and its fallback, from the moment the plan is |
| +// stored, so both are on disk before either execs; |
| +// - a socketless attempt's build and fallback (Task 11). |
| +// |
| +// The plan's builds stay listed while the entry lives, as e.plan does |
| +// (afterWake reads it after the wake): a few more pins than the processes |
| +// strictly need, never fewer. The caller holds a.mu. |
| +func (a *Service) entryGenerationsLocked(e *actEntry, add func(release.StagedIdentity)) { |
| + gen := func(g Gen) { |
| + if id, ok := a.genIdentity(g); ok { |
| + add(id) |
| + } |
| + } |
| + if e.worker { |
| + add(e.workerServing) |
| + } |
| + gen(e.gen) |
| + if e.pin != nil { |
| + gen(*e.pin) |
| + } |
| + if e.plan.planned() { |
| + add(e.plan.Boot) |
| + if e.plan.Fallback != nil { |
| + add(*e.plan.Fallback) |
| + } |
| + } |
| + for _, m := range e.ops.members { |
| + add(m.build) |
| + gen(m.gen) |
| + } |
| + if e.shed != nil { |
| + add(e.shed.build) |
| + } |
| + for _, v := range e.attemptBuilds { |
| + add(release.StagedIdentity{Version: v}) |
| + } |
| +} |
| + |
| +// genIdentity is g as the generations file names it, and whether it is |
| +// listed at all. A generation exec'd from a staged path is listed whatever |
| +// the switch says. Otherwise it is listed only with generations on: with |
| +// them off an instance's generation is the live link's build, which root |
| +// pins as the row's Version already, and plan 4's flag-off file never |
| +// named it. |
| +func (a *Service) genIdentity(g Gen) (release.StagedIdentity, bool) { |
| + if g.StagedPath == "" && (!a.gens || g.Build == "") { |
| + return release.StagedIdentity{}, false |
| + } |
| + return release.StagedIdentity{Version: g.Build, StagedPath: g.StagedPath}, true |
| +} |
| + |
| +// publishesLocked reports whether e names anything in the generations file, |
| +// so a change that removes e must rewrite it. The caller holds a.mu. |
| +func (a *Service) publishesLocked(e *actEntry) bool { |
| + named := false |
| + a.entryGenerationsLocked(e, func(id release.StagedIdentity) { |
| + if id.StagedPath != "" || release.ValidVersion(id.Version) { |
| + named = true |
| + } |
| + }) |
| + return named |
| +} |
| + |
| +// publishBeforeExec writes the generations file before a staged instance |
| +// spawn's exec (spec §4.2, §8.5; the index's generations-file contract). |
| +// The caller has already put the build about to run on its entry (e.plan, |
| +// e.pin, the operation set), so the snapshot names it. When the write fails |
| +// the caller does not exec: root reads only what the file names, and could |
| +// sweep or restage a build it does not name under the process. The refusal |
| +// is ErrBuildNotPublished, which classifyStep reads as abandonment, never a |
| +// failure of the build. The write error itself goes to the operator log |
| +// (persistGenerations), not into the returned error, which may reach a |
| +// member's wake event. Never call it with a.mu held. |
| +func (a *Service) publishBeforeExec() error { |
| + if err := a.persistGenerations(); err != nil { |
| + return ErrBuildNotPublished |
| + } |
| + return nil |
| +} |
| + |
| func sortedLists(m map[string]map[string]bool) map[string][]string { |
| out := make(map[string][]string, len(m)) |
| for k, set := range m { |
| @@ -161,7 +256,7 @@ func (a *Service) persistGenerationsGenLocked() error { |
| a.mu.Unlock() |
| if err := release.WriteGenerations(a.GenerationsFile, g); err != nil { |
| a.genDirty = true |
| - log.Printf("activator: publish worker generations to %s: %v; the sweep retries", a.GenerationsFile, err) |
| + log.Printf("activator: publish generations to %s: %v; the sweep retries", a.GenerationsFile, err) |
| return err |
| } |
| a.genWritten = true |
diff --git a/internal/activator/generations_instances_test.go b/internal/activator/generations_instances_test.go| new file mode 100644 |
| index 000000000..238406ad2 |
| --- /dev/null |
| +++ b/internal/activator/generations_instances_test.go |
| @@ -0,0 +1,423 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "reflect" |
| + "slices" |
| + "strings" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostwrite" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// Zero-downtime plan 5 Task 12: the generations file names every live or |
| +// pending INSTANCE generation too (spec §4.2, §8.5), and the build a staged |
| +// instance spawn is about to exec is in a successfully written file before |
| +// that exec, or the spawn does not happen. |
| + |
| +// hostsOf is the generations file's hosts, or nil when it does not read. |
| +func hostsOf(path string) map[string][]string { |
| + g, err := release.ReadGenerations(path) |
| + if err != nil || g == nil { |
| + return nil |
| + } |
| + return g.Hosts |
| +} |
| + |
| +// waitForPublications waits until the generations file at path has had n |
| +// publications (its seq). A wake goroutine's last write comes after the |
| +// request it served has returned; a test whose file lives in a TempDir |
| +// waits for it, or the write races the directory's removal. |
| +func waitForPublications(t *testing.T, path string, n uint64) *release.Generations { |
| + t.Helper() |
| + var g *release.Generations |
| + waitFor(t, "the wake's last publication", func() bool { |
| + var err error |
| + g, err = release.ReadGenerations(path) |
| + return err == nil && g != nil && g.Seq >= n |
| + }) |
| + return g |
| +} |
| + |
| +// unwritableGenerationsFile is a path whose parent is a regular file, so |
| +// every write fails. |
| +func unwritableGenerationsFile(t *testing.T) string { |
| + t.Helper() |
| + blocker := filepath.Join(t.TempDir(), "not-a-dir") |
| + if err := os.WriteFile(blocker, nil, 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + return filepath.Join(blocker, "zdd", "generations.json") |
| +} |
| + |
| +// breakGenerationsFile makes every later write of path fail: its directory |
| +// is replaced by a regular file. |
| +func breakGenerationsFile(t *testing.T, path string) { |
| + t.Helper() |
| + dir := filepath.Dir(path) |
| + if err := os.RemoveAll(dir); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(dir, nil, 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func TestTheGenerationsFileFollowsEveryChange(t *testing.T) { |
| + dir := t.TempDir() |
| + good := stagedID(t, dir, "v1") |
| + a, _, _ := planActivator(t, WakePlan{Boot: good, Mode: ModeOverlap}) |
| + a.GenerationsFile = release.GenerationsPath(dir) |
| + if err := a.persistGenerations(); err != nil { |
| + t.Fatal(err) |
| + } |
| + g, err := release.ReadGenerations(a.GenerationsFile) |
| + if err != nil || g == nil || len(g.Hosts) != 0 { |
| + t.Fatalf("before any spawn: %+v, %v; want an empty file, written", g, err) |
| + } |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + g, err = release.ReadGenerations(a.GenerationsFile) |
| + if err != nil || !slices.Equal(g.Hosts["one.example.com"], []string{"v1"}) { |
| + t.Fatalf("awake: %+v, %v", g, err) |
| + } |
| + if !slices.Equal(g.Paths["one.example.com"], []string{good.StagedPath}) { |
| + t.Fatalf("awake: paths %v, want the staged %s", g.Paths, good.StagedPath) |
| + } |
| + a.BounceFor("one.example.com", CauseRestart) |
| + // Four publications: the empty file, the plan before the lease, the |
| + // awake generation, the stop (in either order with the third). |
| + if g := waitForPublications(t, a.GenerationsFile, 4); len(g.Hosts) != 0 || len(g.Paths) != 0 { |
| + t.Fatalf("after the stop: %+v, want no generations", g) |
| + } |
| +} |
| + |
| +func TestTheGenerationsFileNamesThePlannedBuildBeforeItSpawns(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + a.GenerationsFile = filepath.Join(dir, "zdd", "generations.json") |
| + b.gate = make(chan struct{}) |
| + woke := make(chan error, 1) |
| + go func() { |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err == nil { |
| + done() |
| + } |
| + woke <- err |
| + }() |
| + waitFor(t, "the spawning build and its fallback pinned before the spawn", func() bool { |
| + return len(b.started()) == 0 && slices.Equal(hostsOf(a.GenerationsFile)["one.example.com"], []string{"v1", "v2"}) |
| + }) |
| + close(b.gate) |
| + if err := <-woke; err != nil { |
| + t.Fatal(err) |
| + } |
| + waitForPublications(t, a.GenerationsFile, 2) // the plan, then the awake generation |
| +} |
| + |
| +// Spec §7.4, §8.1: a planned build the selection gate refuses after the |
| +// restore is planned again, and the replanned build is published before it |
| +// execs; when it cannot be, it is not exec'd, and that is no failure of |
| +// either build. |
| +func TestAReplannedBuildIsPublishedBeforeItExecs(t *testing.T) { |
| + for _, broken := range []bool{false, true} { |
| + name := "published" |
| + if broken { |
| + name = "unpublished" |
| + } |
| + t.Run(name, func(t *testing.T) { |
| + dir := t.TempDir() |
| + v2, v3 := stagedID(t, dir, "v2"), stagedID(t, dir, "v3") |
| + a, b, evs := planActivator(t, WakePlan{}) |
| + a.GenerationsFile = filepath.Join(dir, "zdd", "generations.json") |
| + var plans atomic.Int32 |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + if plans.Add(1) == 1 { |
| + return WakePlan{Boot: v2, Mode: ModeOverlap, Check: func(build string) error { |
| + if build != "v2" { |
| + return nil |
| + } |
| + if broken { |
| + breakGenerationsFile(t, a.GenerationsFile) |
| + } |
| + return errors.New("v2 is held now") |
| + }} |
| + } |
| + return WakePlan{Boot: v3, Mode: ModeOverlap} |
| + } |
| + if broken { |
| + _, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if !errors.Is(err, ErrBuildNotPublished) { |
| + t.Fatalf("err = %v, want ErrBuildNotPublished", err) |
| + } |
| + if gens := b.started(); len(gens) != 0 { |
| + t.Fatalf("spawned %+v with the replanned build unpublished", gens) |
| + } |
| + if k := evs.kinds(); len(k) != 0 { |
| + t.Fatalf("events %v: a publication failure is no failure of the build", k) |
| + } |
| + return |
| + } |
| + b.gate = make(chan struct{}) |
| + woke := make(chan error, 1) |
| + go func() { |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err == nil { |
| + done() |
| + } |
| + woke <- err |
| + }() |
| + waitFor(t, "the replanned v3 pinned before its exec", func() bool { |
| + return len(b.started()) == 0 && slices.Contains(hostsOf(a.GenerationsFile)["one.example.com"], "v3") |
| + }) |
| + close(b.gate) |
| + if err := <-woke; err != nil { |
| + t.Fatal(err) |
| + } |
| + if gens := b.started(); len(gens) != 1 || gens[0].Build != "v3" { |
| + t.Fatalf("spawned %+v, want the replanned v3 alone", gens) |
| + } |
| + waitForPublications(t, a.GenerationsFile, 3) // the plan, the replan, the awake generation |
| + }) |
| + } |
| +} |
| + |
| +// Spec §4.2, §8.5: a failed wake whose generation is not proven terminated |
| +// keeps its lease and its process may still run (plan 3, unprovenWakes), |
| +// though its entry is gone. Its build stays in the generations file until |
| +// the proof holds. |
| +func TestARetainedFailedWakeStaysPinned(t *testing.T) { |
| + a, _, _ := planActivator(t, WakePlan{}) |
| + a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json") |
| + a.mu.Lock() |
| + a.unprovenWakes = map[string]*unprovenWake{"one.example.com": {gen: Gen{N: 4, Build: "v2", StagedPath: "/opt/carlos/versions/v2/app"}}} |
| + a.mu.Unlock() |
| + if err := a.persistGenerations(); err != nil { |
| + t.Fatal(err) |
| + } |
| + g, err := release.ReadGenerations(a.GenerationsFile) |
| + if err != nil || g == nil || !slices.Equal(g.Hosts["one.example.com"], []string{"v2"}) || |
| + !slices.Equal(g.Paths["one.example.com"], []string{"/opt/carlos/versions/v2/app"}) { |
| + t.Fatalf("generations %+v, %v: want the retained failed wake's v2 pinned", g, err) |
| + } |
| +} |
| + |
| +// Spec §8.5: every generation of the operation set is named, the successor |
| +// before its exec, and the retiring one until its termination is proven. |
| +func TestTheGenerationsFileFollowsAnOverlapSwap(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.a.GenerationsFile = release.GenerationsPath(t.TempDir()) |
| + gate := make(chan struct{}) |
| + f.b.gmu.Lock() |
| + f.b.genGate = gate |
| + f.b.gmu.Unlock() |
| + res := make(chan SwapResult, 1) |
| + go func() { res <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| + waitFor(t, "the successor pinned beside the incumbent before its exec", func() bool { |
| + gens, _, _ := f.b.calls() |
| + return gens == 1 && slices.Equal(hostsOf(f.a.GenerationsFile)[f.host], []string{"v1", "v2"}) |
| + }) |
| + close(gate) |
| + if r := <-res; r.Outcome != SwapDone { |
| + t.Fatalf("Swap = %+v, want done", r) |
| + } |
| + waitForGenerations(t, f.a.GenerationsFile, map[string][]string{f.host: {"v2"}}) |
| +} |
| + |
| +// Spec §4.2, §8.5: root pins only what the file names. A staged instance |
| +// spawn whose build could not be published must not exec it. The refusal is |
| +// no failure of the build, takes no lease, and is never down. |
| +func TestAnUnpublishedBuildIsNeverSpawned(t *testing.T) { |
| + t.Run("planned wake", func(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.GenerationsFile = unwritableGenerationsFile(t) |
| + _, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if !errors.Is(err, ErrBuildNotPublished) { |
| + t.Fatalf("err = %v, want ErrBuildNotPublished", err) |
| + } |
| + if gens := b.started(); len(gens) != 0 { |
| + t.Fatalf("spawned %+v with the generations file unwritten", gens) |
| + } |
| + if acquires, _, _ := b.counts(); acquires != 0 { |
| + t.Fatal("an unpublished wake took the lease") |
| + } |
| + if k := evs.kinds(); len(k) != 0 { |
| + t.Fatalf("events %v: a publication failure is no failure of the build", k) |
| + } |
| + }) |
| + t.Run("overlap successor", func(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.a.GenerationsFile = unwritableGenerationsFile(t) |
| + res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) |
| + if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, ErrBuildNotPublished.Error()) { |
| + t.Fatalf("Swap = %+v, want abandoned: unpublished", res) |
| + } |
| + if gens, raises, restores := f.b.calls(); gens != 0 || raises != restores { |
| + t.Fatalf("spawned %d, raises %d, restores %d", gens, raises, restores) |
| + } |
| + f.a.mu.Lock() |
| + members := len(f.a.entries[f.host].ops.members) |
| + f.a.mu.Unlock() |
| + if members != 0 { |
| + t.Fatalf("%d members left in the operation set", members) |
| + } |
| + }) |
| + t.Run("hold's pinned wake", func(t *testing.T) { |
| + f := newSwapFixture(t) |
| + sink := &lockedSink{} |
| + f.a.Logs = sink |
| + f.a.GenerationsFile = unwritableGenerationsFile(t) |
| + res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) |
| + if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, ErrBuildNotPublished.Error()) { |
| + t.Fatalf("Swap = %+v, want abandoned: unpublished", res) |
| + } |
| + if gens, _, _ := f.b.calls(); gens != 0 { |
| + t.Fatalf("spawned %d generations with the pinned build unpublished", gens) |
| + } |
| + if got := sink.errors(); len(got) != 0 { |
| + t.Fatalf("member error events %q, want none for an abandoned hold", got) |
| + } |
| + }) |
| + t.Run("claimed planned wake", func(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, admits, prepares := claimedPlanActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + a.GenerationsFile = unwritableGenerationsFile(t) |
| + _, err := a.beginRequest(context.Background(), registry.Route{Host: "x.example", Kind: "instance", Hibernate: true}) |
| + if !errors.Is(err, ErrBuildNotPublished) { |
| + t.Fatalf("err = %v, want ErrBuildNotPublished", err) |
| + } |
| + if admits.Load() != 0 || prepares.Load() != 0 { |
| + t.Fatalf("admitted %d, prepared %d: an unpublished claimed wake reached the admission", admits.Load(), prepares.Load()) |
| + } |
| + }) |
| +} |
| + |
| +// claimedPlanActivator is a claims-on Service whose planner answers plan, |
| +// as TestAPlannedClaimedWakeRefusesAtTheLaunchRecordingNothing builds it, |
| +// counting the admissions and preparations a wake reaches. |
| +func claimedPlanActivator(t *testing.T, plan WakePlan) (*Service, *atomic.Int32, *atomic.Int32) { |
| + t.Helper() |
| + const host = "x.example" |
| + obs := claimedObs(host) |
| + obs.Claim.Op.Exec = "box" |
| + obs.Claim.Holder.WriterBox = "box" |
| + admitted := obs |
| + claim := *obs.Claim |
| + holder := *claim.Holder |
| + holder.Session, holder.Generation = 2, "two" |
| + claim.Holder = &holder |
| + admitted.Claim = &claim |
| + work, _ := (hostwrite.Config{}).Select(context.Background(), host, "spawn") |
| + var ( |
| + reads atomic.Int32 |
| + admits atomic.Int32 |
| + prepares atomic.Int32 |
| + ) |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + a := New(Config{Bucket: "test-bucket", Idle: time.Hour, HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", |
| + Read: func(context.Context, string) (hostclaim.Observed, error) { |
| + if reads.Add(1) == 1 { |
| + return obs, nil |
| + } |
| + return admitted, nil |
| + }, |
| + StopForeign: func(context.Context, string) error { return nil }, |
| + FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil }, |
| + Prerequisites: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error { return nil }, |
| + Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) { |
| + admits.Add(1) |
| + return admitted, nil |
| + }, |
| + Prepare: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (ClaimStart, error) { |
| + prepares.Add(1) |
| + return claimedStartFor(admitted, work, p.Pin.Build), nil |
| + }}}, &fakeBackend{}, reg) |
| + a.OnGen = (&genLog{}).add |
| + a.Planner = func(context.Context, registry.Route) WakePlan { return plan } |
| + return a, &admits, &prepares |
| +} |
| + |
| +// The flag-independent behaviour (rulings R17, R19): with generations off |
| +// the file says exactly what plan 4's writer said. A worker's captured build |
| +// is listed by version and by staged path; an instance, which runs from the |
| +// live link, adds nothing. |
| +func TestTheGenerationsFileIsUnchangedWithTheFlagOff(t *testing.T) { |
| + a, _, _, _, _ := stagedActivator(t) // generations off, no planner |
| + a.GenerationsFile = release.GenerationsPath(t.TempDir()) |
| + const bot = "bot.x.internal" |
| + if err := a.WakeWorker(alwaysOnWorkerRoute(bot)); err != nil { |
| + t.Fatal(err) |
| + } |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + a.mu.Lock() |
| + instGen := a.entries["one.example.com"].gen |
| + a.mu.Unlock() |
| + if instGen.Build == "" { |
| + t.Fatal("the instance's generation names no build: this test would prove nothing") |
| + } |
| + want := map[string][]string{bot: {"v1"}} |
| + wantPaths := map[string][]string{bot: {fakeIdentity("v1").StagedPath}} |
| + check := func(when string) { |
| + t.Helper() |
| + if err := a.persistGenerations(); err != nil { |
| + t.Fatal(err) |
| + } |
| + g := readGenerationsFile(t, a.GenerationsFile) |
| + if !reflect.DeepEqual(g.Hosts, want) || !reflect.DeepEqual(g.Paths, wantPaths) { |
| + t.Fatalf("%s: flag-off file names %v and paths %v, want %v and %v", when, g.Hosts, g.Paths, want, wantPaths) |
| + } |
| + } |
| + // With the instance awake, after a sweep, and after its stop: none of |
| + // it may add the instance. |
| + check("both awake") |
| + a.sweepOnce() |
| + check("after a sweep") |
| + a.mu.Lock() |
| + a.beginStopLocked(a.entries["one.example.com"]) |
| + a.mu.Unlock() |
| + waitFor(t, "the instance's stop", func() bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + return a.entries["one.example.com"] == nil |
| + }) |
| + check("after the instance's stop") |
| +} |
| + |
| +// Task 11's socketless attempt names its build and its fallback while it |
| +// runs (its fallback above all: startWorker publishes only the build about |
| +// to exec). |
| +func TestAWorkerAttemptsBuildsArePublished(t *testing.T) { |
| + a, _, _ := planActivator(t, WakePlan{}) |
| + a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json") |
| + a.mu.Lock() |
| + a.entries["bot.x.internal"] = &actEntry{host: "bot.x.internal", worker: true, attemptBuilds: []string{"v3", "v1"}} |
| + a.mu.Unlock() |
| + if err := a.persistGenerations(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := hostsOf(a.GenerationsFile)["bot.x.internal"]; !slices.Equal(got, []string{"v1", "v3"}) { |
| + t.Fatalf("attempt builds published as %v, want [v1 v3]", got) |
| + } |
| +} |
diff --git a/internal/activator/opset.go b/internal/activator/opset.go| index f1809303b..d0fb79bf5 100644 |
| --- a/internal/activator/opset.go |
| +++ b/internal/activator/opset.go |
| @@ -329,7 +329,7 @@ func (a *Service) leaveSet(e *actEntry, m *genMember) { |
| } |
| a.mu.Unlock() |
| close(m.done) |
| - if m.build != (release.StagedIdentity{}) { |
| + if _, listed := a.genIdentity(m.gen); listed || m.build != (release.StagedIdentity{}) { |
| // Its child is proven gone: its build is no longer referenced. |
| _ = a.persistGenerations() |
| } |
| @@ -622,7 +622,7 @@ func (a *Service) reapUnprovenWake(host string, u *unprovenWake) { |
| } |
| u.busy = false |
| a.mu.Unlock() |
| - if u.build != (release.StagedIdentity{}) { |
| + if _, listed := a.genIdentity(u.gen); listed || u.build != (release.StagedIdentity{}) { |
| // Proven gone: its build is no longer referenced. |
| _ = a.persistGenerations() |
| } |
diff --git a/internal/activator/swap.go b/internal/activator/swap.go| index d4b302af5..55238eb06 100644 |
| --- a/internal/activator/swap.go |
| +++ b/internal/activator/swap.go |
| @@ -436,8 +436,17 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n |
| return abandon(fmt.Errorf("raise parent limit: %w", err)) |
| } |
| + // The successor's build is published before its exec (spec §4.2, §8.5): |
| + // it has been in the operation set since admission, so this write names |
| + // it. A failed write abandons the attempt; nothing of it ran, so it is |
| + // never a failure of the build. |
| + if err := a.publishBeforeExec(); err != nil { |
| + a.dropMember(e, m) // restores the raised parent (leaveSet) |
| + return abandon(err) |
| + } |
| + |
| // Step 3a: the §7.4 checks again, at the exec (spec §6.3, §7.4). The |
| - // raise (and plan 5's pin publication, just above this step) took time: |
| + // raise (and the pin publication just above this step) took time: |
| // a newer attempt, a halted cohort, a lost lease or a fence since |
| // admission means nothing execs. Claims on, this runs inside the |
| // selected work as the admission did. |
diff --git a/internal/activator/wakeplan.go b/internal/activator/wakeplan.go| index 49c2a79b8..2b60d260f 100644 |
| --- a/internal/activator/wakeplan.go |
| +++ b/internal/activator/wakeplan.go |
| @@ -80,6 +80,12 @@ func (a *Service) checkPlan(ctx context.Context, e *actEntry, rt registry.Route, |
| a.mu.Lock() |
| e.plan, e.firstWake = next, next.FirstWake |
| a.mu.Unlock() |
| + // The replanned build (and its fallback) is published before it execs |
| + // (spec §4.2, §8.5). A failed write execs neither: the wake fails with |
| + // ErrBuildNotPublished, which is abandonment, never a failure. |
| + if err := a.publishBeforeExec(); err != nil { |
| + return next, err |
| + } |
| return next, nil |
| } |
diff --git a/internal/activator/wakeplan_test.go b/internal/activator/wakeplan_test.go| index d3895b955..79b7d13aa 100644 |
| --- a/internal/activator/wakeplan_test.go |
| +++ b/internal/activator/wakeplan_test.go |
| @@ -920,8 +920,10 @@ func TestAPlannedWorkerRefusedForPublicationRecordsNothing(t *testing.T) { |
| if n := b.workerStarts(); n != 0 { |
| t.Fatalf("%d worker children started for an unpublished build", n) |
| } |
| - if calls := b.calls(); len(calls) != 1 { |
| - t.Fatalf("spawn attempts %v: a refusal before the exec fell back", calls) |
| + // The plan is published before the lease (Task 12), so the refusal |
| + // comes before any spawn is attempted, let alone a fallback. |
| + if calls := b.calls(); len(calls) != 0 { |
| + t.Fatalf("spawn attempts %v: an unpublished plan reached a spawn", calls) |
| } |
| time.Sleep(50 * time.Millisecond) |
| if k := evs.kinds(); len(k) != 0 { |
diff --git a/internal/activator/workerbuild.go b/internal/activator/workerbuild.go| index c17dee2fa..4975ca77f 100644 |
| --- a/internal/activator/workerbuild.go |
| +++ b/internal/activator/workerbuild.go |
| @@ -27,11 +27,13 @@ var ( |
| _ stagedWorkerStarter = (*SplitBackend)(nil) |
| ) |
| -// ErrBuildNotPublished is a worker spawn refused because the build it was |
| -// about to exec could not be written to the generations file first (ruling |
| -// R19 item 3): root could not see that build as referenced, so it is not |
| -// run. The failed-spawn handling applies, and the sweep retries the write. |
| -var ErrBuildNotPublished = errors.New("worker build not published before its exec") |
| +// ErrBuildNotPublished is a spawn refused because the build it was about to |
| +// exec could not be written to the generations file first (ruling R19 item |
| +// 3; plan 5 Task 12 for staged instance spawns, publishBeforeExec): root |
| +// could not see that build as referenced, so it is not run. The |
| +// failed-spawn handling applies, and the sweep retries the write. It is |
| +// never a failure of the build (classifyStep). |
| +var ErrBuildNotPublished = errors.New("build not published before its exec") |
| // beforeExecKey carries the activator's before-exec hook on a spawn's |
| // context (withBeforeExec). |
diff --git a/internal/activator/workerswap.go b/internal/activator/workerswap.go| index 2eef2820e..ba3e30f27 100644 |
| --- a/internal/activator/workerswap.go |
| +++ b/internal/activator/workerswap.go |
| @@ -83,6 +83,9 @@ func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *relea |
| a.mu.Lock() |
| e.attemptBuilds = nil |
| a.mu.Unlock() |
| + // The attempt's builds are no longer pinned on its account; what |
| + // the worker runs now is its captured build, published at install. |
| + _ = a.persistGenerations() |
| }() |
| if !a.awaitRespawn(ctx, done) { |
diff --git a/internal/host/versionsweep.go b/internal/host/versionsweep.go| index 4579615b1..6ee76ffc8 100644 |
| --- a/internal/host/versionsweep.go |
| +++ b/internal/host/versionsweep.go |
| @@ -68,6 +68,11 @@ func sweepStagedVersions(reg routeLister, optDir, genPath string, grace time.Dur |
| if r.PrevVersion != "" { |
| pinned[r.PrevVersion] = true |
| } |
| + // GoodVersion likewise (spec §8.5): it is what a failed first wake |
| + // falls back to, however long ago it last served. |
| + if r.GoodVersion != "" { |
| + pinned[r.GoodVersion] = true |
| + } |
| } |
| // Every live or pending generation the agent published (spec §8.5, |
| // ruling R19 item 1): a worker's captured build stays pinned after the |
diff --git a/internal/host/versionsweep_test.go b/internal/host/versionsweep_test.go| index e2833a04f..493173085 100644 |
| --- a/internal/host/versionsweep_test.go |
| +++ b/internal/host/versionsweep_test.go |
| @@ -253,3 +253,20 @@ func TestAdoptAndRestartLocksAreDistinct(t *testing.T) { |
| } |
| inner() |
| } |
| + |
| +// Spec §8.5: the good build is a pin, with no grace, like PrevVersion: it is |
| +// the fallback a failed first wake boots, however long ago it last served. |
| +func TestSweepStagedVersionsKeepsTheGoodBuild(t *testing.T) { |
| + opt := t.TempDir() |
| + for _, v := range []string{"good", "gone"} { |
| + stageSweepVersion(t, opt, "acct1", v, "km", 48*time.Hour) |
| + } |
| + reg := fakeLister{routes: []registry.Route{{Host: "a.example.com", Kind: "instance", App: "km", Account: "acct1", Version: "x", GoodVersion: "good"}}} |
| + sweepStagedVersions(reg, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| + if !staged(t, opt, "good") { |
| + t.Error("the good build was removed") |
| + } |
| + if staged(t, opt, "gone") { |
| + t.Error("gone survived; nothing names it and it is older than the grace") |
| + } |
| +} |
diff --git a/internal/release/pins.go b/internal/release/pins.go| index 9dcc26cb1..e936e228d 100644 |
| --- a/internal/release/pins.go |
| +++ b/internal/release/pins.go |
| @@ -85,7 +85,8 @@ func (f PinSourceFunc) StagedPins(ctx context.Context, optDir string) (StagedPat |
| // relative hop pinned at the real path the OS reaches, and at its lexical |
| // join too when the two differ), plus the fully resolved real path of the |
| // chain's end when it exists (StagedPaths.addReal), and the staged path of |
| -// every row's Version and PrevVersion. routes is every registry row, unfiltered: a hibernating route, |
| +// every row's Version, PrevVersion and GoodVersion (the good build a failed |
| +// first wake falls back to, spec §8.5). routes is every registry row, unfiltered: a hibernating route, |
| // an expired canary and a mid-provision row all still refer to their builds. |
| // |
| // An entry that cannot be read is an error, never a skipped entry: a pin set |
| @@ -119,7 +120,7 @@ func BoxPins(optDir string, routes []registry.Route) (StagedPaths, error) { |
| if rt.App == "" { |
| continue |
| } |
| - for _, v := range []string{rt.Version, rt.PrevVersion} { |
| + for _, v := range []string{rt.Version, rt.PrevVersion, rt.GoodVersion} { |
| if ValidVersion(v) { |
| pins.add(VersionPath(optDir, RouteAccount(rt), v, rt.App)) |
| } |
diff --git a/internal/release/pins_test.go b/internal/release/pins_test.go| index 1a1ef766b..f076f0009 100644 |
| --- a/internal/release/pins_test.go |
| +++ b/internal/release/pins_test.go |
| @@ -340,3 +340,20 @@ func TestStagedPathsRefersComparesTheRealPath(t *testing.T) { |
| t.Fatalf("Refers(a missing copy) = %v, %v; want false, nil", ok, err) |
| } |
| } |
| + |
| +// Spec §8.5: the good build is a pin. It is what a failed first wake falls |
| +// back to and what a hold's recovery boots, long after the row's Version and |
| +// PrevVersion have moved past it. |
| +func TestBoxPinsNamesTheGoodBuild(t *testing.T) { |
| + opt := t.TempDir() |
| + if err := StageArtifact(opt, "bab", "v0", "km", []byte("v0")); err != nil { |
| + t.Fatal(err) |
| + } |
| + pins, err := BoxPins(opt, []registry.Route{{Host: "b.example.com", Kind: "instance", App: "km", Account: "bab", Version: "v2", PrevVersion: "v1", GoodVersion: "v0"}}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !pins.Has(VersionPath(opt, "bab", "v0", "km")) { |
| + t.Fatal("GoodVersion is not a pin (spec §8.5)") |
| + } |
| +} |