amadan

carlos / platform Public

activator: a one-shot is good only after a clean park, and a waiting wake never starts a second

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude deb4c81176a0d0362fe992039d7bc2396ed7ce71 parent c910d02
3 files changed, +253 −51
  • internal/activator/activator.go +85 −49
  • internal/activator/workerswap.go +8 −2
  • internal/activator/workerswap_test.go +160 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index d815c595e..fc67b3c18 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -612,8 +612,11 @@ type actEntry struct {
// heldServing is the publication holdServing kept back, delivered by
// PublishHeldServing. workerInc is the worker child's incarnation,
// bumped at every install (a wake, a respawn): an attempt counts only if
- // it moved, and a publication is delivered only for the latest. All
- // written under a.mu.
+ // it moved, and a publication is delivered only for the latest.
+ // parkedInc is the incarnation a clean-exit park stopped
+ // (markCleanParkLocked), 0 for none: the only stop after which
+ // ReplaceWorker counts an exited one-shot as a good build. All written
+ // under a.mu.
attemptBuilds []string
retireTook time.Duration
attemptGate func() error
@@ -622,6 +625,7 @@ type actEntry struct {
holdServing bool
heldServing *heldServing
workerInc uint64
+ parkedInc uint64
// shed is a respawn child, refused at its installation with generations
// off, whose exit is not proven yet (final review R2-1). With
@@ -868,6 +872,11 @@ func ConvergeDecision(cur string, readErr error, spawnVer string, staleSince, no
// forever. A var so tests can shrink it.
var wakeWaitBound = 30 * time.Second
+// wakeBoundExtended is a test seam, nil in production: awaitWake calls it
+// right after a first wake's waiter extends its bound (spec §8.2), so a test
+// can fail the wake inside the extension window.
+var wakeBoundExtended func()
+
// errLeaseHeld: the per-host lease is held elsewhere. On one box this means a
// previous incarnation hasn't released (crashed agent; lease expires within
// ~lease.LeaseTTL) — the edge answers 503 with Retry-After.
@@ -1231,32 +1240,16 @@ func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (rel
case actWaking:
ready := e.ready
a.mu.Unlock()
- select {
- case <-ready:
- // A failed wake reports to every coalesced waiter (retrying
- // here would turn one bad wake into a storm); a successful
- // one loops to register against the now-awake entry.
- if e.err != nil {
- return nil, e.err
- }
- continue
- case <-ctx.Done():
- return nil, ctx.Err()
- case <-deadline.C:
- // A first wake onto a new build gives its waiters arrival +
- // 55 s (zero-downtime spec §8.2): enough for B's health
- // window, B's kill and the good build's boot. Only that kind
- // of wake; firstWake is set only by a planner (flag on).
- a.mu.Lock()
- fw := e.firstWake
- a.mu.Unlock()
- if fw && !extended {
- extended = true
- deadline.Reset(time.Until(arrival.Add(firstWakeBudget)))
- continue
- }
- return nil, errWakeTimeout
+ if err := a.awaitWake(ctx, e, ready, deadline, arrival, &extended); err != nil {
+ return nil, err
+ }
+ // A failed wake reports to every coalesced waiter (retrying
+ // here would turn one bad wake into a storm); a successful one
+ // loops to register against the now-awake entry.
+ if e.err != nil {
+ return nil, e.err
}
+ continue
case actStopping:
stopped := e.stopped
@@ -1278,6 +1271,42 @@ func (a *Service) beginRequest(ctx context.Context, rt registry.Route) (release
return a.begin(ctx, rt, true)
}
+// awaitWake is a waiter's wait on one wake: begin's and WakeWorker's. It
+// returns nil once ready closes (the caller then reads e.err), ctx.Err(), or
+// errWakeTimeout at the deadline. A first wake onto a new build gives its
+// waiters arrival + firstWakeBudget (zero-downtime spec §8.2): enough for
+// B's health window, B's kill and the good build's boot. Only that kind of
+// wake (firstWake is set only by a planner, flag on), and once per caller:
+// *extended carries across the caller's loop.
+//
+// After the extension it waits again on the SAME ready and never goes back
+// to the entries map (round 2 Task 6). A wake that fails inside the
+// extension closes ready and deletes its entry; a waiter that re-read the
+// map found nothing and started a second wake, two lease acquires and two
+// spawns for one caller (#184's shape). Called without a.mu.
+func (a *Service) awaitWake(ctx context.Context, e *actEntry, ready <-chan struct{}, deadline *time.Timer, arrival time.Time, extended *bool) error {
+ for {
+ select {
+ case <-ready:
+ return nil
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-deadline.C:
+ a.mu.Lock()
+ fw := e.firstWake
+ a.mu.Unlock()
+ if !fw || *extended {
+ return errWakeTimeout
+ }
+ *extended = true
+ deadline.Reset(time.Until(arrival.Add(firstWakeBudget)))
+ if wakeBoundExtended != nil {
+ wakeBoundExtended()
+ }
+ }
+ }
+}
+
// errSidecarAsleep: a sidecar-marked request arrived for a host with no awake
// instance. Sidecar traffic never wakes anything — the activator itself spawns
// the sidecar as part of a wake, so a marked request for a sleeping host is
@@ -2826,29 +2855,17 @@ func (a *Service) WakeWorker(rt registry.Route) error {
case actWaking:
ready := e.ready
a.mu.Unlock()
- select {
- case <-ready:
- // A failed wake reports to every coalesced caller; retrying
- // here would turn one bad wake into a storm (beginRequest's
- // rule, for the same reason).
- return e.err
- case <-deadline.C:
- // A first wake onto a new build gives its waiters arrival +
- // firstWakeBudget (spec §8.2), a worker's caller as much as
- // a request: B's health window, B's kill and the good
- // build's boot. Without it a fallback that succeeds after
- // the ordinary bound was reported as a timeout (Astra R3
- // flag triage T10).
- a.mu.Lock()
- fw := e.firstWake
- a.mu.Unlock()
- if fw && !extended {
- extended = true
- deadline.Reset(time.Until(arrival.Add(firstWakeBudget)))
- continue
- }
- return errWakeTimeout
+ // A worker's caller gets a first wake's extension as much as a
+ // request does (Astra R3 flag triage T10): without it a
+ // fallback that succeeds after the ordinary bound was reported
+ // as a timeout.
+ if err := a.awaitWake(context.Background(), e, ready, deadline, arrival, &extended); err != nil {
+ return err
}
+ // A failed wake reports to every coalesced caller; retrying
+ // here would turn one bad wake into a storm (beginRequest's
+ // rule, for the same reason).
+ return e.err
case actStopping:
stopped := e.stopped
@@ -3942,6 +3959,21 @@ func (a *Service) Start() {
}()
}
+// markCleanParkLocked records that a clean-exit park is stopping e's current
+// worker incarnation (round 2 Task 6). ReplaceWorker's observation forgives
+// an exited one-shot that is no longer awake (spec §6.5) only when this mark
+// names the incarnation it watched: the park ran with no fence, the lease
+// still held and a clean exit. A stop for a lost lease, a fence or a
+// shutdown never marks, so the build is never recorded good on it. The
+// conditions are checked here again so the mark never depends on the
+// caller's order. The caller holds a.mu.
+func (a *Service) markCleanParkLocked(e *actEntry) {
+ if a.fences[e.host] != 0 || e.rep == nil || e.rep.Err() != nil || procCrashed(e.proc) {
+ return
+ }
+ e.parkedInc = e.workerInc
+}
+
// sweepOnce examines every awake instance once: idle ones hibernate; a lost
// lease or dead child stands down immediately regardless of activity. It then
// checks the due-time index and wakes any sleeping host whose scheduled
@@ -4055,6 +4087,10 @@ func (a *Service) sweepOnce() {
}
if e.worker {
log.Printf("activator: sidecar %s exited cleanly — its work is done; parking it", e.host)
+ // Past this sweep's fence skip and lease check: the one
+ // stop of an exited one-shot that a ReplaceWorker
+ // observation may still count as a good build.
+ a.markCleanParkLocked(e)
} else {
log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
}
diff --git a/internal/activator/workerswap.go b/internal/activator/workerswap.go
index a03d4ff8b..5ed469060 100644
--- a/internal/activator/workerswap.go
+++ b/internal/activator/workerswap.go
@@ -185,9 +185,15 @@ func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *relea
// a newer release or a halted cohort since the spawn means this
// build must not be recorded good. Its child stays until the newer
// attempt replaces it. A one-shot that finished may have been parked
- // meanwhile: no longer awake, but still not superseded.
+ // meanwhile: no longer awake, but still not superseded. Only the
+ // sweep's clean park of this very child counts (markCleanParkLocked,
+ // past the sweep's own fence and lease checks), and only while no
+ // fence or shutdown stands now. Any other stop of the exited child
+ // (a lost lease, a fence, the agent's shutdown) leaves it not awake
+ // too, with no ownership behind the exit, and is abandoned.
err := a.swapChecksLocked(e, at, newest)
- if r == settleExited && errors.Is(err, errSwapNotAwake) {
+ if r == settleExited && errors.Is(err, errSwapNotAwake) &&
+ e.parkedInc == inc1 && a.fences[at.Host] == 0 && !a.shuttingDown {
err = nil
if newest(at.Host) != at.SchedRev {
err = errSwapSuperseded
diff --git a/internal/activator/workerswap_test.go b/internal/activator/workerswap_test.go
index a2cbad45d..dc8ded9ad 100644
--- a/internal/activator/workerswap_test.go
+++ b/internal/activator/workerswap_test.go
@@ -848,3 +848,163 @@ func TestASupersededWorkerAttemptRestoresOnlyWhatProvablyNeverRan(t *testing.T)
})
}
}
+
+// Round 2 Task 6 (recon B §1): a one-shot's clean exit inside its
+// observation is a good build (spec §6.5) only when the sweep parked it
+// after its own fence and lease checks. Any other stop of the exited child
+// (a lost lease, a fence) leaves the entry not awake, and that answer was
+// forgiven for every exited one-shot, so the build was recorded good with no
+// ownership behind it. Those stops are abandoned; the sweep's park is done.
+func TestAnExitedOneShotIsGoodOnlyAfterACleanPark(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ stop func(t *testing.T, a *Service, b *stagedFake, host string)
+ clean bool
+ }{
+ {"parked by the sweep", func(t *testing.T, a *Service, b *stagedFake, host string) {
+ a.sweepOnce()
+ }, true},
+ {"lease lost, then the sweep", func(t *testing.T, a *Service, b *stagedFake, host string) {
+ b.lastRep().setErr(errors.New("renewal failed"))
+ a.sweepOnce()
+ }, false},
+ {"fenced", func(t *testing.T, a *Service, b *stagedFake, host string) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ if _, err := a.FenceAndStop(ctx, host, 1); err != nil {
+ t.Errorf("FenceAndStop: %v", err)
+ }
+ }, false},
+ {"parked, then fenced before the result", func(t *testing.T, a *Service, b *stagedFake, host string) {
+ a.sweepOnce()
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ if _, err := a.FenceAndStop(ctx, host, 1); err != nil {
+ t.Errorf("FenceAndStop: %v", err)
+ }
+ }, false},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ // A 2 s window polls every 250 ms (settlePoll): the stop below
+ // lands well before the observation's first look at the exit.
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ waitForWorkerStarts(t, b, 2)
+ waitFor(t, "the observation", func() bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ return e != nil && e.firstWatch != nil
+ })
+ b.lastWorker().exit() // status 0: the one-shot finished
+ tc.stop(t, a, b, host)
+ }()
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
+ <-done
+ if tc.clean {
+ if res.Outcome != SwapDone || !res.CleanExit {
+ t.Fatalf("ReplaceWorker = %+v, want done with a clean exit: the sweep parked a finished one-shot", res)
+ }
+ return
+ }
+ if res.Outcome != SwapAbandoned || res.CleanExit {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned: no clean park with no fence stands behind the exit, so its build must not be recorded good", res)
+ }
+ })
+ }
+}
+
+// awaitExtensionSeam installs wakeBoundExtended for one test: the first
+// extension of a waiter's bound lets the wake fail (release) and waits until
+// that failed wake has deleted its entry, the window in which a waiter that
+// re-read the entries map started a second wake.
+func awaitExtensionSeam(t *testing.T, a *Service, release chan struct{}) *atomic.Int32 {
+ t.Helper()
+ var calls atomic.Int32
+ old := wakeBoundExtended
+ wakeBoundExtended = func() {
+ if calls.Add(1) != 1 {
+ return
+ }
+ close(release)
+ waitFor(t, "the failed wake to delete its entry", func() bool { return a.entryCount() == 0 })
+ }
+ t.Cleanup(func() { wakeBoundExtended = old })
+ return &calls
+}
+
+// Round 2 Task 6 (recon B §2): a first wake's waiter extends its bound once
+// (spec §8.2) and then waits again on the SAME ready. It used to loop back to
+// the entries map: a wake that failed inside the extension had deleted its
+// entry, so the waiter found nothing and started a second wake, a second
+// lease acquire and a second spawn (#184's shape).
+func TestWakeWorkerAWakeFailingInsideTheExtensionStartsNoSecond(t *testing.T) {
+ old := wakeWaitBound
+ wakeWaitBound = 50 * time.Millisecond
+ t.Cleanup(func() { wakeWaitBound = old })
+ a, b, _, _, _ := stagedActivator(t)
+ a.gens = true
+ a.backend = &plannedHostKill{stagedFake: b}
+ next := stagedID(t, t.TempDir(), "v2")
+ a.Planner = func(context.Context, registry.Route) WakePlan {
+ return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true}
+ }
+ boom := errors.New("exec refused")
+ release := make(chan struct{})
+ calls := awaitExtensionSeam(t, a, release)
+ b.mu.Lock()
+ b.workerStartHook = func(registry.Route) {
+ <-release
+ b.mu.Lock()
+ b.workerErr = boom
+ b.mu.Unlock()
+ }
+ b.mu.Unlock()
+ err := a.WakeWorker(testWorkerRoute("bot.x.internal"))
+ if calls.Load() == 0 {
+ t.Fatal("the waiter never extended its bound: the test did not reach the window")
+ }
+ if !errors.Is(err, boom) {
+ t.Fatalf("WakeWorker = %v, want the wake's own error", err)
+ }
+ if acquires, _, _ := b.counts(); acquires != 1 {
+ t.Fatalf("%d lease acquires, want 1: a waiter whose wake failed inside the extension started a second wake", acquires)
+ }
+}
+
+// The same window for a request (begin): the failed wake reports to its
+// waiter, and nothing wakes the instance a second time.
+func TestBeginAWakeFailingInsideTheExtensionStartsNoSecond(t *testing.T) {
+ old := wakeWaitBound
+ wakeWaitBound = 50 * time.Millisecond
+ t.Cleanup(func() { wakeWaitBound = old })
+ next := stagedID(t, t.TempDir(), "v2")
+ a, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true})
+ boom := errors.New("GET / answered 500")
+ release := make(chan struct{})
+ calls := awaitExtensionSeam(t, a, release)
+ a.genHealth = func(context.Context, Gen) error {
+ <-release
+ return boom
+ }
+ _, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if calls.Load() == 0 {
+ t.Fatal("the waiter never extended its bound: the test did not reach the window")
+ }
+ if err == nil || errors.Is(err, errWakeTimeout) {
+ t.Fatalf("beginRequest = %v, want the failed wake's error", err)
+ }
+ if acquires, _, _ := b.counts(); acquires != 1 {
+ t.Fatalf("%d lease acquires, want 1: a waiter whose wake failed inside the extension started a second wake", acquires)
+ }
+ if gens := b.started(); len(gens) != 1 {
+ t.Fatalf("spawned %+v, want one generation", gens)
+ }
+}