//go:build linux

package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"database/sql"
	"database/sql/driver"
	"encoding/hex"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"math/big"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/schedule"

	_ "modernc.org/sqlite"
)

const (
	testHost    = "h1.example.com"
	adminToken  = "admin-token-for-tests"
	probeSecret = "probe-secret-for-tests"
	platformKey = "platform-instance-key-for-tests"
	accessKey   = "AKIDMOVEPROBETEST"
	bucketName  = "probe-bucket"
	storeRegion = "eu-west-1"
)

// binPath is the moveprobe binary the tests start, built once in TestMain,
// so every test starts the app the way the exec backend does: a separate
// process given --socket and --db.
var binPath string

func TestMain(m *testing.M) {
	dir, err := os.MkdirTemp("", "mpbin")
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	binPath = filepath.Join(dir, "moveprobe")
	if out, err := exec.Command("go", "build", "-o", binPath, ".").CombinedOutput(); err != nil {
		fmt.Fprintf(os.Stderr, "build moveprobe: %v\n%s", err, out)
		os.RemoveAll(dir)
		os.Exit(1)
	}
	code := m.Run()
	os.RemoveAll(dir)
	os.Exit(code)
}

// shortTempDir keeps unix socket paths under the kernel's length limit
// (as internal/activator/rolelaunch_root_linux_test.go does); t.TempDir's
// names carry the test's name and can pass 107 bytes.
func shortTempDir(t *testing.T) string {
	t.Helper()
	dir, err := os.MkdirTemp("", "mp")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	return dir
}

// fixture is one instance's world: its data directory (the flat layout,
// <dir>/h1.db and <dir>/h1/), a test S3 holding its store, and a control
// socket served by the platform's own schedule verbs over a registry.
// Each test makes its own, and shares nothing else but the read-only
// binPath, so every test runs in parallel.
type fixture struct {
	t        *testing.T
	sockDir  string
	dataDir  string
	db       string
	bucket   *leasetest.VersionedBucket
	store    http.Handler
	storeURL string
	reg      *registry.Registry
	control  string
	env      map[string]string

	mu       sync.Mutex
	auths    []string
	storeReq atomic.Int64
	// storeDelay slows every store request, in milliseconds.
	storeDelay atomic.Int64
	// controlConns counts the connections the control socket accepted.
	controlConns atomic.Int64
	starts       int
}

func newFixture(t *testing.T) *fixture {
	t.Helper()
	f := &fixture{t: t, sockDir: shortTempDir(t), dataDir: t.TempDir()}
	f.db = filepath.Join(f.dataDir, "h1.db")

	f.bucket = leasetest.NewVersionedBucket()
	f.bucket.AddBucket(bucketName)
	s3h, _ := leasetest.NewS3Server(f.bucket)
	f.store = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		f.storeReq.Add(1)
		time.Sleep(time.Duration(f.storeDelay.Load()) * time.Millisecond)
		f.mu.Lock()
		f.auths = append(f.auths, r.Header.Get("Authorization"))
		f.mu.Unlock()
		s3h.ServeHTTP(w, r)
	})
	srv := httptest.NewServer(f.store)
	t.Cleanup(srv.Close)
	f.storeURL = srv.URL

	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	f.reg = reg
	f.control = filepath.Join(f.sockDir, "c.sock")
	ln, err := net.Listen("unix", f.control)
	if err != nil {
		t.Fatal(err)
	}
	mux := http.NewServeMux()
	schedule.Mount(mux, testHost, reg, func() (string, error) { return adminToken, nil }, time.Now)
	cs := &http.Server{Handler: mux, ConnState: func(_ net.Conn, s http.ConnState) {
		if s == http.StateNew {
			f.controlConns.Add(1)
		}
	}}
	go cs.Serve(ln)
	t.Cleanup(func() { cs.Close() })

	// The environment agentOwnedVars composes over <host>.env: the control
	// socket (when the listener bound), the purge channel, the region (when
	// the box has one) and the host; then the env file's own variables.
	f.env = map[string]string{
		"CARLOS_CONTROL_SOCKET": f.control,
		"CARLOS_PURGE":          "header",
		"CARLOS_REGION":         "uk",
		"CARLOS_HOST":           testHost,
		"CARLOS_ADMIN_TOKEN":    adminToken,
		"CARLOS_INSTANCE_KEY":   platformKey,
		"MOVEPROBE_BUCKET":      bucketName,
		"MOVEPROBE_REGION":      storeRegion,
		"MOVEPROBE_ENDPOINT":    f.storeURL,
		"MOVEPROBE_ACCESS_KEY":  accessKey,
		"MOVEPROBE_SECRET_KEY":  "secret-key-for-tests",
		"MOVEPROBE_SECRET":      probeSecret,
	}
	// Registered first, so it runs last: every probe's main is gone by then.
	t.Cleanup(func() { killWorkers(f.db) })
	return f
}

func (f *fixture) seenAuth() []string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]string(nil), f.auths...)
}

// envWith is the fixture's environment with changes applied; an empty
// value removes the variable.
func (f *fixture) envWith(changes map[string]string) map[string]string {
	out := map[string]string{}
	for k, v := range f.env {
		out[k] = v
	}
	for k, v := range changes {
		if v == "" {
			delete(out, k)
		} else {
			out[k] = v
		}
	}
	return out
}

// childEnv mirrors the backend's: the parent's environment without any
// CARLOS_ variable, then the instance's variables. The test also drops the
// probe's and AWS's own names and every proxy setting, so nothing from the
// shell running the tests reaches the app.
func childEnv(vars map[string]string) []string {
	var out []string
	for _, kv := range os.Environ() {
		k, _, _ := strings.Cut(kv, "=")
		switch {
		case strings.HasPrefix(k, "CARLOS_"), strings.HasPrefix(k, "MOVEPROBE_"), strings.HasPrefix(k, "AWS_"):
			continue
		case strings.EqualFold(k, "HTTP_PROXY"), strings.EqualFold(k, "HTTPS_PROXY"), strings.EqualFold(k, "NO_PROXY"), k == "SSL_CERT_FILE", k == "SSL_CERT_DIR":
			continue
		}
		out = append(out, kv)
	}
	for k, v := range vars {
		out = append(out, k+"="+v)
	}
	return out
}

type lockedBuffer struct {
	mu sync.Mutex
	b  bytes.Buffer
}

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

