package rollout

import (
	"context"
	"errors"
	"strings"
	"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/status"
)

func wakeRoute(installed, good string) registry.Route {
	return registry.Route{Host: "a", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true,
		Version: installed, GoodVersion: good, GoodDigest: "sha-" + good}
}

func inputs(rt registry.Route, b *FailureBook, p AppPolicy) wakeInputs {
	return wakeInputs{Route: rt, OptDir: "/opt/carlos", Policy: p, Book: b,
		Demand: func(ver, _ string) (string, string, error) { return "", "sha-" + ver, nil }}
}

func TestDecideWake(t *testing.T) {
	overlap := DefaultPolicy()
	held := &FailureBook{}
	held.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "x", At: t0})
	down := &FailureBook{}
	down.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, At: t0})

	if p, _ := decideWake(inputs(wakeRoute("", ""), &FailureBook{}, overlap)); p.Boot.StagedPath != "" || p.Refuse != "no build is installed for this instance yet" {
		t.Fatal("nothing stamped: today's wake, unplanned")
	}
	p, _ := decideWake(inputs(wakeRoute("v1", "v1"), &FailureBook{}, overlap))
	if p.Boot.Version != "v1" || p.FirstWake || p.Boot.Digest != "sha-v1" {
		t.Fatalf("installed is the good build: recovery by the retained identity, got %+v", p)
	}
	p, _ = decideWake(inputs(wakeRoute("v2", "v1"), &FailureBook{}, overlap))
	if p.Boot.Version != "v2" || !p.FirstWake || p.Fallback == nil || p.Fallback.Version != "v1" || p.Mode != activator.ModeOverlap {
		t.Fatalf("a new build: a first wake with the good build behind it, got %+v", p)
	}
	p, _ = decideWake(inputs(wakeRoute("v2", "v1"), &FailureBook{}, AppPolicy{Swap: activator.ModeHold, Concurrency: 1}))
	if p.Mode != activator.ModeHold || p.Fallback != nil {
		t.Fatalf("hold: no fallback, got %+v", p)
	}
	in := inputs(wakeRoute("v2", "v1"), &FailureBook{}, overlap)
	in.Demand = func(ver, _ string) (string, string, error) { return "hold", "sha-" + ver, nil }
	if p, _ := decideWake(in); p.Mode != activator.ModeHold || p.Fallback != nil {
		t.Fatalf("the release demands hold: no fallback, got %+v", p)
	}
	if p, _ := decideWake(inputs(wakeRoute("v2", "v1"), held, overlap)); p.Boot.Version != "v1" || p.FirstWake {
		t.Fatalf("installed build held by a record: boot the good build, got %+v", p)
	}
	// Spec §8.5: an instance with no good build has no fallback, which is
	// today's behaviour. A sibling's record must not refuse it outright.
	if p, _ := decideWake(inputs(wakeRoute("v2", ""), held, overlap)); p.Refuse != "" || p.Boot.Version != "v2" || !p.FirstWake ||
		p.Fallback != nil || p.Mode != activator.ModeOverlap {
		t.Fatalf("held, no good build, not held down: boot the installed build as a first wake with no fallback, got %+v", p)
	}
	if p, _ := decideWake(inputs(wakeRoute("v2", ""), down, overlap)); p.Refuse == "" {
		t.Fatal("held down with no good build: it stays down (spec §6.4)")
	}
	if p, _ := decideWake(inputs(wakeRoute("v2", "v1"), down, overlap)); p.Refuse == "" {
		t.Fatal("held down after a hold failure: no automatic fallback (spec §6.4)")
	}
	in = inputs(wakeRoute("v2", "v1"), &FailureBook{}, overlap)
	in.Demand = func(string, string) (string, string, error) { return "", "", errors.New("digest mismatch") }
	p, err := decideWake(in)
	if err == nil || p.Boot.Version != "v1" || p.FirstWake {
		t.Fatalf("a staging refusal with a good build boots the good build, got %+v, %v", p, err)
	}
}

// Review focus 3.
func TestPlanBootsTheGoodBuildForAHeldInstallation(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.install("v2", "a")
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0})
	p := h.m.Plan(context.Background(), h.rows.get("a"))
	if p.Boot.Version != "v1" || p.FirstWake {
		t.Fatalf("plan = %+v, want the good v1 without a repoint", p)
	}
}

