package activator

import (
	"context"
	"errors"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

func (a *Service) stopWhyOf(host string) string {
	a.mu.Lock()
	defer a.mu.Unlock()
	if e := a.entries[host]; e != nil {
		return e.stopWhy
	}
	return ""
}

func TestStopsRecordTheirReason(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Second)
	rt := testRoute("why.example.com")
	arrangeAwake(t, a, rt)
	gate := make(chan struct{})
	b.lastProc().setStopGate(gate) // hold the entry in stopping
	defer close(gate)
	a.Bounce(rt.Host)
	if got := a.stopWhyOf(rt.Host); got != StopRestart {
		t.Fatalf("Bounce: stopWhy %q, want %q", got, StopRestart)
	}
}

// fakeCmdUnit records the stop and whether replication was already closed
// when it came.
type fakeCmdUnit struct {
	mu           sync.Mutex
	stops        int
	repWasClosed bool
	settleErr    error
	rep          func() *fakeRep
	status       instancectl.CmdStatus
}

func (u *fakeCmdUnit) Invocation() string { return "inv1" }
func (u *fakeCmdUnit) Poll(context.Context) (instancectl.CmdStatus, error) {
	u.mu.Lock()
	defer u.mu.Unlock()
	return u.status, nil
}
func (u *fakeCmdUnit) StopAndSettle(context.Context) error {
	u.mu.Lock()
	defer u.mu.Unlock()
	u.stops++
	if u.rep != nil {
		u.repWasClosed = u.rep().isClosed()
	}
	return u.settleErr
}
func (u *fakeCmdUnit) setSettleErr(err error) {
	u.mu.Lock()
	defer u.mu.Unlock()
	u.settleErr = err
}

func (u *fakeCmdUnit) stopped() (int, bool) {
	u.mu.Lock()
	defer u.mu.Unlock()
	return u.stops, u.repWasClosed
}

// cmdFakeBackend is fakeBackend with the run capability and a nested layout.
type cmdFakeBackend struct {
	*fakeBackend
	cmdMu     sync.Mutex
	unit      *fakeCmdUnit
	startErr  error
	cmdStarts int
	gate      chan struct{} // when set, StartCmd blocks until it closes
	entered   chan struct{} // when set, closed as StartCmd is entered
	// errWithUnit makes a failing start return the unit too: the start
	// reached the control service and the unit may be running.
	errWithUnit bool
	// startCmd, when set, replaces the fake start (a real ExecBackend's).
	startCmd func(context.Context, registry.Route, CmdSpec) (CmdUnit, error)
}

func (b *cmdFakeBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
	if b.entered != nil {
		close(b.entered)
	}
	if b.gate != nil {
		<-b.gate
	}
	b.cmdMu.Lock()
	defer b.cmdMu.Unlock()
	b.cmdStarts++
	if b.startCmd != nil {
		return b.startCmd(ctx, rt, spec)
	}
	if b.startErr != nil {
		if b.errWithUnit {
			return b.unit, b.startErr
		}
		return nil, b.startErr
	}
	return b.unit, nil
}

func testCmdActivator(t *testing.T, cfg Config) (*Service, *cmdFakeBackend, *fakeClock) {
	t.Helper()
	fb := &fakeBackend{layoutFn: func(registry.Route) (Layout, string) { return LayoutNested, "" }}
	cb := &cmdFakeBackend{fakeBackend: fb}
	cb.unit = &fakeCmdUnit{rep: fb.lastRep}
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	if cfg.Bucket == "" {
		cfg.Bucket = "test-bucket"
	}
	if cfg.Idle == 0 {
		cfg.Idle = time.Second
	}
	a := New(cfg, cb, reg)
	a.now = clk.now
	return a, cb, clk
}

