//go:build torture
package spindle
import (
"context"
"fmt"
"math/rand/v2"
"net/http"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
)
type pushDelivery struct {
repo *tortureRepo
ref string
oldSha string
newSha string
}
type deliveryStats struct {
first latencies
accepted latencies
attempts atomic.Int64
dropped atomic.Int64
statuses sync.Map
}
func (st *deliveryStats) saw(status int) {
n, _ := st.statuses.LoadOrStore(statusKind(status), new(atomic.Int64))
n.(*atomic.Int64).Add(1)
}
func (st *deliveryStats) seen(kind string) int {
n, ok := st.statuses.Load(kind)
if !ok {
return 0
}
return int(n.(*atomic.Int64).Load())
}
// 503 means busy, which doesn't count as an error
func (st *deliveryStats) serverErrors() int {
total := 0
st.statuses.Range(func(k, v any) bool {
if kind := k.(string); strings.HasPrefix(kind, "http_5") && kind != "http_503" {
total += int(v.(*atomic.Int64).Load())
}
return true
})
return total
}
func (st *deliveryStats) record(reply notifyReply, attempt int, start time.Time) {
st.attempts.Add(1)
st.saw(reply.status)
if attempt == 1 {
st.first.add(reply.elapsed)
}
if reply.status == http.StatusOK {
st.accepted.since(start)
}
}
func (st *deliveryStats) report(r *tortureRun, prefix string) {
r.latency(prefix+"first_attempt", &st.first)
r.latency(prefix+"accepted", &st.accepted)
r.metric(prefix+"attempts", float64(st.attempts.Load()))
r.metric(prefix+"dropped", float64(st.dropped.Load()))
st.statuses.Range(func(k, v any) bool {
r.count(prefix+k.(string), int(v.(*atomic.Int64).Load()))
return true
})
}
func (w *tortureWorld) knotOutbox(ctx context.Context, queue []pushDelivery, st *deliveryStats) {
client := &http.Client{Timeout: time.Minute}
backoff := time.Duration(0)
for _, d := range queue {
start := time.Now()
for attempt := 1; ctx.Err() == nil; attempt++ {
if resp, err := client.Get(w.api.URL + "/.well-known/did.json"); err == nil {
resp.Body.Close()
}
reply := w.notifyWith(ctx, client, d)
st.record(reply, attempt, start)
if reply.status == http.StatusOK {
backoff = 0
break
}
if reply.status >= 400 && reply.status < 500 && reply.status != http.StatusRequestTimeout && reply.status != http.StatusTooManyRequests {
st.dropped.Add(1)
break
}
backoff = min(max(backoff*2, time.Second), time.Minute)
sleepCtx(ctx, backoff)
}
}
}
func (w *tortureWorld) outboxes(ctx context.Context, queues map[*tortureKnot][]pushDelivery, st *deliveryStats) *latencies {
var finished latencies
start := time.Now()
var wg sync.WaitGroup
for _, queue := range queues {
wg.Go(func() {
w.knotOutbox(ctx, queue, st)
finished.since(start)
})
}
wg.Wait()
return &finished
}
func (w *tortureWorld) flood(ctx context.Context, ds []pushDelivery, concurrency, maxAttempts int, st *deliveryStats) {
var wg sync.WaitGroup
work := make(chan pushDelivery)
for range concurrency {
wg.Go(func() {
for d := range work {
start := time.Now()
wait := 50 * time.Millisecond
for attempt := 1; ; attempt++ {
reply := w.notifyWith(ctx, http.DefaultClient, d)
st.record(reply, attempt, start)
if reply.status == http.StatusOK {
break
}
if (reply.status >= 0 && reply.status < 500) || attempt == maxAttempts || ctx.Err() != nil {
st.dropped.Add(1)
break
}
sleepCtx(ctx, wait+rand.N(wait))
wait = min(wait*2, 2*time.Second)
}
}
})
}
for _, d := range ds {
work <- d
}
close(work)
wg.Wait()
}
func (w *tortureWorld) notifyWith(ctx context.Context, client *http.Client, d pushDelivery) notifyReply {
return w.post(ctx, client, w.token(d.repo.knot), w.notifyBody(d.repo, d.ref, d.oldSha, d.newSha))
}
func pushRange(repo *tortureRepo, from, to int) []pushDelivery {
old := zeroSha
if from > 0 {
old = repo.shas[from-1]
}
var out []pushDelivery
for _, sha := range repo.shas[from:to] {
out = append(out, pushDelivery{repo: repo, ref: "refs/heads/main", oldSha: old, newSha: sha})
old = sha
}
return out
}
func pushesOf(repo *tortureRepo, n int) []pushDelivery { return pushRange(repo, 0, n) }
func pushRanges(repos []*tortureRepo, from, to int) []pushDelivery {
var out []pushDelivery
for _, repo := range repos {
out = append(out, pushRange(repo, from, to)...)
}
return out
}
func ownerFor(i, perOwner int) syntax.DID {
return syntax.DID(fmt.Sprintf("did:plc:owner%04d", i/perOwner))
}
func (w *tortureWorld) checkDrained(within time.Duration) {
drained := w.r.waitFor(within, func() bool {
// check jobs first, a job leaves the table before it counts as running
return w.rows("jobs") == 0 && w.eng.running.Load() == 0 && w.unfinishedWorkflows() == 0
})
w.r.check("queue drains", drained, "jobs=%v running=%d", w.rows("jobs"), w.eng.running.Load())
w.r.check("nothing left unfinished", w.unfinishedWorkflows() == 0, "%v workflows not finished", w.unfinishedWorkflows())
}
func (w *tortureWorld) unfinishedWorkflows() float64 {
return w.countRows(`
select count(*) from workflow_statuses ws
where ws.id = (
select max(id) from workflow_statuses latest
where latest.pipeline_id = ws.pipeline_id and latest.workflow = ws.workflow
) and ws.status in ('pending', 'running')`)
}
func (w *tortureWorld) pushWorld(knots, repos, pushes int, workflows map[string]string) []*tortureRepo {
w.addKnots(knots)
rs := w.addRepos(repos, func(i int) (*tortureKnot, syntax.DID) {
return w.knots[i%knots], ownerFor(i, 3)
}, pushes, workflows)
w.boot()
w.seed(rs, 3*time.Minute)
return rs
}
func (w *tortureWorld) checkPipelines(r *tortureRun, want int, what string) {
got := int(w.rows("pipelines"))
r.check("one pipeline per unique push", got == want, "%d pipelines for %d unique pushes (%s)", got, want, what)
}
func init() {
registerTorture("notify-outbox-flush", "knots coming back with full outboxes, the way knot2 sends them", func(r *tortureRun) {
knots := r.load("knots", 12)
perKnot := 4
pushes := 12
dupRate := float64(0.1)
step := "sleep " + r.duration("step", 50*time.Millisecond).String()
w := newTortureWorld(r)
w.maxJobs = int(r.param("workers", 8))
rs := w.pushWorld(knots, knots*perKnot, pushes, map[string]string{"ci.yaml": tortureWorkflow(step)})
baseGoroutines := r.gaugeValue("goroutines")
queues := map[*tortureKnot][]pushDelivery{}
unique := 0
for _, repo := range rs {
for _, d := range pushRange(repo, 0, pushes) {
queues[repo.knot] = append(queues[repo.knot], d)
unique++
// a lost ack makes the knot resend, and some pushes get sent twice on purpose
if rand.Float64() < dupRate {
queues[repo.knot] = append(queues[repo.knot], d)
}
}
}
// only cross repo swaps, a repo's own pushes have to stay in order
for _, q := range queues {
rand.Shuffle(len(q), func(i, j int) {
if q[i].repo != q[j].repo {
q[i], q[j] = q[j], q[i]
}
})
}
var st deliveryStats
finished := w.outboxes(context.Background(), queues, &st)
st.report(r, "notify_")
r.latency("knot_finished", finished)
w.checkDrained(5 * time.Minute)
r.check("no delivery dropped", st.dropped.Load() == 0, "%d deliveries dropped by the knot", st.dropped.Load())
r.check("no server errors besides busy", st.serverErrors() == 0, "%d 5xx answers that weren't 503", st.serverErrors())
w.checkPipelines(r, unique, fmt.Sprintf("%d deliveries", st.attempts.Load()))
r.settles("goroutines", baseGoroutines+20, 30*time.Second)
})
registerTorture("notify-flood", "one knot that ignores the outbox discipline and blasts the caps", func(r *tortureRun) {
repos := r.load("repos", 16)
pushes := 10
concurrency := 64
replays := float64(0.5)
step := "sleep " + r.duration("step", 20*time.Millisecond).String()
w := newTortureWorld(r)
w.maxJobs = int(r.param("workers", 8))
rs := w.pushWorld(1, repos, pushes, map[string]string{"ci.yaml": tortureWorkflow(step)})
baseGoroutines := r.gaugeValue("goroutines")
ds := pushRanges(rs, 0, pushes)
unique := len(ds)
for _, d := range ds[:unique] {
if rand.Float64() < replays {
ds = append(ds, d)
}
}
rand.Shuffle(len(ds), func(i, j int) { ds[i], ds[j] = ds[j], ds[i] })
var st deliveryStats
w.flood(context.Background(), ds, concurrency, 40, &st)
st.report(r, "notify_")
r.metric("busy_ratio", float64(st.seen("http_503"))/float64(st.attempts.Load()))
w.checkDrained(5 * time.Minute)
r.check("every delivery accepted", st.dropped.Load() == 0, "%d deliveries gave up", st.dropped.Load())
r.check("no server errors besides busy", st.serverErrors() == 0, "%d 5xx answers that weren't 503", st.serverErrors())
w.checkPipelines(r, unique, fmt.Sprintf("%d deliveries", len(ds)))
r.settles("goroutines", baseGoroutines+20, 30*time.Second)
})
}