package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/runs"
	"github.com/carlosframework/platform/internal/seeds"
)

type fakeHandle struct {
	mu          sync.Mutex
	polls       chan instancectl.CmdStatus
	stops       int
	released    bool
	interrupted string
	askedAt     time.Time // when the interrupting stop was asked for; zero is "long ago"
	afterStop   *instancectl.CmdStatus
	stopErr     error
}

func (h *fakeHandle) interrupt(why string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.interrupted, h.askedAt = why, time.Now()
}

func (h *fakeHandle) Invocation() string { return "inv1" }
func (h *fakeHandle) Poll(ctx context.Context) (instancectl.CmdStatus, error) {
	h.mu.Lock()
	if h.stops > 0 {
		st := instancectl.CmdStatus{State: instancectl.CmdGone}
		if h.afterStop != nil {
			st = *h.afterStop
		}
		h.mu.Unlock()
		return st, nil
	}
	h.mu.Unlock()
	select {
	case st := <-h.polls:
		return st, nil
	case <-ctx.Done():
		return instancectl.CmdStatus{}, ctx.Err()
	}
}
func (h *fakeHandle) Interrupted() (string, time.Time) {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.interrupted, h.askedAt
}
func (h *fakeHandle) Stop(context.Context) error {
	h.mu.Lock()
	h.stops++
	err := h.stopErr
	h.mu.Unlock()
	return err
}
func (h *fakeHandle) Release() { h.mu.Lock(); h.released = true; h.mu.Unlock() }

// fakeLauncher counts the units it starts. With noUnit set it starts none:
// a test that must start no unit fails at once instead of following one.
type fakeLauncher struct {
	mu     sync.Mutex
	calls  int
	units  int
	noUnit bool
	err    error
	handle *fakeHandle
	out    string
	errOut string
}

func (l *fakeLauncher) started() int { l.mu.Lock(); defer l.mu.Unlock(); return l.units }

func (l *fakeLauncher) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
	l.mu.Lock()
	l.calls++
	l.mu.Unlock()
	if l.err != nil {
		return nil, l.err
	}
	if spec.Prepare != nil {
		if err := spec.Prepare(ctx); err != nil {
			return nil, err
		}
	}
	l.mu.Lock()
	l.units++
	l.mu.Unlock()
	if l.noUnit {
		return nil, errors.New("fake: a unit was started")
	}
	io.WriteString(spec.Stdout, l.out)
	io.WriteString(spec.Stderr, l.errOut)
	return l.handle, nil
}

// pickedRun queues and picks a run for box-1/boot-1 and returns what
// execute is called with.
func pickedRun(t *testing.T, s release.Store, timeoutS int64) (provision.Record, runs.Request) {
	t.Helper()
	rec, id := queueRun(t, s, time.Now().Add(time.Minute))
	req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
	if err != nil {
		t.Fatal(err)
	}
	req.TimeoutS = timeoutS
	got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	if err != nil {
		t.Fatal(err)
	}
	return *got, *req
}

// execWorker is a box-1/boot-1 worker over a data dir that holds the
// instance's directory, as a real box does.
func execWorker(t *testing.T, s release.Store, l runLauncher, boot string) *runsWorker {
	t.Helper()
	dataDir := t.TempDir()
	if err := os.MkdirAll(provision.InstanceDir(dataDir, runHost), 0o700); err != nil {
		t.Fatal(err)
	}
	return newRunsWorker(s, runOwnedRoute(runHost), l, dataDir, "box-1", boot)
}

func runWith(t *testing.T, s release.Store, l *fakeLauncher, rec provision.Record, req runs.Request, h *runHandle) *runs.Result {
	t.Helper()
	return runOn(t, execWorker(t, s, l, "boot-1"), s, rec, req, h)
}

func runOn(t *testing.T, w *runsWorker, s release.Store, rec provision.Record, req runs.Request, h *runHandle) *runs.Result {
	t.Helper()
	rt, _ := runOwnedRoute(runHost)(runHost)
	w.active[req.ID] = h
	w.execute(context.Background(), rec, req, *rt, h)
	res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID)
	if err != nil || res == nil {
		t.Fatalf("no result: %v", err)
	}
	if w.active[req.ID] != nil {
		t.Error("the run was not forgotten")
	}
	if _, err := os.Stat(runDirOf(w, req)); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("the run directory is still there: %v", err)
	}
	return res
}