func TestRunCmdRefusesWithClaimControlArmed(t *testing.T) {
	a, cb, _ := testCmdActivator(t, Config{HostWrites: hostwrite.Config{Enabled: true}})
	if _, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdClaim) {
		t.Fatalf("err = %v, want ErrCmdClaim", err)
	}
	if acquires, _, _ := cb.counts(); acquires != 0 {
		t.Fatal("a claim-armed refusal woke the instance")
	}
}

func TestRunCmdRefusesAFlatInstance(t *testing.T) {
	a, cb, _ := testCmdActivator(t, Config{})
	cb.layoutFn = func(registry.Route) (Layout, string) { return LayoutFlat, "" }
	if _, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotNested) {
		t.Fatalf("err = %v, want ErrCmdNotNested", err)
	}
}

func TestRunCmdWakesAHibernatedInstanceFirst(t *testing.T) {
	a, cb, _ := testCmdActivator(t, Config{})
	h, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"})
	if err != nil {
		t.Fatal(err)
	}
	if _, _, wakes := cb.counts(); h.Invocation() != "inv1" || wakes != 1 || cb.cmdStarts != 1 {
		t.Fatalf("wake starts %d, cmd starts %d", wakes, cb.cmdStarts)
	}
	h.Release()
}

func TestRunCmdWakeFailureIsAWakeRefusal(t *testing.T) {
	a, cb, _ := testCmdActivator(t, Config{})
	cb.acquireErr = errors.New("restore failed")
	_, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"})
	var we *CmdWakeError
	if !errors.As(err, &we) || we.Reason == "" {
		t.Fatalf("err = %v, want a CmdWakeError with a member-safe reason", err)
	}
	if cb.cmdStarts != 0 {
		t.Fatal("a failed wake started the command")
	}
}

func TestRunCmdRefusesASecondRun(t *testing.T) {
	a, _, _ := testCmdActivator(t, Config{})
	rt := testRoute("c.example.com")
	h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
	if err != nil {
		t.Fatal(err)
	}
	defer h.Release()
	if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_b"}); !errors.Is(err, ErrCmdBusy) {
		t.Fatalf("err = %v, want ErrCmdBusy", err)
	}
}

func TestRunCmdHoldsTheIdleTimer(t *testing.T) {
	t.Parallel()
	a, _, clk := testCmdActivator(t, Config{})
	rt := testRoute("c.example.com")
	h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
	if err != nil {
		t.Fatal(err)
	}
	clk.advance(time.Hour)
	a.sweepOnce()
	if a.entryCount() != 1 || a.stopWhyOf(rt.Host) != "" {
		t.Fatal("the idle timer stopped an instance with a run open")
	}
	h.Release()
	clk.advance(time.Hour)
	a.sweepOnce()
	waitFor(t, "hibernation after the run", func() bool { return a.entryCount() == 0 })
}

