package rollout

import (
	"context"
	"errors"
	"fmt"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rolloutwords"
	"github.com/carlosframework/platform/internal/status"
)

func TestAReleaseRollsOutCanaryFirstThenPaced(t *testing.T) {
	h := newHarness(t, "a", "b", "c")
	h.stage("v2")
	h.install("v2", "a", "b", "c")
	h.pass()
	waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b", "c") })
	calls, _, maxFlight := h.act.snapshot()
	if len(calls) != 3 || !calls[0].a.Canary || calls[1].a.Canary {
		t.Fatalf("calls = %+v, want the canary first and then two plain swaps", calls)
	}
	if maxFlight > 1 {
		t.Fatalf("%d swaps ran at once with swap_concurrency 1", maxFlight)
	}
	for _, host := range []string{"a", "b", "c"} {
		if r := h.rows.get(host); r.GoodVersion != "v2" || r.GoodDigest != "sha-v2" || r.ServingVersion != "v2" {
			t.Fatalf("%s row = %+v, want good and serving v2", host, r)
		}
	}
	if !h.events.has(textHead(textSwapped("v2", time.Hour), time.Hour.String())) {
		t.Fatal("no swapped event")
	}
}

// Review focus 1, the index's cross-plan item.
func TestTwoReleasesTwentySecondsApartNeverSwitchToTheFirst(t *testing.T) {
	h := newHarness(t, "a", "b", "c")
	h.stage("v2", "v3")
	gate := h.act.hold("a") // the canary onto v2 sits in its settle
	h.install("v2", "a", "b", "c")
	h.pass()
	waitFor(t, "the v2 canary to start", func() bool { _, started, _ := h.act.snapshot(); return len(started) == 1 })
	h.advance(20 * time.Second)
	h.install("v3", "a", "b", "c")
	h.pass()
	close(gate)
	waitFor(t, "every host on v3", func() bool { return h.servingAll("v3", "a", "b", "c") })
	calls, _, maxFlight := h.act.snapshot()
	for _, c := range calls {
		if c.a.Build == "v2" && c.outcome == activator.SwapDone {
			t.Fatalf("switched %s to the superseded v2", c.a.Host)
		}
	}
	if calls[0].a.Build != "v2" || calls[0].outcome != activator.SwapAbandoned {
		t.Fatalf("first call = %+v, want the v2 canary abandoned at its re-check", calls[0])
	}
	if len(h.book.List()) != 0 {
		t.Fatal("an abandoned attempt is not a build failure")
	}
	if maxFlight > 1 {
		t.Fatal("swap_concurrency 1 exceeded")
	}
}

func TestAFailedCanaryHaltsTheCohort(t *testing.T) {
	h := newHarness(t, "a", "b", "c")
	h.stage("v2")
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Build == "v2" {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.install("v2", "a", "b", "c")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
		t.Fatalf("%d attempts onto v2, want the halt at the first failure", len(calls))
	}
	if r := h.rows.get("a"); r.GoodVersion != "v1" {
		t.Fatalf("good build moved to %q on a failure", r.GoodVersion)
	}
	if !h.events.has(textHead(textCohortHalted("v2", "\x00", status.ReasonHealth), "\x00")) {
		t.Fatal("no cohort-halted event")
	}
}

// Spec §8.1: the failure transition (the record and the halt) happens in
// the same step that frees the failed attempt's slot. The hook looks at the
// queue at every moment done() reads a row after the failure: whenever the
// failed attempt's slot is free, the record must already be in the book
// (which wake plans read) and admission must start nothing onto the build.
// Halting in that step but writing the record after the row read let a
// wake select the failed build in between.
func TestAFailureIsRecordedAndHaltedInTheStepThatFreesItsSlot(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.stage("v2")
	var failed atomic.Bool
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Build != "v2" {
			return activator.SwapDone
		}
		failed.Store(true)
		return activator.SwapFailed
	}
	var bad atomic.Int32
	h.rows.mu.Lock()
	h.rows.getHook = func(host string) {
		if !failed.Load() {
			return
		}
		s := h.m.schedulerOf(host)
		s.mu.Lock()
		hs := s.q.hosts[host]
		free := hs != nil && hs.running == nil
		var ds []dispatch
		if free {
			ds = s.q.admit(currentView(s.q), h.clock())
		}
		s.mu.Unlock()
		if !free {
			return
		}
		if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
			bad.Add(1)
		}
		for _, d := range ds {
			if d.ID.Build == "v2" {
				bad.Add(1)
			}
		}
	}
	h.rows.mu.Unlock()
	h.install("v2", "a", "b")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	h.pass() // reads the rows again, with the slot free
	if n := bad.Load(); n != 0 {
		t.Fatalf("%d moment(s) with the slot free and the record missing or a v2 attempt admissible", n)
	}
}

// Spec §8.5: a successful attempt keeps its host's slot until its state
// (serving and good build) is published, so no newer attempt can be
// admitted and complete first and then be overwritten by this one. Freeing
// the slot before the write let that happen.
func TestASuccessKeepsItsSlotUntilItIsPublished(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	var freed atomic.Int32
	h.rows.mu.Lock()
	h.rows.setGoodHook = func(host string) {
		s := h.m.schedulerOf(host)
		s.mu.Lock()
		if hs := s.q.hosts[host]; hs != nil && hs.running == nil {
			freed.Add(1)
		}
		s.mu.Unlock()
	}
	h.rows.mu.Unlock()
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the good build moved to v2", func() bool { return h.rows.get("a").GoodVersion == "v2" })
	if n := freed.Load(); n != 0 {
		t.Fatalf("%d good-build write(s) made after the attempt's slot was freed", n)
	}
}

