//go:build torture

package spindle

import (
	"context"
	"database/sql"
	"fmt"
	"io/fs"
	"math"
	"net"
	"net/url"
	"os"
	"path/filepath"
	"runtime"
	"runtime/debug"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/gorilla/websocket"

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/engine"
	"tangled.org/core/spindle/models"
	pipelinecodec "tangled.org/core/spindle/pipeline"
	"tangled.org/core/workflow"
)

var tqSeq atomic.Int64

func tqWorld(r *tortureRun, owners, perOwner, workers int) (*tortureWorld, []*tortureRepo) {
	w := newTortureWorld(r)
	w.maxJobs = workers
	k := w.addKnot()
	repos := w.addRepos(owners*perOwner, func(i int) (*tortureKnot, syntax.DID) {
		return k, ownerFor(i, perOwner)
	}, 0, nil)
	w.boot()
	w.seed(repos, 3*time.Minute)
	return w, repos
}

func (w *tortureWorld) tqPipeline(repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, workflow.PipelineSpec, *models.PipelineRecord, error) {
	n := tqSeq.Add(1)
	id := models.PipelineId(fmt.Sprintf("tq%09d", n))
	repoDid := repo.did.String()
	rkey := repo.name
	actor := repo.owner.String()
	tpl := workflow.PipelineSpec{
		TriggerMetadata: &workflow.TriggerMetadata{
			Kind:  "push",
			Actor: &actor,
			Push:  &workflow.PushTriggerData{NewSha: fmt.Sprintf("%040x", n), OldSha: zeroSha, Ref: "refs/heads/main"},
			Repo: &workflow.TriggerRepo{
				Did: repo.owner.String(), Knot: repo.knot.host(), Repo: &rkey, RepoDid: &repoDid, DefaultBranch: "main",
			},
		},
	}
	for i := range workflows {
		name := "ci.yaml"
		if workflows > 1 {
			name = fmt.Sprintf("w%d.yaml", i)
		}
		tpl.Workflows = append(tpl.Workflows, &workflow.WorkflowSpec{
			Name: name, Engine: "torture", Raw: tortureWorkflow(steps...), Clone: &workflow.CloneSpec{Skip: true},
		})
	}
	record, err := pipelinecodec.FromTangled(id, time.Now(), tpl)
	return id, tpl, record, err
}

func (w *tortureWorld) tqEnqueueTimed(lat *latencies, repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, error) {
	id, tpl, record, err := w.tqPipeline(repo, workflows, steps...)
	if err != nil {
		return "", err
	}
	start := time.Now()
	err = w.s.processPipeline(context.Background(), repo.did, tpl, record, nil, false)
	if lat != nil {
		lat.since(start)
	}
	return id, err
}

func (w *tortureWorld) tqEnqueue(repo *tortureRepo, workflows int, steps ...string) (models.PipelineId, error) {
	return w.tqEnqueueTimed(nil, repo, workflows, steps...)
}

func (w *tortureWorld) tqEnqueueMany(repos []*tortureRepo, n, par, workflows int, lat *latencies, steps ...string) []models.PipelineId {
	ids := make([]models.PipelineId, n)
	var next atomic.Int64
	var wg sync.WaitGroup
	for range par {
		wg.Go(func() {
			for {
				i := int(next.Add(1)) - 1
				if i >= n {
					return
				}
				id, err := w.tqEnqueueTimed(lat, repos[i%len(repos)], workflows, steps...)
				if err != nil {
					w.r.count("enqueue_error", 1)
					continue
				}
				ids[i] = id
			}
		})
	}
	wg.Wait()
	return ids
}

func (w *tortureWorld) tqFinished(within time.Duration) bool {
	return w.r.waitFor(within, func() bool { return w.unfinishedWorkflows() == 0 })
}

func (w *tortureWorld) tqWorkflowDone(id models.PipelineId, name string) bool {
	st, err := w.s.db.GetStatus(models.WorkflowId{PipelineId: id, Name: name})
	return err == nil && st.IsFinish()
}

func (w *tortureWorld) tqStartDelays(ids []models.PipelineId, name string, lat *latencies) {
	for _, id := range ids {
		if id == "" {
			continue
		}
		d, ok, err := w.s.db.WorkflowStartupDelay(context.Background(), models.WorkflowId{PipelineId: id, Name: name})
		if err == nil && ok {
			lat.add(d)
		}
	}
}

// must match logWriteTimeout in xrpc/ci_pipeline_subscribe_logs.go
const tqLogWriteTimeout = time.Minute