// registerThenInstall makes the scheduler know host (the pass) while every
// host still wants v1, and only then stamps v2 on host's row. The pass sees
// nothing to adopt, so no swap onto v2 runs beside the wake outcome a test
// hands OnGen: these tests are about the wake, and a swap racing it would
// change what the activator serves under them.
func (h *harness) registerThenInstall(ver, host string) {
	h.pass()
	h.install(ver, host)
}

func TestAFellBackWakeRecordsTheFailureAndKeepsTheGoodBuild(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.registerThenInstall("v2", "a")
	rt := h.rows.get("a")
	h.act.serve("a", "v1") // the good build the fallback left serving
	served, _ := h.act.Current("a")
	h.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: rt, FirstWake: true, Mode: activator.ModeOverlap,
		Gen: served, Failed: release.StagedIdentity{Version: "v2"}, Reason: "health check failed"})
	if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
		t.Fatal("no failure record")
	}
	if r := h.rows.get("a"); r.GoodVersion != "v1" || r.ServingVersion != "v1" {
		t.Fatalf("row = %+v", r)
	}
	if !h.events.has(textFallback("v2", "v1")) {
		t.Fatal("no fallback event")
	}
}

// Spec §8.1, §9.4 (Astra R3 flag triage T11): a first wake that fails on a
// host whose app has no scheduler yet still halts the cohort and says so,
// once, when the record is new. The same failure delivered again (the record
// stands) is not announced twice, and a failure of an execution that began
// before a retry of the build (a pre-retry record) is never announced as a
// halt.
func TestAFirstWakeFailureWithNoSchedulerAnnouncesTheHalt(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.install("v2", "a") // no pass: the manager has no scheduler for the host
	if h.m.schedulerOf("a") != nil {
		t.Fatal("the harness already has a scheduler for the host")
	}
	halted := textCohortHalted("v2", h.m.d.Box, status.ReasonHealth)
	ev := activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("a"), FirstWake: true, Mode: activator.ModeOverlap,
		Failed: release.StagedIdentity{Version: "v2"}, Reason: "GET / answered 500"}
	h.m.OnGen(ev)
	if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
		t.Fatal("no failure record")
	}
	if n := h.events.count(halted); n != 1 {
		t.Fatalf("%d cohort-halted event(s) after a new record, want 1", n)
	}
	h.m.OnGen(ev)
	if n := h.events.count(halted); n != 1 {
		t.Fatalf("%d cohort-halted event(s) after the same failure again, want still 1", n)
	}

	// A wake planned before a retry of its build fails stale: PreRetry.
	p := newHarness(t, "a")
	p.stage("v2")
	p.install("v2", "a")
	p.wakeHost("a")
	plan := p.m.Plan(context.Background(), p.rows.get("a"))
	if !plan.FirstWake || plan.Boot.Version != "v2" {
		t.Fatalf("plan %+v: want a first wake of v2", plan)
	}
	if p.m.schedulerOf("a") != nil {
		t.Fatal("the harness has a scheduler for the host")
	}
	p.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s", PromotedAt: t0, At: t0})
	if ok, err := p.book.RetryHost("bab", "blog", "v2", "a"); !ok || err != nil {
		t.Fatalf("RetryHost = %v, %v: cleared nothing", ok, err)
	}
	p.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: p.rows.get("a"), Failed: plan.Boot, Mode: plan.Mode,
		FirstWake: true, Reason: "health check failed", Token: plan.Token, ConfigRev: plan.ConfigRev})
	if n := p.events.count(halted); n != 0 {
		t.Fatalf("%d cohort-halted event(s) for a failure that began before a retry", n)
	}
	if p.book.Blocks("bab", "blog", "v2", "a", false) {
		t.Fatal("the stale failure halted the retried host")
	}
}

// Spec §8.1: a fallback publishes the failure hold before the serving build
// and the active config. Writing the serving build first showed the good
// build serving with no hold on the failed one, a window in which a wake or
// an admission could select the failed build again.
func TestAFallbackHoldsTheFailedBuildBeforeItPublishesServing(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.registerThenInstall("v2", "a")
	h.act.serve("a", "v1")
	served, _ := h.act.Current("a")
	var early atomic.Int32
	h.rows.mu.Lock()
	h.rows.setServingHook = func(string) {
		if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
			early.Add(1)
		}
	}
	h.rows.mu.Unlock()
	h.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: h.rows.get("a"), FirstWake: true, Mode: activator.ModeOverlap,
		Gen: served, Failed: release.StagedIdentity{Version: "v2"}, Reason: "health check failed"})
	if n := early.Load(); n != 0 {
		t.Fatal("the serving build was published before the failed build was held")
	}
	if r := h.rows.get("a"); r.ServingVersion != "v1" {
		t.Fatalf("serving = %q, want the good v1", r.ServingVersion)
	}
}

