package activator

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/registry"
)

// plantMember puts a generation into host's operation set the way Swap does
// (Task 4). A successor waits for the barrier's cancel and is then stopped by
// its owner. A retiring generation is stopped by its owner at once.
func plantMember(t *testing.T, a *Service, host string, role genRole, p *fakeProc, n uint64) *genMember {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	m := &genMember{gen: Gen{N: n, Build: "v2"}, role: role, proc: p, done: make(chan struct{})}
	if role != roleRetiring {
		m.cancel = cancel
	}
	a.mu.Lock()
	e := a.entries[host]
	if e == nil {
		a.mu.Unlock()
		t.Fatalf("no entry for %s", host)
	}
	a.addMemberLocked(e, m)
	a.mu.Unlock()
	go func() {
		if role != roleRetiring {
			<-ctx.Done()
		}
		a.dropMember(e, m)
	}()
	return m
}

type barrierEnv struct {
	a    *Service
	b    *fakeBackend
	clk  *fakeClock
	host string
}

func awakeForBarrier(t *testing.T, swap bool) barrierEnv {
	t.Helper()
	a, b, clk, _ := testActivator(t, time.Hour)
	a.gens = swap
	host := "barrier.example.com"
	arrangeAwake(t, a, testRoute(host))
	return barrierEnv{a, b, clk, host}
}

// stopSignal is an EventSink that closes began when an entry's stop emits
// "instance stopping": the stop goroutine has started, and from there it
// reaches the barrier without blocking on anything else.
type stopSignal struct {
	once  sync.Once
	began chan struct{}
}

func (s *stopSignal) App(string, string, string, string) {}
func (s *stopSignal) Event(_, _, _, _, msg string) {
	if msg == "instance stopping" {
		s.once.Do(func() { close(s.began) })
	}
}

// Every terminal stop cancels and joins the whole operation set before the
// final checkpoint and the lease release (spec §7.4). The member's stop is
// held open by a gate; until it closes, neither the instance's own stop
// nor the lease release may happen.
func TestSwapWitnessBarrierJoinsBeforeTheLease(t *testing.T) {
	t.Parallel()
	stops := []struct {
		name   string
		stop   func(x barrierEnv)
		rewake bool
	}{
		{"hibernate", func(x barrierEnv) { x.clk.advance(2 * time.Hour); x.a.sweepOnce() }, true},
		{"lease lost", func(x barrierEnv) { x.b.lastRep().setErr(errors.New("renewal failed")); x.a.sweepOnce() }, true},
		{"restart", func(x barrierEnv) { x.a.Bounce(x.host) }, true},
		{"maintenance fence", func(x barrierEnv) {
			go func() {
				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
				defer cancel()
				x.a.FenceAndStop(ctx, x.host, 1) //nolint:errcheck
			}()
		}, false},
		{"move or park retire", func(x barrierEnv) { x.a.RetireExcept(map[string]bool{}) }, true},
		{"shutdown", func(x barrierEnv) { go x.a.Shutdown(context.Background()) }, false},
	}
	for _, tc := range stops {
		for _, role := range []genRole{roleSuccessor, roleRetiring} {
			t.Run(tc.name+"/"+role.String(), func(t *testing.T) {
				x := awakeForBarrier(t, true)
				sig := &stopSignal{began: make(chan struct{})}
				x.a.Logs = sig
				joining := make(chan struct{})
				var joinOnce sync.Once
				x.a.mu.Lock()
				x.a.joinWaiting = func(_, what string) {
					if what == "operation set" {
						joinOnce.Do(func() { close(joining) })
					}
				}
				x.a.mu.Unlock()
				gate := make(chan struct{})
				member := &fakeProc{label: "gen:9", rec: x.b, stopGate: gate}
				plantMember(t, x.a, x.host, role, member, 9)
				incumbent, rep := x.b.lastProc(), x.b.lastRep()
				tc.stop(x)
				select {
				case <-sig.began:
				case <-time.After(5 * time.Second):
					t.Fatal("the stop never began")
				}
				// The stop has begun. Wait for it to reach the barrier (the
				// join seam's positive signal), or, if something skipped the
				// barrier, for the incumbent's stop or the lease release.
				waitFor(t, "the stop at the barrier", func() bool {
					select {
					case <-joining:
						return true
					default:
					}
					return rep.isClosed() || incumbent.stopEntered() > 0
				})
				if rep.isClosed() || incumbent.stopEntered() > 0 {
					t.Fatal("the stop went past the barrier while a generation was still draining")
				}
				close(gate)
				waitFor(t, "the lease release", rep.isClosed)
				ev := x.b.events()
				if i, j := slices.Index(ev, "stop:gen:9"), slices.Index(ev, "stop:instance"); i < 0 || j < 0 || i > j {
					t.Fatalf("order %v: the member must stop before the instance", ev)
				}
				if tc.rewake {
					// #231: Owns stays true while an entry drains, so the stop is
					// proven by a fresh spawn instead.
					arrangeAwake(t, x.a, testRoute(x.host))
					if got := x.b.startCount(); got != 2 {
						t.Fatalf("Start calls = %d after the stop, want 2", got)
					}
				}
			})
		}
	}
}

// A member whose termination cannot be proven (ErrStillRunning) keeps the
// lease, exactly as an unproven main process does (spec §7.4). The reaper
// finishes the member once it exits, and then the stop completes.
func TestBarrierKeepsTheLeaseWhileAGenerationIsUnproven(t *testing.T) {
	x := awakeForBarrier(t, true)
	old := opJoinBound
	opJoinBound = 100 * time.Millisecond
	t.Cleanup(func() { opJoinBound = old })
	member := &fakeProc{label: "gen:9", rec: x.b, stuck: true}
	plantMember(t, x.a, x.host, roleRetiring, member, 9)
	waitFor(t, "the member marked unproven", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		m := x.a.entries[x.host].ops.members[9]
		return m != nil && m.unproven
	})
	rep := x.b.lastRep()
	x.a.Bounce(x.host)
	waitFor(t, "the stop to pend", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		e := x.a.entries[x.host]
		return e != nil && e.pendingStop
	})
	if rep.isClosed() {
		t.Fatal("the lease was released over an unproven generation")
	}
	member.exit()
	x.a.sweepOnce() // reaps the member
	waitFor(t, "the member reaped", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		e := x.a.entries[x.host]
		return e == nil || len(e.ops.members) == 0
	})
	x.a.sweepOnce() // finishes the stop
	waitFor(t, "the lease release", rep.isClosed)
	arrangeAwake(t, x.a, testRoute(x.host))
	if got := x.b.startCount(); got != 2 {
		t.Fatalf("Start calls = %d, want 2", got)
	}
}

// killGenBackend is fakeBackend plus plan 2's KillGen. err, while set, is
// what every kill answers: a generation child that still holds a process.
type killGenBackend struct {
	*fakeBackend
	kmu   sync.Mutex
	err   error
	kills []uint64
}