// Every stop path stops and settles the run before replication closes, and
// the run learns why (run spec §5.3, codex 1).
func TestEveryStopPathSettlesTheRunBeforeReplicationCloses(t *testing.T) {
	t.Parallel()
	cases := []struct {
		name string
		why  string
		stop func(a *Service, cb *cmdFakeBackend, host string)
	}{
		{"bounce", StopRestart, func(a *Service, _ *cmdFakeBackend, h string) { a.Bounce(h) }},
		{"process exit", StopExited, func(a *Service, cb *cmdFakeBackend, _ string) { cb.lastProc().exit(); a.sweepOnce() }},
		{"lease lost", StopLeaseLost, func(a *Service, cb *cmdFakeBackend, _ string) { cb.lastRep().setErr(errors.New("lost")); a.sweepOnce() }},
		{"convergence", StopRelease, func(a *Service, _ *cmdFakeBackend, h string) {
			a.liveVer = func(string) (string, error) { return "v2", nil }
			a.ConvergeAdopted(h, "v2")
		}},
		{"move fence", StopFenced, func(a *Service, _ *cmdFakeBackend, h string) {
			// FenceAndStop refuses a context without a deadline.
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			go func() {
				defer cancel()
				a.FenceAndStop(ctx, h, 1) //nolint:errcheck
			}()
		}},
		{"retire", StopRetired, func(a *Service, _ *cmdFakeBackend, _ string) { a.RetireExcept(map[string]bool{}) }},
		{"agent shutdown", StopShutdown, func(a *Service, _ *cmdFakeBackend, _ string) {
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			a.Shutdown(ctx)
		}},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			a, cb, _ := testCmdActivator(t, Config{})
			cb.execVerFn = func() string { return "v1" }
			rt := testRoute("c.example.com")
			h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
			if err != nil {
				t.Fatal(err)
			}
			c.stop(a, cb, rt.Host)
			waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
			stops, repWasClosed := cb.unit.stopped()
			if stops != 1 {
				t.Fatalf("the cmd unit was stopped %d times, want 1", stops)
			}
			if repWasClosed {
				t.Fatal("replication closed before the run settled")
			}
			if !cb.lastRep().isClosed() {
				t.Fatal("replication never closed")
			}
			if got, at := h.Interrupted(); got != c.why || at.IsZero() {
				t.Fatalf("Interrupted() = %q at %v, want %q and when", got, at, c.why)
			}
		})
	}
}

// The hold stamps the first stop asked for with the activator's clock, so
// the run can tell an instance stop that came after its command finished.
func TestRunCmdInterruptedSaysWhenTheStopWasAsked(t *testing.T) {
	t.Parallel()
	a, _, clk := testCmdActivator(t, Config{})
	rt := testRoute("c.example.com")
	h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
	if err != nil {
		t.Fatal(err)
	}
	clk.advance(5 * time.Minute)
	asked := clk.now()
	a.Bounce(rt.Host)
	waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
	clk.advance(time.Minute)
	if err := h.Stop(context.Background()); err != nil {
		t.Fatal(err)
	}
	if why, at := h.Interrupted(); why != StopRestart || !at.Equal(asked) {
		t.Fatalf("Interrupted() = %q at %v, want %q at %v (the first stop's)", why, at, StopRestart, asked)
	}
}

func TestRunCmdNeverStartsUnderAStoppingEntry(t *testing.T) {
	a, cb, _ := testCmdActivator(t, Config{})
	rt := testRoute("c.example.com")
	_, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Prepare: func(context.Context) error {
		a.Bounce(rt.Host) // the instance starts stopping while the files download
		return nil
	}})
	var se *CmdStoppedError
	if !errors.As(err, &se) || se.Reason != StopRestart {
		t.Fatalf("err = %v, want CmdStoppedError{%q}", err, StopRestart)
	}
	if cb.cmdStarts != 0 {
		t.Fatal("the cmd unit started under a stopping entry")
	}
}

func TestAStopDuringTheStartWaitsForIt(t *testing.T) {
	t.Parallel()
	a, cb, _ := testCmdActivator(t, Config{})
	cb.gate = make(chan struct{})
	cb.entered = make(chan struct{})
	rt := testRoute("c.example.com")
	done := make(chan error, 1)
	go func() {
		_, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
		done <- err
	}()
	waitFor(t, "the run attached", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		e := a.entries[rt.Host]
		return e != nil && e.cmd != nil
	})
	// The attach comes before RunCmd's last awake check; the stop must land
	// after that check, with StartCmd in flight.
	<-cb.entered
	a.Bounce(rt.Host)
	time.Sleep(50 * time.Millisecond)
	if cb.lastRep().isClosed() {
		t.Fatal("replication closed while the cmd unit's start was still in flight")
	}
	close(cb.gate)
	if err := <-done; err != nil {
		t.Fatalf("RunCmd: %v", err)
	}
	waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
	if stops, repWasClosed := cb.unit.stopped(); stops != 1 || repWasClosed {
		t.Fatalf("stops %d, replication closed first %v", stops, repWasClosed)
	}
}