func (l *lockedBuffer) String() string {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.b.String()
}

type probe struct {
	t    *testing.T
	cmd  *exec.Cmd
	sock string
	out  *lockedBuffer
	done chan struct{}
}

// start runs the binary exactly as spawnChildFrom does:
// exec.Command(bin, "--socket", sock, "--db", dbPath) with the composed env.
// Its stdout and stderr are pipes, as the backend's tee makes them.
func (f *fixture) start(env map[string]string) *probe {
	f.t.Helper()
	return f.startBin(binPath, env)
}

// startBin is start with another build of moveprobe.
func (f *fixture) startBin(bin string, env map[string]string) *probe {
	f.t.Helper()
	f.starts++
	p := &probe{t: f.t, sock: filepath.Join(f.sockDir, fmt.Sprintf("a%d.sock", f.starts)), out: &lockedBuffer{}, done: make(chan struct{})}
	p.cmd = exec.Command(bin, "--socket", p.sock, "--db", f.db)
	p.cmd.Env = childEnv(env)
	p.cmd.Stdout = p.out
	p.cmd.Stderr = p.out
	if err := p.cmd.Start(); err != nil {
		f.t.Fatal(err)
	}
	go func() { p.cmd.Wait(); close(p.done) }()
	f.t.Cleanup(p.kill)
	return p
}

func (p *probe) exited() bool {
	select {
	case <-p.done:
		return true
	default:
		return false
	}
}

func (p *probe) kill() {
	if p.exited() {
		return
	}
	p.cmd.Process.Kill()
	<-p.done
}

// stop is the backend's Stop: SIGTERM to the leader only.
func (p *probe) stop() {
	p.t.Helper()
	if p.exited() {
		return
	}
	p.cmd.Process.Signal(syscall.SIGTERM)
	select {
	case <-p.done:
	case <-time.After(10 * time.Second):
		p.t.Errorf("moveprobe did not exit on SIGTERM within 10s; output:\n%s", p.out)
		p.kill()
	}
}

func unixClient(sock string) *http.Client {
	return &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{
		DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
			var d net.Dialer
			return d.DialContext(ctx, "unix", sock)
		},
	}}
}

func (p *probe) do(method, path, bearer string, hdr map[string]string) (int, string, error) {
	req, err := http.NewRequest(method, "http://"+testHost+path, nil)
	if err != nil {
		return 0, "", err
	}
	if bearer != "" {
		req.Header.Set("Authorization", "Bearer "+bearer)
	}
	for k, v := range hdr {
		req.Header.Set(k, v)
	}
	resp, err := unixClient(p.sock).Do(req)
	if err != nil {
		return 0, "", err
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	return resp.StatusCode, string(b), nil
}

// wait polls path until it answers want, and fails the test with the last
// answer and the app's output otherwise.
func (p *probe) wait(path string, want int) string {
	p.t.Helper()
	deadline := time.Now().Add(20 * time.Second)
	var last string
	for time.Now().Before(deadline) {
		code, body, err := p.do(http.MethodGet, path, "", nil)
		if err == nil && code == want {
			return body
		}
		last = fmt.Sprintf("code %d body %q err %v", code, body, err)
		if p.exited() {
			break
		}
		time.Sleep(100 * time.Millisecond)
	}
	p.t.Fatalf("GET %s never answered %d; last: %s; output:\n%s", path, want, last, p.out)
	return ""
}

type reportDoc struct {
	Region     string `json:"region"`
	RegionFrom string `json:"region_from"`
	Mode       string `json:"mode"`
	Rows       []struct {
		Kind   string `json:"kind"`
		Region string `json:"region"`
		Count  int64  `json:"count"`
		MinSeq int64  `json:"min_seq"`
		MaxSeq int64  `json:"max_seq"`
	} `json:"rows"`
	Fired []struct {
		Kind  string `json:"kind"`
		Name  string `json:"name"`
		DueAt int64  `json:"due_at"`
		Count int64  `json:"count"`
	} `json:"fired"`
	FileKeySHA256     string `json:"file_key_sha256"`
	PlatformKeySHA256 string `json:"platform_key_sha256"`
	Object            struct {
		Key   string `json:"key"`
		Body  string `json:"body"`
		Error string `json:"error"`
	} `json:"object"`
	HasSecret             bool   `json:"has_secret"`
	WorkerPID             int    `json:"worker_pid"`
	WorkerWritesFailed    int64  `json:"worker_writes_failed"`
	WorkerWritesCommitted int64  `json:"worker_writes_committed"`
	WorkerFailedInARow    int64  `json:"worker_failed_in_a_row"`
	WorkerLastError       string `json:"worker_last_error"`
	WorkerExit            string `json:"worker_exit"`
	WorkerLastStatusAt    int64  `json:"worker_last_status_at"`
}

func (r reportDoc) count(kind, region string) int64 {
	for _, row := range r.Rows {
		if row.Kind == kind && row.Region == region {
			return row.Count
		}
	}
	return 0
}

func (p *probe) report() reportDoc {
	p.t.Helper()
	code, body, err := p.do(http.MethodGet, "/report", probeSecret, nil)
	if err != nil || code != http.StatusOK {
		p.t.Fatalf("/report: code %d err %v body %q; output:\n%s", code, err, body, p.out)
	}
	var r reportDoc
	if err := json.Unmarshal([]byte(body), &r); err != nil {
		p.t.Fatalf("/report: %v in %q", err, body)
	}
	return r
}

// workerPIDs lists the pids whose command line starts with a moveprobe
// build the tests made (binPath, or a test's own, all named moveprobe*) and
// "worker", and names db.
func workerPIDs(db string) []int {
	var pids []int
	ents, _ := os.ReadDir("/proc")
	for _, e := range ents {
		pid, err := strconv.Atoi(e.Name())
		if err != nil {
			continue
		}
		raw, err := os.ReadFile(filepath.Join("/proc", e.Name(), "cmdline"))
		if err != nil {
			continue
		}
		args := strings.Split(strings.TrimRight(string(raw), "\x00"), "\x00")
		if len(args) < 2 || !strings.HasPrefix(filepath.Base(args[0]), "moveprobe") || args[1] != "worker" || !strings.Contains(string(raw), db) {
			continue
		}
		if alive(pid) {
			pids = append(pids, pid)
		}
	}
	return pids
}

// alive is true while pid exists and is neither a zombie nor dead.
func alive(pid int) bool {
	raw, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid))
	if err != nil {
		return false
	}
	i := strings.LastIndexByte(string(raw), ')')
	if i < 0 || i+2 >= len(raw) {
		return false
	}
	state := raw[i+2]
	return state != 'Z' && state != 'X'
}

