//go:build torture
package spindle
import (
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"slices"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/bluesky-social/indigo/atproto/atcrypto"
"github.com/bluesky-social/indigo/atproto/identity"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/quota"
"tangled.org/core/sqlite"
)
type admUser struct {
did syntax.DID
priv atcrypto.PrivateKey
}
// has to run after addRepo, which overwrites the owner's entry
func (w *tortureWorld) admAddUser(did syntax.DID, pds *tortureKnot) *admUser {
priv, id := w.newKeyedIdentity(did)
if pds != nil {
id.Services = map[string]identity.ServiceEndpoint{
"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: pds.url()},
}
}
w.ids.Insert(id)
return &admUser{did: did, priv: priv}
}
func (w *tortureWorld) admToken(u *admUser, nsid string) string { return w.sign(u.did, u.priv, nsid) }
// same setup as Start, which keeps the timeouts real
func (w *tortureWorld) admServe() string {
srv := newHTTPServer("", w.s.Router())
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
w.r.t.Fatal(err)
}
go srv.Serve(ln)
w.r.t.Cleanup(func() { srv.Close() })
return ln.Addr().String()
}
func admFdKinds() string {
targets, err := fdTargets()
if err != nil {
return "unreadable"
}
kinds := map[string]int{}
for _, target := range targets {
if target == "" {
continue
}
kind, _, found := strings.Cut(target, ":")
if !found {
kind = "file"
}
kinds[kind]++
}
return fmt.Sprint(kinds)
}
// built before the db exists, the real store gets swapped in after boot
type admQuotaStore struct {
qs atomic.Pointer[db.QuotaStore]
reserves atomic.Int64
}
func (s *admQuotaStore) Reserve(ctx context.Context, req quota.ReserveRequest) (quota.Reservation, error) {
s.reserves.Add(1)
return s.qs.Load().Reserve(ctx, req)
}
func (s *admQuotaStore) CheckDenied(ctx context.Context, req quota.ReserveRequest) (quota.Reservation, bool, error) {
return s.qs.Load().CheckDenied(ctx, req)
}
func (s *admQuotaStore) BeginCommit(ctx context.Context, id string) error {
return s.qs.Load().BeginCommit(ctx, id)
}
func (s *admQuotaStore) Commit(ctx context.Context, id string) error {
return s.qs.Load().Commit(ctx, id)
}
func (s *admQuotaStore) Release(ctx context.Context, id string) error {
return s.qs.Load().Release(ctx, id)
}
type admQuotaObserver struct {
depth atomic.Int64
}
func (o *admQuotaObserver) RecordDecision(kind, resource string, allowed, temporary bool, reason string) {
}
func (o *admQuotaObserver) SetWaitDepth(kind string, depth int64) {
if kind == string(quota.KindWorkflow) {
o.depth.Store(depth)
}
}
func (o *admQuotaObserver) RecordWait(kind, resource string, duration time.Duration) {}
func init() {
// twice the read timeout, which means only the server cutting them off ends it
registerTorture("http-slowloris", "connections that never finish their headers or body against a real listener", func(r *tortureRun) {
conns := r.load("conns", 600)
hold := r.duration("hold", 2*newHTTPServer("", nil).ReadTimeout)
drip := time.Second
w := newTortureWorld(r)
w.maxJobs = 2
user := w.admAddUser("did:plc:slowuser", nil)
w.boot()
addr := w.admServe()
// warm up files and pools so the baseline doesn't pay for them
for range 20 {
if resp, err := http.Get("http://" + addr + "/_health"); err == nil {
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
}
}
http.DefaultClient.CloseIdleConnections()
time.Sleep(2 * time.Second)
baseGoroutines := r.gaugeValue("goroutines")
baseFds := r.gaugeValue("fds")
token := w.admToken(user, org_tangled.SecretAddSecretNSID)
var headerOpen, bodyOpen, headerCut, bodyCut atomic.Int64
stop := make(chan struct{})
var attackers sync.WaitGroup
for i := range conns {
attackers.Go(func() {
c, err := net.DialTimeout("tcp", addr, 5*time.Second)
if err != nil {
r.count("dial", 1)
return
}
defer c.Close()
body := i%2 == 1
opening := "GET /_health HTTP/1.1\r\nHost: spindle\r\n"
next := "X-a: b\r\n"
if body {
bodyOpen.Add(1)
opening = fmt.Sprintf("POST /xrpc/%s HTTP/1.1\r\nHost: spindle\r\nAuthorization: Bearer %s\r\nContent-Type: application/json\r\nContent-Length: 1000000\r\n\r\n{\"repo\":\"", org_tangled.SecretAddSecretNSID, token)
next = "a"
} else {
headerOpen.Add(1)
}
if _, err := c.Write([]byte(opening)); err != nil {
return
}
tick := time.NewTicker(drip)
defer tick.Stop()
buf := make([]byte, 1)
for {
select {
case <-stop:
return
case <-tick.C:
}
c.SetWriteDeadline(time.Now().Add(time.Second))
if _, err := c.Write([]byte(next)); err == nil {
c.SetReadDeadline(time.Now().Add(time.Millisecond))
n, rerr := c.Read(buf)
var ne net.Error
if n == 0 && errors.As(rerr, &ne) && ne.Timeout() {
continue
}
}
if body {
bodyCut.Add(1)
} else {
headerCut.Add(1)
}
return
}
})
}
var legit latencies
var legitFailed atomic.Int64
probing := make(chan struct{})
go func() {
defer close(probing)
tick := time.NewTicker(250 * time.Millisecond)
defer tick.Stop()
for {
select {
case <-stop:
return
case <-tick.C:
}
client := &http.Client{Timeout: 3 * time.Second, Transport: &http.Transport{DisableKeepAlives: true}}
start := time.Now()
resp, err := client.Get("http://" + addr + "/_health")
if err != nil {
legitFailed.Add(1)
continue
}
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
if resp.StatusCode != http.StatusOK {
legitFailed.Add(1)
continue
}
legit.since(start)
}
}()
time.Sleep(hold)
r.metric("fds_during", r.gaugeValue("fds"))
r.metric("goroutines_during", r.gaugeValue("goroutines"))
close(stop)
attackers.Wait()
<-probing
r.metric("header_conns", float64(headerOpen.Load()))
r.metric("body_conns", float64(bodyOpen.Load()))
r.metric("header_cut_by_server", float64(headerCut.Load()))
r.metric("body_cut_by_server", float64(bodyCut.Load()))
r.latency("legit_health", &legit)
r.metric("legit_failed", float64(legitFailed.Load()))
r.check("conns that never finish their headers get dropped (A1)", headerCut.Load() == headerOpen.Load(), "%d of %d header-dripping conns cut by the server within %v, want all", headerCut.Load(), headerOpen.Load(), hold)
r.check("conns that never finish their body get dropped (A1)", bodyCut.Load() == bodyOpen.Load(), "%d of %d body-dripping conns cut by the server within %v, want all", bodyCut.Load(), bodyOpen.Load(), hold)
r.check("legit requests keep getting through", legitFailed.Load() == 0, "%d health probes failed or timed out, want 0", legitFailed.Load())
r.note("fds after the attackers left: %s, baseline %.0f", admFdKinds(), baseFds)
r.settles("goroutines", baseGoroutines+50, 30*time.Second)
r.settles("fds", baseFds+30, 30*time.Second)
})
// the 50ms quota tick writes for every waiting owner, that's the cost here
registerTorture("quota-churn", "many owners at a tight workflow limit, what the quota manager costs while they wait", func(r *tortureRun) {
owners := r.load("owners", 40)
queued := int(r.param("queued", 2))
limit := int64(r.param("limit", 1))
hold := r.duration("hold", 20*time.Second)
window := 4 * time.Second
obs := &admQuotaObserver{}
store := &admQuotaStore{}
qm := quota.NewManager(store, 50*time.Millisecond, obs)
r.t.Cleanup(qm.Close)
w := newTortureWorld(r)
w.qm = qm
w.maxJobs = owners*queued + 8
r.gauge("quota_waiters", func() float64 { return float64(obs.depth.Load()) })
r.gauge("quota_reservations", func() float64 { return w.rows("quota_reservations") })
flow := map[string]string{"ci.yaml": tortureWorkflow("sleep " + hold.String())}
knot := w.addKnot()
repos := w.addRepos(owners, func(i int) (*tortureKnot, syntax.DID) { return knot, ownerFor(i, 1) }, 2*queued, flow)
fresh := w.addRepos(1, func(int) (*tortureKnot, syntax.DID) { return knot, "did:plc:fresh" }, 1, flow)
w.boot()
w.seed(append(slices.Clone(repos), fresh...), 3*time.Minute)
defaults := quota.Defaults{quota.ScopeUser: quota.Resources{quota.ResourceWorkflows: limit}}
store.qs.Store(db.NewQuotaStore(w.s.db, defaults))
ctx := context.Background()
w.probeTable()
var quiet latencies
for i := range 50 {
start := time.Now()
if _, err := w.s.db.Exec(`insert into torture_probe (n) values (?)`, -i); err != nil {
r.t.Fatal(err)
}
quiet.since(start)
}
var probe latencies
var probeErrors atomic.Int64
stopProbe := w.probeWrites(100*time.Millisecond, &probe, &probeErrors)
ds := pushRanges(repos, 0, queued)
var st deliveryStats
w.flood(ctx, ds, 32, 20, &st)
st.report(r, "notify_")
want := int64(owners) * (int64(queued) - limit)
formed := r.waitFor(60*time.Second, func() bool { return obs.depth.Load() >= want*9/10 })
r.check("waiters reach the quota queue (setup)", formed, "queue depth %d, want at least 90%% of %d", obs.depth.Load(), want)
r0 := store.reserves.Load()
t0 := time.Now()
time.Sleep(window / 2)
freshAt := time.Now()
w.notifyWith(ctx, http.DefaultClient, pushRange(fresh[0], 0, 1)[0])
freshStarted := r.waitFor(30*time.Second, func() bool { return slices.Contains(w.eng.startOrder(), "did:plc:fresh") })
freshTook := time.Since(freshAt)
if rest := window - time.Since(t0); rest > 0 {
time.Sleep(rest)
}
secs := time.Since(t0).Seconds()
reserves := store.reserves.Load() - r0
stopProbe()
r.metric("reserve_per_s", float64(reserves)/secs)
r.metric("fresh_owner_start_ms", float64(freshTook.Milliseconds()))
r.metric("probe_errors", float64(probeErrors.Load()))
r.latency("probe_write", &probe)
perOwner := float64(reserves) / secs / float64(owners)
r.check("idle waiters don't poll the db (A7)", perOwner <= 1, "%.1f Reserve write txs per waiting owner per second while nothing was released, want at most 1", perOwner)
quietP99 := quiet.quantile(0.99)
r.check("unrelated writes aren't starved by the quota tick (A7)", probe.quantile(0.99) <= 10*quietP99+50*time.Millisecond && probeErrors.Load() == 0, "probe insert p99 %v and %d failed, %v when quiet, want within 10x plus 50ms and 0 failed", probe.quantile(0.99), probeErrors.Load(), quietP99)
r.check("an unrelated owner with quota starts without waiting on the queue (A7)", freshStarted && freshTook < hold/2, "started=%v after %v with %d waiters queued, want started in under half of one workflow's %v hold", freshStarted, freshTook, obs.depth.Load(), hold)
w.checkDrained(5 * time.Minute)
leaked := r.waitFor(10*time.Second, func() bool { return w.rows("quota_reservations") == 0 })
r.check("no reservations leaked after the run (A7)", leaked, "%v quota_reservations rows left, want 0", w.rows("quota_reservations"))
r.settles("quota_waiters", 0, 30*time.Second)
half := repos[:max(1, owners/2)]
ds2 := pushRanges(half, queued, 2*queued)
var st2 deliveryStats
w.flood(ctx, ds2, 32, 20, &st2)
holding := r.waitFor(30*time.Second, func() bool {
return w.rows("quota_reservations") > 0 && obs.depth.Load() > 0
})
r.note("stopped with reservations held and waiters queued: %v", holding)
w.stop()
raw, err := sqlite.Open(w.dbPath())
if err != nil {
r.t.Fatal(err)
}
var left int
if err := raw.QueryRow(`select count(*) from quota_reservations`).Scan(&left); err != nil {
r.t.Fatal(err)
}
raw.Close()
r.metric("reservations_after_clean_stop", float64(left))
r.check("a clean stop leaves no reservations (A7)", left == 0, "%d quota_reservations rows after stopping mid-run, want 0", left)
// Recover isn't called by Start, only server.go's Run does it
d, err := db.Make(ctx, w.dbPath())
if err != nil {
r.t.Fatal(err)
}
qs := db.NewQuotaStore(d, nil)
leaks := owners * queued
for i := range leaks {
id := fmt.Sprintf("crash-%d", i)
if _, err := qs.Reserve(ctx, quota.ReserveRequest{
ID: id,
Kind: quota.KindWorkflow,
Key: id,
Identity: quota.Identity{OwnerDID: ownerFor(i, 1).String(), RepoDID: "did:plc:crashrepo"},
Resources: quota.Resources{quota.ResourceWorkflows: 1},
}); err != nil {
r.t.Fatal(err)
}
}
_, recoverErr := qs.Recover(ctx, nil)
var after int
if err := d.QueryRow(`select count(*) from quota_reservations`).Scan(&after); err != nil {
r.t.Fatal(err)
}
d.Close()
r.check("recover clears what a crash leaves (A7)", recoverErr == nil && after == 0, "err=%v and %d of %d crash reservations still there, want no error and 0", recoverErr, after, leaks)
})
// the first workflow waits on it because resumeWipes runs before the workers start
registerTorture("ban-restart", "a restart with a long ban list, time to the first workflow", func(r *tortureRun) {
bans := r.load("bans", 5000)
w := newTortureWorld(r)
w.maxJobs = 2
knot := w.addKnot()
repo := w.addRepo(knot, "did:plc:banowner", 3, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")})
w.boot()
w.seed([]*tortureRepo{repo}, time.Minute)
ctx := context.Background()
pushes := pushRange(repo, 0, 3)
w.notifyWith(ctx, http.DefaultClient, pushes[0])
if !r.waitFor(30*time.Second, func() bool { return w.eng.started.Load() >= 1 }) {
r.t.Fatal("the first workflow never started")
}
restart := func(i int) time.Duration {
before := w.eng.started.Load()
start := time.Now()
w.boot()
reply := w.notifyWith(ctx, http.DefaultClient, pushes[i])
if reply.status != http.StatusOK {
r.t.Fatalf("push %d after restart answered %d: %s", i, reply.status, reply.body)
}
if !r.waitFor(5*time.Minute, func() bool { return w.eng.started.Load() > before }) {
r.t.Fatalf("no workflow started within 5m of restart %d", i)
}
return time.Since(start)
}
w.stop()
quiet := restart(1)
w.stop()
d, err := db.Make(ctx, w.dbPath())
if err != nil {
r.t.Fatal(err)
}
for i := range bans {
if err := d.PutBan(db.BanEntry{SubjectDid: syntax.DID(fmt.Sprintf("did:plc:banned%06d", i))}); err != nil {
r.t.Fatal(err)
}
}
d.Close()
banned := restart(2)
r.metric("restart_no_bans_s", quiet.Seconds())
r.metric("restart_with_bans_s", banned.Seconds())
r.check("a restart doesn't wait on the ban list (A10)", banned <= quiet+10*time.Second, "first workflow after boot took %v with %d bans and %v without, want at most 10s more", banned, bans, quiet)
})
// unknown repo lookups hit the knot, only lookupSlots of them run at once
// well under the knot client's 30s timeout
registerTorture("unknown-repo-stripes", "made-up repo dids on a stalling knot against legit refreshes", func(r *tortureRun) {
attackers := r.load("attackers", 200)
probes := 40
stall := 2 * time.Second
const lookupSlots = 2
w := newTortureWorld(r)
w.maxJobs = 2
slow := w.addKnot()
fast := w.addKnot()
evilRepos := w.addRepos(attackers, func(int) (*tortureKnot, syntax.DID) { return slow, "did:plc:evil" }, 0, nil)
baseRepos := w.addRepos(probes, func(int) (*tortureKnot, syntax.DID) { return fast, "did:plc:legit" }, 0, nil)
hitRepos := w.addRepos(probes, func(int) (*tortureKnot, syntax.DID) { return fast, "did:plc:legit" }, 0, nil)
// they look unknown since the web made them before spindle saw the event
webRepos := w.addRepos(10, func(int) (*tortureKnot, syntax.DID) { return fast, "did:plc:legit" }, 0, nil)
evil := w.admAddUser("did:plc:evil", slow)
legit := w.admAddUser("did:plc:legit", fast)
w.boot()
w.seed(append(append([]*tortureRepo{}, baseRepos...), hitRepos...), 2*time.Minute)
ctx := context.Background()
client := &http.Client{Timeout: 3 * time.Minute}
listSecrets := func(u *admUser, repo *tortureRepo) notifyReply {
token := w.admToken(u, org_tangled.SecretListSecretsNSID)
return w.call(ctx, client, http.MethodGet, org_tangled.SecretListSecretsNSID, "repo="+repo.did.String(), token, nil)
}
probe := func(repos []*tortureRepo, into *latencies) {
var wg sync.WaitGroup
for _, repo := range repos {
wg.Go(func() {
reply := listSecrets(legit, repo)
into.add(reply.elapsed)
if reply.status != http.StatusOK {
r.count("probe_"+statusKind(reply.status), 1)
}
})
}
wg.Wait()
}
var base, attacked latencies
probe(baseRepos, &base)
slow.delay.Store(int64(stall))
var held sync.WaitGroup
var attackersDone, attackersBusy atomic.Int64
for _, repo := range evilRepos {
held.Go(func() {
reply := listSecrets(evil, repo)
attackersDone.Add(1)
if reply.status == http.StatusServiceUnavailable {
attackersBusy.Add(1)
}
r.count("attacker_"+statusKind(reply.status), 1)
})
}
time.Sleep(time.Second)
stillStalled := int64(attackers) - attackersDone.Load()
busy := attackersBusy.Load()
var web latencies
var webBusy atomic.Int64
var webWG sync.WaitGroup
for _, repo := range webRepos {
webWG.Go(func() {
reply := listSecrets(legit, repo)
web.add(reply.elapsed)
if reply.status == http.StatusServiceUnavailable {
webBusy.Add(1)
}
})
}
probe(hitRepos, &attacked)
webWG.Wait()
held.Wait()
r.metric("attackers_stalled_during_probes", float64(stillStalled))
r.metric("attackers_refused_busy", float64(busy))
r.metric("web_created_repos_refused_busy", float64(webBusy.Load()))
r.latency("web_created_repo_lookup", &web)
r.check("attackers are refused busy at once and only the lookup slots stall (setup)", stillStalled <= lookupSlots && busy >= int64(attackers)-lookupSlots,
"%d of %d attacker requests still in flight after 1s and %d answered busy, want at most %d in flight and the rest refused busy", stillStalled, attackers, busy, lookupSlots)
r.note("a repo the web just created and spindle hasn't heard of takes a lookup slot like an attacker's repo, so while the attackers hold them %d of %d such requests were refused busy", webBusy.Load(), len(webRepos))
baseP90 := base.quantile(0.9)
attackedP90 := attacked.quantile(0.9)
r.check("a served repo's request isn't queued behind stalled lookups (A3)", attackedP90 <= baseP90+time.Second, "p90 %v under attack and %v alone with %d attacker dids on a stalling knot, want within 1s of alone", attackedP90, baseP90, attackers)
})
}