// A run whose stop cannot be proven keeps the instance's stop pending:
// the entry, replication and the lease stay, and the run's handle with
// them. The sweep tries the settle again, and the retry that proves it
// closes replication and releases the lease (merge ruling M1). That holds
// with generations off and on.
func TestAnUnprovenRunStopKeepsReplicationAndTheLease(t *testing.T) {
	t.Parallel()
	for _, gens := range []bool{false, true} {
		t.Run(map[bool]string{false: "generations off", true: "generations on"}[gens], func(t *testing.T) {
			a, cb, _ := testCmdActivator(t, Config{})
			a.gens = gens
			cb.unit.setSettleErr(errors.New("processes remained"))
			rt := testRoute("c.example.com")
			if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); err != nil {
				t.Fatal(err)
			}
			a.Bounce(rt.Host)
			waitFor(t, "the first settle", func() bool { n, _ := cb.unit.stopped(); return n >= 1 })
			// Every sweep retries, and every retry fails: nothing is given up.
			for i := 0; i < 3; i++ {
				before, _ := cb.unit.stopped()
				a.sweepOnce()
				waitFor(t, "a retried settle", func() bool { n, _ := cb.unit.stopped(); return n > before })
				if a.entryCount() != 1 {
					t.Fatal("the entry went while its run was not proven stopped")
				}
				if cb.lastRep().isClosed() {
					t.Fatal("replication closed, and the lease went, while the run was not proven stopped")
				}
			}
			a.mu.Lock()
			held := a.entries[rt.Host] != nil && len(cmdHoldsLocked(a.entries[rt.Host])) == 1
			a.mu.Unlock()
			if !held {
				t.Fatal("the entry let go of the run's handle")
			}
			if !a.StopUncertain(rt.Host) {
				t.Fatal("StopUncertain = false with the run unproven")
			}
			cb.unit.setSettleErr(nil)
			waitFor(t, "the retry that proves the run stopped", func() bool {
				a.sweepOnce()
				return a.entryCount() == 0
			})
			if _, repWasClosed := cb.unit.stopped(); repWasClosed {
				t.Fatal("replication closed before the run was proven stopped")
			}
			if !cb.lastRep().isClosed() {
				t.Fatal("replication never closed, so the lease was never released")
			}
			if a.StopUncertain(rt.Host) {
				t.Fatal("StopUncertain = true after the run was proven stopped")
			}
		})
	}
}

// A run released without a proven settle (its Stop failed, or was never
// called) still belongs to the instance's next stop: that stop settles it
// before replication closes, and holds replication and the lease until
// it has.
func TestAReleasedRunThatNeverSettledStillHoldsTheStop(t *testing.T) {
	t.Parallel()
	for _, c := range []struct {
		name      string
		stopFirst bool
		settleErr error
	}{
		{"stopped and settled", true, nil},
		{"never stopped", false, nil},
		{"stop failed", true, errors.New("processes remained")},
	} {
		t.Run(c.name, func(t *testing.T) {
			a, cb, clk := testCmdActivator(t, Config{})
			cb.unit.setSettleErr(c.settleErr)
			rt := testRoute("c.example.com")
			h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
			if err != nil {
				t.Fatal(err)
			}
			if c.stopFirst {
				if err := h.Stop(context.Background()); (err != nil) != (c.settleErr != nil) {
					t.Fatalf("Stop: %v", err)
				}
			}
			h.Release()
			clk.advance(time.Hour)
			a.sweepOnce() // the release ended the idle hold
			if c.settleErr != nil {
				waitFor(t, "the stop's own settle", func() bool { n, _ := cb.unit.stopped(); return n >= 2 })
				a.sweepOnce()
				if a.entryCount() != 1 || cb.lastRep().isClosed() {
					t.Fatal("the stop finished over a run that is not proven stopped")
				}
				cb.unit.setSettleErr(nil)
			}
			waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
			stops, repWasClosed := cb.unit.stopped()
			// The failing case: the run's own Stop, the instance stop's
			// settle, and at least one retry, the last of which proved it.
			if c.settleErr == nil && stops != 1 || c.settleErr != nil && stops < 3 {
				t.Fatalf("stops %d", stops)
			}
			if repWasClosed {
				t.Fatal("replication closed before the run settled")
			}
			if a.StopUncertain(rt.Host) {
				t.Fatal("StopUncertain = true after every run was proven stopped")
			}
			if got, _ := h.Interrupted(); got != "" && c.stopFirst {
				t.Fatalf("Interrupted() = %q after the run's own stop", got)
			}
		})
	}
}

