+34

.changeset/feed-prune-bounded-sweep.md

Reviewed

··· 1 1 + --- 2 2 + "@atmo-dev/contrail-appview": minor 3 3 + --- 4 4 + 5 5 + fix(feeds): make feed_items pruning bounded so it can't reset the D1 DO 6 6 + 7 7 + The hourly feed prune ran a single global `ROW_NUMBER() OVER (PARTITION BY actor)` 8 8 + window + `(actor, uri) NOT IN (...)` anti-join across the entire `feed_items` 9 9 + table — O(n) CPU in one statement. Once the table grew large this exceeded D1's 10 10 + per-query CPU limit and reset the shared Durable Object, taking down any 11 11 + concurrent read on the same SQLite instance (unrelated user requests 500'd with 12 12 + `was reset` / `Network connection lost`). Because the statement reset before 13 13 + completing, caps were never enforced, the table kept growing, and the prune got 14 14 + more expensive — a death spiral. 15 15 + 16 16 + Changes: 17 17 + 18 18 + - **Bounded per-actor prune.** Pruning is now an index-backed cutoff delete per 19 19 + `(actor, collection)` using `idx_feed_actor_coll_time`, cost O(cap), never 20 20 + O(table). New `pruneActorFeed` / `sweepFeedItems` exports; the ingest loops 21 21 + run one bounded `sweepFeedItems` slice per tick (`FEED_PRUNE_SWEEP_ACTORS` 22 22 + actors), which also serves as recovery for already-bloated tables. 23 23 + - **Persisted prune cursor.** A new `feed_prune_cursor` row tracks the rolling 24 24 + sweep position, so progress survives the cron isolate recycling that 25 25 + previously made the in-memory hourly gate a no-op (it pruned on essentially 26 26 + every tick). The time gate is removed from the cron path; the long-lived 27 27 + persistent loop keeps a short in-memory throttle. 28 28 + - **API:** `pruneFeedItems(db, caps)` now accepts only the per-collection 29 29 + `Map<collection, cap>` (the legacy global-number form is removed) and is 30 30 + reimplemented as a bounded full-table recovery loop — keep it off the hot 31 31 + path. 32 32 + 33 33 + The follow fan-out's `subject` lookup is already covered by `idx_<follow>_subject`, 34 34 + so no unbounded statement remains in the ingest path.

+189

packages/contrail/tests/feed-prune.test.ts

Reviewed

··· 1 1 + import { describe, it, expect, beforeEach } from "vitest"; 2 2 + import { createSqliteDatabase } from "../src/adapters/sqlite"; 3 3 + import type { Database, ResolvedContrailConfig } from "../src/core/types"; 4 4 + import { resolveConfig } from "../src/core/types"; 5 5 + import { initSchema } from "../src/core/db/schema"; 6 6 + import { 7 7 + pruneActorFeed, 8 8 + sweepFeedItems, 9 9 + pruneFeedItems, 10 10 + getFeedPruneCursor, 11 11 + saveFeedPruneCursor, 12 12 + } from "../src/core/db/records"; 13 13 + 14 14 + const EVENT = "community.lexicon.calendar.event"; 15 15 + const RSVP = "community.lexicon.calendar.rsvp"; 16 16 + 17 17 + // Feeds config: event capped at 2 per actor, rsvp at 3. resolveConfig 18 18 + // auto-adds the `follow` collection, so initSchema builds feed_items, the 19 19 + // idx_feed_actor_coll_time index, and feed_prune_cursor. 20 20 + const CONFIG: ResolvedContrailConfig = resolveConfig({ 21 21 + namespace: "com.example", 22 22 + collections: { 23 23 + event: { collection: EVENT }, 24 24 + rsvp: { collection: RSVP }, 25 25 + }, 26 26 + feeds: { 27 27 + main: { 28 28 + targets: [ 29 29 + { collection: "event", maxItems: 2 }, 30 30 + { collection: "rsvp", maxItems: 3 }, 31 31 + ], 32 32 + }, 33 33 + }, 34 34 + }); 35 35 + 36 36 + // caps keyed by NSID, matching what buildFeedTargetCaps / the fanout produce. 37 37 + const CAPS = new Map<string, number>([ 38 38 + [EVENT, 2], 39 39 + [RSVP, 3], 40 40 + ]); 41 41 + 42 42 + let db: Database; 43 43 + 44 44 + async function insertItem( 45 45 + actor: string, 46 46 + collection: string, 47 47 + n: number, 48 48 + timeUs: number 49 49 + ): Promise<void> { 50 50 + await db 51 51 + .prepare( 52 52 + "INSERT INTO feed_items (actor, uri, collection, time_us) VALUES (?, ?, ?, ?)" 53 53 + ) 54 54 + .bind(actor, `at://${actor}/${collection}/${n}`, collection, timeUs) 55 55 + .run(); 56 56 + } 57 57 + 58 58 + /** Insert `count` items for (actor, collection) with increasing time_us. */ 59 59 + async function seed( 60 60 + actor: string, 61 61 + collection: string, 62 62 + count: number 63 63 + ): Promise<void> { 64 64 + for (let i = 0; i < count; i++) { 65 65 + await insertItem(actor, collection, i, 1000 + i); 66 66 + } 67 67 + } 68 68 + 69 69 + async function rows( 70 70 + actor: string, 71 71 + collection: string 72 72 + ): Promise<number[]> { 73 73 + const res = await db 74 74 + .prepare( 75 75 + "SELECT time_us FROM feed_items WHERE actor = ? AND collection = ? ORDER BY time_us DESC" 76 76 + ) 77 77 + .bind(actor, collection) 78 78 + .all<{ time_us: number }>(); 79 79 + return (res.results ?? []).map((r) => Number(r.time_us)); 80 80 + } 81 81 + 82 82 + beforeEach(async () => { 83 83 + db = createSqliteDatabase(":memory:"); 84 84 + await initSchema(db, CONFIG); 85 85 + }); 86 86 + 87 87 + describe("pruneActorFeed", () => { 88 88 + it("keeps the newest `cap` rows and deletes the rest", async () => { 89 89 + await seed("alice", EVENT, 5); // time_us 1000..1004 90 90 + const deleted = await pruneActorFeed(db, "alice", EVENT, 2); 91 91 + expect(deleted).toBe(3); 92 92 + expect(await rows("alice", EVENT)).toEqual([1004, 1003]); 93 93 + }); 94 94 + 95 95 + it("is a no-op when the actor is at or under cap", async () => { 96 96 + await seed("bob", EVENT, 2); 97 97 + expect(await pruneActorFeed(db, "bob", EVENT, 2)).toBe(0); 98 98 + expect(await pruneActorFeed(db, "bob", EVENT, 5)).toBe(0); 99 99 + expect((await rows("bob", EVENT)).length).toBe(2); 100 100 + }); 101 101 + 102 102 + it("only touches the named collection", async () => { 103 103 + await seed("alice", EVENT, 4); 104 104 + await seed("alice", RSVP, 4); 105 105 + await pruneActorFeed(db, "alice", EVENT, 2); 106 106 + expect((await rows("alice", EVENT)).length).toBe(2); 107 107 + expect((await rows("alice", RSVP)).length).toBe(4); // untouched 108 108 + }); 109 109 + }); 110 110 + 111 111 + describe("sweepFeedItems", () => { 112 112 + it("prunes every actor to the per-collection caps in one pass", async () => { 113 113 + await seed("alice", EVENT, 5); 114 114 + await seed("alice", RSVP, 6); 115 115 + await seed("bob", EVENT, 1); 116 116 + await seed("carol", RSVP, 10); 117 117 + 118 118 + const res = await sweepFeedItems(db, CAPS, null, 100); 119 119 + 120 120 + expect(res.done).toBe(true); 121 121 + expect(res.nextCursor).toBeNull(); 122 122 + expect(res.pruned).toBe(3 + 3 + 0 + 7); // alice event/rsvp, bob, carol 123 123 + expect((await rows("alice", EVENT)).length).toBe(2); 124 124 + expect((await rows("alice", RSVP)).length).toBe(3); 125 125 + expect((await rows("bob", EVENT)).length).toBe(1); 126 126 + expect((await rows("carol", RSVP)).length).toBe(3); 127 127 + }); 128 128 + 129 129 + it("pages by actor and resumes via the cursor", async () => { 130 130 + // Three actors, each over the event cap. 131 131 + for (const a of ["a-actor", "b-actor", "c-actor"]) await seed(a, EVENT, 5); 132 132 + 133 133 + // Budget of 1 actor per slice: first slice handles "a-actor". 134 134 + const s1 = await sweepFeedItems(db, CAPS, null, 1); 135 135 + expect(s1.done).toBe(false); 136 136 + expect(s1.nextCursor).toBe("a-actor"); 137 137 + expect((await rows("a-actor", EVENT)).length).toBe(2); 138 138 + expect((await rows("b-actor", EVENT)).length).toBe(5); // not yet reached 139 139 + 140 140 + const s2 = await sweepFeedItems(db, CAPS, s1.nextCursor, 1); 141 141 + expect(s2.nextCursor).toBe("b-actor"); 142 142 + expect((await rows("b-actor", EVENT)).length).toBe(2); 143 143 + 144 144 + const s3 = await sweepFeedItems(db, CAPS, s2.nextCursor, 1); 145 145 + // Last actor — still a full page, so not yet flagged done. 146 146 + expect(s3.nextCursor).toBe("c-actor"); 147 147 + expect((await rows("c-actor", EVENT)).length).toBe(2); 148 148 + 149 149 + // One more slice runs off the end and wraps. 150 150 + const s4 = await sweepFeedItems(db, CAPS, s3.nextCursor, 1); 151 151 + expect(s4.done).toBe(true); 152 152 + expect(s4.nextCursor).toBeNull(); 153 153 + expect(s4.pruned).toBe(0); 154 154 + }); 155 155 + 156 156 + it("returns done with no work for empty caps", async () => { 157 157 + await seed("alice", EVENT, 5); 158 158 + const res = await sweepFeedItems(db, new Map(), null, 100); 159 159 + expect(res).toEqual({ pruned: 0, nextCursor: null, done: true }); 160 160 + expect((await rows("alice", EVENT)).length).toBe(5); 161 161 + }); 162 162 + }); 163 163 + 164 164 + describe("feed prune cursor", () => { 165 165 + it("round-trips and defaults to null", async () => { 166 166 + expect(await getFeedPruneCursor(db)).toBeNull(); 167 167 + await saveFeedPruneCursor(db, "did:plc:xyz"); 168 168 + expect(await getFeedPruneCursor(db)).toBe("did:plc:xyz"); 169 169 + await saveFeedPruneCursor(db, null); 170 170 + expect(await getFeedPruneCursor(db)).toBeNull(); 171 171 + }); 172 172 + }); 173 173 + 174 174 + describe("pruneFeedItems (full recovery loop)", () => { 175 175 + it("brings an already-bloated table within caps in one call", async () => { 176 176 + // Many actors well over cap — the bloated-table recovery scenario. 177 177 + for (let i = 0; i < 25; i++) { 178 178 + await seed(`actor-${String(i).padStart(2, "0")}`, EVENT, 8); 179 179 + await seed(`actor-${String(i).padStart(2, "0")}`, RSVP, 8); 180 180 + } 181 181 + const total = await pruneFeedItems(db, CAPS); 182 182 + expect(total).toBe(25 * (6 + 5)); // event: 8→2, rsvp: 8→3 183 183 + 184 184 + const remaining = await db 185 185 + .prepare("SELECT COUNT(*) AS c FROM feed_items") 186 186 + .first<{ c: number }>(); 187 187 + expect(Number(remaining?.c)).toBe(25 * (2 + 3)); 188 188 + }); 189 189 + });

+26 -10

packages/contrail-appview/src/core/jetstream.ts

Reviewed

··· 6 6 shortNameForNsid, 7 7 buildFeedTargetCaps, 8 8 } from "./types"; 9 9 - import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; 9 9 + import { initSchema, getLastCursor, saveCursor, applyEvents, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./db"; 10 10 import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; 11 11 import { backfillFollowersFromConstellation } from "./constellation"; 12 12 13 13 const BATCH_SIZE = 50; 14 14 - const FEED_PRUNE_INTERVAL_MS = 60 * 60 * 1000; // 1 hour 14 14 + /** Distinct actors pruned per ingest tick by the rolling feed sweep. Each 15 15 + * actor costs a handful of index-backed O(cap) deletes, so this bounds the 16 16 + * prune's per-tick CPU regardless of how large feed_items grows. */ 17 17 + export const FEED_PRUNE_SWEEP_ACTORS = 500; 15 18 16 19 /** Mutable state that persists across ingest cycles within the same process. */ 17 20 export interface IngestState { 18 21 cachedKnownDids?: Set<string>; 19 22 schemaInitialized: boolean; 20 20 - lastFeedPruneMs: number; 23 23 + /** Wall-clock of the last feed sweep — used only by the long-lived 24 24 + * persistent loop to throttle; the recycling cron isolate sweeps every 25 25 + * tick and relies on the persisted cursor instead. */ 26 26 + lastFeedSweepMs: number; 21 27 } 22 28 23 29 export function createIngestState(): IngestState { 24 24 - return { schemaInitialized: false, lastFeedPruneMs: 0 }; 30 30 + return { schemaInitialized: false, lastFeedSweepMs: 0 }; 25 31 } 26 32 27 33 function getLogger(config: ContrailConfig): Logger { ··· 336 342 } 337 343 } 338 344 339 339 - // Prune feed items hourly, per-target so high-volume targets don't 340 340 - // squeeze out lower-volume ones. 341 341 - if (config.feeds && Date.now() - s.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { 345 345 + // Prune feed_items to per-collection caps with a bounded, cursored sweep. 346 346 + // Every statement is index-backed and O(cap) (see sweepFeedItems), so it can 347 347 + // never exhaust D1's per-query CPU budget and reset the shared DO — unlike 348 348 + // the old global window+anti-join. The cron isolate recycles each tick, so we 349 349 + // persist the sweep cursor in the DB rather than gating on in-memory time, 350 350 + // and run an unconditional bounded slice every tick. 351 351 + if (config.feeds) { 342 352 const caps = buildFeedTargetCaps(config); 343 353 if (caps.size > 0) { 344 344 - const pruned = await pruneFeedItems(db, caps); 345 345 - if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); 354 354 + const cursor = await getFeedPruneCursor(db); 355 355 + const { pruned, nextCursor } = await sweepFeedItems( 356 356 + db, 357 357 + caps, 358 358 + cursor, 359 359 + FEED_PRUNE_SWEEP_ACTORS 360 360 + ); 361 361 + await saveFeedPruneCursor(db, nextCursor); 362 362 + if (pruned > 0) log.log(`Pruned ${pruned} feed items (sweep)`); 346 363 } 347 347 - s.lastFeedPruneMs = Date.now(); 348 364 } 349 365 350 366 log.log(`[ingest] cycle complete. stored=${events.length}`);

+20 -7

packages/contrail-appview/src/core/persistent.ts

Reviewed

··· 7 7 resolveConfig, 8 8 shortNameForNsid, 9 9 } from "./types"; 10 10 - import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; 10 10 + import { initSchema, getLastCursor, saveCursor, applyEvents, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./db"; 11 11 import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; 12 12 import { backfillFollowersFromConstellation } from "./constellation"; 13 13 - import { createIngestState } from "./jetstream"; 13 13 + import { createIngestState, FEED_PRUNE_SWEEP_ACTORS } from "./jetstream"; 14 14 import type { IngestState } from "./jetstream"; 15 15 16 16 - const FEED_PRUNE_INTERVAL_MS = 60 * 60 * 1000; 16 16 + /** How often the long-lived persistent loop runs a bounded feed sweep. The 17 17 + * process stays resident, so this in-memory throttle is reliable here (unlike 18 18 + * the recycling cron isolate). */ 19 19 + const FEED_SWEEP_INTERVAL_MS = 10_000; 17 20 18 21 export interface PersistentIngestOptions { 19 22 batchSize?: number; ··· 176 179 } 177 180 } 178 181 179 179 - if (config.feeds && Date.now() - state.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { 182 182 + // Bounded, cursored feed prune (see sweepFeedItems). This process is 183 183 + // long-lived, so the in-memory interval is a reliable throttle; the 184 184 + // cursor is still persisted so progress carries across restarts. 185 185 + if (config.feeds && Date.now() - state.lastFeedSweepMs > FEED_SWEEP_INTERVAL_MS) { 180 186 const caps = buildFeedTargetCaps(config); 181 187 if (caps.size > 0) { 182 182 - const pruned = await pruneFeedItems(db, caps); 183 183 - if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); 188 188 + const cursor = await getFeedPruneCursor(db); 189 189 + const { pruned, nextCursor } = await sweepFeedItems( 190 190 + db, 191 191 + caps, 192 192 + cursor, 193 193 + FEED_PRUNE_SWEEP_ACTORS 194 194 + ); 195 195 + await saveFeedPruneCursor(db, nextCursor); 196 196 + if (pruned > 0) log.log(`Pruned ${pruned} feed items (sweep)`); 184 197 } 185 185 - state.lastFeedPruneMs = Date.now(); 198 198 + state.lastFeedSweepMs = Date.now(); 186 199 } 187 200 188 201 log.log(`Flushed ${batch.length} events. Cursor: ${lastTimeUs}`);

+2 -2

packages/contrail-appview/src/core/db/index.ts

Reviewed

··· 1 1 export { initSchema } from "./schema"; 2 2 - export { getLastCursor, saveCursor, applyEvents, lookupExistingRecords, queryRecords, queryAcrossSources, pruneFeedItems } from "./records"; 3 3 - export type { QueryOptions, SortOption, ExistingRecordInfo } from "./records"; 2 2 + export { getLastCursor, saveCursor, applyEvents, lookupExistingRecords, queryRecords, queryAcrossSources, pruneFeedItems, pruneActorFeed, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./records"; 3 3 + export type { QueryOptions, SortOption, ExistingRecordInfo, FeedSweepResult } from "./records"; 4 4 export type { RecordSource } from "../types";

+159 -33

packages/contrail-appview/src/core/db/records.ts

Reviewed

··· 295 295 296 296 // --- Feed pruning --- 297 297 298 298 - /** Prune feed_items per (actor, collection) to the given cap. 298 298 + /** db.batch chunk size for the sweep — caps statements per transaction. */ 299 299 + const SWEEP_BATCH_SIZE = 50; 300 300 + /** Actor page size for the full-table {@link pruneFeedItems} recovery loop. */ 301 301 + const FEED_PRUNE_RECOVERY_BATCH = 200; 302 302 + 303 303 + /** 304 304 + * Build the bounded per-actor cutoff DELETE for one (actor, collection). 299 305 * 300 300 - * - If `caps` is a number: legacy behavior — global per-actor cap across all collections. 301 301 - * - If `caps` is a Map<collection-NSID, cap>: each collection is pruned independently per actor, 302 302 - * so high-volume collections (e.g. RSVPs) can't squeeze out lower-volume ones (e.g. events). 303 303 - * Collections not present in the map are left alone. 306 306 + * Deletes everything older than the newest `cap` rows, driven directly by 307 307 + * idx_feed_actor_coll_time(actor, collection, time_us DESC). Cost is 308 308 + * O(cap + deleted) — never O(table). This is the ONLY prune shape contrail 309 309 + * issues: an unbounded window/anti-join over the whole table can exhaust D1's 310 310 + * per-query CPU budget and reset the shared Durable Object, which kills any 311 311 + * concurrent read against the same SQLite instance. 312 312 + * 313 313 + * The cutoff is the cap-th newest row (`OFFSET cap - 1`); we delete strictly 314 314 + * older rows. Actors with `cap` or fewer rows: the OFFSET subquery yields no 315 315 + * row, the cutoff is NULL, and `time_us < NULL` matches nothing — a cheap 316 316 + * index no-op. On a tie at the cutoff time_us we keep the extra rows rather 317 317 + * than risk deleting a row we meant to keep (feed_items is a cache; a few over 318 318 + * cap is harmless, dropping a wanted item is not). 319 319 + */ 320 320 + function actorCutoffDelete( 321 321 + db: Database, 322 322 + actor: string, 323 323 + collection: string, 324 324 + cap: number 325 325 + ): Statement { 326 326 + // Plain `?` placeholders (bound repeatedly) rather than numbered params, so 327 327 + // the Postgres adapter's positional `?`→`$n` rewrite stays correct. 328 328 + return db 329 329 + .prepare( 330 330 + `DELETE FROM feed_items 331 331 + WHERE actor = ? AND collection = ? 332 332 + AND time_us < ( 333 333 + SELECT time_us FROM feed_items 334 334 + WHERE actor = ? AND collection = ? 335 335 + ORDER BY time_us DESC LIMIT 1 OFFSET ? 336 336 + )` 337 337 + ) 338 338 + .bind(actor, collection, actor, collection, Math.max(0, cap - 1)); 339 339 + } 340 340 + 341 341 + /** Prune a single actor's feed for one collection to `cap`. Bounded O(cap). */ 342 342 + export async function pruneActorFeed( 343 343 + db: Database, 344 344 + actor: string, 345 345 + collection: string, 346 346 + cap: number 347 347 + ): Promise<number> { 348 348 + const result = await actorCutoffDelete(db, actor, collection, cap).run(); 349 349 + return (result as any)?.changes ?? 0; 350 350 + } 351 351 + 352 352 + export interface FeedSweepResult { 353 353 + /** Rows deleted this slice. */ 354 354 + pruned: number; 355 355 + /** Actor to resume after; null once a full pass completed (wrap to start). */ 356 356 + nextCursor: string | null; 357 357 + /** True when this slice reached the end of the actor list. */ 358 358 + done: boolean; 359 359 + } 360 360 + 361 361 + /** 362 362 + * One bounded slice of a rolling feed-items prune. 363 363 + * 364 364 + * Pages at most `actorBudget` distinct actors (resuming after `cursor`, via the 365 365 + * feed_items (actor, uri) PK) and applies the per-(actor, collection) cutoff 366 366 + * delete for every cap in `caps`. Every issued statement is index-backed and 367 367 + * O(cap), so the slice's per-query CPU stays flat no matter how large 368 368 + * feed_items grows — the property the old global window query lacked. 369 369 + * 370 370 + * Drive it across ticks with a persisted cursor (see getFeedPruneCursor): 371 371 + * feed back `nextCursor` until `done`, at which point the cursor wraps to null 372 372 + * and the next pass starts from the beginning. Because each pass visits every 373 373 + * actor, this doubles as the recovery path for an already-bloated table. 374 374 + */ 375 375 + export async function sweepFeedItems( 376 376 + db: Database, 377 377 + caps: Map<string, number>, 378 378 + cursor: string | null, 379 379 + actorBudget: number 380 380 + ): Promise<FeedSweepResult> { 381 381 + if (caps.size === 0 || actorBudget <= 0) { 382 382 + return { pruned: 0, nextCursor: null, done: true }; 383 383 + } 384 384 + 385 385 + const actorsRes = cursor 386 386 + ? await db 387 387 + .prepare( 388 388 + "SELECT DISTINCT actor FROM feed_items WHERE actor > ? ORDER BY actor LIMIT ?" 389 389 + ) 390 390 + .bind(cursor, actorBudget) 391 391 + .all<{ actor: string }>() 392 392 + : await db 393 393 + .prepare("SELECT DISTINCT actor FROM feed_items ORDER BY actor LIMIT ?") 394 394 + .bind(actorBudget) 395 395 + .all<{ actor: string }>(); 396 396 + 397 397 + const actors = (actorsRes.results ?? []).map((r) => r.actor); 398 398 + if (actors.length === 0) { 399 399 + // Ran off the end (cursor pointed past the last actor) — wrap next tick. 400 400 + return { pruned: 0, nextCursor: null, done: true }; 401 401 + } 402 402 + 403 403 + const stmts: Statement[] = []; 404 404 + for (const actor of actors) { 405 405 + for (const [collection, cap] of caps) { 406 406 + stmts.push(actorCutoffDelete(db, actor, collection, cap)); 407 407 + } 408 408 + } 409 409 + 410 410 + let pruned = 0; 411 411 + for (let i = 0; i < stmts.length; i += SWEEP_BATCH_SIZE) { 412 412 + const results = await db.batch(stmts.slice(i, i + SWEEP_BATCH_SIZE)); 413 413 + for (const r of results) pruned += (r as any)?.changes ?? 0; 414 414 + } 415 415 + 416 416 + // A short page means we exhausted the actor list this slice. 417 417 + const done = actors.length < actorBudget; 418 418 + return { pruned, nextCursor: done ? null : actors[actors.length - 1], done }; 419 419 + } 420 420 + 421 421 + /** 422 422 + * Prune the ENTIRE feed_items table to the per-collection `caps` by looping the 423 423 + * bounded {@link sweepFeedItems} until a full pass completes. 424 424 + * 425 425 + * Every statement is O(cap) and safe against D1's per-query CPU limit, but the 426 426 + * statement count is O(distinct actors), so keep this OFF the hot ingest path — 427 427 + * the cron/persistent loops issue a single bounded slice per tick instead. Use 428 428 + * it for one-shot recovery or admin tooling. 304 429 */ 305 430 export async function pruneFeedItems( 306 431 db: Database, 307 307 - caps: number | Map<string, number> 432 432 + caps: Map<string, number> 308 433 ): Promise<number> { 309 309 - if (typeof caps === "number") { 310 310 - const result = await db 311 311 - .prepare( 312 312 - `DELETE FROM feed_items WHERE (actor, uri) NOT IN ( 313 313 - SELECT actor, uri FROM ( 314 314 - SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn 315 315 - FROM feed_items 316 316 - ) sub WHERE rn <= ? 317 317 - )` 318 318 - ) 319 319 - .bind(caps) 320 320 - .run(); 321 321 - return (result as any)?.changes ?? 0; 322 322 - } 323 434 let total = 0; 324 324 - for (const [collection, cap] of caps) { 325 325 - const result = await db 326 326 - .prepare( 327 327 - `DELETE FROM feed_items WHERE collection = ? AND (actor, uri) NOT IN ( 328 328 - SELECT actor, uri FROM ( 329 329 - SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn 330 330 - FROM feed_items WHERE collection = ? 331 331 - ) sub WHERE rn <= ? 332 332 - )` 333 333 - ) 334 334 - .bind(collection, collection, cap) 335 335 - .run(); 336 336 - total += (result as any)?.changes ?? 0; 435 435 + let cursor: string | null = null; 436 436 + for (;;) { 437 437 + const res = await sweepFeedItems(db, caps, cursor, FEED_PRUNE_RECOVERY_BATCH); 438 438 + total += res.pruned; 439 439 + if (res.done) break; 440 440 + cursor = res.nextCursor; 337 441 } 338 442 return total; 443 443 + } 444 444 + 445 445 + // --- Feed prune cursor --- 446 446 + 447 447 + /** Last actor swept by the rolling feed prune; null = start of a fresh pass. */ 448 448 + export async function getFeedPruneCursor(db: Database): Promise<string | null> { 449 449 + const row = await db 450 450 + .prepare("SELECT actor FROM feed_prune_cursor WHERE id = 1") 451 451 + .first<{ actor: string | null }>(); 452 452 + return row?.actor ?? null; 453 453 + } 454 454 + 455 455 + export async function saveFeedPruneCursor( 456 456 + db: Database, 457 457 + actor: string | null 458 458 + ): Promise<void> { 459 459 + await db 460 460 + .prepare( 461 461 + "INSERT INTO feed_prune_cursor (id, actor) VALUES (1, ?) ON CONFLICT(id) DO UPDATE SET actor = excluded.actor" 462 462 + ) 463 463 + .bind(actor) 464 464 + .run(); 339 465 } 340 466 341 467 // --- Cursor ---

+5

packages/contrail-appview/src/core/db/schema.ts

Reviewed

··· 338 338 )`, 339 339 `CREATE INDEX IF NOT EXISTS idx_feed_actor_coll_time ON feed_items(actor, collection, time_us DESC)`, 340 340 `CREATE INDEX IF NOT EXISTS idx_feed_actor_time ON feed_items(actor, time_us DESC)`, 341 341 + // Single-row cursor for the rolling, bounded feed prune (see sweepFeedItems). 342 342 + `CREATE TABLE IF NOT EXISTS feed_prune_cursor ( 343 343 + id INTEGER PRIMARY KEY CHECK (id = 1), 344 344 + actor TEXT 345 345 + )`, 341 346 `CREATE TABLE IF NOT EXISTS feed_backfills ( 342 347 actor TEXT NOT NULL, 343 348 feed TEXT NOT NULL,