func (k *killGenBackend) KillGen(_ context.Context, _ registry.Route, n uint64) error {
	k.kmu.Lock()
	defer k.kmu.Unlock()
	k.kills = append(k.kills, n)
	return k.err
}

func (k *killGenBackend) setErr(err error) {
	k.kmu.Lock()
	k.err = err
	k.kmu.Unlock()
}

func (k *killGenBackend) killed() []uint64 {
	k.kmu.Lock()
	defer k.kmu.Unlock()
	return slices.Clone(k.kills)
}

// A generation's leader can exit while a process it forked lives on in its
// child cgroup. The leader's exit is not termination (spec §7.3, §7.4).
// After the graceful drain the member's own child is killed and proven
// empty through plan 2's KillGen. While that proof fails, the member stays
// in the set and the lease stays held, exactly as for ErrStillRunning. The
// reaper retries the kill and finishes the stop once the child is empty.
//
// Without the KillGen step, stopMember accepts Exited() alone: the member
// leaves the set at once, "marked unproven" never happens, and the Bounce
// releases the lease over the surviving descendant.
func TestSwapWitnessMemberStopProvesTheGenerationChildEmpty(t *testing.T) {
	x := awakeForBarrier(t, true)
	old := opJoinBound
	opJoinBound = 100 * time.Millisecond
	t.Cleanup(func() { opJoinBound = old })
	kb := &killGenBackend{fakeBackend: x.b, err: errors.New("g9 still holds a process")}
	x.a.backend = kb
	member := &fakeProc{label: "gen:9", rec: x.b} // its leader exits on SIGTERM
	plantMember(t, x.a, x.host, roleRetiring, member, 9)
	waitFor(t, "the member marked unproven", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		m := x.a.entries[x.host].ops.members[9]
		return m != nil && m.unproven
	})
	if !member.isStopped() || !member.Exited() {
		t.Fatal("the graceful stop never ran, or the leader did not exit")
	}
	if got := kb.killed(); !slices.Equal(got, []uint64{9}) {
		t.Fatalf("KillGen calls = %v, want [9] (the member's own generation)", got)
	}
	rep := x.b.lastRep()
	x.a.Bounce(x.host)
	waitFor(t, "the stop to pend", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		e := x.a.entries[x.host]
		return e != nil && e.pendingStop
	})
	if rep.isClosed() {
		t.Fatal("the lease was released while generation 9's child still held a process")
	}
	x.a.sweepOnce() // the reaper retries the kill, which still fails
	waitFor(t, "the reaper's retry to finish", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		m := x.a.entries[x.host].ops.members[9]
		return len(kb.killed()) >= 2 && m != nil && !m.reaping
	})
	if rep.isClosed() {
		t.Fatal("the lease was released on a failed retry")
	}
	kb.setErr(nil)
	x.a.sweepOnce() // the kill proves the child empty now
	waitFor(t, "the member reaped", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		e := x.a.entries[x.host]
		return e == nil || len(e.ops.members) == 0
	})
	x.a.sweepOnce() // finishes the stop
	waitFor(t, "the lease release", rep.isClosed)
}

// A backend with no generation children (containment off answers
// errNoGenContainment) has nothing more to prove than the leader's exit:
// the generation shares the host cgroup, whose emptiness the terminal
// stop's own evidence proves.
func TestMemberStopWithoutContainmentAcceptsTheLeaderExit(t *testing.T) {
	x := awakeForBarrier(t, true)
	kb := &killGenBackend{fakeBackend: x.b, err: fmt.Errorf("wrapped: %w", errNoGenContainment)}
	x.a.backend = kb
	m := plantMember(t, x.a, x.host, roleRetiring, &fakeProc{label: "gen:9", rec: x.b}, 9)
	select {
	case <-m.done:
	case <-time.After(5 * time.Second):
		t.Fatal("the member never left the set")
	}
}

// A member that never got a process (its spawn failed with its kill
// unproven, as startUnitAt answers) is not proven by the missing handle:
// its generation's child is, through KillGen (P2-13 (f)).
func TestMemberStopWithoutAProcessProvesItsChild(t *testing.T) {
	x := awakeForBarrier(t, true)
	kb := &killGenBackend{fakeBackend: x.b, err: errors.New("g12 still holds a process")}
	x.a.backend = kb
	m := &genMember{gen: Gen{N: 12, Build: "v2"}, role: roleSpawning, done: make(chan struct{})}
	x.a.mu.Lock()
	e := x.a.entries[x.host]
	x.a.addMemberLocked(e, m)
	x.a.mu.Unlock()
	if x.a.dropMember(e, m) {
		t.Fatal("a member with no process and an unproven child was taken as terminated")
	}
	if got := kb.killed(); !slices.Equal(got, []uint64{12}) {
		t.Fatalf("KillGen calls = %v, want [12]", got)
	}
	kb.setErr(nil)
	x.a.sweepOnce()
	select {
	case <-m.done:
	case <-time.After(5 * time.Second):
		t.Fatal("the member never left once its child was empty")
	}
}

// stubbornProc is a main process whose Stop fails with an error other than
// ErrStillRunning (a claims-on KillAndEmpty error) and whose leader lives
// on until exit. stops counts Stop calls.
type stubbornProc struct {
	exited atomic.Bool
	stops  atomic.Int32
}

func (p *stubbornProc) Stop(context.Context) error {
	p.stops.Add(1)
	return errors.New("cgroup.kill: the child is not empty")
}
func (p *stubbornProc) Exited() bool  { return p.exited.Load() }
func (p *stubbornProc) Crashed() bool { return false }

// THE retain-until-proven rule (spec §7.4): a stop whose main process is
// still alive after a failed kill keeps its lease. Only ErrStillRunning
// kept it before; any other Stop error, with the leader alive, skipped the
// proof (it ran only for an exited leader) and released the lease.
func TestAStopKeepsTheLeaseWhileItsLeaderLivesOn(t *testing.T) {
	x := awakeForBarrier(t, true)
	p := &stubbornProc{}
	x.a.mu.Lock()
	x.a.entries[x.host].proc = p
	x.a.mu.Unlock()
	rep := x.b.lastRep()
	x.a.Bounce(x.host)
	waitFor(t, "the stop to pend", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		e := x.a.entries[x.host]
		return e != nil && e.pendingStop
	})
	// The leader is alive and the retry cadence has not come round (the
	// clock is fake), so this sweep starts nothing: the stop stays pending.
	x.a.sweepOnce()
	x.a.mu.Lock()
	e := x.a.entries[x.host]
	idle := e != nil && e.pendingStop && !e.genReaping && !e.stopRetrying
	x.a.mu.Unlock()
	if !idle || rep.isClosed() {
		t.Fatal("the lease was released while the leader still ran after a failed kill")
	}
	p.exited.Store(true)
	x.a.sweepOnce()
	waitFor(t, "the lease release", rep.isClosed)
}