// A failure is published with no blocking read between the failure and the
// record (spec §8.1): the pointer the build failed under is read before the
// attempt runs. Reading it in done(), before the record, left every sibling's
// wake free to plan the failed build for as long as that read took.
func TestNoPointerReadSitsBetweenAFailureAndItsRecord(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	var failed, late atomic.Bool
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
		failed.Store(true)
		return activator.SwapFailed
	}
	h.mu.Lock()
	h.pointerHook = func() {
		if failed.Load() {
			if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
				late.Store(true)
			}
		}
	}
	h.mu.Unlock()
	h.install("v2", "a")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	if late.Load() {
		t.Fatal("a pointer read ran after the failure and before its record was published")
	}
}

// Spec §6.3: a build installed while the entry is unstable (a hold
// draining, a wake running) still supersedes the attempt in flight, so its
// gates refuse the older build. Skipping wantBuild for such a host left the
// revision on the older build, and a hold's pinned wake exec'd it. Nothing
// is admitted onto the host until it is stable again.
func TestAnInstallDuringAnUnstableEntryStillSupersedes(t *testing.T) {
	h := newHarness(t, "a")
	h.pass()
	s := h.m.schedulerOf("a")
	h.act.mu.Lock()
	delete(h.act.gens, "a")
	h.act.owned["a"] = true // stopping or waking: a process, no current generation
	h.act.mu.Unlock()
	s.mu.Lock()
	before := s.q.newest("a")
	s.mu.Unlock()
	h.stage("v2")
	h.install("v2", "a")
	h.pass()
	s.mu.Lock()
	after := s.q.newest("a")
	s.mu.Unlock()
	if after == before {
		t.Fatal("an install during an unstable entry left the attempt revision unchanged")
	}
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
		t.Fatalf("attempts %+v admitted onto an unstable entry", calls)
	}
}

// A row that cannot be read at the failure still gets its record: the
// failure is the build's, not the row's (spec §8.1). Skipping the write on
// a failed registry read left the failed build selectable by every wake.
func TestAFailureIsRecordedWhenItsRowIsUnreadable(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	var failed atomic.Bool
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
		failed.Store(true)
		return activator.SwapFailed
	}
	h.rows.mu.Lock()
	h.rows.getErr = func(string) error {
		if failed.Load() {
			return errors.New("registry: database is locked")
		}
		return nil
	}
	h.rows.mu.Unlock()
	h.install("v2", "a")
	h.pass()
	waitFor(t, "a failure record despite the unreadable row", func() bool {
		_, ok := h.book.Lookup("bab", "blog", "v2")
		return ok
	})
}

// Review focus 4.
func TestStagingRefusalWritesNoFailureRecord(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.mismatch["v2"] = true
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the stage relay touch", func() bool { return h.touched.Load() > 0 })
	if len(h.book.List()) != 0 {
		t.Fatal("a staging error wrote a failure record against the build")
	}
	if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
		t.Fatal("a refused attempt reached Swap")
	}
	if g, _ := h.act.Current("a"); g.Build != "v1" {
		t.Fatal("the instance left its serving build")
	}
	if !h.events.has(textStagingRefused("v2")) {
		t.Fatal("no staging event")
	}
}

func TestAnUnstagedBuildIsAStagingRefusal(t *testing.T) {
	h := newHarness(t, "a")
	h.install("v9", "a") // never staged on this box
	h.pass()
	waitFor(t, "the stage relay touch", func() bool { return h.touched.Load() > 0 })
	if len(h.book.List()) != 0 {
		t.Fatal("a missing staged file wrote a failure record")
	}
}

func TestRestartGoesAheadForAnInstanceServingAFailedBuild(t *testing.T) {
	h := newHarness(t, "d")
	h.stage("v2")
	h.act.serve("d", "v2")
	r := h.rows.get("d")
	r.Version, r.GoodVersion, r.GoodDigest = "v2", "v2", "sha-v2"
	h.rows.put(r)
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "e", At: t0})
	h.pass()
	if !h.m.Restart("d", "gen:etag-1") {
		t.Fatal("the restart was not queued")
	}
	waitFor(t, "the restart", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
	calls, _, _ := h.act.snapshot()
	if c := calls[0].a; c.Cause != activator.CauseRestart || c.Build != "v2" || c.RestartID != "gen:etag-1" {
		t.Fatalf("attempt = %+v", c)
	}
}

func TestRecoveryUsesTheRetainedIdentityAndTheCurrentPolicy(t *testing.T) {
	h := newHarness(t, "a")
	h.swap = "hold" // switched to hold after v1 passed under overlap
	h.pass()
	h.m.Restart("a", "gen:etag-1")
	waitFor(t, "the restart", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
	calls, _, _ := h.act.snapshot()
	g, _ := h.act.Current("a")
	if c := calls[0].a; c.Mode != activator.ModeHold || c.StagedPath != "/opt/v1" || c.Digest != "sha-v1" {
		t.Fatalf("attempt = %+v, want hold, by the generation's captured path and digest (current gen %+v)", c, g)
	}
}

func TestAReleaseOntoTheServingBuildMakesNoSwap(t *testing.T) {
	h := newHarness(t, "a")
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
		t.Fatalf("calls = %+v", calls)
	}
}