func tqPeaks(r *tortureRun, fn func(), gauges ...string) []float64 {
	peaks := make([]float64, len(gauges))
	for i, g := range gauges {
		peaks[i] = r.gaugeValue(g)
	}
	done := make(chan struct{})
	var wg sync.WaitGroup
	wg.Go(func() {
		tick := time.NewTicker(25 * time.Millisecond)
		defer tick.Stop()
		for {
			select {
			case <-done:
				return
			case <-tick.C:
				for i, g := range gauges {
					peaks[i] = max(peaks[i], r.gaugeValue(g))
				}
			}
		}
	})
	fn()
	close(done)
	wg.Wait()
	return peaks
}

func tqSettle(r *tortureRun, phase, gauge string, ceiling float64, within time.Duration) {
	ok := r.waitFor(within, func() bool { return r.gaugeValue(gauge) <= ceiling })
	r.check(phase+": "+gauge+" settles", ok, "%s is %.0f after %v, want <= %.0f", gauge, r.gaugeValue(gauge), within, ceiling)
	if ok {
		return
	}
	switch gauge {
	case "goroutines":
		r.dumpGoroutines()
	case "fds":
		tqFdSummary(r)
	}
}

func tqSettleBoth(r *tortureRun, phase string, baseGoroutines, baseFds, fdSlack float64, within time.Duration) {
	tqSettle(r, phase, "goroutines", baseGoroutines+30, within)
	tqSettle(r, phase, "fds", baseFds+fdSlack, within)
}

func tqFdSummary(r *tortureRun) {
	targets, err := fdTargets()
	if err != nil {
		return
	}
	kinds := map[string]int{}
	others := map[string]int{}
	for _, target := range targets {
		switch {
		case target == "":
			kinds["gone"]++
		case strings.HasPrefix(target, "socket:"):
			kinds["socket"]++
		case strings.HasPrefix(target, "pipe:"):
			kinds["pipe"]++
		case strings.HasPrefix(target, "anon_inode:"):
			kinds[target]++
		case strings.HasSuffix(target, ".log"):
			kinds["workflow log"]++
		default:
			kinds["other"]++
			others[filepath.Base(target)]++
		}
	}
	r.note("fds by kind: %v", kinds)
	r.note("fds of kind other by file name: %v", others)
}

func tqLargestFileMB(root string) float64 {
	var largest int64
	_ = filepath.WalkDir(root, func(_ string, d fs.DirEntry, err error) error {
		if err != nil || d.IsDir() {
			return nil
		}
		if info, err := d.Info(); err == nil {
			largest = max(largest, info.Size())
		}
		return nil
	})
	return float64(largest) / (1 << 20)
}

func tqOpenLogs(root string) float64 {
	targets, err := fdTargets()
	if err != nil {
		return math.NaN()
	}
	n := 0
	for _, target := range targets {
		if strings.HasPrefix(target, root) && strings.HasSuffix(target, ".log") {
			n++
		}
	}
	return float64(n)
}

func tqFileCount(root string) float64 {
	entries, err := os.ReadDir(root)
	if err != nil {
		return 0
	}
	return float64(len(entries))
}

func (w *tortureWorld) tqWSURL(nsid string, repo syntax.DID, pipeline string, workflows ...string) string {
	q := url.Values{}
	q.Set("repo", repo.String())
	q.Set("pipeline", pipeline)
	for _, name := range workflows {
		q.Add("workflows", name)
	}
	return "ws" + strings.TrimPrefix(w.api.URL, "http") + "/xrpc/" + nsid + "?" + q.Encode()
}

// small recv buffer so a stalled reader backs up the send queue quickly
func tqDialer(smallRecvBuf bool) *websocket.Dialer {
	d := &websocket.Dialer{HandshakeTimeout: 30 * time.Second}
	if smallRecvBuf {
		d.NetDial = func(network, addr string) (net.Conn, error) {
			c, err := net.Dial(network, addr)
			if err != nil {
				return nil, err
			}
			if tcp, ok := c.(*net.TCPConn); ok {
				_ = tcp.SetReadBuffer(4096)
			}
			return c, nil
		}
	}
	return d
}

func tqDrain(conn *websocket.Conn, bytesRead, msgs *atomic.Int64) {
	for {
		_, p, err := conn.ReadMessage()
		if err != nil {
			return
		}
		bytesRead.Add(int64(len(p)))
		msgs.Add(1)
	}
}