// A stop keeps its lease while a stray sidecar of the host is tracked
// (spec §7.4): its process may still run against the database. The lease
// went with the stop before.
func TestAStopKeepsTheLeaseWhileAStraySidecarRuns(t *testing.T) {
	t.Parallel()
	x := awakeForBarrier(t, true)
	stray := &fakeProc{label: "sidecar", rec: x.b, stuck: true}
	x.a.mu.Lock()
	x.a.straySidecars = map[string][]*straySidecar{x.host: {{proc: stray, rt: testRoute(x.host)}}}
	x.a.mu.Unlock()
	rep := x.b.lastRep()
	x.a.Bounce(x.host)
	waitFor(t, "the stop to pend", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		e := x.a.entries[x.host]
		return e != nil && e.pendingStop
	})
	if rep.isClosed() {
		t.Fatal("the lease was released while a stray sidecar ran")
	}
	x.a.mu.Lock()
	delete(x.a.straySidecars, x.host) // proven gone (Task 7's reaper does this)
	x.a.mu.Unlock()
	x.a.sweepOnce()
	waitFor(t, "the lease release", rep.isClosed)
}

// failingContainment is a claims-on child whose KillAndEmpty fails until
// heal is called.
type failingContainment struct {
	mu  sync.Mutex
	bad bool
}

func (c *failingContainment) KillAndEmpty(context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.bad {
		return errors.New("cgroup.kill: the child is not empty")
	}
	return nil
}
func (c *failingContainment) Close() error { return nil }
func (c *failingContainment) heal() {
	c.mu.Lock()
	c.bad = false
	c.mu.Unlock()
}

// A claims-on generation is terminated only when its own Stop completed
// (spec §7.4): KillAndEmpty on its child included. Here the leader has
// exited but KillAndEmpty fails, so the member stays in the set, unproven,
// until the child is empty. Treating any claimProcess as proven accepted
// the leader's exit alone.
func TestAClaimsOnMemberNeedsItsChildEmptied(t *testing.T) {
	x := awakeForBarrier(t, true)
	cmd := exec.Command("true")
	if err := cmd.Run(); err != nil {
		t.Skipf("no true(1): %v", err)
	}
	done := make(chan struct{})
	close(done) // the leader has exited
	cont := &failingContainment{bad: true}
	p := &claimProcess{cmd: cmd, prepared: cont, done: done}
	m := &genMember{gen: Gen{N: 11, Build: "v2"}, role: roleRetiring, proc: p, done: make(chan struct{})}
	x.a.mu.Lock()
	e := x.a.entries[x.host]
	x.a.addMemberLocked(e, m)
	x.a.mu.Unlock()
	if x.a.dropMember(e, m) {
		t.Fatal("a claims-on generation whose KillAndEmpty failed was taken as terminated")
	}
	cont.heal()
	x.a.sweepOnce()
	select {
	case <-m.done:
	case <-time.After(5 * time.Second):
		t.Fatal("the member never left once its child was empty")
	}
}

// The current generation is held to the same proof (spec §7.4): a terminal
// stop whose main process has exited keeps the lease while that
// generation's child is not proven empty, and finishes once it is. A stop
// that accepts the leader's exit alone releases the lease over a surviving
// descendant. This is the witness for mutation M8.
func TestSwapWitnessStopProvesTheCurrentGenerationEmpty(t *testing.T) {
	t.Parallel()
	x := awakeForBarrier(t, true)
	kb := &killGenBackend{fakeBackend: x.b, err: errors.New("the child still holds a process")}
	x.a.backend = kb
	x.a.mu.Lock()
	x.a.entries[x.host].gen.N = 1 << 41 // the woken generation, as plan 1's wake records it
	x.a.mu.Unlock()
	rep := x.b.lastRep()
	x.a.Bounce(x.host)
	waitFor(t, "the stop to pend", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		e := x.a.entries[x.host]
		return e != nil && e.pendingStop
	})
	if !x.b.lastProc().isStopped() {
		t.Fatal("the graceful stop never ran")
	}
	kills := len(kb.killed())
	x.a.sweepOnce() // retries the proof, which still fails
	waitFor(t, "the retried proof to finish", func() bool {
		x.a.mu.Lock()
		defer x.a.mu.Unlock()
		e := x.a.entries[x.host]
		return len(kb.killed()) > kills && e != nil && !e.genReaping
	})
	if rep.isClosed() {
		t.Fatal("the lease was released while the current generation's child still held a process")
	}
	kb.setErr(nil)
	x.a.sweepOnce()
	waitFor(t, "the lease release", rep.isClosed)
	if k := kb.killed(); len(k) == 0 || k[len(k)-1] != 1<<41 {
		t.Fatalf("KillGen calls = %v, want the current generation's", k)
	}
}

// With swaps on, Shutdown covers a waking entry (spec §7.4): it cancels the
// wake and waits for it. Its child is stopped and its lease given back
// before Shutdown returns.
func TestSwapWitnessShutdownJoinsAWakingEntry(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	a.gens = true
	gate := make(chan struct{})
	b.startGate = gate
	host := "waking.example.com"
	go func() {
		if release, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
			release()
		}
	}()
	waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 1 })
	done := make(chan struct{})
	go func() { a.Shutdown(context.Background()); close(done) }()
	select {
	case <-done:
		t.Fatal("Shutdown returned while a wake was still spawning")
	case <-time.After(100 * time.Millisecond):
	}
	close(gate)
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("Shutdown never returned")
	}
	if !b.lastProc().isStopped() {
		t.Fatal("the woken child outlived Shutdown")
	}
	if got := b.releaseCount(); got != 1 {
		t.Fatalf("lease releases = %d, want 1", got)
	}
	// Shutdown cancelled the wake's context. The release must not run under
	// it: a real release is an S3 DeleteObject that would fail at once and
	// leave the lease held for its whole TTL.
	for _, err := range b.releaseCtxErrors() {
		if err != nil {
			t.Fatalf("the lease was released under a dead context (%v)", err)
		}
	}
}

// Every doWake bail-out gives the lease back under a live context, even
// after the wake's own context is gone (a fence, a Shutdown). Here a fence
// lands while Start is held, so the wake's context is cancelled before it
// releases. Flag off: the fix is not gated on generations.
func TestAFencedWakeReleasesUnderALiveContext(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	gate := make(chan struct{})
	b.startGate = gate
	host := "fenced-wake.example.com"
	failed := make(chan error, 1)
	go func() {
		release, err := a.beginRequest(context.Background(), testRoute(host))
		if err == nil {
			release()
		}
		failed <- err
	}()
	waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 1 })
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	fenceDone := make(chan struct{})
	go func() {
		a.FenceAndStop(ctx, host, 1) //nolint:errcheck
		close(fenceDone)
	}()
	waitFor(t, "the fence", func() bool { return a.Fenced(host) })
	close(gate)
	if err := <-failed; err == nil {
		t.Fatal("a fenced wake succeeded")
	}
	<-fenceDone
	if got := b.releaseCount(); got != 1 {
		t.Fatalf("lease releases = %d, want 1", got)
	}
	for _, err := range b.releaseCtxErrors() {
		if err != nil {
			t.Fatalf("the lease was released under a dead context (%v)", err)
		}
	}
}

