From 0480fc827a44f3a01b0dd181c56044fa8fd68087 Mon Sep 17 00:00:00 2001 From: Bretton <36870434+BrettM86@users.noreply.github.com> Date: Fri, 10 Jul 2026 20:08:15 -0700 Subject: [PATCH] =?UTF-8?q?Task=2012:=20perf=20&=20scale=20=E2=80=94=20MST?= =?UTF-8?q?=20tree=20cache,=20streaming=20reachable-set=20getRepo,=20block?= =?UTF-8?q?s=20GC,=20ClaimNext=20skip-scan?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes the four known scaling cliffs before any big community hits them (and the storage prerequisites for the votes-as-records design revisit). Behavior-preserving throughout: golden TID/at-uri tests untouched, firehose CAR slices identical with/without the cache (test-pinned, ordered comparison, create/update/delete ops), seq order == visibility order for both event kinds, TxSideEffect hook semantics unchanged. Benchmarks (Apple M4 Pro, local test postgres, 2k-record repo fixture; "before" = identical bench file against a throwaway HEAD worktree plus the two-line testutil *testing.T -> testing.TB backport): - PutRecord (per-DID MST tree cache, internal/repo/treecache.go): 140.6ms -> 3.30ms/op (42.6x), 2.34MB -> 115KB/op, 38,482 -> 750 allocs. Was one SELECT per MST node, full tree, per commit. LRU keyed by DID, head-CID-validated, detach-on-take, re-cached only after durable commit. MST_CACHE_SIZE env (default 512, n<=0 disables). Cold first commit per DID still pays one full-tree load. - getRepo / ExportCAR (streaming reachable-set export, repo.ExportCARTo): 38.6ms/4.77MB CAR -> 27.7ms/438KB CAR (10.9x smaller, historical blocks no longer exported); the sync handler streams with residency bounded by one 256-block batch + a CID seen-set, never the full CAR. Batched ANY() block fetches — the naive per-block walk measured 698ms/op (25x worse; recorded so nobody simplifies the batching away). Mid-stream failure now panics http.ErrAbortHandler so a truncated CAR is a transport-level failure, not a clean 200; client disconnects log at Debug; vanished-repo race maps to 404. - blocks GC (internal/repo/gc.go, invariant-first design per the task spec): delete only if unreachable from the current head (one REPEATABLE READ snapshot) AND older than BLOCKS_GC_RETENTION (72h default), created_at re-checked inside the DELETE. The commit path's ON CONFLICT now refreshes blocks.created_at ("last written"), so a re-introduced block survives a concurrent sweep — the retention window is the race guard (must exceed app<->DB clock skew + sweep gap). GetRecord/GetRecordProof/ExportCARTo moved to REPEATABLE READ. Wired via internal/prune.Run, 6h sweeps, batched deletes. - ClaimNext (internal/store): recursive-CTE loose index scan over the existing partial index; per-key head rows only, heads materialized via ARRAY(...) — a plain IN regressed the planner to the O(N) scan. 50k-deep backlogged key: 162.6ms/200,855 buffers -> 0.050ms/22. Same semantics: per-key serial ordering, claimed_until fencing, FOR UPDATE SKIP LOCKED. No schema change. Resumed from a killed predecessor agent's partial working tree: its MST cache and streaming-export skeleton were kept (verified against pinned indigo source: WriteDiffBlocks dirty-clearing makes cached-tree diffs identical; no-op inserts never dirty the tree); its unbatched export walk would have been a getRepo latency regression and was rewritten. 7-reviewer second-opinion pass (5 Claude specialty agents + codex gpt-5.6-sol + gemini 3.1 pro; glm watchdog-killed): no high-severity code findings. Applied: client-visible truncation abort (4/7 flagged); false parenthetical in the gc.go invariant header corrected (only NEWLY referenced blocks are in newBlocks — believing the original would have justified deleting the reachability rule); the DELETE-time created_at re-check, the RR isolation choice, and the 256-block batch boundary are now all load-bearing in tests with fail-then-pass proofs (previously the race-guard clause could be deleted with the suite staying green); ExportCARTo pre-first-byte contract doc corrected; WithTreeCacheSize doc de-contradicted; treecache ABA reasoning documented; clock-skew assumption documented (codex unique catch). Full unit suite green; make e2e green (17/17). `since` diff export deliberately deferred and documented as GC-constrained: a real diff export would read historical blocks, which the GC invariant explicitly does not guarantee. v1.1 loop (tasks 09-12) complete. Co-Authored-By: Claude Fable 5 --- FOLLOWUPS.md | 76 ++++- LOOP_STATE.md | 60 +++- README.md | 2 + cmd/tidepool/main.go | 8 +- internal/config/config.go | 21 ++ internal/repo/bench_streaming_test.go | 22 ++ internal/repo/bench_test.go | 105 +++++++ internal/repo/blocks.go | 31 ++ internal/repo/events.go | 198 +++++++++--- internal/repo/gc.go | 230 ++++++++++++++ internal/repo/gc_test.go | 413 ++++++++++++++++++++++++++ internal/repo/repo.go | 114 +++++-- internal/repo/sync.go | 10 +- internal/repo/treecache.go | 144 +++++++++ internal/repo/treecache_test.go | 284 ++++++++++++++++++ internal/store/inbox_events.go | 87 ++++-- internal/store/inbox_events_test.go | 61 ++++ internal/sync/server.go | 83 +++++- internal/sync/sync_test.go | 48 +++ internal/testutil/db.go | 4 +- 20 files changed, 1877 insertions(+), 124 deletions(-) create mode 100644 internal/repo/bench_streaming_test.go create mode 100644 internal/repo/bench_test.go create mode 100644 internal/repo/gc.go create mode 100644 internal/repo/gc_test.go create mode 100644 internal/repo/treecache.go create mode 100644 internal/repo/treecache_test.go diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md index f8ca317..a3933c3 100644 --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -93,10 +93,21 @@ harness, **(relay)** by task 09's relay pipeline. community repo with an `author` field. Zero new DIDs, votes on the firehose. Requires a new lexicon + Coves AppView consumer — decide WITH Coves. - - Write amplification (one commit + firehose event per vote) is only + - ~~Write amplification (one commit + firehose event per vote) is only viable after the perf items land: per-DID MST cache, block GC, batched pruning (tasks 11–12). Hardening first is a prerequisite, - not a competing priority. + not a competing priority.~~ **The perf prerequisites now hold (task + 12)**: per-DID MST cache (steady-state commit 3.3 ms on a 2k-record + repo, 42.6x over the full-reload path), blocks GC (superseded blocks + reclaimed after `BLOCKS_GC_RETENTION`), batched firehose pruning + (task 11), streaming reachable-set getRepo, and a bounded ClaimNext + scan. What is still NOT solved for votes-as-records: commits remain + globally serialized (the advisory lock — ~300/s ceiling at the + benchmarked commit cost, shared across ALL writes), and every vote + would still be a firehose event relays and the AppView must chew + through. The decision is now genuinely open on design grounds + (lexicon + Coves consumer + DID externality), not blocked on storage + perf. - Write-back symmetry favors records: Coves users' votes on bridged posts are already native `social.coves.feed.vote` records; outbound translation is records→Like. Symmetric records would let frontends @@ -244,20 +255,41 @@ harness, **(relay)** by task 09's relay pipeline. not flap) + a concurrent `subscribeRepos` connection cap (`SYNC_MAX_SUBSCRIBERS`, reserve-then-check, 429 `SubscriberLimitExceeded`). -- `getRepo` buffers full CARs in memory; `ExportCAR` includes unreachable - historical blocks (consider reachable-set-only). +- ~~`getRepo` buffers full CARs in memory; `ExportCAR` includes unreachable + historical blocks (consider reachable-set-only).~~ **Closed by task 12**: + `repo.ExportCARTo` streams the CAR block-by-block (memory bounded by one + fetch batch, `walkBatchSize` = 256 blocks, plus a CID-string seen-set that + grows with the reachable set) and exports only the reachable + set from the current head (10.9x smaller on the 2k-record benchmark + fixture; correct CAR consumers traverse from the root, so omitting + superseded blocks is transparent — verified against indigo's + `LoadRepoFromCAR`). The sync handler streams; a pre-first-byte failure + still returns a clean 500, a mid-stream failure can only truncate + (logged). - ~~`PruneEvents` is one unbatched DELETE per hourly sweep.~~ **Closed by task 11**: batched (1000/statement) from the oldest seq up, so a partial sweep still leaves a contiguous retained suffix. -- MST loads are full-tree (one SELECT per node) → `PutRecord` is O(repo - size); needs a per-DID tree cache before big-community scale. +- ~~MST loads are full-tree (one SELECT per node) → `PutRecord` is O(repo + size); needs a per-DID tree cache before big-community scale.~~ **Closed + by task 12**: per-DID LRU tree cache (`internal/repo/treecache.go`, + `MST_CACHE_SIZE`, default 512 repos) — steady-state commit is an + in-memory mutation + diff write (42.6x faster on the 2k-record fixture). + Coherence model documented in that file: take() detaches (a failed commit + can never leave a mutated tree cached), put() only under a + durably-committed head, head validated against the value read under the + commit locks (a cross-process commit makes the entry stale, not wrong). + A cold first commit still pays one full-tree load. - `SigningKeys` could become a `SignCommit` capability (keeps key plaintext inside identity; enables KMS later) — revisit before the interface calcifies. - `getRepo`'s optional `since` parameter (diff export) is not implemented — - a `since` request gets the full CAR, which the spec permits (extra blocks - are legal); consumers needing incremental sync use subscribeRepos - (`internal/sync/server.go`). + a `since` request gets the full reachable-set CAR, which the spec permits + (extra blocks are legal); consumers needing incremental sync use + subscribeRepos (`internal/sync/server.go`). Deliberately left that way in + task 12: a real diff export would have to read historical blocks, which + the blocks GC invariant (internal/repo/gc.go) explicitly does not + guarantee — implementing it means revisiting that invariant, not just + adding a query. ## Ingestion (task 06 notes) @@ -294,8 +326,14 @@ harness, **(relay)** by task 09's relay pipeline. `materializeContent` checks it (`TestCreateAfterDeleteTombstone` pins the whole ordering, restore included). The README's claim was correct; this entry was the false doc. -- `ClaimNext` does an O(N) row scan when one community's queue backs up - behind a failing event (per-key serialization cost; revisit at scale). +- ~~`ClaimNext` does an O(N) row scan when one community's queue backs up + behind a failing event (per-key serialization cost; revisit at scale).~~ + **Closed by task 12**: the claim query is now a recursive loose index + scan over `idx_inbox_events_queue` — one index descent per distinct + pending ordering key, O(pending keys × log N) regardless of any key's + backlog depth (measured: 162.6 ms → 0.05 ms with a 50k-deep backed-up + key). Same semantics, same fencing contract, no schema change; + `TestInboxEvents_DeepBacklogDoesNotBlockOtherKeys` pins it. - A shutdown-interrupted attempt still consumes its ClaimNext attempt increment (cosmetic). - ~~`MAX_BLOB_BYTES` above 5 MiB is a silent no-op.~~ **Closed by task @@ -399,8 +437,20 @@ harness, **(relay)** by task 09's relay pipeline. row.~~ **Closed by task 11**: renamed to `key_material` (migration 013; the per-row encoding — plaintext PEM vs sealed ciphertext — is documented on `store.ServiceKey`). -- `blocks` is append-only with no GC (load-bearing for GetRecord read - consistency; revisit together with the getRepo memory item). +- ~~`blocks` is append-only with no GC (load-bearing for GetRecord read + consistency; revisit together with the getRepo memory item).~~ **Closed + by task 12**: `repo.GCBlocks` (shared `internal/prune` runner, 6h sweeps, + `BLOCKS_GC_RETENTION` default 72h) deletes blocks that are BOTH + unreachable from their repo's current head AND older than the retention + cutoff. The full invariant + the audit of every consumer is written at + the top of `internal/repo/gc.go` — the load-bearing parts are that + readers moved to REPEATABLE READ snapshots, firehose replay reads the + self-contained `firehose_events.car` (never `blocks`), and the commit + path's `ON CONFLICT ... DO UPDATE SET created_at = clock_timestamp()` + refresh makes the retention window the compute→delete race guard. If a + real `since` diff export ever gets served from `blocks` history, it must + revisit that invariant first (the fallback stays: `since` requests get + the full reachable-set CAR, which the spec permits). ## E2E harness itself (task 08) diff --git a/LOOP_STATE.md b/LOOP_STATE.md index 8b4940d..b5e5b63 100644 --- a/LOOP_STATE.md +++ b/LOOP_STATE.md @@ -34,7 +34,11 @@ write-back design; tasks 11–12 are its prerequisites). | 10 | 09-e2e-relay | done | (see git log) | 7/7 reviewers (4 Claude emulated + codex/gemini/glm); fixes: dev requestCrawl PUBLIC-relay dial guard (codex unique catch — NewPrivateOnlyHTTPClient, inverse SSRF guard), terminal-error classification made pre-flight-only (whole-chain IsValidation was abandoning a relay on attempt 1 for transient DNS), 10s per-attempt timeout (budget arithmetic was 14min worst-case, not 2min), vacuous validation-no-retry test rewritten + 400-is-retried pin, vetEvent per-DID rev-monotonicity (restores per-repo ordering assertion suite-wide), drain() returns+clears pending (closes task-10 vacuous-pass trap), relay poll robustness + pagination cap, doc corrections (RESOLVE_ADDRESS overstatement, spec BGS_CRAWL_INSECURE_WS annotation, FOLLOWUPS 16th-failure off-by-one). KEPT DELIBERATE over 3 reviewers' objection: all wire errors incl. 4xx retried — bigsky answers the describeServer callback race with HTTP 400 (comment + test pin it). Final clean make e2e: 10/10, 96.7s | | 11 | 10-e2e-scenarios | done | (see git log) | 6/7 reviewers (glm watchdog-killed); UNANIMOUS 6/6 finding: tombstone confirm-fetch transient failure → definitive 401 permanently lost legitimate account deletions → fixed with three-way taxonomy (tombstone→202, alive/validation/404→401, transport/5xx→503 defer) + test; codex unique: confirmation fetch followed cross-authority redirects (open-redirect → forged 410) → FetchActorSameAuthority pins every hop; security: unauthenticated durable-write path flagged → encoded into task 11 rate-limit spec; also: zz-sweep replay floor + honest bounds (sentinel-only pass was vacuous), Delete(Actor) over-scrub drain, actor!=object + Announce{Delete} 401 pins, GET / route-level test, cursor 0→1 doc fixes, vote-hammer header de-overclaimed. TASK ITSELF: 7 scenarios + 2 PRODUCTION fixes (apex instance actor — Lemmy silently never delivers Delete{Person} without a Site actor row; tombstone-verified self-delete acceptance). Final clean make e2e: 17/17, 239s | | 12 | 11-hardening | done | (see git log) | 6/7 reviewers (glm watchdog-killed on 5k-line diff); NO high-sev confirmed (gemini's "carry-forward type assertion always fails" was a FALSE POSITIVE — GetRecord returns typed atdata.Blob, test green). Fixes: FollowRetrier atomic UPDATE...RETURNING claim (list-then-update raced Accept + burned attempts on transient failure + silent exhaustion), rate-limit refusal observability (expvar counters + sampled Warn — mistuned limit silently dropped all traffic), community-DID blob orphan now retryable (was swallowed → served forever; required delete-before-soft-delete reorder), ScrubVoter DELETE...RETURNING recompute (phantom-count lost update), carry-forward drops on permanent 404/410 vs carries on transient, DeleteActor terminal-state fixpoint (no double #account), migration-011 CHECK tightened + raw-insert test, /admin/metrics scoped expvar, internal/prune fail-closed test, proxy XFF ops note. TASK: inbox+sync admission control, #account{active:false,status:deleted} frame verified purging repo from bigsky, follow auto-retry, 3 pruners, one-tx record+mapping, blob/vote scrubs, service_keys rename. delete-before-create: README was RIGHT, FOLLOWUPS stale (task 06 already closed it). Final clean make e2e: 17/17, 232s | -| 13 | 12-perf-scale | in-progress | | MST cache, getRepo streaming/reachable-set, blocks GC, ClaimNext scan | +| 13 | 12-perf-scale | done | (see git log) | 7/8 reviewers (5 Claude emulated + codex/gemini; glm watchdog-killed again); NO high-sev code findings — gemini zero-issue "excellent". Fixes: mid-stream getRepo failure now panics http.ErrAbortHandler so a truncated CAR is a transport-level failure, not a clean 200 (4/7 flagged — the diff's sharpest catch); client-disconnect logging downgraded to Debug; vanished-repo race → 404; FALSE parenthetical in the gc.go invariant header corrected ("every block a commit references is in newBlocks" → only NEWLY-referenced blocks are; believing the original justified deleting rule (a)); DELETE-time created_at re-check + RR isolation + 256-block batch boundary all made test-load-bearing with fail-then-pass proofs (pr-test-analyzer: "you could delete the race-guard clause and the suite stayed green" — no longer); CAR-slice identity test extended to update/delete ops + ordered comparison + no swallowed reader errors; ExportCARTo pre-first-byte doc overclaim fixed; app↔DB clock-skew assumption documented (codex unique). TASK: per-DID MST cache (PutRecord 140.6ms→3.30ms, 42.6x, on a 2k-record repo), streaming reachable-set getRepo (CAR 10.9x smaller, batch-bounded residency), blocks GC (invariant-first design in gc.go), ClaimNext loose index scan (162.6ms→0.050ms on a 50k backlog). Resumed from a killed predecessor agent's partial tree — its walk was correct but 1-SELECT-per-block (698ms/op, would have been a getRepo latency REGRESSION; batching fixed it). Final clean make e2e: 17/17 | + +v1.1 loop (tasks 09–12) COMPLETE — `make e2e` green (17/17), full unit +suite green, all FOLLOWUPS items scheduled into this loop closed or +explicitly deferred with rationale. Statuses: pending → in-progress → review → done (or blocked: ). @@ -133,10 +137,10 @@ and deferred TODOs here) session lock 0x7469646570 — keep them distinct). This guarantees seq order == commit-visibility order, so task 04 may tail with naive `WHERE seq > cursor` — any future writer bypassing repo.Manager breaks - that. Per-DID mutex + repo_state row lock remain as backstops. blocks - keeps superseded blocks (no GC; append-only is load-bearing for - GetRecord's read consistency); ExportCAR includes unreachable - historical blocks — task 04's getRepo may want reachable-set-only. + that. Per-DID mutex + repo_state row lock remain as backstops. + [SUPERSEDED by task 12: blocks now has GC (invariant in + internal/repo/gc.go), readers hold REPEATABLE READ snapshots instead + of relying on append-only, and ExportCAR is reachable-set-only.] - identity.Minter.MintActor mints did:plc via MODERN plc_operation genesis ops (indigo's plc package only has the deprecated legacy `create` op — don't use it): rotationKeys=[bridge escrow key], verificationMethods. @@ -598,3 +602,49 @@ and deferred TODOs here) dialBridgeFirehose/readBridgeAccountFrame in tests/e2e/helpers.go) and the repo DISAPPEARING from the relay's listRepos (polled; bigsky processes the frame async). + +### From task 12 (perf & scale — anything touching repo storage MUST know) +- **`blocks` is no longer append-only-forever.** The replacement invariant + lives at the top of internal/repo/gc.go (delete only if unreachable from + the current head in one REPEATABLE READ snapshot AND older than + BLOCKS_GC_RETENTION, created_at re-checked inside the DELETE). Every + new `blocks` reader must hold a REPEATABLE READ snapshot (GetRecord/ + GetRecordProof/ExportCARTo do; the commit path's loadTree is covered by + the FOR UPDATE head pin) — audit any new reader against gc.go's header. + `blocks.created_at` now means "last written" (ON CONFLICT refresh), not + "first written"; it is the GC race guard and only ever makes GC more + conservative. Retention must comfortably exceed app↔DB clock skew + a + sweep's compute→delete gap (72h default dwarfs both). +- **A future `since` diff export is now HARDER, not just missing**: it + would read historical blocks, which the GC invariant explicitly does + not guarantee. Documented in FOLLOWUPS; decide GC-interaction first. +- **Per-DID MST tree cache** (internal/repo/treecache.go): take() detaches + the entry and validates against the FOR UPDATE head (ABA impossible — + revs strictly increase and are embedded in the commit CID); re-cached + only after durable commit (or the provably-unchanged NoOp head). All + access under the per-DID mutex inside commitWrite ONLY — never add a + read-path consumer. repo.NewManager is now variadic (...Option, + WithTreeCacheSize; MST_CACHE_SIZE env, default 512; n <= 0 disables). + Cold first commit per DID still pays one full-tree load. +- **getRepo streams** (repo.ExportCARTo): reachable-set-only, batched + ANY() fetches (walkBatchSize 256 — a naive per-block walk measured + 698ms/op vs 27.7ms batched; don't "simplify" it away). Mid-stream + failure panics http.ErrAbortHandler so consumers see a transport error + instead of a clean truncated 200; client disconnects log at Debug. + walkReachable is SHARED between export and GC — reachability is + definitionally "what getRepo serves", they cannot drift. +- **ClaimNext is a recursive-CTE loose index scan** over + idx_inbox_events_queue. The ARRAY(...) head-materialization is + LOAD-BEARING (a plain IN regressed the planner to the O(N) scan — + EXPLAIN-verified, commented in inbox_events.go). Semantics unchanged: + per-key serial ordering, claimed_until fencing, SKIP LOCKED. +- **Perf ceilings that remain (for the votes-as-records design revisit):** + the GLOBAL commit advisory lock serializes ALL repos' commits — per-DID + throughput is now ~300 commits/s (3.3ms each) but it is one writer at a + time bridge-wide; firehose_events volume is untouched. Those are design + decisions, not storage perf — storage prerequisites for + votes-as-records now HOLD (FOLLOWUPS updated). +- Deferred: GC-vs-commit concurrency argued + mechanism-pinned, no + goroutine stress test; cold-start loadTree could reuse GetMany batching + if it ever matters; sync_test.go grew an exportCAR seam (mirrors + onUpgrade) for mid-stream failure injection. diff --git a/README.md b/README.md index c6b8e9c..4ddf660 100644 --- a/README.md +++ b/README.md @@ -210,6 +210,8 @@ production**: | `SEED_COUNTS_FROM_API` | on | seed backfilled posts' vote aggregates from the origin instance's public API (`/api/v3/post` `counts`); set `0` to disable | | `TOMBSTONE_RETENTION` | `720h` | how long `ap_tombstones` markers (the delete-before-create guard) are kept before the hourly pruner reclaims them | | `VOTE_EVENT_RETENTION` | `2160h` | how long **undone** (superseded/retracted) `vote_events` rows are kept; live rows are the counts and are never pruned | +| `BLOCKS_GC_RETENTION` | `72h` | how long superseded (head-unreachable) repo blocks are kept before the GC sweep (every 6h) reclaims them; the window doubles as the sweep's race guard, so keep it far above sweep duration — and comfortably above any app↔DB clock skew plus the sweep's compute→delete gap, since the cutoff comes from the app clock while `created_at` refreshes use the DB clock (see `internal/repo/gc.go`) | +| `MST_CACHE_SIZE` | `512` | per-DID MST tree cache entry cap (repos held as decoded in-memory trees on the commit path); memory scales with the cached repos' sizes, tune down when bridging many very large communities | | `INBOX_IP_RATE_PER_SECOND` / `INBOX_IP_RATE_BURST` | `50` / `200` | per-client-IP token bucket on `POST /inbox` (refusals are 503 — retryable for federation queues) | | `INBOX_SIGNER_RATE_PER_SECOND` / `INBOX_SIGNER_RATE_BURST` | `20` / `100` | per-verified-signer token bucket on `POST /inbox` | | `INBOX_TOMBSTONE_CONFIRMS_PER_MINUTE` / `INBOX_TOMBSTONE_CONFIRM_BURST` | `6` / `10` | dedicated per-IP cap on the tombstoned-self-delete confirmation branch (an unauthenticated POST that costs an outbound fetch + durable writes); over-limit deliveries defer (503) so legitimate deletions redeliver | diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index 05eecb1..50d1985 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -99,7 +99,8 @@ func run(logger *slog.Logger) error { if err != nil { return err } - repoManager, err := repo.NewManager(database, identity.NewActorKeys(actors, custodian), logger) + repoManager, err := repo.NewManager(database, identity.NewActorKeys(actors, custodian), logger, + repo.WithTreeCacheSize(cfg.MSTCacheSize)) if err != nil { return err } @@ -127,6 +128,11 @@ func run(logger *slog.Logger) error { // Firehose retention: prune events older than FIREHOSE_RETENTION so the // replay window (and the table) stays bounded. go tidepoolsync.RunPruner(ctx, repoManager, cfg.FirehoseRetention, 0, logger) + // Blocks GC (task 12): reclaim head-unreachable blocks older than + // BLOCKS_GC_RETENTION (the invariant lives in internal/repo/gc.go). A + // sweep walks every repo's live MST — heavier than the row pruners — so + // it runs every 6h instead of the runner's hourly default. + go prune.Run(ctx, "blocks(unreachable)", cfg.BlocksGCRetention, 6*time.Hour, repoManager.GCBlocks, logger) // Ask configured relays to crawl us. Development hosts are not publicly // reachable, so dev only logs what it would have sent (never touches a diff --git a/internal/config/config.go b/internal/config/config.go index 6e92c5c..c519c89 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -131,6 +131,17 @@ type Config struct { // duration, default 2160h = 90 days). Live rows are never pruned; see // votes.PruneUndoneEvents for the replay-dedupe trade-off. VoteEventRetention time.Duration + // BlocksGCRetention is how long superseded (head-unreachable) repo + // blocks are kept before the GC sweep reclaims them (BLOCKS_GC_RETENTION, + // a Go duration, default 72h). The window doubles as the sweep's race + // guard against concurrent commits — see internal/repo/gc.go — so it + // must stay far above one sweep's duration (seconds); days is right. + BlocksGCRetention time.Duration + // MSTCacheSize is the per-DID MST tree cache's entry cap (MST_CACHE_SIZE, + // default 512, must be positive). Each entry is one repo's fully decoded + // live tree, so memory scales with the cached repos' sizes; operators + // bridging many very large communities tune this down. + MSTCacheSize int // Inbox admission control (task 11): per-client-IP and per-verified- // signer token buckets on POST /inbox, plus the dedicated tighter cap // on the tombstoned-self-delete confirmation branch. All are generous @@ -331,6 +342,16 @@ func Load(logger *slog.Logger) (*Config, error) { if err != nil { return nil, err } + cfg.BlocksGCRetention, err = durationVar(logger, "BLOCKS_GC_RETENTION", 72*time.Hour) + if err != nil { + return nil, err + } + // 512 mirrors repo.DefaultTreeCacheSize (importing repo here would point + // the dependency the wrong way); if one moves, move the other. + cfg.MSTCacheSize, err = intVar(logger, "MST_CACHE_SIZE", 512) + if err != nil { + return nil, err + } // Admission-control knobs (task 11): tuning knobs with real defaults in // every environment, like the other rate limits. diff --git a/internal/repo/bench_streaming_test.go b/internal/repo/bench_streaming_test.go new file mode 100644 index 0000000..fc221ab --- /dev/null +++ b/internal/repo/bench_streaming_test.go @@ -0,0 +1,22 @@ +package repo + +// Streaming-path benchmark for the task-12 getRepo work. Separate from +// bench_test.go because ExportCARTo does not exist in the "before" tree that +// file also runs against. + +import ( + "io" + "testing" +) + +func BenchmarkExportCARTo_BigRepo(b *testing.B) { + manager := benchFixtureManager(b) + ctx := b.Context() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := manager.ExportCARTo(ctx, benchDID, io.Discard); err != nil { + b.Fatal(err) + } + } +} diff --git a/internal/repo/bench_test.go b/internal/repo/bench_test.go new file mode 100644 index 0000000..0d48d24 --- /dev/null +++ b/internal/repo/bench_test.go @@ -0,0 +1,105 @@ +package repo + +// Task-12 before/after benchmarks. This file deliberately uses only API that +// exists both before and after the task (NewManager's variadic options make +// the 3-arg call compile on both sides), so the identical file can run +// against a HEAD checkout for the "before" numbers — recorded in the task-12 +// commit message. One backport is still needed +// on the HEAD side: the two-line testutil.DB/testutil.Truncate *testing.T → +// testing.TB widening, which HEAD lacks (its *testing.T signatures reject +// this file's *testing.B callers). Run with a fixed iteration +// count so before/after repo growth matches: +// +// go test ./internal/repo -bench 'PutRecord_BigRepo|ExportCAR_BigRepo' \ +// -run '^$' -benchtime 20x -benchmem +// +// The streaming-path benchmark lives in bench_streaming_test.go (new API, +// working tree only). + +import ( + "context" + "fmt" + "sync" + "testing" + + "github.com/bluesky-social/indigo/atproto/atcrypto" + + "tidepool/internal/testutil" +) + +// benchRepoSize is the fixture size: a community repo a few months into +// bridging a mid-size Lemmy community (posts + comments). +const benchRepoSize = 2000 + +const benchDID = "did:plc:bbenchbenchbenchbenchbig" + +var ( + benchOnce sync.Once + benchShared *Manager + benchPreload error +) + +// benchFixtureManager returns ONE shared Manager (steady-state: whatever +// cache the build has is warm after the preload) over a repo preloaded with +// benchRepoSize records. The preload runs once per process; benchmarks that +// append use rkey indexes far above the fixture range. Run with -count=1 — +// a second in-process run would re-put identical records and measure the +// idempotent NoOp path instead. +func benchFixtureManager(b *testing.B) *Manager { + b.Helper() + database := testutil.DB(b) + benchOnce.Do(func() { + testutil.Truncate(b, database, "blocks", "repo_state", "firehose_events") + key, err := atcrypto.GeneratePrivateKeyK256() + if err != nil { + benchPreload = err + return + } + benchShared, err = NewManager(database, &staticKeys{key: key}, nil) + if err != nil { + benchPreload = err + return + } + ctx := context.Background() + for i := 0; i < benchRepoSize; i++ { + if _, err := benchShared.PutRecord(ctx, benchDID, testCollection, testRKey(i), + testRecord(fmt.Sprintf("fixture post %d", i))); err != nil { + benchPreload = fmt.Errorf("preload record %d: %w", i, err) + return + } + } + }) + if benchPreload != nil { + b.Fatal(benchPreload) + } + return benchShared +} + +func BenchmarkPutRecord_BigRepo(b *testing.B) { + manager := benchFixtureManager(b) + ctx := b.Context() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := manager.PutRecord(ctx, benchDID, testCollection, testRKey(1_000_000+i), + testRecord(fmt.Sprintf("bench post %d", i))); err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkExportCAR_BigRepo(b *testing.B) { + manager := benchFixtureManager(b) + ctx := b.Context() + b.ReportAllocs() + b.ResetTimer() + var carLen int + for i := 0; i < b.N; i++ { + carBytes, err := manager.ExportCAR(ctx, benchDID) + if err != nil { + b.Fatal(err) + } + carLen = len(carBytes) + } + b.ReportMetric(float64(carLen), "car-bytes") +} diff --git a/internal/repo/blocks.go b/internal/repo/blocks.go index 3fae48e..fc26e0d 100644 --- a/internal/repo/blocks.go +++ b/internal/repo/blocks.go @@ -9,6 +9,7 @@ import ( blockformat "github.com/ipfs/go-block-format" "github.com/ipfs/go-cid" ipld "github.com/ipfs/go-ipld-format" + "github.com/lib/pq" "github.com/multiformats/go-multihash" ) @@ -41,6 +42,36 @@ type txBlockSource struct { did string } +// GetMany reads a batch of blocks in one query, returning raw bytes keyed by +// CID string. Missing CIDs are simply absent from the map — callers decide +// whether a miss is fatal (the reachable-set walk treats it as corruption). +func (s *txBlockSource) GetMany(ctx context.Context, cids []cid.Cid) (map[string][]byte, error) { + strs := make([]string, len(cids)) + for i, c := range cids { + strs[i] = c.String() + } + rows, err := s.tx.QueryContext(ctx, + `SELECT cid, bytes FROM blocks WHERE did = $1 AND cid = ANY($2)`, + s.did, pq.Array(strs)) + if err != nil { + return nil, fmt.Errorf("repo: read %d blocks for %s: %w", len(cids), s.did, err) + } + defer rows.Close() + out := make(map[string][]byte, len(cids)) + for rows.Next() { + var cidStr string + var raw []byte + if err := rows.Scan(&cidStr, &raw); err != nil { + return nil, fmt.Errorf("repo: scan block for %s: %w", s.did, err) + } + out[cidStr] = raw + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("repo: iterate blocks for %s: %w", s.did, err) + } + return out, nil +} + func (s *txBlockSource) Get(ctx context.Context, c cid.Cid) (blockformat.Block, error) { var raw []byte err := s.tx.QueryRowContext(ctx, diff --git a/internal/repo/events.go b/internal/repo/events.go index 996159d..3e70dc9 100644 --- a/internal/repo/events.go +++ b/internal/repo/events.go @@ -6,9 +6,11 @@ import ( "database/sql" "encoding/json" "fmt" + "io" "strconv" indigorepo "github.com/bluesky-social/indigo/atproto/repo" + "github.com/bluesky-social/indigo/atproto/repo/mst" "github.com/bluesky-social/indigo/atproto/syntax" blockformat "github.com/ipfs/go-block-format" @@ -203,66 +205,186 @@ func writeCARSlice(root cid.Cid, blks []blockformat.Block) ([]byte, error) { return buf.Bytes(), nil } -// ExportCAR writes the DID's full repo as a CARv1 stream rooted at the -// current head commit. NOTE (task 04): until block garbage collection -// exists, this includes every historical block for the DID, not just the -// reachable set — harmless for CAR readers (they traverse from the root) -// but larger than a minimal export. +// ExportCAR writes the DID's full repo as a CARv1 byte slice rooted at the +// current head commit — the reachable set only (commit + MST nodes + record +// blocks), the same content ExportCARTo streams. It buffers the whole CAR in +// memory; the sync surface uses ExportCARTo to stream instead. A missing repo +// satisfies errors.IsNotFound. func (m *Manager) ExportCAR(ctx context.Context, did string) ([]byte, error) { - tx, err := m.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) + var buf bytes.Buffer + if err := m.ExportCARTo(ctx, did, &buf); err != nil { + return nil, err + } + return buf.Bytes(), nil +} + +// ExportCARTo streams the DID's repo as a CARv1 stream rooted at the current +// head commit, writing blocks to w as they are fetched so memory stays +// bounded by one fetch batch (walkBatchSize blocks) plus a CID-string +// seen-set that grows with the reachable set, rather than the whole CAR +// (com.atproto.sync.getRepo can serve a large community repo without +// buffering it). It exports the REACHABLE SET from the +// current head — the commit block, every MST node on the live tree, and every +// record block those nodes reference — NOT every historical block ever stored +// for the DID. Superseded MST nodes and old record versions are omitted: +// correct CAR consumers (indigo's LoadRepoFromCAR, bigsky) traverse from the +// root commit and never look at unreachable blocks, so the export is smaller +// but semantically identical. +// +// This reachable-set export is independent of the read-consistency guarantees +// GetRecord/GetRecordProof and subscribeRepos replay rely on: those read the +// current head's blocks (append-only, content-addressed) and the self- +// contained firehose_events.car respectively — neither depends on the +// unreachable historical blocks omitted here. The read runs in a REPEATABLE +// READ snapshot so a concurrent blocks GC (which deletes only unreachable +// blocks) can never pull a block out from under the walk. +// +// Failure ordering: the repo-state read, head-CID parse, head-block fetch, +// and commit decode all happen before the first byte is written, so a missing +// repo or an unreadable/undecodable head commit returns an error with nothing +// written to w. The reachable walk itself (batched block fetches, MST node +// decodes, the missing-block corruption check) runs after bytes are flowing: +// any failure there truncates the stream mid-write, and the HTTP handler is +// responsible for making that visible to the client (it cannot be turned into +// a clean error response once the header is out). +func (m *Manager) ExportCARTo(ctx context.Context, did string, w io.Writer) error { + tx, err := m.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelRepeatableRead}) if err != nil { - return nil, fmt.Errorf("repo: begin read tx: %w", err) + return fmt.Errorf("repo: begin read tx: %w", err) } defer func() { _ = tx.Rollback() }() state, err := readRepoState(ctx, tx, did, false) if err != nil { - return nil, err + return err } head, err := cid.Parse(state.headCID) if err != nil { - return nil, fmt.Errorf("repo: parse head cid %q for %s: %w", state.headCID, did, err) + return fmt.Errorf("repo: parse head cid %q for %s: %w", state.headCID, did, err) } - - var buf bytes.Buffer - if err := car.WriteHeader(&car.CarHeader{Roots: []cid.Cid{head}, Version: 1}, &buf); err != nil { - return nil, fmt.Errorf("repo: write CAR header: %w", err) - } - - // Head commit block first (readers conventionally expect the root - // early), then everything else. src := &txBlockSource{tx: tx, did: did} headBlk, err := src.Get(ctx, head) if err != nil { - return nil, fmt.Errorf("repo: read head commit %s for %s: %w", state.headCID, did, err) + return fmt.Errorf("repo: read head commit %s for %s: %w", state.headCID, did, err) } - if err := carutil.LdWrite(&buf, headBlk.Cid().Bytes(), headBlk.RawData()); err != nil { - return nil, fmt.Errorf("repo: write CAR block %s: %w", head, err) + var commit indigorepo.Commit + if err := commit.UnmarshalCBOR(bytes.NewReader(headBlk.RawData())); err != nil { + return fmt.Errorf("repo: decode head commit %s for %s: %w", state.headCID, did, err) } - rows, err := tx.QueryContext(ctx, - `SELECT cid, bytes FROM blocks WHERE did = $1 AND cid <> $2 ORDER BY created_at, cid`, - did, state.headCID) - if err != nil { - return nil, fmt.Errorf("repo: list blocks for %s: %w", did, err) + // From here on we write to w; any error truncates the stream. + if err := car.WriteHeader(&car.CarHeader{Roots: []cid.Cid{head}, Version: 1}, w); err != nil { + return fmt.Errorf("repo: write CAR header: %w", err) } - defer rows.Close() - for rows.Next() { - var cidStr string - var raw []byte - if err := rows.Scan(&cidStr, &raw); err != nil { - return nil, fmt.Errorf("repo: scan block for %s: %w", did, err) + // Head commit block first (readers conventionally expect the root early). + if err := carutil.LdWrite(w, head.Bytes(), headBlk.RawData()); err != nil { + return fmt.Errorf("repo: write CAR block %s: %w", head, err) + } + // seen dedupes blocks (a record CID can appear under multiple keys, and an + // MST subtree could be shared): it holds only CID strings, not bytes, so + // memory stays proportional to the number of distinct reachable blocks. + seen := map[string]struct{}{head.String(): {}} + return walkReachableBlocks(ctx, src, commit.Data, seen, w) +} + +// walkReachableBlocks streams the MST rooted at node — node blocks, then the +// record blocks their value entries reference, level by level — to w in CARv1 +// LdWrite framing, skipping anything already emitted (seen). +func walkReachableBlocks(ctx context.Context, src *txBlockSource, node cid.Cid, seen map[string]struct{}, w io.Writer) error { + writeBlock := func(c cid.Cid, raw []byte) error { + if err := carutil.LdWrite(w, c.Bytes(), raw); err != nil { + return fmt.Errorf("repo: write CAR block %s: %w", c, err) } - c, err := cid.Parse(cidStr) + return nil + } + return walkReachable(ctx, src, node, seen, writeBlock, + func(records []cid.Cid) error { + return forEachBlock(ctx, src, records, writeBlock) + }) +} + +// walkBatchSize is how many blocks one walk fetch pulls from postgres. The +// reachable-set walk is round-trip bound (a 2k-record repo is ~2.7k blocks), +// so blocks are fetched in batches of this size — it is also the walk's +// memory bound: at most walkBatchSize block payloads are resident at once. +const walkBatchSize = 256 + +// walkReachable is the reachable-set walk ExportCARTo and GCBlocks share: a +// breadth-first walk over the MST rooted at node, fetching each level's node +// blocks in walkBatchSize batches. visitNode receives every node block (with +// its raw bytes, in batch order); visitRecords receives the record CIDs each +// level references — bytes deliberately not fetched, because GC only needs +// the CIDs (the export's visitRecords fetches them itself, batched). Every +// reachable CID (nodes and records) is added to seen, which both dedupes the +// walk and, for GC, IS the reachable set. +func walkReachable(ctx context.Context, src *txBlockSource, node cid.Cid, seen map[string]struct{}, + visitNode func(cid.Cid, []byte) error, visitRecords func([]cid.Cid) error) error { + var level []cid.Cid + if _, ok := seen[node.String()]; !ok { + seen[node.String()] = struct{}{} + level = append(level, node) + } + for len(level) > 0 { + var children, records []cid.Cid + err := forEachBlock(ctx, src, level, func(c cid.Cid, raw []byte) error { + if err := visitNode(c, raw); err != nil { + return err + } + nd, err := mst.NodeDataFromCBOR(bytes.NewReader(raw)) + if err != nil { + return fmt.Errorf("repo: decode MST node %s: %w", c, err) + } + n := nd.Node(&c) + for i := range n.Entries { + e := &n.Entries[i] + if e.IsValue() && e.Value != nil { + if _, ok := seen[e.Value.String()]; !ok { + seen[e.Value.String()] = struct{}{} + records = append(records, *e.Value) + } + } + if e.IsChild() && e.ChildCID != nil { + if _, ok := seen[e.ChildCID.String()]; !ok { + seen[e.ChildCID.String()] = struct{}{} + children = append(children, *e.ChildCID) + } + } + } + return nil + }) if err != nil { - return nil, fmt.Errorf("repo: parse stored cid %q for %s: %w", cidStr, did, err) + return err } - if err := carutil.LdWrite(&buf, c.Bytes(), raw); err != nil { - return nil, fmt.Errorf("repo: write CAR block %s: %w", c, err) + if len(records) > 0 { + if err := visitRecords(records); err != nil { + return err + } } + level = children } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("repo: iterate blocks for %s: %w", did, err) + return nil +} + +// forEachBlock fetches the given CIDs' bytes in walkBatchSize batches (one +// SELECT ... = ANY per batch) and invokes fn for each in the order given. A +// CID with no stored block is an error: the walk only asks for blocks the +// head's tree references, so a miss is corruption, never skippable. +func forEachBlock(ctx context.Context, src *txBlockSource, cids []cid.Cid, fn func(cid.Cid, []byte) error) error { + for start := 0; start < len(cids); start += walkBatchSize { + batch := cids[start:min(start+walkBatchSize, len(cids))] + blocks, err := src.GetMany(ctx, batch) + if err != nil { + return err + } + for _, c := range batch { + raw, ok := blocks[c.String()] + if !ok { + return fmt.Errorf("repo: block %s missing for %s (corrupt repo?)", c, src.did) + } + if err := fn(c, raw); err != nil { + return err + } + } } - return buf.Bytes(), nil + return nil } diff --git a/internal/repo/gc.go b/internal/repo/gc.go new file mode 100644 index 0000000..16a10a8 --- /dev/null +++ b/internal/repo/gc.go @@ -0,0 +1,230 @@ +package repo + +// Blocks garbage collection (task 12). The commit path writes `blocks` +// append-only; GCBlocks below is the ONLY deleter. Before it existed the +// table simply grew forever ("append-only is load-bearing", task 03) — this +// file replaces that blanket rule with an explicit invariant every reader +// was audited against. +// +// THE INVARIANT: a block may be deleted only if it is BOTH +// +// (a) unreachable from its repo's current head — not the head commit +// block, not a node of the head's live MST, not a record the live +// tree references — as computed in one REPEATABLE READ snapshot, AND +// (b) older than the retention cutoff (blocks.created_at < cutoff), +// re-checked inside the DELETE statement itself. +// +// Why that is sufficient for every consumer of `blocks`: +// +// - GetRecord / GetRecordProof / ExportCARTo read the head pointer and +// walk its blocks inside a single REPEATABLE READ snapshot. A GC DELETE +// committing after that snapshot began is invisible to it (MVCC). A GC +// that committed before it only removed blocks unreachable from the +// head at GC time — and any block reachable from a LATER head was +// (re-)written by an intervening commit, which refreshes created_at +// (the ON CONFLICT DO UPDATE in commitWrite), so rule (b) kept it. +// - subscribeRepos replay reads firehose_events.car, which is +// self-contained (commit + diff + record bytes inline) and never +// touches `blocks`. Pruning superseded blocks cannot break replay. +// - The commit path itself loads the tree behind the current head, whose +// blocks are all reachable — rule (a) never touches them. +// - A commit racing the sweep can make a previously-unreachable block +// reachable again only by re-writing it: every block a commit NEWLY +// references — anything not already reachable from its parent head — +// is dirty in the MST diff or is the put's record block, so it IS in +// newBlocks. (Blocks the commit references but does NOT re-write were +// already reachable from the parent head, so rule (a) never made them +// victims.) The re-write stamps created_at = clock_timestamp() in the +// commit transaction. The DELETE re-evaluates created_at < cutoff +// under READ COMMITTED row-lock semantics, so whichever way the two +// serialize the block survives. This makes the retention window the +// race guard: it only has to exceed one sweep's compute→delete gap +// (seconds) and it defaults to days. One more assumption rides on it: +// the cutoff comes from the APP clock (prune.Run passes +// time.Now()-retention) while the refresh uses the DB's +// clock_timestamp(), so app↔DB clock skew must stay small relative to +// the retention window — retention must comfortably exceed skew plus +// the compute→delete gap (the 72h default dwarfs both). +// +// If getRepo ever grows a real `since` diff export served from `blocks` +// history (rather than firehose_events), that consumer breaks rule (a)'s +// "current head only" audit and this GC must learn about it first. + +import ( + "bytes" + "context" + "database/sql" + stderrors "errors" + "fmt" + "slices" + "time" + + indigorepo "github.com/bluesky-social/indigo/atproto/repo" + + "github.com/ipfs/go-cid" + "github.com/lib/pq" + + "tidepool/internal/errors" +) + +// gcDeleteBatchSize bounds one DELETE statement so a large unreachable +// backlog (first sweep after enabling GC) never holds row locks for a whole +// sweep — same discipline as pruneEventsBatchSize. +const gcDeleteBatchSize = 1000 + +// GCBlocks deletes blocks that are unreachable from their repo's current +// head AND older than cutoff (see the file-header invariant), returning the +// number deleted. It matches prune.Func and is wired through the shared +// internal/prune runner, which fails closed on a non-positive retention. +// One repo's failure (corrupt tree, vanished state) is logged and does not +// stop the sweep for the other repos; the joined error is returned so the +// runner still reports the sweep as failed. +func (m *Manager) GCBlocks(ctx context.Context, cutoff time.Time) (int64, error) { + rows, err := m.db.QueryContext(ctx, `SELECT did FROM repo_state ORDER BY did`) + if err != nil { + return 0, fmt.Errorf("repo: list repos for blocks GC: %w", err) + } + var dids []string + for rows.Next() { + var did string + if err := rows.Scan(&did); err != nil { + rows.Close() + return 0, fmt.Errorf("repo: scan repo did for blocks GC: %w", err) + } + dids = append(dids, did) + } + if err := stderrors.Join(rows.Err(), rows.Close()); err != nil { + return 0, fmt.Errorf("repo: iterate repos for blocks GC: %w", err) + } + + var total int64 + var errs []error + for _, did := range dids { + n, err := m.gcRepoBlocks(ctx, did, cutoff) + total += n + if err != nil { + if ctx.Err() != nil { + errs = append(errs, err) + break + } + m.logger.Error("blocks GC failed for repo", "did", did, "error", err) + errs = append(errs, fmt.Errorf("%s: %w", did, err)) + } + } + return total, stderrors.Join(errs...) +} + +// gcRepoBlocks sweeps one repo: compute the victim set in a snapshot +// (unreachableBlocks), then delete it in short batches outside that snapshot +// (deleteUnreachable). The two phases are separate methods so tests can +// interleave a re-introducing commit between them and pin the DELETE-time +// created_at re-check. +func (m *Manager) gcRepoBlocks(ctx context.Context, did string, cutoff time.Time) (int64, error) { + victims, err := m.unreachableBlocks(ctx, did, cutoff) + if err != nil || len(victims) == 0 { + return 0, err + } + return m.deleteUnreachable(ctx, did, victims, cutoff) +} + +// deleteUnreachable is gcRepoBlocks' delete phase: remove the +// previously-computed victims in bounded batches, re-checking created_at +// against cutoff inside each DELETE (rule (b) of the invariant). +func (m *Manager) deleteUnreachable(ctx context.Context, did string, victims []string, cutoff time.Time) (int64, error) { + // Sorted for determinism and debuggability (reproducible batch contents + // across runs) — NOT a lock-ordering guarantee: within one + // DELETE ... WHERE cid = ANY($2), postgres takes row locks in executor + // scan order, not array order, so a deadlock against a concurrent + // commit's inserts (or another sweep) stays possible in theory. Postgres + // aborts one side, the sweep reports the error, and the next run + // self-heals — that retry, not the sort, is the real guarantee. + slices.Sort(victims) + + var total int64 + for start := 0; start < len(victims); start += gcDeleteBatchSize { + batch := victims[start:min(start+gcDeleteBatchSize, len(victims))] + // created_at < cutoff re-checked HERE, not just in the snapshot: this + // is rule (b) of the invariant. A commit that re-introduced a victim + // block since the snapshot refreshed its created_at, and the DELETE's + // row-lock re-evaluation sees that refresh — the block survives. + res, err := m.db.ExecContext(ctx, ` + DELETE FROM blocks WHERE did = $1 AND cid = ANY($2) AND created_at < $3`, + did, pq.Array(batch), cutoff) + if err != nil { + return total, fmt.Errorf("repo: delete unreachable blocks for %s: %w", did, err) + } + n, err := res.RowsAffected() + if err != nil { + return total, fmt.Errorf("repo: delete unreachable blocks for %s: rows affected: %w", did, err) + } + total += n + } + return total, nil +} + +// unreachableBlocks returns the CIDs of blocks for did that are both older +// than cutoff and unreachable from the current head. The head read, the +// reachable-set walk, and the candidate listing all share one REPEATABLE +// READ snapshot, so the set is internally consistent; staleness against +// concurrent commits is what the created_at re-check in deleteUnreachable +// (and the commit path's created_at refresh) absorbs. +func (m *Manager) unreachableBlocks(ctx context.Context, did string, cutoff time.Time) ([]string, error) { + tx, err := m.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelRepeatableRead}) + if err != nil { + return nil, fmt.Errorf("repo: begin read tx: %w", err) + } + defer func() { _ = tx.Rollback() }() + + state, err := readRepoState(ctx, tx, did, false) + if errors.IsNotFound(err) { + // Repo vanished between the sweep's listing and now; its blocks are + // left alone (conservative — nothing without a head is ever swept). + return nil, nil + } + if err != nil { + return nil, err + } + head, err := cid.Parse(state.headCID) + if err != nil { + return nil, fmt.Errorf("repo: parse head cid %q for %s: %w", state.headCID, did, err) + } + src := &txBlockSource{tx: tx, did: did} + headBlk, err := src.Get(ctx, head) + if err != nil { + return nil, fmt.Errorf("repo: read head commit %s for %s: %w", state.headCID, did, err) + } + var commit indigorepo.Commit + if err := commit.UnmarshalCBOR(bytes.NewReader(headBlk.RawData())); err != nil { + return nil, fmt.Errorf("repo: decode head commit %s for %s: %w", state.headCID, did, err) + } + + reachable := map[string]struct{}{head.String(): {}} + // The walk populates `reachable` (its seen set) as a side effect; record + // bytes are never fetched — GC only needs the CIDs. + if err := walkReachable(ctx, src, commit.Data, reachable, + func(cid.Cid, []byte) error { return nil }, + func([]cid.Cid) error { return nil }); err != nil { + return nil, err + } + + rows, err := tx.QueryContext(ctx, + `SELECT cid FROM blocks WHERE did = $1 AND created_at < $2`, did, cutoff) + if err != nil { + return nil, fmt.Errorf("repo: list expired blocks for %s: %w", did, err) + } + defer rows.Close() + var victims []string + for rows.Next() { + var c string + if err := rows.Scan(&c); err != nil { + return nil, fmt.Errorf("repo: scan expired block for %s: %w", did, err) + } + if _, ok := reachable[c]; !ok { + victims = append(victims, c) + } + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("repo: iterate expired blocks for %s: %w", did, err) + } + return victims, nil +} diff --git a/internal/repo/gc_test.go b/internal/repo/gc_test.go new file mode 100644 index 0000000..ec60e78 --- /dev/null +++ b/internal/repo/gc_test.go @@ -0,0 +1,413 @@ +package repo + +import ( + "bytes" + "database/sql" + "fmt" + "io" + "testing" + "time" + + indigorepo "github.com/bluesky-social/indigo/atproto/repo" + + "github.com/ipfs/go-cid" + car "github.com/ipld/go-car" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/errors" +) + +func countBlocks(t *testing.T, database *sql.DB, did string) int { + t.Helper() + var n int + require.NoError(t, database.QueryRowContext(t.Context(), + `SELECT COUNT(*) FROM blocks WHERE did = $1`, did).Scan(&n)) + return n +} + +func blockExists(t *testing.T, database *sql.DB, did, cid string) bool { + t.Helper() + var n int + require.NoError(t, database.QueryRowContext(t.Context(), + `SELECT COUNT(*) FROM blocks WHERE did = $1 AND cid = $2`, did, cid).Scan(&n)) + return n == 1 +} + +// churn builds a repo with garbage: creates, updates, and a delete, so the +// blocks table holds superseded record versions, dead MST nodes, and old +// commit blocks alongside the live tree. +func churn(t *testing.T, manager *Manager, did string) (liveRecordCID, deadRecordCID string) { + t.Helper() + ctx := t.Context() + for i := 0; i < 8; i++ { + _, err := manager.PutRecord(ctx, did, testCollection, testRKey(i), testRecord(fmt.Sprintf("post %d", i))) + require.NoError(t, err) + } + v1, err := manager.PutRecord(ctx, did, testCollection, testRKey(20), testRecord("v1")) + require.NoError(t, err) + v2, err := manager.PutRecord(ctx, did, testCollection, testRKey(20), testRecord("v2")) + require.NoError(t, err) + _, err = manager.DeleteRecord(ctx, did, testCollection, testRKey(7)) + require.NoError(t, err) + return v2.RecordCID, v1.RecordCID +} + +// verifyRepoIntact asserts every reader still works after a sweep: GetRecord, +// GetRecordProof, and a full export that indigo loads and verifies. +func verifyRepoIntact(t *testing.T, manager *Manager, did string, liveRKeys []int) { + t.Helper() + ctx := t.Context() + for _, i := range liveRKeys { + _, _, err := manager.GetRecord(ctx, did, testCollection, testRKey(i)) + require.NoError(t, err, "GetRecord %d after GC", i) + _, err = manager.GetRecordProof(ctx, did, testCollection, testRKey(i)) + require.NoError(t, err, "GetRecordProof %d after GC", i) + } + carBytes, err := manager.ExportCAR(ctx, did) + require.NoError(t, err) + commit, loaded, err := indigorepo.LoadRepoFromCAR(ctx, bytes.NewReader(carBytes)) + require.NoError(t, err, "repo must load from CAR after GC") + require.NoError(t, commit.VerifyStructure()) + root, err := loaded.MST.RootCID() + require.NoError(t, err) + assert.Equal(t, commit.Data.String(), root.String()) +} + +func TestGCBlocks_ReclaimsUnreachableOnly(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + liveCID, deadCID := churn(t, manager, testDID) + before := countBlocks(t, database, testDID) + require.True(t, blockExists(t, database, testDID, deadCID), "superseded record block present before GC") + + // Everything is older than a future cutoff, so only reachability + // protects blocks — the sharpest version of rule (a). + deleted, err := manager.GCBlocks(ctx, time.Now().Add(time.Minute)) + require.NoError(t, err) + assert.Positive(t, deleted, "churn must have produced unreachable blocks") + assert.Equal(t, before-int(deleted), countBlocks(t, database, testDID)) + + assert.False(t, blockExists(t, database, testDID, deadCID), "superseded record version must be reclaimed") + assert.True(t, blockExists(t, database, testDID, liveCID), "live record must survive") + head, _, err := manager.Head(ctx, testDID) + require.NoError(t, err) + assert.True(t, blockExists(t, database, testDID, head), "head commit must survive") + + verifyRepoIntact(t, manager, testDID, []int{0, 1, 2, 3, 4, 5, 6, 20}) + _, _, err = manager.GetRecord(ctx, testDID, testCollection, testRKey(7)) + require.Error(t, err) + assert.True(t, errors.IsNotFound(err), "deleted record stays deleted") + + // A second sweep over a clean table is a no-op. + deleted, err = manager.GCBlocks(ctx, time.Now().Add(time.Minute)) + require.NoError(t, err) + assert.Zero(t, deleted) + + // And the repo keeps committing normally afterwards. + _, err = manager.PutRecord(ctx, testDID, testCollection, testRKey(30), testRecord("post-GC write")) + require.NoError(t, err) +} + +func TestGCBlocks_RetentionFloorProtectsEverything(t *testing.T) { + manager, database, _ := testManager(t) + + churn(t, manager, testDID) + before := countBlocks(t, database, testDID) + + // Cutoff in the past: every block is younger, rule (b) keeps them all + // no matter how unreachable. + deleted, err := manager.GCBlocks(t.Context(), time.Now().Add(-time.Hour)) + require.NoError(t, err) + assert.Zero(t, deleted, "retention floor must protect young blocks") + assert.Equal(t, before, countBlocks(t, database, testDID)) +} + +func TestGCBlocks_RepoIsolation(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + churn(t, manager, testDID) + // The other repo has no garbage: one clean commit. + _, err := manager.PutRecord(ctx, testOtherDID, testCollection, testRKey(0), testRecord("clean")) + require.NoError(t, err) + otherBefore := countBlocks(t, database, testOtherDID) + + deleted, err := manager.GCBlocks(ctx, time.Now().Add(time.Minute)) + require.NoError(t, err) + assert.Positive(t, deleted) + assert.Equal(t, otherBefore, countBlocks(t, database, testOtherDID), + "a repo with no unreachable blocks must be untouched") + verifyRepoIntact(t, manager, testOtherDID, []int{0}) +} + +// TestGCBlocks_CommitRefreshesCreatedAt pins the commit path's ON CONFLICT +// created_at refresh — the mechanism that closes the compute→delete race in +// the invariant's rule (b). A block whose CID re-enters the live tree via a +// new commit must read as freshly written, so a sweep whose reachability +// snapshot predated that commit still cannot delete it. +func TestGCBlocks_CommitRefreshesCreatedAt(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + v1, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + v2, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v2")) + require.NoError(t, err) + require.NotEqual(t, v1.RecordCID, v2.RecordCID) + + // Age every block far past any cutoff. + _, err = database.ExecContext(ctx, + `UPDATE blocks SET created_at = TIMESTAMPTZ '2000-01-01' WHERE did = $1`, testDID) + require.NoError(t, err) + + // Reverting to v1's exact content re-writes v1's record block (same CID, + // identical bytes) inside the new commit — its created_at must refresh. + v3, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + require.Equal(t, v1.RecordCID, v3.RecordCID, "reverted content must reproduce the CID") + + var ageSeconds float64 + require.NoError(t, database.QueryRowContext(ctx, + `SELECT EXTRACT(EPOCH FROM now() - created_at) FROM blocks WHERE did = $1 AND cid = $2`, + testDID, v1.RecordCID).Scan(&ageSeconds)) + assert.Less(t, ageSeconds, 60.0, "re-written block must read as freshly created") + + // A sweep with a recent cutoff now reclaims v2's superseded block (old + // AND unreachable) but must keep the re-written v1 block despite the + // backdating (rule (b) via the refresh). + deleted, err := manager.GCBlocks(ctx, time.Now().Add(-time.Second)) + require.NoError(t, err) + assert.Positive(t, deleted) + assert.False(t, blockExists(t, database, testDID, v2.RecordCID)) + assert.True(t, blockExists(t, database, testDID, v1.RecordCID)) + + rec, _, err := manager.GetRecord(ctx, testDID, testCollection, testRKey(0)) + require.NoError(t, err) + assert.Equal(t, "v1", rec["text"]) + verifyRepoIntact(t, manager, testDID, []int{0}) +} + +// TestGCBlocks_DeleteRecheckSavesReintroducedBlock pins the DELETE-time +// `AND created_at < cutoff` re-check — rule (b) of the invariant, and the +// only thing standing between a stale victim snapshot and a live block. It +// drives the exact race interleaving: compute victims in a snapshot, +// re-introduce a victim via a real commit (same CID re-enters the live tree; +// the commit's ON CONFLICT refreshes created_at), then run the delete phase +// with the STALE victim list. The re-introduced block must survive. Delete +// the clause from deleteUnreachable's DELETE and this test fails. +func TestGCBlocks_DeleteRecheckSavesReintroducedBlock(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + v1, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + _, err = manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v2")) + require.NoError(t, err) + // Backdate everything so rule (b) alone cannot save the victims. + _, err = database.ExecContext(ctx, + `UPDATE blocks SET created_at = TIMESTAMPTZ '2000-01-01' WHERE did = $1`, testDID) + require.NoError(t, err) + + // Phase 1 (snapshot): v1's superseded record block is a legitimate victim. + cutoff := time.Now() + victims, err := manager.unreachableBlocks(ctx, testDID, cutoff) + require.NoError(t, err) + require.Contains(t, victims, v1.RecordCID, "superseded block must be a victim at snapshot time") + + // The racing commit: reverting to v1's content re-writes v1's record + // block (same CID) into the live tree and refreshes its created_at. + v3, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + require.Equal(t, v1.RecordCID, v3.RecordCID, "reverted content must reproduce the CID") + + // Phase 2 (delete) runs with the stale victim list: the re-introduced + // block must survive on the re-check while genuinely dead victims go. + deleted, err := manager.deleteUnreachable(ctx, testDID, victims, cutoff) + require.NoError(t, err) + assert.Positive(t, deleted, "stale-but-still-dead victims must still be reclaimed") + assert.True(t, blockExists(t, database, testDID, v1.RecordCID), + "re-introduced block must survive the stale delete phase") + verifyRepoIntact(t, manager, testDID, []int{0}) + + // Control: the identical interleaving WITHOUT the re-introducing commit + // deletes the block — proving the survival above is the re-check at + // work, not the victim simply never being deletable. + c1, err := manager.PutRecord(ctx, testOtherDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + _, err = manager.PutRecord(ctx, testOtherDID, testCollection, testRKey(0), testRecord("v2")) + require.NoError(t, err) + _, err = database.ExecContext(ctx, + `UPDATE blocks SET created_at = TIMESTAMPTZ '2000-01-01' WHERE did = $1`, testOtherDID) + require.NoError(t, err) + controlCutoff := time.Now() + controlVictims, err := manager.unreachableBlocks(ctx, testOtherDID, controlCutoff) + require.NoError(t, err) + require.Contains(t, controlVictims, c1.RecordCID) + _, err = manager.deleteUnreachable(ctx, testOtherDID, controlVictims, controlCutoff) + require.NoError(t, err) + assert.False(t, blockExists(t, database, testOtherDID, c1.RecordCID), + "without re-introduction the victim must be deleted") +} + +// TestGCBlocks_BigRepoCrossesWalkBatchSize pins the reachable-set walk across +// the walkBatchSize=256 batching boundary: with ~300 live records, one MST +// level references >256 record blocks, so forEachBlock must split the fetch +// into multiple batches. An off-by-one that drops a batch element would make +// a reachable block invisible to the walk — GC would delete it — and the +// post-sweep full-integrity verification (export → indigo load → verify, plus +// per-record reads) catches exactly that. The same walk feeds getRepo +// exports, so this is also the scale test for that surface. +func TestGCBlocks_BigRepoCrossesWalkBatchSize(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + const n = 300 // > walkBatchSize=256 record blocks on the MST's bottom level + rkeys := make([]int, 0, n) + for i := 0; i < n; i++ { + _, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(i), testRecord(fmt.Sprintf("post %d", i))) + require.NoError(t, err) + rkeys = append(rkeys, i) + } + before := countBlocks(t, database, testDID) + require.Greater(t, before, walkBatchSize, + "fixture must be large enough to force multi-batch walks") + + // Future cutoff: reachability alone protects blocks, at scale. + deleted, err := manager.GCBlocks(ctx, time.Now().Add(time.Minute)) + require.NoError(t, err) + assert.Positive(t, deleted, "300 sequential commits must leave unreachable garbage") + assert.Equal(t, before-int(deleted), countBlocks(t, database, testDID)) + + verifyRepoIntact(t, manager, testDID, rkeys) +} + +// TestGCBlocks_SnapshotReaderSurvivesConcurrentSweep pins the isolation-level +// half of the invariant's reader audit: a REPEATABLE READ snapshot opened +// before a sweep keeps seeing the OLD head's entire block set even after GC +// (running against a newer head) deletes those blocks. This is the MVCC +// argument the file header makes for GetRecord/GetRecordProof/ExportCARTo — +// it fails if the reader transaction is demoted to READ COMMITTED, where each +// statement would see the committed deletes mid-walk. +func TestGCBlocks_SnapshotReaderSurvivesConcurrentSweep(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + _, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + oldHead, _, err := manager.Head(ctx, testDID) + require.NoError(t, err) + + // The reader: a REPEATABLE READ read-only tx, exactly what the audited + // readers open. Its first query pins the snapshot at the old head. + tx, err := database.BeginTx(ctx, &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelRepeatableRead}) + require.NoError(t, err) + defer func() { _ = tx.Rollback() }() + snapState, err := readRepoState(ctx, tx, testDID, false) + require.NoError(t, err) + require.Equal(t, oldHead, snapState.headCID, "snapshot must be pinned at the old head") + + // Advance the repo past the snapshot: the old head's commit block, MST + // root, and superseded record versions all become unreachable garbage... + for _, text := range []string{"v2", "v3"} { + _, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord(text)) + require.NoError(t, err) + } + // ...which a sweep with a future cutoff reclaims, including the old head. + deleted, err := manager.GCBlocks(ctx, time.Now().Add(time.Minute)) + require.NoError(t, err) + require.Positive(t, deleted) + require.False(t, blockExists(t, database, testDID, oldHead), + "the old head commit block must actually be swept") + + // Inside the still-open snapshot, the old head must remain fully + // walkable: commit block, every MST node, every record block. Under + // READ COMMITTED this walk errors on the first swept block. + oldHeadCID, err := cid.Parse(oldHead) + require.NoError(t, err) + src := &txBlockSource{tx: tx, did: testDID} + headBlk, err := src.Get(ctx, oldHeadCID) + require.NoError(t, err, "snapshot must still see the swept head commit block") + var commit indigorepo.Commit + require.NoError(t, commit.UnmarshalCBOR(bytes.NewReader(headBlk.RawData()))) + seen := map[string]struct{}{oldHead: {}} + fetched := 0 + err = walkReachable(ctx, src, commit.Data, seen, + func(cid.Cid, []byte) error { fetched++; return nil }, + func(records []cid.Cid) error { + return forEachBlock(ctx, src, records, func(cid.Cid, []byte) error { + fetched++ + return nil + }) + }) + require.NoError(t, err, "old head's full block set must stay visible to the snapshot") + assert.Equal(t, len(seen), fetched+1, "every reachable CID must have been fetched (head via Get)") +} + +// TestGCBlocks_SkipsRepolessDIDs: blocks with no repo_state row (impossible +// by construction, conceivable after manual surgery) are never swept — +// nothing without a head is ever considered. +func TestGCBlocks_SkipsRepolessDIDs(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + _, err := database.ExecContext(ctx, ` + INSERT INTO blocks (did, cid, bytes, created_at) + VALUES ($1, $2, $3, TIMESTAMPTZ '2000-01-01')`, + testDID, "bafyreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku", []byte{0x01}) + require.NoError(t, err) + + deleted, err := manager.GCBlocks(ctx, time.Now()) + require.NoError(t, err) + assert.Zero(t, deleted) + assert.Equal(t, 1, countBlocks(t, database, testDID)) +} + +// TestExportCAR_OmitsUnreachableBlocks pins the reachable-set export: a +// superseded record version stays in `blocks` (until GC) but must not ride +// getRepo responses. +func TestExportCAR_OmitsUnreachableBlocks(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + v1, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + v2, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v2")) + require.NoError(t, err) + require.True(t, blockExists(t, database, testDID, v1.RecordCID), "superseded block still stored") + + carBytes, err := manager.ExportCAR(ctx, testDID) + require.NoError(t, err) + reader, err := car.NewCarReader(bytes.NewReader(carBytes)) + require.NoError(t, err) + exported := map[string]bool{} + for { + blk, err := reader.Next() + if err == io.EOF { + break + } + require.NoError(t, err, "CAR read must end at EOF, not a real error") + exported[blk.Cid().String()] = true + } + assert.True(t, exported[v2.RecordCID], "live record block must be exported") + assert.False(t, exported[v1.RecordCID], "superseded record block must not be exported") + assert.False(t, exported[v1.CommitCID], "old commit block must not be exported") + assert.True(t, exported[v2.CommitCID], "head commit block must be exported") + + _, _, err = indigorepo.LoadRepoFromCAR(ctx, bytes.NewReader(carBytes)) + require.NoError(t, err) +} + +// TestExportCARTo_MissingRepoWritesNothing pins the streaming contract the +// sync handler relies on: every failable read happens before the first byte, +// so a 404/500 can still be sent. +func TestExportCARTo_MissingRepoWritesNothing(t *testing.T) { + manager, _, _ := testManager(t) + + var buf bytes.Buffer + err := manager.ExportCARTo(t.Context(), testDID, &buf) + require.Error(t, err) + assert.True(t, errors.IsNotFound(err)) + assert.Zero(t, buf.Len(), "no bytes may reach the writer before the head is validated") +} diff --git a/internal/repo/repo.go b/internal/repo/repo.go index f61c81f..07e0db4 100644 --- a/internal/repo/repo.go +++ b/internal/repo/repo.go @@ -90,11 +90,31 @@ type Manager struct { // locks is never evicted: it is bounded by the number of bridged // actors, and a stale mutex per DID is 8 bytes of pointer. locks map[string]*sync.Mutex + + // treeCache holds decoded MST trees per DID so the common commit path + // skips the O(repo size) full-tree reload (one SELECT per node). Only + // the write path touches it, and only under the per-DID lock — see + // treecache.go for the coherence model. + treeCache *mstTreeCache +} + +// Option customizes a Manager at construction. Options are additive; the +// zero-option NewManager keeps the previous behavior with a default-sized MST +// tree cache. +type Option func(*Manager) + +// WithTreeCacheSize sets the maximum number of per-DID MST trees held in +// memory (LRU eviction beyond it). Passing n <= 0 disables the cache +// entirely (correct, just unoptimized). Only OMITTING the option uses +// DefaultTreeCacheSize — there is no in-band "default" value. +func WithTreeCacheSize(n int) Option { + return func(m *Manager) { m.treeCache = newTreeCache(n) } } // NewManager builds the repo manager. db and keys must be non-nil; a nil -// logger falls back to slog.Default(). -func NewManager(db *sql.DB, keys SigningKeys, logger *slog.Logger) (*Manager, error) { +// logger falls back to slog.Default(). Without options it uses an MST tree +// cache of DefaultTreeCacheSize entries. +func NewManager(db *sql.DB, keys SigningKeys, logger *slog.Logger, opts ...Option) (*Manager, error) { if db == nil { return nil, errors.NewValidationError("db", "must not be nil") } @@ -104,12 +124,17 @@ func NewManager(db *sql.DB, keys SigningKeys, logger *slog.Logger) (*Manager, er if logger == nil { logger = slog.Default() } - return &Manager{ - db: db, - keys: keys, - logger: logger, - locks: make(map[string]*sync.Mutex), - }, nil + m := &Manager{ + db: db, + keys: keys, + logger: logger, + locks: make(map[string]*sync.Mutex), + treeCache: newTreeCache(DefaultTreeCacheSize), + } + for _, opt := range opts { + opt(m) + } + return m, nil } // CommitResult reports what a successful PutRecord/DeleteRecord did. @@ -181,13 +206,16 @@ func (m *Manager) GetRecord(ctx context.Context, did, collection, rkey string) ( return nil, "", err } - // Note this tx runs READ COMMITTED, i.e. per-statement snapshots — it - // does NOT freeze one snapshot across the reads below. Consistency - // actually rests on blocks being content-addressed and append-only: - // once the head pointer is read, every block it references is immutable - // and present. Future block GC must preserve that property for any head - // a reader may still hold. - tx, err := m.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) + // This tx runs REPEATABLE READ: one snapshot frozen at the first read + // below, so head and every block reachable from it come from a single + // consistent point in time. Blocks are content-addressed and append-only, + // so once the head is read every block it references is present — and the + // snapshot makes that hold even against a concurrent blocks GC: a GC + // DELETE that commits after this snapshot is invisible here, and a GC that + // committed before it only removed blocks unreachable from a head at or + // after this snapshot's head (see GCBlocks). Either way the walk sees a + // complete tree. + tx, err := m.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelRepeatableRead}) if err != nil { return nil, "", fmt.Errorf("repo: begin read tx: %w", err) } @@ -308,12 +336,23 @@ func (m *Manager) commitWrite(ctx context.Context, did, collection, rkey string, tree = &empty } else { prevRev = state.rev - var headData cid.Cid - tree, headData, err = loadTree(ctx, tx, did, state.headCID) - if err != nil { - return nil, err + // Fast path: reuse the in-memory tree for this head if it is cached. + // take() detaches the entry; it is re-cached only under a + // durably-committed head (the successful-commit and NoOp paths below), + // so a commit that errors or rolls back leaves no cached tree and the + // next commit reloads a correct one from postgres. + if cached, root, ok := m.treeCache.take(did, state.headCID); ok { + tree = cached + headData := root + prevData = &headData + } else { + var headData cid.Cid + tree, headData, err = loadTree(ctx, tx, did, state.headCID) + if err != nil { + return nil, err + } + prevData = &headData } - prevData = &headData } // Note: indigo's mst.Tree.Remove returns (nil, nil) for a missing key — @@ -336,6 +375,14 @@ func (m *Manager) commitWrite(ctx context.Context, did, collection, rkey string, Rev: prevRev, NoOp: true, } + // Re-cache the (unmodified) tree under the unchanged head: an + // identical re-put is an indigo no-op that leaves the tree untouched, + // so it still exactly represents state.headCID / *prevData. The + // repo_state head is unchanged whatever the side-effect tx does, so + // caching here is valid even if the write-free commit below fails. + if prevData != nil { + m.cacheTree(did, state.headCID, *prevData, tree) + } if sideEffect != nil { // The side effect still runs (bookkeeping refresh) and must still // be durable, so the — otherwise write-free — transaction commits. @@ -399,8 +446,16 @@ func (m *Manager) commitWrite(ctx context.Context, did, collection, rkey string, } for _, blk := range newBlocks.ordered() { + // ON CONFLICT refreshes created_at rather than DO NOTHING: blocks are + // content-addressed so the bytes are identical, but the timestamp is + // the GC retention floor (see gc.go). A block re-written by this commit + // — e.g. an MST node whose CID reappears after churn — must read as + // "written now" so the GC floor protects it even if an in-flight sweep + // computed it as unreachable from an older head. Refreshing created_at + // can only make GC MORE conservative, never delete a live block sooner. if _, err := tx.ExecContext(ctx, - `INSERT INTO blocks (did, cid, bytes) VALUES ($1, $2, $3) ON CONFLICT (did, cid) DO NOTHING`, + `INSERT INTO blocks (did, cid, bytes) VALUES ($1, $2, $3) + ON CONFLICT (did, cid) DO UPDATE SET created_at = clock_timestamp()`, did, blk.Cid().String(), blk.RawData()); err != nil { return nil, fmt.Errorf("repo: store block %s: %w", blk.Cid(), err) } @@ -451,11 +506,28 @@ func (m *Manager) commitWrite(ctx context.Context, did, collection, rkey string, return nil, fmt.Errorf("repo: commit tx for %s: %w", did, err) } + // The commit is durable: the mutated tree now exactly represents the new + // head, so cache it for the next commit (skips a full-tree reload). Cached + // only after a successful Commit — never for a rolled-back write. + m.cacheTree(did, commitCID.String(), *newRoot, tree) + m.logger.Debug("repo commit", "did", did, "rev", rev.String(), "commit", commitCID.String(), "path", path, "seq", seq) return res, nil } +// cacheTree installs a decoded tree into the per-DID MST cache for reuse by the +// next commit. head/root MUST be the durably-committed head and its MST root. +// A partial tree (a block was missing at load, leaving CID stubs) is never +// cached — a later mutation would need to fault those stubs in from a store +// that is no longer available. +func (m *Manager) cacheTree(did, head string, root cid.Cid, tree *mst.Tree) { + if tree == nil || tree.IsPartial() { + return + } + m.treeCache.put(did, head, root, tree) +} + // lockFor returns the per-DID write mutex, creating it on first use. func (m *Manager) lockFor(did string) *sync.Mutex { m.mu.Lock() diff --git a/internal/repo/sync.go b/internal/repo/sync.go index bb3468e..57ea26f 100644 --- a/internal/repo/sync.go +++ b/internal/repo/sync.go @@ -247,10 +247,12 @@ func (m *Manager) GetRecordProof(ctx context.Context, did, collection, rkey stri return nil, err } - // Read consistency: same reasoning as GetRecord — blocks are - // content-addressed and append-only, so once the head is read every - // block it references is immutable and present. - tx, err := m.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) + // Read consistency: same reasoning as GetRecord — a REPEATABLE READ + // snapshot froze at the first read, so the head and its proof-path blocks + // are consistent and immune to a concurrent blocks GC (append-only, + // content-addressed blocks; GC only removes blocks unreachable from a head + // at or after this snapshot — see GCBlocks). + tx, err := m.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelRepeatableRead}) if err != nil { return nil, fmt.Errorf("repo: begin read tx: %w", err) } diff --git a/internal/repo/treecache.go b/internal/repo/treecache.go new file mode 100644 index 0000000..8ddb5eb --- /dev/null +++ b/internal/repo/treecache.go @@ -0,0 +1,144 @@ +package repo + +import ( + "container/list" + "sync" + + "github.com/bluesky-social/indigo/atproto/repo/mst" + "github.com/ipfs/go-cid" +) + +// DefaultTreeCacheSize is the number of per-DID MST trees kept in memory when +// no explicit size is configured. Each entry holds one fully-decoded tree, so +// memory scales with the sizes of the cached repos, not just the count — +// operators bridging many very large communities should tune MST_CACHE_SIZE +// down (or up, to keep more hot repos resident). +const DefaultTreeCacheSize = 512 + +// mstTreeCache is a bounded, LRU per-DID cache of decoded MST trees, keyed by +// DID and validated against the commit head the tree represents. It exists to +// make PutRecord cheap: loading a repo's tree from postgres is one SELECT per +// MST node (O(repo size) round-trips), so a big community's every commit paid +// a full-tree reload. Caching the live tree turns the steady-state commit into +// an in-memory O(log n) mutation with zero block reads. +// +// COHERENCE MODEL. The cache is a WRITE-PATH optimization only and is only +// ever touched while the caller holds that DID's per-DID write mutex +// (Manager.lockFor) — so all access to a given DID's entry is serialized. Two +// rules keep a cached tree exactly equal to a durably-committed head: +// +// 1. take() DETACHES the entry (removes it from the map) and hands back the +// live tree pointer. The commit path then mutates that tree in place. If +// the commit fails or rolls back, the mutated tree is simply never +// re-inserted, so the next commit reloads a correct tree from postgres — +// a mutated-but-uncommitted tree can never be observed. +// 2. put() re-attaches a tree ONLY under the head that is now durable in +// postgres: the new commit CID after a successful commit, or the +// unchanged head on the idempotent NoOp path (indigo leaves the tree +// unmodified on a no-op insert). A tree is never cached under a head that +// did not commit. +// +// take() also validates head: an entry whose head does not match the head just +// read under the commit's row lock is stale (another process committed) and is +// dropped rather than used. This string-equality check is ABA-safe: head CIDs +// never repeat, because NextRev chains a strictly-increasing rev from the +// stored prev rev under the commit locks and the rev is embedded in the +// signed commit block, so no two commits of a repo can hash to the same head. +// And even a repeated head would be harmless — the head is content-addressed, +// so equal heads mean identical trees, not merely coincidentally-equal labels. +// +// Reads (GetRecord/GetRecordProof/ExportCAR) deliberately do NOT use this +// cache: they run in their own snapshot read transactions and rely on the +// content-addressed blocks table — append-only apart from GC of +// head-unreachable blocks (see gc.go) — for consistency (see repo.go). +// Mixing a shared mutable tree into those paths would add coherence questions +// for no benefit, since reads never contend on the per-DID write mutex. +type mstTreeCache struct { + mu sync.Mutex + max int + ll *list.List // front = most recently used + entries map[string]*list.Element // did -> *cacheItem element +} + +type cacheItem struct { + did string + head string + // root is the MST root CID (the commit's Data field) the tree encodes. + // The commit path needs it as the firehose prevData without recomputing + // it from the tree. + root cid.Cid + tree *mst.Tree +} + +// newTreeCache builds a cache holding at most max trees (LRU eviction). A +// non-positive max disables caching (every take() misses), which keeps the +// commit path correct — just without the optimization. +func newTreeCache(max int) *mstTreeCache { + return &mstTreeCache{ + max: max, + ll: list.New(), + entries: make(map[string]*list.Element), + } +} + +// take removes the cached tree for did and returns it when the cached entry +// matches head (the head just read inside the commit transaction). A miss — +// no entry, or an entry under a different head (a stale cross-process commit) +// — returns nil after evicting any stale entry. The returned tree is detached +// from the cache: the caller owns it until it re-inserts via put (only on a +// durable head) or drops it (on failure). +func (c *mstTreeCache) take(did, head string) (*mst.Tree, cid.Cid, bool) { + if c == nil || c.max <= 0 { + return nil, cid.Undef, false + } + c.mu.Lock() + defer c.mu.Unlock() + el, ok := c.entries[did] + if !ok { + return nil, cid.Undef, false + } + // Detach unconditionally; only return it if the head matches. + delete(c.entries, did) + c.ll.Remove(el) + item := el.Value.(*cacheItem) + if item.head != head { + return nil, cid.Undef, false // stale (another writer/process advanced head) + } + return item.tree, item.root, true +} + +// put installs tree as the cached tree for did under head/root, which MUST be +// the head durably committed in postgres and the MST root it encodes. It +// replaces any existing entry for did and evicts the least-recently-used +// entries beyond the size cap. +func (c *mstTreeCache) put(did, head string, root cid.Cid, tree *mst.Tree) { + if c == nil || c.max <= 0 || tree == nil { + return + } + c.mu.Lock() + defer c.mu.Unlock() + if el, ok := c.entries[did]; ok { + c.ll.Remove(el) + delete(c.entries, did) + } + el := c.ll.PushFront(&cacheItem{did: did, head: head, root: root, tree: tree}) + c.entries[did] = el + for c.ll.Len() > c.max { + back := c.ll.Back() + if back == nil { + break + } + c.ll.Remove(back) + delete(c.entries, back.Value.(*cacheItem).did) + } +} + +// len reports the number of cached trees (test/observability helper). +func (c *mstTreeCache) len() int { + if c == nil { + return 0 + } + c.mu.Lock() + defer c.mu.Unlock() + return c.ll.Len() +} diff --git a/internal/repo/treecache_test.go b/internal/repo/treecache_test.go new file mode 100644 index 0000000..018c9f6 --- /dev/null +++ b/internal/repo/treecache_test.go @@ -0,0 +1,284 @@ +package repo + +import ( + "bytes" + "fmt" + "io" + "testing" + + indigorepo "github.com/bluesky-social/indigo/atproto/repo" + "github.com/bluesky-social/indigo/atproto/repo/mst" + + "github.com/ipfs/go-cid" + car "github.com/ipld/go-car" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func mustCID(t *testing.T, data string) cid.Cid { + t.Helper() + c, err := cidForBlock([]byte(data)) + require.NoError(t, err) + return c +} + +func TestTreeCache_TakeDetachesAndValidatesHead(t *testing.T) { + c := newTreeCache(4) + tree := mst.NewEmptyTree() + root := mustCID(t, "root") + + c.put("did:plc:a", "head1", root, &tree) + require.Equal(t, 1, c.len()) + + // Wrong head: entry is stale and must be dropped, not returned. + _, _, ok := c.take("did:plc:a", "head2") + assert.False(t, ok, "stale head must miss") + assert.Equal(t, 0, c.len(), "stale entry must be evicted by the failed take") + + // Matching head: returned and detached. + c.put("did:plc:a", "head1", root, &tree) + got, gotRoot, ok := c.take("did:plc:a", "head1") + require.True(t, ok) + assert.Same(t, &tree, got) + assert.True(t, root.Equals(gotRoot)) + assert.Equal(t, 0, c.len(), "take must detach the entry") + + // Detached: a second take misses (the caller owns the tree now). + _, _, ok = c.take("did:plc:a", "head1") + assert.False(t, ok) +} + +func TestTreeCache_LRUEvictionAndDisable(t *testing.T) { + c := newTreeCache(2) + tree := mst.NewEmptyTree() + root := mustCID(t, "root") + + c.put("did:plc:a", "h", root, &tree) + c.put("did:plc:b", "h", root, &tree) + // Touch a so b becomes least recently used, then overflow. + _, _, ok := c.take("did:plc:a", "h") + require.True(t, ok) + c.put("did:plc:a", "h", root, &tree) + c.put("did:plc:c", "h", root, &tree) + assert.Equal(t, 2, c.len()) + _, _, ok = c.take("did:plc:b", "h") + assert.False(t, ok, "least-recently-used entry must be evicted") + + // A non-positive size disables caching entirely. + off := newTreeCache(0) + off.put("did:plc:a", "h", root, &tree) + assert.Equal(t, 0, off.len()) + _, _, ok = off.take("did:plc:a", "h") + assert.False(t, ok) + + // nil receiver (defensive) is inert. + var nilCache *mstTreeCache + assert.Equal(t, 0, nilCache.len()) + _, _, ok = nilCache.take("did:plc:a", "h") + assert.False(t, ok) +} + +// TestPutRecord_CachedTreeMatchesReload pins the cache coherence model: after +// a run of commits served from the cached tree, the cached tree's root must +// equal what a cold load of the head from postgres produces, and the exported +// repo must still verify end to end. +func TestPutRecord_CachedTreeMatchesReload(t *testing.T) { + manager, database, _ := testManager(t) + ctx := t.Context() + + for i := 0; i < 10; i++ { + _, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(i), testRecord(fmt.Sprintf("post %d", i))) + require.NoError(t, err) + } + // Update and delete exercise the non-create paths through the cache. + _, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(3), testRecord("post 3 edited")) + require.NoError(t, err) + _, err = manager.DeleteRecord(ctx, testDID, testCollection, testRKey(4)) + require.NoError(t, err) + + require.Equal(t, 1, manager.treeCache.len(), "one DID must occupy one cache slot") + + head, _, err := manager.Head(ctx, testDID) + require.NoError(t, err) + cached, cachedRoot, ok := manager.treeCache.take(testDID, head) + require.True(t, ok, "cache must hold the tree for the current head") + + tx, err := database.BeginTx(ctx, nil) + require.NoError(t, err) + defer func() { _ = tx.Rollback() }() + fresh, freshRoot, err := loadTree(ctx, tx, testDID, head) + require.NoError(t, err) + + cachedCID, err := cached.RootCID() + require.NoError(t, err) + freshCID, err := fresh.RootCID() + require.NoError(t, err) + assert.Equal(t, freshCID.String(), cachedCID.String(), + "cached tree must encode exactly the durable head's MST") + assert.Equal(t, freshRoot.String(), cachedRoot.String()) + + // The repo built through cache-hit commits must round-trip through indigo. + carBytes, err := manager.ExportCAR(ctx, testDID) + require.NoError(t, err) + commit, loaded, err := indigorepo.LoadRepoFromCAR(ctx, bytes.NewReader(carBytes)) + require.NoError(t, err) + require.NoError(t, commit.VerifyStructure()) + root, err := loaded.MST.RootCID() + require.NoError(t, err) + assert.Equal(t, commit.Data.String(), root.String()) +} + +// TestPutRecord_CacheStaleAcrossManagers simulates a second process advancing +// the head: manager A's cached tree must be recognized as stale (head +// mismatch under the commit locks), dropped, and reloaded — never used. +func TestPutRecord_CacheStaleAcrossManagers(t *testing.T) { + managerA, database, _, keys := testManagerWithKeys(t) + ctx := t.Context() + + managerB, err := NewManager(database, keys, nil) + require.NoError(t, err) + + _, err = managerA.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("from A")) + require.NoError(t, err) + // Guard the premise: A's commit must actually have populated A's cache, + // or the staleness scenario below silently tests nothing. + require.Equal(t, 1, managerA.treeCache.len(), "manager A's commit must cache its tree") + _, err = managerB.PutRecord(ctx, testDID, testCollection, testRKey(1), testRecord("from B")) + require.NoError(t, err) + // A's cached tree is now behind B's commit; this commit must detect the + // head mismatch and reload. + _, err = managerA.PutRecord(ctx, testDID, testCollection, testRKey(2), testRecord("from A again")) + require.NoError(t, err) + + for i, want := range []string{"from A", "from B", "from A again"} { + rec, _, err := managerA.GetRecord(ctx, testDID, testCollection, testRKey(i)) + require.NoError(t, err) + assert.Equal(t, want, rec["text"]) + } + carBytes, err := managerA.ExportCAR(ctx, testDID) + require.NoError(t, err) + _, _, err = indigorepo.LoadRepoFromCAR(ctx, bytes.NewReader(carBytes)) + require.NoError(t, err, "repo written by interleaved managers must stay loadable") +} + +// TestPutRecord_NoOpRePutKeepsCacheValid pins the NoOp re-cache path: an +// identical re-put leaves the tree untouched (indigo returns the same value +// without dirtying), so the cache stays valid for the next real commit. +func TestPutRecord_NoOpRePutKeepsCacheValid(t *testing.T) { + manager, _, _ := testManager(t) + ctx := t.Context() + + first, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + noop, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + require.True(t, noop.NoOp) + require.Equal(t, first.CommitCID, noop.CommitCID) + require.Equal(t, 1, manager.treeCache.len(), "NoOp must re-cache the unmodified tree") + + // The next real commit rides the re-cached tree. + second, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(1), testRecord("v2")) + require.NoError(t, err) + require.False(t, second.NoOp) + + carBytes, err := manager.ExportCAR(ctx, testDID) + require.NoError(t, err) + _, _, err = indigorepo.LoadRepoFromCAR(ctx, bytes.NewReader(carBytes)) + require.NoError(t, err) +} + +// TestPutRecord_FailedCommitDropsCacheEntry: take() detaches, and a commit +// that errors must NOT re-insert the tree — the next commit reloads from +// postgres and still succeeds. +func TestPutRecord_FailedCommitDropsCacheEntry(t *testing.T) { + manager, _, _ := testManager(t) + ctx := t.Context() + + _, err := manager.PutRecord(ctx, testDID, testCollection, testRKey(0), testRecord("v1")) + require.NoError(t, err) + require.Equal(t, 1, manager.treeCache.len()) + + // Deleting a missing record fails AFTER the tree was taken from the + // cache (the miss is only detectable post-ApplyOp). + _, err = manager.DeleteRecord(ctx, testDID, testCollection, testRKey(9)) + require.Error(t, err) + assert.Equal(t, 0, manager.treeCache.len(), "failed commit must not re-cache the tree") + + // Reload path still works and repopulates the cache. + _, err = manager.PutRecord(ctx, testDID, testCollection, testRKey(1), testRecord("v2")) + require.NoError(t, err) + assert.Equal(t, 1, manager.treeCache.len()) + rec, _, err := manager.GetRecord(ctx, testDID, testCollection, testRKey(1)) + require.NoError(t, err) + assert.Equal(t, "v2", rec["text"]) +} + +// TestPutRecord_CARSlicesIdenticalWithAndWithoutCache pins that the cache is +// invisible on the firehose: the same logical writes produce CAR slices with +// the same MST-diff and record blocks — in the same order — whether the tree +// came from the cache or a cold load. (indigo's WriteDiffBlocks clears dirty +// flags as it writes, so a reused tree emits only the new commit's diff — +// this is the test that breaks if that upstream behavior ever changes.) The +// sequence deliberately includes an update of an earlier rkey and a delete of +// another: the riskiest ops for a reused tree, whose dirty flags must mark +// exactly the changed path and nothing else. +func TestPutRecord_CARSlicesIdenticalWithAndWithoutCache(t *testing.T) { + manager, database, _, keys := testManagerWithKeys(t) + ctx := t.Context() + + uncachedManager, err := NewManager(database, keys, nil, WithTreeCacheSize(0)) + require.NoError(t, err) + + // Same records, two DIDs: MST node and record CIDs depend only on paths + // and record bytes, so everything except the signed commit block must + // come out identical. + write := func(m *Manager, did string) [][]string { + sliceCIDs := func(res *CommitResult) []string { + var carBytes []byte + require.NoError(t, database.QueryRowContext(ctx, + `SELECT car FROM firehose_events WHERE seq = $1`, res.Seq).Scan(&carBytes)) + reader, err := car.NewCarReader(bytes.NewReader(carBytes)) + require.NoError(t, err) + var cids []string + for { + blk, err := reader.Next() + if err == io.EOF { + break + } + require.NoError(t, err, "CAR slice read must end at EOF, not a real error") + if blk.Cid().String() == res.CommitCID { + continue // the commit block legitimately differs per DID + } + cids = append(cids, blk.Cid().String()) + } + return cids + } + + var perCommit [][]string + for i := 0; i < 8; i++ { + res, err := m.PutRecord(ctx, did, testCollection, testRKey(i), testRecord(fmt.Sprintf("post %d", i))) + require.NoError(t, err) + perCommit = append(perCommit, sliceCIDs(res)) + } + // Update an earlier rkey, then delete another — the non-create paths + // through a reused tree. + res, err := m.PutRecord(ctx, did, testCollection, testRKey(3), testRecord("post 3 edited")) + require.NoError(t, err) + perCommit = append(perCommit, sliceCIDs(res)) + res, err = m.DeleteRecord(ctx, did, testCollection, testRKey(4)) + require.NoError(t, err) + perCommit = append(perCommit, sliceCIDs(res)) + return perCommit + } + + cached := write(manager, testDID) + uncached := write(uncachedManager, testOtherDID) + require.Len(t, cached, len(uncached)) + for i := range cached { + // Ordered equality: both paths write the diff deterministically, so + // the CAR slices must match block-for-block in order, pinning the + // byte layout as closely as the differing signed commit blocks allow. + assert.Equal(t, uncached[i], cached[i], + "commit %d: cached-path CAR slice must carry exactly the cold-load path's blocks, in order", i) + } +} diff --git a/internal/store/inbox_events.go b/internal/store/inbox_events.go index e082c09..a331d7f 100644 --- a/internal/store/inbox_events.go +++ b/internal/store/inbox_events.go @@ -59,36 +59,75 @@ func (r *postgresInboxEvents) ClaimNext(ctx context.Context, lease time.Duration return nil, errors.NewValidationError("lease", "must be positive") } - // The candidate subquery picks the oldest processable event: - // - unprocessed, not poisoned, past its retry schedule; - // - unleased, or leased by a worker whose lease expired (crash); - // - with NO older unprocessed, unpoisoned sibling on the same - // ordering key — this is the per-community serialization: while an - // older event is pending (claimed, backing off, or simply queued), - // every younger event on that key is invisible to workers. A - // poisoned sibling stops blocking (poison → skip). - // FOR UPDATE SKIP LOCKED lets concurrent workers race without - // serializing on row locks; the claiming UPDATE stamps the lease and - // counts the attempt atomically. + // The candidate is the oldest processable event: unprocessed, not + // poisoned, past its retry schedule, unleased (or the lease expired), + // and with NO older unprocessed, unpoisoned sibling on the same + // ordering key — the per-community serialization: while an older event + // is pending (claimed, backing off, or simply queued), every younger + // event on that key is invisible to workers. A poisoned sibling stops + // blocking (poison → skip). + // + // "No older pending sibling" is equivalent to "is the min-id pending + // row of its ordering key", so instead of scanning pending rows in id + // order and probing NOT EXISTS per row — O(backlog) whenever one + // community's queue backs up behind a failing head — the recursive CTE + // emulates a loose index scan over idx_inbox_events_queue + // (ordering_key, id, partial on pending): one index descent per + // DISTINCT pending key jumps straight to each key's head, and only + // those heads are filtered for claimability. Work is O(pending keys × + // log N) regardless of any one key's backlog depth. + // + // The heads are materialized with ARRAY(...) — not a plain IN — so the + // planner fetches exactly those rows by primary key (a merge/semi join + // against an inlined CTE was observed walking the pkey through the + // whole backlog again). The outer SELECT re-applies every claimability + // condition on the locked row: under READ COMMITTED the row is + // re-evaluated after the lock is acquired, so a claim committed between + // the CTE's snapshot and the lock is seen and the row skipped. The + // head-of-its-key property itself needs no re-check — ids only grow and + // processed_at/failed_at are never unset, so a key's pending-min is + // stable once observed (one caveat: id assignment order is not + // commit-visibility order, so a smaller-id enqueue can become visible + // after the claim's snapshot and retroactively lower a key's + // pending-min; the previous NOT EXISTS query had the identical + // single-snapshot blind spot, so this changes nothing about the claim + // semantics). FOR UPDATE SKIP LOCKED lets concurrent workers + // race without serializing on row locks; the claiming UPDATE stamps the + // lease and counts the attempt atomically. query := ` UPDATE inbox_events SET claimed_until = CURRENT_TIMESTAMP + make_interval(secs => $1), attempts = attempts + 1 WHERE id = ( - SELECT e.id FROM inbox_events e - WHERE e.processed_at IS NULL - AND e.failed_at IS NULL - AND e.next_attempt_at <= CURRENT_TIMESTAMP - AND (e.claimed_until IS NULL OR e.claimed_until <= CURRENT_TIMESTAMP) - AND NOT EXISTS ( - SELECT 1 FROM inbox_events prior - WHERE prior.ordering_key = e.ordering_key - AND prior.id < e.id - AND prior.processed_at IS NULL - AND prior.failed_at IS NULL) - ORDER BY e.id + SELECT c.id FROM inbox_events c + WHERE c.id = ANY (ARRAY( + WITH RECURSIVE key_heads AS ( + SELECT h.id, h.ordering_key FROM ( + SELECT e.id, e.ordering_key + FROM inbox_events e + WHERE e.processed_at IS NULL AND e.failed_at IS NULL + ORDER BY e.ordering_key, e.id + LIMIT 1 + ) h + UNION ALL + SELECT n.id, n.ordering_key FROM key_heads k + CROSS JOIN LATERAL ( + SELECT e.id, e.ordering_key + FROM inbox_events e + WHERE e.processed_at IS NULL AND e.failed_at IS NULL + AND e.ordering_key > k.ordering_key + ORDER BY e.ordering_key, e.id + LIMIT 1 + ) n + ) + SELECT id FROM key_heads)) + AND c.processed_at IS NULL + AND c.failed_at IS NULL + AND c.next_attempt_at <= CURRENT_TIMESTAMP + AND (c.claimed_until IS NULL OR c.claimed_until <= CURRENT_TIMESTAMP) + ORDER BY c.id LIMIT 1 - FOR UPDATE SKIP LOCKED) + FOR UPDATE OF c SKIP LOCKED) RETURNING ` + eventColumns event, err := scanInboxEvent(r.db.QueryRowContext(ctx, query, lease.Seconds())) diff --git a/internal/store/inbox_events_test.go b/internal/store/inbox_events_test.go index 54111bf..1256037 100644 --- a/internal/store/inbox_events_test.go +++ b/internal/store/inbox_events_test.go @@ -2,6 +2,7 @@ package store import ( "context" + "fmt" "testing" "time" @@ -141,3 +142,63 @@ func TestInboxEvents_GetEventMissing(t *testing.T) { _, err := repo.GetEvent(ctx, "https://lemmy.world/activities/missing") assert.True(t, errors.IsNotFound(err), "expected IsNotFound, got %v", err) } + +// TestInboxEvents_DeepBacklogDoesNotBlockOtherKeys pins the task-12 claim +// query's semantics AND its reason to exist: a community whose head is +// backing off hides its (arbitrarily deep) younger backlog without hiding +// other keys, and once the head becomes claimable it wins again in id order. +// The old NOT EXISTS scan gave the same answers in O(backlog) per claim; the +// skip-scan gives them in O(pending keys) — the answers must not move. +func TestInboxEvents_DeepBacklogDoesNotBlockOtherKeys(t *testing.T) { + database := testDB(t) + repo := NewInboxEvents(database) + ctx := context.Background() + + const backlog = 500 + const bigKey = "https://lemmy.world/c/big" + for i := 0; i < backlog; i++ { + _, err := repo.Enqueue(ctx, InboxEvent{ + ActivityID: fmt.Sprintf("%s/big/%d", testActivityID, i), + Type: "Announce", + OrderingKey: bigKey, + }) + require.NoError(t, err) + } + _, err := repo.Enqueue(ctx, InboxEvent{ + ActivityID: testActivityID + "/small", + Type: "Announce", + OrderingKey: "https://lemmy.world/c/small", + }) + require.NoError(t, err) + + // Claim the big community's head and put it into backoff — the classic + // failing-event pileup. + head, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + assert.Equal(t, bigKey, head.OrderingKey) + applied, err := repo.Release(ctx, head.ActivityID, "transient failure", + time.Now().Add(time.Hour), *head.ClaimedUntil) + require.NoError(t, err) + require.True(t, applied) + + // The 499 younger big-community events are invisible; the other key's + // event is the only claimable one. + small, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + assert.Equal(t, "https://lemmy.world/c/small", small.OrderingKey) + + // Nothing else claimable while big backs off and small is leased. + _, err = repo.ClaimNext(ctx, time.Minute) + require.Error(t, err) + assert.True(t, errors.IsNotFound(err)) + + // Head becomes due again → it is claimed before every younger sibling. + _, err = database.ExecContext(ctx, + `UPDATE inbox_events SET next_attempt_at = CURRENT_TIMESTAMP WHERE activity_id = $1`, + head.ActivityID) + require.NoError(t, err) + again, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + assert.Equal(t, head.ActivityID, again.ActivityID, + "the key's head must be re-claimed before any younger sibling") +} diff --git a/internal/sync/server.go b/internal/sync/server.go index e3ab349..48260e2 100644 --- a/internal/sync/server.go +++ b/internal/sync/server.go @@ -1,8 +1,10 @@ package sync import ( + "context" "encoding/json" "expvar" + "io" "log/slog" "net/http" "strconv" @@ -89,6 +91,11 @@ type Server struct { // shrinks the kernel send buffer so the write deadline is reachable // without megabytes of backlog. Never set in production. onUpgrade func(*websocket.Conn) + + // exportCAR streams a DID's repo as a CAR; always repo.ExportCARTo in + // production. Test seam: the getRepo mid-stream-failure tests substitute + // exporters that fail before and after the first byte. + exportCAR func(ctx context.Context, did string, w io.Writer) error } // Options configures NewServer. Repo and Broadcaster are required; Hostname @@ -164,6 +171,7 @@ func NewServer(opts Options) (*Server, error) { } s.limiter = ratelimit.New(perSecond, burst) s.refusalLog = ratelimit.NewSampler(time.Second) + s.exportCAR = s.repo.ExportCARTo s.maxSubscribers = int64(opts.MaxSubscribers) if s.maxSubscribers <= 0 { s.maxSubscribers = defaultMaxSubscribers @@ -234,31 +242,74 @@ func (s *Server) loadActiveRepo(w http.ResponseWriter, r *http.Request) *repo.Re return info } -// handleGetRepo serves com.atproto.sync.getRepo: the full repo as a CARv1 -// stream. The optional `since` parameter (diff export) is not implemented — -// consumers that need incremental sync use subscribeRepos; a `since` request -// gets the full CAR, which the spec permits (extra blocks are legal). +// handleGetRepo serves com.atproto.sync.getRepo: the repo as a CARv1 stream. +// The CAR is streamed block-by-block (ExportCARTo) so a large community repo +// never buffers whole in memory, and it carries only the reachable set from +// the current head (commit + live MST + record blocks) — correct CAR readers +// (indigo, bigsky) traverse from the root, so omitting superseded blocks is +// transparent. +// +// The optional `since` parameter (diff export) is not implemented — consumers +// that need incremental sync use subscribeRepos; a `since` request gets the +// full CAR, which the spec permits. // -// NOTE: until block GC exists the export includes historical (unreachable) -// blocks. CAR consumers traverse from the root commit, so extra blocks are -// harmless — Jetstream and indigo's LoadRepoFromCAR both tolerate them — and -// a minimal reachable-set walk over large community repos would cost far -// more than it saves at bridge scale. Revisit alongside block GC. +// Streaming means the response status cannot change once the first block is +// written: existence/consent were already checked by loadActiveRepo, and a +// missing repo (delete race) or unreadable/undecodable head commit fails +// before the first byte — the countingWriter lets those still send a proper +// XRPC error (404 RepoNotFound for the vanished repo, 500 otherwise). A +// failure during the reachable walk happens mid-stream: returning normally +// would let the server write the terminating chunk and hand the client a +// transport-complete 200 wrapping a silently truncated CAR (block-boundary +// truncation is invisible at both the HTTP and CAR-framing layers), so the +// handler aborts the connection instead and the client sees a transport-level +// failure. The exception is the client's own disconnect, which is logged +// quietly and otherwise ignored. func (s *Server) handleGetRepo(w http.ResponseWriter, r *http.Request) { info := s.loadActiveRepo(w, r) if info == nil { return } - carBytes, err := s.repo.ExportCAR(r.Context(), info.DID) - if err != nil { + w.Header().Set("Content-Type", "application/vnd.ipld.car") + cw := &countingWriter{w: w} + err := s.exportCAR(r.Context(), info.DID, cw) + switch { + case err == nil: + case r.Context().Err() != nil: + // The client hung up mid-download. Routine on a public sync surface; + // keep it out of the Error stream so real export failures stay + // visible above the disconnect noise. + s.logger.Debug("sync: export CAR abandoned by client", "did", info.DID, "wrote", cw.n, "error", err) + case cw.n == 0 && errors.IsNotFound(err): + // The repo vanished between loadActiveRepo and the export; nothing + // has been written yet, so the same 404 the earlier existence check + // produces is still deliverable. + writeXRPCError(w, http.StatusNotFound, "RepoNotFound", "repo not found: "+info.DID) + case cw.n == 0: s.logger.Error("sync: export CAR", "did", info.DID, "error", err) writeXRPCError(w, http.StatusInternalServerError, "InternalServerError", "internal error") - return + default: + // Bytes are already on the wire: the 200 cannot be revoked, and a + // normal return would let the server finish the chunked body around + // a truncated CAR. Reset the connection so the truncation is visible + // at the transport layer; http.ErrAbortHandler is the stdlib's + // sanctioned abort (its stack trace is suppressed). + s.logger.Error("sync: export CAR failed mid-stream", "did", info.DID, "wrote", cw.n, "error", err) + panic(http.ErrAbortHandler) } - w.Header().Set("Content-Type", "application/vnd.ipld.car") - w.Header().Set("Content-Length", strconv.Itoa(len(carBytes))) - w.WriteHeader(http.StatusOK) - _, _ = w.Write(carBytes) +} + +// countingWriter tracks whether any bytes have reached the client, so a +// streaming handler knows whether it can still send an HTTP error status. +type countingWriter struct { + w io.Writer + n int64 +} + +func (c *countingWriter) Write(p []byte) (int, error) { + n, err := c.w.Write(p) + c.n += int64(n) + return n, err } // handleGetLatestCommit serves com.atproto.sync.getLatestCommit. diff --git a/internal/sync/sync_test.go b/internal/sync/sync_test.go index 863fc36..6975f80 100644 --- a/internal/sync/sync_test.go +++ b/internal/sync/sync_test.go @@ -26,6 +26,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "tidepool/internal/errors" "tidepool/internal/repo" "tidepool/internal/store" "tidepool/internal/testutil" @@ -746,6 +747,53 @@ func TestHTTPEndpoints_DeactivatedRepo(t *testing.T) { assert.Equal(t, "deleted", *status.Status) } +// TestGetRepo_MidStreamFailureAbortsConnection pins the streaming failure +// contract: once bytes are on the wire the 200 cannot be revoked, and letting +// the server write the terminating chunk would hand the client a +// transport-complete response wrapping a silently truncated CAR. The handler +// must abort the connection instead, so the client sees a transport-level +// read error rather than a clean EOF. +func TestGetRepo_MidStreamFailureAbortsConnection(t *testing.T) { + h := newHarness(t) + putRecord(t, h, "s01", "repo exists") + + h.server.exportCAR = func(ctx context.Context, did string, w io.Writer) error { + if _, err := w.Write([]byte("carv1 header and first block")); err != nil { + return err + } + // Push the 200 header and first bytes to the client before failing, + // so the truncation lands mid-body rather than pre-header. + w.(*countingWriter).w.(http.Flusher).Flush() + return fmt.Errorf("reachable walk: block row unreadable") + } + + client := &http.Client{Timeout: 5 * time.Second} + resp, err := client.Get(h.http.URL + "/xrpc/com.atproto.sync.getRepo?did=" + testDID) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + _, readErr := io.ReadAll(resp.Body) + require.Error(t, readErr, "truncated CAR must surface as a transport failure, not a clean EOF") +} + +// TestGetRepo_VanishedRepoIs404 pins the delete race: a repo that disappears +// between loadActiveRepo and the export has written nothing yet, so the +// handler must still answer with the same 404 RepoNotFound the earlier +// existence check uses, not a 500. +func TestGetRepo_VanishedRepoIs404(t *testing.T) { + h := newHarness(t) + putRecord(t, h, "s02", "repo exists") + + h.server.exportCAR = func(ctx context.Context, did string, w io.Writer) error { + return errors.NewNotFoundError("repo", did) + } + + var body map[string]string + resp := getJSON(t, h.http.URL+"/xrpc/com.atproto.sync.getRepo?did="+testDID, &body) + assert.Equal(t, http.StatusNotFound, resp.StatusCode) + assert.Equal(t, "RepoNotFound", body["error"]) +} + // TestGetBlob pins the com.atproto.sync.getBlob surface task 05 added: a // stored blob round-trips with its content type, misses 404, and // deactivated repos refuse to serve. diff --git a/internal/testutil/db.go b/internal/testutil/db.go index 19e7a4c..74a9ddd 100644 --- a/internal/testutil/db.go +++ b/internal/testutil/db.go @@ -42,7 +42,7 @@ var ( // database fails loudly (skipping every postgres-backed test would let the // suite go green while testing nothing). `make test` starts the // postgres-test container and sets the variable. -func DB(t *testing.T) *sql.DB { +func DB(t testing.TB) *sql.DB { t.Helper() databaseURL := os.Getenv("TIDEPOOL_TEST_DATABASE_URL") @@ -83,7 +83,7 @@ func DB(t *testing.T) *sql.DB { // Truncate empties the given tables and resets their sequences, so a test // starts from a clean slate. -func Truncate(t *testing.T, conn *sql.DB, tables ...string) { +func Truncate(t testing.TB, conn *sql.DB, tables ...string) { t.Helper() if len(tables) == 0 { return -- 2.51.2