amadan

carlos / platform Public

zdd-5 final review: deliver a planned wake's outcome before the next plan (F3)

A per-host outcome fence (outcomeFences, under a.mu). wake() takes it
for a failed or fallen-back first wake inside the critical section that
removes the entry (or makes it awake), before e.ready closes, and
releases it once afterWake's OnGen has returned. A first wake's
observation takes it when it is armed and releases it after its
terminal event has been delivered. A new wake of the host waits on it
before it calls the Planner (doWake and doClaimWake), bounded by
outcomeFenceWait (5 s: OnGen is in-memory work plus one local book
write; past the bound the wake plans as before, and the exec-time
check still reads the record).

The observer now delivers GenFailed and GenCancelled before it clears
firstWatch, so Current keeps hiding the failed generation until the
failure is recorded. GenSettled still clears it first: the scheduler
records a settled build as good only while Current shows it; the fence
still covers that delivery.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 9f15c771427a9d15e73aaa0fa3908daf8459e20e parent 35772c9
4 files changed, +354 −7
  • internal/activator/activator.go +28 −0
  • internal/activator/claimwake.go +3 −0
  • internal/activator/outcome_fence_test.go +211 −0
  • internal/activator/wakeplan.go +112 −7
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 04c073897..1eb57b6f4 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -512,6 +512,10 @@ type actEntry struct {
unexpectedExit bool
firstWatch context.CancelFunc
firstWatchCtx context.Context
+ // firstWatchRelease releases the host's outcome fence the observation
+ // took when it was armed (final review F3), once its terminal event has
+ // been delivered. Under a.mu.
+ firstWatchRelease func()
// Standalone sidecar ("worker") entries. A worker is this entry's own
// proc, not a passenger on someone else's: the co-spawn fields above stay
@@ -947,6 +951,13 @@ type Service struct {
// the host's next wake until the sweep proves it
// (reapUnprovenWakesLocked). Under a.mu.
unprovenWakes map[string]*unprovenWake
+ // outcomeFences are planned-wake outcomes taken but not yet delivered
+ // to OnGen, per host, made lazily (final review F3): a failed or
+ // fallen-back first wake's, from before its entry goes until afterWake's
+ // OnGen has returned, and a first wake's observation's, from its arming
+ // until its terminal event has been delivered. A new wake of the host
+ // waits for them (bounded) before it plans. Under a.mu.
+ outcomeFences map[string]*outcomeFence
sweepStop chan struct{}
sweepDone chan struct{}
@@ -1379,8 +1390,18 @@ func (a *Service) wake(e *actEntry) {
servingInc uint64
publishServing bool
republish bool
+ releaseOutcome func()
)
a.mu.Lock()
+ if a.outcomeOwedLocked(e, w) {
+ // A failed or fallen-back first wake (final review F3; spec §8.1,
+ // O7): its failure is recorded only by afterWake's OnGen, after this
+ // critical section and after e.ready closes. The host's outcome
+ // fence is taken here, before the entry goes and before anyone
+ // waiting is released, so a new wake of the host does not plan
+ // until that record exists. Released once OnGen has returned.
+ releaseOutcome = a.holdOutcomeLocked(e.host)
+ }
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
@@ -1493,6 +1514,9 @@ func (a *Service) wake(e *actEntry) {
// A planned wake's outcome, for the scheduler: off a.mu, since OnGen
// reaches its goodMu, which calls back into Current.
a.afterWake(e, w, err)
+ if releaseOutcome != nil {
+ releaseOutcome()
+ }
if publishServing {
// A worker's identity, after afterWake and never before it (spec
// §8.1; P5 review finding 7): for a fallback wake, afterWake's
@@ -1674,6 +1698,10 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
// planner, off a.mu, before the lease: a refused plan takes no lease.
var plan WakePlan
if a.Planner != nil && e.pin == nil {
+ // An earlier planned wake's outcome still on its way to the
+ // scheduler is waited for, bounded (final review F3): planning
+ // before its failure record exists would pick the failed build.
+ a.awaitOutcomes(ctx, host)
plan = a.Planner(ctx, rt)
if plan.Refuse != "" {
return w, fmt.Errorf("%s: refusing to wake: %s", host, plan.Refuse)
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
index b368192a0..a6fe26055 100644
--- a/internal/activator/claimwake.go
+++ b/internal/activator/claimwake.go
@@ -190,6 +190,9 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance,
// is called off a.mu. On this base every claims-on launch refuses
// before its exec hook (RoleLaunch.Start), so none of it runs a build.
if a.Planner != nil && e.pin == nil {
+ // As doWake: an earlier planned wake's outcome is delivered before
+ // this one plans (final review F3), bounded.
+ a.awaitOutcomes(ctx, rt.Host)
plan := a.Planner(ctx, rt)
if plan.Refuse != "" {
return nil, fmt.Errorf("%s: refusing to wake: %s", rt.Host, plan.Refuse)
diff --git a/internal/activator/outcome_fence_test.go b/internal/activator/outcome_fence_test.go
new file mode 100644
index 000000000..fb92e5651
--- /dev/null
+++ b/internal/activator/outcome_fence_test.go
@@ -0,0 +1,211 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "slices"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// blockedOnGen is an OnGen that blocks on events of kind until release
+// closes, as a scheduler whose failure transition is slow would, and marks
+// recorded once that event's handling has returned. Only the first such
+// event is the one a test waits on.
+type blockedOnGen struct {
+ kind GenEventKind
+ once sync.Once
+ entered chan struct{}
+ release chan struct{}
+ recorded atomic.Bool
+ evs genLog
+}
+
+func newBlockedOnGen(kind GenEventKind) *blockedOnGen {
+ return &blockedOnGen{kind: kind, entered: make(chan struct{}), release: make(chan struct{})}
+}
+
+func (o *blockedOnGen) on(ev GenEvent) {
+ o.evs.add(ev)
+ if ev.Kind != o.kind {
+ return
+ }
+ o.once.Do(func() { close(o.entered) })
+ <-o.release
+ o.recorded.Store(true)
+}
+
+func (o *blockedOnGen) waitEntered(t *testing.T) {
+ t.Helper()
+ select {
+ case <-o.entered:
+ case <-time.After(5 * time.Second):
+ t.Fatalf("OnGen never received %s; saw %v", o.kind, o.evs.kinds())
+ }
+}
+
+// secondPlanner answers the first plan with first, and records, for every
+// later plan, whether the blocked outcome had been recorded when it was
+// asked. A later plan refuses once it has, as the scheduler's planner does
+// for a held-down build.
+func secondPlanner(o *blockedOnGen, first WakePlan, saw chan<- bool) func(context.Context, registry.Route) WakePlan {
+ var n atomic.Int32
+ return func(context.Context, registry.Route) WakePlan {
+ if n.Add(1) == 1 {
+ return first
+ }
+ rec := o.recorded.Load()
+ saw <- rec
+ if rec {
+ return WakePlan{Refuse: "held down"}
+ }
+ return first
+ }
+}
+
+// Final review F3 (spec §8.1, O7): a failed first wake's outcome is
+// delivered before admission reopens. A request that arrives after the
+// failed wake's entry is gone, while OnGen is still recording the failure,
+// starts a wake that waits for that delivery and only then plans, with the
+// record in place. Before the fence the second wake planned B again in the
+// gap, before the hold-down existed.
+func TestANewWakeWaitsForAFailedFirstWakesOutcome(t *testing.T) {
+ dir := t.TempDir()
+ next := stagedID(t, dir, "v2")
+ a, b, _ := planActivator(t, WakePlan{})
+ const host = "one.example.com"
+ a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") }
+ o := newBlockedOnGen(GenFailed)
+ a.OnGen = o.on
+ saw := make(chan bool, 4)
+ a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true}, saw)
+ if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
+ t.Fatal("the failed hold first wake served")
+ }
+ o.waitEntered(t)
+ if a.Owns(host) {
+ t.Fatal("fixture: the failed wake's entry is still there")
+ }
+ done := make(chan error, 1)
+ go func() {
+ _, err := a.beginRequest(context.Background(), testRoute(host))
+ done <- err
+ }()
+ select {
+ case rec := <-saw:
+ close(o.release)
+ t.Fatalf("the second wake planned while the first wake's failure was still being delivered (record in place: %v)", rec)
+ case <-time.After(300 * time.Millisecond):
+ }
+ close(o.release)
+ select {
+ case rec := <-saw:
+ if !rec {
+ t.Fatal("the second wake planned before the failure was recorded")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("the second wake never planned")
+ }
+ if err := <-done; err == nil || !strings.Contains(err.Error(), "held down") {
+ t.Fatalf("second wake = %v, want refused by the recorded hold-down", err)
+ }
+ if gens := b.started(); len(gens) != 1 {
+ t.Fatalf("spawned %+v: B ran again before its failure was recorded", gens)
+ }
+}
+
+// The fence is bounded (outcomeFenceWait): a wake behind an OnGen that never
+// returns proceeds as it did before the fence, so a wedged scheduler cannot
+// keep a host down. The planner's exec-time check still reads whatever the
+// record says by then.
+func TestTheOutcomeFenceIsBounded(t *testing.T) {
+ old := outcomeFenceWait
+ outcomeFenceWait = 200 * time.Millisecond
+ t.Cleanup(func() { outcomeFenceWait = old })
+ dir := t.TempDir()
+ next := stagedID(t, dir, "v2")
+ a, _, _ := planActivator(t, WakePlan{})
+ const host = "one.example.com"
+ a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") }
+ o := newBlockedOnGen(GenFailed)
+ t.Cleanup(func() { close(o.release) })
+ a.OnGen = o.on
+ saw := make(chan bool, 4)
+ a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true}, saw)
+ if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
+ t.Fatal("the failed hold first wake served")
+ }
+ o.waitEntered(t)
+ start := time.Now()
+ go func() { _, _ = a.beginRequest(context.Background(), testRoute(host)) }()
+ select {
+ case <-saw:
+ if took := time.Since(start); took < 150*time.Millisecond {
+ t.Fatalf("planned after %v: it did not wait for the fence", took)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("the second wake never planned: the fence is not bounded")
+ }
+}
+
+// Final review F3, the observer's half: a first wake's 10 s observation
+// delivers its terminal event (here GenFailed, an exit inside the window)
+// BEFORE it clears firstWatch, and holds the host's outcome fence until
+// the delivery returns. While OnGen records the failure, Current still
+// hides the generation, and a request that stops the dead entry and wakes
+// it again waits for the record before planning. Before the fix the
+// observer cleared firstWatch first and nothing held the next wake back.
+func TestAFirstWakeObserverDeliversItsOutcomeBeforeAdmissionReopens(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
+ dir := t.TempDir()
+ good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
+ a, b, _ := planActivator(t, WakePlan{})
+ const host = "one.example.com"
+ o := newBlockedOnGen(GenFailed)
+ a.OnGen = o.on
+ saw := make(chan bool, 4)
+ a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}, saw)
+ done, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatal(err)
+ }
+ done()
+ o.evs.wait(t, GenHealthy)
+ b.procOf(b.started()[0].N).crash()
+ o.waitEntered(t)
+ if _, ok := a.Current(host); ok {
+ close(o.release)
+ t.Fatal("Current showed the first wake while its failure was still being delivered: firstWatch was cleared first")
+ }
+ second := make(chan error, 1)
+ go func() {
+ _, err := a.beginRequest(context.Background(), testRoute(host))
+ second <- err
+ }()
+ select {
+ case rec := <-saw:
+ close(o.release)
+ t.Fatalf("the next wake planned while the observer's failure was still being delivered (record in place: %v)", rec)
+ case <-time.After(300 * time.Millisecond):
+ }
+ close(o.release)
+ select {
+ case rec := <-saw:
+ if !rec {
+ t.Fatal("the next wake planned before the failure was recorded")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("the next wake never planned")
+ }
+ if err := <-second; err == nil || !strings.Contains(err.Error(), "held down") {
+ t.Fatalf("next wake = %v, want refused by the recorded failure", err)
+ }
+ if k := o.evs.kinds(); !slices.Equal(k, []GenEventKind{GenHealthy, GenFailed}) {
+ t.Fatalf("events %v, want healthy then failed", k)
+ }
+}
diff --git a/internal/activator/wakeplan.go b/internal/activator/wakeplan.go
index 7582b73e7..2b4f702ea 100644
--- a/internal/activator/wakeplan.go
+++ b/internal/activator/wakeplan.go
@@ -10,6 +10,7 @@ import (
"errors"
"fmt"
"log"
+ "sync"
"time"
"github.com/carlosframework/platform/internal/registry"
@@ -194,6 +195,15 @@ var (
// a failure would be the clock's, not the build's, and the good build
// boots instead with nothing recorded against B.
firstWakeMinVerdict = 20 * time.Second
+ // outcomeFenceWait bounds how long a new wake waits for a planned wake's
+ // outcome to be delivered before it plans (final review F3). OnGen's
+ // work is in memory plus one local failure-book write, so it normally
+ // returns in milliseconds; 5 s covers a slow disk with room to spare and
+ // still leaves the waiting requests most of their wakeWaitBound (30 s).
+ // Past it the wake proceeds as it did before the fence, so a wedged
+ // scheduler cannot keep a host down, and the plan's exec-time check
+ // still reads the record if it has landed by then.
+ outcomeFenceWait = 5 * time.Second
)
// genKiller (plan 3 Task 2, opset.go) is plan 2's generation kill
@@ -489,6 +499,75 @@ func plannedWorkerReady(ctx context.Context, p Proc) error {
return context.Canceled
}
+// outcomeFence counts a host's planned-wake outcomes that have been taken
+// but not yet delivered to OnGen (final review F3). done closes when the
+// count returns to zero.
+type outcomeFence struct {
+ n int
+ done chan struct{}
+}
+
+// holdOutcomeLocked takes one of host's outcome fences and returns its
+// release. The release takes a.mu itself, so it is called off a.mu, after
+// the outcome's OnGen has returned; only its first call counts. The caller
+// holds a.mu.
+func (a *Service) holdOutcomeLocked(host string) func() {
+ if a.outcomeFences == nil {
+ a.outcomeFences = make(map[string]*outcomeFence)
+ }
+ f := a.outcomeFences[host]
+ if f == nil {
+ f = &outcomeFence{done: make(chan struct{})}
+ a.outcomeFences[host] = f
+ }
+ f.n++
+ var once sync.Once
+ return func() {
+ once.Do(func() {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if f.n--; f.n == 0 {
+ close(f.done)
+ if a.outcomeFences[host] == f {
+ delete(a.outcomeFences, host)
+ }
+ }
+ })
+ }
+}
+
+// awaitOutcomes is a new wake's wait, before it plans, for host's
+// planned-wake outcomes still being delivered (final review F3; spec §8.1,
+// O7): a failed first wake's entry goes before afterWake's OnGen records
+// the failure, and a first wake's observer reaches its verdict before
+// OnGen hears it. Planning in that gap would pick the failed build again,
+// with no record yet to hold it. Bounded by outcomeFenceWait (and the
+// wake's own context): past it the wake plans as it did before the fence.
+// Called off a.mu.
+func (a *Service) awaitOutcomes(ctx context.Context, host string) {
+ a.mu.Lock()
+ f := a.outcomeFences[host]
+ a.mu.Unlock()
+ if f == nil {
+ return
+ }
+ bound := time.NewTimer(outcomeFenceWait)
+ defer bound.Stop()
+ select {
+ case <-f.done:
+ case <-ctx.Done():
+ case <-bound.C:
+ log.Printf("activator: %s: an earlier planned wake's outcome is still being delivered after %v; planning without waiting for it", host, outcomeFenceWait)
+ }
+}
+
+// outcomeOwedLocked reports whether wake w of e ends with a failure that
+// afterWake will deliver (GenFailed, GenFellBack): the outcome fence covers
+// it. The caller holds a.mu.
+func (a *Service) outcomeOwedLocked(e *actEntry, w *wokenInstance) bool {
+ return a.OnGen != nil && w != nil && w.failed != nil && e.plan.planned()
+}
+
// emitGen hands ev to OnGen. It is never called with a.mu held: OnGen
// reaches the scheduler's served, publishServing and writeGood, which take
// its goodMu and call back into Current.
@@ -539,14 +618,18 @@ func (a *Service) afterWake(e *actEntry, w *wokenInstance, err error) {
// generation from the instant it is reachable, and an exit or a stop
// before the observer goroutine runs is classified by it, never lost.
// wake() calls it after e.gen is set, and before the fence and bounce
-// checks there, so a stop they begin cancels the observation. The caller
-// holds a.mu.
+// checks there, so a stop they begin cancels the observation. It also takes
+// the host's outcome fence (final review F3), which watchFirstWake releases
+// once its terminal event has been delivered: the entry can be stopped and
+// a new wake begun before then, and that wake must not plan until the
+// verdict is recorded. The caller holds a.mu.
func (a *Service) armFirstWatchLocked(e *actEntry, w *wokenInstance) {
if a.Planner == nil || a.OnGen == nil || e.worker || !e.plan.planned() || !e.plan.FirstWake || w.fellBack || w.failed != nil || w.untried {
return
}
ctx, cancel := context.WithCancel(context.Background())
e.firstWatch, e.firstWatchCtx = cancel, ctx
+ e.firstWatchRelease = a.holdOutcomeLocked(e.host)
}
// watchFirstWake is a first wake's 10 s observation (spec §6.4), on plan 3's
@@ -575,22 +658,44 @@ func (a *Service) watchFirstWake(e *actEntry, g Gen, plan WakePlan) {
go func() {
r := observeSettle(ctx, proc, firstWakeSettle)
a.mu.Lock()
- cancel := e.firstWatch
- e.firstWatch, e.firstWatchCtx = nil, nil
- rt, unexpected := e.route, e.unexpectedExit
+ rt, unexpected, release := e.route, e.unexpectedExit, e.firstWatchRelease
a.mu.Unlock()
- if cancel != nil {
- cancel()
+ // end closes the observation: Current shows the generation again,
+ // and a stop no longer has an observation to cancel.
+ end := func() {
+ a.mu.Lock()
+ cancel := e.firstWatch
+ e.firstWatch, e.firstWatchCtx, e.firstWatchRelease = nil, nil, nil
+ a.mu.Unlock()
+ if cancel != nil {
+ cancel()
+ }
}
+ // The verdict is delivered before the host's outcome fence is
+ // released (final review F3): a new wake that began meanwhile plans
+ // only once the scheduler has recorded it.
+ defer func() {
+ if release != nil {
+ release()
+ }
+ }()
exited := r == settleExited || (r == settleCancelled && unexpected)
switch {
case r == settleOK:
+ // Ended first: the scheduler records a settled build as good
+ // only while Current shows it (rollout's acknowledged).
+ end()
a.emitGen(GenEvent{Kind: GenSettled, Route: rt, Gen: g, Mode: plan.Mode, FirstWake: true, At: time.Now()})
case exited:
+ // Delivered first, so nothing sees the failed generation as
+ // current (and admits an attempt onto it) before the failure is
+ // recorded.
a.emitGen(GenEvent{Kind: GenFailed, Route: rt, Failed: plan.Boot, Mode: plan.Mode, FirstWake: true,
Reason: "exited within " + firstWakeSettle.String() + " of becoming healthy", At: time.Now(), FailedUnder: plan.FailedUnder})
+ end()
default: // an intentional stop cancelled it
cancelled(rt)
+ end()
}
}()
}