// Flag off: Shutdown joins a waking entry exactly as it does with swaps on
// (#292). It used to leave one to resolve on its own, and the agent then
// exited with the wake still holding its lease. Here the start ignores its
// cancelled context, so Shutdown must wait it out, stop what it woke and
// give the lease back under a live context.
func TestShutdownJoinsAWakingEntryWithSwapOff(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	gate := make(chan struct{})
	b.startGate = gate
	host := "waking.example.com"
	go func() {
		if release, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
			release()
		}
	}()
	waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 1 })
	done := make(chan struct{})
	go func() { a.Shutdown(context.Background()); close(done) }()
	select {
	case <-done:
		t.Fatal("Shutdown returned while a wake was still spawning, with swaps off")
	case <-time.After(100 * time.Millisecond):
	}
	close(gate)
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("Shutdown never returned")
	}
	if !b.lastProc().isStopped() {
		t.Fatal("the woken child outlived Shutdown")
	}
	if got := b.releaseCount(); got != 1 {
		t.Fatalf("lease releases = %d, want 1", got)
	}
	for _, err := range b.releaseCtxErrors() {
		if err != nil {
			t.Fatalf("the lease was released under a dead context (%v)", err)
		}
	}
	if n := a.entryCount(); n != 0 {
		t.Fatalf("Shutdown left %d entr(ies) behind", n)
	}
}

// execGenBackend wakes through a real ExecBackend's StartGen (a placed flat
// spawn under GenContainment, in a delegated cgroup tree of plain files)
// while the fake keeps the lease, restore and replica. Both kills, the
// spawn path's own (ExecBackend.killGen) and the reaper's (KillGen), answer
// the embedded killGenBackend's err.
type execGenBackend struct {
	*killGenBackend
	x    *ExecBackend
	gmu  sync.Mutex
	gens []uint64
}

func (g *execGenBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
	g.gmu.Lock()
	g.gens = append(g.gens, n)
	g.gmu.Unlock()
	return g.x.StartGen(ctx, rt, n, perGen)
}

func (g *execGenBackend) NextDueVia(ctx context.Context, rt registry.Route, _ string) (int64, error) {
	return g.fakeBackend.NextDue(ctx, rt)
}

func (g *execGenBackend) firstGen() uint64 {
	g.gmu.Lock()
	defer g.gmu.Unlock()
	return g.gens[0]
}

// A placed flat generation whose health wait fails is stopped by its
// leader only, and a process it forked can live on in g<n> (spec §7.3).
// startChildAt kills g<n> after the leader's stop. When that kill fails,
// the process goes back with an error wrapping errGenUnproven and the wake
// keeps its lease until the sweep proves the child empty. When it
// succeeds, nothing is left and the lease goes back at once. With
// containment off nothing is placed and startChildAt makes no kill. Its
// handle now comes back with the health error (Task 15 fix round 1, ruling
// a), so the failed-wake cleanup's proof asks KillGen once, which answers
// errNoGenContainment as ExecBackend's does with containment off; this
// backend has no host kill, so the lease goes back.
func TestAPlacedSpawnWithAFailedHealthWaitProvesItsChild(t *testing.T) {
	old := spawnHealthTimeout
	spawnHealthTimeout = 300 * time.Millisecond
	t.Cleanup(func() { spawnHealthTimeout = old })
	rows := []struct {
		name        string
		containment bool
		killErr     error
		wantKept    bool
	}{
		{"placed, kill fails", true, errors.New("g still holds a process"), true},
		{"placed, kill proves the child empty", true, nil, false},
		{"containment off", false, errNoGenContainment, false},
	}
	for _, row := range rows {
		t.Run(row.name, func(t *testing.T) {
			host := "placed.example.com"
			x, _ := delegatedBackend(t, host)
			cgroup.FakePlacementForTest(t) // a temp dir is no cgroupfs, and a generation is never moved by PID
			x.GenContainment = row.containment
			x.RunDirOverride = shortRunDir(t)
			stageLive(t, x, host, "v1") // exits at once without binding: the health wait fails
			a, b, _, _ := testActivator(t, time.Hour)
			a.gens = true
			kb := &killGenBackend{fakeBackend: b, err: row.killErr}
			gb := &execGenBackend{killGenBackend: kb, x: x}
			x.killGen = kb.KillGen
			a.backend = gb
			rt := flatGenRoute(x, host)

			_, err := a.beginRequest(context.Background(), rt)
			if err == nil {
				t.Fatal("a wake whose health wait failed succeeded")
			}
			n := gb.firstGen()
			a.mu.Lock()
			_, kept := a.unprovenWakes[host]
			a.mu.Unlock()
			if kept != row.wantKept {
				t.Fatalf("retained = %v, want %v (err %v)", kept, row.wantKept, err)
			}
			if !row.containment {
				if k := kb.killed(); !slices.Equal(k, []uint64{n}) {
					t.Fatalf("KillGen calls = %v with containment off, want only the cleanup's proof of generation %d", k, n)
				}
				if got := b.releaseCount(); got != 1 {
					t.Fatalf("lease releases = %d, want 1", got)
				}
				return
			}
			if k := kb.killed(); len(k) == 0 || k[0] != n {
				t.Fatalf("KillGen calls = %v, want generation %d's first", k, n)
			}
			if !row.wantKept {
				if got := b.releaseCount(); got != 1 {
					t.Fatalf("lease releases = %d after a proven kill, want 1", got)
				}
				return
			}
			if got := b.releaseCount(); got != 0 {
				t.Fatalf("lease releases = %d while g%d is unproven, want 0", got, n)
			}
			kb.setErr(nil)
			a.sweepOnce()
			waitFor(t, "the lease release", func() bool { return b.releaseCount() == 1 })
		})
	}
}

// unprovenStartBackend is a GenBackend whose next StartGen fails the way a
// generation start fails when its kill cannot be proven (plan 2,
// startUnitAt): fail, while set, answers that one call. Every later call
// spawns through the fake's Start. KillGen comes from killGenBackend.
type unprovenStartBackend struct {
	*killGenBackend
	smu  sync.Mutex
	fail func(n uint64) (Proc, error)
	gens []uint64 // every generation number StartGen was handed
}