// Review focus 5.
func TestConfigDeliveredMidSwapIsAckedOnlyWhenActive(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	gate := h.act.hold("a")
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the release swap to start", func() bool { _, s, _ := h.act.snapshot(); return len(s) == 1 })
	// The ack is written from the scheduler's goroutine: counted atomically.
	var acks atomic.Int32
	ack := h.m.DeferAck(func(context.Context, appconfig.ConfigAck) { acks.Add(1) })
	h.m.ConfigDelivered("a", "sha256:c1")
	ack(appconfig.ConfigAck{Host: "a"})
	if acks.Load() != 0 {
		t.Fatal("acknowledged a config no generation has started with")
	}
	close(gate)
	waitFor(t, "the config attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 2 })
	calls, _, _ := h.act.snapshot()
	if c := calls[1].a; c.Cause != activator.CauseConfig || c.ConfigRev != "sha256:c1" || c.Build != "v2" {
		t.Fatalf("second attempt = %+v, want a config attempt onto v2", c)
	}
	waitFor(t, "the ack", func() bool { return acks.Load() == 1 })
}

// Spec §8.1: a successful config attempt's activation is published before
// its slot frees, and with no registry read in front of it. Published after
// the slot and the read, a newer config attempt could complete first and be
// overwritten (active went back to the older revision), and an unreadable
// row lost the activation, leaving delivered and active apart for good. The
// held ack is released (handed to the writer, its order fixed) in the same
// step; the write itself is the writer's, after (Task 11).
func TestConfigActivationIsPublishedBeforeTheSlotFrees(t *testing.T) {
	h := newHarness(t, "a")
	h.pass()
	var acks, released, freed atomic.Int32
	h.m.acks.onSubmit = func(host string) {
		released.Add(1)
		s := h.m.schedulerOf("a")
		s.mu.Lock()
		if hs := s.q.hosts["a"]; hs != nil && hs.running == nil {
			freed.Add(1)
		}
		s.mu.Unlock()
	}
	ack := h.m.DeferAck(func(context.Context, appconfig.ConfigAck) { acks.Add(1) })
	var swapped atomic.Bool
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
		swapped.Store(true)
		return activator.SwapDone
	}
	h.rows.mu.Lock()
	h.rows.getErr = func(string) error {
		if swapped.Load() {
			return errors.New("registry: database is locked")
		}
		return nil
	}
	h.rows.mu.Unlock()
	// The config attempt waits at the gate until the ack is held, so the ack
	// is released by the activation and not written at once by a config sync
	// call that came after it.
	gate := h.act.hold("a")
	h.m.ConfigDelivered("a", "sha256:c1")
	ack(appconfig.ConfigAck{Host: "a"})
	if acks.Load() != 0 {
		t.Fatal("acknowledged a config no generation has started with")
	}
	close(gate)
	waitFor(t, "the config activated and acknowledged", func() bool { return acks.Load() == 1 })
	if released.Load() != 1 {
		t.Fatalf("%d acks released, want 1", released.Load())
	}
	if n := freed.Load(); n != 0 {
		t.Fatal("the activation was published after the attempt's slot was freed")
	}
}

func TestAFailedConfigAttemptRetriesWithBackoff(t *testing.T) {
	h := newHarness(t, "a")
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Cause == activator.CauseConfig {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.pass()
	h.m.ConfigDelivered("a", "sha256:c1")
	waitFor(t, "the first config attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
	waitFor(t, "the pending event", func() bool { return h.events.has(configPendingHead) })
	h.advance(59 * time.Second)
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
		t.Fatal("retried inside the one-minute backoff")
	}
	h.advance(2 * time.Second)
	h.pass()
	waitFor(t, "the retry after the backoff", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 2 })
	if len(h.book.List()) != 0 {
		t.Fatal("a failed config attempt wrote a failure record")
	}
}

func TestBootstrapRecordsAwakeInstancesOnly(t *testing.T) {
	h := newHarness(t, "awake", "asleep")
	for _, host := range []string{"awake", "asleep"} {
		r := h.rows.get(host)
		r.GoodVersion, r.GoodDigest = "", ""
		h.rows.put(r)
	}
	h.act.mu.Lock()
	delete(h.act.gens, "asleep")
	h.act.mu.Unlock()
	rs, _ := h.rows.List()
	h.m.Bootstrap(rs)
	if r := h.rows.get("awake"); r.Good() != (registry.GoodBuild{Version: "v1", Digest: "sha-v1"}) {
		t.Fatalf("awake row good = %+v", r.Good())
	}
	if r := h.rows.get("asleep"); r.GoodVersion != "" {
		t.Fatal("nothing may be inferred for a sleeping instance (spec §8.5)")
	}
}

// `carlos promote --retry` re-signs the channel pointer with a fresh
// PromotedAt. The record keeps the PromotedAt it failed under, survives that
// same pointer being read again, and clears on the re-signed one.
func TestAReSignedPointerClearsTheRecord(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.stage("v2")
	h.failing.Store(true)
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Build == "v2" && h.failing.Load() {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.pointer("stable", "v2", promotedAt)
	h.install("v2", "a", "b")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	if r, _ := h.book.Lookup("bab", "blog", "v2"); !r.PromotedAt.Equal(promotedAt) {
		t.Fatalf("record keeps PromotedAt %v, want the pointer's %v", r.PromotedAt, promotedAt)
	}
	h.advance(pointerReadEvery)
	h.pass()
	if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
		t.Fatal("the pointer the build failed under cleared its record")
	}
	h.failing.Store(false)
	h.pointer("stable", "v2", promotedAt.Add(time.Minute))
	h.advance(pointerReadEvery)
	h.pass()
	waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b") })
	calls, _, _ := h.act.snapshot()
	if c := calls[1].a; c.Cause != activator.CauseRetry || !c.Canary {
		t.Fatalf("first attempt after the retry = %+v, want a retry canary", c)
	}
}

// `carlos instances retry <host>`: that host alone, as a canary.
func TestRetryHostClearsOneInstance(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.stage("v2")
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "c", PromotedAt: promotedAt, At: t0})
	h.install("v2", "a", "b")
	h.pass()
	if !h.retry("b", "v2") {
		t.Fatal("RetryHost did not clear b")
	}
	waitFor(t, "b on v2", func() bool { return h.servingAll("v2", "b") })
	if h.servingAll("v2", "a") {
		t.Fatal("a host retry moved another host")
	}
}

