amadan

carlos / platform Public

zdd-readiness: Task 11 fix round 1, Close joins planned wakes until their terminal event and coalesced worker attempts, a caller's cancellation is not a failed policy read

Each planned wake is counted from the intent beginWake writes until the OnGen that ends it (per token, given back exactly once), and Plan is counted for its length, so a wake whose terminal event reaches OnGen after svc.Shutdown returns has its End on disk when Close returns. CoalesceRecycle counts its admission before admitOne and skips the recycle once Close has begun. The policy cache no longer stamps failedAt when the caller's own context ended.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 4416872ecef85740985f43b0b2452af4cfa89999 parent f46d314
7 files changed, +417 −17
  • internal/rollout/manager.go +62 −12
  • internal/rollout/manager_test.go +202 −0
  • internal/rollout/policy.go +7 −2
  • internal/rollout/policy_test.go +43 −0
  • internal/rollout/wakeplan.go +17 −0
  • internal/rollout/worker.go +18 −3
  • internal/rollout/worker_test.go +68 −0
diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go
index 1bc74b5f6..44c9fa18f 100644
--- a/internal/rollout/manager.go
+++ b/internal/rollout/manager.go
@@ -173,13 +173,20 @@ type Manager struct {
goodMu sync.Mutex
// The shutdown join (Close, spec §8.6). joinMu is a leaf: nothing else
- // is taken while it is held. inflight counts the attempts' executions and
- // the OnGen calls under way; closed stops admission; joined, when set,
- // is closed once inflight is zero.
+ // is taken while it is held, so it may be taken under any other lock
+ // (CoalesceRecycle takes it under activator.Service.mu and a
+ // scheduler.mu). inflight counts the attempts' executions, the Plan and
+ // OnGen calls under way, and the planned wakes in wakes; closed stops
+ // admission; joined, when set, is closed once inflight is zero.
joinMu sync.Mutex
closed bool
inflight int
joined chan struct{}
+ // wakes holds the token of each planned wake whose intent beginWake
+ // wrote and whose terminal event has not yet reached OnGen (Task 11 fix
+ // round 1). Each is one count in inflight, taken and given back exactly
+ // once whatever the order of plans and events.
+ wakes map[uint64]struct{}
}
// NewManager builds the manager. Its schedulers run until ctx ends.
@@ -204,12 +211,20 @@ func NewManager(ctx context.Context, d Deps) *Manager {
// it after the activator's Shutdown, which has stopped every entry and so
// cancelled any swap in flight. It stops admitting attempts, then waits,
// until ctx ends, for every attempt's execution to finish its done() (the
-// outcome's book writes and the intent's End) and for every OnGen call
-// under way to return, and then writes the acks still pending and stops the
-// ack writer. A planned restart in the middle of a swap so leaves the
-// attempt's outcome on disk and no intent behind, and the next agent
-// reconciles nothing. A Close that runs out of ctx returns anyway: what was
-// not finished is what OpenFailureBook reconciles.
+// outcome's book writes and the intent's End), for every Plan and OnGen
+// call under way to return, and for every planned wake whose intent is on
+// disk to deliver the terminal event that ends it, and then writes the acks
+// still pending and stops the ack writer. A planned restart in the middle
+// of a swap or a wake so leaves the outcome on disk and no intent behind,
+// and the next agent reconciles nothing. A Close that runs out of ctx
+// returns anyway: what was not finished is what OpenFailureBook reconciles.
+//
+// The planned wakes are counted here, not left to Shutdown (Task 11 fix
+// round 1): Shutdown waits for each entry's e.ready, and wake() closes it
+// before afterWake sends the terminal event, while the first-wake
+// observer's cancellation is sent from its own goroutine. A wake already
+// past doWake's context check when Shutdown began can also still plan and
+// write its intent. Each of those reaches OnGen after Shutdown returns.
func (m *Manager) Close(ctx context.Context) {
m.joinMu.Lock()
m.closed = true
@@ -226,15 +241,15 @@ func (m *Manager) Close(ctx context.Context) {
select {
case <-wait:
case <-ctx.Done():
- m.d.Logf("rollout: stopping with %d attempt(s) or wake event(s) still running: %v", n, ctx.Err())
+ m.d.Logf("rollout: stopping with %d attempt(s), plan(s), wake(s) or wake event(s) still running: %v", n, ctx.Err())
}
}
m.acks.close(ctx)
}
// enter counts one operation Close joins. An admission (admit true) is
-// refused once Close has begun; an OnGen call is always counted, so one that
-// arrives while Close waits is waited for too.
+// refused once Close has begun; a Plan or OnGen call is always counted, so
+// one that arrives while Close waits is waited for too.
func (m *Manager) enter(admit bool) bool {
m.joinMu.Lock()
defer m.joinMu.Unlock()
@@ -249,6 +264,10 @@ func (m *Manager) enter(admit bool) bool {
func (m *Manager) leave() {
m.joinMu.Lock()
defer m.joinMu.Unlock()
+ m.leaveLocked()
+}
+
+func (m *Manager) leaveLocked() {
m.inflight--
if m.inflight == 0 && m.joined != nil {
close(m.joined)
@@ -256,6 +275,37 @@ func (m *Manager) leave() {
}
}
+// enterWake counts the planned wake with token tok, whose intent beginWake
+// has just written, until leaveWake (its terminal event). Always counted,
+// even once Close has begun: the intent is on disk, and only that event's
+// OnGen ends it. A token already counted is not counted again.
+func (m *Manager) enterWake(tok uint64) {
+ m.joinMu.Lock()
+ defer m.joinMu.Unlock()
+ if _, ok := m.wakes[tok]; ok {
+ return
+ }
+ if m.wakes == nil {
+ m.wakes = map[uint64]struct{}{}
+ }
+ m.wakes[tok] = struct{}{}
+ m.inflight++
+}
+
+// leaveWake gives back the count enterWake took for tok, once. Every
+// terminal event of a planned wake ends here; one whose wake wrote no
+// intent (a refused plan, a failed Begin, an ordinary wake), or one
+// delivered again, finds nothing and changes nothing.
+func (m *Manager) leaveWake(tok uint64) {
+ m.joinMu.Lock()
+ defer m.joinMu.Unlock()
+ if _, ok := m.wakes[tok]; !ok {
+ return
+ }
+ delete(m.wakes, tok)
+ m.leaveLocked()
+}
+
// closing reports whether Close has begun.
func (m *Manager) closing() bool {
m.joinMu.Lock()
diff --git a/internal/rollout/manager_test.go b/internal/rollout/manager_test.go
index d0337b8fa..8a569fdde 100644
--- a/internal/rollout/manager_test.go
+++ b/internal/rollout/manager_test.go
@@ -1281,3 +1281,205 @@ func TestClosingTheAckWriterWritesWhatIsPending(t *testing.T) {
t.Fatalf("written %v, want b2 and c3 written before close returned, and nothing after", got)
}
}
+
+// joinCount is Close's count of operations under way (test-only).
+func (m *Manager) joinCount() int {
+ m.joinMu.Lock()
+ defer m.joinMu.Unlock()
+ return m.inflight
+}
+
+// Task 11 fix round 1, review finding 1 (spec §8.6): a wake already past
+// doWake's context check when the activator's Shutdown begins can still plan
+// and write its intent. Its terminal event (afterWake's, here a refusal at
+// its lease: nothing ran) reaches OnGen only after svc.Shutdown has
+// returned, because Shutdown waits for e.ready and afterWake runs after it.
+// Close waits for that event, so the wake's End is on disk when Close
+// returns, and the next agent reconciles nothing.
+func TestCloseJoinsAPlannedWakeWhoseEventArrivesAfterShutdown(t *testing.T) {
+ h := newHarness(t, "a")
+ dir := h.onDisk()
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.wakeHost("a")
+ p := h.m.Plan(context.Background(), h.rows.get("a"))
+ if !p.FirstWake || p.Boot.Version != "v2" {
+ t.Fatalf("plan %+v: want a first wake of v2", p)
+ }
+ if _, ok := fileOf(t, dir).Intents["a"]; !ok {
+ t.Fatal("no intent on disk after the plan")
+ }
+ rt := h.rows.get("a")
+ go func() {
+ time.Sleep(200 * time.Millisecond) // afterWake, after Shutdown saw e.ready close
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: rt, Mode: p.Mode, FirstWake: true,
+ Token: p.Token, ConfigRev: p.ConfigRev})
+ }()
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ h.m.Close(ctx)
+ if f := fileOf(t, dir); len(f.Intents) != 0 {
+ t.Fatalf("intents %+v on disk after Close: a restart would reconcile the wake as a failure", f.Intents)
+ }
+}
+
+// Task 11 fix round 1, review finding 1: a planned wake is counted from the
+// intent its plan wrote until the one event that ends that intent, whatever
+// its kind; a plan that wrote no intent is never counted; and an event
+// delivered twice, or one for a wake that wrote nothing, never takes the
+// count below what is still under way.
+func TestAPlannedWakeIsCountedUntilItsIntentEnds(t *testing.T) {
+ ctx := context.Background()
+ setup := func(t *testing.T) (*harness, string, activator.WakePlan) {
+ h := newHarness(t, "a")
+ dir := h.onDisk()
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.wakeHost("a")
+ p := h.m.Plan(ctx, h.rows.get("a"))
+ if !p.FirstWake || p.Boot.Version != "v2" {
+ t.Fatalf("plan %+v: want a first wake of v2", p)
+ }
+ if n := h.m.joinCount(); n != 1 {
+ t.Fatalf("count %d after a plan that wrote its intent, want 1", n)
+ }
+ return h, dir, p
+ }
+ ev := func(p activator.WakePlan, h *harness, kind activator.GenEventKind, first, ran bool) activator.GenEvent {
+ e := activator.GenEvent{Kind: kind, Route: h.rows.get("a"), Gen: activator.Gen{N: 3, Build: "v2", Digest: "sha-v2"},
+ Mode: p.Mode, FirstWake: first, Token: p.Token, ConfigRev: p.ConfigRev, Executed: ran}
+ if kind == activator.GenFailed || kind == activator.GenFellBack {
+ e.Failed = p.Boot
+ }
+ if kind == activator.GenFellBack {
+ e.Gen = activator.Gen{N: 4, Build: "v1", Digest: "sha-v1"}
+ }
+ return e
+ }
+ type step struct {
+ kind activator.GenEventKind
+ first bool
+ ran bool
+ count int // the count once the event has returned
+ }
+ for _, tc := range []struct {
+ name string
+ steps []step
+ }{
+ {"settled", []step{{activator.GenHealthy, true, false, 1}, {activator.GenSettled, true, false, 0}}},
+ {"failed", []step{{activator.GenFailed, true, false, 0}}},
+ {"fell back", []step{{activator.GenFellBack, true, false, 0}}},
+ {"refused at its lease", []step{{activator.GenCancelled, true, false, 0}}},
+ {"stopped in its observation", []step{{activator.GenHealthy, true, false, 1}, {activator.GenCancelled, true, true, 0}}},
+ {"the good build booted untried", []step{{activator.GenHealthy, false, false, 0}}},
+ {"its terminal event twice", []step{{activator.GenCancelled, true, false, 0}, {activator.GenCancelled, true, false, 0}}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ h, dir, p := setup(t)
+ for i, s := range tc.steps {
+ h.m.OnGen(ev(p, h, s.kind, s.first, s.ran))
+ if n := h.m.joinCount(); n != s.count {
+ t.Fatalf("after event %d (%s): count %d, want %d", i, s.kind, n, s.count)
+ }
+ }
+ if _, stands := fileOf(t, dir).Intents["a"]; stands {
+ t.Fatal("the intent stands after its terminal event")
+ }
+ })
+ }
+ t.Run("a replaced wake and its replacement each end once", func(t *testing.T) {
+ h, _, old := setup(t)
+ cur := h.m.Plan(ctx, h.rows.get("a"))
+ if n := h.m.joinCount(); n != 2 {
+ t.Fatalf("count %d with two planned wakes, want 2", n)
+ }
+ h.m.OnGen(ev(old, h, activator.GenCancelled, true, false))
+ if n := h.m.joinCount(); n != 1 {
+ t.Fatalf("count %d after the replaced wake's event, want 1", n)
+ }
+ h.m.OnGen(ev(cur, h, activator.GenCancelled, true, false))
+ if n := h.m.joinCount(); n != 0 {
+ t.Fatalf("count %d after both events, want 0", n)
+ }
+ })
+ t.Run("a plan whose intent could not be written", func(t *testing.T) {
+ h := newHarness(t, "a")
+ dir := h.onDisk()
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.wakeHost("a")
+ readOnly(t, dir)
+ p := h.m.Plan(ctx, h.rows.get("a"))
+ if p.Refuse != BookUnreadableRefusal {
+ t.Fatalf("plan %+v: want the book refusal", p)
+ }
+ if n := h.m.joinCount(); n != 0 {
+ t.Fatalf("count %d after a plan that wrote nothing, want 0", n)
+ }
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), FirstWake: true, Token: p.Token})
+ if n := h.m.joinCount(); n != 0 {
+ t.Fatalf("count %d after an event for a wake that wrote nothing, want 0", n)
+ }
+ })
+ t.Run("a plan refused while the attempt's intent stands", func(t *testing.T) {
+ h := newHarness(t, "a")
+ h.onDisk()
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.wakeHost("a")
+ if err := h.book.Begin("a", Intent{Account: "bab", App: "blog", Build: "v3", Mode: "hold", ID: "a1"}); err != nil {
+ t.Fatal(err)
+ }
+ if p := h.m.Plan(ctx, h.rows.get("a")); p.Refuse == "" {
+ t.Fatalf("plan %+v: want a refusal", p)
+ }
+ if n := h.m.joinCount(); n != 0 {
+ t.Fatalf("count %d after a refused plan, want 0", n)
+ }
+ })
+ t.Run("a plan under way", func(t *testing.T) {
+ h := newHarness(t, "a")
+ h.onDisk()
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.wakeHost("a")
+ entered, release := make(chan struct{}), make(chan struct{})
+ var once sync.Once
+ h.mu.Lock()
+ h.demandHook = func(ver string) {
+ if ver == "v2" {
+ once.Do(func() { close(entered); <-release })
+ }
+ }
+ h.mu.Unlock()
+ planned := make(chan activator.WakePlan, 1)
+ go func() { planned <- h.m.Plan(ctx, h.rows.get("a")) }()
+ <-entered
+ if n := h.m.joinCount(); n != 1 {
+ t.Fatalf("count %d while a plan reads the demand, want 1", n)
+ }
+ closed := make(chan struct{})
+ go func() {
+ cctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+ defer cancel()
+ h.m.Close(cctx)
+ close(closed)
+ }()
+ select {
+ case <-closed:
+ t.Fatal("Close returned while a plan was under way")
+ case <-time.After(100 * time.Millisecond):
+ }
+ close(release)
+ p := <-planned
+ if n := h.m.joinCount(); n != 1 || !p.FirstWake {
+ t.Fatalf("count %d once the plan (%+v) returned, want 1: its wake's", n, p)
+ }
+ h.m.OnGen(ev(p, h, activator.GenCancelled, true, false))
+ select {
+ case <-closed:
+ case <-time.After(5 * time.Second):
+ t.Fatal("Close did not return once the wake ended")
+ }
+ })
+}
diff --git a/internal/rollout/policy.go b/internal/rollout/policy.go
index 2d86cb2b7..924d9a798 100644
--- a/internal/rollout/policy.go
+++ b/internal/rollout/policy.go
@@ -131,9 +131,14 @@ func (c *policyCache) get(ctx context.Context, acct, app string) AppPolicy {
defer c.mu.Unlock()
if fill > e.stored {
// No newer read has stored since this one began.
- if err != nil {
+ switch {
+ case err != nil && ctx.Err() != nil:
+ // The caller's own context ended (a wake or an attempt that
+ // stopped), not the read bound: that says nothing about the
+ // store, so it holds off no other caller (Task 11 fix round 1).
+ case err != nil:
e.failedAt = c.now()
- } else {
+ default:
e.p, e.ok, e.at, e.failedAt, e.stored = PolicyFromManifest(m), true, c.now(), time.Time{}, fill
}
}
diff --git a/internal/rollout/policy_test.go b/internal/rollout/policy_test.go
index 570fbf31b..080087a70 100644
--- a/internal/rollout/policy_test.go
+++ b/internal/rollout/policy_test.go
@@ -207,3 +207,46 @@ func TestManagedIsTheActivatorsExecRoutes(t *testing.T) {
}
}
}
+
+// Task 11 fix round 1, review finding 3: a read the caller abandoned (its
+// own context ended, a wake or an attempt that stopped) says nothing about
+// the store, so it is not a failed read and does not hold off the next
+// caller for policyRetryAfter. The read bound's own deadline, with the
+// caller still waiting, is a failed read.
+func TestPolicyCacheACallersCancellationIsNotAFailedRead(t *testing.T) {
+ now := time.Unix(1_700_000_000, 0)
+ calls := 0
+ var fail error
+ c := &policyCache{
+ load: func(ctx context.Context, _, _ string) (*appconfig.Manifest, error) {
+ calls++
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ if fail != nil {
+ return nil, fail
+ }
+ return &appconfig.Manifest{Schema: 1, App: "blog", Swap: appconfig.SwapOverlap, SwapConcurrency: 2}, nil
+ },
+ ttl: 30 * time.Second,
+ now: func() time.Time { return now },
+ m: map[appKey]*policyEntry{},
+ }
+ hold := AppPolicy{Swap: activator.ModeHold, Concurrency: 1}
+ gone, cancel := context.WithCancel(context.Background())
+ cancel()
+ if got := c.get(gone, "bab", "blog"); got != hold {
+ t.Fatalf("a cancelled caller's read = %+v, want hold", got)
+ }
+ want := AppPolicy{Swap: activator.ModeOverlap, Concurrency: 2}
+ if got := c.get(context.Background(), "bab", "blog"); got != want || calls != 2 {
+ t.Fatalf("the next caller's read = %+v after %d loads, want %+v from a second load: a cancelled caller held it off", got, calls, want)
+ }
+ now = now.Add(31 * time.Second)
+ fail = context.DeadlineExceeded // the read bound ran out, the caller still waiting
+ c.get(context.Background(), "bab", "blog")
+ now = now.Add(time.Second)
+ if got := c.get(context.Background(), "bab", "blog"); got != want || calls != 3 {
+ t.Fatalf("read inside policyRetryAfter of a timed-out read = %+v after %d loads, want the last value and 3 loads", got, calls)
+ }
+}
diff --git a/internal/rollout/wakeplan.go b/internal/rollout/wakeplan.go
index 06675ba58..422256910 100644
--- a/internal/rollout/wakeplan.go
+++ b/internal/rollout/wakeplan.go
@@ -211,6 +211,12 @@ func (m *Manager) Plan(ctx context.Context, rt registry.Route) (plan activator.W
if !Managed(rt) {
return activator.WakePlan{}
}
+ // Joined by Close for its whole length (spec §8.6, Task 11 fix round 1):
+ // a plan under way may still write its wake's intent. Deferred first, so
+ // it is given back last, after beginWake below has counted the wake
+ // whose intent it wrote: the count never touches zero in between.
+ m.enter(false)
+ defer m.leave()
tok, planAt := m.newWakeToken(rt.Host)
defer func() {
plan.Token = tok
@@ -305,6 +311,10 @@ func (m *Manager) beginWake(rt registry.Route, plan activator.WakePlan, tok uint
ID: wakeExecID(tok), Start: planAt}
err := m.d.Book.Begin(rt.Host, in)
if err == nil {
+ // The intent is on disk: Close now waits for the terminal event
+ // that ends it (OnGen, leaveWake). A refused plan, and a Begin that
+ // failed, wrote nothing and is never counted.
+ m.enterWake(tok)
return plan
}
m.d.Logf("rollout: %s: write the intent of its first wake of %s: %v; refusing the wake", rt.Host, plan.Boot.Version, err)
@@ -535,6 +545,13 @@ func (m *Manager) OnGen(ev activator.GenEvent) {
if err := m.d.Book.End(rt.Host, wakeExecID(ev.Token)); err != nil {
m.d.Logf("rollout: %s: end the intent of its wake: %v (it stays on disk until a save lands)", rt.Host, err)
}
+ // The wake's count in Close's join goes with the End (beginWake
+ // took it). This call is itself counted until it returns, so the
+ // join cannot see zero before the End above is written. A wake
+ // replaced by a newer plan has its intent replaced on disk, but its
+ // own terminal event still ends its count here: the activator sends
+ // exactly one per planned wake (afterWake, cancelPlanned).
+ m.leaveWake(ev.Token)
}
}
diff --git a/internal/rollout/worker.go b/internal/rollout/worker.go
index eb06786ec..72ec50f29 100644
--- a/internal/rollout/worker.go
+++ b/internal/rollout/worker.go
@@ -64,7 +64,8 @@ func (q *appQueue) admitOne(host string, now time.Time) (dispatch, bool) {
// order, and the attempt it admits runs on its own goroutine. True means an
// attempt makes the spawn and the recycle is skipped: one admitted now, or
// one already running for the host (ReplaceWorker respawns that child
-// itself, and its observation reads a crash as the build's).
+// itself, and its observation reads a crash as the build's). Once Close has
+// begun it is also true, with nothing admitted.
func (m *Manager) CoalesceRecycle(host string) bool {
s := m.schedulerOf(host)
if s == nil {
@@ -75,11 +76,25 @@ func (m *Manager) CoalesceRecycle(host string) bool {
s.mu.Unlock()
return true
}
+ // Counted before it is admitted, as the scheduler loop counts its
+ // admissions, so Close joins its done() (spec §8.6, Task 11 fix round
+ // 1). joinMu is a leaf, so it may be taken here under activator.Service.mu
+ // and s.mu. Once Close has begun nothing is admitted and the recycle is
+ // skipped: the agent is stopping, and the child goes with it.
+ if !m.enter(true) {
+ s.mu.Unlock()
+ return true
+ }
d, ok := s.q.admitOne(host, m.d.Now())
- s.mu.Unlock()
if !ok {
+ m.leave()
+ s.mu.Unlock()
return false
}
- go s.execute(m.ctx, d)
+ s.mu.Unlock()
+ go func() {
+ defer m.leave()
+ s.execute(m.ctx, d)
+ }()
return true
}
diff --git a/internal/rollout/worker_test.go b/internal/rollout/worker_test.go
index 07848a96a..e0b74372e 100644
--- a/internal/rollout/worker_test.go
+++ b/internal/rollout/worker_test.go
@@ -1,6 +1,7 @@
package rollout
import (
+ "context"
"testing"
"time"
@@ -190,3 +191,70 @@ func TestCoalesceRecycleNeverCallsTheActivator(t *testing.T) {
}
h.act.mu.Unlock()
}
+
+// Task 11 fix round 1, review finding 2 (spec §8.6): an adoption a recycle
+// coalesces with runs on its own goroutine, and Close joins it as it joins
+// one the scheduler loop admits: its outcome and its End are on disk when
+// Close returns. Once Close has begun a recycle admits nothing; it is
+// skipped, which is all a stopping agent needs.
+func TestCloseJoinsACoalescedWorkerAttempt(t *testing.T) {
+ h := workerHarness(t, "bot")
+ dir := h.onDisk()
+ h.stage("v2", "v3")
+ h.pass()
+ waitFor(t, "the host's scheduler to see it serve v1", func() bool {
+ s := h.m.schedulerOf("bot")
+ if s == nil {
+ return false
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ hs := s.q.hosts["bot"]
+ return hs != nil && hs.serving == "v1"
+ })
+ running, proceed := make(chan struct{}), make(chan struct{})
+ h.act.mu.Lock()
+ h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult {
+ close(running)
+ <-proceed
+ return activator.SwapResult{Outcome: activator.SwapDone}
+ }
+ h.act.mu.Unlock()
+ queue := func(build string) {
+ h.install(build, "bot")
+ s := h.m.schedulerOf("bot")
+ s.mu.Lock()
+ s.q.wantBuild("bot", activator.CauseRelease, build, h.clock()) // queued, not yet admitted by the loop
+ s.mu.Unlock()
+ }
+ queue("v2")
+ if !h.m.CoalesceRecycle("bot") {
+ t.Fatal("the recycle did not coalesce with the admissible release")
+ }
+ select {
+ case <-running:
+ case <-time.After(5 * time.Second):
+ t.Fatal("the coalesced attempt never started")
+ }
+ if n := len(fileOf(t, dir).Intents); n != 1 {
+ t.Fatalf("%d intents on disk while the coalesced attempt runs, want 1", n)
+ }
+ go func() { time.Sleep(200 * time.Millisecond); close(proceed) }()
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ h.m.Close(ctx)
+ if f := fileOf(t, dir); len(f.Intents) != 0 {
+ t.Fatalf("intents %+v on disk after Close: a restart would reconcile the coalesced attempt as a failure", f.Intents)
+ }
+ if g := h.rows.get("bot").GoodVersion; g != "v2" {
+ t.Fatalf("good build %q after Close, want v2: the coalesced attempt's outcome", g)
+ }
+ queue("v3")
+ if !h.m.CoalesceRecycle("bot") {
+ t.Fatal("a recycle after Close went ahead: want it skipped")
+ }
+ time.Sleep(100 * time.Millisecond)
+ if r := h.act.replacements(); len(r) != 1 {
+ t.Fatalf("replacements %+v: an attempt was admitted by a recycle after Close", r)
+ }
+}