func (u *unprovenStartBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, _ bool) (Proc, Gen, error) {
	u.smu.Lock()
	fail := u.fail
	u.fail = nil
	u.gens = append(u.gens, n)
	u.smu.Unlock()
	if fail != nil {
		p, err := fail(n)
		return p, Gen{N: n}, err
	}
	p, ver, err := u.fakeBackend.Start(ctx, rt)
	return p, canonicalGen(rt, n, ver), err
}

func (u *unprovenStartBackend) NextDueVia(ctx context.Context, rt registry.Route, _ string) (int64, error) {
	return u.fakeBackend.NextDue(ctx, rt)
}

func (u *unprovenStartBackend) firstGen() uint64 {
	u.smu.Lock()
	defer u.smu.Unlock()
	return u.gens[0]
}

var (
	_ GenBackend = (*unprovenStartBackend)(nil)
	_ genKiller  = (*unprovenStartBackend)(nil)
)

// P2-13 (plan 2's parked L1): a generation start that failed with its kill
// unproven may still be running. The wake keeps the lease (spec §7.4,
// "unproven, the lease is kept"), refuses the host's next wake, and gives
// the lease back only once the sweep proves the generation's child empty
// through KillGen. Before, a start that returned no process released the
// lease at once, and one whose Stop failed released it anyway.
//
// The sentinel rows hold with generations off too: errGenUnproven and
// instancectl.ErrGenUnproven exist only under plan 2's GenContainment,
// which can be on while swaps are off. This is the witness for M9.
func TestSwapWitnessUnprovenWakeKeepsTheLease(t *testing.T) {
	t.Parallel()
	lost := errors.New("the unit exited before it was observed")
	rows := []struct {
		name   string
		gens   bool
		leader *stubbornProc // the row's process, alive until told to exit; nil: none
		err    error
	}{
		{"errGenUnproven with no process", true, nil,
			fmt.Errorf("start unit: %w; %w: kill timed out", lost, errGenUnproven)},
		{"instancectl.ErrGenUnproven with no process", true, nil,
			fmt.Errorf("h: start unit: %w", fmt.Errorf("%w: %w", instancectl.ErrGenUnproven, lost))},
		{"a failed stop while the leader lives", true, &stubbornProc{},
			errors.New("health: no answer within 30s")},
		{"generations off, sentinel present", false, nil,
			fmt.Errorf("start unit: %w; %w: kill timed out", lost, errGenUnproven)},
	}
	for _, row := range rows {
		t.Run(row.name, func(t *testing.T) {
			a, b, _, _ := testActivator(t, time.Hour)
			a.gens = row.gens
			kb := &killGenBackend{fakeBackend: b, err: errors.New("the generation's child still holds a process")}
			ub := &unprovenStartBackend{killGenBackend: kb, fail: func(uint64) (Proc, error) {
				if row.leader != nil {
					return row.leader, row.err
				}
				return nil, row.err
			}}
			a.backend = ub
			host := "unproven.example.com"
			rt := testRoute(host)
			ctx := context.Background()
			retained := func() *unprovenWake {
				a.mu.Lock()
				defer a.mu.Unlock()
				return a.unprovenWakes[host]
			}
			settled := func() bool {
				a.mu.Lock()
				defer a.mu.Unlock()
				u := a.unprovenWakes[host]
				return u == nil || !u.busy
			}

			if _, err := a.beginRequest(ctx, rt); err == nil {
				t.Fatal("the wake should have failed")
			}
			if got := b.releaseCount(); got != 0 {
				t.Fatalf("lease releases = %d after a failed wake whose generation is unproven, want 0", got)
			}
			if retained() == nil {
				t.Fatal("the failed wake was not retained")
			}
			n := ub.firstGen()

			// The host's next wake is refused before it takes a lease.
			_, err := a.beginRequest(ctx, rt)
			if !errors.Is(err, errWakeUnproven) {
				t.Fatalf("second wake: err = %v, want errWakeUnproven", err)
			}
			if acq, _, _ := b.counts(); acq != 1 {
				t.Fatalf("Acquire calls = %d, want 1 (the refusal comes before the lease)", acq)
			}

			// A sweep whose proof still fails keeps the lease.
			a.sweepOnce()
			waitFor(t, "the retained wake's retry to finish", settled)
			if got := b.releaseCount(); got != 0 || retained() == nil {
				t.Fatalf("lease releases = %d after a failed proof, want 0 and the wake retained", got)
			}
			if row.leader != nil {
				if got := row.leader.stops.Load(); got < 2 {
					t.Fatalf("leader Stop calls = %d, want the sweep to have stopped it again", got)
				}
				// The leader exits, but its generation's child is still not
				// proven empty: the lease stays.
				row.leader.exited.Store(true)
				a.sweepOnce()
				waitFor(t, "the retained wake's retry to finish", settled)
				if got := b.releaseCount(); got != 0 || retained() == nil {
					t.Fatalf("lease releases = %d with the leader gone but the child unproven, want 0", got)
				}
			}

			kb.setErr(nil)
			a.sweepOnce()
			waitFor(t, "the lease release", func() bool { return b.releaseCount() == 1 })
			waitFor(t, "the retained wake dropped", func() bool { return retained() == nil })
			if k := kb.killed(); !slices.Contains(k, n) {
				t.Fatalf("KillGen calls = %v, want generation %d's", k, n)
			}

			release, err := a.beginRequest(ctx, rt)
			if err != nil {
				t.Fatalf("third wake: %v", err)
			}
			release()
			if got := b.startCount(); got != 1 {
				t.Fatalf("Start calls = %d, want 1 (the third wake)", got)
			}
			a.Bounce(host)
			waitFor(t, "cleanup", func() bool { return a.entryCount() == 0 })
		})
	}
}

// syncLog is a log destination that goroutines may write concurrently (the
// reapers log off the test's goroutine).
type syncLog struct {
	mu sync.Mutex
	b  strings.Builder
}

func (s *syncLog) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.b.Write(p)
}

func (s *syncLog) String() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.b.String()
}