func init() {
	// the only worker gets stuck here and the backlog piles up behind it
	registerTorture("queue-backlog", "a big backlog nobody is draining yet, then a restart that works it off", func(r *tortureRun) {
		depth := r.load("jobs", 4000)
		probes := 50
		par := 4

		w, repos := tqWorld(r, 10, 3, 1)
		if _, err := w.tqEnqueue(repos[0], 1, "hang"); err != nil {
			r.t.Fatal(err)
		}
		if !r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() == 1 }) {
			r.t.Fatal("the hung pipeline never started")
		}
		baseMB := r.gaugeValue("db_mb")

		var enqueue latencies
		var firstMedian, lastMedian, firstEnqueue time.Duration
		var writes latencies
		var quiet latencies
		for range 50 {
			start := time.Now()
			if err := w.s.db.StatusPending(models.WorkflowId{PipelineId: "tq-quiet", Name: "ci.yaml"}, nil); err != nil {
				r.count("status_write", 1)
			}
			quiet.since(start)
		}
		quietWrite := quiet.quantile(0.99)
		levels := []int{depth / 8, depth / 4, depth / 2, depth}
		for _, level := range levels {
			have := int(w.rows("jobs"))
			w.tqEnqueueMany(repos, max(level-have, 0), par, 1, &enqueue, "sleep 1ms")
			if level == levels[0] {
				firstEnqueue = enqueue.quantile(0.99)
			}

			// probes only pull 50 jobs a level, not enough to shift the depth
			var lat latencies
			stopWrites := make(chan struct{})
			var wg sync.WaitGroup
			wg.Go(func() {
				wid := models.WorkflowId{PipelineId: "tq-probe", Name: "ci.yaml"}
				for {
					select {
					case <-stopWrites:
						return
					case <-time.After(2 * time.Millisecond):
					}
					start := time.Now()
					if err := w.s.db.StatusPending(wid, nil); err != nil {
						r.count("status_write", 1)
					}
					writes.since(start)
				}
			})
			for range probes {
				start := time.Now()
				if _, err := w.s.db.DequeueJob(context.Background(), nil); err != nil {
					r.count("dequeue_probe", 1)
				}
				lat.since(start)
			}
			close(stopWrites)
			wg.Wait()

			if level == levels[0] {
				firstMedian = lat.quantile(0.5)
			}
			lastMedian = lat.quantile(0.5)
		}
		r.latency("enqueue", &enqueue)
		r.latency("status_write_during_dequeue", &writes)
		have := int(w.rows("jobs"))
		r.metric("db_kb_per_queued_job", (r.gaugeValue("db_mb")-baseMB)*1024/float64(max(have, 1)))

		r.metric("dequeue_ms_per_1k_queued", float64(lastMedian.Microseconds())/1000/float64(max(have, 1))*1000)
		r.check("D1 dequeue cost is flat in queue depth", lastMedian <= 3*firstMedian+500*time.Microsecond,
			"median dequeue %v at depth %d vs %v at depth %d", lastMedian, levels[len(levels)-1], firstMedian, levels[0])
		r.check("D1 a dequeue doesn't stall other writers", writes.quantile(0.99) <= 10*quietWrite+50*time.Millisecond,
			"status write p99 %v while dequeues ran, %v when quiet, want within 10x plus 50ms", writes.quantile(0.99), quietWrite)
		r.check("enqueue stays fast under the backlog", enqueue.quantile(0.99) <= 10*firstEnqueue+50*time.Millisecond,
			"enqueue p99 %v over the whole run, %v at the first depth, want within 10x plus 50ms", enqueue.quantile(0.99), firstEnqueue)

		w.stop()
		w.maxJobs = int(r.param("drain_workers", 8))
		w.boot()

		total := int(w.rows("jobs"))
		marks := []float64{0.75, 0.5, 0.25, 0}
		at := make([]time.Duration, len(marks))
		drainStart := time.Now()
		drained := true
		for i, m := range marks {
			target := int(float64(total) * m)
			if !r.waitFor(4*time.Minute, func() bool { return int(w.rows("jobs")) <= target }) {
				drained = false
				break
			}
			at[i] = time.Since(drainStart)
		}
		r.check("queue drains after the restart", drained, "%v jobs still queued", w.rows("jobs"))
		if drained && at[0] > 0 && at[3] > at[2] {
			first := float64(total) / 4 / at[0].Seconds()
			last := float64(total) / 4 / (at[3] - at[2]).Seconds()
			r.metric("drain_first_quarter_per_s", first)
			r.metric("drain_last_quarter_per_s", last)
			r.check("D1 drain rate doesn't depend on how deep the queue is", first >= 0.7*last,
				"%.0f jobs/s over the first quarter vs %.0f jobs/s over the last", first, last)
		}
		r.note("the repo schedule refresh warnings in this run are the records-only repos having no git, harness noise")
		r.note("the fake engine does no work, drain rates are about dequeue and bookkeeping only. probe-dequeued jobs and the hung pipeline are left unfinished on purpose")
	})

	registerTorture("queue-fairness", "one owner floods the queue while a few small owners push a little", func(r *tortureRun) {
		flood := r.load("flood", 1500)
		smallOwners := 5
		smallJobs := 3
		workers := int(r.param("workers", 4))
		floodStep := r.duration("flood_step", 100*time.Millisecond)

		w, repos := tqWorld(r, 1+smallOwners, 1, workers)
		floodRepo, smallRepos := repos[0], repos[1:]
		floodOwner := floodRepo.owner.String()

		w.tqEnqueueMany(repos[:1], flood, 4, 1, nil, "sleep "+floodStep.String())
		if !r.waitFor(time.Minute, func() bool { return w.eng.started.Load() >= int64(workers*3) }) {
			r.t.Fatal("the flood never started running")
		}
		ahead := len(w.eng.startOrder())
		r.metric("queued_when_smalls_arrive", w.rows("jobs"))

		var smallIDs []models.PipelineId
		for range smallJobs {
			smallIDs = append(smallIDs, w.tqEnqueueMany(smallRepos, smallOwners, 1, 1, nil, "sleep "+floodStep.String())...)
		}
		done := r.waitFor(5*time.Minute, func() bool {
			for _, id := range smallIDs {
				if !w.tqWorkflowDone(id, "ci.yaml") {
					return false
				}
			}
			return true
		})
		var delay latencies
		w.tqStartDelays(smallIDs, "ci.yaml", &delay)
		r.latency("small_owner_start_delay", &delay)

		order := w.eng.startOrder()
		firstSmall := -1
		for i := ahead; i < len(order); i++ {
			if order[i] != floodOwner {
				firstSmall = i
				break
			}
		}
		r.metric("flood_starts_ahead_of_first_small", float64(firstSmall-ahead))
		r.metric("flood_starts_total", float64(flood))
		r.check("small owners finish while the flood is still queued", done && w.rows("jobs") > 0,
			"done=%v queued=%v", done, w.rows("jobs"))
		r.check("Q3 a small owner is only ever behind about one round of flood jobs", firstSmall >= 0 && firstSmall-ahead <= 2*workers,
			"%d flood jobs started after the small ones were queued and before the first small one", firstSmall-ahead)
		r.check("Q3 small owners start within a few job lengths", delay.quantile(0.99) <= 20*floodStep,
			"p99 wait %v with %v jobs and %d workers", delay.quantile(0.99), floodStep, workers)
		r.check("the flood drains", w.tqFinished(10*time.Minute), "%v workflows unfinished", w.unfinishedWorkflows())
	})

	registerTorture("queue-restart-mid-run", "restarts with work in flight, and a worker that wakes into a database error", func(r *tortureRun) {
		hung := r.load("hung", 6)
		queued := r.load("queued", 30)
		recoverWait := 20 * time.Second
		rounds := r.load("wake_rounds", 20)
		lockHold := 7 * time.Second

		w, repos := tqWorld(r, 3, 1, hung+2)
		startHung := func(tag string) []models.PipelineId {
			ids := w.tqEnqueueMany(repos, hung, 2, 1, nil, "hang")
			if !r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() >= int64(hung) }) {
				r.t.Fatalf("%s: only %d of %d hung workflows started", tag, w.eng.running.Load(), hung)
			}
			return ids
		}

		startHung("clean")
		w.tqEnqueueMany(repos, queued, 2, 1, nil, "sleep 50ms")
		w.stop()
		r.check("a clean stop destroys every running workflow", w.eng.running.Load() == 0, "%d still running", w.eng.running.Load())
		w.boot()
		finished := w.tqFinished(2 * time.Minute)
		r.check("queued jobs survive a clean restart and finish", finished, "%v workflows pending or running", w.unfinishedWorkflows())

		// erase them here, a killed process never writes its terminal row
		crashed := startHung("crash")
		w.stop()
		w.tqEraseTerminals(crashed)
		w.boot()
		recovered := r.waitFor(recoverWait, func() bool { return w.unfinishedWorkflows() == 0 })
		r.metric("unfinished_after_crash", w.unfinishedWorkflows())
		notFailed := 0
		for _, id := range crashed {
			if id == "" {
				continue
			}
			if status, err := w.s.db.GetStatus(models.WorkflowId{PipelineId: id, Name: "ci.yaml"}); err != nil || status != models.StatusKindFailed {
				notFailed++
			}
		}
		r.check("Q1 a crash restart marks what it was running as failed", notFailed == 0, "%d of %d crashed workflows are not failed, want 0", notFailed, len(crashed))
		r.check("Q1 nothing stays pending or running forever after a crash restart", recovered,
			"%v workflows pending or running %v after the restart, want 0", w.unfinishedWorkflows(), recoverWait)

		if !w.tqFinished(time.Minute) {
			r.note("workflows from the crash step are still unfinished, wake rounds run on top of them")
		}
		stuck := 0
		for range rounds {
			var ids [2]models.PipelineId
			var wg sync.WaitGroup
			for i := range ids {
				wg.Go(func() {
					id, err := w.tqEnqueue(repos[i%len(repos)], 1, "hang")
					if err != nil {
						r.count("enqueue_error", 1)
					}
					ids[i] = id
				})
			}
			wg.Wait()
			if !r.waitFor(10*time.Second, func() bool { return w.eng.running.Load() >= 2 }) {
				stuck++
			}
			for _, id := range ids {
				engine.CancelWorkflow(models.WorkflowId{PipelineId: id, Name: "ci.yaml"})
			}
			r.waitFor(30*time.Second, func() bool { return w.eng.running.Load() == 0 })
		}
		r.metric("rounds_with_a_second_job_not_started", float64(stuck))
		r.check("W2 a second enqueue wakes a second idle worker", stuck == 0,
			"%d of %d rounds had a job that never started with idle workers around", stuck, rounds)
		r.note("the wake race is narrow, a clean result here doesn't prove the channel is safe")

		startedBefore := w.eng.started.Load()
		id, tpl, record, err := w.tqPipeline(repos[0], 1, "sleep 10ms")
		if err != nil {
			r.t.Fatal(err)
		}
		if err := w.s.db.EnqueuePipeline(context.Background(), repos[0].did.String(), record, nil, tpl, false, "", "", nil); err != nil {
			r.t.Fatal(err)
		}
		conn, err := w.s.db.Conn(context.Background())
		if err != nil {
			r.t.Fatal(err)
		}
		if _, err := conn.ExecContext(context.Background(), "begin immediate"); err != nil {
			r.t.Fatal(err)
		}
		select {
		case w.s.jobWake <- struct{}{}:
		default:
		}
		time.Sleep(lockHold)
		_, _ = conn.ExecContext(context.Background(), "rollback")
		conn.Close()
		resumed := r.waitFor(15*time.Second, func() bool { return w.eng.started.Load() > startedBefore })
		r.check("W2 the queue resumes after a dequeue error with no new enqueue", resumed,
			"job %s started=%v within %v of the lock being released, want started", id, resumed, 15*time.Second)
		select {
		case w.s.jobWake <- struct{}{}:
		default:
		}
	})

	registerTorture("logs-flood", "log output without newlines, a lot of output, and many workflows at once", func(r *tortureRun) {
		rawMB := 64
		dribbleKB := 2048
		logMB := 24
		logWorkflows := 4
		many := r.load("many_workflows", 150)
		slots := int(r.param("slots", 4))
		capMB := r.param("log_cap_mb", 33)
		heapBudget := r.param("heap_budget_mb", 48)

		w, repos := tqWorld(r, 1, 1, 4)
		eng := w.eng
		repo := repos[0]
		logDir := filepath.Join(w.dir, "logs")
		baseGoroutines := r.gaugeValue("goroutines")
		baseFds := r.gaugeValue("fds")

		// no newline means the whole thing waits in the writer's pending buffer
		rawCmd := fmt.Sprintf("raw %d", rawMB<<20)
		// tight gc so heap numbers are live memory, not garbage
		defer debug.SetGCPercent(debug.SetGCPercent(10))
		runtime.GC()
		baseHeap := r.gaugeValue("heap_mb")
		peaks := tqPeaks(r, func() {
			if _, err := w.tqEnqueue(repo, 1, rawCmd); err != nil {
				r.t.Fatal(err)
			}
			if !w.tqFinished(5 * time.Minute) {
				r.check("raw output workflow finishes", false, "still unfinished after 5m")
			}
		}, "heap_mb", "rss_mb")
		debug.SetGCPercent(100)
		r.metric("raw_heap_growth_mb", peaks[0]-baseHeap)
		r.metric("raw_rss_peak_mb", peaks[1])
		r.check("L1 output without newlines doesn't pile up in memory", peaks[0]-baseHeap <= heapBudget,
			"heap grew %.0fMB while a step wrote %dMB with no newline, budget %.0fMB", peaks[0]-baseHeap, rawMB, heapBudget)

		// doubling the bytes should only double the time, small writes rescan the pending line
		dribble := func(kb int) time.Duration {
			cmd := fmt.Sprintf("dribble %d", kb<<10)
			if _, err := w.tqEnqueue(repo, 1, cmd); err != nil {
				r.t.Fatal(err)
			}
			w.tqFinished(5 * time.Minute)
			return eng.tookFor(cmd)
		}
		once, twice := dribble(dribbleKB), dribble(2*dribbleKB)
		r.metric("dribble_step_s", once.Seconds())
		r.check("L1 a long line written in small chunks costs linear time", twice <= 3*once+50*time.Millisecond,
			"%dKB in 16 byte writes took %v, %dKB took %v, want the double within 3x plus 50ms", dribbleKB, once, 2*dribbleKB, twice)

		logCmd := fmt.Sprintf("log %d", logMB<<20)
		// sample the size while it still grows, logs get deleted once archived
		r.gauge("largest_log_mb", func() float64 { return tqLargestFileMB(logDir) })
		largest := tqPeaks(r, func() {
			if _, err := w.tqEnqueue(repo, logWorkflows, logCmd); err != nil {
				r.t.Fatal(err)
			}
			w.tqFinished(5 * time.Minute)
		}, "largest_log_mb")[0]
		r.metric("largest_log_file_mb", largest)
		r.check("L1 a workflow's log file is capped", largest > 1 && largest <= capMB,
			"largest log file seen while running is %.1fMB, want over 1MB so the flood was seen and at most %.0fMB", largest, capMB)
		r.note("log_cap_mb is the default LogMaxBytes (32MB) plus a line and a marker, dawn picks the real number")

		w.eng.slots = engine.NewSemaphoreSlotter(slots)
		r.gauge("open_workflow_logs", func() float64 { return tqOpenLogs(logDir) })
		peaks = tqPeaks(r, func() {
			if _, err := w.tqEnqueue(repo, many, "lines 100; sleep 200ms"); err != nil {
				r.t.Fatal(err)
			}
			w.tqFinished(5 * time.Minute)
		}, "fds", "goroutines", "open_workflow_logs")
		logPeak := peaks[2]
		r.metric("many_open_logs_peak", logPeak)
		r.check("Q2 log files are only open for workflows that hold a slot", logPeak <= float64(slots)+16,
			"%d workflows on %d slots peaked %.0f open workflow log fds, want <= %d (a few more are archives in flight)", many, slots, logPeak, slots+16)

		r.note("many_workflows has no cap in spindle (Q2), this only measures what a pipeline that wide costs. the log engine writes real files, the numbers are about spindle's writer and bookkeeping, nothing an engine does")

		tqSettleBoth(r, "after the floods", baseGoroutines, baseFds, 64, 30*time.Second)
	})

	registerTorture("logs-subscribers", "many log subscribers, stalled readers and vanishing clients", func(r *tortureRun) {
		subs := r.load("subscribers", 300)
		slow := r.load("slow_readers", 60)
		churn := r.load("churn", 400)
		slowWait := tqLogWriteTimeout + 10*time.Second
		backlogMB := 32

		w, repos := tqWorld(r, 1, 1, 4)
		repo := repos[0]
		logDir := filepath.Join(w.dir, "logs")
		dbStat := func(read func(sql.DBStats) int) func() float64 {
			return func() float64 {
				return w.fromSpindle(func(s *Spindle) float64 { return float64(read(s.db.Stats())) })
			}
		}
		r.gauge("db_open_conns", dbStat(func(st sql.DBStats) int { return st.OpenConnections }))
		r.gauge("db_conns_in_use", dbStat(func(st sql.DBStats) int { return st.InUse }))
		logSize := func(id models.PipelineId) int64 {
			fi, err := os.Stat(models.LogFilePath(logDir, models.WorkflowId{PipelineId: id, Name: "ci.yaml"}))
			if err != nil {
				return 0
			}
			return fi.Size()
		}

		small, err := w.tqEnqueue(repo, 1, "lines 200", "hang")
		if err != nil {
			r.t.Fatal(err)
		}
		big, err := w.tqEnqueue(repo, 1, fmt.Sprintf("log %d", backlogMB<<20), "hang")
		if err != nil {
			r.t.Fatal(err)
		}
		if !r.waitFor(time.Minute, func() bool { return w.eng.running.Load() >= 2 && logSize(big) >= int64(backlogMB)<<20 }) {
			r.t.Fatal("the running pipelines never got their logs written")
		}
		smallURL := w.tqWSURL(org_tangled.CiSubscribePipelineLogsNSID, repo.did, string(small), "ci.yaml")
		bigURL := w.tqWSURL(org_tangled.CiSubscribePipelineLogsNSID, repo.did, string(big), "ci.yaml")
		// take the baseline after the first viewer warms lazy state
		if conn, _, err := tqDialer(false).Dial(smallURL, nil); err == nil {
			_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
			_, _, _ = conn.ReadMessage()
			conn.Close()
		}
		time.Sleep(2 * time.Second)
		baseGoroutines := r.gaugeValue("goroutines")
		baseFds := r.gaugeValue("fds")

		var first latencies
		var bytesRead, msgs atomic.Int64
		conns := make([]*websocket.Conn, subs)
		sem := make(chan struct{}, 50)
		var wg sync.WaitGroup
		for i := range subs {
			wg.Go(func() {
				sem <- struct{}{}
				start := time.Now()
				conn, _, err := tqDialer(false).Dial(smallURL, nil)
				<-sem
				if err != nil {
					r.count("dial", 1)
					return
				}
				_ = conn.SetReadDeadline(time.Now().Add(30 * time.Second))
				_, p, err := conn.ReadMessage()
				if err != nil {
					r.count("first_message", 1)
					conn.Close()
					return
				}
				first.since(start)
				bytesRead.Add(int64(len(p)))
				_ = conn.SetReadDeadline(time.Time{})
				conns[i] = conn
				go tqDrain(conn, &bytesRead, &msgs)
			})
		}
		wg.Wait()
		r.latency("herd_first_message", &first)
		r.metric("goroutines_per_subscriber", (r.gaugeValue("goroutines")-baseGoroutines)/float64(subs))
		r.metric("fds_per_subscriber", (r.gaugeValue("fds")-baseFds)/float64(subs))
		time.Sleep(3 * time.Second)
		for _, c := range conns {
			if c != nil {
				c.Close()
			}
		}
		tqSettleBoth(r, "herd leaves", baseGoroutines, baseFds, 30, 30*time.Second)

		// stalled readers, with more backlog than the socket buffers hold
		baseGoroutines, baseFds = r.gaugeValue("goroutines"), r.gaugeValue("fds")
		var slowConns []*websocket.Conn
		for range slow {
			conn, _, err := tqDialer(true).Dial(bigURL, nil)
			if err != nil {
				r.count("slow_dial", 1)
				continue
			}
			slowConns = append(slowConns, conn)
		}
		time.Sleep(5 * time.Second)
		r.metric("slow_goroutines_per_reader_at_5s", (r.gaugeValue("goroutines")-baseGoroutines)/float64(max(len(slowConns), 1)))
		time.Sleep(max(slowWait-5*time.Second, 0))
		r.metric("slow_fds_over_base", r.gaugeValue("fds")-baseFds)
		for _, c := range slowConns {
			c.UnderlyingConn().Close()
		}
		tqSettleBoth(r, "slow readers leave", baseGoroutines, baseFds, 30, 15*time.Second)

		// they leave with backlog still unsent on the server
		for _, abrupt := range []bool{false, true} {
			baseGoroutines, baseFds = r.gaugeValue("goroutines"), r.gaugeValue("fds")
			sem = make(chan struct{}, 25)
			for range churn {
				wg.Go(func() {
					sem <- struct{}{}
					defer func() { <-sem }()
					conn, _, err := tqDialer(false).Dial(smallURL, nil)
					if err != nil {
						r.count("churn_dial", 1)
						return
					}
					_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
					_, _, _ = conn.ReadMessage()
					if abrupt {
						conn.UnderlyingConn().Close()
					} else {
						conn.Close()
					}
				})
			}
			wg.Wait()
			phase := "churn with a close frame"
			if abrupt {
				phase = "churn without a close frame"
			}
			tqSettleBoth(r, phase, baseGoroutines, baseFds, 30, 15*time.Second)
		}
	})

	registerTorture("db-growth", "pipelines and statuses at a steady rate, what grows and what slows", func(r *tortureRun) {
		soak := 3 * time.Minute
		perSec := float64(30)
		workflows := int(r.param("workflows", 3))
		owners := r.load("owners", 10)

		w, repos := tqWorld(r, owners, 2, 8)
		logDir := filepath.Join(w.dir, "logs")
		r.gauge("log_files", func() float64 { return tqFileCount(logDir) })
		baseMB := r.gaugeValue("db_mb")

		var recentEarly, recentLate, rareEarly, rareLate latencies
		probe := func(repo *tortureRepo, elapsed time.Duration) {
			query := func(kinds []string, limit int, early, late *latencies) {
				start := time.Now()
				if _, _, _, err := w.s.db.QueryPipelines(context.Background(), repo.did.String(), nil, "", kinds, limit); err != nil {
					r.count("query", 1)
				}
				switch {
				case elapsed < soak/3:
					early.since(start)
				case elapsed > 2*soak/3:
					late.since(start)
				}
			}
			query(nil, 50, &recentEarly, &recentLate)
			query([]string{"manual"}, 30, &rareEarly, &rareLate)
		}

		var enqueue latencies
		interval := time.Duration(float64(time.Second) / perSec)
		begin := time.Now()
		next := begin
		lastProbe := begin
		sent := 0
		for time.Since(begin) < soak {
			// half go to repos[0] so its history is long enough for the query probes
			repo := repos[0]
			if sent%2 == 1 {
				repo = repos[sent%len(repos)]
			}
			if _, err := w.tqEnqueueTimed(&enqueue, repo, workflows, "lines 20"); err != nil {
				r.count("enqueue_error", 1)
			}
			sent++
			if time.Since(lastProbe) >= 5*time.Second {
				probe(repos[0], time.Since(begin))
				lastProbe = time.Now()
			}
			next = next.Add(interval)
			time.Sleep(time.Until(next))
		}
		r.latency("enqueue", &enqueue)
		r.metric("queued_at_end_of_load", w.rows("jobs"))
		r.check("the queue keeps up with the load", w.rows("jobs") <= 50,
			"%v jobs still queued at the end of the load", w.rows("jobs"))
		w.tqFinished(2 * time.Minute)

		pipelines := w.rows("pipelines")
		statuses := w.rows("workflow_statuses")
		growthMB := r.gaugeValue("db_mb") - baseMB
		r.metric("pipelines", pipelines)
		r.metric("status_rows", statuses)
		r.metric("db_kb_per_pipeline", growthMB*1024/max(pipelines, 1))
		r.metric("db_gb_per_month_at_this_rate", growthMB*1024/max(pipelines, 1)*perSec*86400*30/1e6)
		r.metric("log_dir_files", tqFileCount(logDir))
		r.latency("query_recent_early", &recentEarly)
		r.latency("query_recent_late", &recentLate)
		r.latency("query_rare_kind_early", &rareEarly)
		r.latency("query_rare_kind_late", &rareLate)

		r.check("QP1 listing a repo's recent pipelines doesn't slow down as history grows",
			recentLate.len() == 0 || recentLate.quantile(0.5) <= 2*recentEarly.quantile(0.5)+300*time.Microsecond,
			"median %v late vs %v early", recentLate.quantile(0.5), recentEarly.quantile(0.5))
		r.check("QP1 a rare kind filter doesn't walk the whole history",
			rareLate.len() == 0 || rareLate.quantile(0.5) <= 2*rareEarly.quantile(0.5)+300*time.Microsecond,
			"median %v late vs %v early with no matching rows", rareLate.quantile(0.5), rareEarly.quantile(0.5))
		archived := w.rows("mill_artifacts")
		r.metric("archived_logs", archived)
		if archived > 0 {
			r.check("G1 a log that is archived leaves local disk", tqFileCount(logDir) <= 10,
				"%.0f log files in the log dir with %.0f archived, want <= 10", tqFileCount(logDir), archived)
		} else {
			r.note("no log was archived in this setup, so log disk growth shows in log_dir_files but isn't checked")
		}
		r.note("one workflow is %d statuses at least (pending, running, success). the fake engine writes %s of log per workflow, the db numbers are bookkeeping only", 3, "20 lines")
	})
}

func (w *tortureWorld) tqEraseTerminals(ids []models.PipelineId) {
	d, err := db.Make(context.Background(), w.dbPath())
	if err != nil {
		w.r.t.Fatal(err)
	}
	defer d.Close()
	for _, id := range ids {
		if id == "" {
			continue
		}
		if _, err := d.Exec(`delete from workflow_statuses where pipeline_id = ? and status not in ('pending', 'running')`, string(id)); err != nil {
			w.r.t.Fatal(err)
		}
	}
}