// Carried from Task 3: a canonical manifest that cannot be read, or a store
// that cannot compare digests, must reach AttemptMode as a hold demand, never
// as "". Only release.ErrStagedDigestMismatch is a staging refusal.
func TestAnUnreadableCanonicalManifestRunsInHold(t *testing.T) {
	for name, set := range map[string]func(h *harness){
		"demand error": func(h *harness) { h.demandErr["v2"] = errors.New("store: GetObject: connection reset") },
		"no digest":    func(h *harness) { h.noDigest["v2"] = true },
	} {
		t.Run(name, func(t *testing.T) {
			h := newHarness(t, "a")
			h.stage("v2")
			set(h)
			h.install("v2", "a")
			h.pass()
			waitFor(t, "the attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
			want, err := release.FileSHA256(release.VersionPath(h.opt, "bab", "v2", "blog"))
			if err != nil {
				t.Fatal(err)
			}
			calls, _, _ := h.act.snapshot()
			if c := calls[0].a; c.Mode != activator.ModeHold || c.Demand != release.DemandHold || c.Digest != want {
				t.Fatalf("attempt = %+v, want hold on the staged file's own digest %s", c, want)
			}
			if h.touched.Load() != 0 || len(h.book.List()) != 0 {
				t.Fatal("an unreadable manifest is neither a staging refusal nor a build failure")
			}
		})
	}
}

// Carried from Task 4: a record keeps a real PromotedAt. With no pointer to
// read, the failure's own time stands in, never a zero time.
func TestAFailureWithNoReadablePointerKeepsItsOwnTime(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapFailed }
	h.install("v2", "a")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	r, _ := h.book.Lookup("bab", "blog", "v2")
	if r.PromotedAt.IsZero() || !r.PromotedAt.Equal(r.At) {
		t.Fatalf("record %+v: want the failure's own time as PromotedAt", r)
	}
}

// Carried from Task 5: a second `carlos promote --retry` of the same build,
// after the retry failed too, clears the record again and the next attempt is
// a canary again. Nothing may rest on wantBuild reporting a change.
func TestASecondRetryOfTheSameBuildIsACanaryAgain(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.stage("v2")
	h.failing.Store(true)
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Build == "v2" && h.failing.Load() {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.pointer("stable", "v2", promotedAt)
	h.install("v2", "a", "b")
	h.pass()
	waitFor(t, "the first failure", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })

	h.pointer("stable", "v2", promotedAt.Add(time.Minute))
	h.advance(pointerReadEvery)
	h.pass()
	waitFor(t, "the retry to fail too", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 2 })
	waitFor(t, "the record again", func() bool {
		r, ok := h.book.Lookup("bab", "blog", "v2")
		return ok && r.PromotedAt.Equal(promotedAt.Add(time.Minute))
	})

	h.failing.Store(false)
	h.pointer("stable", "v2", promotedAt.Add(2*time.Minute))
	h.advance(pointerReadEvery)
	h.pass()
	waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b") })
	calls, _, _ := h.act.snapshot()
	for i, c := range calls[:3] {
		if c.a.Host != "a" || !c.a.Canary {
			t.Fatalf("call %d = %+v, want a canary on a", i, c.a)
		}
	}
	if c := calls[2].a; c.Cause != activator.CauseRetry {
		t.Fatalf("second retry's attempt = %+v, want cause retry", c)
	}
}

// Carried from Task 6 and plan 3 (O7): a hold attempt that fails is held
// down in the book before its slot frees, so no wake planned in between can
// boot the failed hold build. The mode is the one Swap ran (a foreign-layout
// overlap runs as a hold, SwapResult.Mode), and the record's build comes from
// the dispatch: a failed hold returns no generation.
func TestAFailedHoldIsHeldDownBeforeItsSlotFrees(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.act.mode = activator.ModeHold
	var failed atomic.Bool
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
		failed.Store(true)
		return activator.SwapDown
	}
	var bad, seen atomic.Int32
	h.rows.mu.Lock()
	h.rows.getHook = func(host string) {
		if !failed.Load() {
			return
		}
		s := h.m.schedulerOf(host)
		s.mu.Lock()
		free := s.q.hosts[host] != nil && s.q.hosts[host].running == nil
		s.mu.Unlock()
		if !free {
			return
		}
		seen.Add(1)
		if !h.book.HeldDown("bab", "blog", "v2", host) {
			bad.Add(1)
		}
	}
	h.rows.mu.Unlock()
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the held-down record", func() bool { return h.book.HeldDown("bab", "blog", "v2", "a") })
	waitFor(t, "the down event", func() bool { return h.events.has(textSwapDown("v2")) })
	if bad.Load() != 0 || seen.Load() == 0 {
		t.Fatalf("%d of %d look(s) with the slot free found the host not held down", bad.Load(), seen.Load())
	}
	r, _ := h.book.Lookup("bab", "blog", "v2")
	if r.Mode != string(activator.ModeHold) || r.Build != "v2" || r.Host != "a" {
		t.Fatalf("record = %+v, want build v2 held down in the mode Swap ran", r)
	}
}

// Carried from Task 4: every Pass prunes, a held-down instance whose route
// still names the failed build keeps the record, and a record no route
// installs any more goes.
func TestPassPrunesOnlyRecordsNoRouteInstalls(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, PromotedAt: promotedAt, At: t0})
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v0", Host: "b", PromotedAt: promotedAt, At: t0})
	h.install("v2", "a")
	h.pass()
	if _, ok := h.book.Lookup("bab", "blog", "v0"); ok {
		t.Fatal("a record no route installs survived the pass")
	}
	if !h.book.HeldDown("bab", "blog", "v2", "a") {
		t.Fatal("the pass pruned a held-down instance's record while its route still names the build")
	}
	h.install("v1", "a") // rolled back by hand
	h.pass()
	if len(h.book.List()) != 0 {
		t.Fatalf("records = %+v after the last route left v2", h.book.List())
	}
}