func TestOnlyASettledFirstWakeBecomesTheGoodBuild(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.registerThenInstall("v2", "a")
	rt := h.rows.get("a")
	h.act.serve("a", "v2") // the first wake's generation, which the activator serves
	g, _ := h.act.Current("a")
	h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: g, FirstWake: true})
	if h.rows.get("a").GoodVersion != "v1" {
		t.Fatal("a first wake became the good build before its 10 s (spec §8.5)")
	}
	h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: rt, Gen: g, FirstWake: true})
	if r := h.rows.get("a"); r.GoodVersion != "v2" || r.ServingVersion != "v2" {
		t.Fatalf("row = %+v: the settled first wake must be the good and the serving build", r)
	}
}

// Task 9's verdict floor: a first wake left too little health window boots
// the good build untried and reports an ordinary healthy wake of it
// (FirstWake false). The scheduler records the good build as serving and
// records nothing against the untried build, which stays installed for a
// later wake.
func TestAnUntriedFirstWakeIsAnOrdinaryWakeOfTheGoodBuild(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.registerThenInstall("v2", "a")
	h.act.serve("a", "v1")
	g, _ := h.act.Current("a")
	h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, Mode: activator.ModeOverlap})
	if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
		t.Fatal("the untried build was recorded as failed")
	}
	if r := h.rows.get("a"); r.ServingVersion != "v1" || r.GoodVersion != "v1" || r.Version != "v2" {
		t.Fatalf("row = %+v: want v1 serving and good, v2 still installed", r)
	}
	if p := h.m.Plan(context.Background(), h.rows.get("a")); !p.FirstWake || p.Boot.Version != "v2" {
		t.Fatalf("plan %+v: the untried build must stay a candidate for the next wake", p)
	}
}

// Spec §6.4: an intentional stop during the 10 s observation records
// nothing. GenCancelled only closes the observation.
//
// What it must not publish is the observed build, v2 (fix wave 1, item 6).
// The row's serving build is not required to stay empty: the scheduler last
// saw host a wanting v1 (registerThenInstall), and with the fake activator
// now serving v2 its loop may admit a swap back onto v1 at any moment, whose
// success acks v1 off this goroutine. Asserting "" made the test fail
// whenever that swap won the race (seen 3 in 300 runs beside the pointer
// test). The cancel's own writes are flushed before the row is read.
func TestACancelledObservationRecordsNothing(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.registerThenInstall("v2", "a")
	h.act.serve("a", "v2")
	g, _ := h.act.Current("a")
	h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), Gen: g, FirstWake: true})
	h.flushAcks()
	if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
		t.Fatal("a cancelled observation recorded a failure")
	}
	if r := h.rows.get("a"); r.GoodVersion != "v1" || r.ServingVersion == "v2" {
		t.Fatalf("row = %+v: a cancelled observation published a build", r)
	}
}

// Spec §7.9: a worker's serving build has one writer, the activator
// (ordered by incarnation). The scheduler records a worker's good build but
// leaves its ServingVersion alone, so its write can never interleave with
// the activator's and restore an older value.
func TestTheSchedulerLeavesAWorkersServingVersionAlone(t *testing.T) {
	h := newHarness(t, "a")
	h.act.serve("a", "v2")
	g, _ := h.act.Current("a")
	g.Worker = true
	h.m.served("a", g, 0)
	if r := h.rows.get("a"); r.ServingVersion != "" || r.GoodVersion != "v2" {
		t.Fatalf("row = %+v: want the good build recorded and the serving build left to the activator", r)
	}
}

// Spec §8.5: a good build is written expecting the current one. A late
// success of an older generation (a completion delayed past a newer swap, a
// stale wake event) is not the generation the activator serves, so it never
// overwrites the newer good or serving build. Writing whatever arrived last
// regressed both to the older build.
func TestAStaleSuccessNeverOverwritesANewerBuild(t *testing.T) {
	h := newHarness(t, "a")
	h.act.serve("a", "v2")
	older, _ := h.act.Current("a")
	h.act.serve("a", "v3") // a newer generation took over
	newer, _ := h.act.Current("a")
	h.m.served("a", newer, 0)
	h.m.served("a", older, 0) // the older one's completion arrives late
	h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: h.rows.get("a"), Gen: older, FirstWake: true})
	if r := h.rows.get("a"); r.GoodVersion != "v3" || r.ServingVersion != "v3" {
		t.Fatalf("row = %+v: an older generation overwrote the newer good and serving build", r)
	}
}