func TestRunCmdRefusesAnythingButAnInstance(t *testing.T) {
	a, cb, _ := testCmdActivator(t, Config{})
	rt := testRoute("c.example.com")
	rt.Kind = "sidecar"
	if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotInstance) {
		t.Fatalf("err = %v, want ErrCmdNotInstance", err)
	}
	if acquires, _, _ := cb.counts(); acquires != 0 {
		t.Fatal("a refused route was woken")
	}
}

// A stop during a slow Prepare (the file downloads) does not wait for it:
// no unit has started, so the stop is prompt and complete, Prepare's
// context is cancelled, and the run ends interrupted with no unit started.
func TestAStopDuringASlowPrepareDoesNotWaitForIt(t *testing.T) {
	t.Parallel()
	a, cb, _ := testCmdActivator(t, Config{})
	rt := testRoute("c.example.com")
	inPrepare, unblock := make(chan struct{}), make(chan struct{})
	var cancelled atomic.Bool
	done := make(chan error, 1)
	go func() {
		_, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Prepare: func(ctx context.Context) error {
			close(inPrepare)
			<-unblock // a download that does not watch its context
			cancelled.Store(ctx.Err() != nil)
			return nil
		}})
		done <- err
	}()
	<-inPrepare
	a.Bounce(rt.Host)
	waitFor(t, "the instance stopped while Prepare was still running", func() bool { return a.entryCount() == 0 })
	if a.StopUncertain(rt.Host) {
		t.Fatal("a stop that no unit outlived was recorded uncertain")
	}
	close(unblock)
	var se *CmdStoppedError
	if err := <-done; !errors.As(err, &se) || se.Reason != StopRestart {
		t.Fatalf("err = %v, want CmdStoppedError{%q}", err, StopRestart)
	}
	if !cancelled.Load() {
		t.Error("the stop did not cancel Prepare's context")
	}
	if cb.cmdStarts != 0 {
		t.Fatal("a unit started after the stop")
	}
}

// A start that failed after it reached the control service may have left
// the unit running: RunCmd settles it, and a unit that will not settle
// keeps the instance from another run and makes its next stop uncertain.
func TestAFailedStartThatMayHaveRunIsSettled(t *testing.T) {
	t.Parallel()
	for _, c := range []struct {
		name      string
		settleErr error
	}{{"settles", nil}, {"will not settle", errors.New("processes remained")}} {
		t.Run(c.name, func(t *testing.T) {
			a, cb, _ := testCmdActivator(t, Config{})
			cb.startErr, cb.errWithUnit = errors.New("502 mid-roll"), true
			cb.unit.setSettleErr(c.settleErr)
			rt := testRoute("c.example.com")
			if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, cb.startErr) {
				t.Fatalf("err = %v, want the start's error", err)
			}
			if stops, _ := cb.unit.stopped(); stops != 1 {
				t.Fatalf("the unit was stopped %d times, want 1", stops)
			}
			_, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_b"})
			if busy := errors.Is(err, ErrCmdBusy); busy != (c.settleErr != nil) {
				t.Fatalf("second run: err = %v, busy = %v", err, busy)
			}
			a.Bounce(rt.Host)
			if c.settleErr != nil {
				waitFor(t, "the stop's settle", func() bool { n, _ := cb.unit.stopped(); return n >= 2 })
				a.sweepOnce()
				if a.entryCount() != 1 || cb.lastRep().isClosed() {
					t.Fatal("the stop finished over a unit that is not proven stopped")
				}
				cb.unit.setSettleErr(nil)
			}
			waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
			if _, repWasClosed := cb.unit.stopped(); repWasClosed {
				t.Fatal("replication closed before the unit settled")
			}
			if a.StopUncertain(rt.Host) {
				t.Fatal("StopUncertain = true after the unit was proven stopped")
			}
		})
	}
}

