//go:build torture
package spindle
import (
"context"
"fmt"
"slices"
"sync"
"sync/atomic"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
)
// knots includes the caller's own knot
func poolFairShare(pool, knots int) int {
return min(max(2, pool/knots), max(1, pool-poolReserve(pool)))
}
func init() {
// one size per run to fit under MemoryMax, e.g. -torture.set pool=64,notify=16
registerTorture("pool-sizing", "what a repo job pool size and a notify slot count cost idle and under a catch-up with pushes", func(r *tortureRun) {
n := r.load("repos", 5000)
repoPool := int(r.param("pool", 256))
notifySlots := int(r.param("notify", 64))
smallKnots := 4
pushRepos := 40
slowDelay := 2 * time.Second
budget := r.duration("drain_budget", 30*time.Minute)
w := newTortureWorld(r)
w.repoPool, w.notifySlots = repoPool, notifySlots
w.maxJobs = 8
big := w.addKnot()
small := w.addKnots(smallKnots)
slow := w.addKnot()
pick := func(i int) (*tortureKnot, syntax.DID) {
owner := syntax.DID(fmt.Sprintf("did:plc:poolowner%05d", i))
switch rem := i % 10; {
case rem < 6:
return big, owner
case rem < 9:
return small[i%smallKnots], owner
}
return slow, owner
}
repos := w.addRepos(n, pick, 0, nil)
pushes := w.addRepos(pushRepos, func(i int) (*tortureKnot, syntax.DID) {
return small[i%smallKnots], syntax.DID(fmt.Sprintf("did:plc:poolpush%03d", i))
}, 2, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")})
r.metric("repo_pool", float64(repoPool))
r.metric("notify_slots", float64(notifySlots))
r.gauge("knots_inflight", func() float64 {
var total int64
for _, k := range w.knots {
total += k.inflight.Load()
}
return float64(total)
})
r.gauge("slow_knot_inflight", func() float64 { return float64(slow.inflight.Load()) })
r.gauge("notify_inflight", w.abuseNotifyInflight)
r.gauge("notify_held_max_per_knot", func() float64 {
most := 0.0
for _, k := range w.knots {
most = max(most, w.abuseHeld(k.host()))
}
return most
})
w.boot()
time.Sleep(5 * time.Second)
r.metric("idle_rss_mb", r.gaugeValue("rss_mb"))
r.metric("idle_goroutines", r.gaugeValue("goroutines"))
r.metric("idle_fds", r.gaugeValue("fds"))
all := append(slices.Clone(repos), pushes...)
w.seed(all, 10*time.Minute)
r.metric("served_rss_mb", r.gaugeValue("rss_mb"))
w.consumed(w.js.filler(), 30*time.Second)
w.stop()
var changed []syntax.DID
var changedOn []*tortureRepo
for i := 0; i < len(repos); i += 5 {
who := syntax.DID(fmt.Sprintf("did:plc:poolnew%05d", i))
ingPutCollaborator(repos[i], who)
changed = append(changed, who)
changedOn = append(changedOn, repos[i])
}
for _, k := range w.knots {
k.delay.Store((20 * time.Millisecond).Nanoseconds())
}
slow.delay.Store(slowDelay.Nanoseconds())
w.js.forget()
w.js.filler()
readsBefore := ingManifestReads(w.knots)
w.boot()
start := time.Now()
ctx, cancel := context.WithTimeout(context.Background(), budget)
defer cancel()
var wg sync.WaitGroup
var st deliveryStats
ds := pushRanges(pushes, 0, 2)
wg.Go(func() { w.flood(ctx, ds, min(2*notifySlots, 512), 40, &st) })
var applied grantProbes
stopLive := make(chan struct{})
var liveSent atomic.Int64
wg.Go(func() {
for i := 0; ; i++ {
select {
case <-stopLive:
return
case <-time.After(250 * time.Millisecond):
}
w.probeGrant(repos[0], syntax.DID(fmt.Sprintf("did:plc:poollive%05d", i)), 2*time.Minute, &applied)
liveSent.Add(1)
}
})
// running jobs can stay over the share, only starting one over it is a violation
slowHost := slow.host()
var jobsPeak, violations atomic.Int64
stopSample := make(chan struct{})
sampling := make(chan struct{})
go func() {
defer close(sampling)
prevRunning, prevOthers := 0, 0
for {
select {
case <-stopSample:
return
case <-time.After(2 * time.Millisecond):
}
jobs := &w.spindle().retries.jobs
jobs.mu.Lock()
running := jobs.running[slowHost]
others := 0
for knot := range jobs.queued {
if knot != slowHost {
others++
}
}
jobs.mu.Unlock()
if others > 0 {
if int64(running) > jobsPeak.Load() {
jobsPeak.Store(int64(running))
}
if prevOthers > 0 && running > prevRunning && running > poolFairShare(repoPool, others+1) {
violations.Add(1)
}
}
prevRunning, prevOthers = running, others
}
}()
w.probeTable()
var probe latencies
var probeErrors atomic.Int64
stopProbe := w.probeWrites(50*time.Millisecond, &probe, &probeErrors)
caughtUp := r.waitFor(budget, func() bool {
pending, err := w.spindle().db.PendingCatchUp()
return err == nil && len(pending) == 0
})
catchUp := time.Since(start)
applied2 := r.waitFor(max(time.Until(start.Add(budget)), time.Second), func() bool {
for i, who := range changed {
if !w.isCollab(changedOn[i], who) {
return false
}
}
return true
})
elapsed := time.Since(start)
close(stopLive)
stopProbe()
close(stopSample)
cancel()
wg.Wait()
<-sampling
applied.wg.Wait()
// over budget there's no real time to report, only finished ones do
if caughtUp {
r.metric("catchup_s", catchUp.Seconds())
}
if applied2 {
r.metric("apply_all_s", elapsed.Seconds())
}
r.metric("run_s", elapsed.Seconds())
r.metric("refreshes_per_s", float64(ingManifestReads(w.knots)-readsBefore)/elapsed.Seconds())
r.metric("live_events_per_s", float64(liveSent.Load())/elapsed.Seconds())
r.metric("notify_accepted_per_s", float64(st.accepted.len())/elapsed.Seconds())
r.metric("probe_write_errors", float64(probeErrors.Load()))
r.metric("slow_knot_jobs_peak_while_others_wait", float64(jobsPeak.Load()))
r.metric("fair_share_violations", float64(violations.Load()))
r.latency("probe_write", &probe)
r.latency("live_apply", &applied.latencies)
r.metric("live_missed", float64(applied.missed.Load()))
st.report(r, "notify_")
r.note("rss, goroutines, fds and the knots' open requests are sampled all run, read them as _max in the metrics. busy and locked errors from spindle's own writes show up in errors by message, the probe write errors are only the probe's")
r.check("catch-up finished within the budget", caughtUp, "the pending catch-up was not empty after %v", budget)
r.check("every changed repo applied within the budget", applied2, "%d changed repos, not all applied %v after the boot", len(changed), budget)
r.check("a slow knot never starts another repo job past its share while others wait", violations.Load() == 0,
"the slow knot started %d jobs while over its share of %d workers with other knots waiting, its most running at once then was %d", violations.Load(), repoPool, jobsPeak.Load())
})
}
func poolReserve(pool int) int { return max(2, pool/8) }
func init() {
registerTorture("pool-priority", "a priority knot's repo jobs run at once while a hostile knot pins the shared repo pool", func(r *tortureRun) {
hostileRepos := r.load("hostile_repos", 300)
pool := int(r.param("pool", 16))
hold := 4 * time.Second
budget := r.duration("priority_budget", 3*time.Second)
w := newTortureWorld(r)
w.repoPool = pool
prio, plain, hostile := w.addKnot(), w.addKnot(), w.addKnot()
w.priorityKnots = []string{prio.host()}
prioRepos := w.addRepos(10, func(i int) (*tortureKnot, syntax.DID) {
return prio, syntax.DID(fmt.Sprintf("did:plc:prioowner%03d", i))
}, 0, nil)
plainRepos := w.addRepos(10, func(i int) (*tortureKnot, syntax.DID) {
return plain, syntax.DID(fmt.Sprintf("did:plc:plainowner%03d", i))
}, 0, nil)
bad := w.addRepos(hostileRepos, func(i int) (*tortureKnot, syntax.DID) {
return hostile, syntax.DID(fmt.Sprintf("did:plc:hostileowner%04d", i))
}, 0, nil)
all := append(append(append([]*tortureRepo{}, prioRepos...), plainRepos...), bad...)
w.boot()
w.seed(all, 5*time.Minute)
w.consumed(w.js.filler(), 30*time.Second)
w.stop()
for i, repo := range bad {
ingPutCollaborator(repo, syntax.DID(fmt.Sprintf("did:plc:hostilenew%04d", i)))
}
hostile.delay.Store(hold.Nanoseconds())
w.js.forget()
w.js.filler()
w.boot()
hostileHost := hostile.host()
running := func() int {
jobs := &w.spindle().retries.jobs
jobs.mu.Lock()
defer jobs.mu.Unlock()
return jobs.running[hostileHost]
}
pinned := r.waitFor(time.Minute, func() bool { return running() >= pool-poolReserve(pool) })
r.metric("hostile_jobs_running", float64(running()))
r.check("the hostile knot pins the shared pool (setup)", pinned, "the hostile knot had %d of %d workers running, want %d", running(), pool, pool-poolReserve(pool))
var prioProbe, plainProbe grantProbes
w.probeGrant(prioRepos[0], syntax.DID("did:plc:prioprobe"), time.Minute, &prioProbe)
w.probeGrant(plainRepos[0], syntax.DID("did:plc:plainprobe"), time.Minute, &plainProbe)
prioProbe.wg.Wait()
plainProbe.wg.Wait()
missed := prioProbe.missed.Load() + plainProbe.missed.Load()
r.latency("priority_knot_apply", &prioProbe.latencies)
r.latency("plain_knot_apply", &plainProbe.latencies)
r.metric("probes_missed", float64(missed))
r.check("a priority knot's job runs at once while a hostile knot pins the pool", missed == 0 && prioProbe.quantile(1) <= budget,
"priority knot's event applied after %v with %d missed, want within %v while the hostile knot ran %d workers", prioProbe.quantile(1), missed, budget, running())
})
}