// Carried from Task 6, spec §8.2/§8.4: an install by hand that clears the
// record lifts the halt with it. A halt is in memory and outlived its record,
// so installing the build again later was refused for good.
func TestAnInstallByHandThatClearsTheRecordLiftsTheHalt(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.stage("v2")
	h.failing.Store(true)
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Build == "v2" && h.failing.Load() {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.install("v2", "a", "b")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	h.install("v1", "a", "b")
	h.pass()
	if len(h.book.List()) != 0 {
		t.Fatal("the record outlived every route installing the build")
	}
	h.failing.Store(false)
	h.install("v2", "a", "b")
	h.pass()
	waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b") })
	calls, _, _ := h.act.snapshot()
	if c := calls[1].a; !c.Canary {
		t.Fatalf("first attempt after the hold lifted = %+v, want a fresh canary", c)
	}
}

// Carried from Task 6: `carlos instances retry` runs that host as a canary
// through its cohort's halt, and only that host.
func TestRetryHostRunsACanaryThroughAHalt(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.stage("v2")
	h.failing.Store(true)
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Build == "v2" && h.failing.Load() {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.install("v2", "a", "b")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	h.failing.Store(false)
	if !h.retry("b", "v2") {
		t.Fatal("RetryHost did not clear b")
	}
	waitFor(t, "b on v2", func() bool { return h.servingAll("v2", "b") })
	calls, _, _ := h.act.snapshot()
	if c := calls[1].a; c.Host != "b" || !c.Canary || c.Cause != activator.CauseRetry {
		t.Fatalf("attempt after the host retry = %+v, want b's retry canary", c)
	}
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if h.servingAll("v2", "a") {
		t.Fatal("a host retry lifted the halt for another host")
	}
}

// Carried from Task 6: a failed config attempt backs off even when its row
// cannot be read afterwards; the backoff is set in the step that frees the
// slot, so nothing can re-admit the attempt before it. The row is unreadable
// only for the read that follows the failure: the next attempt's own read
// succeeds, so only the backoff can keep it from running.
func TestAFailedConfigAttemptBacksOffWithItsRowUnreadable(t *testing.T) {
	h := newHarness(t, "a")
	var failed, refused atomic.Bool
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Cause == activator.CauseConfig {
			failed.Store(true)
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.rows.mu.Lock()
	h.rows.getErr = func(string) error {
		if failed.Load() && refused.CompareAndSwap(false, true) {
			return errors.New("registry: database is locked")
		}
		return nil
	}
	h.rows.mu.Unlock()
	h.pass()
	h.m.ConfigDelivered("a", "sha256:c1")
	waitFor(t, "the first config attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) >= 1 })
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
		t.Fatalf("%d config attempts with no backoff after an unreadable row", len(calls))
	}
}

// Carried from Task 7: activating an older revision than the delivered one
// leaves the delivered revision's failure count, and so its backoff, alone.
func TestAnOlderActivationKeepsTheConfigFailureCount(t *testing.T) {
	h := newHarness(t, "a")
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Cause == activator.CauseConfig {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.pass()
	h.m.ConfigDelivered("a", "sha256:c2")
	waitFor(t, "the pending event", func() bool { return h.events.has(configPendingHead) })
	h.m.configActivated("a", "sha256:c1")
	h.m.mu.Lock()
	fails := h.m.cfg["a"].fails
	h.m.mu.Unlock()
	if fails != 1 {
		t.Fatalf("fails = %d after an older revision activated, want 1", fails)
	}
}

// Carried from Task 6, spec §8.1: an instance that goes to sleep with a
// config attempt pending needs no attempt any more. Its next wake starts with
// the delivered file, so the revision is active, and the held ack goes out as
// a sleeper's does, marked Hibernating. After it wakes, no config attempt
// restarts a generation that already runs the revision.
func TestASleepersPendingConfigIsNoAttemptAfterItWakes(t *testing.T) {
	h := newHarness(t, "a")
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Cause == activator.CauseConfig {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.pass()
	var mu sync.Mutex
	var acks []appconfig.ConfigAck
	ack := h.m.DeferAck(func(_ context.Context, a appconfig.ConfigAck) {
		mu.Lock()
		defer mu.Unlock()
		acks = append(acks, a)
	})
	h.m.ConfigDelivered("a", "sha256:c1")
	ack(appconfig.ConfigAck{Host: "a", Restarted: true})
	waitFor(t, "the failed config attempt", func() bool { return h.events.has(configPendingHead) })
	h.act.mu.Lock()
	delete(h.act.gens, "a") // hibernated inside its backoff
	h.act.mu.Unlock()
	h.pass()
	waitFor(t, "the held ack", func() bool { mu.Lock(); defer mu.Unlock(); return len(acks) == 1 })
	mu.Lock()
	if !acks[0].Hibernating || acks[0].Restarted {
		t.Fatalf("ack = %+v, want a sleeper's ack, Hibernating", acks[0])
	}
	mu.Unlock()
	h.act.serve("a", "v1") // woke, with the delivered file
	h.advance(2 * time.Minute)
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
		t.Fatalf("calls = %+v: a woken sleeper ran its old pending config attempt", calls)
	}
}

// Spec §8.1, fix round 1: a config attempt pending for c1 while the instance
// was awake is no attempt any more once c2 is delivered to it asleep, even
// if it wakes before the scheduler ever saw it asleep (configSlept never
// fires). The wake starts with c2. The stale c1 attempt would restart a
// generation already running c2 and then mark c1 active.
func TestAConfigDeliveredToASleeperClearsTheOlderPendingAttempt(t *testing.T) {
	h := newHarness(t, "a")
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Cause == activator.CauseConfig {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.pass()
	h.m.ConfigDelivered("a", "sha256:c1")
	waitFor(t, "the failed c1 attempt", func() bool { return h.events.has(configPendingHead) })
	s := h.m.schedulerOf("a")
	waitFor(t, "the slot to free with c1 pending", func() bool {
		s.mu.Lock()
		defer s.mu.Unlock()
		hs := s.q.hosts["a"]
		return hs != nil && hs.running == nil && hs.wantConfig == "sha256:c1"
	})
	h.act.mu.Lock()
	delete(h.act.gens, "a") // hibernated inside the backoff; no scheduler iteration runs
	h.act.mu.Unlock()
	h.m.ConfigDelivered("a", "sha256:c2")
	h.act.serve("a", "v1") // woke at once, with c2 in the file
	h.m.configActivated("a", "sha256:c2")
	h.advance(2 * time.Minute)
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
		t.Fatalf("calls = %+v: a stale c1 attempt ran against a generation already on c2", calls)
	}
	h.m.mu.Lock()
	tr := h.m.track("a")
	delivered, active := tr.delivered, tr.active
	h.m.mu.Unlock()
	if delivered != "sha256:c2" || active != "sha256:c2" {
		t.Fatalf("delivered %q, active %q: want both c2", delivered, active)
	}
}

// Spec §8.4: a record stays while any instance is on the failed build. A
// held-down instance whose row is "draining" during a Pass is not managed
// for that Pass, but it still names the build: the record and the halt stay.
func TestPruneKeepsTheRecordOfAHeldDownHostWhoseRowLeavesReady(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, PromotedAt: promotedAt, At: t0})
	r := h.rows.get("a")
	r.Version, r.Status = "v2", "draining"
	h.rows.put(r)
	h.pass()
	if !h.book.HeldDown("bab", "blog", "v2", "a") {
		t.Fatal("a Pass dropped the record of a held-down host whose row names the build")
	}
	r.Status = "ready"
	h.rows.put(r)
	h.pass()
	if !h.book.HeldDown("bab", "blog", "v2", "a") {
		t.Fatal("the held-down host lost its record when its row returned to ready")
	}
}

// Plan 3's deferral (O5): an attempt abandoned while it is still the host's
// newest (a hold whose stop withdrew, a lost re-check) waits before it runs
// again, rather than looping stop cycles. One abandoned by a newer desire
// does not wait (TestTwoReleasesTwentySecondsApartNeverSwitchToTheFirst).
func TestAnAbandonedAttemptStillNewestWaitsBeforeItRunsAgain(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapAbandoned }
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the first attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) >= 1 })
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
		t.Fatalf("%d attempts: an abandoned attempt ran again at once", len(calls))
	}
	h.advance(abandonDeferral + time.Second)
	h.pass()
	waitFor(t, "the attempt after the wait", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 2 })
	if len(h.book.List()) != 0 {
		t.Fatal("an abandoned attempt wrote a failure record")
	}
}