// The real ExecBackend: POST /cmd starts the unit and then fails (a lost
// response). The cmd role is stopped and its cgroup waited for before
// replication closes. A cgroup that cannot be read empty keeps the stop
// pending with replication open, and the retry once it reads empty
// finishes it (merge ruling M1).
func TestALostCmdResponseStopsTheCmdRoleBeforeReplicationCloses(t *testing.T) {
	for _, c := range []struct {
		name      string
		cgroupErr error
	}{{"cgroup empties", nil}, {"cgroup unreadable", errors.New("cgroup.events: permission denied")}} {
		t.Run(c.name, func(t *testing.T) {
			shrinkCmdAcceptGrace(t, 50*time.Millisecond) // the stop drains two streams the fake unit never connected
			a, cb, _ := testCmdActivator(t, Config{})
			ctl := &fakeControl{cmdErr: errors.New("502 mid-roll")}
			unit := &fakeUnit{}
			unit.set("active", "inv1", true, true) // the POST reached systemd
			b, rt := nestedBackend(t, ctl, unit)
			b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
			var repClosedAtEmpty, unreadable atomic.Bool
			unreadable.Store(c.cgroupErr != nil)
			b.CgroupEvents = func(slice, u string) (bool, bool, error) {
				if unreadable.Load() {
					return true, true, c.cgroupErr
				}
				exists, populated, err := unit.events(slice, u)
				if !populated {
					repClosedAtEmpty.Store(cb.lastRep().isClosed())
				}
				return exists, populated, err
			}
			ctl.stopHook = func(_ string, role instancectl.Role) {
				if role == instancectl.RoleCmd {
					go func() {
						time.Sleep(50 * time.Millisecond)
						unit.set("inactive", "inv1", false, false)
					}()
				}
			}
			cb.startCmd = b.StartCmd
			dir := filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), CmdRunDirName, "run_a")
			if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", RunDir: dir}); err == nil || !strings.Contains(err.Error(), "502 mid-roll") {
				t.Fatalf("err = %v, want the lost response", err)
			}
			a.Bounce(rt.Host)
			if c.cgroupErr != nil {
				waitFor(t, "the stop kept pending", func() bool {
					a.mu.Lock()
					defer a.mu.Unlock()
					e := a.entries[rt.Host]
					return e != nil && e.pendingStop
				})
				if cb.lastRep().isClosed() {
					t.Fatal("replication closed while the cmd cgroup could not be read empty")
				}
				unreadable.Store(false)
			}
			waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
			if !strings.Contains(ctl.joined(), "stop "+rt.Host+" cmd") {
				t.Fatalf("the cmd role was never stopped:\n%s", ctl.joined())
			}
			if repClosedAtEmpty.Load() {
				t.Fatal("replication closed before the cmd cgroup emptied")
			}
			if a.StopUncertain(rt.Host) {
				t.Fatal("StopUncertain = true after the cmd role was proven stopped")
			}
		})
	}
}