func TestAFailedHoldFirstWakeHoldsTheInstanceDown(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.registerThenInstall("v2", "a")
	h.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("a"), FirstWake: true, Mode: activator.ModeHold,
		Failed: release.StagedIdentity{Version: "v2"}, Reason: "health check failed"})
	if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" {
		t.Fatalf("plan = %+v, want a refusal: recovery after a hold failure is explicit", p)
	}
	// Task 4's rule: a record never carries a zero PromotedAt. No pointer was
	// read for this wake, so the failure's own time stands in.
	if rec, ok := h.book.Lookup("bab", "blog", "v2"); !ok || !rec.PromotedAt.Equal(t0) || !rec.At.Equal(t0) {
		t.Fatalf("record %+v %v: want PromotedAt = At = the failure's time", rec, ok)
	}
	if !h.events.has(textCohortHalted("v2", "box-1", status.ReasonHealth)) {
		t.Fatal("no cohort-halted event with the reason word")
	}
}

func TestAWakeActivatesTheConfigItStartedWith(t *testing.T) {
	h := newHarness(t, "a")
	h.pass()
	h.act.mu.Lock()
	delete(h.act.gens, "a")
	h.act.owned["a"] = true // waking: a process, not yet current
	h.act.mu.Unlock()
	var acks int
	ack := h.m.DeferAck(func(context.Context, appconfig.ConfigAck) { acks++ })
	h.m.ConfigDelivered("a", "sha256:c1")
	ack(appconfig.ConfigAck{Host: "a"})
	h.flushAcks()
	if acks != 0 {
		t.Fatal("acknowledged before any generation started with the config")
	}
	rt := h.rows.get("a")
	p := h.m.Plan(context.Background(), rt)
	// The activator carries the plan's token and revision onto the event.
	h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: activator.Gen{Build: "v1", Digest: "sha-v1"},
		Token: p.Token, ConfigRev: p.ConfigRev})
	h.flushAcks()
	if acks != 1 {
		t.Fatalf("acks = %d after the wake, want 1", acks)
	}
}

// Spec §8.1: a sleeping instance needs no config attempt. Its next wake
// starts with the delivered file, and config sync's ack goes out at
// delivery, as today, marked Hibernating.
func TestASleepersConfigNeedsNoAttempt(t *testing.T) {
	h := newHarness(t, "a")
	h.pass()
	h.act.mu.Lock()
	delete(h.act.gens, "a") // asleep: no process at all
	h.act.mu.Unlock()
	var acks []appconfig.ConfigAck
	ack := h.m.DeferAck(func(_ context.Context, a appconfig.ConfigAck) { acks = append(acks, a) })
	h.m.ConfigDelivered("a", "sha256:c1")
	ack(appconfig.ConfigAck{Host: "a", Restarted: true})
	h.flushAcks()
	if len(acks) != 1 || !acks[0].Hibernating || acks[0].Restarted {
		t.Fatalf("acks = %+v, want one at delivery, Hibernating", acks)
	}
	h.pass()
	time.Sleep(50 * time.Millisecond)
	if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
		t.Fatalf("attempts %+v: a sleeper got a config attempt", calls)
	}
}

// Spec §8.1: a failed first wake is published with no blocking read in
// between. The failing pointer's PromotedAt is read when the wake is
// planned and travels on the event. buildFailed read it again after the
// failure, before the record, leaving every sibling's wake free to plan the
// failed build meanwhile.
func TestAFailedFirstWakeIsRecordedWithNoPointerRead(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.install("v2", "a")
	promoted := t0.Add(-time.Hour)
	h.pointer("stable", "v2", promoted)
	p := h.m.Plan(context.Background(), h.rows.get("a"))
	if !p.FirstWake || !p.FailedUnder.Equal(promoted) {
		t.Fatalf("plan %+v: want a first wake carrying the pointer's PromotedAt", p)
	}
	var reads atomic.Int32
	h.mu.Lock()
	h.pointerHook = func() { reads.Add(1) }
	h.mu.Unlock()
	h.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("a"), FirstWake: true, Mode: activator.ModeOverlap,
		Failed: p.Boot, Reason: "GET / answered 500", FailedUnder: p.FailedUnder})
	if n := reads.Load(); n != 0 {
		t.Fatalf("%d pointer read(s) between the failure and its record", n)
	}
	rec, ok := h.book.Lookup("bab", "blog", "v2")
	if !ok || !rec.PromotedAt.Equal(promoted) {
		t.Fatalf("record %+v %v: want one carrying the planned PromotedAt", rec, ok)
	}
}