func killWorkers(db string) {
	for _, pid := range workerPIDs(db) {
		syscall.Kill(pid, syscall.SIGKILL)
	}
	deadline := time.Now().Add(5 * time.Second)
	for len(workerPIDs(db)) > 0 && time.Now().Before(deadline) {
		time.Sleep(50 * time.Millisecond)
	}
}

func testDB(t *testing.T, db string) *sql.DB {
	t.Helper()
	d, err := sql.Open("sqlite", "file:"+db+"?_pragma=busy_timeout(5000)")
	if err != nil {
		t.Fatal(err)
	}
	return d
}

func countRows(t *testing.T, db, kind, region string) int64 {
	t.Helper()
	d := testDB(t, db)
	defer d.Close()
	var n int64
	if err := d.QueryRow(`SELECT count(*) FROM rows WHERE kind=? AND region=?`, kind, region).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

// checkpoint folds the WAL into the database file and closes, so the file
// is what a move would install.
func checkpoint(t *testing.T, db string) {
	t.Helper()
	d := testDB(t, db)
	if _, err := d.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil {
		t.Fatal(err)
	}
	d.Close()
}

// snapshot is every entry under dir with its size and SHA-256: the
// database file's digest and the files directory's listing, and anything
// else that appears beside them (a -wal or -shm file included).
func snapshot(t *testing.T, dir string) string {
	t.Helper()
	var lines []string
	err := filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}
		rel, _ := filepath.Rel(dir, p)
		if d.IsDir() {
			lines = append(lines, rel+"/")
			return nil
		}
		b, err := os.ReadFile(p)
		if err != nil {
			return err
		}
		sum := sha256.Sum256(b)
		lines = append(lines, fmt.Sprintf("%s %d %x", rel, len(b), sum))
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	sort.Strings(lines)
	return strings.Join(lines, "\n")
}

func sha(s string) string {
	sum := sha256.Sum256([]byte(s))
	return hex.EncodeToString(sum[:])
}

// fire delivers a timer the way the platform's runner does: the request
// schedule.BuildRequest builds, sent over the instance's socket.
func (p *probe) fire(tm registry.Timer, token string) (int, string) {
	p.t.Helper()
	req, err := schedule.BuildRequest(registry.Route{Host: testHost}, tm, token, "test")
	if err != nil {
		p.t.Fatal(err)
	}
	resp, err := unixClient(p.sock).Do(req)
	if err != nil {
		p.t.Fatal(err)
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	return resp.StatusCode, string(b)
}

// ---- the launch contract --------------------------------------------------

func TestMoveprobeLaunchContract(t *testing.T) {
	t.Parallel()
	t.Run("the agent's variables and a store endpoint", func(t *testing.T) {
		f := newFixture(t)
		p := f.start(f.env)
		if body := p.wait("/ready", http.StatusOK); !strings.Contains(body, "ready") {
			t.Errorf("/ready body %q", body)
		}
		// The platform probes "/" when moves.ready_path is unset.
		p.wait("/", http.StatusOK)

		vs, err := f.bucket.ListVersions(context.Background(), bucketName, "probe/")
		if err != nil || len(vs) != 1 {
			t.Fatalf("probe objects: %v %v", vs, err)
		}
		for _, a := range f.seenAuth() {
			if !strings.Contains(a, "Credential="+accessKey+"/") {
				t.Errorf("a store request was not signed with MOVEPROBE_ACCESS_KEY: %q", a)
			}
		}

		// /arm-once and /report need the probe's bearer.
		if code, _, _ := p.do(http.MethodGet, "/report", "", nil); code != http.StatusUnauthorized {
			t.Errorf("GET /report without a bearer: %d, want 401", code)
		}
		if code, _, _ := p.do(http.MethodGet, "/report", "wrong", nil); code != http.StatusUnauthorized {
			t.Errorf("GET /report with the wrong bearer: %d, want 401", code)
		}
		if code, _, _ := p.do(http.MethodPost, "/arm-once?in=600", "", nil); code != http.StatusUnauthorized {
			t.Errorf("POST /arm-once without a bearer: %d, want 401", code)
		}
		before := time.Now().Unix()
		code, body, err := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil)
		if err != nil || code != http.StatusOK {
			t.Fatalf("POST /arm-once: %d %q %v", code, body, err)
		}
		tm, err := f.reg.GetTimer(testHost, "once-1")
		if err != nil {
			t.Fatalf("once-1 not registered through the control socket: %v (arm answered %q)", err, body)
		}
		if tm.Kind != schedule.KindOnce || tm.Path != "/once" || tm.DueAt < before+600 || tm.DueAt > time.Now().Unix()+601 {
			t.Errorf("once-1 = %+v, want a once-timer for /once due in 600s", tm)
		}
		if code, body, _ := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil); code != http.StatusOK {
			t.Errorf("the second arm: %d %q", code, body)
		}
		if _, err := f.reg.GetTimer(testHost, "once-2"); err != nil {
			t.Errorf("the second arm is not once-2: %v", err)
		}

		r := p.report()
		if r.Region != "uk" || r.RegionFrom != "CARLOS_REGION" {
			t.Errorf("report region %q from %q, want uk from CARLOS_REGION", r.Region, r.RegionFrom)
		}
		if r.PlatformKeySHA256 != sha(platformKey) || r.FileKeySHA256 == "" || !r.HasSecret || r.Object.Body == "" || r.Object.Error != "" {
			t.Errorf("report = %+v", r)
		}
		p.stop()
	})

	t.Run("no CARLOS_REGION and no control socket", func(t *testing.T) {
		f := newFixture(t)
		p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "", "CARLOS_CONTROL_SOCKET": ""}))
		p.wait("/ready", http.StatusOK)
		code, body, _ := p.do(http.MethodPost, "/arm-once?in=60", probeSecret, nil)
		if code != http.StatusServiceUnavailable || !strings.Contains(body, "CARLOS_CONTROL_SOCKET") {
			t.Errorf("/arm-once with no control socket: %d %q, want 503 naming CARLOS_CONTROL_SOCKET", code, body)
		}
		p.wait("/ready", http.StatusOK) // unaffected
		r := p.report()
		if r.Region != "home" || !strings.Contains(r.RegionFrom, "unset") {
			t.Errorf("report region %q from %q, want home, with CARLOS_REGION unset", r.Region, r.RegionFrom)
		}
		time.Sleep(700 * time.Millisecond)
		if n := countRows(t, f.db, "bg", "home"); n == 0 {
			t.Errorf("no bg rows written as region home")
		}
		p.stop()
	})

	t.Run("no MOVEPROBE_ENDPOINT: AWS's own host, the delivered keys", func(t *testing.T) {
		f := newFixture(t)
		proxy, caFile, connects := awsLookalike(t, f.store)
		p := f.start(f.envWith(map[string]string{
			"MOVEPROBE_ENDPOINT": "",
			"HTTPS_PROXY":        proxy,
			"SSL_CERT_FILE":      caFile,
		}))
		p.wait("/ready", http.StatusOK)
		want := bucketName + ".s3." + storeRegion + ".amazonaws.com:443"
		got := connects()
		if len(got) == 0 {
			t.Fatalf("no store request went to AWS's host")
		}
		for _, h := range got {
			if h != want {
				t.Errorf("store request to %q, want %q (virtual-hosted, no endpoint)", h, want)
			}
		}
		for _, a := range f.seenAuth() {
			if !strings.Contains(a, "Credential="+accessKey+"/") {
				t.Errorf("a store request was not signed with MOVEPROBE_ACCESS_KEY: %q", a)
			}
		}
		p.stop()
	})

	t.Run("the store keys come only from MOVEPROBE_", func(t *testing.T) {
		f := newFixture(t)
		p := f.start(f.envWith(map[string]string{
			"MOVEPROBE_ACCESS_KEY":  "",
			"AWS_ACCESS_KEY_ID":     accessKey,
			"AWS_SECRET_ACCESS_KEY": "x",
			"AWS_ENDPOINT_URL":      f.storeURL,
			"AWS_ENDPOINT_URL_S3":   f.storeURL,
		}))
		body := p.wait("/ready", http.StatusInternalServerError)
		if !strings.Contains(body, "MOVEPROBE_ACCESS_KEY") {
			t.Errorf("/ready body %q does not name MOVEPROBE_ACCESS_KEY", body)
		}
		if n := f.storeReq.Load(); n != 0 {
			t.Errorf("%d store requests made without the binding's keys", n)
		}
		p.stop()
	})

	t.Run("the sealed secret is required", func(t *testing.T) {
		f := newFixture(t)
		p := f.start(f.envWith(map[string]string{"MOVEPROBE_SECRET": ""}))
		body := p.wait("/ready", http.StatusInternalServerError)
		if !strings.Contains(body, "MOVEPROBE_SECRET") {
			t.Errorf("/ready body %q does not name MOVEPROBE_SECRET", body)
		}
		p.stop()
	})
}

