From 8f0b87ed1b618b1630e411084d0306e38dd7dec7 Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Sun, 7 Jun 2026 17:05:39 +0200 Subject: [PATCH 1/2] fix: bounded prune of follow feeds --- .changeset/feed-prune-bounded-sweep.md | 34 ++++ .../contrail-appview/src/core/db/index.ts | 4 +- .../contrail-appview/src/core/db/records.ts | 192 +++++++++++++++--- .../contrail-appview/src/core/db/schema.ts | 5 + .../contrail-appview/src/core/jetstream.ts | 36 +++- .../contrail-appview/src/core/persistent.ts | 27 ++- packages/contrail/tests/feed-prune.test.ts | 189 +++++++++++++++++ 7 files changed, 435 insertions(+), 52 deletions(-) create mode 100644 .changeset/feed-prune-bounded-sweep.md create mode 100644 packages/contrail/tests/feed-prune.test.ts diff --git a/.changeset/feed-prune-bounded-sweep.md b/.changeset/feed-prune-bounded-sweep.md new file mode 100644 index 0000000..e459924 --- /dev/null +++ b/.changeset/feed-prune-bounded-sweep.md @@ -0,0 +1,34 @@ +--- +"@atmo-dev/contrail-appview": minor +--- + +fix(feeds): make feed_items pruning bounded so it can't reset the D1 DO + +The hourly feed prune ran a single global `ROW_NUMBER() OVER (PARTITION BY actor)` +window + `(actor, uri) NOT IN (...)` anti-join across the entire `feed_items` +table — O(n) CPU in one statement. Once the table grew large this exceeded D1's +per-query CPU limit and reset the shared Durable Object, taking down any +concurrent read on the same SQLite instance (unrelated user requests 500'd with +`was reset` / `Network connection lost`). Because the statement reset before +completing, caps were never enforced, the table kept growing, and the prune got +more expensive — a death spiral. + +Changes: + +- **Bounded per-actor prune.** Pruning is now an index-backed cutoff delete per + `(actor, collection)` using `idx_feed_actor_coll_time`, cost O(cap), never + O(table). New `pruneActorFeed` / `sweepFeedItems` exports; the ingest loops + run one bounded `sweepFeedItems` slice per tick (`FEED_PRUNE_SWEEP_ACTORS` + actors), which also serves as recovery for already-bloated tables. +- **Persisted prune cursor.** A new `feed_prune_cursor` row tracks the rolling + sweep position, so progress survives the cron isolate recycling that + previously made the in-memory hourly gate a no-op (it pruned on essentially + every tick). The time gate is removed from the cron path; the long-lived + persistent loop keeps a short in-memory throttle. +- **API:** `pruneFeedItems(db, caps)` now accepts only the per-collection + `Map` (the legacy global-number form is removed) and is + reimplemented as a bounded full-table recovery loop — keep it off the hot + path. + +The follow fan-out's `subject` lookup is already covered by `idx__subject`, +so no unbounded statement remains in the ingest path. diff --git a/packages/contrail-appview/src/core/db/index.ts b/packages/contrail-appview/src/core/db/index.ts index 2469f31..fcc52bd 100644 --- a/packages/contrail-appview/src/core/db/index.ts +++ b/packages/contrail-appview/src/core/db/index.ts @@ -1,4 +1,4 @@ export { initSchema } from "./schema"; -export { getLastCursor, saveCursor, applyEvents, lookupExistingRecords, queryRecords, queryAcrossSources, pruneFeedItems } from "./records"; -export type { QueryOptions, SortOption, ExistingRecordInfo } from "./records"; +export { getLastCursor, saveCursor, applyEvents, lookupExistingRecords, queryRecords, queryAcrossSources, pruneFeedItems, pruneActorFeed, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./records"; +export type { QueryOptions, SortOption, ExistingRecordInfo, FeedSweepResult } from "./records"; export type { RecordSource } from "../types"; diff --git a/packages/contrail-appview/src/core/db/records.ts b/packages/contrail-appview/src/core/db/records.ts index 51414b8..57f696e 100644 --- a/packages/contrail-appview/src/core/db/records.ts +++ b/packages/contrail-appview/src/core/db/records.ts @@ -295,49 +295,175 @@ function buildFeedStatements( // --- Feed pruning --- -/** Prune feed_items per (actor, collection) to the given cap. +/** db.batch chunk size for the sweep — caps statements per transaction. */ +const SWEEP_BATCH_SIZE = 50; +/** Actor page size for the full-table {@link pruneFeedItems} recovery loop. */ +const FEED_PRUNE_RECOVERY_BATCH = 200; + +/** + * Build the bounded per-actor cutoff DELETE for one (actor, collection). + * + * Deletes everything older than the newest `cap` rows, driven directly by + * idx_feed_actor_coll_time(actor, collection, time_us DESC). Cost is + * O(cap + deleted) — never O(table). This is the ONLY prune shape contrail + * issues: an unbounded window/anti-join over the whole table can exhaust D1's + * per-query CPU budget and reset the shared Durable Object, which kills any + * concurrent read against the same SQLite instance. * - * - If `caps` is a number: legacy behavior — global per-actor cap across all collections. - * - If `caps` is a Map: each collection is pruned independently per actor, - * so high-volume collections (e.g. RSVPs) can't squeeze out lower-volume ones (e.g. events). - * Collections not present in the map are left alone. + * The cutoff is the cap-th newest row (`OFFSET cap - 1`); we delete strictly + * older rows. Actors with `cap` or fewer rows: the OFFSET subquery yields no + * row, the cutoff is NULL, and `time_us < NULL` matches nothing — a cheap + * index no-op. On a tie at the cutoff time_us we keep the extra rows rather + * than risk deleting a row we meant to keep (feed_items is a cache; a few over + * cap is harmless, dropping a wanted item is not). */ -export async function pruneFeedItems( +function actorCutoffDelete( + db: Database, + actor: string, + collection: string, + cap: number +): Statement { + // Plain `?` placeholders (bound repeatedly) rather than numbered params, so + // the Postgres adapter's positional `?`→`$n` rewrite stays correct. + return db + .prepare( + `DELETE FROM feed_items + WHERE actor = ? AND collection = ? + AND time_us < ( + SELECT time_us FROM feed_items + WHERE actor = ? AND collection = ? + ORDER BY time_us DESC LIMIT 1 OFFSET ? + )` + ) + .bind(actor, collection, actor, collection, Math.max(0, cap - 1)); +} + +/** Prune a single actor's feed for one collection to `cap`. Bounded O(cap). */ +export async function pruneActorFeed( db: Database, - caps: number | Map + actor: string, + collection: string, + cap: number ): Promise { - if (typeof caps === "number") { - const result = await db - .prepare( - `DELETE FROM feed_items WHERE (actor, uri) NOT IN ( - SELECT actor, uri FROM ( - SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn - FROM feed_items - ) sub WHERE rn <= ? - )` - ) - .bind(caps) - .run(); - return (result as any)?.changes ?? 0; + const result = await actorCutoffDelete(db, actor, collection, cap).run(); + return (result as any)?.changes ?? 0; +} + +export interface FeedSweepResult { + /** Rows deleted this slice. */ + pruned: number; + /** Actor to resume after; null once a full pass completed (wrap to start). */ + nextCursor: string | null; + /** True when this slice reached the end of the actor list. */ + done: boolean; +} + +/** + * One bounded slice of a rolling feed-items prune. + * + * Pages at most `actorBudget` distinct actors (resuming after `cursor`, via the + * feed_items (actor, uri) PK) and applies the per-(actor, collection) cutoff + * delete for every cap in `caps`. Every issued statement is index-backed and + * O(cap), so the slice's per-query CPU stays flat no matter how large + * feed_items grows — the property the old global window query lacked. + * + * Drive it across ticks with a persisted cursor (see getFeedPruneCursor): + * feed back `nextCursor` until `done`, at which point the cursor wraps to null + * and the next pass starts from the beginning. Because each pass visits every + * actor, this doubles as the recovery path for an already-bloated table. + */ +export async function sweepFeedItems( + db: Database, + caps: Map, + cursor: string | null, + actorBudget: number +): Promise { + if (caps.size === 0 || actorBudget <= 0) { + return { pruned: 0, nextCursor: null, done: true }; + } + + const actorsRes = cursor + ? await db + .prepare( + "SELECT DISTINCT actor FROM feed_items WHERE actor > ? ORDER BY actor LIMIT ?" + ) + .bind(cursor, actorBudget) + .all<{ actor: string }>() + : await db + .prepare("SELECT DISTINCT actor FROM feed_items ORDER BY actor LIMIT ?") + .bind(actorBudget) + .all<{ actor: string }>(); + + const actors = (actorsRes.results ?? []).map((r) => r.actor); + if (actors.length === 0) { + // Ran off the end (cursor pointed past the last actor) — wrap next tick. + return { pruned: 0, nextCursor: null, done: true }; + } + + const stmts: Statement[] = []; + for (const actor of actors) { + for (const [collection, cap] of caps) { + stmts.push(actorCutoffDelete(db, actor, collection, cap)); + } + } + + let pruned = 0; + for (let i = 0; i < stmts.length; i += SWEEP_BATCH_SIZE) { + const results = await db.batch(stmts.slice(i, i + SWEEP_BATCH_SIZE)); + for (const r of results) pruned += (r as any)?.changes ?? 0; } + + // A short page means we exhausted the actor list this slice. + const done = actors.length < actorBudget; + return { pruned, nextCursor: done ? null : actors[actors.length - 1], done }; +} + +/** + * Prune the ENTIRE feed_items table to the per-collection `caps` by looping the + * bounded {@link sweepFeedItems} until a full pass completes. + * + * Every statement is O(cap) and safe against D1's per-query CPU limit, but the + * statement count is O(distinct actors), so keep this OFF the hot ingest path — + * the cron/persistent loops issue a single bounded slice per tick instead. Use + * it for one-shot recovery or admin tooling. + */ +export async function pruneFeedItems( + db: Database, + caps: Map +): Promise { let total = 0; - for (const [collection, cap] of caps) { - const result = await db - .prepare( - `DELETE FROM feed_items WHERE collection = ? AND (actor, uri) NOT IN ( - SELECT actor, uri FROM ( - SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn - FROM feed_items WHERE collection = ? - ) sub WHERE rn <= ? - )` - ) - .bind(collection, collection, cap) - .run(); - total += (result as any)?.changes ?? 0; + let cursor: string | null = null; + for (;;) { + const res = await sweepFeedItems(db, caps, cursor, FEED_PRUNE_RECOVERY_BATCH); + total += res.pruned; + if (res.done) break; + cursor = res.nextCursor; } return total; } +// --- Feed prune cursor --- + +/** Last actor swept by the rolling feed prune; null = start of a fresh pass. */ +export async function getFeedPruneCursor(db: Database): Promise { + const row = await db + .prepare("SELECT actor FROM feed_prune_cursor WHERE id = 1") + .first<{ actor: string | null }>(); + return row?.actor ?? null; +} + +export async function saveFeedPruneCursor( + db: Database, + actor: string | null +): Promise { + await db + .prepare( + "INSERT INTO feed_prune_cursor (id, actor) VALUES (1, ?) ON CONFLICT(id) DO UPDATE SET actor = excluded.actor" + ) + .bind(actor) + .run(); +} + // --- Cursor --- export async function getLastCursor(db: Database): Promise { diff --git a/packages/contrail-appview/src/core/db/schema.ts b/packages/contrail-appview/src/core/db/schema.ts index df93e72..7089827 100644 --- a/packages/contrail-appview/src/core/db/schema.ts +++ b/packages/contrail-appview/src/core/db/schema.ts @@ -338,6 +338,11 @@ function buildFeedTables(config: ContrailConfig, dialect: SqlDialect): string[] )`, `CREATE INDEX IF NOT EXISTS idx_feed_actor_coll_time ON feed_items(actor, collection, time_us DESC)`, `CREATE INDEX IF NOT EXISTS idx_feed_actor_time ON feed_items(actor, time_us DESC)`, + // Single-row cursor for the rolling, bounded feed prune (see sweepFeedItems). + `CREATE TABLE IF NOT EXISTS feed_prune_cursor ( + id INTEGER PRIMARY KEY CHECK (id = 1), + actor TEXT + )`, `CREATE TABLE IF NOT EXISTS feed_backfills ( actor TEXT NOT NULL, feed TEXT NOT NULL, diff --git a/packages/contrail-appview/src/core/jetstream.ts b/packages/contrail-appview/src/core/jetstream.ts index 486d398..ee5cab3 100644 --- a/packages/contrail-appview/src/core/jetstream.ts +++ b/packages/contrail-appview/src/core/jetstream.ts @@ -6,22 +6,28 @@ import { shortNameForNsid, buildFeedTargetCaps, } from "./types"; -import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; +import { initSchema, getLastCursor, saveCursor, applyEvents, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./db"; import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; import { backfillFollowersFromConstellation } from "./constellation"; const BATCH_SIZE = 50; -const FEED_PRUNE_INTERVAL_MS = 60 * 60 * 1000; // 1 hour +/** Distinct actors pruned per ingest tick by the rolling feed sweep. Each + * actor costs a handful of index-backed O(cap) deletes, so this bounds the + * prune's per-tick CPU regardless of how large feed_items grows. */ +export const FEED_PRUNE_SWEEP_ACTORS = 500; /** Mutable state that persists across ingest cycles within the same process. */ export interface IngestState { cachedKnownDids?: Set; schemaInitialized: boolean; - lastFeedPruneMs: number; + /** Wall-clock of the last feed sweep — used only by the long-lived + * persistent loop to throttle; the recycling cron isolate sweeps every + * tick and relies on the persisted cursor instead. */ + lastFeedSweepMs: number; } export function createIngestState(): IngestState { - return { schemaInitialized: false, lastFeedPruneMs: 0 }; + return { schemaInitialized: false, lastFeedSweepMs: 0 }; } function getLogger(config: ContrailConfig): Logger { @@ -336,15 +342,25 @@ export async function runIngestCycle( } } - // Prune feed items hourly, per-target so high-volume targets don't - // squeeze out lower-volume ones. - if (config.feeds && Date.now() - s.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { + // Prune feed_items to per-collection caps with a bounded, cursored sweep. + // Every statement is index-backed and O(cap) (see sweepFeedItems), so it can + // never exhaust D1's per-query CPU budget and reset the shared DO — unlike + // the old global window+anti-join. The cron isolate recycles each tick, so we + // persist the sweep cursor in the DB rather than gating on in-memory time, + // and run an unconditional bounded slice every tick. + if (config.feeds) { const caps = buildFeedTargetCaps(config); if (caps.size > 0) { - const pruned = await pruneFeedItems(db, caps); - if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); + const cursor = await getFeedPruneCursor(db); + const { pruned, nextCursor } = await sweepFeedItems( + db, + caps, + cursor, + FEED_PRUNE_SWEEP_ACTORS + ); + await saveFeedPruneCursor(db, nextCursor); + if (pruned > 0) log.log(`Pruned ${pruned} feed items (sweep)`); } - s.lastFeedPruneMs = Date.now(); } log.log(`[ingest] cycle complete. stored=${events.length}`); diff --git a/packages/contrail-appview/src/core/persistent.ts b/packages/contrail-appview/src/core/persistent.ts index 94ca666..63a3ff0 100644 --- a/packages/contrail-appview/src/core/persistent.ts +++ b/packages/contrail-appview/src/core/persistent.ts @@ -7,13 +7,16 @@ import { resolveConfig, shortNameForNsid, } from "./types"; -import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; +import { initSchema, getLastCursor, saveCursor, applyEvents, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./db"; import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; import { backfillFollowersFromConstellation } from "./constellation"; -import { createIngestState } from "./jetstream"; +import { createIngestState, FEED_PRUNE_SWEEP_ACTORS } from "./jetstream"; import type { IngestState } from "./jetstream"; -const FEED_PRUNE_INTERVAL_MS = 60 * 60 * 1000; +/** How often the long-lived persistent loop runs a bounded feed sweep. The + * process stays resident, so this in-memory throttle is reliable here (unlike + * the recycling cron isolate). */ +const FEED_SWEEP_INTERVAL_MS = 10_000; export interface PersistentIngestOptions { batchSize?: number; @@ -176,13 +179,23 @@ async function streamAndFlush( } } - if (config.feeds && Date.now() - state.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { + // Bounded, cursored feed prune (see sweepFeedItems). This process is + // long-lived, so the in-memory interval is a reliable throttle; the + // cursor is still persisted so progress carries across restarts. + if (config.feeds && Date.now() - state.lastFeedSweepMs > FEED_SWEEP_INTERVAL_MS) { const caps = buildFeedTargetCaps(config); if (caps.size > 0) { - const pruned = await pruneFeedItems(db, caps); - if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); + const cursor = await getFeedPruneCursor(db); + const { pruned, nextCursor } = await sweepFeedItems( + db, + caps, + cursor, + FEED_PRUNE_SWEEP_ACTORS + ); + await saveFeedPruneCursor(db, nextCursor); + if (pruned > 0) log.log(`Pruned ${pruned} feed items (sweep)`); } - state.lastFeedPruneMs = Date.now(); + state.lastFeedSweepMs = Date.now(); } log.log(`Flushed ${batch.length} events. Cursor: ${lastTimeUs}`); diff --git a/packages/contrail/tests/feed-prune.test.ts b/packages/contrail/tests/feed-prune.test.ts new file mode 100644 index 0000000..a1f6e5c --- /dev/null +++ b/packages/contrail/tests/feed-prune.test.ts @@ -0,0 +1,189 @@ +import { describe, it, expect, beforeEach } from "vitest"; +import { createSqliteDatabase } from "../src/adapters/sqlite"; +import type { Database, ResolvedContrailConfig } from "../src/core/types"; +import { resolveConfig } from "../src/core/types"; +import { initSchema } from "../src/core/db/schema"; +import { + pruneActorFeed, + sweepFeedItems, + pruneFeedItems, + getFeedPruneCursor, + saveFeedPruneCursor, +} from "../src/core/db/records"; + +const EVENT = "community.lexicon.calendar.event"; +const RSVP = "community.lexicon.calendar.rsvp"; + +// Feeds config: event capped at 2 per actor, rsvp at 3. resolveConfig +// auto-adds the `follow` collection, so initSchema builds feed_items, the +// idx_feed_actor_coll_time index, and feed_prune_cursor. +const CONFIG: ResolvedContrailConfig = resolveConfig({ + namespace: "com.example", + collections: { + event: { collection: EVENT }, + rsvp: { collection: RSVP }, + }, + feeds: { + main: { + targets: [ + { collection: "event", maxItems: 2 }, + { collection: "rsvp", maxItems: 3 }, + ], + }, + }, +}); + +// caps keyed by NSID, matching what buildFeedTargetCaps / the fanout produce. +const CAPS = new Map([ + [EVENT, 2], + [RSVP, 3], +]); + +let db: Database; + +async function insertItem( + actor: string, + collection: string, + n: number, + timeUs: number +): Promise { + await db + .prepare( + "INSERT INTO feed_items (actor, uri, collection, time_us) VALUES (?, ?, ?, ?)" + ) + .bind(actor, `at://${actor}/${collection}/${n}`, collection, timeUs) + .run(); +} + +/** Insert `count` items for (actor, collection) with increasing time_us. */ +async function seed( + actor: string, + collection: string, + count: number +): Promise { + for (let i = 0; i < count; i++) { + await insertItem(actor, collection, i, 1000 + i); + } +} + +async function rows( + actor: string, + collection: string +): Promise { + const res = await db + .prepare( + "SELECT time_us FROM feed_items WHERE actor = ? AND collection = ? ORDER BY time_us DESC" + ) + .bind(actor, collection) + .all<{ time_us: number }>(); + return (res.results ?? []).map((r) => Number(r.time_us)); +} + +beforeEach(async () => { + db = createSqliteDatabase(":memory:"); + await initSchema(db, CONFIG); +}); + +describe("pruneActorFeed", () => { + it("keeps the newest `cap` rows and deletes the rest", async () => { + await seed("alice", EVENT, 5); // time_us 1000..1004 + const deleted = await pruneActorFeed(db, "alice", EVENT, 2); + expect(deleted).toBe(3); + expect(await rows("alice", EVENT)).toEqual([1004, 1003]); + }); + + it("is a no-op when the actor is at or under cap", async () => { + await seed("bob", EVENT, 2); + expect(await pruneActorFeed(db, "bob", EVENT, 2)).toBe(0); + expect(await pruneActorFeed(db, "bob", EVENT, 5)).toBe(0); + expect((await rows("bob", EVENT)).length).toBe(2); + }); + + it("only touches the named collection", async () => { + await seed("alice", EVENT, 4); + await seed("alice", RSVP, 4); + await pruneActorFeed(db, "alice", EVENT, 2); + expect((await rows("alice", EVENT)).length).toBe(2); + expect((await rows("alice", RSVP)).length).toBe(4); // untouched + }); +}); + +describe("sweepFeedItems", () => { + it("prunes every actor to the per-collection caps in one pass", async () => { + await seed("alice", EVENT, 5); + await seed("alice", RSVP, 6); + await seed("bob", EVENT, 1); + await seed("carol", RSVP, 10); + + const res = await sweepFeedItems(db, CAPS, null, 100); + + expect(res.done).toBe(true); + expect(res.nextCursor).toBeNull(); + expect(res.pruned).toBe(3 + 3 + 0 + 7); // alice event/rsvp, bob, carol + expect((await rows("alice", EVENT)).length).toBe(2); + expect((await rows("alice", RSVP)).length).toBe(3); + expect((await rows("bob", EVENT)).length).toBe(1); + expect((await rows("carol", RSVP)).length).toBe(3); + }); + + it("pages by actor and resumes via the cursor", async () => { + // Three actors, each over the event cap. + for (const a of ["a-actor", "b-actor", "c-actor"]) await seed(a, EVENT, 5); + + // Budget of 1 actor per slice: first slice handles "a-actor". + const s1 = await sweepFeedItems(db, CAPS, null, 1); + expect(s1.done).toBe(false); + expect(s1.nextCursor).toBe("a-actor"); + expect((await rows("a-actor", EVENT)).length).toBe(2); + expect((await rows("b-actor", EVENT)).length).toBe(5); // not yet reached + + const s2 = await sweepFeedItems(db, CAPS, s1.nextCursor, 1); + expect(s2.nextCursor).toBe("b-actor"); + expect((await rows("b-actor", EVENT)).length).toBe(2); + + const s3 = await sweepFeedItems(db, CAPS, s2.nextCursor, 1); + // Last actor — still a full page, so not yet flagged done. + expect(s3.nextCursor).toBe("c-actor"); + expect((await rows("c-actor", EVENT)).length).toBe(2); + + // One more slice runs off the end and wraps. + const s4 = await sweepFeedItems(db, CAPS, s3.nextCursor, 1); + expect(s4.done).toBe(true); + expect(s4.nextCursor).toBeNull(); + expect(s4.pruned).toBe(0); + }); + + it("returns done with no work for empty caps", async () => { + await seed("alice", EVENT, 5); + const res = await sweepFeedItems(db, new Map(), null, 100); + expect(res).toEqual({ pruned: 0, nextCursor: null, done: true }); + expect((await rows("alice", EVENT)).length).toBe(5); + }); +}); + +describe("feed prune cursor", () => { + it("round-trips and defaults to null", async () => { + expect(await getFeedPruneCursor(db)).toBeNull(); + await saveFeedPruneCursor(db, "did:plc:xyz"); + expect(await getFeedPruneCursor(db)).toBe("did:plc:xyz"); + await saveFeedPruneCursor(db, null); + expect(await getFeedPruneCursor(db)).toBeNull(); + }); +}); + +describe("pruneFeedItems (full recovery loop)", () => { + it("brings an already-bloated table within caps in one call", async () => { + // Many actors well over cap — the bloated-table recovery scenario. + for (let i = 0; i < 25; i++) { + await seed(`actor-${String(i).padStart(2, "0")}`, EVENT, 8); + await seed(`actor-${String(i).padStart(2, "0")}`, RSVP, 8); + } + const total = await pruneFeedItems(db, CAPS); + expect(total).toBe(25 * (6 + 5)); // event: 8→2, rsvp: 8→3 + + const remaining = await db + .prepare("SELECT COUNT(*) AS c FROM feed_items") + .first<{ c: number }>(); + expect(Number(remaining?.c)).toBe(25 * (2 + 3)); + }); +}); -- 2.51.2 From 162bf96a6d660d19c3f408b0d11eba1c6d642058 Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Sun, 7 Jun 2026 17:58:15 +0200 Subject: [PATCH 2/2] add test --- .../tests/feed-prune-guardrail.test.ts | 243 ++++++++++++++++++ 1 file changed, 243 insertions(+) create mode 100644 packages/contrail/tests/feed-prune-guardrail.test.ts diff --git a/packages/contrail/tests/feed-prune-guardrail.test.ts b/packages/contrail/tests/feed-prune-guardrail.test.ts new file mode 100644 index 0000000..a538902 --- /dev/null +++ b/packages/contrail/tests/feed-prune-guardrail.test.ts @@ -0,0 +1,243 @@ +import { describe, it, expect, beforeEach } from "vitest"; +import { createSqliteDatabase } from "../src/adapters/sqlite"; +import type { + Database, + Statement, + ResolvedContrailConfig, + IngestEvent, +} from "../src/core/types"; +import { resolveConfig } from "../src/core/types"; +import { initSchema } from "../src/core/db/schema"; +import { + applyEvents, + sweepFeedItems, + pruneActorFeed, + pruneFeedItems, +} from "../src/core/db/records"; + +// --------------------------------------------------------------------------- +// Guardrail: no contrail-issued maintenance statement may be unbounded-O(n) +// over a table. A single full-table SCAN can exhaust D1's per-query CPU budget +// and reset the shared Durable Object, which kills every concurrent read on the +// same SQLite instance (the feed-prune outage). This test exercises the real +// prune + feed-fanout code, captures every SQL it issues, and asserts each one +// is index-bounded via EXPLAIN QUERY PLAN. +// --------------------------------------------------------------------------- + +const EVENT = "community.lexicon.calendar.event"; +const RSVP = "community.lexicon.calendar.rsvp"; +const FOLLOW = "app.bsky.graph.follow"; + +const CONFIG: ResolvedContrailConfig = resolveConfig({ + namespace: "com.example", + collections: { + event: { collection: EVENT }, + rsvp: { collection: RSVP }, + }, + feeds: { + main: { + targets: [ + { collection: "event", maxItems: 2 }, + { collection: "rsvp", maxItems: 3 }, + ], + }, + }, +}); + +const CAPS = new Map([ + [EVENT, 2], + [RSVP, 3], +]); + +/** Wrap a Database so every SQL string passed to prepare() is recorded, while + * delegating to the real DB so the exercised code still reads/writes data. */ +function recordingDb(real: Database): { db: Database; sqls: string[] } { + const sqls: string[] = []; + const db: Database = { + prepare(sql: string): Statement { + sqls.push(sql); + return real.prepare(sql); + }, + batch(stmts: Statement[]): Promise { + return real.batch(stmts); + }, + dialect: real.dialect, + }; + return { db, sqls }; +} + +/** Return the EXPLAIN QUERY PLAN `detail` lines for a statement. Params are + * irrelevant to the plan, so we bind dummy values to satisfy the placeholders. */ +async function queryPlan(db: Database, sql: string): Promise { + const placeholders = (sql.match(/\?/g) ?? []).length; + const binds = Array.from({ length: placeholders }, () => 1); + const res = await db + .prepare("EXPLAIN QUERY PLAN " + sql) + .bind(...binds) + .all<{ detail: string }>(); + return (res.results ?? []).map((r) => r.detail); +} + +/** + * Plan lines that represent an UNBOUNDED full-table scan: a `SCAN ` that + * is not driven by an index. Index SEARCHes and LIMIT-bounded index SCANs are + * fine — their cost is keyed/bounded, not proportional to the whole table. + */ +function unboundedScans(plan: string[]): string[] { + return plan.filter( + (d) => /^SCAN\b/i.test(d.trim()) && !/\bINDEX\b/i.test(d) + ); +} + +/** Statements with no rows in their plan (e.g. INSERT ... VALUES) read nothing. */ +async function assertAllBounded(db: Database, sqls: string[]): Promise { + for (const sql of [...new Set(sqls)]) { + const plan = await queryPlan(db, sql); + const bad = unboundedScans(plan); + expect( + bad, + `Unbounded full-table scan in maintenance SQL:\n ${sql}\n plan: ${plan.join(" | ")}` + ).toEqual([]); + } +} + +function makeFollowRow(db: Database, follower: string, subject: string) { + return db + .prepare( + "INSERT INTO records_follow (uri, did, rkey, cid, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?)" + ) + .bind( + `at://${follower}/${FOLLOW}/${subject}`, + follower, + subject, + "bafyfollow", + JSON.stringify({ subject }), + 1000, + 1000 + ) + .run(); +} + +let real: Database; + +beforeEach(async () => { + real = createSqliteDatabase(":memory:"); + await initSchema(real, CONFIG); +}); + +describe("feed maintenance stays index-bounded", () => { + it("sweepFeedItems issues only index-bounded statements", async () => { + // Several actors over cap so the sweep actually deletes. + for (const a of ["did:plc:a", "did:plc:b", "did:plc:c"]) { + for (let i = 0; i < 6; i++) { + await real + .prepare( + "INSERT INTO feed_items (actor, uri, collection, time_us) VALUES (?, ?, ?, ?)" + ) + .bind(a, `at://${a}/e/${i}`, EVENT, 1000 + i) + .run(); + } + } + + const { db, sqls } = recordingDb(real); + const res = await sweepFeedItems(db, CAPS, null, 100); + expect(res.pruned).toBeGreaterThan(0); + expect(sqls.length).toBeGreaterThan(0); + await assertAllBounded(real, sqls); + }); + + it("pruneActorFeed / pruneFeedItems issue only index-bounded statements", async () => { + for (let i = 0; i < 8; i++) { + await real + .prepare( + "INSERT INTO feed_items (actor, uri, collection, time_us) VALUES (?, ?, ?, ?)" + ) + .bind("did:plc:z", `at://did:plc:z/e/${i}`, EVENT, 2000 + i) + .run(); + } + + const a = recordingDb(real); + await pruneActorFeed(a.db, "did:plc:z", EVENT, 2); + await assertAllBounded(real, a.sqls); + + const b = recordingDb(real); + await pruneFeedItems(b.db, CAPS); + await assertAllBounded(real, b.sqls); + }); + + it("feed fan-out on a target create issues only index-bounded statements", async () => { + // A follower pointing at the event's author, so the fan-out has work. + await makeFollowRow(real, "did:plc:follower", "did:plc:author"); + + const { db, sqls } = recordingDb(real); + const event: IngestEvent = { + uri: "at://did:plc:author/" + EVENT + "/evt1", + did: "did:plc:author", + collection: EVENT, + rkey: "evt1", + cid: "bafyevt", + record: JSON.stringify({ name: "Party", startsAt: "2026-04-01T10:00:00Z" }), + time_us: 5000, + indexed_at: 5000, + operation: "create", + }; + await applyEvents(db, [event], CONFIG); + + // The fan-out INSERT must have been issued and it must hit the table. + const fanout = sqls.find((s) => /INSERT.*feed_items/is.test(s)); + expect(fanout, "expected a feed_items fan-out INSERT").toBeTruthy(); + expect((await real.prepare("SELECT COUNT(*) AS c FROM feed_items").first<{ c: number }>())?.c).toBe(1); + + await assertAllBounded(real, sqls); + }); + + it("the fan-out follower lookup uses idx_follow_subject (not a full scan)", async () => { + await makeFollowRow(real, "did:plc:follower", "did:plc:author"); + + const { db, sqls } = recordingDb(real); + await applyEvents( + db, + [ + { + uri: "at://did:plc:author/" + EVENT + "/evt2", + did: "did:plc:author", + collection: EVENT, + rkey: "evt2", + cid: "bafyevt2", + record: JSON.stringify({ name: "x" }), + time_us: 6000, + indexed_at: 6000, + operation: "create", + }, + ], + CONFIG + ); + + const fanout = sqls.find((s) => /records_follow/is.test(s))!; + const plan = (await queryPlan(real, fanout)).join(" | "); + expect(plan).toMatch(/idx_follow_subject/i); + }); +}); + +describe("the guardrail has teeth", () => { + it("rejects the old global window + anti-join prune", async () => { + // The original pruneFeedItems statement that reset the D1 DO in production. + const oldGlobalPrune = `DELETE FROM feed_items WHERE collection = ? AND (actor, uri) NOT IN ( + SELECT actor, uri FROM ( + SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn + FROM feed_items WHERE collection = ? + ) sub WHERE rn <= ? + )`; + const plan = await queryPlan(real, oldGlobalPrune); + // It must trip the guard with at least one full-table SCAN of feed_items. + expect(unboundedScans(plan).length).toBeGreaterThan(0); + }); + + it("flags a contrived unindexed scan", async () => { + const plan = await queryPlan( + real, + "SELECT * FROM feed_items WHERE time_us = ?" + ); + expect(unboundedScans(plan).length).toBeGreaterThan(0); + }); +}); -- 2.51.2