// Shutdown covers a retained failed wake (spec §7.4, I2: nothing runs after
// Shutdown returns). No entry waits for one, so Shutdown polls its reaper
// until the generation is proven stopped and the lease given back, or the
// budget runs out, which it logs with the lease still held. Before, Shutdown
// saw no entries and returned at once, leaving the generation running and
// its lease held until the TTL. The retention has no generations gate (the
// sentinel exists only under GenContainment), so neither has the wait.
func TestSwapWitnessShutdownWaitsForAnUnprovenWake(t *testing.T) {
	for _, row := range []struct {
		name   string
		gens   bool
		budget time.Duration
		heal   bool
	}{
		{"healed within the budget", true, 10 * time.Second, true},
		{"healed within the budget, generations off", false, 10 * time.Second, true},
		{"the budget runs out", true, 300 * time.Millisecond, false},
	} {
		t.Run(row.name, func(t *testing.T) {
			a, b, _, _ := testActivator(t, time.Hour)
			a.gens = row.gens
			kb := &killGenBackend{fakeBackend: b, err: errors.New("the generation's child still holds a process")}
			a.backend = &unprovenStartBackend{killGenBackend: kb, fail: func(uint64) (Proc, error) {
				return nil, fmt.Errorf("start unit: health wait failed; %w: kill timed out", errGenUnproven)
			}}
			host := "unproven.example.com"
			retained := func() bool {
				a.mu.Lock()
				defer a.mu.Unlock()
				return a.unprovenWakes[host] != nil
			}
			if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
				t.Fatal("the wake should have failed")
			}
			if !retained() || b.releaseCount() != 0 {
				t.Fatalf("retained %v, lease releases %d: want the failed wake retained with its lease", retained(), b.releaseCount())
			}
			logs := &syncLog{}
			if !row.heal {
				flags := log.Flags()
				log.SetOutput(logs)
				t.Cleanup(func() { log.SetOutput(os.Stderr); log.SetFlags(flags) })
			}
			kills := len(kb.killed())

			shut := make(chan struct{})
			go func() {
				ctx, cancel := context.WithTimeout(context.Background(), row.budget)
				defer cancel()
				a.Shutdown(ctx)
				close(shut)
			}()
			returned := func() bool {
				select {
				case <-shut:
					return true
				default:
					return false
				}
			}
			// Two more failed proofs since Shutdown began: it is still
			// polling the reaper, so it has not returned while the wake is
			// retained.
			waitFor(t, "two failed proofs during Shutdown, or its return", func() bool {
				return len(kb.killed()) >= kills+2 || returned()
			})
			if returned() {
				t.Fatalf("Shutdown returned while the failed wake was retained (KillGen calls during it: %d)", len(kb.killed())-kills)
			}

			if row.heal {
				kb.setErr(nil)
			}
			select {
			case <-shut:
			case <-time.After(5 * time.Second):
				t.Fatal("Shutdown did not return")
			}
			if row.heal {
				if retained() || b.releaseCount() != 1 {
					t.Fatalf("after Shutdown: retained %v, lease releases %d; want the wake proven and its lease given back once", retained(), b.releaseCount())
				}
				return
			}
			if !retained() || b.releaseCount() != 0 {
				t.Fatalf("after the budget ran out: retained %v, lease releases %d; want the lease kept", retained(), b.releaseCount())
			}
			if got := logs.String(); !strings.Contains(got, host) || !strings.Contains(got, "shutdown") || !strings.Contains(got, "not proven stopped") {
				t.Fatalf("Shutdown's log does not name the retained wake:\n%s", got)
			}
		})
	}
}

// sigtermIgnorer is a leader that does not stop when asked: Stop answers
// ErrStillRunning, as unitProc does when SIGKILL could not prove the unit
// gone, until the process exits on its own (exit).
type sigtermIgnorer struct {
	exited atomic.Bool
	stops  atomic.Int32
}

func (p *sigtermIgnorer) Stop(context.Context) error {
	p.stops.Add(1)
	if p.exited.Load() {
		return nil
	}
	return ErrStillRunning
}
func (p *sigtermIgnorer) Exited() bool { return p.exited.Load() }

// fencedStartBackend is a GenBackend whose StartGen has spawned p but
// returns it only once gate opens: a fence lands between the spawn and the
// wake's post-spawn check. No KillGen: the leader's exit is the proof.
type fencedStartBackend struct {
	*fakeBackend
	p       *sigtermIgnorer
	entered chan struct{}
	gate    chan struct{}
}

func (s *fencedStartBackend) StartGen(_ context.Context, rt registry.Route, n uint64, _ bool) (Proc, Gen, error) {
	close(s.entered)
	<-s.gate
	return s.p, canonicalGen(rt, n, "v1"), nil
}

func (s *fencedStartBackend) NextDueVia(context.Context, registry.Route, string) (int64, error) {
	return 0, nil
}

var _ GenBackend = (*fencedStartBackend)(nil)

// A wake fenced after its spawn (the post-spawn ctx/wakeAllowed check) stops
// its process. If that process does not stop, the lease is kept until it is
// proven gone, with generations off too: a fail-closed dormancy exception
// (plan 3 task 2 fix), because releasing it would let another box take the
// host while this process still writes. Before, with generations off,
// cleanupFailedWake released the lease over the live process, as main does.
// The wake is retained (unprovenWakes): the host's next wake is refused,
// FenceAndStop reports the stop unproven, and the sweep gives the lease back
// once the leader has exited.
func TestSwapWitnessAFencedWakeKeepsItsLeaseWhileItsProcessRuns(t *testing.T) {
	t.Parallel()
	for _, gens := range []bool{false, true} {
		t.Run(fmt.Sprintf("generations %v", gens), func(t *testing.T) {
			a, b, _, _ := testActivator(t, time.Hour)
			a.gens = gens
			p := &sigtermIgnorer{}
			s := &fencedStartBackend{fakeBackend: b, p: p, entered: make(chan struct{}), gate: make(chan struct{})}
			a.backend = s
			host := "fenced.example.com"
			woke := make(chan error, 1)
			go func() {
				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
				defer cancel()
				_, err := a.beginRequest(ctx, testRoute(host))
				woke <- err
			}()
			<-s.entered
			fenced := make(chan error, 1)
			go func() {
				ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
				defer cancel()
				_, err := a.FenceAndStop(ctx, host, 1)
				fenced <- err
			}()
			waitFor(t, "the fence", func() bool { return !a.wakeAllowed(host) })
			close(s.gate)
			if err := <-woke; err == nil {
				t.Fatal("the fenced wake should have failed")
			}
			if p.stops.Load() == 0 {
				t.Fatal("the fenced wake did not stop its process")
			}
			if got := b.releaseCount(); got != 0 {
				t.Fatalf("lease releases = %d while the fenced wake's process runs, want 0", got)
			}
			retained := func() bool {
				a.mu.Lock()
				defer a.mu.Unlock()
				return a.unprovenWakes[host] != nil
			}
			settled := func() bool {
				a.mu.Lock()
				defer a.mu.Unlock()
				u := a.unprovenWakes[host]
				return u == nil || !u.busy
			}
			if !retained() {
				t.Fatal("the fenced wake was not retained")
			}
			if err := <-fenced; !errors.Is(err, ErrStopUnproven) {
				t.Fatalf("FenceAndStop = %v, want ErrStopUnproven while the process runs", err)
			}

			// A sweep while it still runs keeps the lease.
			a.sweepOnce()
			waitFor(t, "the retained wake's retry", settled)
			if got := b.releaseCount(); got != 0 || !retained() {
				t.Fatalf("lease releases = %d after a sweep with the process alive, want 0 and the wake retained", got)
			}

			// It exits; the next sweep proves it and gives the lease back.
			p.exited.Store(true)
			a.sweepOnce()
			waitFor(t, "the lease release", func() bool { return b.releaseCount() == 1 })
			waitFor(t, "the retained wake dropped", func() bool { return !retained() })
			a.LiftFence(host, 1)
		})
	}
}