// Spec §6.4, §8.4: a cold managed host (nothing installed yet) has its build
// installed and stamped before the wake is planned, so the plan boots that
// identity, as a first wake, and respects its failure records. An empty
// plan sent the wake down the live-link path, which materialized the
// channel head and ran it unplanned, a held build included.
func TestAColdManagedHostIsInstalledBeforeItIsPlanned(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	cold := func() {
		r := h.rows.get("a")
		r.Version = ""
		h.rows.put(r)
	}
	cold()
	installs := 0
	h.m.d.Materialize = func(_ context.Context, host string) error {
		installs++
		r := h.rows.get(host)
		r.Version = "v2"
		h.rows.put(r)
		return nil
	}
	if p := h.m.Plan(context.Background(), h.rows.get("a")); installs != 1 || p.Boot.Version != "v2" || !p.FirstWake {
		t.Fatalf("installs %d, plan %+v: want v2 installed, then planned as a first wake", installs, p)
	}
	if _, err := h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", At: t0}); err != nil {
		t.Fatal(err)
	}
	cold()
	if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Boot.Version == "v2" {
		t.Fatalf("plan %+v: a cold host was booted onto a build its app holds", p)
	}
	h.m.d.Materialize = nil
	cold()
	if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" || p.Boot.StagedPath != "" {
		t.Fatalf("plan %+v: with nothing installed and no way to install, the wake must be refused", p)
	}
}

// A refusal's text reaches the member's "wake failed" event (Task 9). A
// failed install's error, which can name box paths, goes to the operator log
// only.
func TestAFailedColdInstallRefusesWithoutItsErrorText(t *testing.T) {
	h := newHarness(t, "a")
	r := h.rows.get("a")
	r.Version = ""
	h.rows.put(r)
	h.m.d.Materialize = func(context.Context, string) error {
		return errors.New("stage /opt/carlos/versions/bab/v2/blog: no space left on device")
	}
	p := h.m.Plan(context.Background(), h.rows.get("a"))
	if p.Refuse == "" || p.Boot.StagedPath != "" {
		t.Fatalf("plan %+v: a failed install must refuse the wake", p)
	}
	if strings.Contains(p.Refuse, "/opt") || strings.Contains(p.Refuse, "no space") {
		t.Fatalf("refusal %q carries the raw error", p.Refuse)
	}
}

// Spec §4.2 (Task 3's carry, as resolve applies it to an attempt): a first
// wake whose canonical demand cannot be read or compared boots in hold, on
// the staged bytes hashed here, with no fallback. Missing or unverifiable
// data never permits an overlap. A staged file that is not the release's
// artifact is a staging refusal: the good build boots and root is asked to
// stage again.
func TestAPlannedWakeReadsTheDemandAsAnAttemptDoes(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.install("v2", "a")
	sum, err := release.FileSHA256(release.VersionPath(h.opt, "bab", "v2", "blog"))
	if err != nil {
		t.Fatal(err)
	}

	h.mu.Lock()
	h.demandErr["v2"] = errors.New("manifest unreadable")
	h.mu.Unlock()
	if p := h.m.Plan(context.Background(), h.rows.get("a")); !p.FirstWake || p.Mode != activator.ModeHold || p.Fallback != nil || p.Boot.Digest != sum {
		t.Fatalf("plan %+v: an unreadable demand must boot v2 in hold on its hashed bytes", p)
	}

	h.mu.Lock()
	delete(h.demandErr, "v2")
	h.noDigest["v2"] = true
	h.mu.Unlock()
	if p := h.m.Plan(context.Background(), h.rows.get("a")); !p.FirstWake || p.Mode != activator.ModeHold || p.Fallback != nil || p.Boot.Digest != sum {
		t.Fatalf("plan %+v: a demand nothing compared must boot v2 in hold on its hashed bytes", p)
	}

	h.mu.Lock()
	delete(h.noDigest, "v2")
	h.mismatch["v2"] = true
	h.mu.Unlock()
	p := h.m.Plan(context.Background(), h.rows.get("a"))
	if p.FirstWake || p.Boot.Version != "v1" {
		t.Fatalf("plan %+v: a mis-staged v2 must boot the good v1", p)
	}
	if h.touched.Load() != 1 || !h.events.has(textStagingRefused("v2")) {
		t.Fatal("a staging refusal must ask root to stage again and say so")
	}
}