// awsLookalike stands in for S3's regional host: an HTTPS proxy whose
// CONNECT tunnels end at a TLS server holding a certificate for
// *.s3.<region>.amazonaws.com, signed by a CA written to the returned file.
// The tunnelled requests are virtual-hosted; they reach store path-style.
func awsLookalike(t *testing.T, store http.Handler) (proxyURL, caFile string, connects func() []string) {
	t.Helper()
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	tmpl := &x509.Certificate{
		SerialNumber:          big.NewInt(1),
		Subject:               pkix.Name{CommonName: "moveprobe test s3"},
		NotBefore:             time.Now().Add(-time.Hour),
		NotAfter:              time.Now().Add(time.Hour),
		DNSNames:              []string{"*.s3." + storeRegion + ".amazonaws.com", "s3." + storeRegion + ".amazonaws.com"},
		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
		BasicConstraintsValid: true,
		IsCA:                  true,
	}
	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
	if err != nil {
		t.Fatal(err)
	}
	caFile = filepath.Join(t.TempDir(), "ca.pem")
	if err := os.WriteFile(caFile, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600); err != nil {
		t.Fatal(err)
	}
	tlsCfg := &tls.Config{Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}}

	vhost := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		bucket, _, _ := strings.Cut(r.Host, ".")
		r.URL.Path = "/" + bucket + r.URL.Path
		r.URL.RawPath = ""
		store.ServeHTTP(w, r)
	})
	var mu sync.Mutex
	var hosts []string
	proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodConnect {
			http.Error(w, "CONNECT only", http.StatusMethodNotAllowed)
			return
		}
		mu.Lock()
		hosts = append(hosts, r.Host)
		mu.Unlock()
		conn, brw, err := w.(http.Hijacker).Hijack()
		if err != nil {
			return
		}
		brw.WriteString("HTTP/1.1 200 Connection established\r\n\r\n")
		brw.Flush()
		srv := &http.Server{Handler: vhost}
		srv.Serve(&oneConn{c: tls.Server(&bufferedConn{Conn: conn, r: brw.Reader}, tlsCfg)})
	}))
	t.Cleanup(proxy.Close)
	return proxy.URL, caFile, func() []string {
		mu.Lock()
		defer mu.Unlock()
		return append([]string(nil), hosts...)
	}
}

type bufferedConn struct {
	net.Conn
	r *bufio.Reader
}

func (b *bufferedConn) Read(p []byte) (int, error) { return b.r.Read(p) }

// oneConn is a listener that yields one connection; Serve returns once it
// has been handed out and the connection is served on its own goroutine.
type oneConn struct {
	c    net.Conn
	used bool
}

func (o *oneConn) Accept() (net.Conn, error) {
	if o.used {
		return nil, io.EOF
	}
	o.used = true
	return o.c, nil
}
func (o *oneConn) Close() error   { return nil }
func (o *oneConn) Addr() net.Addr { return o.c.LocalAddr() }

// ---- the forked worker ------------------------------------------------------