// PF5 (plan 3's O8): member-facing events carry the reason word, never the
// swap's diagnostic text, which can name paths and internals.
func TestMemberEventsCarryOnlyTheReasonWord(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.act.reason = "spawn: fork/exec /opt/carlos/versions/bab/v2/blog: permission denied"
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapFailed }
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the halt event", func() bool {
		return h.events.has(textHead(textCohortHalted("v2", "\x00", status.ReasonHealth), "\x00"))
	})
	waitFor(t, "the failed event", func() bool { return h.events.has(textHead(textSwapFailed("v2", "\x00", status.ReasonHealth), "\x00")) })
	for _, raw := range []string{"permission denied", "fork/exec", "blog:"} {
		if h.events.has(raw) {
			t.Fatalf("a member event carries the raw reason %q", raw)
		}
	}
	if !h.events.has(textSwapFailed("v2", "v1", status.ReasonHealth)) ||
		!h.events.has(textCohortHalted("v2", "box-1", status.ReasonHealth)) {
		t.Fatal("the events do not carry the reason word")
	}
}

// Astra R3 §5 item 4 (execution admission): an attempt is admitted from the
// pass's view of a row, and the row may change before the scheduler runs it.
// execute revalidates the row it reads: a row that has become a named-unit
// route, or that now belongs to another app, is no longer this scheduler's,
// so the attempt ends abandoned with nothing run (no Swap, no ReplaceWorker,
// no record). Before the fix the attempt went ahead on the changed row.
func TestARowThatLeavesTheSchedulerBeforeDispatchIsAbandoned(t *testing.T) {
	for _, worker := range []bool{false, true} {
		for name, change := range map[string]func(*registry.Route){
			"a named-unit route": func(r *registry.Route) { r.Unit = "blog.service" },
			"another app":        func(r *registry.Route) { r.App = "shop" },
		} {
			kind := "socket instance"
			if worker {
				kind = "worker"
			}
			t.Run(kind+"/"+name, func(t *testing.T) {
				var h *harness
				if worker {
					h = workerHarness(t, "h")
				} else {
					h = newHarness(t, "h")
				}
				var (
					mu        sync.Mutex
					abandoned []string
				)
				h.m.d.Logf = func(format string, args ...any) {
					if line := fmt.Sprintf(format, args...); strings.Contains(line, "abandoned") {
						mu.Lock()
						abandoned = append(abandoned, line)
						mu.Unlock()
					}
				}
				h.stage("v2")
				h.install("v2", "h")
				// The row changes between the pass that queued the attempt and
				// the scheduler's read of it.
				h.rows.mu.Lock()
				h.rows.getHook = func(host string) {
					r := h.rows.get(host)
					change(&r)
					h.rows.put(r)
				}
				h.rows.mu.Unlock()
				moves := func() int {
					calls, _, _ := h.act.snapshot()
					return len(calls) + len(h.act.replacements())
				}
				h.pass()
				waitFor(t, "the attempt to end", func() bool {
					mu.Lock()
					n := len(abandoned)
					mu.Unlock()
					return (n > 0 || moves() > 0) && !h.m.attemptRunning("h")
				})
				if n := moves(); n != 0 {
					t.Fatalf("%d Swap/ReplaceWorker calls on a row that left the scheduler, want none", n)
				}
				mu.Lock()
				got := abandoned[0]
				mu.Unlock()
				if !strings.Contains(got, "route no longer managed") {
					t.Fatalf("abandoned as %q, want the route no longer managed", got)
				}
				if recs := h.book.List(); len(recs) != 0 {
					t.Fatalf("records %+v for an attempt that never ran", recs)
				}
			})
		}
	}
}

