diff --git a/docs/src/content/docs/reference/spindle-config.md b/docs/src/content/docs/reference/spindle-config.md index 6bbcc90c6..7ebd6040f 100644 --- a/docs/src/content/docs/reference/spindle-config.md +++ b/docs/src/content/docs/reference/spindle-config.md @@ -61,7 +61,11 @@ roles. | `SPINDLE_SERVER_DEV_EXTRA_HOSTS` | unset | Declared but not read by the current code. | | `SPINDLE_SERVER_LOG_DIR` | `/var/log/spindle` | Standalone only: used as the disk artifact store when `SPINDLE_ARTIFACT_STORES_DISK_DIR` is unset (with a warning). | | `SPINDLE_SERVER_MAX_JOB_COUNT` | `2` | Pipelines that run at once. | -| `SPINDLE_SERVER_REPO_JOB_WORKERS` | `64` | Repos refreshed at once, including their record marks and pull events. One knot may use all of them while no other knot has work waiting, and an equal share each once others do. | +| `SPINDLE_SERVER_REPO_JOB_WORKERS` | `256` | Repos refreshed at once, including their record marks and pull events. One knot may use all but a small reserve (an eighth, at least 2) while no other knot has work waiting, and an equal share each once others do. | +| `SPINDLE_SERVER_NOTIFY_SLOTS` | `64` | Notify events from knots that fetch or sync at once. While no other knot is waiting, one knot may use all but a reserve of the pool (an eighth, at least 2), and once others wait each gets an equal share (at least 2). A knot past its share gets a 503 and its outbox retries. | +| `SPINDLE_SERVER_PRIORITY_NOTIFY_SLOTS` | `0` | Notify slots only the knots in `SPINDLE_SERVER_PRIORITY_KNOTS` can use, on top of their share of the pool above. `0` means a quarter of `SPINDLE_SERVER_NOTIFY_SLOTS`. | +| `SPINDLE_SERVER_PRIORITY_KNOTS` | unset | Comma-separated knots, as hosts or `did:web:` dids, that get their own repo job workers (and notify slots) no other knot can use, on top of their share of the common pool. | +| `SPINDLE_SERVER_PRIORITY_REPO_JOB_WORKERS` | `0` | Repo job workers only the priority knots use. `0` means a quarter of `SPINDLE_SERVER_REPO_JOB_WORKERS`. Unused when no priority knots are set. | | `SPINDLE_SERVER_DOCKER_SOCKET` | unset | Docker socket exposed to nixery workflow containers. | | `SPINDLE_SERVER_CORS_ORIGINS` | unset | Comma-separated origins allowed to call the XRPC API from a browser, or `*` for any. Unset means CORS is off. | diff --git a/nix/modules/spindle.nix b/nix/modules/spindle.nix index 6c14ed915..1345c4154 100644 --- a/nix/modules/spindle.nix +++ b/nix/modules/spindle.nix @@ -89,6 +89,27 @@ in description = "Maximum number of concurrent jobs to run"; }; + priorityKnots = mkOption { + type = types.listOf types.str; + default = []; + example = ["knot1.tangled.sh" "did:web:knot2.example.com"]; + description = "Knots, as hosts or did:web dids, whose repos get repo job workers and notify slots no other knot can use. Empty gives nobody priority."; + }; + + notifySlots = mkOption { + type = types.ints.positive; + default = 64; + example = 128; + description = "Notify events from knots that fetch or sync at once. One knot can use all but an eighth of them while no other knot is waiting, and gets an equal share once others are."; + }; + + priorityNotifySlots = mkOption { + type = types.ints.unsigned; + default = 0; + example = 16; + description = "Notify slots only the priority knots can use, on top of their share of the pool. 0 means a quarter of notifySlots."; + }; + corsOrigins = mkOption { type = types.listOf types.str; default = []; @@ -723,6 +744,9 @@ in "SPINDLE_SERVER_ADMINS=${concatStringsSep "," instance.server.admins}" "SPINDLE_SERVER_ADMISSION=${instance.server.admission}" "SPINDLE_SERVER_MAX_JOB_COUNT=${toString instance.server.maxJobCount}" + "SPINDLE_SERVER_PRIORITY_KNOTS=${concatStringsSep "," instance.server.priorityKnots}" + "SPINDLE_SERVER_NOTIFY_SLOTS=${toString instance.server.notifySlots}" + "SPINDLE_SERVER_PRIORITY_NOTIFY_SLOTS=${toString instance.server.priorityNotifySlots}" "SPINDLE_SERVER_CORS_ORIGINS=${concatStringsSep "," instance.server.corsOrigins}" "SPINDLE_SERVER_SECRETS_PROVIDER=${instance.server.secrets.provider}" "SPINDLE_SERVER_SECRETS_OPENBAO_PROXY_ADDR=${instance.server.secrets.openbao.proxyAddr}" diff --git a/spindle/TORTURE.md b/spindle/TORTURE.md index 155c38065..41f6dd541 100644 --- a/spindle/TORTURE.md +++ b/spindle/TORTURE.md @@ -66,6 +66,28 @@ flags: - `-torture.log warn` prints spindle logs from that level, the default is `error`. warnings and errors are 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. + +```sh +for pool in 16 64 256 1024; do + /tmp/torture.test -test.run 'TestTorture/pool-sizing' -torture.set pool=$pool,notify=64 -torture.out /tmp/pool.jsonl +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). + +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. + ## results each json line has `params`, `metrics` (latency percentiles as `_p50_ms` .. diff --git a/spindle/config/config.go b/spindle/config/config.go index 5443b9be8..ea85afeaf 100644 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -32,10 +32,15 @@ type Server struct { Secrets Secrets `env:",prefix=SECRETS_"` LogDir string `env:"LOG_DIR, default=/var/log/spindle"` MaxJobCount int `env:"MAX_JOB_COUNT, default=2"` // max number of pipelines that run at a time - RepoJobWorkers int `env:"REPO_JOB_WORKERS, default=64"` // repos refreshed, and their record marks and pull events worked through, at once + RepoJobWorkers int `env:"REPO_JOB_WORKERS, default=256"` // repos refreshed, and their record marks and pull events worked through, at once + NotifySlots int `env:"NOTIFY_SLOTS, default=64"` // notify events from knots that fetch or sync at once, shared between knots LogMaxBytes int64 `env:"LOG_MAX_BYTES, default=33554432"` // size of one workflow's stored jsonl log, envelope included, not raw output. output past it is dropped, 0 means no cap DockerSocket string `env:"DOCKER_SOCKET"` // path to a docker socket to expose to workflow containers CorsOrigins []string `env:"CORS_ORIGINS"` // origins browsers may call xrpc from, "*" for any. empty leaves cors off + + PriorityKnots []string `env:"PRIORITY_KNOTS"` // knot dids or hosts whose repos get workers and notify slots no other knot can use + PriorityRepoJobWorkers int `env:"PRIORITY_REPO_JOB_WORKERS, default=0"` // repo job workers only the priority knots use, 0 means a quarter of REPO_JOB_WORKERS + PriorityNotifySlots int `env:"PRIORITY_NOTIFY_SLOTS, default=0"` // notify slots only the priority knots use, 0 means a quarter of NOTIFY_SLOTS } type Hydrant struct { @@ -306,9 +311,18 @@ func (c *Config) validate() error { if c.Schedule.MinimumInterval < time.Minute || c.Schedule.MinimumInterval%time.Minute != 0 { return fmt.Errorf("SPINDLE_SCHEDULE_MINIMUM_INTERVAL must be a positive whole number of minutes") } + if c.Server.PriorityRepoJobWorkers < 0 { + return fmt.Errorf("SPINDLE_SERVER_PRIORITY_REPO_JOB_WORKERS must not be negative") + } if c.Server.RepoJobWorkers < 1 { return fmt.Errorf("SPINDLE_SERVER_REPO_JOB_WORKERS must be greater than zero") } + if c.Server.PriorityNotifySlots < 0 { + return fmt.Errorf("SPINDLE_SERVER_PRIORITY_NOTIFY_SLOTS must not be negative") + } + if c.Server.NotifySlots < 1 { + return fmt.Errorf("SPINDLE_SERVER_NOTIFY_SLOTS must be greater than zero") + } if c.Schedule.Concurrency < 1 { return fmt.Errorf("SPINDLE_SCHEDULE_CONCURRENCY must be greater than zero") } diff --git a/spindle/config/config_test.go b/spindle/config/config_test.go index 78fa5197e..9955c0151 100644 --- a/spindle/config/config_test.go +++ b/spindle/config/config_test.go @@ -539,8 +539,8 @@ func TestLoadRepoJobWorkersDefaultsAndIsRead(t *testing.T) { if err != nil { t.Fatal(err) } - if cfg.Server.RepoJobWorkers != 64 { - t.Fatalf("repo job workers = %d, want 64 when nothing is set", cfg.Server.RepoJobWorkers) + if cfg.Server.RepoJobWorkers != 256 { + t.Fatalf("repo job workers = %d, want 256 when nothing is set", cfg.Server.RepoJobWorkers) } t.Setenv("SPINDLE_SERVER_REPO_JOB_WORKERS", "12") @@ -554,3 +554,76 @@ func TestLoadRepoJobWorkersDefaultsAndIsRead(t *testing.T) { t.Fatal("Load accepted no repo job workers") } } + +func TestLoadNotifySlotsDefaultsAndIsRead(t *testing.T) { + standaloneEnv(t) + + cfg, err := Load(context.Background()) + if err != nil { + t.Fatal(err) + } + if cfg.Server.NotifySlots != 64 { + t.Fatalf("notify slots = %d, want 64 when nothing is set", cfg.Server.NotifySlots) + } + + t.Setenv("SPINDLE_SERVER_NOTIFY_SLOTS", "8") + cfg, err = Load(context.Background()) + if err != nil || cfg.Server.NotifySlots != 8 { + t.Fatalf("notify slots = %d, %v, want the configured 8", cfg.Server.NotifySlots, err) + } + + t.Setenv("SPINDLE_SERVER_NOTIFY_SLOTS", "0") + if _, err := Load(context.Background()); err == nil { + t.Fatal("Load accepted no notify slots") + } +} + +func TestLoadPriorityKnotsAndTheirWorkersAreRead(t *testing.T) { + standaloneEnv(t) + + cfg, err := Load(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(cfg.Server.PriorityKnots) != 0 || cfg.Server.PriorityRepoJobWorkers != 0 { + t.Fatalf("priority knots = %v, workers = %d, want none set by default", cfg.Server.PriorityKnots, cfg.Server.PriorityRepoJobWorkers) + } + + t.Setenv("SPINDLE_SERVER_PRIORITY_KNOTS", "did:web:knot1.tangled.sh,knot2.example.com") + t.Setenv("SPINDLE_SERVER_PRIORITY_REPO_JOB_WORKERS", "9") + cfg, err = Load(context.Background()) + if err != nil { + t.Fatal(err) + } + if got := cfg.Server.PriorityKnots; len(got) != 2 || got[0] != "did:web:knot1.tangled.sh" || got[1] != "knot2.example.com" { + t.Fatalf("priority knots = %v, want both spellings as given", got) + } + if cfg.Server.PriorityRepoJobWorkers != 9 { + t.Fatalf("priority workers = %d, want 9", cfg.Server.PriorityRepoJobWorkers) + } + + t.Setenv("SPINDLE_SERVER_PRIORITY_REPO_JOB_WORKERS", "-1") + if _, err := Load(context.Background()); err == nil { + t.Fatal("Load accepted a negative number of priority workers") + } +} + +func TestLoadPriorityNotifySlotsDefaultsAndIsRead(t *testing.T) { + standaloneEnv(t) + + cfg, err := Load(context.Background()) + if err != nil || cfg.Server.PriorityNotifySlots != 0 { + t.Fatalf("priority notify slots = %d, %v, want 0 (a quarter of the pool) when nothing is set", cfg.Server.PriorityNotifySlots, err) + } + + t.Setenv("SPINDLE_SERVER_PRIORITY_NOTIFY_SLOTS", "6") + cfg, err = Load(context.Background()) + if err != nil || cfg.Server.PriorityNotifySlots != 6 { + t.Fatalf("priority notify slots = %d, %v, want the configured 6", cfg.Server.PriorityNotifySlots, err) + } + + t.Setenv("SPINDLE_SERVER_PRIORITY_NOTIFY_SLOTS", "-1") + if _, err := Load(context.Background()); err == nil { + t.Fatal("Load accepted a negative number of priority notify slots") + } +} diff --git a/spindle/db/pull_events_limit_test.go b/spindle/db/pull_events_limit_test.go new file mode 100644 index 000000000..1149ad103 --- /dev/null +++ b/spindle/db/pull_events_limit_test.go @@ -0,0 +1,31 @@ +package db + +import ( + "fmt" + "testing" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +func TestDuePullEventsReadsOnlyALimitOldestFirst(t *testing.T) { + d := newTestDB(t) + repo := syntax.DID("did:plc:a") + if err := d.AddPullTicket(repo, "t1", PullTicket{Author: "did:plc:bob", State: "open", Version: "v0"}); err != nil { + t.Fatal(err) + } + for i := 1; i <= 5; i++ { + ev := PullEvent{RepoDid: repo, Ticket: "t1", Event: "closed", Pull: []byte("{}")} + if err := d.AdvancePullTicket(repo, "t1", fmt.Sprintf("v%d", i-1), fmt.Sprintf("v%d", i), "closed", &ev); err != nil { + t.Fatal(err) + } + } + + due, err := d.DuePullEvents(repo, time.Now().Add(time.Hour), 3) + if err != nil || len(due) != 3 { + t.Fatalf("read %d events, %v, want the limit of 3", len(due), err) + } + if due[0].ID >= due[1].ID || due[1].ID >= due[2].ID { + t.Fatalf("ids %d %d %d, want oldest first", due[0].ID, due[1].ID, due[2].ID) + } +} diff --git a/spindle/db/pull_tickets_test.go b/spindle/db/pull_tickets_test.go index 637b5bd49..bfa1414dc 100644 --- a/spindle/db/pull_tickets_test.go +++ b/spindle/db/pull_tickets_test.go @@ -16,7 +16,6 @@ func TestAStaleTicketAdvanceIsRefusedAndLeavesNoEvent(t *testing.T) { } ev := PullEvent{RepoDid: repo, Ticket: "t1", Event: "closed", Pull: []byte("{}")} - // two syncs read v1, the first one gets to v2 and the second still thinks it is at v1 if err := d.AdvancePullTicket(repo, "t1", "v1", "v2", "closed", &ev); err != nil { t.Fatal(err) } @@ -29,7 +28,7 @@ func TestAStaleTicketAdvanceIsRefusedAndLeavesNoEvent(t *testing.T) { if err != nil || got.Version != "v2" || got.State != "closed" { t.Fatalf("ticket = %+v, %v, want the first advance kept", got, err) } - due, err := d.DuePullEvents(repo, time.Now().Add(time.Hour)) + due, err := d.DuePullEvents(repo, time.Now().Add(time.Hour), 100) if err != nil || len(due) != 1 { t.Fatalf("events = %d, %v, want only the first advance's", len(due), err) } @@ -48,7 +47,7 @@ func TestAStaleStateSetIsRefusedAndLeavesNoEvent(t *testing.T) { if err := d.SetPullTicketState(repo, "t1", "open", "completed", ev); !errors.Is(err, ErrPullRaced) { t.Fatalf("a second set from open = %v, want ErrPullRaced", err) } - due, err := d.DuePullEvents(repo, time.Now().Add(time.Hour)) + due, err := d.DuePullEvents(repo, time.Now().Add(time.Hour), 100) if err != nil || len(due) != 1 { t.Fatalf("events = %d, %v, want only the first set's", len(due), err) } diff --git a/spindle/db/pulls.go b/spindle/db/pulls.go index 79619f153..8cd0891bf 100644 --- a/spindle/db/pulls.go +++ b/spindle/db/pulls.go @@ -13,7 +13,7 @@ import ( type PullTicket struct { Author syntax.DID State string - // the knot version the state was last read at, empty for tickets stored before versions were tracked + // empty on rows written before versions were tracked Version string } @@ -74,8 +74,7 @@ func (d *DB) AddPullTicket(repoDid syntax.DID, rkey string, ticket PullTicket) e return err } -// another sync moved the ticket since this one read it, so what it computed is stale. the caller -// retries from the row as it is now +// the ticket moved since it was read, re-read the row and retry var ErrPullRaced = errors.New("the ticket changed since it was read") func movedOne(res sql.Result, err error) error { @@ -88,8 +87,6 @@ func movedOne(res sql.Result, err error) error { return nil } -// the version moves with the state it was read at, and the event that came with it commits in the -// same tx. a version that changed nothing passes no event. prev is the version the caller read func (d *DB) AdvancePullTicket(repoDid syntax.DID, rkey, prev, version, state string, ev *PullEvent) error { tx, err := d.Begin() if err != nil { @@ -111,7 +108,6 @@ func (d *DB) AdvancePullTicket(repoDid syntax.DID, rkey, prev, version, state st } // the state and its event commit together, or a crash between them loses the event for good -// prev is the state the caller read func (d *DB) SetPullTicketState(repoDid syntax.DID, rkey, prev, state string, ev PullEvent) error { tx, err := d.Begin() if err != nil { @@ -196,11 +192,11 @@ func addPullEvent(tx *sql.Tx, ev PullEvent) error { } // oldest first, so a pull's webhooks go out in the order its events happened -func (d *DB) DuePullEvents(repoDid syntax.DID, now time.Time) ([]PullEvent, error) { +func (d *DB) DuePullEvents(repoDid syntax.DID, now time.Time, limit int) ([]PullEvent, error) { rows, err := d.Query( `select id, ticket, event, author, sender, pull, rounds, head, source_repo, source_branch, pipeline_id, webhook_fired, attempts - from pull_events where repo_did = ? and due <= ? order by id`, - repoDid.String(), now.Unix(), + from pull_events where repo_did = ? and due <= ? order by id limit ?`, + repoDid.String(), now.Unix(), limit, ) if err != nil { return nil, err diff --git a/spindle/notify_event.go b/spindle/notify_event.go index 062d77052..58f05218e 100644 --- a/spindle/notify_event.go +++ b/spindle/notify_event.go @@ -29,10 +29,13 @@ import ( ) const ( - notifyWorkBudget = 5 * time.Minute - notifyMaxInflight = 16 - // a hostile knot can sit on its slots for the whole budget, so one knot never gets them all - notifyMaxPerKnot = 4 + notifyWorkBudget = 5 * time.Minute + // a hostile knot can sit on its slots for the whole budget, so a knot only keeps the pool to itself + // while nobody else is waiting. a turned away knot retries with a backoff of up to a minute, so it + // counts as waiting until it gets in, or for twice that if it never comes back + defaultNotifySlots = 64 + notifyMinShare = 2 + notifyWaitWindow = 2 * time.Minute // repos nobody serves yet cost a knot read and a pds read before anything else, so they get a // small pool of their own and can't crowd out the repos that are served @@ -59,31 +62,146 @@ func (e notifyStatus) Error() string { return e.err.Error() } func (e notifyStatus) Unwrap() error { return e.err } func (e notifyStatus) HTTPStatus() int { return e.status } -// the map only holds knots with work running, so it can't grow with the knots seen -type knotSlots struct { - mu sync.Mutex - held map[string]int +// the repos table and the knot's did doc spell a host differently, and every cap is per host +func canonicalKnotHost(host string) string { + name, port, hasPort := strings.Cut(strings.ToLower(host), ":") + name = strings.TrimSuffix(name, ".") + if hasPort { + return name + ":" + port + } + return name +} + +// what a request for a slot is measured against +type slotLimits struct { + pool int + // never taken by one knot alone, so a newcomer finds it on an idle pool + reserve int + // slots only the priority knots can take + dedicated int + priority map[string]bool } -func (k *knotSlots) acquire(host string) bool { - k.mu.Lock() - defer k.mu.Unlock() - if k.held[host] >= notifyMaxPerKnot { +// pool/8 with a floor of 2 +func slotReserve(pool int) int { return max(2, pool/8) } + +// the pool is shared by the knots that have work, so it only holds knots with work running or +// waiting, and it can't grow with the knots seen +type notifySlots struct { + mu sync.Mutex + held map[string]int + total int + prioHeld map[string]int + prioTotal int + waiting map[string]time.Time + swept time.Time + now func() time.Time +} + +func (n *notifySlots) clock() time.Time { + if n.now != nil { + return n.now() + } + return time.Now() +} + +// a priority knot takes a dedicated slot first, and every knot then goes by the shared rules. a knot may +// take all but the reserve of the pool while no other knot is waiting. once others are, each gets an +// equal share of it, and a knot at its share is turned away and counted as waiting itself +func (n *notifySlots) acquire(host string, lim slotLimits) bool { + host = canonicalKnotHost(host) + n.mu.Lock() + defer n.mu.Unlock() + now := n.clock() + n.forgetStale(now) + + if lim.priority[host] && n.prioTotal < lim.dedicated { + if n.prioHeld == nil { + n.prioHeld = map[string]int{} + } + n.prioHeld[host]++ + n.prioTotal++ + return true + } + + limit := max(1, lim.pool-lim.reserve) + if others := n.othersWaiting(host); others > 0 { + limit = max(notifyMinShare, lim.pool/(others+1)) + } + if n.total >= lim.pool || n.held[host] >= limit { + if n.waiting == nil { + n.waiting = map[string]time.Time{} + } + n.waiting[host] = now return false } - if k.held == nil { - k.held = map[string]int{} + if n.held == nil { + n.held = map[string]int{} } - k.held[host]++ + n.held[host]++ + n.total++ + delete(n.waiting, host) return true } -func (k *knotSlots) release(host string) { - k.mu.Lock() - defer k.mu.Unlock() - if k.held[host]--; k.held[host] == 0 { - delete(k.held, host) +// a shared slot goes back first, it's the one other knots wait for +func (n *notifySlots) release(host string) { + host = canonicalKnotHost(host) + n.mu.Lock() + defer n.mu.Unlock() + if n.held[host] == 0 { + n.prioTotal-- + if n.prioHeld[host]--; n.prioHeld[host] == 0 { + delete(n.prioHeld, host) + } + return + } + n.total-- + if n.held[host]--; n.held[host] == 0 { + delete(n.held, host) + } +} + +func (n *notifySlots) inflight() int { + n.mu.Lock() + defer n.mu.Unlock() + return n.total + n.prioTotal +} + +func (n *notifySlots) othersWaiting(host string) int { + count := len(n.waiting) + if _, ok := n.waiting[host]; ok { + count-- } + return count +} + +// once a window is enough, the entries are only read against the window +func (n *notifySlots) forgetStale(now time.Time) { + if now.Sub(n.swept) < notifyWaitWindow { + return + } + n.swept = now + for host, at := range n.waiting { + if now.Sub(at) >= notifyWaitWindow { + delete(n.waiting, host) + } + } +} + +// the hosts of the configured priority knots, which may be named by host or by did:web +func priorityKnotHosts(knots []string) map[string]bool { + hosts := map[string]bool{} + for _, knot := range knots { + host := strings.TrimSpace(knot) + if rest, ok := strings.CutPrefix(strings.ToLower(host), "did:web:"); ok { + host = strings.ReplaceAll(rest, "%3a", ":") + } + if host != "" { + hosts[canonicalKnotHost(host)] = true + } + } + return hosts } // the pool for repos the db doesn't know, plus a short memory of the ones that turned out not served here @@ -97,6 +215,7 @@ type notifyAdmission struct { } func (a *notifyAdmission) acquire(host string) bool { + host = canonicalKnotHost(host) a.mu.Lock() defer a.mu.Unlock() if a.total >= notifyMaxUnknown || a.held[host] >= notifyMaxUnknownPerKnot { @@ -111,6 +230,7 @@ func (a *notifyAdmission) acquire(host string) bool { } func (a *notifyAdmission) release(host string) { + host = canonicalKnotHost(host) a.mu.Lock() defer a.mu.Unlock() a.total-- @@ -160,20 +280,29 @@ func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid sy } } -// the budget for events that fetch from a knot: one global cap and one per knot +// the budget for events that fetch from a knot, shared fairly between the knots that have work func (s *Spindle) acquireNotifySlot(host string) (release func(), ok bool) { - if s.notifyInflight.Add(1) > notifyMaxInflight { - s.notifyInflight.Add(-1) + if !s.notifySlots.acquire(host, s.notifyLimits()) { return nil, false } - if !s.knotInflight.acquire(host) { - s.notifyInflight.Add(-1) - return nil, false + return func() { s.notifySlots.release(host) }, true +} + +func (s *Spindle) notifySlotCount() int { + if n := s.cfg.Server.NotifySlots; n > 0 { + return n + } + return defaultNotifySlots +} + +func (s *Spindle) notifyLimits() slotLimits { + pool := s.notifySlotCount() + dedicated := s.cfg.Server.PriorityNotifySlots + if dedicated <= 0 { + dedicated = pool / 4 } - return func() { - s.knotInflight.release(host) - s.notifyInflight.Add(-1) - }, true + s.notifyPriorityOnce.Do(func() { s.notifyPriorityHosts = s.priorityKnots() }) + return slotLimits{pool: pool, reserve: slotReserve(pool), dedicated: dedicated, priority: s.notifyPriorityHosts} } func (s *Spindle) runPush(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, push *org_tangled.EventPush) (pipeline syntax.ATURI, err error) { @@ -280,7 +409,7 @@ func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Re } l.Info("synced git repo") - changedFiles := changedFilesUnderBudget(ctx, l, repoPath, oldSha, newSha) + changedFiles, _ := changedFilesUnderBudget(ctx, l, repoPath, oldSha, newSha) triggerRepo, err := s.buildTriggerRepo(ctx, repo) if err != nil { @@ -361,16 +490,23 @@ func isMaterializedRef(refname string) bool { return strings.HasPrefix(refname, "refs/heads/") || strings.HasPrefix(refname, "refs/tags/") } -func changedFilesUnderBudget(ctx context.Context, l *slog.Logger, repoPath string, oldSha, newSha knotfeed.ObjectID) []string { +// known is false when the changed files can't be told, and the nil list then makes the compiler run +// the workflows a paths filter would have gated, because missing CI is worse than extra CI +func changedFilesUnderBudget(ctx context.Context, l *slog.Logger, repoPath string, oldSha, newSha knotfeed.ObjectID) (files []string, known bool) { if oldSha.IsZero() { - return nil + return nil, false } paths, err := changedFiles(ctx, repoPath, oldSha, newSha) if err != nil { l.Warn("changed files unavailable between revisions", "oldSha", oldSha, "newSha", newSha, "err", err) - return nil + return nil, false + } + admitted, ok := admitChangedFiles(paths) + if !ok { + l.Info("too many changed files to filter by path, running the workflows that have a paths filter", "oldSha", oldSha, "newSha", newSha, "files", len(paths)) + return nil, false } - return admitChangedFiles(paths) + return admitted, true } // the sync is shallow, so a missing oldSha gets fetched here, and go-git can't do sha256 @@ -390,15 +526,16 @@ func changedFiles(ctx context.Context, repoPath string, oldSha, newSha knotfeed. return strings.FieldsFunc(string(out), func(r rune) bool { return r == 0 }), nil } -func admitChangedFiles(paths []string) []string { +// false when the listing is over budget, an empty list is a push that changed nothing +func admitChangedFiles(paths []string) (admitted []string, ok bool) { spent := 0 - admitted := make([]string, 0, min(len(paths), changedFilesMaxCount)) + admitted = make([]string, 0, min(len(paths), changedFilesMaxCount)) for _, path := range paths { spent += changedFileOverheadBytes + len(path) if spent > changedFilesMaxBytes || len(admitted) == changedFilesMaxCount { - return nil + return nil, false } admitted = append(admitted, path) } - return admitted + return admitted, true } diff --git a/spindle/notify_event_test.go b/spindle/notify_event_test.go index 3193ef99e..f542aaa60 100644 --- a/spindle/notify_event_test.go +++ b/spindle/notify_event_test.go @@ -3,6 +3,7 @@ package spindle import ( "context" "errors" + "fmt" "io" "log/slog" "net/http" @@ -151,7 +152,6 @@ func (h *hookEnv) seedPushPipeline(t *testing.T, id, ref, oldSha, newSha string) if err := h.s.db.CreatePipeline(record); err != nil { t.Fatal(err) } - // a push that made a pipeline was announced when it was first handled if _, err := h.s.db.RecordPush(testRepoDid, ref, oldSha, newSha, time.Now()); err != nil { t.Fatal(err) } @@ -308,14 +308,22 @@ func TestNotifyEventAnswersAnErrorWhenTheWorkFails(t *testing.T) { h.expectPush(t) } -func TestNotifyEventRefusesPastTheInflightCap(t *testing.T) { +func TestNotifyEventRefusesPastThePool(t *testing.T) { h := newHookEnv(t) - h.s.notifyInflight.Store(notifyMaxInflight) + h.s.cfg.Server.NotifySlots = 3 + others := []string{"one.test", "two.test", "three.test"} + for _, host := range others { + if !h.s.notifySlots.acquire(host, h.s.notifyLimits()) { + t.Fatal("slot refused below the pool") + } + } _, err := h.s.NotifyEvent(context.Background(), h.k, testRepoDid, pushEvent("refs/heads/main", "skip-ci")) if !errors.Is(err, xrpc.ErrNotifyBusy) { t.Fatalf("err = %v, want ErrNotifyBusy", err) } - h.s.notifyInflight.Store(0) + for _, host := range others { + h.s.notifySlots.release(host) + } h.notify(t, pushEvent("refs/heads/main", "skip-ci")) h.expectPush(t) } @@ -369,29 +377,32 @@ func TestAdmitChangedFilesUnderBudget(t *testing.T) { return paths } - if got := admitChangedFiles([]string{"a.go", "b/c.go"}); !slices.Equal(got, []string{"a.go", "b/c.go"}) { - t.Errorf("small listing = %v, want both paths", got) + if got, ok := admitChangedFiles([]string{"a.go", "b/c.go"}); !ok || !slices.Equal(got, []string{"a.go", "b/c.go"}) { + t.Errorf("small listing = %v, %v, want both paths", got, ok) } - if got := admitChangedFiles(repeatedPaths(11000)); got != nil { - t.Errorf("listing over the entry cap = %v, want nil", got) + if got, ok := admitChangedFiles(repeatedPaths(11000)); ok || got != nil { + t.Errorf("listing over the entry cap = %v, %v, want nothing admitted", got, ok) } wide := repeatedPaths(8192) - if got := admitChangedFiles(wide); !slices.Equal(got, wide) { - t.Errorf("listing at the entry cap = %d paths, want all %d", len(got), len(wide)) + if got, ok := admitChangedFiles(wide); !ok || !slices.Equal(got, wide) { + t.Errorf("listing at the entry cap = %d paths, %v, want all %d", len(got), ok, len(wide)) + } + if got, ok := admitChangedFiles(nil); !ok || got == nil || len(got) != 0 { + t.Errorf("an empty listing = %v, %v, want an admitted list that is empty and not nil", got, ok) } big := make([]string, 11000) for i := range big { big[i] = string(make([]byte, 64)) } - if got := admitChangedFiles(big); got != nil { - t.Errorf("listing over the byte budget = %d paths, want nil", len(got)) + if got, ok := admitChangedFiles(big); ok || got != nil { + t.Errorf("listing over the byte budget = %d paths, %v, want nothing admitted", len(got), ok) } } func TestChangedFilesUnderBudgetGuardsFirstSight(t *testing.T) { quiet := slog.New(slog.NewTextHandler(io.Discard, nil)) - if got := changedFilesUnderBudget(context.Background(), quiet, "/nonexistent/repo", knotfeed.ObjectID{}, mustObjectID(testHexNew)); got != nil { - t.Errorf("first sight of a ref = %v, want nil without touching the repo", got) + if got, known := changedFilesUnderBudget(context.Background(), quiet, "/nonexistent/repo", knotfeed.ObjectID{}, mustObjectID(testHexNew)); known || got != nil { + t.Errorf("first sight of a ref = %v, %v, want unknown without touching the repo", got, known) } } @@ -422,7 +433,10 @@ func changedFilesInRealRepo(t *testing.T, format string) { if err := gitutil.SparseSync(context.Background(), repoPath, synced, shaSecond.String(), "/.tangled/workflows"); err != nil { t.Fatalf("sync: %v", err) } - got := changedFilesUnderBudget(context.Background(), quiet, synced, shaFirst, shaSecond) + got, known := changedFilesUnderBudget(context.Background(), quiet, synced, shaFirst, shaSecond) + if !known { + t.Error("shallow checkout changed files are unknown") + } slices.Sort(got) if want := []string{".tangled/workflows/ci.yaml", "src/app.go"}; !slices.Equal(got, want) { t.Errorf("shallow checkout changed files = %v, want %v", got, want) @@ -439,41 +453,19 @@ func changedFilesInRealRepo(t *testing.T, format string) { {"a no-op revision range lists nothing", shaThird, shaThird, nil}, } { t.Run(tc.name, func(t *testing.T) { - got := changedFilesUnderBudget(context.Background(), quiet, repoPath, tc.oldSha, tc.newSha) + got, known := changedFilesUnderBudget(context.Background(), quiet, repoPath, tc.oldSha, tc.newSha) slices.Sort(got) if !slices.Equal(got, tc.want) { t.Errorf("changed files = %v, want %v", got, tc.want) } + // a nil list reads as unknown, so an empty change set still has to be non-nil + if !known || got == nil { + t.Errorf("known = %v, list nil = %v, want a known list that is not nil", known, got == nil) + } }) } } -func TestNotifyEventRefusesPastTheKnotCap(t *testing.T) { - h := newHookEnv(t) - for range notifyMaxPerKnot { - if !h.s.knotInflight.acquire(h.k.Host) { - t.Fatal("slot refused below the cap") - } - } - _, err := h.s.NotifyEvent(context.Background(), h.k, testRepoDid, pushEvent("refs/heads/main", "skip-ci")) - if !errors.Is(err, xrpc.ErrNotifyBusy) { - t.Fatalf("err = %v, want ErrNotifyBusy for a knot at its cap", err) - } - if !h.s.knotInflight.acquire("other-knot.test") { - t.Error("one knot at its cap blocked another") - } - h.s.knotInflight.release("other-knot.test") - - for range notifyMaxPerKnot { - h.s.knotInflight.release(h.k.Host) - } - if len(h.s.knotInflight.held) != 0 { - t.Errorf("held = %v, want the map emptied at zero", h.s.knotInflight.held) - } - h.notify(t, pushEvent("refs/heads/main", "skip-ci")) - h.expectPush(t) -} - func TestNotifyEventDoesNotLetGitNearAnInternalKnot(t *testing.T) { h := newHookEnv(t) h.s.cfg.Server.Dev = false @@ -487,3 +479,61 @@ func TestNotifyEventDoesNotLetGitNearAnInternalKnot(t *testing.T) { t.Errorf("the knot saw %d requests, git should not have dialed a loopback host", got-before) } } + +func TestPathsGatedWorkflowsRunWhenTheChangedFilesCantBeTold(t *testing.T) { + gated := workflow.Workflow{ + Name: ".tangled/workflows/ci.yml", + Engine: "nixery", + When: []workflow.Constraint{{Event: []string{"push"}, Branch: []string{"main"}, Paths: []string{"src/**"}}}, + } + compile := func(changed []string) int { + c := workflow.Compiler{ + Trigger: tangled.Pipeline_TriggerMetadata{ + Kind: string(workflow.TriggerKindPush), + Push: &tangled.Pipeline_PushTriggerData{Ref: "refs/heads/main"}, + }, + ChangedFiles: changed, + } + return len(c.Compile([]workflow.Workflow{gated}).Workflows) + } + + small, ok := admitChangedFiles([]string{"docs/guide.md"}) + if !ok || compile(small) != 0 { + t.Errorf("a push that only touched docs ran the src workflow (known %v)", ok) + } + + over := make([]string, changedFilesMaxCount+1) + for i := range over { + over[i] = "docs/f.md" + } + tooMany, ok := admitChangedFiles(over) + if ok || compile(tooMany) != 1 { + t.Errorf("a push with %d changed files didn't run the paths-gated workflow (known %v)", len(over), ok) + } +} + +func TestOverBudgetPushIsLoggedWithItsFileCount(t *testing.T) { + repoPath := newSourceRepo(t, "sha1") + commit := func() knotfeed.ObjectID { + commitEverything(t, repoPath) + return mustObjectID(gitInDir(t, repoPath, "rev-parse", "HEAD")) + } + writeTestFile(t, filepath.Join(repoPath, "README.md"), "one\n") + before := commit() + for i := range changedFilesMaxCount + 1 { + writeTestFile(t, filepath.Join(repoPath, "gen", fmt.Sprintf("f%d.txt", i)), "x\n") + } + after := commit() + + var logged strings.Builder + l := slog.New(slog.NewTextHandler(&logged, nil)) + got, known := changedFilesUnderBudget(context.Background(), l, repoPath, before, after) + if known || got != nil { + t.Fatalf("changed files = %d, known %v, want unknown", len(got), known) + } + for _, want := range []string{before.String(), after.String(), fmt.Sprintf("files=%d", changedFilesMaxCount+1)} { + if !strings.Contains(logged.String(), want) { + t.Errorf("log %q lacks %q", logged.String(), want) + } + } +} diff --git a/spindle/notify_slots_test.go b/spindle/notify_slots_test.go new file mode 100644 index 000000000..18e41e2d7 --- /dev/null +++ b/spindle/notify_slots_test.go @@ -0,0 +1,277 @@ +package spindle + +import ( + "testing" + "time" +) + +type slotClock struct{ at time.Time } + +func (c *slotClock) now() time.Time { return c.at } + +func newSlots() (*notifySlots, *slotClock) { + clock := &slotClock{at: time.Unix(1000, 0)} + return ¬ifySlots{now: clock.now}, clock +} + +func shared(pool int) slotLimits { return slotLimits{pool: pool, reserve: slotReserve(pool)} } + +func take(n *notifySlots, host string, lim slotLimits, count int) (got int) { + for range count { + if n.acquire(host, lim) { + got++ + } + } + return got +} + +func fill(t *testing.T, n *notifySlots, lim slotLimits) { + t.Helper() + if got := take(n, "hog.test", lim, lim.pool); got != lim.pool-lim.reserve { + t.Fatalf("setup: the hog got %d of %d, want all but the reserve %d", got, lim.pool, lim.reserve) + } + if got := take(n, "second.test", lim, lim.reserve); got != lim.reserve { + t.Fatalf("setup: the second knot got %d of the reserve of %d", got, lim.reserve) + } +} + +func TestNotifySlotsAnIdlePoolLetsOneKnotPastTheOldCap(t *testing.T) { + n, _ := newSlots() + lim := shared(16) + + if got := take(n, "a.test", lim, 16); got != 14 { + t.Fatalf("one knot got %d of 16 slots while nobody else waited, want all but the reserve of 2", got) + } +} + +func TestNotifySlotsAnIdlePoolKeepsAReserveForANewcomer(t *testing.T) { + for pool, reserve := range map[int]int{64: 8, 16: 2, 8: 2, 3: 2} { + n, _ := newSlots() + lim := shared(pool) + if lim.reserve != reserve { + t.Fatalf("reserve of a pool of %d = %d, want %d", pool, lim.reserve, reserve) + } + got := take(n, "hog.test", lim, pool) + if want := max(1, pool-reserve); got != want { + t.Errorf("pool of %d: the hog got %d, want %d", pool, got, want) + } + if !n.acquire("newcomer.test", lim) { + t.Errorf("pool of %d: a newcomer found no slot on an idle pool", pool) + } + } +} + +func TestNotifySlotsKnotsThatWaitGetAnEqualShare(t *testing.T) { + n, _ := newSlots() + lim := shared(16) + fill(t, n, lim) + take(n, "b.test", lim, 1) + take(n, "c.test", lim, 1) + + for range 4 { + n.release("hog.test") + } + if n.acquire("hog.test", lim) { + t.Fatal("the hog took a freed slot back while two knots waited") + } + for _, host := range []string{"b.test", "c.test", "b.test", "c.test"} { + if !n.acquire(host, lim) { + t.Fatalf("%s was refused a slot inside its share", host) + } + } + if n.held["b.test"] != 2 || n.held["c.test"] != 2 { + t.Fatalf("held = %v, want the freed slots split between the knots that waited", n.held) + } +} + +func TestNotifySlotsAHostileKnotCantGrowPastItsShareWhileAnotherHasWork(t *testing.T) { + n, _ := newSlots() + lim := shared(16) + fill(t, n, lim) + if n.acquire("honest.test", lim) { + t.Fatal("the pool was full, so the honest knot should wait") + } + + n.release("hog.test") + if n.acquire("hog.test", lim) { + t.Fatalf("the hostile knot took a freed slot back while holding %d and another knot waited", n.held["hog.test"]) + } + if !n.acquire("honest.test", lim) { + t.Fatal("the slot the hostile knot let go of went to nobody") + } +} + +func TestNotifySlotsAKnotThatStopsWaitingStopsCounting(t *testing.T) { + n, clock := newSlots() + lim := shared(8) + fill(t, n, lim) + n.acquire("b.test", lim) + + clock.at = clock.at.Add(2 * notifyWaitWindow) + n.release("second.test") + n.release("second.test") + n.release("hog.test") + n.release("hog.test") + if got := take(n, "hog.test", lim, 2); got != 2 { + t.Fatalf("the hog got %d slots after b stopped waiting, want both", got) + } +} + +func TestNotifySlotsAKnotBackingOffStillLimitsTheHog(t *testing.T) { + n, clock := newSlots() + lim := shared(16) + fill(t, n, lim) + if n.acquire("legit.test", lim) { + t.Fatal("the pool was full, so the legit knot should wait") + } + + clock.at = clock.at.Add(30 * time.Second) + for range 6 { + n.release("hog.test") + } + if n.acquire("hog.test", lim) { + t.Fatalf("the hog took a slot back at %d held while the legit knot was only backing off", n.held["hog.test"]) + } + clock.at = clock.at.Add(30 * time.Second) + if !n.acquire("legit.test", lim) { + t.Fatal("the legit knot found no slot when its retry came") + } +} + +func TestNotifySlotsAPriorityKnotGetsInWhileAHostileKnotPinsThePool(t *testing.T) { + n, _ := newSlots() + lim := shared(16) + lim.dedicated = 4 + lim.priority = priorityKnotHosts([]string{"prio.test"}) + fill(t, n, lim) + + if got := take(n, "prio.test", lim, 4); got != 4 { + t.Fatalf("the priority knot got %d of its 4 dedicated slots with the shared pool full", got) + } + if n.acquire("prio.test", lim) { + t.Fatal("a fifth slot came from a full shared pool") + } + if n.prioTotal != 4 || n.total != 16 { + t.Fatalf("dedicated %d, shared %d, want 4 and 16", n.prioTotal, n.total) + } +} + +func TestNotifySlotsANonPriorityKnotCantUseTheDedicatedSlots(t *testing.T) { + n, _ := newSlots() + lim := shared(16) + lim.dedicated = 4 + lim.priority = priorityKnotHosts([]string{"prio.test"}) + fill(t, n, lim) + + if n.acquire("plain.test", lim) { + t.Fatal("a knot that isn't priority took a slot while the shared pool was full") + } + if n.prioTotal != 0 { + t.Fatalf("%d dedicated slots in use, want none", n.prioTotal) + } +} + +func TestNotifySlotsAPriorityKnotFallsBackToTheSharedPoolUnderTheNormalShare(t *testing.T) { + n, _ := newSlots() + lim := shared(16) + lim.dedicated = 2 + lim.priority = priorityKnotHosts([]string{"prio.test"}) + + if got := take(n, "prio.test", lim, 20); got != 2+14 { + t.Fatalf("the priority knot got %d, want its 2 dedicated and all but the reserve of the shared pool", got) + } + // shared slots go back first, the other knots are waiting on them + n.release("prio.test") + if n.held["prio.test"] != 13 || n.prioTotal != 2 { + t.Fatalf("shared held %d, dedicated %d after one release, want 13 and 2", n.held["prio.test"], n.prioTotal) + } +} + +func TestPriorityKnotsMatchByHostOrDid(t *testing.T) { + hosts := priorityKnotHosts([]string{"Knot1.Example.com", "did:web:knot2.example.com", " did:web:Knot3.Example.com%3A8443 ", ""}) + for _, host := range []string{"knot1.example.com", "KNOT1.example.com.", "knot2.example.com", "Knot3.example.com:8443"} { + if !hosts[canonicalKnotHost(host)] { + t.Errorf("%s isn't a priority knot, want it matched by host and by did", host) + } + } + if hosts[canonicalKnotHost("knot3.example.com")] || hosts[canonicalKnotHost("other.example.com")] || len(hosts) != 3 { + t.Fatalf("hosts = %v, want three, and the port to count", hosts) + } +} + +func TestNotifySlotsAreSizedFromTheConfig(t *testing.T) { + h := newHookEnv(t) + if got := h.s.notifySlotCount(); got != defaultNotifySlots { + t.Fatalf("pool = %d with nothing configured, want %d", got, defaultNotifySlots) + } + h.s.cfg.Server.NotifySlots = 5 + if got := h.s.notifySlotCount(); got != 5 { + t.Fatalf("pool = %d, want the configured 5", got) + } + var release []func() + for _, host := range []string{"a.test", "b.test", "c.test", "d.test", "e.test"} { + r, ok := h.s.acquireNotifySlot(host) + if !ok { + t.Fatalf("slot refused for %s below the configured pool", host) + } + release = append(release, r) + } + if _, ok := h.s.acquireNotifySlot("f.test"); ok { + t.Fatal("a sixth slot was granted from a pool of 5") + } + for _, r := range release { + r() + } + if got := h.s.notifySlots.inflight(); got != 0 { + t.Fatalf("%d slots held after releasing all", got) + } +} + +func TestPriorityNotifySlotsAreSizedFromTheConfig(t *testing.T) { + h := newHookEnv(t) + h.s.cfg.Server.PriorityKnots = []string{"did:web:Prio.Test"} + + lim := h.s.notifyLimits() + if lim.dedicated != defaultNotifySlots/4 || !lim.priority["prio.test"] { + t.Fatalf("limits = %+v, want a quarter of the pool dedicated and prio.test priority", lim) + } + h.s.cfg.Server.PriorityNotifySlots = 3 + if got := h.s.notifyLimits().dedicated; got != 3 { + t.Fatalf("dedicated = %d, want the configured 3", got) + } +} + +func TestNotifyCapsKeyOnOneSpellingOfAKnotHost(t *testing.T) { + n, _ := newSlots() + for _, host := range []string{"Knot.Test", "knot.test.", "KNOT.test"} { + if !n.acquire(host, shared(8)) { + t.Fatalf("slot refused for %s below the pool", host) + } + } + if len(n.held) != 1 || n.held["knot.test"] != 3 { + t.Fatalf("held = %v, want one knot holding 3", n.held) + } + for _, host := range []string{"knot.test", "KNOT.test.", "Knot.Test"} { + n.release(host) + } + if len(n.held) != 0 || n.inflight() != 0 { + t.Fatalf("held = %v after releasing under other spellings", n.held) + } + + var unknown notifyAdmission + if !unknown.acquire("Knot.Test") || unknown.acquire("knot.test.") { + t.Fatal("the unknown-repo pool keyed one knot under two spellings") + } +} + +func TestCanonicalKnotHostKeepsThePort(t *testing.T) { + for raw, want := range map[string]string{ + "Knot.Test:8443": "knot.test:8443", + "knot.test.": "knot.test", + "knot.test": "knot.test", + } { + if got := canonicalKnotHost(raw); got != want { + t.Errorf("canonicalKnotHost(%q) = %q, want %q", raw, got, want) + } + } +} diff --git a/spindle/pull_backlog_test.go b/spindle/pull_backlog_test.go new file mode 100644 index 000000000..58e053bb2 --- /dev/null +++ b/spindle/pull_backlog_test.go @@ -0,0 +1,61 @@ +package spindle + +import ( + "fmt" + "testing" + "time" + + "tangled.org/core/spindle/db" +) + +func TestAPullEventBacklogDrainsInBatches(t *testing.T) { + e := newPullEnv(t) + d := e.n.s.db + if err := d.AddPullTicket(testRepoDid, "bulk", db.PullTicket{Author: testBob, State: "open", Version: "v0"}); err != nil { + t.Fatal(err) + } + const events = 3 * pullEventBatch + for i := 1; i <= events; i++ { + ev := db.PullEvent{RepoDid: testRepoDid, Ticket: "bulk", Event: "closed", Author: testBob, Sender: testBob, Pull: []byte("{}")} + if err := d.AdvancePullTicket(testRepoDid, "bulk", fmt.Sprintf("v%d", i-1), fmt.Sprintf("v%d", i), "closed", &ev); err != nil { + t.Fatal(err) + } + } + + e.n.s.retries.jobs.kick(testRepoDid) + eventually(t, 10*time.Second, "the backlog to be worked through, a batch at a time", func() bool { + due, err := d.DuePullEvents(testRepoDid, time.Now().Add(time.Hour), events+1) + return err == nil && len(due) == 0 + }) +} + +func TestPullEventsStayDueWhileThePoolIsFull(t *testing.T) { + e := newPullEnv(t) + d := e.n.s.db + e.n.s.cfg.Server.NotifySlots = 1 + repo, err := e.n.s.servedRepo(testRepoDid) + if err != nil || repo == nil { + t.Fatalf("setup: the repo isn't served: %v", err) + } + if !e.n.s.notifySlots.acquire("other-knot.test", e.n.s.notifyLimits()) { + t.Fatal("setup: the pool was already full") + } + if err := d.AddPullTicket(testRepoDid, "held", db.PullTicket{Author: testBob, State: "open", Version: "v0"}); err != nil { + t.Fatal(err) + } + ev := db.PullEvent{RepoDid: testRepoDid, Ticket: "held", Event: "closed", Author: testBob, Sender: testBob, Pull: []byte("{}")} + if err := d.AdvancePullTicket(testRepoDid, "held", "v0", "v1", "closed", &ev); err != nil { + t.Fatal(err) + } + + e.n.s.runDuePullEvents(e.n.s.rootCtx, testRepoDid) + due, err := d.DuePullEvents(testRepoDid, time.Now().Add(time.Hour), 10) + if err != nil || len(due) != 1 { + t.Fatalf("%d events due, %v, want the one that found no slot", len(due), err) + } + e.n.s.notifySlots.release("other-knot.test") + e.n.s.runDuePullEvents(e.n.s.rootCtx, testRepoDid) + if due, _ := d.DuePullEvents(testRepoDid, time.Now().Add(time.Hour), 10); len(due) != 0 { + t.Fatalf("%d events left after the slot freed", len(due)) + } +} diff --git a/spindle/pull_versions_test.go b/spindle/pull_versions_test.go index c581a052e..543c0e4cc 100644 --- a/spindle/pull_versions_test.go +++ b/spindle/pull_versions_test.go @@ -37,7 +37,7 @@ func TestEveryStateChangeBetweenTwoSyncsFiresInOrder(t *testing.T) { } events := func() []string { t.Helper() - due, err := env.s.db.DuePullEvents(testRepoDid, time.Now()) + due, err := env.s.db.DuePullEvents(testRepoDid, time.Now(), 100) if err != nil { t.Fatal(err) } @@ -51,7 +51,6 @@ func TestEveryStateChangeBetweenTwoSyncsFiresInOrder(t *testing.T) { put("open", testBob) sync() - // closed and reopened while the repo's job was stuck, so the newest state is the one spindle already has put("closed", testCarol) put("open", testBob) sync() @@ -98,7 +97,7 @@ func TestAFailedVersionListDoesNotSkipTheStatesInBetween(t *testing.T) { if err := sync(); err != nil { t.Fatal(err) } - due, err := env.s.db.DuePullEvents(testRepoDid, time.Now()) + due, err := env.s.db.DuePullEvents(testRepoDid, time.Now(), 100) if err != nil || len(due) != 2 { t.Fatalf("pull events = %d, %v, want the closed and the reopened that the failed sync must not skip", len(due), err) } diff --git a/spindle/pulls.go b/spindle/pulls.go index adcc86375..639f8ba0b 100644 --- a/spindle/pulls.go +++ b/spindle/pulls.go @@ -21,6 +21,9 @@ import ( "tangled.org/core/workflow" ) +// read in batches so a big backlog never loads in one go +const pullEventBatch = 16 + // a pull request is a ticket with a target branch plus the patches that name it type knotTicket struct { org_tangled.TrackTicket @@ -75,7 +78,7 @@ func (s *Spindle) applyPullRecord(ctx context.Context, rec *tapc.RecordEventData if err != nil || repo == nil { return err } - // a mark that lands while this applies bumps the epoch, so its row has to outlive this + // a mark that lands during apply bumps the epoch, so the clear leaves its row alone epoch, err := s.db.RecordRetryEpoch(retry) if err != nil { return err @@ -143,8 +146,7 @@ func (s *Spindle) applyPatch(ctx context.Context, repo *db.Repo, rec *tapc.Recor return s.whileServed(repo, func() error { return s.db.AddPullPatch(repo.RepoDid, stored, ev) }) } -// pull handlers read from the knot with no lock, the repo job already runs one at a time, so only -// the write takes the repo lock and a wipe that landed meanwhile means the rows must not come back +// the served check and the write share the repo lock, so a wipe can't let rows come back func (s *Spindle) whileServed(repo *db.Repo, write func() error) error { defer s.lockRepo(repo.RepoDid).Unlock() if served, err := s.servedRepo(repo.RepoDid); err != nil || served == nil { @@ -199,20 +201,17 @@ func (s *Spindle) syncPullTicket(ctx context.Context, k knotclient.Knot, repo *d return &ticket, row, nil } -// how many versions past the stored one a single sync will replay, so a ticket edited a -// thousand times costs a bounded walk +// a sync replays at most this many of the newest versions, so a heavily edited ticket stays bounded const maxPullVersionWalk = 64 -// events only say a ticket changed, so a state that went closed and back before a sync would -// look like nothing happened. the versions say what happened, in order +// events don't say what changed, so a close and reopen between syncs needs the versions func (s *Spindle) walkPullTicket(ctx context.Context, k knotclient.Knot, repo *db.Repo, rkey string, current *knotTicket, row *db.PullTicket) error { versions, err := s.knots.ListRecordVersions(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey) if errors.Is(err, knotclient.ErrNotFound) { - // no history to walk, so the newest state is all there is to compare return s.applyPullState(repo, rkey, current, row, "", current.Editor) } if err != nil { - // comparing only the newest state would skip the transitions in between for good + // falling back to the newest state would skip the middle transitions for good return fmt.Errorf("listing the versions of ticket %s/%s: %w", repo.RepoDid, rkey, err) } newest := len(versions) - 1 @@ -248,7 +247,7 @@ func (s *Spindle) walkPullTicket(ctx context.Context, k knotclient.Knot, repo *d return nil } -// version is empty when there is no history, then only the state moves +// an empty version means no history, so there is no version to record func (s *Spindle) applyPullState(repo *db.Repo, rkey string, at *knotTicket, row *db.PullTicket, version string, sender syntax.DID) error { state := *at.State event, moved := pullTransition(row.State, state) @@ -328,15 +327,14 @@ func newPullEvent(repo *db.Repo, rkey string, event models.WebhookEvent, ticket }, nil } -// the repo's job runs its pull events, and a job per repo at a time is what keeps two runners off one event +// one job per repo at a time keeps two runners off the same event func (s *Spindle) startPullEvents(repo *db.Repo) { s.retries.jobs.kick(repo.RepoDid) } -// runs in the repo job, after its record marks. a full notify budget leaves the rows due, they're -// durable and the ticker kicks the repo again +// a full notify budget leaves the rows due for the next pass func (s *Spindle) runDuePullEvents(ctx context.Context, did syntax.DID) { - events, err := s.db.DuePullEvents(did, s.retries.clock()) + events, err := s.db.DuePullEvents(did, s.retries.clock(), pullEventBatch) if err != nil { s.l.Error("reading the pull events", "repo", did, "err", err) return @@ -349,35 +347,40 @@ func (s *Spindle) runDuePullEvents(ctx context.Context, did syntax.DID) { s.l.Error("looking up a repo with pull events", "repo", did, "err", err) return } - // a repo this spindle stopped serving runs nothing, so its events would wait forever + // events for an unserved repo would sit due forever, so they get dropped if repo == nil { if err := s.db.DeletePullEventsByRepo(did); err != nil { s.l.Error("dropping the pull events of a repo no longer served", "repo", did, "err", err) } return } - release, ok := s.acquireNotifySlot(repo.Knot) - if !ok { - return + if s.runPullEvents(ctx, repo, events) && len(events) == pullEventBatch { + // back of the queue, so one long backlog can't hog a worker + s.retries.jobs.kick(did) } - defer release() - s.runPullEvents(ctx, repo, events) } -func (s *Spindle) runPullEvents(ctx context.Context, repo *db.Repo, events []db.PullEvent) { +// takes a notify slot per event, so a backlog doesn't pin the pool +func (s *Spindle) runPullEvents(ctx context.Context, repo *db.Repo, events []db.PullEvent) bool { for _, ev := range events { - // the budget is per event, so only the outer ctx ending (shutdown) leaves a row without a failure + release, ok := s.acquireNotifySlot(repo.Knot) + if !ok { + return false + } + // only a shutdown leaves the row untouched, a budget timeout counts as a failure work, cancel := context.WithTimeout(ctx, notifyWorkBudget) err := s.runPullEvent(work, repo, ev) cancel() + release() if err != nil && ctx.Err() == nil { s.l.Warn("a pull request event failed", "repo", repo.RepoDid, "pull", ev.Ticket, "event", ev.Event, "err", err) s.failPullEvent(ev) } } + return true } -// a banned user's pull request makes nothing, no webhook and no pipeline, whether they opened it or acted on it +// a banned author or sender means no webhook and no pipeline func (s *Spindle) bannedInPull(ev db.PullEvent) (bool, error) { for _, did := range []syntax.DID{ev.Author, ev.Sender} { if ban, err := s.db.GetBan(did); err != nil || ban != nil { diff --git a/spindle/repo_job_limits_test.go b/spindle/repo_job_limits_test.go index 4d765e5e8..d4b210d16 100644 --- a/spindle/repo_job_limits_test.go +++ b/spindle/repo_job_limits_test.go @@ -73,14 +73,16 @@ func TestARepoJobWorksThroughALimitedNumberOfMarksThenYieldsItsWorker(t *testing } } -// a pool of the given size whose repos belong to the knots given, for the fair share tests -func fairJobs(pool int, knots map[syntax.DID]string) *repoJobs { +func fairJobs(pool int, knots map[syntax.DID]string, priority ...string) *repoJobs { j := &repoJobs{} - j.configure(pool, func(did syntax.DID) string { return knots[did] }) + set := map[string]bool{} + for _, knot := range priority { + set[knot] = true + } + j.configure(pool, func(did syntax.DID) string { return knots[did] }, set) return j } -// n repos on one knot, named so they sort in the order they were kicked func reposOn(knot string, n int, knots map[syntax.DID]string) []syntax.DID { var dids []syntax.DID for i := range n { @@ -91,7 +93,6 @@ func reposOn(knot string, n int, knots map[syntax.DID]string) []syntax.DID { return dids } -// takes jobs without finishing any, until none is allowed to start func startAll(t *testing.T, j *repoJobs, knots map[syntax.DID]string) (started []syntax.DID, perKnot map[string]int) { t.Helper() perKnot = map[string]int{} @@ -107,14 +108,33 @@ func startAll(t *testing.T, j *repoJobs, knots map[syntax.DID]string) (started [ } } -func TestAnIdlePoolLetsOneKnotRunEveryWorker(t *testing.T) { +func TestOneKnotOnAnIdlePoolLeavesTheReserveForANewcomer(t *testing.T) { knots := map[syntax.DID]string{} - j := fairJobs(8, knots) - for _, did := range reposOn("a", 12, knots) { + j := fairJobs(16, knots) + for _, did := range reposOn("a", 30, knots) { + j.kick(did) + } + if started, _ := startAll(t, j, knots); len(started) != 14 { + t.Fatalf("%d jobs started on a pool of 16 with only one knot waiting, want all but the reserve of 2", len(started)) + } + + newcomer := reposOn("b", 1, knots)[0] + j.kick(newcomer) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if did, ok := j.next(ctx); !ok || did != newcomer { + t.Fatalf("next = %q, %v, want a new knot to get a worker at once while another is busy", did, ok) + } +} + +func TestTheReserveIsAnEighthOfALargePool(t *testing.T) { + knots := map[syntax.DID]string{} + j := fairJobs(64, knots) + for _, did := range reposOn("a", 80, knots) { j.kick(did) } - if started, _ := startAll(t, j, knots); len(started) != 8 { - t.Fatalf("%d jobs started on a pool of 8 with only one knot waiting, want all 8", len(started)) + if started, _ := startAll(t, j, knots); len(started) != 56 { + t.Fatalf("%d jobs started on a pool of 64, want 56 with an eighth kept in reserve", len(started)) } } @@ -142,7 +162,6 @@ func TestASlowKnotWithManyReposCantHoldMoreThanItsShareWhileAnotherKnotWaits(t * healthy := reposOn("healthy", 1, knots)[0] j.kick(healthy) - // the first five that start: the hostile knot is capped at its share and the healthy one is next var firstFive []syntax.DID var hostile int for range 5 { @@ -206,3 +225,78 @@ func TestTheRepoJobWorkerCountComesFromTheConfig(t *testing.T) { t.Fatalf("workers = %d, want the configured 7", got) } } + +func TestAPriorityKnotsJobRunsAtOnceWhileAHostileKnotHoldsTheSharedPool(t *testing.T) { + knots := map[syntax.DID]string{} + j := fairJobs(8, knots, "priority.example.com") + for _, did := range reposOn("hostile", 30, knots) { + j.kick(did) + } + if started, _ := startAll(t, j, knots); len(started) != 6 { + t.Fatalf("%d hostile jobs started, want them held to the shared pool less its reserve", len(started)) + } + + vip := reposOn("priority.example.com", 1, knots)[0] + j.kick(vip) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if did, ok := j.nextPriority(ctx); !ok || did != vip { + t.Fatalf("priority worker got %q, %v, want the priority knot's job at once", did, ok) + } +} + +func TestANonPriorityKnotCantUseThePriorityWorkers(t *testing.T) { + knots := map[syntax.DID]string{} + j := fairJobs(8, knots, "priority.example.com") + for _, did := range reposOn("other", 5, knots) { + j.kick(did) + } + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + if did, ok := j.nextPriority(ctx); ok { + t.Fatalf("a priority worker took %q of a knot that isn't a priority one", did) + } +} + +func TestAPriorityKnotsJobCanAlsoUseASharedWorker(t *testing.T) { + knots := map[syntax.DID]string{} + j := fairJobs(8, knots, "priority.example.com") + vip := reposOn("priority.example.com", 1, knots)[0] + j.kick(vip) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if did, ok := j.next(ctx); !ok || did != vip { + t.Fatalf("shared worker got %q, %v, want it to take a priority knot's job under the usual rules", did, ok) + } +} + +func TestPriorityKnotsMatchAsDidsAndAsHosts(t *testing.T) { + env := newEnsureEnv(t) + env.s.cfg.Server.PriorityKnots = []string{"did:web:knot1.tangled.sh", " KNOT2.example.com ", "did:web:knot3.example.com%3A8443"} + got := env.s.priorityKnots() + for _, want := range []string{"knot1.tangled.sh", "knot2.example.com", "knot3.example.com:8443"} { + if !got[want] { + t.Fatalf("priority knots = %v, want %s to match however it was written", got, want) + } + } + byDid, byHost := priorityKnotHosts([]string{"did:web:knot1.tangled.sh"}), priorityKnotHosts([]string{"knot1.tangled.sh"}) + if len(byDid) != 1 || !byDid["knot1.tangled.sh"] || !byHost["knot1.tangled.sh"] { + t.Fatalf("a knot's did gave %v and its host %v, want the same key", byDid, byHost) + } +} + +func TestPriorityWorkersOnlyExistWithPriorityKnotsAndDefaultToAQuarter(t *testing.T) { + env := newEnsureEnv(t) + env.s.cfg.Server.RepoJobWorkers = 64 + if got := env.s.priorityJobWorkerCount(); got != 0 { + t.Fatalf("priority workers = %d with no priority knots, want none", got) + } + env.s.cfg.Server.PriorityKnots = []string{"knot1.tangled.sh"} + if got := env.s.priorityJobWorkerCount(); got != 16 { + t.Fatalf("priority workers = %d, want a quarter of 64", got) + } + env.s.cfg.Server.PriorityRepoJobWorkers = 5 + if got := env.s.priorityJobWorkerCount(); got != 5 { + t.Fatalf("priority workers = %d, want the configured 5", got) + } +} diff --git a/spindle/repo_jobs.go b/spindle/repo_jobs.go index b59209758..8970ecc1d 100644 --- a/spindle/repo_jobs.go +++ b/spindle/repo_jobs.go @@ -16,9 +16,8 @@ import ( ) const ( - // when the config doesn't say, which is only a test config - defaultRepoJobWorkers = 64 - // what one repo may have marked at once, a repo that isn't served has only a pending refresh to go on + defaultRepoJobWorkers = 256 + // unserved repos can only have a refresh pending, which keeps the cap low recordMarksPerRepo = 1000 recordMarksPerUnservedRepo = 20 ) @@ -28,7 +27,6 @@ type jobState uint8 const ( jobQueued jobState = iota + 1 jobRunning - // kicked while running, so it goes once more when it finishes jobAgain ) @@ -37,41 +35,47 @@ type queuedJob struct { knot string } -// one job per repo at a time, so kicks for a repo that is queued or running pile into one more run -// instead of a queue of them, and a hung repo only ever holds its own worker. -// -// the pool is shared fairly between knots, but only when that matters: a knot may use every worker -// while no other knot has jobs waiting, and once others do each one gets an equal slice, so a knot -// that is slow or hostile can't starve the rest and a big healthy one isn't throttled on an idle pool +// one job per repo, a kick during a run folds into one more run +// knots share the pool, and a reserve stays back so new knots can start type repoJobs struct { mu sync.Mutex state map[syntax.DID]jobState queue []queuedJob - wake chan struct{} + // two channels so one kind of worker can't eat the other's wakeup + wake chan struct{} + wakePriority chan struct{} - // both are set by configure, without them every job counts as one knot's and nothing is limited - pool int - knotOf func(syntax.DID) string + // zero until configure runs, and zero means no limits + pool int + knotOf func(syntax.DID) string + priority map[string]bool queued map[string]int running map[string]int - // the knot each running job was counted under, so finishing gives back the same one - runningKnot map[syntax.DID]string + // the slot goes back under the knot it was taken from, not whatever the repo says now + runningKnot map[syntax.DID]runningJob } -func (j *repoJobs) configure(pool int, knotOf func(syntax.DID) string) { +type runningJob struct { + knot string + shared bool +} + +// priority is keyed by knot, the same string knotOf returns +func (j *repoJobs) configure(pool int, knotOf func(syntax.DID) string, priority map[string]bool) { j.mu.Lock() defer j.mu.Unlock() - j.pool, j.knotOf = pool, knotOf + j.pool, j.knotOf, j.priority = pool, knotOf, priority } func (j *repoJobs) wakeLocked() chan struct{} { if j.wake == nil { j.wake = make(chan struct{}, 1) + j.wakePriority = make(chan struct{}, 1) j.state = map[syntax.DID]jobState{} j.queued = map[string]int{} j.running = map[string]int{} - j.runningKnot = map[syntax.DID]string{} + j.runningKnot = map[syntax.DID]runningJob{} } return j.wake } @@ -83,6 +87,11 @@ func nudge(wake chan struct{}) { } } +func (j *repoJobs) nudgeAllLocked() { + nudge(j.wake) + nudge(j.wakePriority) +} + func (j *repoJobs) enqueueLocked(did syntax.DID, knot string) { j.state[did] = jobQueued j.queue = append(j.queue, queuedJob{did, knot}) @@ -91,7 +100,7 @@ func (j *repoJobs) enqueueLocked(did syntax.DID, knot string) { func (j *repoJobs) kick(did syntax.DID) { j.mu.Lock() - wake := j.wakeLocked() + j.wakeLocked() switch j.state[did] { case jobRunning: j.state[did] = jobAgain @@ -103,7 +112,7 @@ func (j *repoJobs) kick(did syntax.DID) { knotOf := j.knotOf j.mu.Unlock() - // the knot comes from the db, which isn't read under the lock + // knotOf hits the db, call it outside the lock knot := "" if knotOf != nil { knot = knotOf(did) @@ -113,14 +122,12 @@ func (j *repoJobs) kick(did syntax.DID) { switch j.state[did] { case 0: j.enqueueLocked(did, knot) - nudge(wake) + j.nudgeAllLocked() case jobRunning: j.state[did] = jobAgain } } -// how many jobs a knot may run now: the whole pool unless other knots have jobs waiting, then an equal -// slice of it. repos that aren't served yet share a smaller one inside that func (j *repoJobs) limitLocked(knot string) int { limit := j.pool if others := len(j.queued) - 1; others > 0 { @@ -129,12 +136,17 @@ func (j *repoJobs) limitLocked(knot string) int { if knot == "" { limit = max(2, limit/2) } - return limit + return min(limit, max(1, j.pool-max(2, j.pool/8))) } -func (j *repoJobs) pickLocked() int { +func (j *repoJobs) pickLocked(dedicated bool) int { for i, q := range j.queue { - if j.pool == 0 || j.running[q.knot] < j.limitLocked(q.knot) { + switch { + case dedicated: + if j.priority[q.knot] { + return i + } + case j.pool == 0 || j.running[q.knot] < j.limitLocked(q.knot): return i } } @@ -142,20 +154,34 @@ func (j *repoJobs) pickLocked() int { } func (j *repoJobs) next(ctx context.Context) (syntax.DID, bool) { + return j.take(ctx, false) +} + +func (j *repoJobs) nextPriority(ctx context.Context) (syntax.DID, bool) { + return j.take(ctx, true) +} + +func (j *repoJobs) take(ctx context.Context, dedicated bool) (syntax.DID, bool) { for { j.mu.Lock() - wake := j.wakeLocked() - if i := j.pickLocked(); i >= 0 { + j.wakeLocked() + wake := j.wake + if dedicated { + wake = j.wakePriority + } + if i := j.pickLocked(dedicated); i >= 0 { q := j.queue[i] j.queue = slices.Delete(j.queue, i, i+1) if j.queued[q.knot]--; j.queued[q.knot] <= 0 { delete(j.queued, q.knot) } j.state[q.did] = jobRunning - j.running[q.knot]++ - j.runningKnot[q.did] = q.knot + if !dedicated { + j.running[q.knot]++ + } + j.runningKnot[q.did] = runningJob{q.knot, !dedicated} if len(j.queue) > 0 { - nudge(wake) + j.nudgeAllLocked() } j.mu.Unlock() return q.did, true @@ -172,28 +198,37 @@ func (j *repoJobs) next(ctx context.Context) (syntax.DID, bool) { func (j *repoJobs) finish(did syntax.DID) { j.mu.Lock() defer j.mu.Unlock() - wake := j.wakeLocked() - knot := j.runningKnot[did] + j.wakeLocked() + ran := j.runningKnot[did] delete(j.runningKnot, did) - if j.running[knot]--; j.running[knot] <= 0 { - delete(j.running, knot) + if ran.shared { + if j.running[ran.knot]--; j.running[ran.knot] <= 0 { + delete(j.running, ran.knot) + } } - // a slot is free, a job that was waiting for one may be able to go now - defer nudge(wake) + defer j.nudgeAllLocked() if j.state[did] != jobAgain { delete(j.state, did) return } - j.enqueueLocked(did, knot) + j.enqueueLocked(did, ran.knot) } func (j *repoJobs) work(ctx context.Context, run func(context.Context, syntax.DID)) { + j.workWith(ctx, run, j.next) +} + +func (j *repoJobs) workPriority(ctx context.Context, run func(context.Context, syntax.DID)) { + j.workWith(ctx, run, j.nextPriority) +} + +func (j *repoJobs) workWith(ctx context.Context, run func(context.Context, syntax.DID), next func(context.Context) (syntax.DID, bool)) { for { - did, ok := j.next(ctx) + did, ok := next(ctx) if !ok { return } - // a job that panics must not take its worker or the process with it, its marks stay for the next run + // a panic can't kill the worker, and the marks are still there for the next run _ = recovered("repo job", func() error { run(ctx, did); return nil }) j.finish(did) } @@ -201,7 +236,6 @@ func (j *repoJobs) work(ctx context.Context, run func(context.Context, syntax.DI type markOnlyKey struct{} -// handlers under it only write the repo's mark, a job does the knot and pds reads later func withMarkOnly(ctx context.Context) context.Context { return context.WithValue(ctx, markOnlyKey{}, true) } @@ -210,7 +244,7 @@ func markOnly(ctx context.Context) bool { return ctx.Value(markOnlyKey{}) != nil } -// db work only, so the jetstream shards can call it +// a mark is a db row saying the repo needs a refresh, the repo's job picks it up func (s *Spindle) markRepo(repoDid syntax.DID) error { known, err := s.knowsRepo(repoDid) if err != nil { @@ -228,8 +262,7 @@ func (s *Spindle) markRepo(repoDid syntax.DID) error { return nil } -// db work only. a record for a repo whose refresh is still marked waits for that refresh, one for -// a repo spindle doesn't know at all is dropped +// same as markRepo but for one record, applied by the repo's job func (s *Spindle) markRecord(rec *tapc.RecordEventData) error { repo, err := s.servedRepo(rec.Did) if err != nil { @@ -250,7 +283,6 @@ func (s *Spindle) markRecord(rec *tapc.RecordEventData) error { return nil } -// a fresh claim is due at once and a failed one waits out its backoff func (s *Spindle) runDueWipe(ctx context.Context, did syntax.DID) { w, err := s.db.GetWipe(did) if err != nil || w.Due.After(s.retries.clock()) { @@ -261,7 +293,6 @@ func (s *Spindle) runDueWipe(ctx context.Context, did syntax.DID) { } } -// everything with work due goes to the repo jobs, which is the only place any of it runs func (s *Spindle) kickDueRepos() { now := s.retries.clock() for _, due := range []struct { @@ -284,20 +315,17 @@ func (s *Spindle) kickDueRepos() { } } -// what one job may spend. a knot that answers slowly could otherwise keep a worker for as long as -// it likes, and the pool is every repo's +// a slow knot would pin a worker forever, and the pool is shared var repoJobBudget = 5 * time.Minute -// the most record marks one job works through before the repo goes to the back of the queue +// past this many marks the job requeues itself, which stops one repo from hogging a worker const recordMarksPerJob = 200 var errRepoJobBudget = errors.New("the repo job ran out of time") -// a repo's job refreshes it if it is marked, then works through its record marks func (s *Spindle) runRepoJob(ctx context.Context, did syntax.DID) { - // the pull events of a repo run here too, so one job per repo is all that orders them + // pull events run in this job too, and the one-job-per-repo rule keeps them in order defer s.runDuePullEvents(ctx, did) - // a repo being wiped is not refreshed, its job finishes the wipe once that is due if wiping, err := s.db.WipeExists(did); err != nil { s.l.Error("reading a wipe", "repo", did, "err", err) return @@ -308,14 +336,13 @@ func (s *Spindle) runRepoJob(ctx context.Context, did syntax.DID) { jobCtx, cancel := context.WithTimeoutCause(ctx, repoJobBudget, errRepoJobBudget) defer cancel() s.refreshAndRunMarks(jobCtx, did) - // out of time is a failure of the repo, or it comes back every tick and holds a worker again + // count a timeout as a failure, or the repo comes back every tick and eats a worker if ctx.Err() == nil && errors.Is(context.Cause(jobCtx), errRepoJobBudget) { s.l.Warn("a repo job ran out of time, backing the repo off", "repo", did) s.failRepoRefresh(did) } } -// only repos that are served have a knot to count against, the rest share one bucket func (s *Spindle) jobKnot(did syntax.DID) string { repo, err := s.servedRepo(did) if err != nil || repo == nil { @@ -330,7 +357,7 @@ func (s *Spindle) refreshAndRunMarks(ctx context.Context, did syntax.DID) { s.l.Error("reading a repo retry", "repo", did, "err", err) return } - // anything can kick a repo, a failing one still waits out its backoff + // a kick doesn't skip the backoff, a failed refresh still waits it out if marked && !due { return } @@ -340,7 +367,7 @@ func (s *Spindle) refreshAndRunMarks(ctx context.Context, did syntax.DID) { case refreshFailed: return case refreshSynced: - // the full sync read every collaborator, so their marks have nothing left to say + // a full sync already read every collaborator, their marks can go for _, m := range covered { if err := s.db.ClearRecordRetryAt(m.RecordRetry, m.Epoch); err != nil { s.l.Error("clearing a record mark", "repo", did, "rkey", m.Rkey, "err", err) @@ -355,11 +382,8 @@ type refreshResult int const ( refreshSynced refreshResult = iota - // the repo is served but its collaborators are too many to sync whole refreshTooLong - // nothing to sync, the repo has no knot refreshNoKnot - // the repo's records wait for its retry refreshFailed ) @@ -381,13 +405,13 @@ func (s *Spindle) refreshForJob(ctx context.Context, did syntax.DID) refreshResu case ctx.Err() != nil: return refreshFailed case errors.Is(err, knotclient.ErrNoKnot): - // a did with no knot won't grow one by waiting + // waiting won't give it a knot, the retry gets dropped if err := s.retries.clear(s.db, did); err != nil { s.l.Error("clearing a repo retry", "repo", did, "err", err) } return refreshNoKnot case errors.Is(err, knotclient.ErrTooManyRecords): - // retrying can't shorten the list, the collaborator marks change it one record at a time + // retrying won't shorten the list, the collaborator marks work through it one at a time if err := s.retries.clearAt(s.db, did, epoch); err != nil { s.l.Error("clearing a repo retry", "repo", did, "err", err) } @@ -413,7 +437,6 @@ func (s *Spindle) runRecordMarks(ctx context.Context, did syntax.DID) { if ctx.Err() != nil { return } - // a repo with a long list goes to the back of the queue instead of keeping its worker if worked++; worked > recordMarksPerJob { s.retries.jobs.kick(did) return @@ -422,7 +445,7 @@ func (s *Spindle) runRecordMarks(ctx context.Context, did syntax.DID) { failed = true } } - // a failed record is backed off, so the next pass would only find the others again + // a failed mark is in backoff, looping again would just spin if failed { return } @@ -440,7 +463,7 @@ func (s *Spindle) runRecordMark(ctx context.Context, m db.RecordMark) bool { } return false } - // a repo that stopped being served returns early, before the handler clears it + // clear it here, handlers drop unserved repos without clearing if err := s.db.ClearRecordRetryAt(m.RecordRetry, m.Epoch); err != nil { s.l.Error("clearing a record mark", "repo", m.RepoDid, "rkey", m.Rkey, "err", err) return false diff --git a/spindle/repo_retry.go b/spindle/repo_retry.go index caa54c245..5269cf994 100644 --- a/spindle/repo_retry.go +++ b/spindle/repo_retry.go @@ -42,7 +42,7 @@ func (r *repoRetries) fail(d *db.DB, did syntax.DID) (bool, error) { if err != nil || exists { return attempts, err } - // made-up dids cost nothing to send, so a new one only gets in while there's room + // anyone can send a fake did, and new rows only fit while there's room rows, err := d.CountRepoRetries() if err == nil && rows >= repoRetryCap { err = errRetryTableFull @@ -75,7 +75,7 @@ func (r *repoRetries) next(prior int) (attempts int, due time.Time, retained boo return attempts, time.Time{}, false } backoff := min(repoRetryBase<<(attempts-1), repoRetryMax) - // repos that failed in the same outage would otherwise all come due in the same second + // jitter stops everything that failed in one outage from coming due together backoff -= rand.N(backoff / 4) return attempts, r.clock().Add(backoff), true } @@ -93,20 +93,38 @@ func (s *Spindle) repoJobWorkerCount() int { return defaultRepoJobWorkers } -// the refresh jobs live here too, so shutdown waits for them before the db closes +func (s *Spindle) priorityJobWorkerCount() int { + if len(s.priorityKnots()) == 0 { + return 0 + } + if n := s.cfg.Server.PriorityRepoJobWorkers; n > 0 { + return n + } + return max(1, s.repoJobWorkerCount()/4) +} + +// the config can list a did or a host, and both get reduced to a host here +func (s *Spindle) priorityKnots() map[string]bool { + return priorityKnotHosts(s.cfg.Server.PriorityKnots) +} + +// refresh jobs run here too, and returning waits for them func (s *Spindle) retryFailedRepos(ctx context.Context) { workers := s.repoJobWorkerCount() - s.retries.jobs.configure(workers, s.jobKnot) + s.retries.jobs.configure(workers, s.jobKnot, s.priorityKnots()) var pool sync.WaitGroup for range workers { pool.Go(func() { s.retries.jobs.work(ctx, s.runRepoJob) }) } + for range s.priorityJobWorkerCount() { + pool.Go(func() { s.retries.jobs.workPriority(ctx, s.runRepoJob) }) + } defer pool.Wait() ticker := time.NewTicker(repoRetryTick) defer ticker.Stop() var lastPrune time.Time - // a crash can leave marks and pull events that were never tried, so they don't wait for the first tick + // after a crash there can be untried marks, kick now instead of waiting a tick s.kickDueRepos() for { select { diff --git a/spindle/server.go b/spindle/server.go index 4fe3af25c..4fe978a9c 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -20,7 +20,6 @@ import ( "strconv" "strings" "sync" - "sync/atomic" "time" "github.com/bluesky-social/indigo/atproto/syntax" @@ -123,11 +122,12 @@ type Spindle struct { retries repoRetries knots *knotclient.Client - pushGroup singleflight.Group - notifyInflight atomic.Int32 - knotInflight knotSlots - notifyAdmission notifyAdmission - workflowDefs workflowDefCache + pushGroup singleflight.Group + notifySlots notifySlots + notifyPriorityOnce sync.Once + notifyPriorityHosts map[string]bool + notifyAdmission notifyAdmission + workflowDefs workflowDefCache } func newCacheStore(ctx context.Context, cfg *config.Config) (storage.Storage, error) { diff --git a/spindle/torture_ingest_test.go b/spindle/torture_ingest_test.go index 3e7c9cb76..0781aeb41 100644 --- a/spindle/torture_ingest_test.go +++ b/spindle/torture_ingest_test.go @@ -22,13 +22,11 @@ import ( "tangled.org/core/spindle/knotclient" ) -// copies of production values the scenarios reason with. when one changes the scenario that -// uses it needs the new number, and a failing check is how that shows +// copies of spindle/knotclient/knotclient.go values, change these with it const ( - // spindle/knotclient/knotclient.go:27 pageSize and :28 maxListPages ingListPageSize = 100 ingListPageCap = 32 - // the knot requests one repo refresh makes: manifest, owner declaration, collaborator list (spindle/repos.go ensureRepo) + // manifest, owner declaration, collaborator list (ensureRepo in spindle/repos.go) ingRequestsPerRefresh = 3 ) @@ -41,7 +39,7 @@ func (w *tortureWorld) isCollab(repo *tortureRepo, who syntax.DID) bool { return err == nil && ok } -// the knot knows first, jetstream only says that something changed +// knot gets the record before jetstream hears about it 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"}) } @@ -56,7 +54,6 @@ func (w *tortureWorld) revokeCollaborator(repo *tortureRepo, who syntax.DID) int return w.emitCollaborator(repo, who, jmodels.CommitOperationDelete) } -// true once the jetstream reader is past the event at us func ingConsumed(w *tortureWorld, us int64, within time.Duration) bool { return w.r.waitFor(within, func() bool { s := w.spindle() @@ -89,7 +86,7 @@ func ingManifestReads(knots []*tortureKnot) int64 { return n } -// times a collaborator grant from the moment it is emitted to the moment casbin has it +// 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 { start := time.Now() us := w.grantCollaborator(repo, who) @@ -103,7 +100,7 @@ func (w *tortureWorld) probeGrant(repo *tortureRepo, who syntax.DID, within time return us } -// the most requests one counter saw in any second, sampled every 100ms +// peak growth of count over any 1s, sampled every 100ms type ingBurst struct { peak atomic.Int64 stop chan struct{} @@ -143,7 +140,6 @@ func (b *ingBurst) finish() int64 { return b.peak.Load() } -// most retries sharing one due second, and how many different seconds there are func ingDueSpread(w *tortureWorld) (sameSecond, distinct int) { s := w.spindle() if s == nil { @@ -167,9 +163,6 @@ func ingDueSpread(w *tortureWorld) (sameSecond, distinct int) { } func init() { - // crafted manifests from dids nobody admitted, on a closed spindle. some point at a slow - // knot and pds, some have no knot service at all, and a served repo keeps getting real - // collaborator events while it all queues up on the single jetstream reader registerTorture("ingest-hostile-manifests", "crafted manifests from unadmitted dids while a served repo keeps getting events", func(r *tortureRun) { slow := r.load("slow_attackers", 20) noKnot := r.load("noknot_attackers", 200) @@ -203,7 +196,7 @@ func init() { last = w.js.commit(did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, jmodels.CommitOperationUpdate, manifestRecord(owner, "x", w.spindleDid())) } - // these owners are on the allow-list, so only the missing knot stops them (ING-2 past the enrollment check) + // owners are allowlisted, so only the missing knot stops 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)) @@ -223,7 +216,7 @@ func init() { } waiters.Wait() drained := ingConsumed(w, last, within) - // the reader is past the events once they are marked, the jobs behind them take longer + // 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 }) r.latency("legit_apply", &applied) @@ -254,8 +247,6 @@ func init() { } }) - // a jetstream gap on a spindle with many repos, a changed few that only hydrant can bring - // back, live events arriving the whole time, and then a second gap while it runs registerTorture("ingest-gap-catchup", "a jetstream gap with many repos, live events during it, then a second gap", func(r *tortureRun) { n := r.load("repos", 60) knots := int(r.param("knots", 3)) @@ -288,7 +279,7 @@ func init() { return float64(len(pending)) }) - // the cursor has to be saved past everything, or a boot just replays + // the cursor needs to pass everything or the next boot replays it at := w.js.filler() ingConsumed(w, at, 30*time.Second) w.stop() @@ -334,7 +325,7 @@ func init() { started := r.waitFor(time.Minute, func() bool { return len(w.hyd.erasedDids()) > erasedBefore }) gaps := 1 if secondGap && started { - // probes in flight would be forgotten with the history, so let them land first + // forget drops the history, so probes in flight have to land first paused.Store(true) ingConsumed(w, lastUS.Load(), 30*time.Second) w.js.forget() @@ -379,18 +370,17 @@ func init() { 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) - // what a first catch-up backfilled lacks the events of a second gap, so every did goes again once per gap + // each gap can erase every did once, so the limit scales with gaps 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) - // the probes are live events that refresh their repos on purpose, so they are not the catch-up's + // probes refresh their repos too, so take them out catchUpRefreshes := refreshes - float64(applied.len())/float64(n) r.metric("catchup_refreshes_per_repo", catchUpRefreshes) - // the catch-up's own mark, plus a second one when the owner's backfilled declaration got a job started before it - r.check("ING-3 each served repo is refreshed at most 1.5 times per gap", catchUpRefreshes <= 1.5*float64(gaps), + // the 0.2 is slack for a record that lands while the repo job runs + 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()) }) - // a burst of collaborator grants and revokes on one repo, then a list past the page cap registerTorture("ingest-collab-churn", "a burst of collaborator grants and revokes, then a list past the page cap", func(r *tortureRun) { events := r.load("events", 200) pool := 40 @@ -465,7 +455,7 @@ func init() { if oversize == 0 { return } - // past the page cap the list errors out, so a revoke has to still land + // the list errors past the page cap, so the revoke has to land anyway victim := syntax.DID("did:plc:churnvictim") w.grantCollaborator(repo, victim) if !r.waitFor(30*time.Second, func() bool { return w.isCollab(repo, victim) }) { @@ -480,14 +470,13 @@ func init() { ingConsumed(w, us, time.Minute) revoked := r.waitFor(10*time.Second, func() bool { return !w.isCollab(repo, victim) }) perEvent := float64(k.xrpc.Load()-oversizeBefore) / 2 - // the job's failed full sync is a page cap of list requests plus a few, and each collaborator record one more + // a failed full sync is the page cap in lists plus a few more requests r.metric("oversize_requests_per_event", perEvent) r.metric("oversize_retry_rows", w.countRows("select count(*) from 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) }) - // every repo's refresh fails while the knots are down, then the knots come back registerTorture("ingest-retry-herd", "every repo's refresh fails while the knots are down, then they come back", func(r *tortureRun) { n := r.load("repos", 100) knots := int(r.param("knots", 4)) @@ -572,7 +561,6 @@ func init() { 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) }) - // the owner flips the manifest between this spindle and another one, as fast as spindle reacts registerTorture("ingest-manifest-flap", "the owner flips the manifest between this spindle and another one", func(r *tortureRun) { flips := r.load("flips", 40) gitDelay := r.duration("git_delay", 2*time.Second) @@ -589,7 +577,7 @@ func init() { r.t.Fatal("the repo never got served with its collaborator") } - // the seed already cost one schedule load, wait for its fetches to stop to price a single one + // wait for the seed's fetches to stop, so one load can be priced seen, stableSince := int64(-1), time.Now() r.waitFor(30*time.Second, func() bool { if now := k.fetches.Load(); now != seen { @@ -624,7 +612,7 @@ func init() { forgetting.since(flipStart) } peak = max(peak, r.gaugeValue("goroutines")) - // where the goroutines sit is the whole answer, and they're gone by the time the check runs + // dump now, the goroutines are gone by the time the check runs if !dumped && peak-baseGoroutines > goroutineSlack { dumped = true r.dumpGoroutines() @@ -641,14 +629,13 @@ func init() { 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 removes the clone, so every paced return to this spindle needs its own load + // forgetting drops the clone, so each return to this spindle loads again r.check("ING-10 a paced flap loads schedules once per return", loads <= float64(flips/2)+1, "%.1f schedule loads (%d git fetches at %d per load) for %d returns to this spindle", loads, k.fetches.Load()-fetchesBefore, fetchesPerLoad, flips/2) r.check("ING-10 forgetting a repo isn't held up by a clone in flight", forgetting.quantile(1) <= gitDelay/2, "slowest away flip took %v, a clone fetch is %v, want under half of it", forgetting.quantile(1), gitDelay) r.settles("goroutines", baseGoroutines+goroutineSlack, time.Minute) - // the same flips with no waiting between them, all of them behind one repo job burstFetches := k.fetches.Load() var burstLast int64 for i := range flips { @@ -669,10 +656,9 @@ 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)) }) - // a manifest event that lands while every db write fails. the mark is the only thing between the - // event and the repo job, so if the event counts as done the repo never learns of it + // the mark is the only way to the repo job, so a failed mark can't count as done registerTorture("ingest-mark-fails", "the db is locked past its busy timeout while a manifest event lands", func(r *tortureRun) { - // sqlite's busy timeout is 5s (sqlite/sqlite.go) + // past the 5s sqlite busy timeout in sqlite/sqlite.go lockHold := 8 * time.Second w := newTortureWorld(r) @@ -703,8 +689,6 @@ func init() { "repo still served a minute after the db unlocked, the event that moved it away was dropped") }) - // one repo's knot hangs while the reader keeps going: other repos have to apply at once, the - // hung repo's own later events must land after it unsticks, and the cursor must not wait registerTorture("ingest-hung-repo", "one repo's knot hangs while the reader keeps going", func(r *tortureRun) { others := r.load("other_repos", 8) events := int(r.param("other_events", 4)) diff --git a/spindle/torture_notify_abuse_test.go b/spindle/torture_notify_abuse_test.go index cd354470c..2eb2254e6 100644 --- a/spindle/torture_notify_abuse_test.go +++ b/spindle/torture_notify_abuse_test.go @@ -23,7 +23,7 @@ import ( "tangled.org/core/spindle/models" ) -// one owner per knot, because tortureWorld points an owner's pds at the knot of the last repo it added +// 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 { @@ -47,7 +47,6 @@ func (w *tortureWorld) abuseServe(r *tortureRun, repos []*tortureRepo) { r.metric("seed_s", time.Since(start).Seconds()) } -// repos on one knot take turns, like real pushes would func abuseQueues(repos []*tortureRepo, from, to int) map[*tortureKnot][]pushDelivery { queues := map[*tortureKnot][]pushDelivery{} for i := from; i < to; i++ { @@ -65,7 +64,6 @@ func abuseRatio(n, of int64) float64 { return float64(n) / float64(of) } -// n requests from conc clients, or until ctx ends when n is 0 func (w *tortureWorld) abuseHammer(ctx context.Context, n, conc int, token func() string, body []byte, st *deliveryStats) { client := &http.Client{Timeout: 30 * time.Second, Transport: &http.Transport{MaxIdleConnsPerHost: conc}} defer client.CloseIdleConnections() @@ -90,7 +88,6 @@ func (w *tortureWorld) abuseHammer(ctx context.Context, n, conc int, token func( wg.Wait() } -// a webhook target that answers at once, or sits on the request for hang type abuseHook struct { srv *httptest.Server arrived atomic.Int64 @@ -130,7 +127,7 @@ func newAbuseHook(r *tortureRun, hang time.Duration) *abuseHook { h.mu.Unlock() rw.WriteHeader(http.StatusOK) })) - // cleanups run last in first out, so the hung requests are let go before the server waits on them + // cleanups run last in first out, so release frees the hung requests before srv.Close waits on them r.t.Cleanup(h.srv.Close) r.t.Cleanup(h.release) return h @@ -149,7 +146,7 @@ func abuseHookKey(d pushDelivery) string { return d.newSha + "|" + d.repo.owner.String() + "/" + strings.ToLower(d.repo.name) } -// straight into the db, so the per repo and per owner quotas don't get in the way of the load +// goes straight to the db, so the per repo and per owner quotas don't apply 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) @@ -163,10 +160,29 @@ func (w *tortureWorld) abuseNotifyInflight() float64 { if s == nil { return math.NaN() } - return float64(s.notifyInflight.Load()) + 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) +} + +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]) } -// hostile knots that keep every slot they can get busy with repos the spindle has never heard of func (w *tortureWorld) abuseSybil(ctx context.Context, k *tortureKnot, repos []*tortureRepo, next *atomic.Int64, st *deliveryStats) { client := &http.Client{Timeout: time.Minute} sha := strings.Repeat("a", 40) @@ -185,9 +201,7 @@ func (w *tortureWorld) abuseSybil(ctx context.Context, k *tortureKnot, repos []* } func init() { - // N2. a few hostile hosts whose manifest and pds answers crawl, each notifying for repos nobody - // serves yet, while real knots flush their outboxes. the hold is shorter than the 30s knot - // client timeout a real attacker would sit on, so slots free up more often than in production + // hold is shorter than the knot client's 30s timeout, so slots free up more often than in prod 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 @@ -201,6 +215,7 @@ func init() { w := newTortureWorld(r) w.maxJobs = 8 r.gauge("notify_inflight", w.abuseNotifyInflight) + r.gauge("notify_unknown_inflight", w.abuseUnknownInflight) legit := w.abuseKnots(legitKnots) rs := w.abuseRepos(legit, "sybillegit", perKnot, pushes, map[string]string{"ci.yaml": tortureWorkflow("sleep 20ms")}) hostile := w.abuseKnots(hosts) @@ -233,9 +248,10 @@ func init() { atk.Go(func() { w.abuseSybil(atkCtx, k, hostileRepos[k], next, &hst) }) } } - want := float64(min(16, hosts*4)) - saturated := r.waitFor(3*hold+5*time.Second, func() bool { return w.abuseNotifyInflight() >= want }) - r.metric("slots_saturated", map[bool]float64{true: 1, false: 0}[saturated]) + want := float64(min(notifyMaxUnknown, hosts*notifyMaxUnknownPerKnot)) + saturated := r.waitFor(3*hold+5*time.Second, func() bool { return w.abuseUnknownInflight() >= want }) + r.metric("unknown_pool_saturated", map[bool]float64{true: 1, false: 0}[saturated]) + r.check("hostile knots fill the pool for unknown repos (setup)", saturated, "%.0f of %.0f unknown-repo seats held", w.abuseUnknownInflight(), want) legitCtx, cancel := context.WithTimeout(context.Background(), budget) var st deliveryStats @@ -253,11 +269,11 @@ func init() { want2 := len(rs) * (pushes - half) maxWait := st.accepted.quantile(1) if st.accepted.len() < want2 { - // pushes that never got in waited the whole budget, accepted ones alone would read as no wait + // pushes that never got in count as waiting the whole budget, or they'd hide as no wait maxWait = max(maxWait, attack) } r.metric("legit_max_wait_s", maxWait.Seconds()) - r.note("hostile hold is %v per call and ensureRepo makes three calls to the hostile knot, production clients hold up to 30s each", hold) + r.note("hostile hold is %v per call and ensureRepo makes three calls to the hostile knot, production clients hold up to 30s each. the notify slots stay free of them, they only hold the unknown-repo seats", hold) r.check("legit pushes all delivered under sybil load (N2)", st.accepted.len() == want2, "%d of %d legit pushes accepted within %v", st.accepted.len(), want2, budget) r.check("legit knots mostly not told busy (N2)", abuseRatio(int64(st.seen("http_503")), st.attempts.Load()) <= 0.1, "%d of %d legit attempts got 503, want at most 10%%", st.seen("http_503"), st.attempts.Load()) @@ -265,9 +281,90 @@ 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)) }) - // N1. a flood of forged tokens naming a real knot, and what that does to the real knot's own - // notifies. what each kind of bad token is answered with is pinned by ordinary tests in - // xrpc/serviceauth and xrpc/org_tangled_event_notify_event_test.go + // 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) + budget := r.duration("legit_budget", 90*time.Second) + const commits = 6 + workflows := map[string]string{"ci.yaml": tortureWorkflow("sleep 20ms")} + + w := newTortureWorld(r) + w.maxJobs = 8 + w.notifySlots = pool + r.gauge("notify_inflight", w.abuseNotifyInflight) + legit := w.abuseKnots(3) + rs := w.abuseRepos(legit, "sharelegit", 2, commits, workflows) + hostile := w.abuseKnots(1) + hrs := w.abuseRepos(hostile, "sharehostile", 12, commits, workflows) + w.boot() + w.abuseServe(r, append(append([]*tortureRepo{}, rs...), hrs...)) + 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)...) + } + var hst deliveryStats + var atk sync.WaitGroup + atk.Go(func() { w.flood(atkCtx, ds, 4*pool, 1000, &hst) }) + shared := pool - poolReserve(pool) + reached := r.waitFor(30*time.Second, func() bool { return w.abuseHeld(hk.host()) >= float64(shared) }) + r.check("an idle pool lets one knot use all of it but the reserve", reached, "the hostile knot held %.0f of %d slots, want the %d that aren't the reserve while nobody else waits", w.abuseHeld(hk.host()), pool, shared) + first := w.notifyWith(context.Background(), http.DefaultClient, pushRange(rs[0], 0, 1)[0]) + r.metric("first_legit_notify_ms", float64(first.elapsed.Milliseconds())) + r.check("a real knot gets the reserve at once while a hanging knot holds the rest", first.status == http.StatusOK && first.elapsed <= 2*time.Second, + "the first legit push answered %d after %v, want 200 within 2s", first.status, first.elapsed.Round(time.Millisecond)) + r.check("a hanging knot never holds the reserve", w.abuseHeld(hk.host()) <= float64(shared), "the hostile knot held %.0f of %d slots, want at most %d", w.abuseHeld(hk.host()), pool, shared) + + // sample only after the first hostile calls have run out their hold + var peak atomic.Int64 + stopSample := make(chan struct{}) + sampled := make(chan struct{}) + go func() { + defer close(sampled) + select { + case <-time.After(hold + 2*time.Second): + case <-stopSample: + return + } + for { + select { + case <-stopSample: + return + case <-time.After(10 * time.Millisecond): + } + if held := int64(w.abuseHeld(hk.host())); held > peak.Load() { + peak.Store(held) + } + } + }() + + legitCtx, cancel := context.WithTimeout(context.Background(), budget) + var st deliveryStats + start := time.Now() + w.outboxes(legitCtx, abuseQueues(rs, 0, commits), &st) + took := time.Since(start) + cancel() + close(stopSample) + <-sampled + stopAtk() + atk.Wait() + + st.report(r, "legit_") + hst.report(r, "hostile_") + r.metric("legit_flush_s", took.Seconds()) + r.metric("hostile_peak_held_after_hold", float64(peak.Load())) + want := len(rs) * commits + r.check("real knots get all their pushes delivered while a knot hangs on the slots", st.accepted.len() == want, "%d of %d legit pushes accepted within %v", st.accepted.len(), want, budget) + // 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) + }) + registerTorture("notify-bad-tokens", "a flood of forged tokens naming a real knot against that knot's own notifies", func(r *tortureRun) { conc := 16 resolveDelay := 20 * time.Millisecond @@ -280,7 +377,7 @@ func init() { w.abuseServe(r, rs) k, repo := ks[0], rs[0] body := w.notifyBody(repo, "refs/heads/main", zeroSha, repo.shas[0]) - // the first push does the real work, every later one is a cheap replay + // 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) } @@ -301,7 +398,6 @@ func init() { } forged := sign(w.spindleDid(), time.Minute, bad) - // does a real knot's traffic feel the forged flood, with every lookup costing resolveDelay w.ids.delay.Store(int64(resolveDelay)) probe := func(ctx context.Context, lat *latencies, notOK *atomic.Int64) { client := &http.Client{Timeout: 10 * time.Second} @@ -345,8 +441,6 @@ func init() { r.check("legit median under forged flood stays near its quiet median (N1)", floodP50 <= 3*baseP50+50*time.Millisecond, "median %v quiet, %v under flood, want <= 3x quiet + 50ms", baseP50.Round(time.Millisecond), floodP50.Round(time.Millisecond)) }) - // N7. repos whose webhooks accept the request and never answer, pushes that keep coming, and - // healthy repos on another knot that share the 8 delivery workers and the 256 slot queue registerTorture("webhooks-hang", "webhooks that never answer, sharing the delivery pool with healthy repos", func(r *tortureRun) { badRepos := r.load("bad_repos", 6) hooksPer := 4 @@ -366,7 +460,7 @@ func init() { w.abuseServe(r, append(append([]*tortureRepo{}, bad...), good...)) hung, fast := newAbuseHook(r, hang), newAbuseHook(r, 0) - // a delivery lane is a hostname, so the healthy target gets another name for the same loopback + // lanes are keyed by hostname, so localhost gets its own lane on the same loopback fastURL := strings.Replace(fast.srv.URL, "127.0.0.1", "localhost", 1) for _, repo := range bad { for i := range hooksPer { @@ -409,7 +503,7 @@ func init() { } hookLat.add(seen.Sub(at)) } - // a delivery that never came waited the whole settle time, or the percentile only sees the lucky ones + // missing deliveries count as the full settle time, or the p99 only sees the lucky ones for range missing { hookLat.add(settle) } @@ -436,9 +530,6 @@ func init() { r.check("a push isn't slowed by the delivery backlog (N7)", pushP99 <= 3*time.Second, "healthy notify p99 %v, want <= 3s", pushP99.Round(time.Millisecond)) }) - // N4. one repo whose git fetch hangs while other repos get pushes, the repo itself gets a - // refresh and a new repo shows up through jetstream. the hold is per git request, so the - // real hold is several times it registerTorture("notify-hung-fetch", "one repo whose git fetch hangs while other repos get pushes", func(r *tortureRun) { victims := 3 hold := r.duration("git_hold", 10*time.Second) @@ -480,7 +571,6 @@ func init() { }) } - // a refresh of the stuck repo itself, then a new repo that has nothing to do with it var refresh time.Duration var refreshErr error wg.Go(func() { diff --git a/spindle/torture_notify_test.go b/spindle/torture_notify_test.go index ebad28736..f0364077a 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()) } -// 5xx answers that aren't the busy signal, which a knot retries forever +// 503 is the busy signal, any other 5xx counts as a failure func (st *deliveryStats) serverErrors() int { total := 0 st.statuses.Range(func(k, v any) bool { @@ -66,9 +66,7 @@ func (st *deliveryStats) report(r *tortureRun, prefix string) { }) } -// what knot2's outbox does: one delivery in flight per spindle, oldest first, -// a did.json fetch and a fresh token per attempt, backoff 1s doubling to 60s -// with no jitter that a success clears, and any 4xx but 408/429 drops the event +// 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) @@ -99,7 +97,6 @@ func (w *tortureWorld) knotOutbox(ctx context.Context, queue []pushDelivery, st } } -// many knots coming back at once, each with its own outbox func (w *tortureWorld) outboxes(ctx context.Context, queues map[*tortureKnot][]pushDelivery, st *deliveryStats) *latencies { var finished latencies start := time.Now() @@ -114,8 +111,7 @@ func (w *tortureWorld) outboxes(ctx context.Context, queues map[*tortureKnot][]p return &finished } -// a knot that ignores the outbox rules: many deliveries at once, retrying -// busy answers on its own short backoff +// 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) @@ -156,7 +152,6 @@ func (w *tortureWorld) notifyWith(ctx context.Context, client *http.Client, d pu return w.post(ctx, client, w.token(d.repo.knot), w.notifyBody(d.repo, d.ref, d.oldSha, d.newSha)) } -// commits from..to of a repo as consecutive pushes to main, in history order func pushRange(repo *tortureRepo, from, to int) []pushDelivery { old := zeroSha if from > 0 { @@ -170,7 +165,6 @@ func pushRange(repo *tortureRepo, from, to int) []pushDelivery { return out } -// every commit as its own push to main func pushesOf(repo *tortureRepo, n int) []pushDelivery { return pushRange(repo, 0, n) } func ownerFor(i, perOwner int) syntax.DID { @@ -179,12 +173,11 @@ func ownerFor(i, perOwner int) syntax.DID { func (w *tortureWorld) drained(within time.Duration) bool { return w.r.waitFor(within, func() bool { - // a dequeued job is off the jobs table before the engine counts it as running + // 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 }) } -// once the queue is empty anything still pending or running is stuck func (w *tortureWorld) unfinishedWorkflows() float64 { return w.countRows(` select count(*) from workflow_statuses ws @@ -194,7 +187,6 @@ func (w *tortureWorld) unfinishedWorkflows() float64 { ) and ws.status in ('pending', 'running')`) } -// knots with repos with pushes, all served before the load starts func (w *tortureWorld) pushWorld(knots, repos, pushes int, workflows map[string]string) []*tortureRepo { for range knots { w.addKnot() @@ -217,7 +209,6 @@ func (w *tortureWorld) checkPipelines(r *tortureRun, want int, what string) { } func init() { - // knots coming back with full outboxes, the way knot2 sends them registerTorture("notify-outbox-flush", "knots coming back with full outboxes, the way knot2 sends them", func(r *tortureRun) { knots := r.load("knots", 12) perKnot := 4 @@ -236,13 +227,13 @@ func init() { for _, d := range pushesOf(repo, pushes) { queues[repo.knot] = append(queues[repo.knot], d) unique++ - // a knot that lost its ack resends, so the same push comes twice + // a lost ack makes the knot resend, so some pushes come twice if rand.Float64() < dupRate { queues[repo.knot] = append(queues[repo.knot], d) } } } - // repos on one knot interleave like real pushes would + // shuffles across repos, a repo's own pushes keep their order for _, q := range queues { rand.Shuffle(len(q), func(i, j int) { if q[i].repo != q[j].repo { @@ -265,7 +256,6 @@ func init() { r.settles("goroutines", baseGoroutines+20, 30*time.Second) }) - // one knot that ignores the outbox discipline and blasts the per-knot and global caps registerTorture("notify-flood", "one knot that ignores the outbox discipline and blasts the caps", func(r *tortureRun) { repos := r.load("repos", 16) pushes := 10 diff --git a/spindle/torture_pool_test.go b/spindle/torture_pool_test.go new file mode 100644 index 000000000..8f43526ba --- /dev/null +++ b/spindle/torture_pool_test.go @@ -0,0 +1,307 @@ +//go:build torture + +package spindle + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +// knots includes this one, and the share stays under the pool less its reserve +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 + 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)) + notifySlots := int(r.param("notify", 64)) + smallKnots := 4 + pushRepos := 40 + slowDelay := 2 * time.Second + budget := r.duration("drain_budget", 30*time.Minute) + + w := newTortureWorld(r) + w.repoPool, w.notifySlots = repoPool, notifySlots + w.maxJobs = 8 + big := w.addKnot() + var small []*tortureKnot + for range smallKnots { + small = append(small, w.addKnot()) + } + 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; { + case rem < 6: + return big, owner + case rem < 9: + return small[i%smallKnots], owner + } + return slow, owner + } + repos := w.addRepos(n, pick, 0, nil) + pushes := w.addRepos(pushRepos, func(i int) (*tortureKnot, syntax.DID) { + return small[i%smallKnots], syntax.DID(fmt.Sprintf("did:plc:poolpush%03d", i)) + }, 2, map[string]string{"ci.yaml": tortureWorkflow("sleep 10ms")}) + r.metric("repo_pool", float64(repoPool)) + r.metric("notify_slots", float64(notifySlots)) + + r.gauge("knots_inflight", func() float64 { + var total int64 + for _, k := range w.knots { + total += k.inflight.Load() + } + return float64(total) + }) + r.gauge("slow_knot_inflight", func() float64 { return float64(slow.inflight.Load()) }) + r.gauge("notify_inflight", w.abuseNotifyInflight) + r.gauge("notify_held_max_per_knot", func() float64 { + most := 0.0 + for _, k := range w.knots { + most = max(most, w.abuseHeld(k.host())) + } + return most + }) + + w.boot() + time.Sleep(5 * time.Second) + r.metric("idle_rss_mb", r.gaugeValue("rss_mb")) + 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()) + r.metric("served_rss_mb", r.gaugeValue("rss_mb")) + + ingConsumed(w, w.js.filler(), 30*time.Second) + w.stop() + var changed []syntax.DID + var changedOn []*tortureRepo + for i := 0; i < len(repos); i += 5 { + who := syntax.DID(fmt.Sprintf("did:plc:poolnew%05d", i)) + ingPutCollaborator(repos[i], who) + changed = append(changed, who) + changedOn = append(changedOn, repos[i]) + } + for _, k := range w.knots { + k.delay.Store((20 * time.Millisecond).Nanoseconds()) + } + slow.delay.Store(slowDelay.Nanoseconds()) + w.js.forget() + w.js.filler() + readsBefore := ingManifestReads(w.knots) + + w.boot() + start := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), budget) + defer cancel() + var wg sync.WaitGroup + + var st deliveryStats + var ds []pushDelivery + for _, repo := range pushes { + ds = append(ds, pushRange(repo, 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 + stopLive := make(chan struct{}) + var liveSent atomic.Int64 + wg.Go(func() { + for i := 0; ; i++ { + select { + case <-stopLive: + 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) + liveSent.Add(1) + } + }) + + // a job already running can go past the share, starting a new one over it while others wait is the bug + slowHost := slow.host() + var jobsPeak, violations atomic.Int64 + stopSample := make(chan struct{}) + sampling := make(chan struct{}) + go func() { + defer close(sampling) + prevRunning, prevOthers := 0, 0 + for { + select { + case <-stopSample: + return + case <-time.After(2 * time.Millisecond): + } + jobs := &w.spindle().retries.jobs + jobs.mu.Lock() + running := jobs.running[slowHost] + others := 0 + for knot := range jobs.queued { + if knot != slowHost { + others++ + } + } + jobs.mu.Unlock() + if others > 0 { + if int64(running) > jobsPeak.Load() { + jobsPeak.Store(int64(running)) + } + if prevOthers > 0 && running > prevRunning && running > poolFairShare(repoPool, others+1) { + violations.Add(1) + } + } + prevRunning, prevOthers = running, others + } + }() + + if _, err := w.spindle().db.Exec(`create table if not exists torture_probe (n integer)`); err != nil { + r.t.Fatal(err) + } + 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) + } + }) + + caughtUp := r.waitFor(budget, func() bool { + pending, err := w.spindle().db.PendingCatchUp() + return err == nil && len(pending) == 0 + }) + catchUp := time.Since(start) + applied2 := r.waitFor(max(time.Until(start.Add(budget)), time.Second), func() bool { + for i, who := range changed { + if !w.isCollab(changedOn[i], who) { + return false + } + } + return true + }) + elapsed := time.Since(start) + close(stopLive) + close(stopProbe) + close(stopSample) + cancel() + wg.Wait() + <-sampling + waiters.Wait() + + // a run that hit the budget has no real time, so only finished waits report one + if caughtUp { + r.metric("catchup_s", catchUp.Seconds()) + } + if applied2 { + r.metric("apply_all_s", elapsed.Seconds()) + } + r.metric("run_s", elapsed.Seconds()) + r.metric("refreshes_per_s", float64(ingManifestReads(w.knots)-readsBefore)/elapsed.Seconds()) + r.metric("live_events_per_s", float64(liveSent.Load())/elapsed.Seconds()) + r.metric("notify_accepted_per_s", float64(st.accepted.len())/elapsed.Seconds()) + r.metric("probe_write_errors", float64(probeErrors.Load())) + 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())) + 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") + + r.check("catch-up finished within the budget", caughtUp, "the pending catch-up was not empty after %v", budget) + r.check("every changed repo applied within the budget", applied2, "%d changed repos, not all applied %v after the boot", len(changed), budget) + r.check("a slow knot never starts another repo job past its share while others wait", violations.Load() == 0, + "the slow knot started %d jobs while over its share of %d workers with other knots waiting, its most running at once then was %d", violations.Load(), repoPool, jobsPeak.Load()) + }) +} + +func poolReserve(pool int) int { return max(2, pool/8) } + +func init() { + registerTorture("pool-priority", "a priority knot's repo jobs run at once while a hostile knot pins the shared repo pool", func(r *tortureRun) { + hostileRepos := r.load("hostile_repos", 300) + pool := int(r.param("pool", 16)) + hold := 4 * time.Second + budget := r.duration("priority_budget", 3*time.Second) + + w := newTortureWorld(r) + w.repoPool = pool + prio, plain, hostile := w.addKnot(), w.addKnot(), w.addKnot() + w.priorityKnots = []string{prio.host()} + prioRepos := w.addRepos(10, func(i int) (*tortureKnot, syntax.DID) { + return prio, syntax.DID(fmt.Sprintf("did:plc:prioowner%03d", i)) + }, 0, nil) + plainRepos := w.addRepos(10, func(i int) (*tortureKnot, syntax.DID) { + return plain, syntax.DID(fmt.Sprintf("did:plc:plainowner%03d", i)) + }, 0, nil) + bad := w.addRepos(hostileRepos, func(i int) (*tortureKnot, syntax.DID) { + return hostile, syntax.DID(fmt.Sprintf("did:plc:hostileowner%04d", i)) + }, 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.stop() + + for i, repo := range bad { + ingPutCollaborator(repo, syntax.DID(fmt.Sprintf("did:plc:hostilenew%04d", i))) + } + hostile.delay.Store(hold.Nanoseconds()) + w.js.forget() + w.js.filler() + w.boot() + + hostileHost := hostile.host() + running := func() int { + jobs := &w.spindle().retries.jobs + jobs.mu.Lock() + defer jobs.mu.Unlock() + return jobs.running[hostileHost] + } + pinned := r.waitFor(time.Minute, func() bool { return running() >= pool-poolReserve(pool) }) + 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()) + }) +} diff --git a/spindle/torture_world_test.go b/spindle/torture_world_test.go index 12999cc73..da23769ed 100644 --- a/spindle/torture_world_test.go +++ b/spindle/torture_world_test.go @@ -40,8 +40,7 @@ import ( "tangled.org/core/xrpc/serviceauth" ) -// one directory for every did, so knots can be did:web on loopback and still sign. it counts -// what would be a cold fetch from plc or did:web +// one directory for every did, so a knot can be did:web on loopback type tortureDirectory struct { *identity.MockDirectory delay atomic.Int64 @@ -87,8 +86,6 @@ func sleepCtx(ctx context.Context, d time.Duration) error { } } -// a knot that also plays every owner's pds, with knobs for latency, outages and git failures -// in front of fakeKnot's records type tortureKnot struct { *fakeKnot did syntax.DID @@ -104,8 +101,7 @@ type tortureKnot struct { fetches atomic.Int64 inflight atomic.Int64 peak atomic.Int64 - // getRecord and listRecords calls by collection - reads sync.Map + reads sync.Map } func (k *tortureKnot) readsOf(collection string) int64 { @@ -174,7 +170,7 @@ type tortureRepo struct { owner syntax.DID name string knot *tortureKnot - // every commit sits on its own branch so it stays fetchable by sha + // every commit gets a branch, or git won't fetch it by sha shas []string } @@ -191,14 +187,16 @@ type tortureWorld struct { pdsMu sync.Mutex pds map[syntax.DID]*tortureKnot - // knobs read at boot + // read at boot, so set these before the world starts maxJobs int admission config.Admission qm *quota.Manager - // moves the repo retry clock forward, so backoffs fall due without waiting - skew atomic.Int64 - // skips the identity cache, for scenarios that change a did doc mid run + skew atomic.Int64 + // the cache would hide a did doc that changes mid run uncachedIDs bool + // zero keeps the config default + repoPool, notifySlots int + priorityKnots []string s *Spindle sp atomic.Pointer[Spindle] @@ -264,10 +262,10 @@ func (w *tortureWorld) spindleDid() string { return config.Server{Hostname: faultHostname}.Did().String() } -// the one place that reads the spindle from other goroutines, nil once it is stopped +// nil once stopped, safe to read from other goroutines func (w *tortureWorld) spindle() *Spindle { return w.sp.Load() } -// a missing table or a stopped spindle reads as NaN, which no ceiling check lets through +// NaN on error, so any `<=` ceiling check fails closed func (w *tortureWorld) countRows(query string) float64 { s := w.spindle() if s == nil { @@ -311,7 +309,6 @@ func (w *tortureWorld) addKnot() *tortureKnot { return k } -// a workflow file run by the torture engine, steps are its little command language func tortureWorkflow(steps ...string) string { var b strings.Builder b.WriteString("engine: torture\nwhen:\n - event: push\n branch: main\nsteps:\n") @@ -321,8 +318,7 @@ func tortureWorkflow(steps ...string) string { return b.String() } -// git fast-import makes the whole history in one process, a commit per branch. -// with no commits the repo only has records, enough for membership load +// no commits skips git, the repo only has 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 { @@ -331,7 +327,7 @@ func (w *tortureWorld) addRepo(k *tortureKnot, owner syntax.DID, commits int, wo return repo } -// no t.Fatal in here, addRepos runs it off the test goroutine +// no t.Fatal here, addRepos calls this off the test goroutine func (w *tortureWorld) tryAddRepo(k *tortureKnot, owner syntax.DID, commits int, workflows map[string]string) (*tortureRepo, error) { n := w.nextRepo.Add(1) repo := &tortureRepo{did: syntax.DID(fmt.Sprintf("did:plc:torture%06d", n)), owner: owner, name: fmt.Sprintf("repo%d", n), knot: k} @@ -349,7 +345,6 @@ func (w *tortureWorld) tryAddRepo(k *tortureKnot, owner syntax.DID, commits int, return repo, nil } -// an owner's first repo picks the knot that plays their pds, later ones declare there too func (w *tortureWorld) pdsOf(owner syntax.DID, k *tortureKnot) *tortureKnot { w.pdsMu.Lock() defer w.pdsMu.Unlock() @@ -429,7 +424,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()}) } -// repos in parallel, git is most of the setup time +// git is most of the setup time, so repos go in parallel 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 @@ -471,7 +466,7 @@ func (w *tortureWorld) emitCollaborator(repo *tortureRepo, who syntax.DID, op st return w.js.collaborator(repo.did, who, op) } -// serves repos the way production learns of them, through jetstream +// production learns repos from jetstream, so seed does too func (w *tortureWorld) seed(repos []*tortureRepo, within time.Duration) bool { var last int64 for _, repo := range repos { @@ -506,7 +501,7 @@ func (w *tortureWorld) boot() { if err != nil { t.Fatal(err) } - // production defaults, so nothing reads a zero that a real deploy never has + // 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, @@ -526,6 +521,15 @@ func (w *tortureWorld) boot() { cfg.Server.Owner = testOwner.String() cfg.Server.Admission = w.admission cfg.Server.MaxJobCount = w.maxJobs + if w.repoPool > 0 { + cfg.Server.RepoJobWorkers = w.repoPool + } + if w.notifySlots > 0 { + cfg.Server.NotifySlots = w.notifySlots + } + if len(w.priorityKnots) > 0 { + cfg.Server.PriorityKnots = w.priorityKnots + } cfg.ArtifactStores.Disk.Dir = filepath.Join(w.dir, "artifacts") for _, dir := range []string{cfg.Server.RepoDir, cfg.Server.LogDir, cfg.ArtifactStores.Disk.Dir} { if err := os.MkdirAll(dir, 0o755); err != nil { @@ -538,7 +542,7 @@ func (w *tortureWorld) boot() { if err != nil { t.Fatalf("New: %v", err) } - // the same cache production puts in front of plc, so lookups count cold fetches + // 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 @@ -562,7 +566,7 @@ func (w *tortureWorld) boot() { } } -// a clean shutdown, the db stays for the next boot +// keeps the db on disk so the next boot can reopen it func (w *tortureWorld) stop() { if w.cancel == nil { return @@ -603,7 +607,6 @@ func (w *tortureWorld) notifyBody(repo *tortureRepo, ref, oldSha, newSha string) return body } -// signs a fresh token like a knot does for every delivery 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)