// The worker's grace is a minute in every build but this test's: it links
// its own moveprobe with an 8 s grace (workerGraceOverride) and watches
// that build's worker outlive its main, keep writing, and leave on its own
// at the grace, as the shipped build does at a minute.
func TestMoveprobeWorkerOutlivesMain(t *testing.T) {
	t.Parallel()
	if workerGrace != 60*time.Second || workerGraceOverride != "" || orphanGrace() != workerGrace {
		t.Fatalf("a plain build's worker grace is %s (const %s, override %q), want a minute", orphanGrace(), workerGrace, workerGraceOverride)
	}
	const grace = 8 * time.Second
	bin := filepath.Join(t.TempDir(), "moveprobe-grace")
	build := exec.Command("go", "build", "-ldflags", "-X main.workerGraceOverride="+grace.String(), "-o", bin, ".")
	if out, err := build.CombinedOutput(); err != nil {
		t.Fatalf("build moveprobe with an %s grace: %v\n%s", grace, err, out)
	}
	f := newFixture(t)
	p := f.startBin(bin, f.env)
	p.wait("/ready", http.StatusOK)
	r := p.report()
	if r.WorkerPID <= 0 || !alive(r.WorkerPID) {
		t.Fatalf("no live worker in the report: %d", r.WorkerPID)
	}
	if got := workerPIDs(f.db); len(got) != 1 || got[0] != r.WorkerPID {
		t.Fatalf("worker processes %v, report says %d", got, r.WorkerPID)
	}

	// The main dies outright. Its stdout and stderr are pipes; a worker
	// still holding them would keep Wait from returning (the backend's tee
	// waits up to its drain grace for exactly that).
	killed := time.Now()
	p.cmd.Process.Kill()
	select {
	case <-p.done:
	case <-time.After(2 * time.Second):
		t.Fatalf("the main's Wait did not return within 2s of its death: the worker holds its stdio")
	}

	n0 := countRows(t, f.db, "worker", "uk")
	time.Sleep(3 * time.Second)
	if !alive(r.WorkerPID) {
		t.Fatalf("the worker died with the main")
	}
	if n1 := countRows(t, f.db, "worker", "uk"); n1 < n0+2 {
		t.Fatalf("the worker stopped writing after the main died: %d then %d rows", n0, n1)
	}

	// It exits on its own about its grace after its parent has gone.
	for alive(r.WorkerPID) && time.Since(killed) < grace+20*time.Second {
		time.Sleep(50 * time.Millisecond)
	}
	gone := time.Since(killed)
	if alive(r.WorkerPID) {
		t.Fatalf("the worker is still alive %s after the main died", gone.Round(time.Second))
	}
	// The worker notices its parent has gone on its next 1 s tick and
	// leaves on the first tick at or past the grace, so it exits between
	// the grace and about 2 s after it. 4 s of slack still refuses a worker
	// that waits twice its grace.
	if gone < grace-time.Second || gone > grace+4*time.Second {
		t.Errorf("the worker exited %s after the main died, want about %s", gone.Round(time.Second), grace)
	}
}

// ---- the failure switch ---------------------------------------------------

func TestReadyFailsWhereTold(t *testing.T) {
	t.Parallel()
	f := newFixture(t)

	// The installed state: a normal boot in uk, stopped, its worker killed
	// (the stop evidence does that on a real box) and the WAL folded in.
	seed := f.start(f.env)
	seed.wait("/ready", http.StatusOK)
	time.Sleep(time.Second)
	seed.stop()
	killWorkers(f.db)
	checkpoint(t, f.db)
	installed := snapshot(t, f.dataDir)
	if !strings.Contains(installed, "h1/instance.key") || !strings.Contains(installed, "h1.db ") {
		t.Fatalf("the seed boot left no database or key file:\n%s", installed)
	}
	timers, _ := f.reg.TimersFor(testHost)

	t.Run("before-write writes nothing", func(t *testing.T) {
		p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie", "MOVEPROBE_FAIL_READY_IN": "ie"}))
		p.wait("/ready", http.StatusServiceUnavailable)
		p.wait("/", http.StatusServiceUnavailable)
		if code, _, _ := p.do(http.MethodPost, "/arm-once?in=60", probeSecret, nil); code != http.StatusServiceUnavailable {
			t.Errorf("/arm-once in before-write: %d, want 503", code)
		}
		code, _ := p.fire(registry.Timer{Host: testHost, Name: "every-minute", Kind: schedule.KindSchedule, Path: "/tick", DueAt: time.Now().Unix()}, adminToken)
		if code != http.StatusServiceUnavailable {
			t.Errorf("/tick in before-write: %d, want 503", code)
		}
		code, _ = p.fire(registry.Timer{Host: testHost, Name: "once-9", Kind: schedule.KindOnce, Path: "/once", DueAt: time.Now().Unix()}, adminToken)
		if code != http.StatusServiceUnavailable {
			t.Errorf("/once in before-write: %d, want 503", code)
		}
		time.Sleep(2 * time.Second)
		if w := workerPIDs(f.db); len(w) != 0 {
			t.Errorf("before-write forked a worker: %v", w)
		}
		p.stop()
		if got := snapshot(t, f.dataDir); got != installed {
			t.Errorf("before-write changed the data directory\ninstalled:\n%s\nafter:\n%s", installed, got)
		}
		if got, _ := f.reg.TimersFor(testHost); len(got) != len(timers) {
			t.Errorf("before-write changed the timers: %v then %v", timers, got)
		}
	})

	t.Run("after-write writes one bg row, then fails", func(t *testing.T) {
		p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie", "MOVEPROBE_FAIL_READY_IN": "ie,after-write"}))
		p.wait("/ready", http.StatusServiceUnavailable)
		p.wait("/", http.StatusServiceUnavailable)
		time.Sleep(2 * time.Second)
		r := p.report()
		if r.Mode != "after-write" {
			t.Errorf("mode %q, want after-write", r.Mode)
		}
		if n := r.count("bg", "ie"); n != 1 {
			t.Errorf("after-write wrote %d bg rows in ie, want exactly 1", n)
		}
		if n := r.count("worker", "ie"); n < 1 {
			t.Errorf("after-write's worker wrote no rows in ie")
		}
		p.stop()
		killWorkers(f.db)
	})

	t.Run("another region's switch does not apply", func(t *testing.T) {
		p := f.start(f.envWith(map[string]string{"MOVEPROBE_FAIL_READY_IN": "ie"}))
		p.wait("/ready", http.StatusOK)
		p.stop()
		killWorkers(f.db)
	})

	t.Run("a box with no region is home", func(t *testing.T) {
		checkpoint(t, f.db)
		before := snapshot(t, f.dataDir)
		p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "", "MOVEPROBE_FAIL_READY_IN": "home"}))
		p.wait("/ready", http.StatusServiceUnavailable)
		time.Sleep(time.Second)
		p.stop()
		if got := snapshot(t, f.dataDir); got != before {
			t.Errorf("before-write at home changed the data directory")
		}
	})
}