func runDirOf(w *runsWorker, req runs.Request) string {
	return provision.InstanceDir(w.dataDir, runHost) + "/" + runs.DirName + "/" + req.ID
}

func TestRunExecutesAndWritesTheResultAfterTheLastChunk(t *testing.T) {
	s := &putOrder{Store: release.NewDirStore(t.TempDir())}
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "exit-code", Status: "3", Code: "1"}
	l := &fakeLauncher{handle: h, out: "hello\n", errOut: "oops\n"}
	res := runWith(t, s, l, rec, req, newRunHandle())
	if res.Exit != 3 || res.Signal != "" || res.LastSeq != 2 || res.Box != "box-1" || res.Started.IsZero() || res.Ended.IsZero() {
		t.Fatalf("result %+v", res)
	}
	if h.stops == 0 || !h.released {
		t.Fatal("the unit was not stopped after done, or the hold not released")
	}
	if r := currentRun(t, s.Store); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
	if !s.resultAfterChunks() {
		t.Fatalf("puts %q: result.json must come after every chunk", s.keys)
	}
}

// putOrder records the order of Puts.
type putOrder struct {
	release.Store
	mu   sync.Mutex
	keys []string
}

func (p *putOrder) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	p.mu.Lock()
	p.keys = append(p.keys, key)
	p.mu.Unlock()
	return p.Store.Put(ctx, key, body, ifMatch)
}

func (p *putOrder) resultAfterChunks() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	result := -1
	lastChunk := -1
	for i, k := range p.keys {
		if strings.HasSuffix(k, "/result.json") {
			result = i
		}
		if strings.Contains(k, "/out/") {
			lastChunk = i
		}
	}
	return result > lastChunk
}

func TestRunKilledBySignal(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "signal", Status: "9", Code: "2"}
	if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Signal != "9" || res.Exit != 0 {
		t.Fatalf("result %+v", res)
	}
}

// ctxStore refuses a Put whose context has ended, as S3 does; DirStore
// ignores it.
type ctxStore struct{ release.Store }

func (c ctxStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if err := ctx.Err(); err != nil {
		return "", err
	}
	return c.Store.Put(ctx, key, body, ifMatch)
}

func TestRunCancelledAfterLaunch(t *testing.T) {
	t.Parallel()
	s := ctxStore{release.NewDirStore(t.TempDir())}
	rec, req := pickedRun(t, s, 60)
	done := instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "exit-code", Status: "143", Code: "1"}
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus), afterStop: &done}
	rh := newRunHandle()
	go func() { time.Sleep(50 * time.Millisecond); rh.requestCancel() }()
	res := runWith(t, s, &fakeLauncher{handle: h, out: "partial\n"}, rec, req, rh)
	if !res.Cancelled || res.Exit != 143 || res.LastSeq != 1 {
		t.Fatalf("result %+v (the last chunk lands after a cancel)", res)
	}
	if r := currentRun(t, s); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
}

func TestRunTimesOut(t *testing.T) {
	s := ctxStore{release.NewDirStore(t.TempDir())}
	rec, req := pickedRun(t, s, 1)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
	start := time.Now()
	res := runWith(t, s, &fakeLauncher{handle: h, out: "slow\n"}, rec, req, newRunHandle())
	if !res.TimedOut || time.Since(start) < time.Second || h.stops == 0 {
		t.Fatalf("result %+v after %s, stops %d", res, time.Since(start), h.stops)
	}
	if res.LastSeq != 1 || res.Interrupted != "" {
		t.Fatalf("result %+v: the last chunk lands after a timeout, and a stopped unit is not interrupted", res)
	}
}

