//go:build torture
package spindle
import (
"context"
"fmt"
"math"
"math/rand/v2"
"sync"
"sync/atomic"
"time"
"github.com/bluesky-social/indigo/atproto/identity"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/jetstream"
"tangled.org/core/rbac"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/knotclient"
)
// copied from spindle/knotclient, change both together
const (
ingListPageSize = 100
ingListPageCap = 32
// a refresh makes 3 requests: manifest, owner declaration, collaborator list
ingRequestsPerRefresh = 3
)
func (w *tortureWorld) isCollab(repo *tortureRepo, who syntax.DID) bool {
s := w.spindle()
if s == nil {
return false
}
ok, err := s.e.IsRepoCollaborator(who.String(), rbac.ThisServer, repo.did.String())
return err == nil && ok
}
// knot first, jetstream only hears about it after
func ingPutCollaborator(repo *tortureRepo, who syntax.DID) {
repo.knot.put(repo.did, org_tangled.GraphCollaboratorNSID, who.String(), map[string]string{"createdAt": "2026-01-01T00:00:00Z"})
}
func (w *tortureWorld) grantCollaborator(repo *tortureRepo, who syntax.DID) int64 {
ingPutCollaborator(repo, who)
return w.emitCollaborator(repo, who, jetstream.CommitOperationCreate)
}
func (w *tortureWorld) revokeCollaborator(repo *tortureRepo, who syntax.DID) int64 {
repo.knot.erase(repo.did, org_tangled.GraphCollaboratorNSID, who.String())
return w.emitCollaborator(repo, who, jetstream.CommitOperationDelete)
}
func ingXrpc(knots []*tortureKnot) int64 {
var n int64
for _, k := range knots {
n += k.xrpc.Load()
}
return n
}
func ingManifestReads(knots []*tortureKnot) int64 {
var n int64
for _, k := range knots {
n += k.readsOf(org_tangled.RepoManifestNSID)
}
return n
}
type grantProbes struct {
latencies
missed atomic.Int64
wg sync.WaitGroup
}
func (w *tortureWorld) probeGrant(repo *tortureRepo, who syntax.DID, within time.Duration, p *grantProbes) int64 {
start := time.Now()
us := w.grantCollaborator(repo, who)
p.wg.Go(func() {
if w.r.waitFor(within, func() bool { return w.isCollab(repo, who) }) {
p.since(start)
return
}
p.missed.Add(1)
})
return us
}
type ingBurst struct {
peak atomic.Int64
stop chan struct{}
done chan struct{}
}
func ingWatchBurst(count func() int64) *ingBurst {
b := &ingBurst{stop: make(chan struct{}), done: make(chan struct{})}
ring := make([]int64, 10)
base := count()
for i := range ring {
ring[i] = base
}
go func() {
defer close(b.done)
tick := time.NewTicker(100 * time.Millisecond)
defer tick.Stop()
for i := 0; ; i++ {
select {
case <-b.stop:
return
case <-tick.C:
}
n := count()
if d := n - ring[i%len(ring)]; d > b.peak.Load() {
b.peak.Store(d)
}
ring[i%len(ring)] = n
}
}()
return b
}
func (b *ingBurst) finish() int64 {
close(b.stop)
<-b.done
return b.peak.Load()
}
func ingDueSpread(w *tortureWorld) (sameSecond, distinct int) {
s := w.spindle()
if s == nil {
return 0, 0
}
rows, err := s.db.Query("select due, count(*) from repo_retries group by due")
if err != nil {
return 0, 0
}
defer rows.Close()
for rows.Next() {
var due int64
var n int
if rows.Scan(&due, &n) != nil {
return 0, 0
}
sameSecond = max(sameSecond, n)
distinct++
}
return sameSecond, distinct
}
func init() {
registerTorture("ingest-hostile-manifests", "crafted manifests from unadmitted dids while a served repo keeps getting events", func(r *tortureRun) {
slow := r.load("slow_attackers", 20)
noKnot := r.load("noknot_attackers", 200)
enrolledNoKnot := r.load("enrolled_noknot_attackers", 100)
delay := time.Second
probes := 5
budget := r.duration("legit_budget", 5*time.Second)
retryRounds := 2
r.note("attacker_delay is 1s where production waits up to 30s per request, and spindle makes two such requests per unadmitted manifest (ING-1), so real stalls are 30x these")
w := newTortureWorld(r)
w.admission = config.AdmissionClosed
good, evil := w.addKnot(), w.addKnot()
legit := w.addRepo(good, testOwner, 0, nil)
w.boot()
w.seed([]*tortureRepo{legit}, 30*time.Second)
evil.delay.Store(delay.Nanoseconds())
lookupsBefore := w.ids.lookups.Load()
var last int64
for i := range slow {
attacker := w.addRepo(evil, syntax.DID(fmt.Sprintf("did:plc:evil%04d", i)), 0, nil)
last = w.emitManifest(attacker)
}
for i := range noKnot {
did := syntax.DID(fmt.Sprintf("did:plc:noknot%05d", i))
owner := syntax.DID(fmt.Sprintf("did:plc:nkowner%05d", i))
w.ids.Insert(identity.Identity{DID: did})
last = w.js.commit(did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, jetstream.CommitOperationUpdate, manifestRecord(owner, "x", w.spindleDid()))
}
// the owner is allowlisted, which leaves the missing knot as the only thing stopping these
for i := range enrolledNoKnot {
did := syntax.DID(fmt.Sprintf("did:plc:enrollednk%05d", i))
owner := syntax.DID(fmt.Sprintf("did:plc:enrolledowner%05d", i))
if err := db.AddToAllowlist(w.s.db, owner); err != nil {
r.t.Fatal(err)
}
w.ids.Insert(identity.Identity{DID: did})
last = w.js.commit(did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, jetstream.CommitOperationUpdate, manifestRecord(owner, "x", w.spindleDid()))
}
within := time.Duration(slow)*2*delay + time.Minute
var applied grantProbes
for i := range probes {
last = w.probeGrant(legit, syntax.DID(fmt.Sprintf("did:plc:probe%04d", i)), within, &applied)
}
applied.wg.Wait()
drained := w.consumed(last, within)
// past an event doesn't mean its repo job finished
r.waitFor(time.Minute, func() bool { return w.rows("repo_retries") == 0 })
r.latency("legit_apply", &applied.latencies)
r.metric("legit_missed", float64(applied.missed.Load()))
r.metric("attacker_requests", float64(evil.xrpc.Load()))
r.metric("dir_cold_lookups", float64(w.ids.lookups.Load()-lookupsBefore))
retryRows := w.rows("repo_retries")
r.metric("retry_rows", retryRows)
r.check("ING-1 legit events apply while hostile manifests queue", applied.missed.Load() == 0 && applied.quantile(1) <= budget,
"%d of %d probes applied, slowest %v, want all within %v", applied.len(), probes, applied.quantile(1), budget)
r.check("ING-1 unadmitted dids cost no fetches to hosts they pick", evil.xrpc.Load() == 0,
"%d requests reached the attacker knot for %d unadmitted manifests", evil.xrpc.Load(), slow)
r.check("ING-1 reader gets through the hostile backlog", drained, "reader caught up %v, waited up to %v", drained, within)
r.check("ING-2 unknown unadmitted dids stay out of the retry set", retryRows == 0,
"%v rows in repo_retries for %d hostile dids, %d of them with an enrolled owner", retryRows, slow+noKnot+enrolledNoKnot, enrolledNoKnot)
if retryRows > 0 {
before := w.ids.lookups.Load()
for i := range retryRounds {
w.skew.Add(int64(2 * time.Hour))
r.waitFor(time.Minute, func() bool {
return w.countRows("select coalesce(min(attempts), 0) from repo_retries") >= float64(i+2)
})
}
r.metric("retry_lookups", float64(w.ids.lookups.Load()-before))
r.note("lookups count cold directory fetches now, and a did doc with no knot service is cached for 24h, so retry_lookups near 0 means the retries cost cpu and log lines but no plc traffic. rows x attempts is the retry work")
}
})
registerTorture("ingest-gap-catchup", "a jetstream gap with many repos, live events during it, then a second gap", func(r *tortureRun) {
n := r.load("repos", 60)
knots := int(r.param("knots", 3))
changedFrac := float64(0.2)
delay := 25 * time.Millisecond
probeEvery := 500 * time.Millisecond
probeBudget := r.duration("probe_budget", 5*time.Second)
secondGap := r.param("second_gap", 1) > 0
w := newTortureWorld(r)
w.addKnots(knots)
repos := w.addRepos(n, func(i int) (*tortureKnot, syntax.DID) {
return w.knots[i%knots], syntax.DID(fmt.Sprintf("did:plc:gapowner%04d", i))
}, 0, nil)
w.boot()
w.seed(repos, 3*time.Minute)
r.gauge("catchup_pending", func() float64 {
return w.fromSpindle(func(s *Spindle) float64 {
pending, err := s.db.PendingCatchUp()
if err != nil {
return math.NaN()
}
return float64(len(pending))
})
})
// the cursor has to pass everything, or the next boot replays it
at := w.js.filler()
w.consumed(at, 30*time.Second)
w.stop()
changed := make([]syntax.DID, int(float64(n)*changedFrac))
for i := range changed {
changed[i] = syntax.DID(fmt.Sprintf("did:plc:gapnew%04d", i))
ingPutCollaborator(repos[i], changed[i])
}
for _, k := range w.knots {
k.delay.Store(delay.Nanoseconds())
}
w.js.forget()
w.js.filler()
erasedBefore := len(w.hyd.erasedDids())
reqBefore := ingXrpc(w.knots)
readsBefore := ingManifestReads(w.knots)
w.boot()
probeRepos := max(1, min(5, n/4))
var applied grantProbes
var probing sync.WaitGroup
var lastUS atomic.Int64
var paused atomic.Bool
stopProbes := make(chan struct{})
probing.Go(func() {
for i := 0; ; i++ {
select {
case <-stopProbes:
return
case <-time.After(probeEvery):
}
if paused.Load() {
continue
}
repo := repos[n-1-i%probeRepos]
lastUS.Store(w.probeGrant(repo, syntax.DID(fmt.Sprintf("did:plc:gapprobe%04d", i)), 2*time.Minute, &applied))
}
})
started := r.waitFor(time.Minute, func() bool { return len(w.hyd.erasedDids()) > erasedBefore })
gaps := 1
if secondGap && started {
// wait for the probes in flight first, forget drops history
paused.Store(true)
w.consumed(lastUS.Load(), 30*time.Second)
w.js.forget()
filler := w.js.filler()
w.js.dropConnections()
w.consumed(filler, time.Minute)
paused.Store(false)
gaps = 2
}
budget := time.Duration(n)*time.Second + time.Minute
finished := r.waitFor(budget, func() bool {
s := w.spindle()
pending, err := s.db.PendingCatchUp()
if err != nil || len(pending) > 0 {
return false
}
for i, who := range changed {
if !w.isCollab(repos[i], who) {
return false
}
}
return true
})
close(stopProbes)
probing.Wait()
applied.wg.Wait()
erased := len(w.hyd.erasedDids()) - erasedBefore
reqs := ingXrpc(w.knots) - reqBefore
refreshes := float64(ingManifestReads(w.knots)-readsBefore) / float64(n)
dids := 2*n + 1
r.latency("live_apply", &applied.latencies)
r.metric("live_missed", float64(applied.missed.Load()))
r.metric("erased", float64(erased))
r.metric("erased_per_did", float64(erased)/float64(dids))
r.metric("knot_requests_per_repo", float64(reqs)/float64(n))
r.metric("gaps", float64(gaps))
r.metric("refreshes_per_repo", refreshes)
r.note("every refresh reads the manifest once, so manifest reads per repo is refreshes per repo. the fake hydrant replays every record of the owner and the repo did, and replayed records only mark the repo, so a repo's job coalesces them. the exact number still depends on the fake")
r.check("ING-3 catch-up brings back what changed during the gap", finished, "finished %v, waited up to %v for no pending dids and every changed repo synced", finished, budget)
r.check("ING-3 live events apply during the catch-up", applied.missed.Load() == 0 && applied.quantile(1) <= probeBudget,
"%d probes applied, %d missed, slowest %v, want within %v", applied.len(), applied.missed.Load(), applied.quantile(1), probeBudget)
r.check("ING-3 a gap erases each did at most once", float64(erased) <= float64(dids*gaps)*1.05,
"%d erases for %d dids over %d gaps", erased, dids, gaps)
// probes refresh the repo too and shouldn't count here
catchUpRefreshes := refreshes - float64(applied.len())/float64(n)
r.metric("catchup_refreshes_per_repo", catchUpRefreshes)
// 0.2 is slack for a record that lands during the repo job
r.check("ING-3 each served repo is refreshed at most 1.2 times per gap", catchUpRefreshes <= 1.2*float64(gaps),
"%.2f refreshes per repo over %d gaps without the %d probe events", catchUpRefreshes, gaps, applied.len())
})
registerTorture("ingest-collab-churn", "a burst of collaborator grants and revokes, then a list past the page cap", func(r *tortureRun) {
events := r.load("events", 200)
pool := 40
base := min(int(r.param("base", 20)), pool)
oversize := int(r.param("oversize", (ingListPageCap+1)*ingListPageSize))
delay := 5 * time.Millisecond
w := newTortureWorld(r)
k := w.addKnot()
repo := w.addRepo(k, syntax.DID("did:plc:churnowner"), 0, nil)
subjects := make([]syntax.DID, pool)
live := map[syntax.DID]bool{}
for i := range subjects {
subjects[i] = syntax.DID(fmt.Sprintf("did:plc:churn%04d", i))
if i < base {
ingPutCollaborator(repo, subjects[i])
live[subjects[i]] = true
}
}
w.boot()
w.seed([]*tortureRepo{repo}, 30*time.Second)
casbinMatches := func() bool {
for _, s := range subjects {
if w.isCollab(repo, s) != live[s] {
return false
}
}
return true
}
rowsMatch := func() bool {
rows, err := w.s.db.ListCollaboratorsByRepoDid(repo.did)
if err != nil || len(rows) != len(live) {
return false
}
for _, c := range rows {
if !live[c.Subject] {
return false
}
}
return true
}
if !r.waitFor(30*time.Second, casbinMatches) {
r.t.Fatal("the base collaborators never synced")
}
k.delay.Store(delay.Nanoseconds())
rng := rand.New(rand.NewPCG(7, 11))
reqBefore := k.xrpc.Load()
var last int64
for range events {
s := subjects[rng.IntN(pool)]
if live[s] {
delete(live, s)
last = w.revokeCollaborator(repo, s)
} else {
live[s] = true
last = w.grantCollaborator(repo, s)
}
}
consumed := w.consumed(last, 3*time.Minute)
settled := r.waitFor(30*time.Second, func() bool { return casbinMatches() && rowsMatch() })
reqs := k.xrpc.Load() - reqBefore
r.metric("knot_requests", float64(reqs))
r.metric("knot_requests_per_event", float64(reqs)/float64(events))
r.check("ING-11 final casbin and db state equal the knot state", settled && consumed, "casbin match %v, rows match %v after the burst", casbinMatches(), rowsMatch())
r.check("ING-11 a burst of events coalesces into fewer refreshes", float64(reqs)/ingRequestsPerRefresh/float64(events) <= 0.5,
"%.2f refreshes per event (%d knot requests for %d events), want at most 0.5", float64(reqs)/ingRequestsPerRefresh/float64(events), reqs, events)
if oversize == 0 {
return
}
// past the knot client's page limit the list errors, the revoke still has to land
victim := syntax.DID("did:plc:churnvictim")
w.grantCollaborator(repo, victim)
if !r.waitFor(30*time.Second, func() bool { return w.isCollab(repo, victim) }) {
r.t.Fatal("the victim never got access")
}
for i := range oversize {
ingPutCollaborator(repo, syntax.DID(fmt.Sprintf("did:plc:bulk%05d", i)))
}
oversizeBefore := k.xrpc.Load()
w.grantCollaborator(repo, syntax.DID("did:plc:churntrigger"))
us := w.revokeCollaborator(repo, victim)
w.consumed(us, time.Minute)
revoked := r.waitFor(10*time.Second, func() bool { return !w.isCollab(repo, victim) })
perEvent := float64(k.xrpc.Load()-oversizeBefore) / 2
// a failed full sync costs one list per page up to the limit, plus a few more
r.metric("oversize_requests_per_event", perEvent)
r.metric("oversize_retry_rows", w.rows("repo_retries"))
r.check("ING-6 a revoke applies even when the list is past the page cap", revoked, "revoked collaborator has access %v with %d collaborators on the knot, want none", !revoked, oversize)
r.check("ING-6 an oversized repo costs at most one failed sync per event", perEvent <= ingListPageCap+8, "%.0f knot requests per event, want at most %d", perEvent, ingListPageCap+8)
})
registerTorture("ingest-retry-herd", "every repo's refresh fails while the knots are down, then they come back", func(r *tortureRun) {
n := r.load("repos", 100)
knots := int(r.param("knots", 4))
delay := 20 * time.Millisecond
outageRounds := 3
burstCap := r.param("burst_cap_per_s", 200)
r.note("outage_rounds skews the retry clock 2h per 10s retry tick, so 3 rounds is a 6h outage in 30s. set outage_rounds=13 to pass the 12 attempts and see ING-5 give up")
w := newTortureWorld(r)
w.addKnots(knots)
repos := w.addRepos(n, func(i int) (*tortureKnot, syntax.DID) {
return w.knots[i%knots], syntax.DID(fmt.Sprintf("did:plc:herdowner%04d", i))
}, 0, nil)
w.boot()
w.seed(repos, 3*time.Minute)
for _, k := range w.knots {
k.delay.Store(delay.Nanoseconds())
k.down.Store(true)
}
subjects := make([]syntax.DID, n)
var last int64
for i, repo := range repos {
subjects[i] = syntax.DID(fmt.Sprintf("did:plc:herd%04d", i))
last = w.grantCollaborator(repo, subjects[i])
}
w.consumed(last, 3*time.Minute)
failed := r.waitFor(time.Minute, func() bool {
return w.rows("repo_retries") >= float64(n) &&
w.countRows("select coalesce(min(attempts), 0) from repo_retries") >= 1
})
sameSecond, distinct := ingDueSpread(w)
r.metric("retry_rows_after_events", w.rows("repo_retries"))
r.metric("due_same_second_max", float64(sameSecond))
r.check("ING-2 retries get jitter instead of all falling due together", failed && sameSecond*4 <= n,
"%d of %d retries share one due second, %d distinct seconds", sameSecond, n, distinct)
for i := range outageRounds {
w.skew.Add(int64(2 * time.Hour))
r.waitFor(time.Minute, func() bool {
return w.rows("repo_retries") == 0 ||
w.countRows("select coalesce(min(attempts), 0) from repo_retries") >= float64(i+2)
})
}
r.metric("retry_rows_before_recovery", w.rows("repo_retries"))
var bursts []*ingBurst
for _, k := range w.knots {
k.down.Store(false)
bursts = append(bursts, ingWatchBurst(k.xrpc.Load))
}
reqBefore := ingXrpc(w.knots)
t0 := time.Now()
w.skew.Add(int64(2 * time.Hour))
drained := r.waitFor(3*time.Minute, func() bool { return w.rows("repo_retries") == 0 })
var peak int64
for _, b := range bursts {
peak = max(peak, b.finish())
}
reqs := ingXrpc(w.knots) - reqBefore
r.metric("peak_requests_per_knot_per_s", float64(peak))
r.metric("requests_per_repo", float64(reqs)/float64(n))
r.metric("drain_repos_per_s", float64(n)/time.Since(t0).Seconds())
synced := 0
r.waitFor(time.Minute, func() bool {
synced = 0
for i, repo := range repos {
if w.isCollab(repo, subjects[i]) {
synced++
}
}
return synced == n
})
r.metric("repos_synced", float64(synced))
r.check("ING-2 the retry table drains after the knots come back", drained, "%v rows still in repo_retries", w.rows("repo_retries"))
r.check("ING-5 membership converges after a long outage", synced == n, "%d of %d repos picked up the collaborator granted during the outage", synced, n)
r.check("ING-2 the retry burst per knot stays bounded", float64(peak) <= burstCap, "%d requests in one second against one knot, want at most %.0f", peak, burstCap)
})
registerTorture("ingest-manifest-flap", "the owner flips the manifest between this spindle and another one", func(r *tortureRun) {
flips := r.load("flips", 40)
gitDelay := r.duration("git_delay", 2*time.Second)
commits := 2
goroutineSlack := r.param("goroutine_slack", 10)
w := newTortureWorld(r)
k := w.addKnot()
repo := w.addRepo(k, syntax.DID("did:plc:flapowner"), commits, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")})
bob := syntax.DID("did:plc:flapbob")
ingPutCollaborator(repo, bob)
w.boot()
w.seed([]*tortureRepo{repo}, 30*time.Second)
if !r.waitFor(30*time.Second, func() bool { return w.isCollab(repo, bob) }) {
r.t.Fatal("the collaborator never synced")
}
// let the setup's fetches finish before measuring one load
seen, stableSince := int64(-1), time.Now()
r.waitFor(30*time.Second, func() bool {
if now := k.fetches.Load(); now != seen {
seen, stableSince = now, time.Now()
}
return time.Since(stableSince) >= 500*time.Millisecond
})
fetchesPerLoad := max(1, k.fetches.Load())
baseGoroutines := r.gaugeValue("goroutines")
peak := baseGoroutines
var forgetting latencies
dumped := false
fetchesBefore := k.fetches.Load()
reqBefore := k.xrpc.Load()
k.gitDelay.Store(gitDelay.Nanoseconds())
start := time.Now()
stuck := 0
var last int64
for i := range flips {
away := i%2 == 0
spindle := w.spindleDid()
if away {
spindle = "did:web:other.example.com"
}
k.put(repo.did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(repo.owner, repo.name, spindle))
flipStart := time.Now()
last = w.emitManifest(repo)
if !r.waitFor(30*time.Second, func() bool { return w.served(repo) != away }) {
stuck++
} else if away {
forgetting.since(flipStart)
}
peak = max(peak, r.gaugeValue("goroutines"))
// dump now, the goroutines are gone by the time the check runs
if !dumped && peak-baseGoroutines > goroutineSlack {
dumped = true
r.dumpGoroutines()
}
}
w.consumed(last, time.Minute)
r.metric("flap_s", time.Since(start).Seconds())
r.metric("stuck_flips", float64(stuck))
r.metric("goroutines_peak_over_base", peak-baseGoroutines)
r.latency("forget_apply", &forgetting)
r.metric("knot_requests_per_flip", float64(k.xrpc.Load()-reqBefore)/float64(flips))
recovered := r.waitFor(30*time.Second, func() bool { return w.served(repo) && w.isCollab(repo, bob) })
r.check("ING-10 the repo ends up where the last manifest says", recovered && stuck == 0, "served with its collaborator %v, %d flips never took effect", recovered, stuck)
r.check("ING-10 flaps don't pile up goroutines behind clones", peak-baseGoroutines <= goroutineSlack, "goroutines peaked %.0f over the start with %.0f of slack", peak-baseGoroutines, goroutineSlack)
loads := float64(k.fetches.Load()-fetchesBefore) / float64(fetchesPerLoad)
// forgetting drops the clone, and every return to this spindle loads it again
r.check("ING-10 a paced flap loads schedules once per return", loads <= float64(flips/2)+1,
"%.1f schedule loads (%d git fetches at %d per load) for %d returns to this spindle", loads, k.fetches.Load()-fetchesBefore, fetchesPerLoad, flips/2)
r.check("ING-10 forgetting a repo isn't held up by a clone in flight", forgetting.quantile(1) <= gitDelay/2,
"slowest away flip took %v, a clone fetch is %v, want under half of it", forgetting.quantile(1), gitDelay)
r.settles("goroutines", baseGoroutines+goroutineSlack, time.Minute)
burstFetches := k.fetches.Load()
var burstLast int64
for i := range flips {
spindle := w.spindleDid()
if i%2 == 0 {
spindle = "did:web:other.example.com"
}
k.put(repo.did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(repo.owner, repo.name, spindle))
burstLast = w.emitManifest(repo)
}
w.consumed(burstLast, time.Minute)
burstSettled := r.waitFor(time.Minute, func() bool {
return w.served(repo) && w.rows("repo_retries") == 0
})
burstLoads := float64(k.fetches.Load()-burstFetches) / float64(fetchesPerLoad)
r.metric("burst_schedule_loads", burstLoads)
r.check("ING-10 an unpaced burst of flips costs few schedule loads", burstSettled && burstLoads <= max(2, float64(flips)/8),
"%.1f schedule loads for %d unpaced flips, settled on the last manifest %v, want at most %.0f", burstLoads, flips, burstSettled, max(2, float64(flips)/8))
})
// a failed mark can't count as done because it's the only way into the repo job
registerTorture("ingest-mark-fails", "the db is locked past its busy timeout while a manifest event lands", func(r *tortureRun) {
// past the 5s sqlite busy timeout in sqlite/sqlite.go
lockHold := 8 * time.Second
w := newTortureWorld(r)
k := w.addKnot()
repo := w.addRepo(k, syntax.DID("did:plc:markowner"), 1, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")})
w.boot()
w.seed([]*tortureRepo{repo}, 30*time.Second)
ctx := context.Background()
conn, err := w.spindle().db.Conn(ctx)
if err != nil {
r.t.Fatal(err)
}
if _, err := conn.ExecContext(ctx, "begin immediate"); err != nil {
r.t.Fatal(err)
}
k.put(repo.did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(repo.owner, repo.name, "did:web:other.example.com"))
last := w.emitManifest(repo)
time.Sleep(lockHold)
_, _ = conn.ExecContext(ctx, "rollback")
conn.Close()
w.consumed(last, time.Minute)
forgotten := r.waitFor(time.Minute, func() bool { return !w.served(repo) })
r.check("ING-12 a manifest event whose mark failed is still applied", forgotten,
"repo still served a minute after the db unlocked, the event that moved it away was dropped")
})
registerTorture("ingest-hung-repo", "one repo's knot hangs while the reader keeps going", func(r *tortureRun) {
others := r.load("other_repos", 8)
events := int(r.param("other_events", 4))
hang := 6 * time.Second
budget := r.duration("other_budget", 2*time.Second)
r.note("hung_delay is 6s where production waits up to 30s per request, a repo's job makes several of them")
w := newTortureWorld(r)
slow, fast := w.addKnot(), w.addKnot()
stuck := w.addRepo(slow, "did:plc:stuckowner", 0, nil)
repos := w.addRepos(others, func(i int) (*tortureKnot, syntax.DID) {
return fast, syntax.DID(fmt.Sprintf("did:plc:fastowner%04d", i))
}, 0, nil)
w.boot()
w.seed(append([]*tortureRepo{stuck}, repos...), 2*time.Minute)
slow.delay.Store(hang.Nanoseconds())
first, second := syntax.DID("did:plc:stuckfirst"), syntax.DID("did:plc:stucksecond")
w.grantCollaborator(stuck, first)
w.revokeCollaborator(stuck, first)
lastStuck := w.grantCollaborator(stuck, second)
cursorFree := w.consumed(lastStuck, 5*time.Second)
var applied grantProbes
for e := range events {
for i, repo := range repos {
w.probeGrant(repo, syntax.DID(fmt.Sprintf("did:plc:fastprobe%d-%04d", e, i)), time.Minute, &applied)
}
}
applied.wg.Wait()
r.latency("other_apply", &applied.latencies)
r.metric("other_missed", float64(applied.missed.Load()))
stillHung := !w.isCollab(stuck, second)
slow.delay.Store(0)
converged := r.waitFor(time.Minute+hang, func() bool { return w.isCollab(stuck, second) && !w.isCollab(stuck, first) })
r.check("INGEST the cursor passes events of a hung repo", cursorFree, "cursor passed the hung repo's events %v, waited up to 5s", cursorFree)
r.check("INGEST other repos apply while one repo hangs", stillHung && applied.missed.Load() == 0 && applied.quantile(1) <= budget,
"%d of %d probes applied, slowest %v, want within %v, hung repo still unsynced %v", applied.len(), others*events, applied.quantile(1), budget, stillHung)
r.check("INGEST the hung repo ends on the knot's last word", converged, "granted then revoked first, then granted second: second collaborator %v, first revoked %v", w.isCollab(stuck, second), !w.isCollab(stuck, first))
})
}