// Spec §7.4, §8.1–8.4 (P5 review finding 3): a plan carries its selection
// gate. A sibling's failure recorded after the plan was made refuses the
// planned build at the exec; the good build still passes.
func TestAPlansCheckRefusesABuildHeldSinceThePlan(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.install("v2", "a")
	p := h.m.Plan(context.Background(), h.rows.get("a"))
	if !p.FirstWake || p.Boot.Version != "v2" || p.Check == nil {
		t.Fatalf("plan %+v: want a first wake onto v2 with its check", p)
	}
	if err := p.Check("v2"); err != nil {
		t.Fatalf("nothing holds v2 yet: %v", err)
	}
	if _, err := h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0}); err != nil {
		t.Fatal(err)
	}
	err := p.Check("v2")
	if err == nil {
		t.Fatal("v2 failed on a sibling after the plan, and the check still let it boot")
	}
	// Approved wording (zdd.refuse.held-since-plan).
	if want := "build v2 failed on this box while this instance was starting, so it was not started"; err.Error() != want {
		t.Fatalf("refusal %q, want the draft %q", err, want)
	}
	if err := p.Check("v1"); err != nil {
		t.Fatalf("the good build was refused: %v", err)
	}
}

// Spec §8.5: a record written after the plan does not refuse an instance with
// no good build either. It is not moved onto anything else, and refusing it
// would turn a sibling's failure into an outage. A hold-down still refuses.
func TestAPlansCheckPassesAHostWithNoGoodBuildWhenARecordLandsAfterThePlan(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.install("v2", "a")
	r := h.rows.get("a")
	r.GoodVersion, r.GoodDigest = "", ""
	h.rows.put(r)
	p := h.m.Plan(context.Background(), h.rows.get("a"))
	if !p.FirstWake || p.Boot.Version != "v2" || p.Fallback != nil || p.Check == nil {
		t.Fatalf("plan %+v: want a first wake onto v2, no fallback, with its check", p)
	}
	if _, err := h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0}); err != nil {
		t.Fatal(err)
	}
	if err := p.Check("v2"); err != nil {
		t.Fatalf("no good build, a sibling's record since the plan: the check must pass, got %v", err)
	}
	if _, err := h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, At: t0}); err != nil {
		t.Fatal(err)
	}
	if err := p.Check("v2"); err == nil {
		t.Fatal("held down on this instance since the plan: the check must still refuse")
	}
}

// Spec §2.9 (rev 17), §6.4, §8.3: an overlap attempt on a foreign-layout row
// runs as a hold, and the activator says so (SwapResult.Mode). Its failure
// is a hold build's failure: held down, so no wake boots the good build in
// its place. And while an attempt runs for an instance, no wake of it is
// planned, so a hold that went down is recorded before anything wakes it.
// Recording under the dispatched ModeOverlap left it un-held, and the next
// wake booted the good build against the database the hold build ran on.
func TestAForeignRowsFailedHoldIsHeldDown(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapDown }
	h.act.mode = activator.ModeHold // the effective mode, as plan 3's Swap reports it
	gate := h.act.hold("a")
	h.install("v2", "a")
	h.pass()
	waitFor(t, "the attempt to start", func() bool { _, started, _ := h.act.snapshot(); return len(started) == 1 })
	// Approved wording (zdd.refuse.attempt-running, zdd.refuse.held-down).
	if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse != "this instance is still moving to another build" {
		t.Fatalf("a wake was planned while the instance's attempt ran: %+v", p)
	}
	close(gate)
	waitFor(t, "the hold-down record", func() bool { return h.book.HeldDown("bab", "blog", "v2", "a") })
	wantDown := "build v2 failed in hold mode on this instance, so it stays down until you roll back or retry the build"
	if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse != wantDown || p.Boot.StagedPath != "" {
		t.Fatalf("after the hold went down the wake planned %+v, want refused %q (held down)", p, wantDown)
	}
}