// The run's deadline is set at pickup, so slow staging does not buy a full
// extra timeout: a run whose deadline is nearer than its timeout times out
// at the deadline, or the console sweep marks a still-running run lost.
func TestRunTimesOutAtItsDeadlineNotAFullTimeoutLater(t *testing.T) {
	s := ctxStore{release.NewDirStore(t.TempDir())}
	rec, id := queueRun(t, s, time.Now().Add(time.Minute))
	req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
	if err != nil {
		t.Fatal(err)
	}
	req.TimeoutS = 60
	got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(1500*time.Millisecond)))
	if err != nil {
		t.Fatal(err)
	}
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
	start := time.Now()
	res := runWith(t, s, &fakeLauncher{handle: h, out: "slow\n"}, *got, *req, newRunHandle())
	if !res.TimedOut || time.Since(start) > 10*time.Second || h.stops == 0 {
		t.Fatalf("result %+v after %s, stops %d: a run times out at its deadline, not %ds after launch", res, time.Since(start), h.stops, req.TimeoutS)
	}
}

// Review Focus 2: a cancel after the pickup CAS and before the unit starts
// ends the run cancelled with no unit started.
func TestRunCancelledBeforeLaunchNeverStarts(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	if _, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID)); err != nil {
		t.Fatal(err)
	}
	l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
	res := runWith(t, s, l, rec, req, newRunHandle())
	if !res.Cancelled || l.started() != 0 || !res.Started.IsZero() {
		t.Fatalf("result %+v, units started %d", res, l.started())
	}
	if r := currentRun(t, s); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
}

// cancelOnInput cancels the run the moment its input is read: a cancel that
// lands while the files download.
type cancelOnInput struct {
	release.Store
	acct, app, host, id string
}

func (c *cancelOnInput) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.Contains(key, "/in/") {
		if _, err := provision.SetRecordRun(ctx, c.Store, c.acct, c.app, c.host, provision.RunCancelTransition(c.id)); err != nil {
			return nil, "", err
		}
	}
	return c.Store.Get(ctx, key)
}

func TestRunCancelledWhileFilesDownloadNeverStarts(t *testing.T) {
	dir := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, dir, 60)
	body := []byte("a,b\n")
	sum := sha256.Sum256(body)
	req.Files = []runs.File{{Name: "in.csv", Size: int64(len(body)), SHA256: hex.EncodeToString(sum[:])}}
	if _, err := dir.Put(context.Background(), runs.InputKey(rec.Account, rec.App, req.ID, "in.csv"), body, ""); err != nil {
		t.Fatal(err)
	}
	s := &cancelOnInput{Store: dir, acct: rec.Account, app: rec.App, host: rec.Host, id: req.ID}
	l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
	res := runWith(t, s, l, rec, req, newRunHandle())
	if !res.Cancelled || l.started() != 0 || !res.Started.IsZero() {
		t.Fatalf("result %+v, units started %d: the record is read again after the download", res, l.started())
	}
}

func TestRunStagesItsFiles(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	body := []byte("a,b\n")
	sum := sha256.Sum256(body)
	req.Files = []runs.File{{Name: "in.csv", Size: int64(len(body)), SHA256: hex.EncodeToString(sum[:])}}
	if _, err := s.Put(context.Background(), runs.InputKey(rec.Account, rec.App, req.ID, "in.csv"), body, ""); err != nil {
		t.Fatal(err)
	}
	w := execWorker(t, s, nil, "boot-1")
	var staged []byte
	l := &stagedReader{read: func(spec activator.CmdSpec) {
		staged, _ = os.ReadFile(spec.RunDir + "/in.csv")
		if spec.RunDir != runDirOf(w, req) {
			t.Errorf("RunDir %s, want %s", spec.RunDir, runDirOf(w, req))
		}
	}}
	w.launch = l
	runOn(t, w, s, rec, req, newRunHandle())
	if string(staged) != string(body) {
		t.Fatalf("staged %q, want %q", staged, body)
	}
}

// stagedReader prepares, reads what was staged, and refuses to start.
type stagedReader struct{ read func(activator.CmdSpec) }

func (l *stagedReader) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
	if err := spec.Prepare(ctx); err != nil {
		return nil, err
	}
	l.read(spec)
	return nil, errors.New("start cmd unit: 502")
}

func TestRunRefusesAnIncarnationThatDoesNotMatch(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	req.Incarnation = strings.Repeat("0", 64)
	l := &fakeLauncher{handle: &fakeHandle{}}
	if res := runWith(t, s, l, rec, req, newRunHandle()); res.Refused != runs.RefusedIncarnation || l.calls != 0 {
		t.Fatalf("result %+v, launches %d", res, l.calls)
	}
	if r := currentRun(t, s); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
}