// ---- the report -----------------------------------------------------------

func TestReportCountsRowsByRegion(t *testing.T) {
	t.Parallel()
	f := newFixture(t)
	uk := f.start(f.env)
	uk.wait("/ready", http.StatusOK)

	code, body, _ := uk.do(http.MethodPost, "/arm-once?in=1", probeSecret, nil)
	if code != http.StatusOK {
		t.Fatalf("/arm-once: %d %q", code, body)
	}
	once, err := f.reg.GetTimer(testHost, "once-1")
	if err != nil {
		t.Fatal(err)
	}
	// A tick and the once-timer, delivered as the runner delivers them.
	tick := registry.Timer{Host: testHost, Name: "every-minute", Kind: schedule.KindSchedule, Path: "/tick", DueAt: 1_700_000_040}
	if code, body := uk.fire(tick, adminToken); code != http.StatusOK {
		t.Fatalf("/tick: %d %q", code, body)
	}
	if code, body := uk.fire(*once, adminToken); code != http.StatusOK {
		t.Fatalf("/once: %d %q", code, body)
	}
	// Neither fires without the platform's token.
	if code, _ := uk.fire(tick, "wrong"); code != http.StatusUnauthorized {
		t.Errorf("/tick with the wrong token: %d, want 401", code)
	}
	if code, _ := uk.fire(*once, ""); code != http.StatusUnauthorized {
		t.Errorf("/once with no token: %d, want 401", code)
	}
	time.Sleep(1500 * time.Millisecond)
	ukReport := uk.report()
	uk.stop()
	killWorkers(f.db)

	// The same database, now in ie.
	ie := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie"}))
	ie.wait("/ready", http.StatusOK)
	time.Sleep(1500 * time.Millisecond)
	r := ie.report()
	ie.stop()

	if r.Region != "ie" {
		t.Errorf("region %q, want ie", r.Region)
	}
	for _, c := range []struct {
		kind, region string
		min          int64
	}{{"bg", "uk", 2}, {"bg", "ie", 2}, {"worker", "uk", 1}, {"worker", "ie", 1}, {"tick", "uk", 1}, {"once", "uk", 1}} {
		if n := r.count(c.kind, c.region); n < c.min {
			t.Errorf("%s rows in %s: %d, want at least %d (rows %+v)", c.kind, c.region, n, c.min, r.Rows)
		}
	}
	if n := r.count("tick", "ie") + r.count("once", "ie"); n != 0 {
		t.Errorf("%d timer rows in ie, none fired there", n)
	}
	// No background row is missing: the sequence runs on across boots.
	var bgCount, bgMax int64
	for _, row := range r.Rows {
		if row.Kind == "bg" {
			bgCount += row.Count
			bgMax = max(bgMax, row.MaxSeq)
		}
	}
	if bgCount != bgMax {
		t.Errorf("bg rows %d, highest seq %d: a row is missing or repeated", bgCount, bgMax)
	}

	var gotOnce, gotTick bool
	for _, fr := range r.Fired {
		switch {
		case fr.Kind == "once" && fr.Name == "once-1":
			gotOnce = fr.Count == 1 && fr.DueAt == once.DueAt
		case fr.Kind == "tick" && fr.Name == "every-minute":
			gotTick = fr.Count == 1 && fr.DueAt == 1_700_000_040
		}
	}
	if !gotOnce || !gotTick {
		t.Errorf("fired = %+v, want once-1 once at %d and every-minute once at 1700000040", r.Fired, once.DueAt)
	}
	if r.FileKeySHA256 == "" || r.FileKeySHA256 != ukReport.FileKeySHA256 {
		t.Errorf("file key fingerprint %q in ie, %q in uk", r.FileKeySHA256, ukReport.FileKeySHA256)
	}
	if r.PlatformKeySHA256 != sha(platformKey) {
		t.Errorf("platform key fingerprint %q, want %q", r.PlatformKeySHA256, sha(platformKey))
	}
	if !r.HasSecret || r.Object.Body == "" || !strings.HasPrefix(r.Object.Key, "probe/") {
		t.Errorf("report = %+v", r)
	}
}

// ---- the instance key -----------------------------------------------------

func TestReadyChecksBothKeys(t *testing.T) {
	t.Parallel()
	f := newFixture(t)
	p := f.start(f.env)
	p.wait("/ready", http.StatusOK)
	p.stop()
	killWorkers(f.db)

	keyFile := filepath.Join(strings.TrimSuffix(f.db, ".db"), "instance.key")
	orig, err := os.ReadFile(keyFile)
	if err != nil {
		t.Fatalf("first boot wrote no key file beside the database: %v", err)
	}
	if b, err := hex.DecodeString(strings.TrimSpace(string(orig))); err != nil || len(b) != 32 {
		t.Errorf("instance.key is not 32 bytes of hex: %q", orig)
	}

	check := func(name string, env map[string]string, want int, says string) {
		t.Helper()
		p := f.start(env)
		body := p.wait("/ready", want)
		if says != "" && !strings.Contains(body, says) {
			t.Errorf("%s: /ready body %q, want it to say %q", name, body, says)
		}
		p.stop()
		killWorkers(f.db)
	}

	if err := os.WriteFile(keyFile, []byte(strings.Repeat("ab", 32)), 0o600); err != nil {
		t.Fatal(err)
	}
	check("changed file", f.env, http.StatusInternalServerError, "key mismatch")
	if err := os.Remove(keyFile); err != nil {
		t.Fatal(err)
	}
	check("missing file", f.env, http.StatusInternalServerError, "key mismatch")
	if err := os.WriteFile(keyFile, orig, 0o600); err != nil {
		t.Fatal(err)
	}
	check("restored file", f.env, http.StatusOK, "")
	check("platform key changed", f.envWith(map[string]string{"CARLOS_INSTANCE_KEY": "another-key"}), http.StatusInternalServerError, "CARLOS_INSTANCE_KEY")
	check("platform key absent", f.envWith(map[string]string{"CARLOS_INSTANCE_KEY": ""}), http.StatusInternalServerError, "CARLOS_INSTANCE_KEY")
	check("both restored", f.env, http.StatusOK, "")
}

