diff --git a/spindle/TORTURE.md b/spindle/TORTURE.md index 41f6dd541..b5b79047b 100644 --- a/spindle/TORTURE.md +++ b/spindle/TORTURE.md @@ -1,41 +1,39 @@ # spindle torture -load, abuse and recovery scenarios for spindle's control plane: the job queue, -jetstream ingestion, hydrant catch-up, repo membership, retries, notify -handling, webhooks, logs, admission and quotas. it is not about how many -workflows an engine can run, every scenario uses a fake engine that only -sleeps and prints, so no number here says anything about vms. +these are load, abuse and recovery scenarios for spindle's control plane: the job queue, +jetstream ingest, hydrant catch-up, repo membership, retries, notifies, webhooks, logs, +admission and quotas. they don't measure how much an engine can run. every scenario uses a +fake engine that only sleeps and prints, so none of these numbers tell you anything about vms. -linux only: rss and fds are read from `/proc`, and a gauge that can't be read -fails the scenario instead of reading as zero. +it only runs on linux, since rss and fds come from `/proc`. a gauge it can't read fails the +scenario instead of quietly reading as zero. ## how it works -the scenarios are go tests in `spindle/torture_*_test.go` behind the -`torture` build tag. each one boots a real spindle in process (`New`, `Start`, -the real router, xrpc handlers, jetstream client, db and job workers) against: +the scenarios are go tests in `spindle/torture_*_test.go` behind the `torture` build tag. +each one boots a real spindle in process (`New`, `Start`, the real router and xrpc handlers, +the jetstream client, the db and the job workers) and points it at fakes: -- fake knots (`tortureKnot`) that serve records and git over http-backend, - sign service auth with real keys as `did:web` on loopback, and also play - every owner's pds. they have knobs for latency, outages and git failures -- the fake jetstream and fake hydrant from `fakes_test.go` / `hydrant_test.go` -- one in-memory identity directory for every did behind the same cache the - real resolver has, so a lookup it counts is a cold fetch. a scenario that - changes a did doc after spindle looked it up has to set `uncachedIDs` -- a knot2-faithful outbox (`knotOutbox`): one delivery in flight per spindle, - oldest first, backoff 1s doubling to 60s without jitter, 4xx drops +- fake knots (`tortureKnot`) serve records and git over http-backend, sign service auth with + real keys as `did:web` on loopback, and double as every owner's pds. you can make them slow, + take them down or break their git +- the fake jetstream and hydrant from `fakes_test.go` and `hydrant_test.go` +- one in-memory identity directory for every did, behind the same cache the real resolver + uses, so a lookup it counts really was a cold fetch +- `knotOutbox`, which delivers notifies the way knot2 does: one in flight per spindle, oldest + first, backoff from 1s doubling up to 60s with no jitter, and any 4xx gets dropped -spindle runs with production config defaults except dev mode, which loopback -knots need and which also skips netguard. +spindle runs with the production defaults except for dev mode. loopback knots need it, and it +also turns netguard off. -running in process is what lets a scenario read the queue depth, swap in the -fake engine and hydrant, and restart spindle on the same db. the fakes share -the process, so cpu and rss numbers include them. +running in process is how a scenario can read the queue depth, swap in the fakes and restart +spindle on the same db. the catch is that the fakes share the process too, so cpu and rss +include them. ## running -the build needs cgo, sqlite and the hydrant ffi like the normal spindle tests, -so on a machine without them go through the repo's nix shell first. +it needs cgo, sqlite and the hydrant ffi like the normal spindle tests, so go through the +repo's nix shell if your machine doesn't have those. ```sh go test -tags 'torture libsqlite3' -c -o /tmp/torture.test ./spindle/ @@ -45,8 +43,8 @@ cd spindle -torture.out /tmp/torture.jsonl ``` -one scenario takes between a few seconds and about ten minutes, all of them -together about 40 minutes, so give a full run `-test.timeout 90m`. +one scenario takes anywhere from a few seconds to about ten minutes, and all of them take +about 40, so give a full run `-test.timeout 90m`. on a shared box, cap it: @@ -56,20 +54,20 @@ systemd-run --user --scope -p MemoryMax=4G -p CPUQuota=800% -- /tmp/torture.test flags: -- `-torture.list` prints every scenario with what it does and runs none -- `-test.run 'TestTorture/'` picks scenarios, `TestTorture` runs all +- `-torture.list` prints every scenario and what it does, without running any +- `-test.run 'TestTorture/'` picks scenarios, plain `TestTorture` runs them all - `-torture.out ` appends one json line per scenario - `-torture.scale 2` doubles every load param (repos, knots, events) -- `-torture.set repos=200,step_ms=500` sets params directly, they're listed in - each result's `params`. only loads and the knobs a check or this doc relies on are - params, everything else is a literal in the scenario -- `-torture.log warn` prints spindle logs from that level, the default is - `error`. warnings and errors are counted per message in `errors` either way +- `-torture.set repos=200,step_ms=500` sets params directly. each result lists its `params`. + only the loads and the knobs a check or this doc depends on are params, the rest are + literals in the scenario +- `-torture.log warn` prints spindle's logs from that level up (the default is `error`). + warnings and errors get counted per message in `errors` either way ## sizing the pools -`pool-sizing` is a measurement, not a pass/fail run. it takes one size per run so it can sit -under `MemoryMax`: `pool` is the repo job pool, `notify` the notify slot count. +`pool-sizing` measures, it doesn't pass or fail on speed. it runs one size at a time so it fits +under `MemoryMax`. `pool` is the repo job pool and `notify` is the number of notify slots. ```sh for pool in 16 64 256 1024; do @@ -77,34 +75,46 @@ for pool in 16 64 256 1024; do done ``` -per size it reports idle cost (`idle_rss_mb`, `idle_goroutines`, `idle_fds`) and the cost of a -catch-up over ~5000 repos (60% on one big knot, 30% over four small ones, 10% on a slow one) with -live events and pushes on top: the sampled `rss_mb`, `goroutines`, `fds` and `knots_inflight` -as `_max`, `catchup_s` (boot to the catch-up done), `apply_all_s` (every changed repo applied), `refreshes_per_s`, `notify_accepted_per_s`, `probe_write` latency and -errors. `drain_budget` (default 30m) is how long the catch-up may take, a run that runs out reports no time and fails 'catch-up finished within the budget', so give it a `-test.timeout` past that. its other check is that the slow knot never holds more than its fair share of the pool -while others have repos waiting, measured on the repo pool's own bookkeeping (jobs running per knot against knots with queued jobs). +for each size you get what it costs idle (`idle_rss_mb`, `idle_goroutines`, `idle_fds`) and what +it costs to catch up on about 5000 repos (60% on one big knot, 30% over four small ones and 10% +on a slow one) with live events and pushes landing on top: -the defaults (256 repo job workers, 64 notify slots) come from this: behind a 5000-repo catch-up the -live event p99 was ~119s at 16 workers and 2.5s at 256 (~685MB peak rss), 1024 only got it to 1.4s for -~990MB, 1775 fds and 2258 goroutines, and 16 vs 64 notify slots made no real difference. +- `rss_mb`, `goroutines`, `fds` and `knots_inflight`, sampled, as `_max` +- `catchup_s`, from boot until the catch-up is done +- `apply_all_s`, until every changed repo got applied +- `refreshes_per_s`, `notify_accepted_per_s`, `probe_write` latency and errors + +`drain_budget` (30m by default) is how long the catch-up gets. a run that runs out of it reports +no time and fails "catch-up finished within the budget", so set `-test.timeout` past it. the +other check is that the slow knot never holds more than its fair share of the pool while other +knots have repos waiting. that's read off the pool's own bookkeeping, jobs running per knot +against knots with jobs queued. + +the defaults (256 repo job workers, 64 notify slots) come from this. behind a 5000 repo +catch-up, the p99 for a live event was about 119s with 16 workers and 2.5s with 256, at about +685MB peak rss. 1024 workers only got it to 1.4s and cost about 990MB, 1775 fds and 2258 +goroutines. 16 or 64 notify slots made no real difference. ## results -each json line has `params`, `metrics` (latency percentiles as `_p50_ms` .. -`_max_ms`, every sampled gauge as `_start` / `_max` / `_end` / -`_slope_per_min`), `errors` (http answers and log messages by kind) and -`invariants`, each one a thing spindle should do. a broken invariant or a -failed setup (`t.Fatal`) fails the test and sets `pass: false`. everything -should pass on the current code, so a red scenario is a regression or a new -finding, not a known one. +each json line has: + +- `params` +- `metrics`: latency percentiles as `_p50_ms` up to `_max_ms`, and every sampled gauge as + `_start`, `_max`, `_end` and `_slope_per_min` +- `errors`: http answers and log messages, counted by kind +- `invariants`: things spindle should always do + +a broken invariant or a setup that fails (`t.Fatal`) fails the test and sets `pass: false`. +everything should pass on the current code, so a red scenario means a regression or something +new, never a known failure. -sampled gauges: goroutines, heap_mb, rss_mb, fds, queued_jobs, pipelines, -repo_retries, db_mb, engine_running, plus whatever a scenario adds. +the sampled gauges are goroutines, heap_mb, rss_mb, fds, queued_jobs, pipelines, repo_retries, +db_mb and engine_running, plus whatever a scenario adds. -invariant names often start with the id of the finding they came from -(A1, N2, ING-1, Q1...). those ids are from the audit the scenarios were written -for and aren't defined in the repo, the name and the detail say what is wanted. +some invariant names start with a short id like A1, N2, ING-1 or Q1. those ids aren't defined +anywhere in the repo, so go by the name and its detail. -pass or fail checks that need no load live in ordinary tests next to the code -they pin, so `go test ./spindle/...` catches them without the tag. the tag is -for what needs scale or time. +checks that are just pass or fail and don't need load live in ordinary tests next to the code +they cover, so `go test ./spindle/...` catches them without the tag. the tag is only for what +needs scale or time. diff --git a/spindle/torture_admission_test.go b/spindle/torture_admission_test.go index 43f82f229..bc3335977 100644 --- a/spindle/torture_admission_test.go +++ b/spindle/torture_admission_test.go @@ -9,8 +9,6 @@ import ( "io" "net" "net/http" - "os" - "path/filepath" "slices" "strings" "sync" @@ -18,7 +16,6 @@ import ( "time" "github.com/bluesky-social/indigo/atproto/atcrypto" - "github.com/bluesky-social/indigo/atproto/auth" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" @@ -33,20 +30,9 @@ type admUser struct { priv atcrypto.PrivateKey } -// call after addRepo, which overwrites the owner's directory entry +// has to run after addRepo, which overwrites the owner's entry func (w *tortureWorld) admAddUser(did syntax.DID, pds *tortureKnot) *admUser { - priv, err := atcrypto.GeneratePrivateKeyP256() - if err != nil { - w.r.t.Fatal(err) - } - pub, err := priv.PublicKey() - if err != nil { - w.r.t.Fatal(err) - } - id := identity.Identity{ - DID: did, - Keys: map[string]identity.VerificationMethod{"atproto": {Type: "Multikey", PublicKeyMultibase: pub.Multibase()}}, - } + priv, id := w.newKeyedIdentity(did) if pds != nil { id.Services = map[string]identity.ServiceEndpoint{ "atproto_pds": {Type: "AtprotoPersonalDataServer", URL: pds.url()}, @@ -56,45 +42,9 @@ func (w *tortureWorld) admAddUser(did syntax.DID, pds *tortureKnot) *admUser { return &admUser{did: did, priv: priv} } -func (w *tortureWorld) admToken(u *admUser, nsid string) string { - lxm := syntax.NSID(nsid) - token, err := auth.SignServiceAuth(u.did, w.spindleDid(), time.Minute, &lxm, u.priv) - if err != nil { - panic(err) - } - return token -} - -// no length set, so a body goes out chunked -func (w *tortureWorld) admCall(ctx context.Context, client *http.Client, method, nsid, query, token string, body io.Reader, length int64) notifyReply { - start := time.Now() - target := w.api.URL + "/xrpc/" + nsid - if query != "" { - target += "?" + query - } - req, err := http.NewRequestWithContext(ctx, method, target, body) - if err != nil { - panic(err) - } - if length > 0 { - req.ContentLength = length - } - if body != nil { - req.Header.Set("Content-Type", "application/json") - } - if token != "" { - req.Header.Set("Authorization", "Bearer "+token) - } - resp, err := client.Do(req) - if err != nil { - return notifyReply{status: -1, body: err.Error(), elapsed: time.Since(start)} - } - defer resp.Body.Close() - raw, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<20)) - return notifyReply{status: resp.StatusCode, body: string(raw), elapsed: time.Since(start)} -} +func (w *tortureWorld) admToken(u *admUser, nsid string) string { return w.sign(u.did, u.priv, nsid) } -// same constructor Start uses, so the timeouts are the real ones +// 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") @@ -107,14 +57,13 @@ func (w *tortureWorld) admServe() string { } func admFdKinds() string { - entries, err := os.ReadDir("/proc/self/fd") + targets, err := fdTargets() if err != nil { return "unreadable" } kinds := map[string]int{} - for _, e := range entries { - target, err := os.Readlink(filepath.Join("/proc/self/fd", e.Name())) - if err != nil { + for _, target := range targets { + if target == "" { continue } kind, _, found := strings.Cut(target, ":") @@ -126,7 +75,7 @@ func admFdKinds() string { return fmt.Sprint(kinds) } -// the manager is built before the db, so the real store gets set after boot +// built before the db exists, the real store gets swapped in after boot type admQuotaStore struct { qs atomic.Pointer[db.QuotaStore] reserves atomic.Int64 @@ -137,7 +86,6 @@ func (s *admQuotaStore) Reserve(ctx context.Context, req quota.ReserveRequest) ( return s.qs.Load().Reserve(ctx, req) } -// the polling read, so it doesn't count as a reserve func (s *admQuotaStore) CheckDenied(ctx context.Context, req quota.ReserveRequest) (quota.Reservation, bool, error) { return s.qs.Load().CheckDenied(ctx, req) } @@ -170,7 +118,7 @@ func (o *admQuotaObserver) SetWaitDepth(kind string, depth int64) { func (o *admQuotaObserver) RecordWait(kind, resource string, duration time.Duration) {} func init() { - // the hold is twice the read timeout, so these only end if the server cuts them + // 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) @@ -181,7 +129,7 @@ func init() { user := w.admAddUser("did:plc:slowuser", nil) w.boot() addr := w.admServe() - // the first requests open files and pools, so the baseline waits those out + // 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) @@ -298,7 +246,7 @@ func init() { r.settles("fds", baseFds+30, 30*time.Second) }) - // the 50ms quota tick writes for every waiting owner, that's the cost under test + // 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)) @@ -315,24 +263,19 @@ func init() { 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.countRows("select count(*) from quota_reservations") }) + 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() - if !w.seed(append(slices.Clone(repos), fresh...), 3*time.Minute) { - r.t.Fatalf("only %d of %d repos served", w.servedCount(repos), len(repos)) - } + 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() - if _, err := w.s.db.Exec(`create table if not exists torture_probe (n integer)`); err != nil { - r.t.Fatal(err) - } - // measured with nothing else running, so the loaded numbers have a baseline + w.probeTable() var quiet latencies for i := range 50 { start := time.Now() @@ -343,30 +286,9 @@ func init() { } var probe latencies var probeErrors atomic.Int64 - probeStop := make(chan struct{}) - var probing sync.WaitGroup - probing.Go(func() { - tick := time.NewTicker(100 * time.Millisecond) - defer tick.Stop() - for i := 0; ; i++ { - select { - case <-probeStop: - return - case <-tick.C: - } - start := time.Now() - if _, err := w.s.db.Exec(`insert into torture_probe (n) values (?)`, i); err != nil { - probeErrors.Add(1) - continue - } - probe.since(start) - } - }) + stopProbe := w.probeWrites(100*time.Millisecond, &probe, &probeErrors) - var ds []pushDelivery - for _, repo := range repos { - ds = append(ds, pushRange(repo, 0, queued)...) - } + ds := pushRanges(repos, 0, queued) var st deliveryStats w.flood(ctx, ds, 32, 20, &st) st.report(r, "notify_") @@ -375,7 +297,6 @@ func init() { 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) - // nothing gets released in this window, so any new reserve is polling r0 := store.reserves.Load() t0 := time.Now() time.Sleep(window / 2) @@ -388,12 +309,9 @@ func init() { } secs := time.Since(t0).Seconds() reserves := store.reserves.Load() - r0 - close(probeStop) - probing.Wait() + stopProbe() r.metric("reserve_per_s", float64(reserves)/secs) - if reserves > 0 { - } r.metric("fresh_owner_start_ms", float64(freshTook.Milliseconds())) r.metric("probe_errors", float64(probeErrors.Load())) r.latency("probe_write", &probe) @@ -404,21 +322,17 @@ func init() { 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) - drained := w.drained(5 * time.Minute) - r.check("queue drains", drained, "%v jobs queued and %d workflows running, want 0 and 0", w.countRows("select count(*) from jobs"), w.eng.running.Load()) - leaked := r.waitFor(10*time.Second, func() bool { return w.countRows("select count(*) from quota_reservations") == 0 }) - r.check("no reservations leaked after the run (A7)", leaked, "%v quota_reservations rows left, want 0", w.countRows("select count(*) from quota_reservations")) + 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)] - var ds2 []pushDelivery - for _, repo := range half { - ds2 = append(ds2, pushRange(repo, queued, 2*queued)...) - } + 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.countRows("select count(*) from quota_reservations") > 0 && obs.depth.Load() > 0 + return w.rows("quota_reservations") > 0 && obs.depth.Load() > 0 }) r.note("stopped with reservations held and waiters queued: %v", holding) w.stop() @@ -435,7 +349,7 @@ func init() { 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 clears these, server.go's Run calls it but Start doesn't + // 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) @@ -463,7 +377,7 @@ func init() { 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) }) - // resumeWipes runs before the job workers start, so the first workflow waits on it + // 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) @@ -472,11 +386,9 @@ func init() { knot := w.addKnot() repo := w.addRepo(knot, "did:plc:banowner", 3, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")}) w.boot() - if !w.seed([]*tortureRepo{repo}, time.Minute) { - r.t.Fatal("repo never got served") - } + w.seed([]*tortureRepo{repo}, time.Minute) ctx := context.Background() - pushes := pushesOf(repo, 3) + 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 }) { @@ -518,8 +430,8 @@ func init() { 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 read from the knot, so only lookupSlots run at once - // the stall stays under the knot client's 30s, so the run doesn't take minutes + // 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 @@ -533,20 +445,18 @@ func init() { 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) - // the web just made this one, spindle hasn't seen the event, so it looks up like an attacker + // 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() - if !w.seed(append(append([]*tortureRepo{}, baseRepos...), hitRepos...), 2*time.Minute) { - r.t.Fatal("the legit repos never got served") - } + 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.admCall(ctx, client, http.MethodGet, org_tangled.SecretListSecretsNSID, "repo="+repo.did.String(), token, nil, 0) + 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 diff --git a/spindle/torture_engine_test.go b/spindle/torture_engine_test.go index 5d0f17007..a6ed5015d 100644 --- a/spindle/torture_engine_test.go +++ b/spindle/torture_engine_test.go @@ -22,21 +22,20 @@ import ( "tangled.org/core/spindle/secrets" ) -// runs no real work, so the numbers only measure spindle's bookkeeping +// no real work runs here, which means the numbers only measure bookkeeping type tortureEngine struct { - slots *engine.SemaphoreSlotter - timeout time.Duration + slots *engine.SemaphoreSlotter running atomic.Int64 started atomic.Int64 - mu sync.Mutex + mu sync.Mutex order []string took map[string]time.Duration } func newTortureEngine() *tortureEngine { - return &tortureEngine{slots: engine.NewSemaphoreSlotter(0), timeout: 5 * time.Minute, took: map[string]time.Duration{}} + return &tortureEngine{slots: engine.NewSemaphoreSlotter(0), took: map[string]time.Duration{}} } type tortureStep struct { @@ -81,7 +80,7 @@ func (e *tortureEngine) SetupWorkflow(_ context.Context, _ models.WorkflowId, wf return nil } -func (e *tortureEngine) WorkflowTimeout() time.Duration { return e.timeout } +func (e *tortureEngine) WorkflowTimeout() time.Duration { return 5 * time.Minute } func (e *tortureEngine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode engine.AcquireMode) (engine.WorkflowSlot, error) { return e.slots.AcquireWorkflowSlot(ctx, wid, wf, mode) @@ -106,6 +105,7 @@ func (e *tortureEngine) RunStep(ctx context.Context, _ models.WorkflowId, w *mod }() for _, command := range strings.Split(w.Steps[idx].Command(), ";") { verb, arg, _ := strings.Cut(strings.TrimSpace(command), " ") + n, numErr := strconv.Atoi(arg) switch verb { case "sleep": d, err := time.ParseDuration(arg) @@ -115,36 +115,11 @@ func (e *tortureEngine) RunStep(ctx context.Context, _ models.WorkflowId, w *mod if err := sleepCtx(ctx, d); err != nil { return err } - case "log": - n, err := strconv.Atoi(arg) - if err != nil { - return err - } - chunk := bytes.Repeat([]byte("x"), 4095) - chunk = append(chunk, '\n') - for n > 0 { - if _, err := out.Write(chunk[:min(n, len(chunk))]); err != nil { - return err - } - n -= len(chunk) - } - case "lines": - n, err := strconv.Atoi(arg) - if err != nil { - return err - } - for i := range n { - if _, err := fmt.Fprintf(out, "line %d\n", i); err != nil { - return err - } - } - case "raw", "dribble", "longline": - n, err := strconv.Atoi(arg) - if err != nil { - return err + case "log", "lines", "raw", "dribble", "longline": + if numErr != nil { + return numErr } - chunk := map[string]int{"raw": 64 << 10, "dribble": 16, "longline": 64 << 10}[verb] - if err := writeChunks(out, n, chunk, verb == "longline"); err != nil { + if err := writeOutput(out, verb, n); err != nil { return err } case "fail": @@ -172,6 +147,30 @@ func (e *tortureEngine) tookFor(command string) time.Duration { return e.took[command] } +func writeOutput(out io.Writer, verb string, n int) error { + switch verb { + case "log": + chunk := bytes.Repeat([]byte("x"), 4095) + chunk = append(chunk, '\n') + for n > 0 { + if _, err := out.Write(chunk[:min(n, len(chunk))]); err != nil { + return err + } + n -= len(chunk) + } + return nil + case "lines": + for i := range n { + if _, err := fmt.Fprintf(out, "line %d\n", i); err != nil { + return err + } + } + return nil + } + chunk := map[string]int{"raw": 64 << 10, "dribble": 16, "longline": 64 << 10}[verb] + return writeChunks(out, n, chunk, verb == "longline") +} + func writeChunks(out io.Writer, total, chunk int, newline bool) error { buf := bytes.Repeat([]byte("x"), chunk) for total > 0 { diff --git a/spindle/torture_ingest_test.go b/spindle/torture_ingest_test.go index 0781aeb41..d6da114bf 100644 --- a/spindle/torture_ingest_test.go +++ b/spindle/torture_ingest_test.go @@ -22,11 +22,11 @@ import ( "tangled.org/core/spindle/knotclient" ) -// copies of spindle/knotclient/knotclient.go values, change these with it +// copied from spindle/knotclient, change both together const ( ingListPageSize = 100 ingListPageCap = 32 - // manifest, owner declaration, collaborator list (ensureRepo in spindle/repos.go) + // a refresh makes 3 requests: manifest, owner declaration, collaborator list ingRequestsPerRefresh = 3 ) @@ -39,7 +39,7 @@ func (w *tortureWorld) isCollab(repo *tortureRepo, who syntax.DID) bool { return err == nil && ok } -// knot gets the record before jetstream hears about it +// 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"}) } @@ -54,22 +54,6 @@ func (w *tortureWorld) revokeCollaborator(repo *tortureRepo, who syntax.DID) int return w.emitCollaborator(repo, who, jmodels.CommitOperationDelete) } -func ingConsumed(w *tortureWorld, us int64, within time.Duration) bool { - return w.r.waitFor(within, func() bool { - s := w.spindle() - return s != nil && s.jc.LastSeenUs() > us - }) -} - -func ingServed(w *tortureWorld, repo *tortureRepo) bool { - s := w.spindle() - if s == nil { - return false - } - rows, err := s.db.ReposByDid(repo.did) - return err == nil && len(rows) > 0 -} - func ingXrpc(knots []*tortureKnot) int64 { var n int64 for _, k := range knots { @@ -86,21 +70,25 @@ func ingManifestReads(knots []*tortureKnot) int64 { return n } -// the timer starts before the knot write, so latency includes it -func (w *tortureWorld) probeGrant(repo *tortureRepo, who syntax.DID, within time.Duration, applied *latencies, missed *atomic.Int64, waiters *sync.WaitGroup) int64 { +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) - waiters.Go(func() { + p.wg.Go(func() { if w.r.waitFor(within, func() bool { return w.isCollab(repo, who) }) { - applied.since(start) + p.since(start) return } - missed.Add(1) + p.missed.Add(1) }) return us } -// peak growth of count over any 1s, sampled every 100ms type ingBurst struct { peak atomic.Int64 stop chan struct{} @@ -178,9 +166,7 @@ func init() { good, evil := w.addKnot(), w.addKnot() legit := w.addRepo(good, testOwner, 0, nil) w.boot() - if !w.seed([]*tortureRepo{legit}, 30*time.Second) { - r.t.Fatal("the legit repo never got served") - } + w.seed([]*tortureRepo{legit}, 30*time.Second) evil.delay.Store(delay.Nanoseconds()) lookupsBefore := w.ids.lookups.Load() @@ -196,7 +182,7 @@ func init() { last = w.js.commit(did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, jmodels.CommitOperationUpdate, manifestRecord(owner, "x", w.spindleDid())) } - // owners are allowlisted, so only the missing knot stops these + // 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)) @@ -208,25 +194,23 @@ func init() { } within := time.Duration(slow)*2*delay + time.Minute - var applied latencies - var missed atomic.Int64 - var waiters sync.WaitGroup + var applied grantProbes for i := range probes { - last = w.probeGrant(legit, syntax.DID(fmt.Sprintf("did:plc:probe%04d", i)), within, &applied, &missed, &waiters) + last = w.probeGrant(legit, syntax.DID(fmt.Sprintf("did:plc:probe%04d", i)), within, &applied) } - waiters.Wait() - drained := ingConsumed(w, last, within) - // the reader being past an event doesn't mean its repo job finished - r.waitFor(time.Minute, func() bool { return w.countRows("select count(*) from repo_retries") == 0 }) + 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) - r.metric("legit_missed", float64(missed.Load())) + 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.countRows("select count(*) from repo_retries") + retryRows := w.rows("repo_retries") r.metric("retry_rows", retryRows) - r.check("ING-1 legit events apply while hostile manifests queue", missed.Load() == 0 && applied.quantile(1) <= budget, + 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) @@ -257,31 +241,25 @@ func init() { secondGap := r.param("second_gap", 1) > 0 w := newTortureWorld(r) - for range knots { - w.addKnot() - } + 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() - if !w.seed(repos, 3*time.Minute) { - r.t.Fatalf("only %d of %d repos served after the seed", w.servedCount(repos), n) - } + w.seed(repos, 3*time.Minute) r.gauge("catchup_pending", func() float64 { - s := w.spindle() - if s == nil { - return math.NaN() - } - pending, err := s.db.PendingCatchUp() - if err != nil { - return math.NaN() - } - return float64(len(pending)) + return w.fromSpindle(func(s *Spindle) float64 { + pending, err := s.db.PendingCatchUp() + if err != nil { + return math.NaN() + } + return float64(len(pending)) + }) }) - // the cursor needs to pass everything or the next boot replays it + // the cursor has to pass everything, or the next boot replays it at := w.js.filler() - ingConsumed(w, at, 30*time.Second) + w.consumed(at, 30*time.Second) w.stop() changed := make([]syntax.DID, int(float64(n)*changedFrac)) @@ -301,9 +279,8 @@ func init() { w.boot() probeRepos := max(1, min(5, n/4)) - var applied latencies - var missed atomic.Int64 - var waiters, probing sync.WaitGroup + var applied grantProbes + var probing sync.WaitGroup var lastUS atomic.Int64 var paused atomic.Bool stopProbes := make(chan struct{}) @@ -318,20 +295,20 @@ func init() { 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, &missed, &waiters)) + 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 { - // forget drops the history, so probes in flight have to land first + // wait for the probes in flight first, forget drops history paused.Store(true) - ingConsumed(w, lastUS.Load(), 30*time.Second) + w.consumed(lastUS.Load(), 30*time.Second) w.js.forget() filler := w.js.filler() w.js.dropConnections() - ingConsumed(w, filler, time.Minute) + w.consumed(filler, time.Minute) paused.Store(false) gaps = 2 } @@ -352,14 +329,14 @@ func init() { }) close(stopProbes) probing.Wait() - waiters.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) - r.metric("live_missed", float64(missed.Load())) + 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)) @@ -368,15 +345,14 @@ func init() { 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", missed.Load() == 0 && applied.quantile(1) <= probeBudget, - "%d probes applied, %d missed, slowest %v, want within %v", applied.len(), missed.Load(), applied.quantile(1), probeBudget) - // each gap can erase every did once, so the limit scales with gaps + 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 their repos too, so take them out + // probes refresh the repo too and shouldn't count here catchUpRefreshes := refreshes - float64(applied.len())/float64(n) r.metric("catchup_refreshes_per_repo", catchUpRefreshes) - // the 0.2 is slack for a record that lands while the repo job runs + // 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()) }) @@ -401,9 +377,7 @@ func init() { } } w.boot() - if !w.seed([]*tortureRepo{repo}, 30*time.Second) { - r.t.Fatal("the repo never got served") - } + w.seed([]*tortureRepo{repo}, 30*time.Second) casbinMatches := func() bool { for _, s := range subjects { if w.isCollab(repo, s) != live[s] { @@ -442,7 +416,7 @@ func init() { last = w.grantCollaborator(repo, s) } } - consumed := ingConsumed(w, last, 3*time.Minute) + consumed := w.consumed(last, 3*time.Minute) settled := r.waitFor(30*time.Second, func() bool { return casbinMatches() && rowsMatch() }) reqs := k.xrpc.Load() - reqBefore @@ -455,7 +429,7 @@ func init() { if oversize == 0 { return } - // the list errors past the page cap, so the revoke has to land anyway + // 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) }) { @@ -467,12 +441,12 @@ func init() { oversizeBefore := k.xrpc.Load() w.grantCollaborator(repo, syntax.DID("did:plc:churntrigger")) us := w.revokeCollaborator(repo, victim) - ingConsumed(w, us, time.Minute) + 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 is the page cap in lists plus a few more requests + // 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.countRows("select count(*) from repo_retries")) + 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) }) @@ -486,16 +460,12 @@ func init() { 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) - for range knots { - w.addKnot() - } + 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() - if !w.seed(repos, 3*time.Minute) { - r.t.Fatalf("only %d of %d repos served after the seed", w.servedCount(repos), n) - } + w.seed(repos, 3*time.Minute) for _, k := range w.knots { k.delay.Store(delay.Nanoseconds()) @@ -507,13 +477,13 @@ func init() { subjects[i] = syntax.DID(fmt.Sprintf("did:plc:herd%04d", i)) last = w.grantCollaborator(repo, subjects[i]) } - ingConsumed(w, last, 3*time.Minute) + w.consumed(last, 3*time.Minute) failed := r.waitFor(time.Minute, func() bool { - return w.countRows("select count(*) from repo_retries") >= float64(n) && + 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.countRows("select count(*) from repo_retries")) + 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) @@ -521,11 +491,11 @@ func init() { for i := range outageRounds { w.skew.Add(int64(2 * time.Hour)) r.waitFor(time.Minute, func() bool { - return w.countRows("select count(*) from repo_retries") == 0 || + 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.countRows("select count(*) from repo_retries")) + r.metric("retry_rows_before_recovery", w.rows("repo_retries")) var bursts []*ingBurst for _, k := range w.knots { @@ -535,7 +505,7 @@ func init() { reqBefore := ingXrpc(w.knots) t0 := time.Now() w.skew.Add(int64(2 * time.Hour)) - drained := r.waitFor(3*time.Minute, func() bool { return w.countRows("select count(*) from repo_retries") == 0 }) + 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()) @@ -556,7 +526,7 @@ func init() { 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.countRows("select count(*) from repo_retries")) + 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) }) @@ -573,11 +543,12 @@ func init() { bob := syntax.DID("did:plc:flapbob") ingPutCollaborator(repo, bob) w.boot() - if !w.seed([]*tortureRepo{repo}, 30*time.Second) || !r.waitFor(30*time.Second, func() bool { return w.isCollab(repo, bob) }) { - r.t.Fatal("the repo never got served with its collaborator") + 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") } - // wait for the seed's fetches to stop, so one load can be priced + // 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 { @@ -606,7 +577,7 @@ func init() { 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 ingServed(w, repo) != away }) { + if !r.waitFor(30*time.Second, func() bool { return w.served(repo) != away }) { stuck++ } else if away { forgetting.since(flipStart) @@ -618,18 +589,18 @@ func init() { r.dumpGoroutines() } } - ingConsumed(w, last, time.Minute) + 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 ingServed(w, repo) && w.isCollab(repo, bob) }) + 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, so each return to this spindle loads again + // 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, @@ -646,9 +617,9 @@ func init() { k.put(repo.did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(repo.owner, repo.name, spindle)) burstLast = w.emitManifest(repo) } - ingConsumed(w, burstLast, time.Minute) + w.consumed(burstLast, time.Minute) burstSettled := r.waitFor(time.Minute, func() bool { - return ingServed(w, repo) && w.countRows("select count(*) from repo_retries") == 0 + return w.served(repo) && w.rows("repo_retries") == 0 }) burstLoads := float64(k.fetches.Load()-burstFetches) / float64(fetchesPerLoad) r.metric("burst_schedule_loads", burstLoads) @@ -656,7 +627,7 @@ func init() { "%.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)) }) - // the mark is the only way to the repo job, so a failed mark can't count as done + // 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 @@ -665,9 +636,7 @@ func init() { k := w.addKnot() repo := w.addRepo(k, syntax.DID("did:plc:markowner"), 1, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")}) w.boot() - if !w.seed([]*tortureRepo{repo}, 30*time.Second) { - r.t.Fatal("the repo never got served") - } + w.seed([]*tortureRepo{repo}, 30*time.Second) ctx := context.Background() conn, err := w.spindle().db.Conn(ctx) @@ -683,8 +652,8 @@ func init() { _, _ = conn.ExecContext(ctx, "rollback") conn.Close() - ingConsumed(w, last, time.Minute) - forgotten := r.waitFor(time.Minute, func() bool { return !ingServed(w, repo) }) + 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") }) @@ -703,9 +672,7 @@ func init() { return fast, syntax.DID(fmt.Sprintf("did:plc:fastowner%04d", i)) }, 0, nil) w.boot() - if !w.seed(append([]*tortureRepo{stuck}, repos...), 2*time.Minute) { - r.t.Fatal("the repos never got served") - } + 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") @@ -713,26 +680,24 @@ func init() { w.revokeCollaborator(stuck, first) lastStuck := w.grantCollaborator(stuck, second) - cursorFree := ingConsumed(w, lastStuck, 5*time.Second) + cursorFree := w.consumed(lastStuck, 5*time.Second) - var applied latencies - var missed atomic.Int64 - var waiters sync.WaitGroup + 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, &missed, &waiters) + w.probeGrant(repo, syntax.DID(fmt.Sprintf("did:plc:fastprobe%d-%04d", e, i)), time.Minute, &applied) } } - waiters.Wait() - r.latency("other_apply", &applied) - r.metric("other_missed", float64(missed.Load())) + 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 && missed.Load() == 0 && applied.quantile(1) <= budget, + 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)) }) diff --git a/spindle/torture_notify_abuse_test.go b/spindle/torture_notify_abuse_test.go index 2eb2254e6..32fa03b89 100644 --- a/spindle/torture_notify_abuse_test.go +++ b/spindle/torture_notify_abuse_test.go @@ -7,9 +7,9 @@ import ( "encoding/json" "fmt" "io" - "math" "net/http" "net/http/httptest" + "slices" "strings" "sync" "sync/atomic" @@ -23,15 +23,6 @@ import ( "tangled.org/core/spindle/models" ) -// one owner per knot, the owner's pds follows the last repo added -func (w *tortureWorld) abuseKnots(n int) []*tortureKnot { - ks := make([]*tortureKnot, n) - for i := range ks { - ks[i] = w.addKnot() - } - return ks -} - 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) @@ -39,14 +30,6 @@ func (w *tortureWorld) abuseRepos(ks []*tortureKnot, tag string, perKnot, commit }, commits, workflows) } -func (w *tortureWorld) abuseServe(r *tortureRun, repos []*tortureRepo) { - start := time.Now() - if !w.seed(repos, 3*time.Minute) { - r.t.Fatalf("only %d of %d repos served after the jetstream seed", w.servedCount(repos), len(repos)) - } - r.metric("seed_s", time.Since(start).Seconds()) -} - func abuseQueues(repos []*tortureRepo, from, to int) map[*tortureKnot][]pushDelivery { queues := map[*tortureKnot][]pushDelivery{} for i := from; i < to; i++ { @@ -64,17 +47,13 @@ func abuseRatio(n, of int64) float64 { return float64(n) / float64(of) } -func (w *tortureWorld) abuseHammer(ctx context.Context, n, conc int, token func() string, body []byte, st *deliveryStats) { +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 issued atomic.Int64 var wg sync.WaitGroup for range conc { wg.Go(func() { for ctx.Err() == nil { - if n > 0 && issued.Add(1) > int64(n) { - return - } st.attempts.Add(1) reply := w.post(ctx, client, token(), body) if ctx.Err() != nil { @@ -89,11 +68,10 @@ func (w *tortureWorld) abuseHammer(ctx context.Context, n, conc int, token func( } type abuseHook struct { - srv *httptest.Server - arrived atomic.Int64 - got atomic.Int64 - stop chan struct{} - once sync.Once + srv *httptest.Server + got atomic.Int64 + stop chan struct{} + once sync.Once mu sync.Mutex firstSeen map[string]time.Time @@ -102,7 +80,6 @@ type abuseHook struct { 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) { - h.arrived.Add(1) raw, _ := io.ReadAll(io.LimitReader(req.Body, 1<<20)) if hang > 0 { select { @@ -127,7 +104,7 @@ func newAbuseHook(r *tortureRun, hang time.Duration) *abuseHook { h.mu.Unlock() rw.WriteHeader(http.StatusOK) })) - // cleanups run last in first out, so release frees the hung requests before srv.Close waits on them + // 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 @@ -146,7 +123,7 @@ func abuseHookKey(d pushDelivery) string { return d.newSha + "|" + d.repo.owner.String() + "/" + strings.ToLower(d.repo.name) } -// goes straight to the db, so the per repo and per owner quotas don't apply +// 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) @@ -156,31 +133,23 @@ func (w *tortureWorld) abuseAddHook(repo *tortureRepo, url string) { } func (w *tortureWorld) abuseNotifyInflight() float64 { - s := w.spindle() - if s == nil { - return math.NaN() - } - return float64(s.notifySlots.inflight()) + return w.fromSpindle(func(s *Spindle) float64 { return float64(s.notifySlots.inflight()) }) } func (w *tortureWorld) abuseUnknownInflight() float64 { - s := w.spindle() - if s == nil { - return math.NaN() - } - s.notifyAdmission.mu.Lock() - defer s.notifyAdmission.mu.Unlock() - return float64(s.notifyAdmission.total) + 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 { - s := w.spindle() - if s == nil { - return math.NaN() - } - s.notifySlots.mu.Lock() - defer s.notifySlots.mu.Unlock() - return float64(s.notifySlots.held[host]) + 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) { @@ -201,7 +170,7 @@ func (w *tortureWorld) abuseSybil(ctx context.Context, k *tortureKnot, repos []* } func init() { - // hold is shorter than the knot client's 30s timeout, so slots free up more often than in prod + // 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 @@ -216,9 +185,9 @@ func init() { w.maxJobs = 8 r.gauge("notify_inflight", w.abuseNotifyInflight) r.gauge("notify_unknown_inflight", w.abuseUnknownInflight) - legit := w.abuseKnots(legitKnots) + legit := w.addKnots(legitKnots) rs := w.abuseRepos(legit, "sybillegit", perKnot, pushes, map[string]string{"ci.yaml": tortureWorkflow("sleep 20ms")}) - hostile := w.abuseKnots(hosts) + hostile := w.addKnots(hosts) hostileRepos := map[*tortureKnot][]*tortureRepo{} for i, k := range hostile { for j := range perHost { @@ -227,7 +196,7 @@ func init() { } } w.boot() - w.abuseServe(r, rs) + w.seed(rs, 3*time.Minute) half := pushes / 2 var base deliveryStats @@ -269,7 +238,7 @@ func init() { want2 := len(rs) * (pushes - half) maxWait := st.accepted.quantile(1) if st.accepted.len() < want2 { - // pushes that never got in count as waiting the whole budget, or they'd hide as no wait + // 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()) @@ -281,7 +250,6 @@ func init() { r.check("no legit push waits more than 20s (N2)", maxWait <= 20*time.Second, "longest wait %v, want <= 20s", maxWait.Round(time.Millisecond)) }) - // a knot may take the whole pool until others get turned away, then it's cut to an equal share 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) @@ -293,21 +261,18 @@ func init() { w.maxJobs = 8 w.notifySlots = pool r.gauge("notify_inflight", w.abuseNotifyInflight) - legit := w.abuseKnots(3) + legit := w.addKnots(3) rs := w.abuseRepos(legit, "sharelegit", 2, commits, workflows) - hostile := w.abuseKnots(1) + hostile := w.addKnots(1) hrs := w.abuseRepos(hostile, "sharehostile", 12, commits, workflows) w.boot() - w.abuseServe(r, append(append([]*tortureRepo{}, rs...), hrs...)) + 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() - var ds []pushDelivery - for _, repo := range hrs { - ds = append(ds, pushRange(repo, 0, commits)...) - } + ds := pushRanges(hrs, 0, commits) var hst deliveryStats var atk sync.WaitGroup atk.Go(func() { w.flood(atkCtx, ds, 4*pool, 1000, &hst) }) @@ -320,7 +285,7 @@ func init() { "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) - // sample only after the first hostile calls have run out their hold + // 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{}) @@ -360,7 +325,6 @@ func init() { 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) - // one other knot waiting gets half the pool, with a floor of 2 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) }) @@ -371,13 +335,12 @@ func init() { window := 4 * time.Second w := newTortureWorld(r) - ks := w.abuseKnots(1) + ks := w.addKnots(1) rs := w.abuseRepos(ks, "badtok", 2, 2, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")}) w.boot() - w.abuseServe(r, rs) + w.seed(rs, 3*time.Minute) k, repo := ks[0], rs[0] body := w.notifyBody(repo, "refs/heads/main", zeroSha, repo.shas[0]) - // the first push does the real work, the flood only ever replays it 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) } @@ -429,7 +392,7 @@ func init() { floodCtx, cancelFlood := context.WithTimeout(context.Background(), window) var flood deliveryStats var wg sync.WaitGroup - wg.Go(func() { w.abuseHammer(floodCtx, 0, conc, forged, body, &flood) }) + wg.Go(func() { w.abuseHammer(floodCtx, conc, forged, body, &flood) }) wg.Go(func() { probe(floodCtx, &floodLat, &floodBad) }) wg.Wait() cancelFlood() @@ -452,15 +415,15 @@ func init() { settle := r.duration("settle", 45*time.Second) w := newTortureWorld(r) - r.gauge("webhook_rows", func() float64 { return w.countRows("select count(*) from webhook_deliveries") }) - ks := w.abuseKnots(2) + 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.abuseServe(r, append(append([]*tortureRepo{}, bad...), good...)) + w.seed(append(slices.Clone(bad), good...), 3*time.Minute) hung, fast := newAbuseHook(r, hang), newAbuseHook(r, 0) - // lanes are keyed by hostname, so localhost gets its own lane on the same loopback + // 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 { @@ -503,7 +466,7 @@ func init() { } hookLat.add(seen.Sub(at)) } - // missing deliveries count as the full settle time, or the p99 only sees the lucky ones + // otherwise a dropped delivery never shows up in the p99 for range missing { hookLat.add(settle) } @@ -537,7 +500,7 @@ func init() { w := newTortureWorld(r) r.gauge("notify_inflight", w.abuseNotifyInflight) - ks := w.abuseKnots(2) + ks := w.addKnots(2) hungKnot, okKnot := ks[0], ks[1] stuck := w.addRepo(hungKnot, "did:plc:hungowner", 2, nil) var others []*tortureRepo @@ -546,7 +509,7 @@ func init() { } late := w.addRepo(okKnot, "did:plc:hungok", 0, nil) w.boot() - w.abuseServe(r, append([]*tortureRepo{stuck}, others...)) + w.seed(append([]*tortureRepo{stuck}, others...), 3*time.Minute) hungKnot.gitDelay.Store(int64(hold)) ctx := context.Background() diff --git a/spindle/torture_notify_test.go b/spindle/torture_notify_test.go index f0364077a..45e452500 100644 --- a/spindle/torture_notify_test.go +++ b/spindle/torture_notify_test.go @@ -43,7 +43,7 @@ func (st *deliveryStats) seen(kind string) int { return int(n.(*atomic.Int64).Load()) } -// 503 is the busy signal, any other 5xx counts as a failure +// 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 { @@ -55,6 +55,17 @@ func (st *deliveryStats) serverErrors() int { 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) @@ -66,7 +77,6 @@ func (st *deliveryStats) report(r *tortureRun, prefix string) { }) } -// 4xx but 408 and 429 drops the event, everything else backs off like knot2's outbox func (w *tortureWorld) knotOutbox(ctx context.Context, queue []pushDelivery, st *deliveryStats) { client := &http.Client{Timeout: time.Minute} backoff := time.Duration(0) @@ -76,14 +86,9 @@ func (w *tortureWorld) knotOutbox(ctx context.Context, queue []pushDelivery, st if resp, err := client.Get(w.api.URL + "/.well-known/did.json"); err == nil { resp.Body.Close() } - st.attempts.Add(1) reply := w.notifyWith(ctx, client, d) - st.saw(reply.status) - if attempt == 1 { - st.first.add(reply.elapsed) - } + st.record(reply, attempt, start) if reply.status == http.StatusOK { - st.accepted.since(start) backoff = 0 break } @@ -111,7 +116,6 @@ func (w *tortureWorld) outboxes(ctx context.Context, queues map[*tortureKnot][]p return &finished } -// no outbox rules here, retries start at 50ms and cap at 2s func (w *tortureWorld) flood(ctx context.Context, ds []pushDelivery, concurrency, maxAttempts int, st *deliveryStats) { var wg sync.WaitGroup work := make(chan pushDelivery) @@ -121,14 +125,9 @@ func (w *tortureWorld) flood(ctx context.Context, ds []pushDelivery, concurrency start := time.Now() wait := 50 * time.Millisecond for attempt := 1; ; attempt++ { - st.attempts.Add(1) reply := w.notifyWith(ctx, http.DefaultClient, d) - st.saw(reply.status) - if attempt == 1 { - st.first.add(reply.elapsed) - } + st.record(reply, attempt, start) if reply.status == http.StatusOK { - st.accepted.since(start) break } if (reply.status >= 0 && reply.status < 500) || attempt == maxAttempts || ctx.Err() != nil { @@ -167,15 +166,25 @@ func pushRange(repo *tortureRepo, from, to int) []pushDelivery { 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) drained(within time.Duration) bool { - return w.r.waitFor(within, func() bool { - // jobs goes first, a dequeued job leaves the table before it shows as running - return w.countRows("select count(*) from jobs") == 0 && w.eng.running.Load() == 0 && w.unfinishedWorkflows() == 0 +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 { @@ -188,23 +197,17 @@ func (w *tortureWorld) unfinishedWorkflows() float64 { } func (w *tortureWorld) pushWorld(knots, repos, pushes int, workflows map[string]string) []*tortureRepo { - for range knots { - w.addKnot() - } + 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() - seedStart := time.Now() - if !w.seed(rs, 3*time.Minute) { - w.r.t.Fatalf("only %d of %d repos served after the jetstream seed", w.servedCount(rs), len(rs)) - } - w.r.metric("seed_s", time.Since(seedStart).Seconds()) + w.seed(rs, 3*time.Minute) return rs } func (w *tortureWorld) checkPipelines(r *tortureRun, want int, what string) { - got := int(w.countRows("select count(*) from pipelines")) + got := int(w.rows("pipelines")) r.check("one pipeline per unique push", got == want, "%d pipelines for %d unique pushes (%s)", got, want, what) } @@ -224,16 +227,16 @@ func init() { queues := map[*tortureKnot][]pushDelivery{} unique := 0 for _, repo := range rs { - for _, d := range pushesOf(repo, pushes) { + for _, d := range pushRange(repo, 0, pushes) { queues[repo.knot] = append(queues[repo.knot], d) unique++ - // a lost ack makes the knot resend, so some pushes come twice + // 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) } } } - // shuffles across repos, a repo's own pushes keep their order + // 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 { @@ -247,12 +250,10 @@ func init() { st.report(r, "notify_") r.latency("knot_finished", finished) - drained := w.drained(5 * time.Minute) - r.check("queue drains", drained, "jobs=%v running=%d", w.countRows("select count(*) from jobs"), w.eng.running.Load()) + 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.check("nothing left unfinished", w.unfinishedWorkflows() == 0, "%v workflows not finished", w.unfinishedWorkflows()) r.settles("goroutines", baseGoroutines+20, 30*time.Second) }) @@ -268,10 +269,7 @@ func init() { rs := w.pushWorld(1, repos, pushes, map[string]string{"ci.yaml": tortureWorkflow(step)}) baseGoroutines := r.gaugeValue("goroutines") - var ds []pushDelivery - for _, repo := range rs { - ds = append(ds, pushesOf(repo, pushes)...) - } + ds := pushRanges(rs, 0, pushes) unique := len(ds) for _, d := range ds[:unique] { if rand.Float64() < replays { @@ -285,11 +283,10 @@ func init() { st.report(r, "notify_") r.metric("busy_ratio", float64(st.seen("http_503"))/float64(st.attempts.Load())) - r.check("queue drains", w.drained(5*time.Minute), "jobs=%v running=%d", w.countRows("select count(*) from jobs"), w.eng.running.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.check("nothing left unfinished", w.unfinishedWorkflows() == 0, "%v workflows not finished", w.unfinishedWorkflows()) r.settles("goroutines", baseGoroutines+20, 30*time.Second) }) } diff --git a/spindle/torture_pool_test.go b/spindle/torture_pool_test.go index 8f43526ba..1d43e08c8 100644 --- a/spindle/torture_pool_test.go +++ b/spindle/torture_pool_test.go @@ -5,6 +5,7 @@ package spindle import ( "context" "fmt" + "slices" "sync" "sync/atomic" "time" @@ -12,13 +13,13 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" ) -// knots includes this one, and the share stays under the pool less its reserve +// knots includes the caller's own knot func poolFairShare(pool, knots int) int { return min(max(2, pool/knots), max(1, pool-poolReserve(pool))) } func init() { - // one size per run to fit under MemoryMax, pick with -torture.set pool=64,notify=16 + // one size per run to fit under MemoryMax, e.g. -torture.set pool=64,notify=16 registerTorture("pool-sizing", "what a repo job pool size and a notify slot count cost idle and under a catch-up with pushes", func(r *tortureRun) { n := r.load("repos", 5000) repoPool := int(r.param("pool", 256)) @@ -32,12 +33,8 @@ func init() { w.repoPool, w.notifySlots = repoPool, notifySlots w.maxJobs = 8 big := w.addKnot() - var small []*tortureKnot - for range smallKnots { - small = append(small, w.addKnot()) - } + small := w.addKnots(smallKnots) slow := w.addKnot() - // 60% big knot, 30% small knots, 10% slow knot pick := func(i int) (*tortureKnot, syntax.DID) { owner := syntax.DID(fmt.Sprintf("did:plc:poolowner%05d", i)) switch rem := i % 10; { @@ -78,15 +75,11 @@ func init() { r.metric("idle_goroutines", r.gaugeValue("goroutines")) r.metric("idle_fds", r.gaugeValue("fds")) - all := append(append([]*tortureRepo{}, repos...), pushes...) - seedStart := time.Now() - if !w.seed(all, 10*time.Minute) { - r.t.Fatalf("only %d of %d repos served after the seed", w.servedCount(all), len(all)) - } - r.metric("seed_s", time.Since(seedStart).Seconds()) + all := append(slices.Clone(repos), pushes...) + w.seed(all, 10*time.Minute) r.metric("served_rss_mb", r.gaugeValue("rss_mb")) - ingConsumed(w, w.js.filler(), 30*time.Second) + w.consumed(w.js.filler(), 30*time.Second) w.stop() var changed []syntax.DID var changedOn []*tortureRepo @@ -111,15 +104,10 @@ func init() { var wg sync.WaitGroup var st deliveryStats - var ds []pushDelivery - for _, repo := range pushes { - ds = append(ds, pushRange(repo, 0, 2)...) - } + ds := pushRanges(pushes, 0, 2) wg.Go(func() { w.flood(ctx, ds, min(2*notifySlots, 512), 40, &st) }) - var applied latencies - var missed atomic.Int64 - var waiters sync.WaitGroup + var applied grantProbes stopLive := make(chan struct{}) var liveSent atomic.Int64 wg.Go(func() { @@ -129,12 +117,12 @@ func init() { return case <-time.After(250 * time.Millisecond): } - w.probeGrant(repos[0], syntax.DID(fmt.Sprintf("did:plc:poollive%05d", i)), 2*time.Minute, &applied, &missed, &waiters) + w.probeGrant(repos[0], syntax.DID(fmt.Sprintf("did:plc:poollive%05d", i)), 2*time.Minute, &applied) liveSent.Add(1) } }) - // a job already running can go past the share, starting a new one over it while others wait is the bug + // running jobs can stay over the share, only starting one over it is a violation slowHost := slow.host() var jobsPeak, violations atomic.Int64 stopSample := make(chan struct{}) @@ -170,29 +158,10 @@ func init() { } }() - if _, err := w.spindle().db.Exec(`create table if not exists torture_probe (n integer)`); err != nil { - r.t.Fatal(err) - } + w.probeTable() var probe latencies var probeErrors atomic.Int64 - stopProbe := make(chan struct{}) - wg.Go(func() { - tick := time.NewTicker(50 * time.Millisecond) - defer tick.Stop() - for i := 0; ; i++ { - select { - case <-stopProbe: - return - case <-tick.C: - } - began := time.Now() - if _, err := w.spindle().db.Exec(`insert into torture_probe (n) values (?)`, i); err != nil { - probeErrors.Add(1) - continue - } - probe.since(began) - } - }) + stopProbe := w.probeWrites(50*time.Millisecond, &probe, &probeErrors) caughtUp := r.waitFor(budget, func() bool { pending, err := w.spindle().db.PendingCatchUp() @@ -209,14 +178,14 @@ func init() { }) elapsed := time.Since(start) close(stopLive) - close(stopProbe) + stopProbe() close(stopSample) cancel() wg.Wait() <-sampling - waiters.Wait() + applied.wg.Wait() - // a run that hit the budget has no real time, so only finished waits report one + // over budget there's no real time to report, only finished ones do if caughtUp { r.metric("catchup_s", catchUp.Seconds()) } @@ -231,8 +200,8 @@ func init() { r.metric("slow_knot_jobs_peak_while_others_wait", float64(jobsPeak.Load())) r.metric("fair_share_violations", float64(violations.Load())) r.latency("probe_write", &probe) - r.latency("live_apply", &applied) - r.metric("live_missed", float64(missed.Load())) + r.latency("live_apply", &applied.latencies) + r.metric("live_missed", float64(applied.missed.Load())) st.report(r, "notify_") r.note("rss, goroutines, fds and the knots' open requests are sampled all run, read them as _max in the metrics. busy and locked errors from spindle's own writes show up in errors by message, the probe write errors are only the probe's") @@ -267,10 +236,8 @@ func init() { }, 0, nil) all := append(append(append([]*tortureRepo{}, prioRepos...), plainRepos...), bad...) w.boot() - if !w.seed(all, 5*time.Minute) { - r.t.Fatalf("only %d of %d repos served after the seed", w.servedCount(all), len(all)) - } - ingConsumed(w, w.js.filler(), 30*time.Second) + w.seed(all, 5*time.Minute) + w.consumed(w.js.filler(), 30*time.Second) w.stop() for i, repo := range bad { @@ -292,16 +259,16 @@ func init() { r.metric("hostile_jobs_running", float64(running())) r.check("the hostile knot pins the shared pool (setup)", pinned, "the hostile knot had %d of %d workers running, want %d", running(), pool, pool-poolReserve(pool)) - var prioLat, plainLat latencies - var missed atomic.Int64 - var waiters sync.WaitGroup - w.probeGrant(prioRepos[0], syntax.DID("did:plc:prioprobe"), time.Minute, &prioLat, &missed, &waiters) - w.probeGrant(plainRepos[0], syntax.DID("did:plc:plainprobe"), time.Minute, &plainLat, &missed, &waiters) - waiters.Wait() - r.latency("priority_knot_apply", &prioLat) - r.latency("plain_knot_apply", &plainLat) - r.metric("probes_missed", float64(missed.Load())) - r.check("a priority knot's job runs at once while a hostile knot pins the pool", missed.Load() == 0 && prioLat.quantile(1) <= budget, - "priority knot's event applied after %v with %d missed, want within %v while the hostile knot ran %d workers", prioLat.quantile(1), missed.Load(), budget, running()) + var prioProbe, plainProbe grantProbes + w.probeGrant(prioRepos[0], syntax.DID("did:plc:prioprobe"), time.Minute, &prioProbe) + w.probeGrant(plainRepos[0], syntax.DID("did:plc:plainprobe"), time.Minute, &plainProbe) + prioProbe.wg.Wait() + plainProbe.wg.Wait() + missed := prioProbe.missed.Load() + plainProbe.missed.Load() + r.latency("priority_knot_apply", &prioProbe.latencies) + r.latency("plain_knot_apply", &plainProbe.latencies) + r.metric("probes_missed", float64(missed)) + r.check("a priority knot's job runs at once while a hostile knot pins the pool", missed == 0 && prioProbe.quantile(1) <= budget, + "priority knot's event applied after %v with %d missed, want within %v while the hostile knot ran %d workers", prioProbe.quantile(1), missed, budget, running()) }) } diff --git a/spindle/torture_queue_test.go b/spindle/torture_queue_test.go index 671a14a2f..93178fd32 100644 --- a/spindle/torture_queue_test.go +++ b/spindle/torture_queue_test.go @@ -30,11 +30,8 @@ import ( pipelinecodec "tangled.org/core/spindle/pipeline" ) -var ( - tqSeq atomic.Int64 -) +var tqSeq atomic.Int64 -// records only, no git, the queue only cares about owners func tqWorld(r *tortureRun, owners, perOwner, workers int) (*tortureWorld, []*tortureRepo) { w := newTortureWorld(r) w.maxJobs = workers @@ -43,9 +40,7 @@ func tqWorld(r *tortureRun, owners, perOwner, workers int) (*tortureWorld, []*to return k, ownerFor(i, perOwner) }, 0, nil) w.boot() - if !w.seed(repos, 3*time.Minute) { - r.t.Fatalf("only %d of %d repos served after the jetstream seed", w.servedCount(repos), len(repos)) - } + w.seed(repos, 3*time.Minute) return w, repos } @@ -140,11 +135,9 @@ func (w *tortureWorld) tqStartDelays(ids []models.PipelineId, name string, lat * } } -// has to match logWriteTimeout in xrpc/ci_pipeline_subscribe_logs.go +// must match logWriteTimeout in xrpc/ci_pipeline_subscribe_logs.go const tqLogWriteTimeout = time.Minute -func tqMS(d time.Duration) float64 { return float64(d.Microseconds()) / 1000 } - func tqPeaks(r *tortureRun, fn func(), gauges ...string) []float64 { peaks := make([]float64, len(gauges)) for i, g := range gauges { @@ -172,7 +165,6 @@ func tqPeaks(r *tortureRun, fn func(), gauges ...string) []float64 { return peaks } -// the ceiling is set per phase from its own baseline, so one leak can't fail the next func tqSettle(r *tortureRun, phase, gauge string, ceiling float64, within time.Duration) { ok := r.waitFor(within, func() bool { return r.gaugeValue(gauge) <= ceiling }) r.check(phase+": "+gauge+" settles", ok, "%s is %.0f after %v, want <= %.0f", gauge, r.gaugeValue(gauge), within, ceiling) @@ -187,17 +179,21 @@ func tqSettle(r *tortureRun, phase, gauge string, ceiling float64, within time.D } } +func tqSettleBoth(r *tortureRun, phase string, baseGoroutines, baseFds, fdSlack float64, within time.Duration) { + tqSettle(r, phase, "goroutines", baseGoroutines+30, within) + tqSettle(r, phase, "fds", baseFds+fdSlack, within) +} + func tqFdSummary(r *tortureRun) { - entries, err := os.ReadDir("/proc/self/fd") + targets, err := fdTargets() if err != nil { return } kinds := map[string]int{} others := map[string]int{} - for _, e := range entries { - target, err := os.Readlink(filepath.Join("/proc/self/fd", e.Name())) + for _, target := range targets { switch { - case err != nil: + case target == "": kinds["gone"]++ case strings.HasPrefix(target, "socket:"): kinds["socket"]++ @@ -216,20 +212,6 @@ func tqFdSummary(r *tortureRun) { r.note("fds of kind other by file name: %v", others) } -func tqDirMB(root string) float64 { - var total int64 - _ = filepath.WalkDir(root, func(_ string, d fs.DirEntry, err error) error { - if err != nil || d.IsDir() { - return nil - } - if info, err := d.Info(); err == nil { - total += info.Size() - } - return nil - }) - return float64(total) / (1 << 20) -} - func tqLargestFileMB(root string) float64 { var largest int64 _ = filepath.WalkDir(root, func(_ string, d fs.DirEntry, err error) error { @@ -245,14 +227,13 @@ func tqLargestFileMB(root string) float64 { } func tqOpenLogs(root string) float64 { - entries, err := os.ReadDir("/proc/self/fd") + targets, err := fdTargets() if err != nil { return math.NaN() } n := 0 - for _, e := range entries { - target, err := os.Readlink(filepath.Join("/proc/self/fd", e.Name())) - if err == nil && strings.HasPrefix(target, root) && strings.HasSuffix(target, ".log") { + for _, target := range targets { + if strings.HasPrefix(target, root) && strings.HasSuffix(target, ".log") { n++ } } @@ -276,7 +257,7 @@ func (w *tortureWorld) tqWSURL(nsid, pipeline string, workflows ...string) strin return "ws" + strings.TrimPrefix(w.api.URL, "http") + "/xrpc/" + nsid + "?" + q.Encode() } -// tiny recv buffer, a reader that stops reading backs up the send queue fast +// small recv buffer so a stalled reader backs up the send queue quickly func tqDialer(smallRecvBuf bool) *websocket.Dialer { d := &websocket.Dialer{HandshakeTimeout: 30 * time.Second} if smallRecvBuf { @@ -306,7 +287,7 @@ func tqDrain(conn *websocket.Conn, bytesRead, msgs *atomic.Int64) { } func init() { - // the one worker sits on a hung pipeline, so the backlog builds behind it + // the only worker gets stuck here and the backlog piles up behind it registerTorture("queue-backlog", "a big backlog nobody is draining yet, then a restart that works it off", func(r *tortureRun) { depth := r.load("jobs", 4000) probes := 50 @@ -335,13 +316,13 @@ func init() { quietWrite := quiet.quantile(0.99) levels := []int{depth / 8, depth / 4, depth / 2, depth} for _, level := range levels { - have := int(w.countRows("select count(*) from jobs")) + have := int(w.rows("jobs")) w.tqEnqueueMany(repos, max(level-have, 0), par, 1, &enqueue, "sleep 1ms") if level == levels[0] { firstEnqueue = enqueue.quantile(0.99) } - // the probe dequeues 50 jobs per level, too few to move the depth + // probes only pull 50 jobs a level, not enough to shift the depth var lat latencies stopWrites := make(chan struct{}) var wg sync.WaitGroup @@ -377,10 +358,10 @@ func init() { } r.latency("enqueue", &enqueue) r.latency("status_write_during_dequeue", &writes) - have := int(w.countRows("select count(*) from jobs")) + have := int(w.rows("jobs")) r.metric("db_kb_per_queued_job", (r.gaugeValue("db_mb")-baseMB)*1024/float64(max(have, 1))) - r.metric("dequeue_ms_per_1k_queued", tqMS(lastMedian)/float64(max(have, 1))*1000) + r.metric("dequeue_ms_per_1k_queued", float64(lastMedian.Microseconds())/1000/float64(max(have, 1))*1000) r.check("D1 dequeue cost is flat in queue depth", lastMedian <= 3*firstMedian+500*time.Microsecond, "median dequeue %v at depth %d vs %v at depth %d", lastMedian, levels[len(levels)-1], firstMedian, levels[0]) r.check("D1 a dequeue doesn't stall other writers", writes.quantile(0.99) <= 10*quietWrite+50*time.Millisecond, @@ -392,20 +373,20 @@ func init() { w.maxJobs = int(r.param("drain_workers", 8)) w.boot() - total := int(w.countRows("select count(*) from jobs")) + total := int(w.rows("jobs")) marks := []float64{0.75, 0.5, 0.25, 0} at := make([]time.Duration, len(marks)) drainStart := time.Now() drained := true for i, m := range marks { target := int(float64(total) * m) - if !r.waitFor(4*time.Minute, func() bool { return int(w.countRows("select count(*) from jobs")) <= target }) { + if !r.waitFor(4*time.Minute, func() bool { return int(w.rows("jobs")) <= target }) { drained = false break } at[i] = time.Since(drainStart) } - r.check("queue drains after the restart", drained, "%v jobs still queued", w.countRows("select count(*) from jobs")) + r.check("queue drains after the restart", drained, "%v jobs still queued", w.rows("jobs")) if drained && at[0] > 0 && at[3] > at[2] { first := float64(total) / 4 / at[0].Seconds() last := float64(total) / 4 / (at[3] - at[2]).Seconds() @@ -434,7 +415,7 @@ func init() { r.t.Fatal("the flood never started running") } ahead := len(w.eng.startOrder()) - r.metric("queued_when_smalls_arrive", w.countRows("select count(*) from jobs")) + r.metric("queued_when_smalls_arrive", w.rows("jobs")) var smallIDs []models.PipelineId for range smallJobs { @@ -462,8 +443,8 @@ func init() { } r.metric("flood_starts_ahead_of_first_small", float64(firstSmall-ahead)) r.metric("flood_starts_total", float64(flood)) - r.check("small owners finish while the flood is still queued", done && w.countRows("select count(*) from jobs") > 0, - "done=%v queued=%v", done, w.countRows("select count(*) from jobs")) + r.check("small owners finish while the flood is still queued", done && w.rows("jobs") > 0, + "done=%v queued=%v", done, w.rows("jobs")) r.check("Q3 a small owner is only ever behind about one round of flood jobs", firstSmall >= 0 && firstSmall-ahead <= 2*workers, "%d flood jobs started after the small ones were queued and before the first small one", firstSmall-ahead) r.check("Q3 small owners start within a few job lengths", delay.quantile(0.99) <= 20*floodStep, @@ -495,7 +476,7 @@ func init() { finished := w.tqFinished(2 * time.Minute) r.check("queued jobs survive a clean restart and finish", finished, "%v workflows pending or running", w.unfinishedWorkflows()) - // a killed process never writes its terminal row, so erase those after a clean stop + // erase them here, a killed process never writes its terminal row crashed := startHung("crash") w.stop() w.tqEraseTerminals(crashed) @@ -515,7 +496,6 @@ func init() { r.check("Q1 nothing stays pending or running forever after a crash restart", recovered, "%v workflows pending or running %v after the restart, want 0", w.unfinishedWorkflows(), recoverWait) - // both jobs hang, so the check is that both run, not how long they take if !w.tqFinished(time.Minute) { r.note("workflows from the crash step are still unfinished, wake rounds run on top of them") } @@ -546,7 +526,6 @@ func init() { "%d of %d rounds had a job that never started with idle workers around", stuck, rounds) r.note("the wake race is narrow, a clean result here doesn't prove the channel is safe") - // a dequeue that fails on a locked db must not park the worker for good startedBefore := w.eng.started.Load() id, tpl, record, err := w.tqPipeline(repos[0], 1, "sleep 10ms") if err != nil { @@ -592,13 +571,12 @@ func init() { eng := w.eng repo := repos[0] logDir := filepath.Join(w.dir, "logs") - r.gauge("log_dir_mb", func() float64 { return tqDirMB(logDir) }) baseGoroutines := r.gaugeValue("goroutines") baseFds := r.gaugeValue("fds") - // no newline, so the whole thing sits in the writer's pending buffer + // no newline means the whole thing waits in the writer's pending buffer rawCmd := fmt.Sprintf("raw %d", rawMB<<20) - // tight gc so the heap numbers are live memory, not garbage + // tight gc so heap numbers are live memory, not garbage defer debug.SetGCPercent(debug.SetGCPercent(10)) runtime.GC() baseHeap := r.gaugeValue("heap_mb") @@ -616,7 +594,7 @@ func init() { r.check("L1 output without newlines doesn't pile up in memory", peaks[0]-baseHeap <= heapBudget, "heap grew %.0fMB while a step wrote %dMB with no newline, budget %.0fMB", peaks[0]-baseHeap, rawMB, heapBudget) - // small writes rescan the long pending line, so double the bytes should take double the time + // doubling the bytes should only double the time, small writes rescan the pending line dribble := func(kb int) time.Duration { cmd := fmt.Sprintf("dribble %d", kb<<10) if _, err := w.tqEnqueue(repo, 1, cmd); err != nil { @@ -631,7 +609,7 @@ func init() { "%dKB in 16 byte writes took %v, %dKB took %v, want the double within 3x plus 50ms", dribbleKB, once, 2*dribbleKB, twice) logCmd := fmt.Sprintf("log %d", logMB<<20) - // logs get deleted once archived, so the size is sampled while it still grows + // sample the size while it still grows, logs get deleted once archived r.gauge("largest_log_mb", func() float64 { return tqLargestFileMB(logDir) }) largest := tqPeaks(r, func() { if _, err := w.tqEnqueue(repo, logWorkflows, logCmd); err != nil { @@ -659,11 +637,9 @@ func init() { r.note("many_workflows has no cap in spindle (Q2), this only measures what a pipeline that wide costs. the log engine writes real files, the numbers are about spindle's writer and bookkeeping, nothing an engine does") - tqSettle(r, "after the floods", "goroutines", baseGoroutines+30, 30*time.Second) - tqSettle(r, "after the floods", "fds", baseFds+64, 30*time.Second) + tqSettleBoth(r, "after the floods", baseGoroutines, baseFds, 64, 30*time.Second) }) - // crafted requests are covered by xrpc/ci_subscribe_logs_*_test.go registerTorture("logs-subscribers", "many log subscribers, stalled readers and vanishing clients", func(r *tortureRun) { subs := r.load("subscribers", 300) slow := r.load("slow_readers", 60) @@ -676,10 +652,7 @@ func init() { logDir := filepath.Join(w.dir, "logs") dbStat := func(read func(sql.DBStats) int) func() float64 { return func() float64 { - if s := w.spindle(); s != nil { - return float64(read(s.db.Stats())) - } - return math.NaN() + return w.fromSpindle(func(s *Spindle) float64 { return float64(read(s.db.Stats())) }) } } r.gauge("db_open_conns", dbStat(func(st sql.DBStats) int { return st.OpenConnections })) @@ -705,7 +678,7 @@ func init() { } smallURL := w.tqWSURL(org_tangled.CiSubscribePipelineLogsNSID, string(small), "ci.yaml") bigURL := w.tqWSURL(org_tangled.CiSubscribePipelineLogsNSID, string(big), "ci.yaml") - // the first viewer warms lazy state, so the baseline below already has it + // take the baseline after the first viewer warms lazy state if conn, _, err := tqDialer(false).Dial(smallURL, nil); err == nil { _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) _, _, _ = conn.ReadMessage() @@ -754,10 +727,9 @@ func init() { c.Close() } } - tqSettle(r, "herd leaves", "goroutines", baseGoroutines+30, 30*time.Second) - tqSettle(r, "herd leaves", "fds", baseFds+30, 30*time.Second) + tqSettleBoth(r, "herd leaves", baseGoroutines, baseFds, 30, 30*time.Second) - // stalled readers, with more backlog than the socket buffers can hold + // stalled readers, with more backlog than the socket buffers hold baseGoroutines, baseFds = r.gaugeValue("goroutines"), r.gaugeValue("fds") var slowConns []*websocket.Conn for range slow { @@ -775,8 +747,7 @@ func init() { for _, c := range slowConns { c.UnderlyingConn().Close() } - tqSettle(r, "slow readers leave", "goroutines", baseGoroutines+30, 15*time.Second) - tqSettle(r, "slow readers leave", "fds", baseFds+30, 15*time.Second) + tqSettleBoth(r, "slow readers leave", baseGoroutines, baseFds, 30, 15*time.Second) // they leave with backlog still unsent on the server for _, abrupt := range []bool{false, true} { @@ -805,8 +776,7 @@ func init() { if abrupt { phase = "churn without a close frame" } - tqSettle(r, phase, "goroutines", baseGoroutines+30, 15*time.Second) - tqSettle(r, phase, "fds", baseFds+30, 15*time.Second) + tqSettleBoth(r, phase, baseGoroutines, baseFds, 30, 15*time.Second) } }) @@ -823,28 +793,20 @@ func init() { var recentEarly, recentLate, rareEarly, rareLate latencies probe := func(repo *tortureRepo, elapsed time.Duration) { - early := elapsed < soak/3 - late := elapsed > 2*soak/3 - start := time.Now() - if _, _, _, err := w.s.db.QueryPipelines(context.Background(), repo.did.String(), nil, "", nil, 50); err != nil { - r.count("query", 1) - } - switch { - case early: - recentEarly.since(start) - case late: - recentLate.since(start) - } - start = time.Now() - if _, _, _, err := w.s.db.QueryPipelines(context.Background(), repo.did.String(), nil, "", []string{"manual"}, 30); err != nil { - r.count("query", 1) - } - switch { - case early: - rareEarly.since(start) - case late: - rareLate.since(start) + query := func(kinds []string, limit int, early, late *latencies) { + start := time.Now() + if _, _, _, err := w.s.db.QueryPipelines(context.Background(), repo.did.String(), nil, "", kinds, limit); err != nil { + r.count("query", 1) + } + switch { + case elapsed < soak/3: + early.since(start) + case elapsed > 2*soak/3: + late.since(start) + } } + query(nil, 50, &recentEarly, &recentLate) + query([]string{"manual"}, 30, &rareEarly, &rareLate) } var enqueue latencies @@ -871,13 +833,13 @@ func init() { time.Sleep(time.Until(next)) } r.latency("enqueue", &enqueue) - r.metric("queued_at_end_of_load", w.countRows("select count(*) from jobs")) - r.check("the queue keeps up with the load", w.countRows("select count(*) from jobs") <= 50, - "%v jobs still queued at the end of the load", w.countRows("select count(*) from jobs")) + r.metric("queued_at_end_of_load", w.rows("jobs")) + r.check("the queue keeps up with the load", w.rows("jobs") <= 50, + "%v jobs still queued at the end of the load", w.rows("jobs")) w.tqFinished(2 * time.Minute) - pipelines := w.countRows("select count(*) from pipelines") - statuses := w.countRows("select count(*) from workflow_statuses") + pipelines := w.rows("pipelines") + statuses := w.rows("workflow_statuses") growthMB := r.gaugeValue("db_mb") - baseMB r.metric("pipelines", pipelines) r.metric("status_rows", statuses) @@ -895,7 +857,7 @@ func init() { r.check("QP1 a rare kind filter doesn't walk the whole history", rareLate.len() == 0 || rareLate.quantile(0.5) <= 2*rareEarly.quantile(0.5)+300*time.Microsecond, "median %v late vs %v early with no matching rows", rareLate.quantile(0.5), rareEarly.quantile(0.5)) - archived := w.countRows("select count(*) from mill_artifacts") + archived := w.rows("mill_artifacts") r.metric("archived_logs", archived) if archived > 0 { r.check("G1 a log that is archived leaves local disk", tqFileCount(logDir) <= 10, diff --git a/spindle/torture_soak_test.go b/spindle/torture_soak_test.go index 0786ccbb1..c30f38758 100644 --- a/spindle/torture_soak_test.go +++ b/spindle/torture_soak_test.go @@ -55,12 +55,8 @@ func init() { if n >= len(repo.shas) { continue } - old := zeroSha - if n > 0 { - old = repo.shas[n-1] - } next[repo] = n + 1 - w.knotOutbox(ctx, []pushDelivery{{repo: repo, ref: "refs/heads/main", oldSha: old, newSha: repo.shas[n]}}, &st) + w.knotOutbox(ctx, pushRange(repo, n, n+1), &st) } }) } @@ -87,17 +83,16 @@ func init() { }) wg.Wait() st.report(r, "notify_") - // no heap check, HeapInuse moves with the gc phase and the fake jetstream holds every event + // 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) } - r.check("queue drains", w.drained(2*time.Minute), "jobs=%v running=%d", w.countRows("select count(*) from jobs"), w.eng.running.Load()) - r.check("nothing left unfinished", w.unfinishedWorkflows() == 0, "%v workflows not finished", w.unfinishedWorkflows()) + 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.countRows("select count(*) from repo_retries") == 0 }), "%v repo retries left", w.countRows("select count(*) from repo_retries")) - r.metric("pipelines_per_min", w.countRows("select count(*) from pipelines")/length.Minutes()) - r.metric("db_mb_per_1k_pipelines", r.gaugeValue("db_mb")/w.countRows("select count(*) from pipelines")*1000) + 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) }) } diff --git a/spindle/torture_test.go b/spindle/torture_test.go index 81488b5d1..3c8dd7381 100644 --- a/spindle/torture_test.go +++ b/spindle/torture_test.go @@ -30,20 +30,18 @@ var ( tortureList = flag.Bool("torture.list", false, "print every scenario with what it does and run none") ) -// spindle logs through slog.Default, so this is where warnings get counted var tortureCurrent atomic.Pointer[tortureRun] type tortureLogHandler struct { slog.Handler } -// below warn is never built, so the count only ever sees warn and up func (h tortureLogHandler) Enabled(_ context.Context, level slog.Level) bool { return level >= slog.LevelWarn } func (h tortureLogHandler) Handle(ctx context.Context, rec slog.Record) error { - if r := tortureCurrent.Load(); r != nil && rec.Level >= slog.LevelWarn { + if r := tortureCurrent.Load(); r != nil { r.count("log_"+strings.ToLower(rec.Level.String())+": "+rec.Message, 1) } if !h.Handler.Enabled(ctx, rec.Level) { @@ -72,7 +70,7 @@ func registerTorture(name, desc string, run func(r *tortureRun)) { tortureScenarios = append(tortureScenarios, tortureScenario{name, desc, run}) } -// each scenario is a subtest, run one with -run TestTorture/ +// run one scenario with -run TestTorture/ func TestTorture(t *testing.T) { slices.SortFunc(tortureScenarios, func(a, b tortureScenario) int { return strings.Compare(a.name, b.name) }) if *tortureList { @@ -157,7 +155,7 @@ func newTortureRun(t *testing.T, name string) *tortureRun { }) r.gauge("rss_mb", func() float64 { return procStatusKB("VmRSS") / 1024 }) r.gauge("fds", func() float64 { - fds, err := os.ReadDir("/proc/self/fd") + fds, err := fdTargets() if err != nil { return math.NaN() } @@ -212,7 +210,7 @@ func (r *tortureRun) record(name string, v float64) { r.res.Params[name] = v } -// json can't encode NaN or Inf, so they go to errors instead +// json can't encode NaN or Inf func (r *tortureRun) metric(name string, v float64) { r.mu.Lock() defer r.mu.Unlock() @@ -382,7 +380,7 @@ func (r *tortureRun) finish() { r.mu.Lock() r.res.Seconds = time.Since(r.start).Seconds() for name, all := range r.samples { - // nan means nothing was read (no /proc, spindle stopped), so don't count it as zero + // nan means the read failed, skip it or it counts as zero vs := slices.DeleteFunc(slices.Clone(all), math.IsNaN) if len(vs) < len(all) && (name == "rss_mb" || name == "fds") { r.res.Invariants = append(r.res.Invariants, tortureCheck{Name: name + " is readable", Detail: fmt.Sprintf("%d of %d samples could not be read", len(all)-len(vs), len(all))}) @@ -438,7 +436,7 @@ func (r *tortureRun) growth(name string) float64 { return g } -// the first half is warmup, so only the second half counts +// first half is warmup func growthPerMin(vs []float64) (float64, bool) { half := vs[len(vs)/2:] if len(half) < 8 { @@ -447,7 +445,7 @@ func growthPerMin(vs []float64) (float64, bool) { return slope(half) * 240, true } -// units per sample, samples are 250ms apart +// per sample, not per second, samples are 250ms apart func slope(vs []float64) float64 { n := float64(len(vs)) var sx, sy, sxy, sxx float64 @@ -461,6 +459,19 @@ func slope(vs []float64) float64 { return (n*sxy - sx*sy) / (n*sxx - sx*sx) } +// an fd that closed mid listing reads back as an empty target +func fdTargets() ([]string, error) { + entries, err := os.ReadDir("/proc/self/fd") + if err != nil { + return nil, err + } + targets := make([]string, len(entries)) + for i, e := range entries { + targets[i], _ = os.Readlink(filepath.Join("/proc/self/fd", e.Name())) + } + return targets, nil +} + func procStatusKB(field string) float64 { raw, err := os.ReadFile("/proc/self/status") if err != nil { diff --git a/spindle/torture_world_test.go b/spindle/torture_world_test.go index 3ca0e7e3a..2f20f1163 100644 --- a/spindle/torture_world_test.go +++ b/spindle/torture_world_test.go @@ -39,12 +39,11 @@ import ( "tangled.org/core/xrpc/serviceauth" ) -// one directory for every did, so a knot can be did:web on loopback +// one directory for every did, knots too, since a knot is did:web on loopback type tortureDirectory struct { *identity.MockDirectory delay atomic.Int64 lookups atomic.Int64 - purges atomic.Int64 } func newTortureDirectory() *tortureDirectory { @@ -66,11 +65,6 @@ func (d *tortureDirectory) Lookup(ctx context.Context, atid syntax.AtIdentifier) return d.MockDirectory.Lookup(ctx, atid) } -func (d *tortureDirectory) Purge(ctx context.Context, atid syntax.AtIdentifier) error { - d.purges.Add(1) - return d.MockDirectory.Purge(ctx, atid) -} - func sleepCtx(ctx context.Context, d time.Duration) error { if d <= 0 { return nil @@ -95,29 +89,25 @@ type tortureKnot struct { delay atomic.Int64 gitDelay atomic.Int64 down atomic.Bool - gitDown atomic.Bool xrpc atomic.Int64 fetches atomic.Int64 inflight atomic.Int64 - peak atomic.Int64 - reads sync.Map + + readsMu sync.Mutex + reads map[string]int64 } func (k *tortureKnot) readsOf(collection string) int64 { - n, ok := k.reads.Load(collection) - if !ok { - return 0 - } - return n.(*atomic.Int64).Load() + k.readsMu.Lock() + defer k.readsMu.Unlock() + return k.reads[collection] } func (k *tortureKnot) url() string { return k.srv.URL } func (k *tortureKnot) middleware(records http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if n := k.inflight.Add(1); n > k.peak.Load() { - k.peak.Store(n) - } + k.inflight.Add(1) defer k.inflight.Add(-1) if strings.HasPrefix(r.URL.Path, "/did:") { k.serveGit(w, r) @@ -126,8 +116,9 @@ func (k *tortureKnot) middleware(records http.Handler) http.Handler { k.xrpc.Add(1) q := r.URL.Query() if collection := q.Get("collection"); collection != "" { - n, _ := k.reads.LoadOrStore(collection, new(atomic.Int64)) - n.(*atomic.Int64).Add(1) + k.readsMu.Lock() + k.reads[collection]++ + k.readsMu.Unlock() } if sleepCtx(r.Context(), time.Duration(k.delay.Load())) != nil { return @@ -157,10 +148,6 @@ func (k *tortureKnot) serveGit(w http.ResponseWriter, r *http.Request) { if sleepCtx(r.Context(), time.Duration(k.gitDelay.Load())) != nil { return } - if k.gitDown.Load() { - http.Error(w, "boom", http.StatusInternalServerError) - return - } k.git.ServeHTTP(w, r) } @@ -169,8 +156,7 @@ type tortureRepo struct { owner syntax.DID name string knot *tortureKnot - // every commit gets a branch, or git won't fetch it by sha - shas []string + shas []string } type tortureWorld struct { @@ -186,14 +172,11 @@ type tortureWorld struct { pdsMu sync.Mutex pds map[syntax.DID]*tortureKnot - // read at boot, so set these before the world starts - maxJobs int - admission config.Admission - qm *quota.Manager - skew atomic.Int64 - // the cache would hide a did doc that changes mid run - uncachedIDs bool - // zero keeps the config default + // read at boot, set these before the world starts + maxJobs int + admission config.Admission + qm *quota.Manager + skew atomic.Int64 repoPool, notifySlots int priorityKnots []string @@ -239,9 +222,9 @@ func newTortureWorld(r *tortureRun) *tortureWorld { } w.ids.Insert(identity.Identity{DID: testOwner}) - r.gauge("queued_jobs", func() float64 { return w.countRows("select count(*) from jobs") }) - r.gauge("pipelines", func() float64 { return w.countRows("select count(*) from pipelines") }) - r.gauge("repo_retries", func() float64 { return w.countRows("select count(*) from repo_retries") }) + r.gauge("queued_jobs", func() float64 { return w.rows("jobs") }) + r.gauge("pipelines", func() float64 { return w.rows("pipelines") }) + r.gauge("repo_retries", func() float64 { return w.rows("repo_retries") }) r.gauge("db_mb", func() float64 { var total int64 for _, suffix := range []string{"", "-wal", "-shm"} { @@ -261,33 +244,91 @@ func (w *tortureWorld) spindleDid() string { return config.Server{Hostname: faultHostname}.Did().String() } -// nil once stopped, safe to read from other goroutines +// nil once stopped func (w *tortureWorld) spindle() *Spindle { return w.sp.Load() } -// NaN on error, so any `<=` ceiling check fails closed -func (w *tortureWorld) countRows(query string) float64 { +// NaN once stopped, since a `<=` ceiling check fails closed on it +func (w *tortureWorld) fromSpindle(read func(s *Spindle) float64) float64 { s := w.spindle() if s == nil { return math.NaN() } - var n float64 - if err := s.db.QueryRow(query).Scan(&n); err != nil { - return math.NaN() - } - return n + return read(s) } -func (w *tortureWorld) addKnot() *tortureKnot { - t := w.r.t +func (w *tortureWorld) countRows(query string) float64 { + return w.fromSpindle(func(s *Spindle) float64 { + var n float64 + if err := s.db.QueryRow(query).Scan(&n); err != nil { + return math.NaN() + } + return n + }) +} + +func (w *tortureWorld) newKeyedIdentity(did syntax.DID) (atcrypto.PrivateKey, identity.Identity) { priv, err := atcrypto.GeneratePrivateKeyP256() if err != nil { - t.Fatal(err) + w.r.t.Fatal(err) } pub, err := priv.PublicKey() if err != nil { - t.Fatal(err) + w.r.t.Fatal(err) + } + return priv, identity.Identity{ + DID: did, + Keys: map[string]identity.VerificationMethod{"atproto": {Type: "Multikey", PublicKeyMultibase: pub.Multibase()}}, + } +} + +func (w *tortureWorld) probeTable() { + if _, err := w.spindle().db.Exec(`create table if not exists torture_probe (n integer)`); err != nil { + w.r.t.Fatal(err) + } +} + +// inserts on a timer, because a starved writer shows up as latency +func (w *tortureWorld) probeWrites(every time.Duration, lat *latencies, errs *atomic.Int64) (stop func()) { + done := make(chan struct{}) + var wg sync.WaitGroup + wg.Go(func() { + tick := time.NewTicker(every) + defer tick.Stop() + for i := 0; ; i++ { + select { + case <-done: + return + case <-tick.C: + } + start := time.Now() + if _, err := w.spindle().db.Exec(`insert into torture_probe (n) values (?)`, i); err != nil { + errs.Add(1) + continue + } + lat.since(start) + } + }) + return func() { + close(done) + wg.Wait() + } +} + +func (w *tortureWorld) addKnots(n int) []*tortureKnot { + ks := make([]*tortureKnot, n) + for i := range ks { + ks[i] = w.addKnot() } - k := &tortureKnot{priv: priv} + return ks +} + +func (w *tortureWorld) rows(table string) float64 { + return w.countRows("select count(*) from " + table) +} + +func (w *tortureWorld) addKnot() *tortureKnot { + t := w.r.t + k := &tortureKnot{reads: map[string]int64{}} k.fakeKnot = newFakeKnotBehind(t, k.middleware) k.root = filepath.Join(w.dir, fmt.Sprintf("knot%d", len(w.knots))) if err := os.MkdirAll(k.root, 0o755); err != nil { @@ -300,10 +341,9 @@ func (w *tortureWorld) addKnot() *tortureKnot { InheritEnv: []string{"PATH", "HOME", "GIT_CONFIG_GLOBAL", "GIT_CONFIG_SYSTEM"}, } k.did = serviceauth.DidWeb(k.host()) - w.ids.Insert(identity.Identity{ - DID: k.did, - Keys: map[string]identity.VerificationMethod{"atproto": {Type: "Multikey", PublicKeyMultibase: pub.Multibase()}}, - }) + var id identity.Identity + k.priv, id = w.newKeyedIdentity(k.did) + w.ids.Insert(id) w.knots = append(w.knots, k) return k } @@ -317,7 +357,7 @@ func tortureWorkflow(steps ...string) string { return b.String() } -// no commits skips git, the repo only has records for membership load +// no commits skips git, the repo is only records for membership load func (w *tortureWorld) addRepo(k *tortureKnot, owner syntax.DID, commits int, workflows map[string]string) *tortureRepo { repo, err := w.tryAddRepo(k, owner, commits, workflows) if err != nil { @@ -423,7 +463,7 @@ func (k *tortureKnot) declare(owner syntax.DID, repo *tortureRepo) { k.put(owner, org_tangled.RepoDeclarationNSID, strings.ToLower(repo.name), org_tangled.RepoDeclaration{LexiconTypeID: org_tangled.RepoDeclarationNSID, Repo: repo.did.String()}) } -// git is most of the setup time, so repos go in parallel +// repos go in parallel because git is most of the setup time func (w *tortureWorld) addRepos(n int, pick func(i int) (*tortureKnot, syntax.DID), commits int, workflows map[string]string) []*tortureRepo { repos := make([]*tortureRepo, n) var wg sync.WaitGroup @@ -465,26 +505,46 @@ func (w *tortureWorld) emitCollaborator(repo *tortureRepo, who syntax.DID, op st return w.js.collaborator(repo.did, who, op) } -// production learns repos from jetstream, so seed does too -func (w *tortureWorld) seed(repos []*tortureRepo, within time.Duration) bool { +// production learns repos from jetstream and seed does too +func (w *tortureWorld) seed(repos []*tortureRepo, within time.Duration) { + start := time.Now() var last int64 for _, repo := range repos { w.emitDeclaration(repo) last = w.emitManifest(repo) } - return w.r.waitFor(within, func() bool { return w.servedCount(repos) == len(repos) && w.s.jc.LastSeenUs() > last }) + if !w.r.waitFor(within, func() bool { return w.servedCount(repos) == len(repos) && w.s.jc.LastSeenUs() > last }) { + w.r.t.Fatalf("only %d of %d repos served after the jetstream seed", w.servedCount(repos), len(repos)) + } + w.r.metric("seed_s", time.Since(start).Seconds()) } func (w *tortureWorld) servedCount(repos []*tortureRepo) int { n := 0 for _, repo := range repos { - if rows, err := w.s.db.ReposByDid(repo.did); err == nil && len(rows) > 0 { + if w.served(repo) { n++ } } return n } +func (w *tortureWorld) served(repo *tortureRepo) bool { + s := w.spindle() + if s == nil { + return false + } + rows, err := s.db.ReposByDid(repo.did) + return err == nil && len(rows) > 0 +} + +func (w *tortureWorld) consumed(us int64, within time.Duration) bool { + return w.r.waitFor(within, func() bool { + s := w.spindle() + return s != nil && s.jc.LastSeenUs() > us + }) +} + func (w *tortureWorld) backfill(did syntax.DID) { var events []tapc.RecordEventData for _, k := range w.knots { @@ -500,7 +560,6 @@ func (w *tortureWorld) boot() { if err != nil { t.Fatal(err) } - // production defaults, so nothing runs on a zero a real deploy never has cfg := &config.Config{} if err := envconfig.ProcessWith(ctx, &envconfig.Config{ Target: cfg, @@ -541,11 +600,7 @@ func (w *tortureWorld) boot() { if err != nil { t.Fatalf("New: %v", err) } - // the same cache production puts in front of plc - var dir identity.Directory = identity.NewCacheDirectory(w.ids, 250_000, 24*time.Hour, 2*time.Minute, 5*time.Minute) - if w.uncachedIDs { - dir = w.ids - } + dir := identity.NewCacheDirectory(w.ids, 250_000, 24*time.Hour, 2*time.Minute, 5*time.Minute) s.res = idresolver.NewMockResolver(dir) s.knots = knotclient.New(dir, knotclient.HTTPClient(true), true) s.backfill.start = w.hyd.start @@ -564,7 +619,7 @@ func (w *tortureWorld) boot() { } } -// keeps the db on disk so the next boot can reopen it +// the db stays on disk for the next boot to reopen func (w *tortureWorld) stop() { if w.cancel == nil { return @@ -605,32 +660,46 @@ func (w *tortureWorld) notifyBody(repo *tortureRepo, ref, oldSha, newSha string) return body } -func (w *tortureWorld) token(k *tortureKnot) string { - lxm := syntax.NSID(org_tangled.EventNotifyEventNSID) - token, err := auth.SignServiceAuth(k.did, w.spindleDid(), time.Minute, &lxm, k.priv) +func (w *tortureWorld) sign(did syntax.DID, priv atcrypto.PrivateKey, nsid string) string { + lxm := syntax.NSID(nsid) + token, err := auth.SignServiceAuth(did, w.spindleDid(), time.Minute, &lxm, priv) if err != nil { panic(err) } return token } -func (w *tortureWorld) post(ctx context.Context, client *http.Client, token string, body []byte) notifyReply { +func (w *tortureWorld) token(k *tortureKnot) string { + return w.sign(k.did, k.priv, org_tangled.EventNotifyEventNSID) +} + +func (w *tortureWorld) call(ctx context.Context, client *http.Client, method, nsid, query, token string, body []byte) notifyReply { start := time.Now() - req, err := http.NewRequestWithContext(ctx, http.MethodPost, w.api.URL+"/xrpc/"+org_tangled.EventNotifyEventNSID, bytes.NewReader(body)) + target := w.api.URL + "/xrpc/" + nsid + if query != "" { + target += "?" + query + } + req, err := http.NewRequestWithContext(ctx, method, target, bytes.NewReader(body)) if err != nil { panic(err) } req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", "Bearer "+token) + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } resp, err := client.Do(req) if err != nil { return notifyReply{status: -1, body: err.Error(), elapsed: time.Since(start)} } defer resp.Body.Close() - raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<20)) return notifyReply{status: resp.StatusCode, body: string(raw), elapsed: time.Since(start)} } +func (w *tortureWorld) post(ctx context.Context, client *http.Client, token string, body []byte) notifyReply { + return w.call(ctx, client, http.MethodPost, org_tangled.EventNotifyEventNSID, "", token, body) +} + func statusKind(status int) string { if status < 0 { return "transport"