func TestRunRefusesAnImportUncommittedAtLaunch(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	// Admission refuses an uncommitted import, so no walk of the seed
	// transitions reaches a picked run on one: the launch check is the
	// backstop. Give the record its seed through the seed writer directly.
	if _, err := provision.SetRecordSeed(context.Background(), s, rec.Account, rec.App, rec.Host, func(r *provision.Record) error {
		r.Seed = &provision.SeedSource{ID: "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa", Kind: seeds.KindImport, State: seeds.StatePlaced}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if res := runWith(t, s, &fakeLauncher{handle: &fakeHandle{}}, rec, req, newRunHandle()); res.Refused != runs.RefusedImport {
		t.Fatalf("result %+v (codex R2-2: checked again at launch)", res)
	}
}

func TestRunLaunchErrorsBecomeResults(t *testing.T) {
	cases := []struct {
		err  error
		want func(runs.Result) bool
	}{
		{activator.ErrCmdClaim, func(r runs.Result) bool { return r.Refused == runs.RefusedClaim }},
		{activator.ErrCmdNotNested, func(r runs.Result) bool { return r.Refused == runs.RefusedFlat }},
		{activator.ErrCmdNotInstance, func(r runs.Result) bool { return r.Refused == runs.RefusedFlat }},
		{&activator.CmdWakeError{Reason: "wake failed: restore", Err: errors.New("x")}, func(r runs.Result) bool {
			return r.Refused == runs.RefusedWake && r.Detail == "wake failed: restore"
		}},
		{&activator.CmdStoppedError{Reason: activator.StopRestart}, func(r runs.Result) bool { return r.Interrupted == activator.StopRestart }},
		{activator.ErrCmdBusy, func(r runs.Result) bool { return r.Interrupted == runs.InterruptedStartFailed }},
		{errors.New("start cmd unit: 502"), func(r runs.Result) bool { return r.Interrupted == runs.InterruptedStartFailed }},
	}
	for _, c := range cases {
		s := release.NewDirStore(t.TempDir())
		rec, req := pickedRun(t, s, 60)
		if res := runWith(t, s, &fakeLauncher{err: c.err}, rec, req, newRunHandle()); !c.want(*res) {
			t.Errorf("%v: result %+v", c.err, res)
		}
		if r := currentRun(t, s); r.State != provision.RunDone {
			t.Errorf("%v: run %+v, want done", c.err, r)
		}
	}
}

func TestRunInterruptedByAnInstanceStop(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1), interrupted: activator.StopRestart}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "143", Code: "1"}
	if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != activator.StopRestart {
		t.Fatalf("result %+v", res)
	}
}

// An instance stop that lands after the command's done status arrived, but
// before the run's own stop, did not end the command: the result is the
// command's exit, not interrupted.
func TestRunFinishedBeforeAnInstanceStopKeepsItsExit(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1), interrupted: activator.StopRestart, askedAt: time.Now().Add(time.Hour)}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "0", Code: "1"}
	if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != "" || res.Exit != 0 {
		t.Fatalf("result %+v, want exit 0 and not interrupted", res)
	}
}

// A run interrupted by a retired generation while its unit is still
// running, and whose stop keeps failing, ends interrupted (exit 125) as
// soon as the interruption is seen: the worker does not follow the command
// to its own end or its timeout (re-review 6).
func TestRunInterruptedWhileItsStopIsRetriedEndsAt125(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 3600)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus), stopErr: errors.New("processes remained")}
	go func() {
		time.Sleep(100 * time.Millisecond)
		h.interrupt(activator.StopRelease)
	}()
	start := time.Now()
	res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
	if res.Interrupted != activator.StopRelease || res.TimedOut || res.Cancelled {
		t.Fatalf("result %+v, want interrupted by the retirement", res)
	}
	if took := time.Since(start); took > runInterruptGrace+5*time.Second {
		t.Fatalf("the result took %s", took)
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.stops == 0 || !h.released {
		t.Fatalf("stops %d released %v: the worker must still stop and release the handle", h.stops, h.released)
	}
}