func TestReasonWords(t *testing.T) {
	for text, want := range map[string]string{
		"health check failed":                                               status.ReasonHealth,
		"swap: successor exited during its settle":                          status.ReasonSettle,
		"exited within 10s of becoming healthy":                             status.ReasonExit,
		"staging: digest mismatch":                                          status.ReasonStaging,
		"version name reused with different content, ship under a new name": status.ReasonReused,
		"GET / answered 500":                                                status.ReasonHealth,
		"":                                                                  status.ReasonHealth,
	} {
		if got := reasonWord(text); got != want {
			t.Errorf("reasonWord(%q) = %q, want %q", text, got, want)
		}
	}
}

// configPendingHead is the config-pending event up to its reason: the tests
// that wait for it do not know the reason or the backoff.
var configPendingHead = textHead(rolloutwords.EventConfigPending, "%s")

// Task 11 rule 3 (Astra R1 Flag triage T8): the held ack an activation
// releases is handed to the manager's writer, not written by done(). A store
// write that never returns cannot keep the attempt's slot, so the host's
// next attempt is admitted while the writer is still blocked.
func TestABlockedAckWriterDoesNotHoldTheSlot(t *testing.T) {
	h := newHarness(t, "a")
	h.pass()
	h.stage("v2")
	entered, block := make(chan struct{}, 4), make(chan struct{})
	t.Cleanup(func() { close(block) })
	ack := h.m.DeferAck(func(context.Context, appconfig.ConfigAck) {
		entered <- struct{}{}
		<-block
	})
	h.m.ConfigDelivered("a", "sha256:c1")
	ack(appconfig.ConfigAck{Host: "a"})
	select {
	case <-entered:
	case <-time.After(5 * time.Second):
		t.Fatal("the config attempt's ack was never written")
	}
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the release attempt, admitted while the ack writer blocks", func() bool { return h.servingAll("v2", "a") })
}

// Task 11 rule 2 (Astra R1 Flag triage T10): the failing pointer a wake's
// plan reads runs under the wake's context. A store read that never answers
// ends when the wake is cancelled, and the plan goes on with no PromotedAt
// (the failure's own time stands in, the safe direction).
func TestABlockedPointerReadEndsWithTheWake(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.install("v2", "a")
	reading := make(chan struct{})
	var once sync.Once
	h.m.d.Pointer = func(ctx context.Context, _, _, _ string) (*release.ChannelPointer, error) {
		once.Do(func() { close(reading) })
		<-ctx.Done()
		return nil, ctx.Err()
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	plans := make(chan activator.WakePlan, 1)
	go func() { plans <- h.m.Plan(ctx, h.rows.get("a")) }()
	<-reading
	cancel()
	select {
	case p := <-plans:
		if !p.FirstWake || !p.FailedUnder.IsZero() {
			t.Fatalf("plan %+v: want the first wake of v2 with no PromotedAt", p)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("the plan's pointer read outlived the wake's context")
	}
}

// Task 11 Corrections (Task 5 review; spec §8.6): a planned agent restart
// stops the activator first, which cancels a swap mid-flight, and then joins
// the scheduler. Close returns only once that attempt's done() has written
// its outcome (here the hold-ran marker of the hold build that ran) and
// ended its intent, so the restart reconciles nothing. After Close no
// attempt is admitted.
func TestCloseJoinsAnAttemptCancelledByShutdown(t *testing.T) {
	h := newHarness(t, "a")
	dir := h.onDisk()
	h.swap = "hold"
	h.stage("v2", "v3")
	gate := h.act.hold("a")
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
		time.Sleep(200 * time.Millisecond) // the swap winding down after its entry stopped
		return activator.SwapAbandoned
	}
	h.act.executed = func(activator.Attempt, activator.SwapOutcome) bool { return true } // the hold build ran
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the hold swap to start", func() bool { _, s, _ := h.act.snapshot(); return len(s) == 1 })
	if n := len(fileOf(t, dir).Intents); n != 1 {
		t.Fatalf("%d intents on disk while the swap runs, want 1", n)
	}
	close(gate) // svc.Shutdown stops the entry under the swap
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	h.m.Close(ctx)
	f := fileOf(t, dir)
	if len(f.Intents) != 0 {
		t.Fatalf("intents %+v on disk after Close: a restart would reconcile the attempt as a failure", f.Intents)
	}
	if mk, ok := f.HoldRan["a"]; !ok || mk.Build != "v2" {
		t.Fatalf("hold-ran markers %+v after Close, want a's for v2: the attempt's outcome", f.HoldRan)
	}
	h.advance(time.Hour)
	h.install("v3", "a")
	h.pass()
	time.Sleep(100 * time.Millisecond)
	if _, s, _ := h.act.snapshot(); len(s) != 1 {
		t.Fatalf("swaps started %v: an attempt was admitted after Close", s)
	}
}

// Task 10 review minor (1), restored in Task 11: the pointer pass requeues a
// cleared build at once, before it reads the app's next channel, as the
// armed pass did before the pointer work was shared (onCleared). Both
// channels re-sign v2, so the second read always follows a clear.
//
// Only the pass's reads are counted (fix wave 1, item 6). The requeued
// retry's attempt runs on the scheduler's goroutine and reads the pointer
// too, for its FailedUnder (scheduler.execute, failedUnder); counted, it
// made a third read about once in a hundred runs.
func TestAPointerClearRequeuesBeforeTheNextChannelIsRead(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.stage("v2")
	rb := h.rows.get("b")
	rb.Channel = "beta"
	h.rows.put(rb)
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "c", PromotedAt: promotedAt, At: t0})
	h.install("v2", "a", "b")
	h.pass()
	h.pointer("stable", "v2", promotedAt.Add(time.Minute))
	h.pointer("beta", "v2", promotedAt.Add(time.Minute))
	var reads atomic.Int32
	var requeuedLate atomic.Bool
	h.mu.Lock()
	h.pointerHook = func() {
		if !calledFrom("(*Manager).applyPointers") {
			return
		}
		if reads.Add(1) != 2 {
			return
		}
		h.m.mu.Lock()
		n := len(h.m.retried)
		h.m.mu.Unlock()
		if n == 0 {
			requeuedLate.Store(true)
		}
	}
	h.mu.Unlock()
	h.advance(pointerReadEvery)
	h.pass()
	if reads.Load() != 2 {
		t.Fatalf("%d pointer reads, want 2 (stable and beta)", reads.Load())
	}
	if requeuedLate.Load() {
		t.Fatal("the cleared build was requeued only after every channel was read")
	}
	waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b") })
}