// The activator's sweep calls OnDrift (Nudge) on a goroutine of its own for
// every drifted entry on every sweep, with no dedupe of its own (Task 9).
// Nudges for one host collapse: while one is reading the row, the others
// return at once and leave it one more look, so nothing piles up behind a
// slow read and nothing a later nudge was for is lost. And a nudge for the
// build an attempt is already running onto neither queues it again nor
// supersedes it.
func TestNudgesForOneHostCollapse(t *testing.T) {
	h := newHarness(t, "a")
	h.stage("v2")
	h.pass()
	release1 := make(chan struct{})
	var gets atomic.Int32
	h.rows.mu.Lock()
	h.rows.getHook = func(host string) {
		if host == "a" && gets.Add(1) == 1 {
			<-release1
		}
	}
	h.rows.mu.Unlock()
	first := make(chan struct{})
	go func() { h.m.Nudge("a"); close(first) }()
	waitFor(t, "the first nudge to read the row", func() bool { return gets.Load() == 1 })
	for range 20 {
		h.m.Nudge("a") // returns at once: the first nudge owns this host
	}
	if n := gets.Load(); n != 1 {
		t.Fatalf("%d row reads while one nudge was in flight, want 1", n)
	}
	close(release1)
	<-first
	if n := gets.Load(); n != 2 {
		t.Fatalf("%d row reads, want 2: the in-flight nudge looks once more for the ones it absorbed", n)
	}

	h.rows.mu.Lock()
	h.rows.getHook = nil
	h.rows.mu.Unlock()
	gate := h.act.hold("a")
	h.install("v2", "a")
	h.m.Nudge("a")
	waitFor(t, "the attempt onto v2 to start", func() bool { _, started, _ := h.act.snapshot(); return len(started) == 1 })
	for range 5 {
		h.m.Nudge("a")
	}
	close(gate)
	waitFor(t, "the attempt to finish", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
	calls, started, _ := h.act.snapshot()
	if calls[0].outcome != activator.SwapDone || len(started) != 1 {
		t.Fatalf("calls %+v started %v: repeated nudges re-queued or superseded the running attempt", calls, started)
	}
}

// wakeHost makes host a sleeper whose wake is in flight: no current
// generation, but the activator owns its entry, as it does from the wake's
// plan to its outcome.
func (h *harness) wakeHost(host string) {
	h.act.mu.Lock()
	defer h.act.mu.Unlock()
	delete(h.act.gens, host)
	h.act.owned[host] = true
}

// I1 (Task 16's simulation, seeds 8, 70 and 102): a wake planned onto build
// B, a sibling's failure of B recorded while the wake is still on its way to
// the exec, then a promotion that moves every route on the box off B. The
// next Pass must not prune B's record: the plan's check reads only the book,
// so a pruned record let the wake boot the failed B, in hold mode against
// its database. The record goes once nothing planned still carries B: the
// wake's outcome, a newer plan for the host, or the activator letting the
// host go.
func TestAPlannedWakeKeepsItsBuildsFailureRecordThroughAPrune(t *testing.T) {
	sibling := FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0}
	setup := func(t *testing.T) (*harness, activator.WakePlan) {
		h := newHarness(t, "a")
		h.stage("v2", "v3")
		h.install("v2", "a")
		h.wakeHost("a")
		p := h.m.Plan(context.Background(), h.rows.get("a"))
		if !p.FirstWake || p.Boot.Version != "v2" || p.Check == nil {
			t.Fatalf("plan %+v: want a first wake onto v2 with its check", p)
		}
		if _, err := h.book.Record(sibling); err != nil {
			t.Fatal(err)
		}
		h.install("v3", "a") // the promotion: no route on the box installs v2 now
		h.pass()
		if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
			t.Fatal("the Pass pruned v2's record while a planned wake still carried v2")
		}
		if err := p.Check("v2"); err == nil {
			t.Fatal("the wake's check let v2 boot: it failed on a sibling after the plan")
		}
		return h, p
	}
	pruned := func(t *testing.T, h *harness, why string) {
		t.Helper()
		h.pass()
		if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
			t.Fatalf("%s, and the next Pass still kept v2's record", why)
		}
	}
	t.Run("the wake's outcome", func(t *testing.T) {
		h, p := setup(t)
		h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), FirstWake: true, Token: p.Token})
		pruned(t, h, "the wake reported its outcome")
	})
	t.Run("a newer plan", func(t *testing.T) {
		h, _ := setup(t)
		if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Boot.Version != "v3" {
			t.Fatalf("replanned %+v, want v3", p)
		}
		pruned(t, h, "the host's newer plan boots v3")
	})
	t.Run("the activator let the host go", func(t *testing.T) {
		h, _ := setup(t)
		h.act.mu.Lock()
		delete(h.act.owned, "a") // the wake ended with no event: refused at its lease
		h.act.mu.Unlock()
		pruned(t, h, "the activator no longer owns the host")
	})
}

// blockDemand makes the demand read for ver block (once per call to it)
// until the returned release is called. blocked closes when the read starts.
func (h *harness) blockDemand(ver string) (blocked <-chan struct{}, release func()) {
	in, out := make(chan struct{}), make(chan struct{})
	var n atomic.Int32
	h.mu.Lock()
	h.demandHook = func(v string) {
		if v == ver && n.Add(1) == 1 {
			close(in)
			<-out
		}
	}
	h.mu.Unlock()
	var once atomic.Bool
	return in, func() {
		if once.CompareAndSwap(false, true) {
			close(out)
		}
	}
}