func TestRunWhoseUnitWentAway(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdGone, InvocationID: "inv2"}
	if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != runs.InterruptedGone {
		t.Fatalf("result %+v", res)
	}
}

// The record-read backstop: a cancel whose serial bump this box missed is
// still seen while the run executes.
func TestRunSeesACancelByRereadingTheRecord(t *testing.T) {
	defer func(d time.Duration) { runRecheckEvery = d }(runRecheckEvery)
	runRecheckEvery = 20 * time.Millisecond
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
	errc := make(chan error, 1)
	go func() {
		time.Sleep(50 * time.Millisecond)
		_, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID))
		errc <- err
	}()
	res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
	if err := <-errc; err != nil {
		t.Fatal(err)
	}
	if !res.Cancelled {
		t.Fatalf("result %+v", res)
	}
}

// failingPuts fails every Put whose key holds part.
type failingPuts struct {
	release.Store
	part string
}

func (f *failingPuts) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if strings.Contains(key, f.part) {
		return "", errors.New("AccessDenied")
	}
	return f.Store.Put(ctx, key, body, ifMatch)
}

// A store that refuses every chunk cannot hold the run open: the output's
// bound ends the chunker, and the result says truncated.
func TestRunOutputIsBoundedWhenEveryChunkFails(t *testing.T) {
	defer func(d time.Duration) { runOutputGrace = d }(runOutputGrace)
	runOutputGrace = 300 * time.Millisecond
	dir := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, dir, 60)
	s := &failingPuts{Store: dir, part: "/out/"}
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "0", Code: "1"}
	start := time.Now()
	res := runWith(t, s, &fakeLauncher{handle: h, out: "lost\n"}, rec, req, newRunHandle())
	if !res.Truncated || res.LastSeq != 0 || time.Since(start) > 10*time.Second {
		t.Fatalf("result %+v after %s", res, time.Since(start))
	}
	if r := currentRun(t, dir); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
}

// A result that cannot be written leaves the run picked, for the sweep to
// end as lost: done is never set without a result.
func TestRunWithoutAResultIsNotDone(t *testing.T) {
	defer func(d time.Duration) { runFinishBound = d }(runFinishBound)
	runFinishBound = 300 * time.Millisecond
	dir := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, dir, 60)
	s := &failingPuts{Store: dir, part: "/result.json"}
	w := execWorker(t, s, &fakeLauncher{err: activator.ErrCmdClaim}, "boot-1")
	rt, _ := runOwnedRoute(runHost)(runHost)
	w.execute(context.Background(), rec, req, *rt, newRunHandle())
	if r := currentRun(t, dir); r.State != provision.RunPicked {
		t.Fatalf("run %+v, want still picked", r)
	}
}

// The done CAS is only for a run the record names as this box's under this
// boot (RunFinish does not check), and a box writes no result for a run the
// record gives to another: whoever holds it ends it.
func TestRunFinishesOnlyItsOwnPick(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60) // picked under boot-1
	h := &fakeHandle{}
	w := execWorker(t, s, &fakeLauncher{handle: h}, "boot-2")
	rt, _ := runOwnedRoute(runHost)(runHost)
	w.active[req.ID] = newRunHandle()
	w.execute(context.Background(), rec, req, *rt, w.active[req.ID])
	if res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID); err != nil || res != nil {
		t.Fatalf("result %+v, %v: a box wrote a result for another boot's run", res, err)
	}
	if r := currentRun(t, s); h.stops != 0 || r.State != provision.RunPicked || r.BootID != "boot-1" {
		t.Fatalf("run %+v, stops %d: another boot's run was launched or finished", r, h.stops)
	}
	if w.active[req.ID] != nil {
		t.Error("the run was not forgotten")
	}
	for _, mut := range []func(*provision.RunState){
		func(r *provision.RunState) { r.Box = "box-2" },
		func(r *provision.RunState) { r.BootID = "boot-2" },
		func(r *provision.RunState) { r.ID = "run_bbbbbbbbbbbbbbbbbbbbbbbbbb" },
	} {
		cur := rec
		r := *rec.ActiveRun
		mut(&r)
		cur.ActiveRun = &r
		if err := runFinishOurs(req.ID, "box-1", "boot-1")(&cur); !errors.Is(err, provision.ErrRunConflict) {
			t.Errorf("run %+v: err %v, want a conflict", r, err)
		}
	}
	cur := rec
	if err := runFinishOurs(req.ID, "box-1", "boot-1")(&cur); err != nil || cur.ActiveRun.State != provision.RunDone {
		t.Fatalf("own run: %v, %+v", err, cur.ActiveRun)
	}
}