// uncontainedStartBackend is the flag-off exec backend's shape for a
// retained failed wake (Task 15, Astra IMPL-3 F6): KillGen answers
// errNoGenContainment, as ExecBackend's does with GenContainment off, and
// KillHost proves carlos-exec@<host> empty, failing while populated.
type uncontainedStartBackend struct {
	*unprovenStartBackend
	hmu       sync.Mutex
	populated bool
	err       error // when set, every KillHost answers it
	hostKills int
}

func (h *uncontainedStartBackend) KillHost(context.Context, registry.Route) error {
	h.hmu.Lock()
	defer h.hmu.Unlock()
	h.hostKills++
	if h.err != nil {
		return h.err
	}
	if h.populated {
		return errors.New("carlos-exec@<host> still holds a process")
	}
	return nil
}

func (h *uncontainedStartBackend) setPopulated(on bool) {
	h.hmu.Lock()
	h.populated = on
	h.hmu.Unlock()
}

func (h *uncontainedStartBackend) hostKillCount() int {
	h.hmu.Lock()
	defer h.hmu.Unlock()
	return h.hostKills
}

var _ hostKiller = (*uncontainedStartBackend)(nil)

// Task 15, Astra IMPL-3 F6: with the flag off a retained failed wake was
// proven stopped by its leader's exit alone. KillGen answers
// errNoGenContainment (there is no g<n>), and genChildEmpty took that as
// proof, so the lease went back while a process the leader forked could
// still run in carlos-exec@<host>. The proof now also asks KillHost to
// empty the host cgroup: while it cannot, the wake stays retained with its
// lease, and once it can the lease goes back.
func TestARetainedFlagOffWakeProvesItsHostCgroupEmpty(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	a.gens = false
	leader := &sigtermIgnorer{}
	kb := &killGenBackend{fakeBackend: b, err: errNoGenContainment}
	ub := &unprovenStartBackend{killGenBackend: kb, fail: func(uint64) (Proc, error) {
		return leader, errors.New("health: no answer within 30s")
	}}
	hb := &uncontainedStartBackend{unprovenStartBackend: ub, populated: true}
	a.backend = hb
	host := "forked.example.com"
	rt := testRoute(host)
	retained := func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		return a.unprovenWakes[host] != nil
	}
	settled := func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		u := a.unprovenWakes[host]
		return u == nil || !u.busy
	}
	if _, err := a.beginRequest(context.Background(), rt); err == nil {
		t.Fatal("the wake should have failed")
	}
	if !retained() || b.releaseCount() != 0 {
		t.Fatalf("retained %v, lease releases %d: want the wake whose leader lives retained with its lease", retained(), b.releaseCount())
	}

	// The leader exits, but a descendant it forked is still in the host
	// cgroup: the lease stays and the wake stays retained.
	leader.exited.Store(true)
	a.sweepOnce()
	waitFor(t, "the retained wake's retry", settled)
	if got := b.releaseCount(); got != 0 || !retained() {
		t.Fatalf("lease releases = %d with the leader gone and carlos-exec@<host> populated, want 0 and the wake retained", got)
	}
	if hb.hostKillCount() == 0 {
		t.Fatal("the proof never asked KillHost about the host cgroup")
	}
	if k := kb.killed(); !slices.Contains(k, ub.firstGen()) {
		t.Fatalf("KillGen calls = %v, want generation %d's (answered errNoGenContainment)", k, ub.firstGen())
	}
	if _, err := a.beginRequest(context.Background(), rt); !errors.Is(err, errWakeUnproven) {
		t.Fatalf("second wake: err = %v, want errWakeUnproven while the host cgroup is unproven", err)
	}

	// The host cgroup empties: the next sweep proves the wake stopped.
	hb.setPopulated(false)
	a.sweepOnce()
	waitFor(t, "the lease release", func() bool { return b.releaseCount() == 1 })
	waitFor(t, "the retained wake dropped", func() bool { return !retained() })
}

// The same proof for a flag-off worker's failed wake: it is never numbered
// with generations off (Gen{}), so genChildEmpty's "no number, nothing to
// prove" answered for it. A worker runs in carlos-exec@<host> as well, so
// its leader's exit is not enough either. A backend with no host kill and
// a contained generation are as before, and so is a box with no delegated
// cgroup, where the wake ran unaccounted and nothing could be proven.
func TestFailedWakeStoppedProvesTheHostCgroupWithoutContainment(t *testing.T) {
	t.Parallel()
	exited := &sigtermIgnorer{}
	exited.exited.Store(true)
	for _, row := range []struct {
		name      string
		gen       Gen
		killGen   error // nil: a contained generation whose child KillGen proved empty
		hostKill  bool  // the backend has KillHost
		populated bool
		hostErr   error // what KillHost answers, when set
		want      bool
		wantHost  bool // KillHost asked
	}{
		{"uncontained instance, host populated", Gen{N: 4}, errNoGenContainment, true, true, nil, false, true},
		{"uncontained instance, host empty", Gen{N: 4}, errNoGenContainment, true, false, nil, true, true},
		{"flag-off worker, host populated", Gen{}, errNoGenContainment, true, true, nil, false, true},
		{"flag-off worker, host empty", Gen{}, errNoGenContainment, true, false, nil, true, true},
		{"contained generation", Gen{N: 4}, nil, true, true, nil, true, false},
		{"no host kill", Gen{N: 4}, errNoGenContainment, false, false, nil, true, false},
		{"no delegated cgroup", Gen{N: 4}, errNoGenContainment, true, false, cgroup.ErrNoDelegation, true, true},
	} {
		t.Run(row.name, func(t *testing.T) {
			a, b, _, _ := testActivator(t, time.Hour)
			kb := &killGenBackend{fakeBackend: b, err: row.killGen}
			ub := &unprovenStartBackend{killGenBackend: kb}
			hb := &uncontainedStartBackend{unprovenStartBackend: ub, populated: row.populated, err: row.hostErr}
			if row.hostKill {
				a.backend = hb
			} else {
				a.backend = ub
			}
			if got := a.failedWakeStopped(testRoute("f6.example.com"), row.gen, exited, ""); got != row.want {
				t.Fatalf("failedWakeStopped = %v, want %v", got, row.want)
			}
			if asked := hb.hostKillCount() > 0; asked != row.wantHost {
				t.Fatalf("KillHost asked = %v, want %v", asked, row.wantHost)
			}
		})
	}
}