// ---- write failures -------------------------------------------------------

// failInserts makes every insert of kind into rows abort, through a trigger
// the test adds to the instance's own database: a real write failure, with
// nothing in the app to switch it on.
func failInserts(t *testing.T, db, kind string) {
	t.Helper()
	d := testDB(t, db)
	defer d.Close()
	q := fmt.Sprintf(`CREATE TRIGGER fail_%[1]s BEFORE INSERT ON rows WHEN NEW.kind = '%[1]s' BEGIN SELECT RAISE(ABORT, 'forced %[1]s failure'); END`, kind)
	if _, err := d.Exec(q); err != nil {
		t.Fatal(err)
	}
}

func allowInserts(t *testing.T, db, kind string) {
	t.Helper()
	d := testDB(t, db)
	defer d.Close()
	if _, err := d.Exec(`DROP TRIGGER fail_` + kind); err != nil {
		t.Fatal(err)
	}
}

// TestReadyFollowsTheWorkersLastWrite: /ready is 200 only once a worker row
// has committed since this start and the worker's latest write succeeded.
func TestReadyFollowsTheWorkersLastWrite(t *testing.T) {
	t.Parallel()
	f := newFixture(t)
	seed := f.start(f.env)
	seed.wait("/ready", http.StatusOK)
	seed.stop()
	killWorkers(f.db)

	// Worker rows from the earlier boot are in the database; none can
	// commit in this one.
	failInserts(t, f.db, "worker")
	p := f.start(f.env)
	body := p.wait("/ready", http.StatusInternalServerError)
	if !strings.Contains(body, "worker: no row committed since start") {
		t.Errorf("/ready before the first worker row: %q", body)
	}
	time.Sleep(1500 * time.Millisecond)
	if code, body, _ := p.do(http.MethodGet, "/ready", "", nil); code != http.StatusInternalServerError {
		t.Errorf("/ready with no worker row yet: %d %q, want 500", code, body)
	}
	r := p.report()
	if r.WorkerWritesFailed == 0 || r.WorkerWritesCommitted != 0 || !strings.Contains(r.WorkerLastError, "forced worker failure") {
		t.Errorf("report: committed %d, failed %d, last %q", r.WorkerWritesCommitted, r.WorkerWritesFailed, r.WorkerLastError)
	}

	// The first good write makes it ready.
	allowInserts(t, f.db, "worker")
	p.wait("/ready", http.StatusOK)

	// One failed write is enough to fail it.
	failInserts(t, f.db, "worker")
	deadline := time.Now().Add(5 * time.Second)
	for {
		r = p.report()
		if r.WorkerFailedInARow > 0 || time.Now().After(deadline) {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	code, body, _ := p.do(http.MethodGet, "/ready", "", nil)
	if r.WorkerFailedInARow != 1 || code != http.StatusInternalServerError || !strings.Contains(body, "worker: the last write failed") {
		t.Errorf("after one failed write (in a row %d): /ready %d %q, want 500", r.WorkerFailedInARow, code, body)
	}

	// The next good write makes it ready again; the counters stay.
	allowInserts(t, f.db, "worker")
	p.wait("/ready", http.StatusOK)
	if r := p.report(); r.WorkerFailedInARow != 0 || r.WorkerWritesFailed == 0 || r.WorkerWritesCommitted == 0 {
		t.Errorf("after recovery: committed %d, failed %d, in a row %d", r.WorkerWritesCommitted, r.WorkerWritesFailed, r.WorkerFailedInARow)
	}
	p.stop()
}

func TestAfterWriteExitsWhenItsRowFails(t *testing.T) {
	t.Parallel()
	f := newFixture(t)
	seed := f.start(f.env)
	seed.wait("/ready", http.StatusOK)
	seed.stop()
	killWorkers(f.db)

	failInserts(t, f.db, "bg")
	p := f.start(f.envWith(map[string]string{"CARLOS_REGION": "ie", "MOVEPROBE_FAIL_READY_IN": "ie,after-write"}))
	select {
	case <-p.done:
	case <-time.After(15 * time.Second):
		t.Fatalf("after-write kept running without its bg row; output:\n%s", p.out)
	}
	if code := p.cmd.ProcessState.ExitCode(); code == 0 {
		t.Errorf("after-write exited 0 without its bg row; output:\n%s", p.out)
	}
	if !strings.Contains(p.out.String(), "forced bg failure") {
		t.Errorf("the exit does not say why; output:\n%s", p.out)
	}
	if w := workerPIDs(f.db); len(w) != 0 {
		t.Errorf("after-write forked a worker before its row committed: %v", w)
	}
	if n := countRows(t, f.db, "bg", "ie"); n != 0 {
		t.Errorf("%d bg rows in ie", n)
	}
}

// ---- report query failures ------------------------------------------------

// failDriver is a database/sql driver whose query matching the DSN (a
// substring) yields one good row and then fails, as a read that breaks part
// way through would. Every other query yields no rows.
type failDriver struct{}

func (failDriver) Open(failOn string) (driver.Conn, error) { return failConn{failOn}, nil }

type failConn struct{ failOn string }

func (c failConn) Prepare(q string) (driver.Stmt, error) { return failStmt{c.failOn, q}, nil }
func (failConn) Close() error                            { return nil }
func (failConn) Begin() (driver.Tx, error)               { return nil, errors.New("no transactions") }

type failStmt struct{ failOn, q string }

func (failStmt) Close() error  { return nil }
func (failStmt) NumInput() int { return -1 }
func (failStmt) Exec([]driver.Value) (driver.Result, error) {
	return nil, errors.New("no writes")
}
func (s failStmt) Query([]driver.Value) (driver.Rows, error) {
	cols := []string{"kind", "region", "count", "min", "max"}
	row := []driver.Value{"bg", "uk", int64(3), int64(1), int64(3)}
	if strings.Contains(s.q, "IN ('tick', 'once')") {
		cols = []string{"kind", "name", "due_at", "count"}
		row = []driver.Value{"once", "once-1", int64(1_700_000_000), int64(1)}
	}
	return &failRows{cols: cols, row: row, fail: strings.Contains(s.q, s.failOn)}, nil
}

type failRows struct {
	cols []string
	row  []driver.Value
	fail bool
	n    int
}

func (r *failRows) Columns() []string { return r.cols }
func (r *failRows) Close() error      { return nil }
func (r *failRows) Next(dest []driver.Value) error {
	r.n++
	switch {
	case !r.fail:
		return io.EOF
	case r.n == 1:
		copy(dest, r.row)
		return nil
	default:
		return errors.New("forced read failure")
	}
}

func init() { sql.Register("moveprobe-fail", failDriver{}) }

func TestReportNeverAnswersPartial(t *testing.T) {
	t.Parallel()
	report := func(t *testing.T, db *sql.DB) (int, string) {
		t.Helper()
		a := &app{db: db, filesDir: t.TempDir(), region: "uk", mode: modeNormal, storeErr: errors.New("store: not under test")}
		rec := httptest.NewRecorder()
		a.handleReport(rec, httptest.NewRequest(http.MethodGet, "/report", nil))
		return rec.Code, rec.Body.String()
	}
	for _, failOn := range []string{"GROUP BY kind, region", "IN ('tick', 'once')"} {
		t.Run("rows fail part way: "+failOn, func(t *testing.T) {
			db, err := sql.Open("moveprobe-fail", failOn)
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			code, body := report(t, db)
			if code != http.StatusInternalServerError || !strings.Contains(body, "forced read failure") {
				t.Errorf("/report: %d %q, want 500 naming the read failure", code, body)
			}
		})
	}
	t.Run("a closed database", func(t *testing.T) {
		db := testDB(t, filepath.Join(t.TempDir(), "closed.db"))
		db.Close()
		if code, body := report(t, db); code != http.StatusInternalServerError {
			t.Errorf("/report on a closed database: %d %q, want 500", code, body)
		}
	})
	t.Run("nothing fails", func(t *testing.T) {
		db, err := sql.Open("moveprobe-fail", "matches no query")
		if err != nil {
			t.Fatal(err)
		}
		defer db.Close()
		if code, body := report(t, db); code != http.StatusOK {
			t.Errorf("/report: %d %q, want 200", code, body)
		}
	})
}

func TestArmOnceReusesOneControlConnection(t *testing.T) {
	t.Parallel()
	f := newFixture(t)
	p := f.start(f.env)
	p.wait("/ready", http.StatusOK)
	for i := 0; i < 3; i++ {
		if code, body, _ := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil); code != http.StatusOK {
			t.Fatalf("arm %d: %d %q", i+1, code, body)
		}
	}
	if n := f.controlConns.Load(); n != 1 {
		t.Errorf("three arms opened %d connections to the control socket, want 1", n)
	}
	p.stop()
}

// TestReadyFailsWhenTheWorkerStops: a worker that has died, or has sent no
// status for 3 s, fails readiness, and the report says so.
func TestReadyFailsWhenTheWorkerStops(t *testing.T) {
	t.Parallel()
	t.Run("the worker dies after its first good write", func(t *testing.T) {
		f := newFixture(t)
		p := f.start(f.env)
		p.wait("/ready", http.StatusOK)
		r := p.report()
		if r.WorkerWritesCommitted == 0 || r.WorkerExit != "" || r.WorkerLastStatusAt == 0 {
			t.Fatalf("before: committed %d, exit %q, last status %d", r.WorkerWritesCommitted, r.WorkerExit, r.WorkerLastStatusAt)
		}
		if err := syscall.Kill(r.WorkerPID, syscall.SIGKILL); err != nil {
			t.Fatal(err)
		}
		body := p.wait("/ready", http.StatusInternalServerError)
		if !strings.Contains(body, "worker: exited") {
			t.Errorf("/ready after the worker died: %q", body)
		}
		if r := p.report(); !strings.Contains(r.WorkerExit, "exited") {
			t.Errorf("report worker_exit %q after the worker died", r.WorkerExit)
		}
		p.stop()
	})

	t.Run("the worker sends no status for 3 s", func(t *testing.T) {
		f := newFixture(t)
		p := f.start(f.env)
		p.wait("/ready", http.StatusOK)
		pid := p.report().WorkerPID
		if err := syscall.Kill(pid, syscall.SIGSTOP); err != nil {
			t.Fatal(err)
		}
		stopped := time.Now()
		body := p.wait("/ready", http.StatusInternalServerError)
		if waited := time.Since(stopped); waited < 2*time.Second {
			t.Errorf("/ready failed %s after the worker stopped, before 3 s without status", waited)
		}
		if !strings.Contains(body, "worker: no status for") {
			t.Errorf("/ready with a stalled worker: %q", body)
		}
		syscall.Kill(pid, syscall.SIGCONT)
		p.wait("/ready", http.StatusOK)
		p.stop()
	})
}

// TestReadyCoalescesTheStoreCheck: anonymous /ready traffic costs at most one
// store check per 5 s, however many requests arrive, and requests that land
// while a check is running share its result.
func TestReadyCoalescesTheStoreCheck(t *testing.T) {
	t.Parallel()
	f := newFixture(t)
	p := f.start(f.env)
	p.wait("/ready", http.StatusOK)

	burst := func() {
		t.Helper()
		var wg sync.WaitGroup
		codes := make(chan int, 50)
		for i := 0; i < 50; i++ {
			wg.Add(1)
			go func() {
				defer wg.Done()
				code, _, err := p.do(http.MethodGet, "/ready", "", nil)
				if err != nil {
					code = 0
				}
				codes <- code
			}()
		}
		wg.Wait()
		close(codes)
		for c := range codes {
			if c != http.StatusOK {
				t.Errorf("a /ready in the burst answered %d", c)
			}
		}
	}

	// A slow store, so the burst lands while a check is in flight.
	f.storeDelay.Store(300)
	time.Sleep(5100 * time.Millisecond) // past any cached result
	n0 := f.storeReq.Load()
	burst()
	n1 := f.storeReq.Load()
	if n1-n0 > 1 {
		t.Errorf("50 concurrent /ready calls made %d store requests, want at most 1", n1-n0)
	}
	burst()
	if n2 := f.storeReq.Load(); n2 != n1 {
		t.Errorf("a second burst inside the 5 s window made %d store requests, want 0", n2-n1)
	}
	time.Sleep(5100 * time.Millisecond)
	n3 := f.storeReq.Load()
	burst()
	if n4 := f.storeReq.Load(); n4-n3 > 1 {
		t.Errorf("the next window's burst made %d store requests, want at most 1", n4-n3)
	}
	p.stop()
}