// stalledInput blocks every read of a run input until its context ends.
type stalledInput struct{ release.Store }

func (s stalledInput) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.Contains(key, "/in/") {
		<-ctx.Done()
		return nil, "", ctx.Err()
	}
	return s.Store.Get(ctx, key)
}

// A download that stalls ends at the run's deadline: the run ends as a
// failed start, with its result and the done CAS, and does not keep the
// activator's hold (and every later run on the instance) forever.
func TestRunStagingEndsAtTheRunsDeadline(t *testing.T) {
	dir := release.NewDirStore(t.TempDir())
	rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
	req, err := runs.ReadRequest(context.Background(), dir, rec.Account, rec.App, id)
	if err != nil {
		t.Fatal(err)
	}
	req.Files = []runs.File{{Name: "in.csv", Size: 4, SHA256: strings.Repeat("0", 64)}}
	got, err := provision.SetRecordRun(context.Background(), dir, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(300*time.Millisecond)))
	if err != nil {
		t.Fatal(err)
	}
	l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
	start := time.Now()
	res := runWith(t, stalledInput{dir}, l, *got, *req, newRunHandle())
	if res.Interrupted != runs.InterruptedStartFailed || l.started() != 0 || time.Since(start) > 5*time.Second {
		t.Fatalf("result %+v after %s, units %d", res, time.Since(start), l.started())
	}
	if r := currentRun(t, dir); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
}

// stoppedDuringPrepare runs Prepare and then reports an instance stop, as
// RunCmd does when a stop overlapped Prepare: the stop wins over Prepare's
// own error.
type stoppedDuringPrepare struct{}

func (stoppedDuringPrepare) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
	_ = spec.Prepare(ctx)
	return nil, &activator.CmdStoppedError{Reason: activator.StopRestart}
}

// A run Prepare found is no longer this box's gets no result, even when a
// stop overlapping Prepare is what RunCmd reports.
func TestRunNotOursGetsNoResultWhenAStopOverlapsPrepare(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60) // picked under boot-1
	w := execWorker(t, s, stoppedDuringPrepare{}, "boot-2")
	rt, _ := runOwnedRoute(runHost)(runHost)
	w.execute(context.Background(), rec, req, *rt, newRunHandle())
	if res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID); err != nil || res != nil {
		t.Fatalf("result %+v, %v: a box wrote a result for a run the record gives another boot", res, err)
	}
	if r := currentRun(t, s); r.State != provision.RunPicked || r.BootID != "boot-1" {
		t.Fatalf("run %+v", r)
	}
}

// waitsForSwitch is RunCmd as it waits for a swap's switch after Prepare:
// it holds until the launch context ends, as the activator's wait does
// under the run's deadline and cancel, and starts nothing.
type waitsForSwitch struct {
	waiting chan struct{}
	units   int
}

func (l *waitsForSwitch) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
	if err := spec.Prepare(ctx); err != nil {
		return nil, err
	}
	close(l.waiting)
	<-ctx.Done()
	return nil, ctx.Err()
}

