//go:build torture
package spindle
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"slices"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/bluesky-social/indigo/atproto/atcrypto"
"github.com/bluesky-social/indigo/atproto/auth"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/models"
)
func (w *tortureWorld) abuseRepos(ks []*tortureKnot, tag string, perKnot, commits int, workflows map[string]string) []*tortureRepo {
return w.addRepos(len(ks)*perKnot, func(i int) (*tortureKnot, syntax.DID) {
idx := i % len(ks)
return ks[idx], syntax.DID(fmt.Sprintf("did:plc:%s%d", tag, idx))
}, commits, workflows)
}
func abuseQueues(repos []*tortureRepo, from, to int) map[*tortureKnot][]pushDelivery {
queues := map[*tortureKnot][]pushDelivery{}
for i := from; i < to; i++ {
for _, repo := range repos {
queues[repo.knot] = append(queues[repo.knot], pushRange(repo, i, i+1)...)
}
}
return queues
}
func abuseRatio(n, of int64) float64 {
if of == 0 {
return 0
}
return float64(n) / float64(of)
}
func (w *tortureWorld) abuseHammer(ctx context.Context, conc int, token func() string, body []byte, st *deliveryStats) {
client := &http.Client{Timeout: 30 * time.Second, Transport: &http.Transport{MaxIdleConnsPerHost: conc}}
defer client.CloseIdleConnections()
var wg sync.WaitGroup
for range conc {
wg.Go(func() {
for ctx.Err() == nil {
st.attempts.Add(1)
reply := w.post(ctx, client, token(), body)
if ctx.Err() != nil {
return
}
st.saw(reply.status)
st.first.add(reply.elapsed)
}
})
}
wg.Wait()
}
type abuseHook struct {
srv *httptest.Server
got atomic.Int64
stop chan struct{}
once sync.Once
mu sync.Mutex
firstSeen map[string]time.Time
}
func newAbuseHook(r *tortureRun, hang time.Duration) *abuseHook {
h := &abuseHook{stop: make(chan struct{}), firstSeen: map[string]time.Time{}}
h.srv = httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
raw, _ := io.ReadAll(io.LimitReader(req.Body, 1<<20))
if hang > 0 {
select {
case <-time.After(hang):
case <-h.stop:
case <-req.Context().Done():
}
}
var p struct {
After string `json:"after"`
Repository struct {
FullName string `json:"full_name"`
} `json:"repository"`
}
_ = json.Unmarshal(raw, &p)
h.got.Add(1)
key := p.After + "|" + p.Repository.FullName
h.mu.Lock()
if _, ok := h.firstSeen[key]; !ok {
h.firstSeen[key] = time.Now()
}
h.mu.Unlock()
rw.WriteHeader(http.StatusOK)
}))
// cleanups run in reverse, and srv.Close waits on hung handlers unless release runs first
r.t.Cleanup(h.srv.Close)
r.t.Cleanup(h.release)
return h
}
func (h *abuseHook) release() { h.once.Do(func() { close(h.stop) }) }
func (h *abuseHook) seenAt(key string) (time.Time, bool) {
h.mu.Lock()
defer h.mu.Unlock()
at, ok := h.firstSeen[key]
return at, ok
}
func abuseHookKey(d pushDelivery) string {
return d.newSha + "|" + d.repo.owner.String() + "/" + strings.ToLower(d.repo.name)
}
// straight to the db, skips the per repo and per owner quotas
func (w *tortureWorld) abuseAddHook(repo *tortureRepo, url string) {
hook := &models.Webhook{RepoDid: repo.did, Url: url, Active: true, Events: []string{string(models.WebhookEventPush)}}
created, err := w.s.db.AddWebhook(context.Background(), hook, 0, 0)
if err != nil || !created {
w.r.t.Fatalf("add webhook: created=%v err=%v", created, err)
}
}
func (w *tortureWorld) abuseNotifyInflight() float64 {
return w.fromSpindle(func(s *Spindle) float64 { return float64(s.notifySlots.inflight()) })
}
func (w *tortureWorld) abuseUnknownInflight() float64 {
return w.fromSpindle(func(s *Spindle) float64 {
s.notifyAdmission.mu.Lock()
defer s.notifyAdmission.mu.Unlock()
return float64(s.notifyAdmission.total)
})
}
func (w *tortureWorld) abuseHeld(host string) float64 {
return w.fromSpindle(func(s *Spindle) float64 {
s.notifySlots.mu.Lock()
defer s.notifySlots.mu.Unlock()
return float64(s.notifySlots.held[host])
})
}
func (w *tortureWorld) abuseSybil(ctx context.Context, k *tortureKnot, repos []*tortureRepo, next *atomic.Int64, st *deliveryStats) {
client := &http.Client{Timeout: time.Minute}
sha := strings.Repeat("a", 40)
for ctx.Err() == nil {
repo := repos[int(next.Add(1))%len(repos)]
st.attempts.Add(1)
reply := w.post(ctx, client, w.token(k), w.notifyBody(repo, "refs/heads/main", zeroSha, sha))
if ctx.Err() != nil {
return
}
st.saw(reply.status)
if reply.status == http.StatusServiceUnavailable {
sleepCtx(ctx, 200*time.Millisecond)
}
}
}
func init() {
// shorter than the knot client's 30s timeout on purpose, which keeps slots turning over faster
registerTorture("notify-sybil", "hostile hosts notifying for repos nobody serves while real knots flush outboxes", func(r *tortureRun) {
hosts := r.load("hostile_hosts", 5)
conc := 6
hold := r.duration("hostile_hold", 6*time.Second)
perHost := 40
legitKnots := 4
perKnot := 2
pushes := int(r.param("pushes", 12))
budget := r.duration("legit_budget", 75*time.Second)
w := newTortureWorld(r)
w.maxJobs = 8
r.gauge("notify_inflight", w.abuseNotifyInflight)
r.gauge("notify_unknown_inflight", w.abuseUnknownInflight)
legit := w.addKnots(legitKnots)
rs := w.abuseRepos(legit, "sybillegit", perKnot, pushes, map[string]string{"ci.yaml": tortureWorkflow("sleep 20ms")})
hostile := w.addKnots(hosts)
hostileRepos := map[*tortureKnot][]*tortureRepo{}
for i, k := range hostile {
for j := range perHost {
owner := syntax.DID(fmt.Sprintf("did:plc:sybilowner%d", i*100000+j))
hostileRepos[k] = append(hostileRepos[k], w.addRepo(k, owner, 0, nil))
}
}
w.boot()
w.seed(rs, 3*time.Minute)
half := pushes / 2
var base deliveryStats
start := time.Now()
w.outboxes(context.Background(), abuseQueues(rs, 0, half), &base)
baseline := time.Since(start)
r.metric("baseline_flush_s", baseline.Seconds())
r.check("baseline delivers everything", base.accepted.len() == len(rs)*half, "%d of %d accepted before any attack", base.accepted.len(), len(rs)*half)
atkCtx, stopAtk := context.WithCancel(context.Background())
defer stopAtk()
var hst deliveryStats
var atk sync.WaitGroup
for _, k := range hostile {
k.delay.Store(int64(hold))
next := &atomic.Int64{}
for range conc {
atk.Go(func() { w.abuseSybil(atkCtx, k, hostileRepos[k], next, &hst) })
}
}
want := float64(min(notifyMaxUnknown, hosts*notifyMaxUnknownPerKnot))
saturated := r.waitFor(3*hold+5*time.Second, func() bool { return w.abuseUnknownInflight() >= want })
r.metric("unknown_pool_saturated", map[bool]float64{true: 1, false: 0}[saturated])
r.check("hostile knots fill the pool for unknown repos (setup)", saturated, "%.0f of %.0f unknown-repo seats held", w.abuseUnknownInflight(), want)
legitCtx, cancel := context.WithTimeout(context.Background(), budget)
var st deliveryStats
start = time.Now()
w.outboxes(legitCtx, abuseQueues(rs, half, pushes), &st)
attack := time.Since(start)
cancel()
stopAtk()
atk.Wait()
st.report(r, "legit_")
hst.report(r, "hostile_")
r.metric("attack_flush_s", attack.Seconds())
r.metric("legit_busy_ratio", abuseRatio(int64(st.seen("http_503")), st.attempts.Load()))
want2 := len(rs) * (pushes - half)
maxWait := st.accepted.quantile(1)
if st.accepted.len() < want2 {
// pushes that never landed waited the whole attack, or they'd look instant
maxWait = max(maxWait, attack)
}
r.metric("legit_max_wait_s", maxWait.Seconds())
r.note("hostile hold is %v per call and ensureRepo makes three calls to the hostile knot, production clients hold up to 30s each. the notify slots stay free of them, they only hold the unknown-repo seats", hold)
r.check("legit pushes all delivered under sybil load (N2)", st.accepted.len() == want2, "%d of %d legit pushes accepted within %v", st.accepted.len(), want2, budget)
r.check("legit knots mostly not told busy (N2)", abuseRatio(int64(st.seen("http_503")), st.attempts.Load()) <= 0.1, "%d of %d legit attempts got 503, want at most 10%%", st.seen("http_503"), st.attempts.Load())
r.check("legit flush time not blown up (N2)", attack <= 3*baseline+10*time.Second, "flush took %v, baseline %v, want <= 3x baseline + 10s", attack.Round(time.Millisecond), baseline.Round(time.Millisecond))
r.check("no legit push waits more than 20s (N2)", maxWait <= 20*time.Second, "longest wait %v, want <= 20s", maxWait.Round(time.Millisecond))
})
registerTorture("notify-slot-share", "a knot with served repos and a hanging fetch takes the notify slots, then real knots arrive", func(r *tortureRun) {
pool := int(r.param("slots", 16))
hold := r.duration("fetch_hold", 5*time.Second)
budget := r.duration("legit_budget", 90*time.Second)
const commits = 6
workflows := map[string]string{"ci.yaml": tortureWorkflow("sleep 20ms")}
w := newTortureWorld(r)
w.maxJobs = 8
w.notifySlots = pool
r.gauge("notify_inflight", w.abuseNotifyInflight)
legit := w.addKnots(3)
rs := w.abuseRepos(legit, "sharelegit", 2, commits, workflows)
hostile := w.addKnots(1)
hrs := w.abuseRepos(hostile, "sharehostile", 12, commits, workflows)
w.boot()
w.seed(append(slices.Clone(rs), hrs...), 3*time.Minute)
hk := hostile[0]
hk.gitDelay.Store(hold.Nanoseconds())
atkCtx, stopAtk := context.WithCancel(context.Background())
defer stopAtk()
ds := pushRanges(hrs, 0, commits)
var hst deliveryStats
var atk sync.WaitGroup
atk.Go(func() { w.flood(atkCtx, ds, 4*pool, 1000, &hst) })
shared := pool - poolReserve(pool)
reached := r.waitFor(30*time.Second, func() bool { return w.abuseHeld(hk.host()) >= float64(shared) })
r.check("an idle pool lets one knot use all of it but the reserve", reached, "the hostile knot held %.0f of %d slots, want the %d that aren't the reserve while nobody else waits", w.abuseHeld(hk.host()), pool, shared)
first := w.notifyWith(context.Background(), http.DefaultClient, pushRange(rs[0], 0, 1)[0])
r.metric("first_legit_notify_ms", float64(first.elapsed.Milliseconds()))
r.check("a real knot gets the reserve at once while a hanging knot holds the rest", first.status == http.StatusOK && first.elapsed <= 2*time.Second,
"the first legit push answered %d after %v, want 200 within 2s", first.status, first.elapsed.Round(time.Millisecond))
r.check("a hanging knot never holds the reserve", w.abuseHeld(hk.host()) <= float64(shared), "the hostile knot held %.0f of %d slots, want at most %d", w.abuseHeld(hk.host()), pool, shared)
// peak only starts once the first hostile calls are done holding their slots
var peak atomic.Int64
stopSample := make(chan struct{})
sampled := make(chan struct{})
go func() {
defer close(sampled)
select {
case <-time.After(hold + 2*time.Second):
case <-stopSample:
return
}
for {
select {
case <-stopSample:
return
case <-time.After(10 * time.Millisecond):
}
if held := int64(w.abuseHeld(hk.host())); held > peak.Load() {
peak.Store(held)
}
}
}()
legitCtx, cancel := context.WithTimeout(context.Background(), budget)
var st deliveryStats
start := time.Now()
w.outboxes(legitCtx, abuseQueues(rs, 0, commits), &st)
took := time.Since(start)
cancel()
close(stopSample)
<-sampled
stopAtk()
atk.Wait()
st.report(r, "legit_")
hst.report(r, "hostile_")
r.metric("legit_flush_s", took.Seconds())
r.metric("hostile_peak_held_after_hold", float64(peak.Load()))
want := len(rs) * commits
r.check("real knots get all their pushes delivered while a knot hangs on the slots", st.accepted.len() == want, "%d of %d legit pushes accepted within %v", st.accepted.len(), want, budget)
share := int64(max(2, pool/2))
r.check("a hanging knot is cut back to its share once others wait", peak.Load() <= share, "the hostile knot held up to %d of %d slots after the real knots were turned away, share with one other waiting is %d", peak.Load(), pool, share)
})
registerTorture("notify-bad-tokens", "a flood of forged tokens naming a real knot against that knot's own notifies", func(r *tortureRun) {
conc := 16
resolveDelay := 20 * time.Millisecond
window := 4 * time.Second
w := newTortureWorld(r)
ks := w.addKnots(1)
rs := w.abuseRepos(ks, "badtok", 2, 2, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")})
w.boot()
w.seed(rs, 3*time.Minute)
k, repo := ks[0], rs[0]
body := w.notifyBody(repo, "refs/heads/main", zeroSha, repo.shas[0])
if reply := w.post(context.Background(), http.DefaultClient, w.token(k), body); reply.status != http.StatusOK {
r.t.Fatalf("first legit notify answered %d: %s", reply.status, reply.body)
}
bad, err := atcrypto.GeneratePrivateKeyP256()
if err != nil {
r.t.Fatal(err)
}
lxm := syntax.NSID(org_tangled.EventNotifyEventNSID)
sign := func(aud string, ttl time.Duration, key atcrypto.PrivateKey) func() string {
return func() string {
token, err := auth.SignServiceAuth(k.did, aud, ttl, &lxm, key)
if err != nil {
panic(err)
}
return token
}
}
forged := sign(w.spindleDid(), time.Minute, bad)
w.ids.delay.Store(int64(resolveDelay))
probe := func(ctx context.Context, lat *latencies, notOK *atomic.Int64) {
client := &http.Client{Timeout: 10 * time.Second}
defer client.CloseIdleConnections()
tick := time.NewTicker(50 * time.Millisecond)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return
case <-tick.C:
}
reply := w.post(ctx, client, w.token(k), body)
if ctx.Err() != nil {
return
}
lat.add(reply.elapsed)
if reply.status != http.StatusOK {
notOK.Add(1)
}
}
}
var baseLat, floodLat latencies
var baseBad, floodBad atomic.Int64
baseCtx, cancelBase := context.WithTimeout(context.Background(), window)
probe(baseCtx, &baseLat, &baseBad)
cancelBase()
floodCtx, cancelFlood := context.WithTimeout(context.Background(), window)
var flood deliveryStats
var wg sync.WaitGroup
wg.Go(func() { w.abuseHammer(floodCtx, conc, forged, body, &flood) })
wg.Go(func() { probe(floodCtx, &floodLat, &floodBad) })
wg.Wait()
cancelFlood()
w.ids.delay.Store(0)
r.rate("forged_flood_requests", int(flood.attempts.Load()), window)
baseP50, floodP50 := baseLat.quantile(0.5), floodLat.quantile(0.5)
r.check("legit replays all accepted during the flood", floodBad.Load() == 0 && baseBad.Load() == 0, "%d not ok alone, %d under flood, want 0", baseBad.Load(), floodBad.Load())
r.check("legit median under forged flood stays near its quiet median (N1)", floodP50 <= 3*baseP50+50*time.Millisecond, "median %v quiet, %v under flood, want <= 3x quiet + 50ms", baseP50.Round(time.Millisecond), floodP50.Round(time.Millisecond))
})
registerTorture("webhooks-hang", "webhooks that never answer, sharing the delivery pool with healthy repos", func(r *tortureRun) {
badRepos := r.load("bad_repos", 6)
hooksPer := 4
pushes := 40
goodRepos := 3
goodPushes := 30
gap := 500 * time.Millisecond
hang := r.duration("hang", 2*time.Minute)
settle := r.duration("settle", 45*time.Second)
w := newTortureWorld(r)
r.gauge("webhook_rows", func() float64 { return w.rows("webhook_deliveries") })
ks := w.addKnots(2)
bad := w.addRepos(badRepos, func(int) (*tortureKnot, syntax.DID) { return ks[0], "did:plc:whbad" }, pushes, nil)
good := w.addRepos(goodRepos, func(int) (*tortureKnot, syntax.DID) { return ks[1], "did:plc:whgood" }, goodPushes, nil)
w.boot()
w.seed(append(slices.Clone(bad), good...), 3*time.Minute)
hung, fast := newAbuseHook(r, hang), newAbuseHook(r, 0)
// webhook queues are per hostname, and localhost gets a different one from 127.0.0.1
fastURL := strings.Replace(fast.srv.URL, "127.0.0.1", "localhost", 1)
for _, repo := range bad {
for i := range hooksPer {
w.abuseAddHook(repo, fmt.Sprintf("%s/hook%d", hung.srv.URL, i))
}
}
for _, repo := range good {
w.abuseAddHook(repo, fastURL)
}
ctx := context.Background()
var bst, gst deliveryStats
var wg sync.WaitGroup
wg.Go(func() { w.knotOutbox(ctx, abuseQueues(bad, 0, pushes)[ks[0]], &bst) })
sent := map[string]time.Time{}
wg.Go(func() {
client := &http.Client{Timeout: time.Minute}
for i := range goodPushes {
for _, repo := range good {
d := pushRange(repo, i, i+1)[0]
sent[abuseHookKey(d)] = time.Now()
reply := w.notifyWith(ctx, client, d)
gst.saw(reply.status)
gst.first.add(reply.elapsed)
}
sleepCtx(ctx, gap)
}
})
wg.Wait()
want := len(sent)
r.waitFor(settle, func() bool { return int(fast.got.Load()) >= want })
var hookLat latencies
missing := 0
for key, at := range sent {
seen, ok := fast.seenAt(key)
if !ok {
missing++
continue
}
hookLat.add(seen.Sub(at))
}
// otherwise a dropped delivery never shows up in the p99
for range missing {
hookLat.add(settle)
}
dids := make([]string, len(good))
for i, repo := range good {
dids[i] = "'" + repo.did.String() + "'"
}
saturated := w.countRows("select count(*) from webhook_deliveries where response_body = 'delivery queue saturated'")
goodSaturated := w.countRows("select count(*) from webhook_deliveries d join webhooks h on h.id = d.webhook_id where d.response_body = 'delivery queue saturated' and h.repo_did in (" + strings.Join(dids, ",") + ")")
hung.release()
bst.report(r, "bad_notify_")
gst.report(r, "good_notify_")
r.metric("good_hooks_missing", float64(missing))
r.metric("saturated_rows", saturated)
r.metric("good_saturated_rows", goodSaturated)
r.note("saturated rows are pruned to 50 per webhook, so the count is a floor")
hookP99 := hookLat.quantile(0.99)
pushP99 := gst.first.quantile(0.99)
r.check("healthy webhooks all delivered while others hang (N7)", missing == 0, "%d of %d healthy deliveries missing %v after the last push, want 0", missing, want, settle)
r.check("healthy webhooks delivered within 10s (N7)", hookP99 <= 10*time.Second, "p99 %v, want <= 10s", hookP99.Round(time.Millisecond))
r.check("no healthy delivery dropped as queue saturated (N7)", goodSaturated == 0, "%v healthy deliveries recorded as saturated, want 0", goodSaturated)
r.check("a push isn't slowed by the delivery backlog (N7)", pushP99 <= 3*time.Second, "healthy notify p99 %v, want <= 3s", pushP99.Round(time.Millisecond))
})
registerTorture("notify-hung-fetch", "one repo whose git fetch hangs while other repos get pushes", func(r *tortureRun) {
victims := 3
hold := r.duration("git_hold", 10*time.Second)
budget := r.duration("victim_budget", 5*time.Second)
w := newTortureWorld(r)
r.gauge("notify_inflight", w.abuseNotifyInflight)
ks := w.addKnots(2)
hungKnot, okKnot := ks[0], ks[1]
stuck := w.addRepo(hungKnot, "did:plc:hungowner", 2, nil)
var others []*tortureRepo
for range victims {
others = append(others, w.addRepo(okKnot, "did:plc:hungok", 2, nil))
}
late := w.addRepo(okKnot, "did:plc:hungok", 0, nil)
w.boot()
w.seed(append([]*tortureRepo{stuck}, others...), 3*time.Minute)
hungKnot.gitDelay.Store(int64(hold))
ctx := context.Background()
client := &http.Client{Timeout: 5 * time.Minute}
defer client.CloseIdleConnections()
var wg sync.WaitGroup
var stuckReply notifyReply
wg.Go(func() { stuckReply = w.notifyWith(ctx, client, pushRange(stuck, 0, 1)[0]) })
if !r.waitFor(10*time.Second, func() bool { return hungKnot.fetches.Load() > 0 }) {
r.t.Fatal("the stuck repo never reached git")
}
var otherLat latencies
var notOK atomic.Int64
for _, repo := range others {
wg.Go(func() {
reply := w.notifyWith(ctx, client, pushRange(repo, 0, 1)[0])
otherLat.add(reply.elapsed)
if reply.status != http.StatusOK {
notOK.Add(1)
}
})
}
var refresh time.Duration
var refreshErr error
wg.Go(func() {
start := time.Now()
refreshErr = w.s.RefreshRepo(ctx, stuck.did)
refresh = time.Since(start)
})
var lateServed time.Duration
var lateOK bool
wg.Go(func() {
start := time.Now()
w.emitDeclaration(late)
w.emitManifest(late)
lateOK = r.waitFor(3*time.Minute, func() bool { return w.servedCount([]*tortureRepo{late}) == 1 })
lateServed = time.Since(start)
})
wg.Wait()
r.latency("other_repo_notify", &otherLat)
r.metric("stuck_notify_s", stuckReply.elapsed.Seconds())
r.metric("stuck_repo_refresh_s", refresh.Seconds())
r.metric("late_repo_ingest_s", lateServed.Seconds())
r.note("git_hold %v is per git request, stuck_notify_s is how long the push really hung, production allows up to the 5 minute work budget", hold)
r.check("the stuck repo's own push finishes", stuckReply.status == http.StatusOK, "answered %d: %s", stuckReply.status, stuckReply.body)
r.check("all other repos' notifies answered ok", notOK.Load() == 0, "%d notifies not answered ok, want 0", notOK.Load())
r.check("other repos aren't held up by a hung fetch (N4)", otherLat.quantile(1) <= budget, "slowest took %v, want <= %v", otherLat.quantile(1).Round(time.Millisecond), budget)
r.check("a refresh of the repo isn't held up by its own hung fetch (N4)", refreshErr == nil && refresh <= budget, "refresh took %v with err %v, want <= %v", refresh.Round(time.Millisecond), refreshErr, budget)
r.check("an unrelated repo is ingested while a fetch hangs (N4)", lateOK && lateServed <= 10*time.Second, "served=%v after %v, want served within 10s", lateOK, lateServed.Round(time.Millisecond))
})
}