// realFlatBackend is fakeBackend's lease, restore and replica with a real
// flag-off ExecBackend's spawn (StartGen, startChildAt and its health
// wait), its KillGen (errNoGenContainment) and its layout. Only KillHost
// is a fake, since the test box has no delegated cgroup v2: it fails while
// populated, as a host cgroup that still holds a process does.
type realFlatBackend struct {
	*fakeBackend
	exec      *ExecBackend
	hmu       sync.Mutex
	populated bool
	hostKills int
}

func (r *realFlatBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
	return r.exec.StartGen(ctx, rt, n, perGen)
}

func (r *realFlatBackend) NextDueVia(ctx context.Context, rt registry.Route, _ string) (int64, error) {
	return r.fakeBackend.NextDue(ctx, rt)
}

func (r *realFlatBackend) KillGen(ctx context.Context, rt registry.Route, n uint64) error {
	return r.exec.KillGen(ctx, rt, n)
}

func (r *realFlatBackend) Layout(rt registry.Route) (Layout, string) { return r.exec.Layout(rt) }

func (r *realFlatBackend) KillHost(context.Context, registry.Route) error {
	r.hmu.Lock()
	defer r.hmu.Unlock()
	r.hostKills++
	if r.populated {
		return errors.New("carlos-exec@<host> still holds a process")
	}
	return nil
}

func (r *realFlatBackend) setPopulated(on bool) {
	r.hmu.Lock()
	r.populated = on
	r.hmu.Unlock()
}

func (r *realFlatBackend) hostKillCount() int {
	r.hmu.Lock()
	defer r.hmu.Unlock()
	return r.hostKills
}

var (
	_ GenBackend = (*realFlatBackend)(nil)
	_ genKiller  = (*realFlatBackend)(nil)
	_ hostKiller = (*realFlatBackend)(nil)
	_ Classifier = (*realFlatBackend)(nil)
)

// Task 15 fix round 1 (rulings a and b), F6 in the shape it really takes:
// a flat exec child whose health wait times out. startChildAt stopped the
// leader and returned no handle, so doWake gave the lease back at once,
// with nothing asked of carlos-exec@<host>, where whatever the app forked
// still ran. The handle now comes back with the health error, and the
// failed-wake cleanup proves the host cgroup empty before any release:
// populated, the wake is retained with its lease; emptied, the sweep gives
// the lease back.
func TestARealFlatHealthFailedWakeProvesItsHostCgroup(t *testing.T) {
	requireUnixSockets(t)
	old := spawnHealthTimeout
	spawnHealthTimeout = 500 * time.Millisecond
	t.Cleanup(func() { spawnHealthTimeout = old })
	a, b, _, _ := testActivator(t, time.Hour)
	a.gens = false
	exec := testExecBackend(t)
	host := "unhealthy-f6.example.com"
	stageServingLive(t, exec, host)
	t.Setenv("ACTIVATOR_FAKE_APP_UNHEALTHY", "1")
	rb := &realFlatBackend{fakeBackend: b, exec: exec, populated: true}
	a.backend = rb
	rt := registry.Route{Host: host, Kind: "instance", Socket: tmpSock(t), DBPath: filepath.Join(t.TempDir(), host+".db"), Hibernate: true}
	if l, _ := exec.Layout(rt); l != LayoutFlat && l != LayoutForeign {
		t.Fatalf("layout = %s, want an exec child's (flat or foreign)", l)
	}
	retained := func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		return a.unprovenWakes[host] != nil
	}
	settled := func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		u := a.unprovenWakes[host]
		return u == nil || !u.busy
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if _, err := a.beginRequest(ctx, rt); err == nil {
		t.Fatal("a never-healthy app woke")
	}
	if got := b.releaseCount(); got != 0 || !retained() {
		t.Fatalf("lease releases = %d, retained %v after a health failure with carlos-exec@<host> populated; want 0 and retained", got, retained())
	}
	if rb.hostKillCount() == 0 {
		t.Fatal("the failed wake's cleanup never asked KillHost")
	}
	a.sweepOnce()
	waitFor(t, "the retained wake's retry", settled)
	if got := b.releaseCount(); got != 0 || !retained() {
		t.Fatalf("lease releases = %d after a sweep with the host cgroup still populated, want 0", got)
	}
	rb.setPopulated(false)
	a.sweepOnce()
	waitFor(t, "the lease release", func() bool { return b.releaseCount() == 1 })
	waitFor(t, "the retained wake dropped", func() bool { return !retained() })
}

// Ruling b with childProc's semantics and no live leader: a flag-off failed
// wake whose leader stopped (Stop nil, Exited true, as childProc always
// answers) still proves the host cgroup before its lease goes. Before, the
// proof ran only for a leader that lived, so this released at once.
// Ruling c: a nested row and a unit-backed row have their own proof and are
// never KillHost'd; their failed wake releases as before.
func TestAFlagOffFailedWakeProvesTheHostCgroupOnlyForAnExecChild(t *testing.T) {
	t.Parallel()
	for _, row := range []struct {
		name     string
		layout   Layout
		backing  string
		retained bool // populated host cgroup: the wake is kept
		killHost bool
	}{
		{"flat", LayoutFlat, "", true, true},
		{"foreign", LayoutForeign, "", true, true},
		{"nested", LayoutNested, "", false, false},
		{"unit-backed", LayoutFlat, registry.BackingUnit, false, false},
	} {
		t.Run(row.name, func(t *testing.T) {
			a, b, _, _ := testActivator(t, time.Hour)
			a.gens = false
			b.layoutFn = func(registry.Route) (Layout, string) { return row.layout, "" }
			leader := &sigtermIgnorer{}
			leader.exited.Store(true)
			kb := &killGenBackend{fakeBackend: b, err: errNoGenContainment}
			ub := &unprovenStartBackend{killGenBackend: kb, fail: func(uint64) (Proc, error) {
				return leader, errors.New("health: no answer within 30s")
			}}
			hb := &uncontainedStartBackend{unprovenStartBackend: ub, populated: true}
			a.backend = hb
			host := "exited-" + row.name + ".example.com"
			rt := testRoute(host)
			rt.Backing = row.backing
			if _, err := a.beginRequest(context.Background(), rt); err == nil {
				t.Fatal("the wake should have failed")
			}
			a.mu.Lock()
			_, retained := a.unprovenWakes[host]
			a.mu.Unlock()
			if retained != row.retained {
				t.Fatalf("retained = %v, want %v", retained, row.retained)
			}
			wantReleases := 1
			if row.retained {
				wantReleases = 0
			}
			if got := b.releaseCount(); got != wantReleases {
				t.Fatalf("lease releases = %d, want %d", got, wantReleases)
			}
			if asked := hb.hostKillCount() > 0; asked != row.killHost {
				t.Fatalf("KillHost asked = %v, want %v", asked, row.killHost)
			}
		})
	}
}