// A run cancelled, or past its deadline, while RunCmd waits for a swap's
// switch ends with its own cancelled or timed-out result, and no unit
// starts (re-review 5): the launch runs under the run's deadline and
// cancel, not the worker's background context.
func TestRunEndedDuringASwitchWaitGetsItsOwnResult(t *testing.T) {
	t.Run("cancelled", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		rec, req := pickedRun(t, s, 60)
		l := &waitsForSwitch{waiting: make(chan struct{})}
		h := newRunHandle()
		go func() { <-l.waiting; h.requestCancel() }()
		res := runOn(t, execWorker(t, s, l, "boot-1"), s, rec, req, h)
		if !res.Cancelled || res.TimedOut || l.units != 0 || !res.Started.IsZero() {
			t.Fatalf("result %+v", res)
		}
	})
	t.Run("expired", func(t *testing.T) {
		dir := release.NewDirStore(t.TempDir())
		rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
		req, err := runs.ReadRequest(context.Background(), dir, rec.Account, rec.App, id)
		if err != nil {
			t.Fatal(err)
		}
		got, err := provision.SetRecordRun(context.Background(), dir, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(300*time.Millisecond)))
		if err != nil {
			t.Fatal(err)
		}
		l := &waitsForSwitch{waiting: make(chan struct{})}
		start := time.Now()
		res := runOn(t, execWorker(t, dir, l, "boot-1"), dir, *got, *req, newRunHandle())
		if !res.TimedOut || res.Cancelled || l.units != 0 || time.Since(start) > 5*time.Second {
			t.Fatalf("result %+v after %s", res, time.Since(start))
		}
	})
}

// blockOneRead holds the first store read after it is armed until the
// read's context ends: the post-switch record read, cut short.
type blockOneRead struct {
	release.Store
	armed   atomic.Bool
	reading chan struct{}
}

func (b *blockOneRead) Get(ctx context.Context, key string) ([]byte, string, error) {
	if b.armed.CompareAndSwap(true, false) {
		close(b.reading)
		<-ctx.Done()
		return nil, "", ctx.Err()
	}
	return b.Store.Get(ctx, key)
}

// rechecksAfterSwitch is RunCmd past a switch wait: Prepare, then the
// post-switch Recheck, whose record read the store holds. It starts
// nothing, and returns the Recheck's answer as RunCmd does.
type rechecksAfterSwitch struct{ store *blockOneRead }

func (l *rechecksAfterSwitch) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
	if err := spec.Prepare(ctx); err != nil {
		return nil, err
	}
	l.store.armed.Store(true)
	return nil, spec.Recheck(ctx)
}

// A cancel or the deadline that cuts short the post-switch record read is
// the run's own end, cancelled (130) or timed out (124), never "start
// failed" (re-review 7).
func TestRunEndedDuringThePostSwitchRecheckGetsItsOwnResult(t *testing.T) {
	t.Parallel()
	t.Run("cancelled", func(t *testing.T) {
		s := &blockOneRead{Store: release.NewDirStore(t.TempDir()), reading: make(chan struct{})}
		rec, req := pickedRun(t, s, 60)
		h := newRunHandle()
		go func() { <-s.reading; h.requestCancel() }()
		res := runOn(t, execWorker(t, s, &rechecksAfterSwitch{store: s}, "boot-1"), s, rec, req, h)
		if !res.Cancelled || res.TimedOut || res.Interrupted != "" || !res.Started.IsZero() {
			t.Fatalf("result %+v, want cancelled", res)
		}
	})
	t.Run("expired", func(t *testing.T) {
		s := &blockOneRead{Store: release.NewDirStore(t.TempDir()), reading: make(chan struct{})}
		rec, id := queueRun(t, s, time.Now().Add(time.Minute))
		req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
		if err != nil {
			t.Fatal(err)
		}
		got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(300*time.Millisecond)))
		if err != nil {
			t.Fatal(err)
		}
		res := runOn(t, execWorker(t, s, &rechecksAfterSwitch{store: s}, "boot-1"), s, *got, *req, newRunHandle())
		if !res.TimedOut || res.Cancelled || res.Interrupted != "" || !res.Started.IsZero() {
			t.Fatalf("result %+v, want timed out", res)
		}
	})
}

func TestExitOfReadsNumbersAndWords(t *testing.T) {
	for _, c := range []struct {
		code, status string
		exit         int
		signal       string
	}{
		{"1", "3", 3, ""}, {"exited", "3", 3, ""}, {"2", "9", 0, "9"}, {"killed", "15", 0, "15"}, {"3", "6", 0, "6"}, {"dumped", "6", 0, "6"},
	} {
		exit, signal := exitOf(instancectl.CmdStatus{Code: c.code, Status: c.status})
		if exit != c.exit || signal != c.signal {
			t.Errorf("%s/%s: %d %q", c.code, c.status, exit, signal)
		}
	}
}