// A run runs what the instance serves (zero-downtime spec §7.5, as for the
// legacy sidecar): the current generation's recorded file by path, captured
// at the last awake check, with generations on or off (settings ruling
// R-EXEC-SIDECAR: a flag-off wake runs its recorded file by path too, so
// the live link may name another build). Only a generation that names no
// file leaves the run the live link. A caller's own setting never
// survives: RunCmd decides.
func TestRunCmdRunsTheServingGenerationsBuild(t *testing.T) {
	t.Parallel()
	rt := testRoute("c.example.com")
	staged := &instancectl.StagedExe{Path: "/opt/carlos/versions/x", Version: "v9", Digest: strings.Repeat("0", 64)}
	want := instancectl.StagedExe{Path: "/opt/carlos/versions/v2/blog", Version: "v2", Digest: strings.Repeat("a", 64)}
	for _, gens := range []bool{false, true} {
		t.Run(map[bool]string{false: "generations off", true: "generations on"}[gens], func(t *testing.T) {
			// The switch is set before any wake, as CARLOS_SWAP=overlap
			// sets it, before anything spawns.
			a, cb, _ := testCmdActivator(t, Config{})
			a.gens = gens
			var got []*instancectl.StagedExe
			cb.startCmd = func(_ context.Context, _ registry.Route, spec CmdSpec) (CmdUnit, error) {
				got = append(got, spec.serving)
				return cb.unit, nil
			}
			run := func(id string, spec CmdSpec) {
				t.Helper()
				spec.RunID = id
				h, err := a.RunCmd(context.Background(), rt, spec)
				if err != nil {
					t.Fatalf("%s: %v", id, err)
				}
				if err := h.Stop(context.Background()); err != nil {
					t.Fatal(err)
				}
				h.Release()
			}
			// A generation that names no file: the live link, whatever the
			// caller put in the spec.
			arrangeAwake(t, a, rt)
			a.mu.Lock()
			e := a.entries[rt.Host]
			e.gen.StagedPath, e.gen.Digest = "", ""
			a.mu.Unlock()
			run("run_livelink", CmdSpec{serving: staged})
			a.mu.Lock()
			e.gen.Build, e.gen.StagedPath, e.gen.Digest = want.Version, want.Path, want.Digest
			a.mu.Unlock()
			run("run_staged", CmdSpec{})
			if len(got) != 2 || got[0] != nil {
				t.Fatalf("serving builds %v, want the live link for a generation that names no file", got)
			}
			if got[1] == nil || *got[1] != want {
				t.Fatalf("serving build %v, want the current generation's %v", got[1], want)
			}
		})
	}
}

// StartCmd names a run's serving build to root through CmdExe, and needs no
// live link for it. A control channel that cannot carry the build refuses
// the run rather than run the live link, which after a fallback names a
// build the instance is not serving.
func TestStartCmdNamesTheServingBuild(t *testing.T) {
	b, rt := nestedBackend(t, &fakeControl{cmdInv: "inv1"}, &fakeUnit{})
	b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
	dir := filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), CmdRunDirName, "run_a")
	x := instancectl.StagedExe{Path: "/opt/carlos/versions/v2/blog", Version: "v2", Digest: strings.Repeat("a", 64)}
	spec := CmdSpec{RunID: "run_a", RunDir: dir, Stdout: &syncBuffer{}, Stderr: &syncBuffer{}, serving: &x, genEnv: map[string]string{}}
	if u, err := b.StartCmd(context.Background(), rt, spec); !errors.Is(err, errNoCmdExe) || u != nil {
		t.Fatalf("StartCmd without CmdExe = %v, %v; want errNoCmdExe and no unit", u, err)
	}
	ctl := &exeControl{}
	b.Control = ctl
	u, err := b.StartCmd(context.Background(), rt, spec)
	if err != nil {
		t.Fatal(err)
	}
	u.(*cmdUnit).out.Close()
	u.(*cmdUnit).errs.Close()
	ctl.mu.Lock()
	defer ctl.mu.Unlock()
	if len(ctl.cmdExes) != 1 || ctl.cmdExes[0] != x {
		t.Fatalf("cmd starts %v, want one naming %v", ctl.cmdExes, x)
	}
}