amadan

carlos / platform Public

zdd-5 task 12: Start writes the generations file first; restore the Task 9 publication test; assert no events on the claimed refusal

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 13c5b8cdbfc1d43ebab1f0cf755d2df7104bf82e parent 221f63e
4 files changed, +136 −7
  • internal/activator/activator.go +8 −0
  • internal/activator/generations_file.go +1 −1
  • internal/activator/generations_instances_test.go +76 −5
  • internal/activator/wakeplan_test.go +51 −1
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index efc47a9fe..22b9518f2 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -3482,6 +3482,14 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
// Start launches the background sweeper that puts idle instances to sleep and
// stands down lost leases/dead children between requests.
func (a *Service) Start() {
+ // The generations file is written before anything spawns, whatever
+ // CARLOS_SWAP says (the index contract): a file left by a previous
+ // incarnation is stale, and one a crash cut short reads to root as an
+ // incomplete pin set, so root would neither restage nor sweep until this
+ // process changes it. The write is off a.mu, and a failure is logged and
+ // marks the publication dirty for the sweep to retry; it never fails or
+ // blocks Start.
+ _ = a.persistGenerations()
a.sweepStop = make(chan struct{})
a.sweepDone = make(chan struct{})
// Sweep well inside the idle window so a sleep happens promptly after it
diff --git a/internal/activator/generations_file.go b/internal/activator/generations_file.go
index 2cbcd5848..19fcce447 100644
--- a/internal/activator/generations_file.go
+++ b/internal/activator/generations_file.go
@@ -41,7 +41,7 @@ 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.
//
-// The file is written before the first worker spawn (so it exists, even
+// 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),
// after every spawn or respawn installs its captured identity in place of
diff --git a/internal/activator/generations_instances_test.go b/internal/activator/generations_instances_test.go
index 238406ad2..d9afc3f35 100644
--- a/internal/activator/generations_instances_test.go
+++ b/internal/activator/generations_instances_test.go
@@ -295,7 +295,7 @@ func TestAnUnpublishedBuildIsNeverSpawned(t *testing.T) {
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, admits, prepares, evs := 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) {
@@ -304,13 +304,17 @@ func TestAnUnpublishedBuildIsNeverSpawned(t *testing.T) {
if admits.Load() != 0 || prepares.Load() != 0 {
t.Fatalf("admitted %d, prepared %d: an unpublished claimed wake reached the admission", admits.Load(), prepares.Load())
}
+ if k := evs.kinds(); len(k) != 0 {
+ t.Fatalf("events %v: a publication failure is no failure of the build", k)
+ }
})
}
// 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) {
+// counting the admissions and preparations a wake reaches, and logging the
+// generation events it records.
+func claimedPlanActivator(t *testing.T, plan WakePlan) (*Service, *atomic.Int32, *atomic.Int32, *genLog) {
t.Helper()
const host = "x.example"
obs := claimedObs(host)
@@ -350,9 +354,10 @@ func claimedPlanActivator(t *testing.T, plan WakePlan) (*Service, *atomic.Int32,
prepares.Add(1)
return claimedStartFor(admitted, work, p.Pin.Build), nil
}}}, &fakeBackend{}, reg)
- a.OnGen = (&genLog{}).add
+ evs := &genLog{}
+ a.OnGen = evs.add
a.Planner = func(context.Context, registry.Route) WakePlan { return plan }
- return a, &admits, &prepares
+ return a, &admits, &prepares, evs
}
// The flag-independent behaviour (rulings R17, R19): with generations off
@@ -421,3 +426,69 @@ func TestAWorkerAttemptsBuildsArePublished(t *testing.T) {
t.Fatalf("attempt builds published as %v, want [v1 v3]", got)
}
}
+
+// The index contract: the generations file is written before the agent's
+// first spawn, whatever CARLOS_SWAP says. A file left by a previous
+// incarnation is stale (it may over-pin, or have been cut short by a crash,
+// which root reads as incomplete and answers by neither restaging nor
+// sweeping), so Start replaces it with this incarnation's snapshot, which is
+// empty: nothing has spawned yet.
+func TestStartReplacesAStaleGenerationsFile(t *testing.T) {
+ stale := func(t *testing.T, path string) {
+ t.Helper()
+ if err := release.WriteGenerations(path, release.Generations{
+ Hosts: map[string][]string{"gone.example.com": {"v9"}},
+ Paths: map[string][]string{"gone.example.com": {"/opt/carlos/versions/v9/app"}},
+ }); err != nil {
+ t.Fatal(err)
+ }
+ }
+ for _, tc := range []struct {
+ name string
+ setup func(t *testing.T, path string)
+ }{
+ {"a complete file from a previous run", stale},
+ {"an intent with the file missing", func(t *testing.T, path string) {
+ stale(t, path)
+ if err := os.Remove(path); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ {"an intent ahead of the file's seq", func(t *testing.T, path string) {
+ stale(t, path)
+ if err := os.WriteFile(release.GenerationsSeqPath(path), []byte("7\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ a, _, _, _, _ := stagedActivator(t) // generations off, no planner, no instances
+ a.GenerationsFile = release.GenerationsPath(t.TempDir())
+ tc.setup(t, a.GenerationsFile)
+ a.Start()
+ t.Cleanup(func() { a.Shutdown(context.Background()) })
+ g, err := release.ReadGenerations(a.GenerationsFile)
+ if err != nil || g == nil {
+ t.Fatalf("after Start the file reads %+v, %v; want a complete publication", g, err)
+ }
+ if len(g.Hosts) != 0 || len(g.Paths) != 0 {
+ t.Fatalf("after Start the file names %v and %v, want nothing", g.Hosts, g.Paths)
+ }
+ })
+ }
+}
+
+// A write that fails at Start is logged and left to the sweep: it does not
+// block or fail Start.
+func TestStartDoesNotFailOnAnUnwritableGenerationsFile(t *testing.T) {
+ a, _, _, _, _ := stagedActivator(t)
+ a.GenerationsFile = unwritableGenerationsFile(t)
+ a.Start()
+ t.Cleanup(func() { a.Shutdown(context.Background()) })
+ a.genMu.Lock()
+ dirty := a.genDirty
+ a.genMu.Unlock()
+ if !dirty {
+ t.Fatal("a failed write at Start did not mark the publication dirty for the sweep")
+ }
+}
diff --git a/internal/activator/wakeplan_test.go b/internal/activator/wakeplan_test.go
index 79b7d13aa..e04a9f852 100644
--- a/internal/activator/wakeplan_test.go
+++ b/internal/activator/wakeplan_test.go
@@ -894,10 +894,60 @@ func TestAFailedWorkerFirstWakeFallsBackToTheGoodBuild(t *testing.T) {
}
}
+// breakOnAcquire is a staged backend that makes the generations file
+// unwritable as the wake takes its lease: after the wake's own publication
+// (doWake's step 0.8) and before startWorker's before-exec publication.
+type breakOnAcquire struct {
+ *stagedFake
+ t *testing.T
+ path string
+}
+
+func (b *breakOnAcquire) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
+ breakGenerationsFile(b.t, b.path)
+ return b.stagedFake.Acquire(ctx, rt)
+}
+
// PF9: a planned worker spawn refused because its build could not be
// published first (ErrBuildNotPublished) ran nothing of the build. It is
-// abandonment: no failure recorded, nothing falls back.
+// abandonment: no failure recorded, nothing falls back. The wake's own
+// publication (step 0.8) succeeds here; the file breaks afterwards, so the
+// refusal is startWorker's before-exec publication, with the spawn attempt
+// reaching the backend and refused inside it.
func TestAPlannedWorkerRefusedForPublicationRecordsNothing(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ a.gens = true
+ a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json")
+ a.backend = &breakOnAcquire{stagedFake: b, t: t, path: a.GenerationsFile}
+ evs := &genLog{}
+ a.OnGen = evs.add
+ dir := t.TempDir()
+ good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
+ a.Planner = func(context.Context, registry.Route) WakePlan {
+ return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
+ }
+ err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal"))
+ if !errors.Is(err, ErrBuildNotPublished) {
+ t.Fatalf("WakeWorker = %v, want ErrBuildNotPublished", err)
+ }
+ 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)
+ }
+ time.Sleep(50 * time.Millisecond)
+ if k := evs.kinds(); len(k) != 0 {
+ t.Fatalf("events %v: a refusal before the exec was recorded", k)
+ }
+}
+
+// The wake's own publication (Task 12, doWake step 0.8) refuses a planned
+// worker before the lease when the file cannot be written at all: nothing is
+// taken, no spawn is attempted, let alone a fallback, and nothing is
+// recorded.
+func TestAPlannedWorkerRefusedAtItsPublicationAttemptsNoSpawn(t *testing.T) {
shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond)
a, b, _, _, _ := stagedActivator(t)
a.gens = true