//go:build torture
package spindle
import (
"context"
"fmt"
"sync"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/jetstream"
)
func init() {
registerTorture("soak", "steady mixed traffic for a long time, to see what grows that shouldn't", func(r *tortureRun) {
knots := r.load("knots", 4)
perKnot := 3
length := 3 * time.Minute
interval := time.Second
churn := 500 * time.Millisecond
step := "lines 200; sleep " + r.duration("step", 100*time.Millisecond).String()
commits := int(length/interval)/perKnot + 2
w := newTortureWorld(r)
w.maxJobs = int(r.param("workers", 8))
rs := w.pushWorld(knots, knots*perKnot, commits, map[string]string{"ci.yaml": tortureWorkflow(step)})
r.note("pushes per knot are serial and spaced like a busy knot, the engine is the fake one")
ctx, cancel := context.WithTimeout(context.Background(), length)
defer cancel()
var st deliveryStats
var wg sync.WaitGroup
for _, k := range w.knots {
var mine []*tortureRepo
for _, repo := range rs {
if repo.knot == k {
mine = append(mine, repo)
}
}
wg.Go(func() {
next := make(map[*tortureRepo]int)
tick := time.NewTicker(interval)
defer tick.Stop()
for i := 0; ; i++ {
select {
case <-ctx.Done():
return
case <-tick.C:
}
repo := mine[i%len(mine)]
n := next[repo]
if n >= len(repo.shas) {
continue
}
next[repo] = n + 1
w.knotOutbox(ctx, pushRange(repo, n, n+1), &st)
}
})
}
wg.Go(func() {
tick := time.NewTicker(churn)
defer tick.Stop()
for i := 0; ; i++ {
select {
case <-ctx.Done():
return
case <-tick.C:
}
repo := rs[i%len(rs)]
who := syntax.DID(fmt.Sprintf("did:plc:collab%02d", i%26))
op := jetstream.CommitOperationCreate
if (i/len(rs))%2 == 1 {
op = jetstream.CommitOperationDelete
repo.knot.erase(repo.did, org_tangled.GraphCollaboratorNSID, who.String())
} else {
repo.knot.put(repo.did, org_tangled.GraphCollaboratorNSID, who.String(), map[string]string{"createdAt": "2026-01-01T00:00:00Z"})
}
w.emitCollaborator(repo, who, op)
}
})
wg.Wait()
st.report(r, "notify_")
// no heap check, HeapInuse swings with the gc phase and the fake jetstream holds every event
for gauge, limit := range map[string]float64{"goroutines": 10, "fds": 2} {
growth := r.growth(gauge)
r.check(gauge+" stays flat", growth <= limit, "%s grows %.1f per minute over the second half, want <= %.0f", gauge, growth, limit)
}
w.checkDrained(2 * time.Minute)
r.check("no delivery dropped", st.dropped.Load() == 0, "%d deliveries dropped", st.dropped.Load())
r.check("retry table drains", r.waitFor(time.Minute, func() bool { return w.rows("repo_retries") == 0 }), "%v repo retries left", w.rows("repo_retries"))
r.metric("pipelines_per_min", w.rows("pipelines")/length.Minutes())
r.metric("db_mb_per_1k_pipelines", r.gaugeValue("db_mb")/w.rows("pipelines")*1000)
})
}