// Task 11 Corrections (plan review #10): acks are ordered at the transition.
// The writer keeps one pending ack per host, the newest by sequence; an older
// one is dropped, never written after a newer one; hosts are written in the
// order they were first queued; each write is bounded by ackWriteBound.
func TestAcksAreWrittenNewestLastPerHost(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	w := newAckWriter(ctx, func(string, ...any) {})
	var mu sync.Mutex
	var got []string
	write := func(tag string) func(context.Context) {
		return func(ctx context.Context) {
			if dl, ok := ctx.Deadline(); !ok || time.Until(dl) > ackWriteBound {
				t.Errorf("ack %s written with deadline %v (%v), want at most ackWriteBound", tag, dl, ok)
			}
			mu.Lock()
			got = append(got, tag)
			mu.Unlock()
		}
	}
	entered, block := make(chan struct{}), make(chan struct{})
	first := write("a1")
	w.submit("a", ackJob{seq: 1, write: func(ctx context.Context) { close(entered); <-block; first(ctx) }})
	<-entered
	w.submit("a", ackJob{seq: 3, write: write("a3")})
	w.submit("a", ackJob{seq: 2, write: write("a2")}) // older than the pending a3
	w.submit("b", ackJob{seq: 4, write: write("b4")})
	w.submit("a", ackJob{seq: 5, write: write("a5")}) // replaces a3, keeps a's place
	close(block)
	waitFor(t, "three writes", func() bool { mu.Lock(); defer mu.Unlock(); return len(got) == 3 })
	w.submit("a", ackJob{seq: 2, write: write("a2-late")}) // older than one written
	time.Sleep(50 * time.Millisecond)
	mu.Lock()
	defer mu.Unlock()
	if want := []string{"a1", "a5", "b4"}; strings.Join(got, ",") != strings.Join(want, ",") {
		t.Fatalf("written %v, want %v", got, want)
	}
}

// Task 11 Corrections (spec §8.6): Close also joins a wake event under way.
// A terminal OnGen writes the wake's outcome and ends its intent; Close does
// not return while it is still running.
func TestCloseWaitsForAnOnGenUnderWay(t *testing.T) {
	h := newHarness(t, "a")
	h.pass()
	entered, release := make(chan struct{}), make(chan struct{})
	var once sync.Once
	h.rows.mu.Lock()
	h.rows.setServingHook = func(string) {
		once.Do(func() { close(entered); <-release })
	}
	h.rows.mu.Unlock()
	g, _ := h.act.Current("a")
	go h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: h.rows.get("a"), Gen: g, FirstWake: true})
	<-entered
	closed := make(chan struct{})
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		h.m.Close(ctx)
		close(closed)
	}()
	select {
	case <-closed:
		t.Fatal("Close returned while a wake event was still being recorded")
	case <-time.After(100 * time.Millisecond):
	}
	close(release)
	select {
	case <-closed:
	case <-time.After(5 * time.Second):
		t.Fatal("Close did not return once the wake event finished")
	}
}

// Task 11 Corrections: closing the writer writes what is pending, each
// write still bounded, and an ack submitted after it is dropped.
func TestClosingTheAckWriterWritesWhatIsPending(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	w := newAckWriter(ctx, func(string, ...any) {})
	var mu sync.Mutex
	var got []string
	write := func(tag string) func(context.Context) {
		return func(context.Context) { mu.Lock(); got = append(got, tag); mu.Unlock() }
	}
	entered, block := make(chan struct{}), make(chan struct{})
	w.submit("a", ackJob{seq: 1, write: func(context.Context) { close(entered); <-block }})
	<-entered
	w.submit("b", ackJob{seq: 2, write: write("b2")})
	w.submit("c", ackJob{seq: 3, write: write("c3")})
	go func() { time.Sleep(50 * time.Millisecond); close(block) }()
	cctx, ccancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer ccancel()
	w.close(cctx)
	w.submit("d", ackJob{seq: 4, write: write("d4")})
	time.Sleep(50 * time.Millisecond)
	mu.Lock()
	defer mu.Unlock()
	if strings.Join(got, ",") != "b2,c3" {
		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")
		}
	})
}

// calledFrom reports whether a function whose name ends in fn is on the
// calling goroutine's stack.
func calledFrom(fn string) bool {
	pc := make([]uintptr, 64)
	frames := runtime.CallersFrames(pc[:runtime.Callers(2, pc)])
	for {
		f, more := frames.Next()
		if strings.HasSuffix(f.Function, fn) {
			return true
		}
		if !more {
			return false
		}
	}
}