// planAsync runs Plan for host on its own goroutine, as the activator's wake
// does, and returns its plan on the channel.
func (h *harness) planAsync(rt registry.Route) <-chan activator.WakePlan {
	done := make(chan activator.WakePlan, 1)
	go func() { done <- h.m.Plan(context.Background(), rt) }()
	return done
}

func waitClosed(t *testing.T, what string, c <-chan struct{}) {
	t.Helper()
	select {
	case <-c:
	case <-time.After(5 * time.Second):
		t.Fatalf("timed out waiting for %s", what)
	}
}

// I1, the planner's own window (Task 10 round 3): Plan reads the book, then
// makes its slow reads (the canonical demand, the failing pointer, the
// policy) before it returns. A sibling's failure of B recorded while Plan is
// in those reads, then a promotion that moves the box off B, then a Pass:
// had Plan registered B only at its return, the Pass would prune B's record
// (no route installs B, nothing tracks it) and the exec would boot the failed
// B. Plan registers B before its reads, so the record stands, and it reads
// the row again after them, so a wake whose row moved is planned against the
// fresh row.
func TestAPlanInItsSlowReadsKeepsItsBuildsFailureRecord(t *testing.T) {
	sibling := FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0}
	notBootable := func(t *testing.T, p activator.WakePlan) {
		t.Helper()
		if p.Boot.Version == "v2" && (p.Check == nil || p.Check("v2") == nil) {
			t.Fatalf("plan %+v: the exec may boot v2, which failed on a sibling during the plan", p)
		}
	}

	t.Run("the route moves during the reads", func(t *testing.T) {
		h := newHarness(t, "a")
		h.stage("v2", "v3")
		h.install("v2", "a")
		h.wakeHost("a")
		blocked, release := h.blockDemand("v2")
		defer release()
		done := h.planAsync(h.rows.get("a"))
		waitClosed(t, "Plan to read v2's demand", blocked)
		if _, err := h.book.Record(sibling); err != nil {
			t.Fatal(err)
		}
		h.install("v3", "a") // the promotion: no route on the box installs v2 now
		h.pass()
		_, kept := h.book.Lookup("bab", "blog", "v2")
		release()
		p := <-done
		notBootable(t, p)
		if !kept {
			t.Fatal("the Pass pruned v2's record while Plan was still planning v2")
		}
		if !p.FirstWake || p.Boot.Version != "v3" {
			t.Fatalf("plan %+v: the row moved to v3 during the plan, want a first wake onto v3", p)
		}
	})

	t.Run("the row was stale before the plan", func(t *testing.T) {
		// The activator read the row (v2) before a promotion landed; the
		// sibling's record was pruned before Plan began. Plan's registration
		// comes too late for that prune, and the read of the row after the
		// slow reads catches it.
		h := newHarness(t, "a")
		h.stage("v2", "v3")
		h.install("v2", "a")
		h.wakeHost("a")
		stale := h.rows.get("a")
		if _, err := h.book.Record(sibling); err != nil {
			t.Fatal(err)
		}
		h.install("v3", "a")
		h.pass()
		if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
			t.Fatal("setup: nothing installs or plans v2, so the Pass should have pruned it")
		}
		p := h.m.Plan(context.Background(), stale)
		if p.Boot.Version != "v3" {
			t.Fatalf("plan %+v from a stale row: want the fresh row's v3", p)
		}
	})

	t.Run("the route moves twice", func(t *testing.T) {
		// The row moves during the first plan and again during the replan:
		// the plan refuses rather than chase the row.
		h := newHarness(t, "a")
		h.stage("v2", "v3", "v4")
		h.install("v2", "a")
		h.wakeHost("a")
		blocked, release := h.blockDemand("v2")
		defer release()
		done := h.planAsync(h.rows.get("a"))
		waitClosed(t, "Plan to read v2's demand", blocked)
		h.install("v3", "a")
		blocked3, release3 := h.blockDemand("v3")
		defer release3()
		release()
		waitClosed(t, "the replan to read v3's demand", blocked3)
		h.install("v4", "a")
		release3()
		if p := <-done; p.Refuse == "" || p.Boot.StagedPath != "" {
			t.Fatalf("plan %+v: the row moved twice during the plan, want refused", p)
		}
	})
}