diff --git a/.changeset/feed-prune-skip-idle-ticks.md b/.changeset/feed-prune-skip-idle-ticks.md new file mode 100644 index 0000000..a34e5f2 --- /dev/null +++ b/.changeset/feed-prune-skip-idle-ticks.md @@ -0,0 +1,8 @@ +--- +"@atmo-dev/contrail-appview": patch +"@atmo-dev/contrail-base": patch +--- + +Stop running the `feed_items` prune sweep on every ingest tick. + +A feed only exceeds its cap right after a feed-mutating record, so the per-tick sweep was a no-op on the vast majority of ticks yet still issued a cutoff `DELETE` per actor (~98% of all D1 queries on one deployment). It now sweeps only when a feed-mutating collection was ingested, plus a recovery pass that becomes due ~6h after the previous one completed and then laps one slice per tick — including on idle persistent streams and the `notifyOfUpdate` path. New `getFeedMutatingNsids(config)` derives the gating set. See `docs/04-feeds.md` for sweep timing (and why the full-pass cadence is interval + lap time, not a hard 6h) and the fan-out promptness trade-off. diff --git a/docs/04-feeds.md b/docs/04-feeds.md index 16e384a..3ecf741 100644 --- a/docs/04-feeds.md +++ b/docs/04-feeds.md @@ -87,7 +87,22 @@ Three moments: ## Pruning -The persistent worker (or cron run) trims `feed_items` per actor down to the largest configured `maxItems` across feeds, keeping newest by `time_us`. There is no per-feed prune — one global cap. +Feeds are capped: each actor keeps at most `maxItems` rows per target collection (default 200, newest first). Older rows past the cap are deleted by a background cleanup that piggybacks on ingestion — there is no separate prune job. + +A few terms used below: + +- **Tick** — one cycle of the ingest loop. In cron mode the worker wakes on a schedule (e.g. once a minute) and each wake-up is a tick; in the persistent loop it's each batch flush. +- **Sweep** — the cleanup that walks `feed_items` actor by actor and deletes whatever is over an actor's cap. +- **Slice** — a sweep doesn't scan the whole table at once. Each tick it handles a chunk of up to `FEED_PRUNE_SWEEP_ACTORS` actors (default 500). That chunk is one slice. +- **Cursor / full pass** — a bookmark for the last actor a slice stopped on, so the next slice resumes after it instead of restarting. When the cursor reaches the last actor it *wraps* back to the start; one start-to-end trip is a *full pass*. + +**When the sweep runs.** A feed can only go over its cap right after a feed-mutating record (a target fan-out or a follow backfill) is applied, so the sweep is skipped entirely on ticks that ingested nothing feed-relevant. It runs when the current tick — a cron run, a persistent-loop flush, or a `notifyOfUpdate` call — applied a feed-mutating record. As a safety net it also runs on a recovery interval (`FEED_PRUNE_RECOVERY_INTERVAL_MS`, 6h), so rows that went over cap without a fresh ingest (a lowered cap, a bulk import) still get cleaned up — including on a stream that is otherwise idle. + +Doing one slice per tick keeps each tick's cost flat no matter how big the table grows. The recovery timer measures from the last *completed full pass* (not the last slice): a fresh pass becomes due one recovery interval after the previous one finished, then advances a slice per tick until the cursor wraps. So a full pass *completes* roughly every `recovery interval + lap time`, where lap time is `ceil(actors / FEED_PRUNE_SWEEP_ACTORS)` ticks — e.g. with 100k actors and one-minute cron ticks, ~6h + ~3h20m. That keeps the whole table draining on a bounded cadence; it is not a hard "fully clean every 6h" guarantee. Raise `FEED_PRUNE_SWEEP_ACTORS` if you need the lap time shorter at large actor counts. + +**Fan-out isn't cleaned up instantly.** A slice cleans up whatever actors the cursor lands on next — not specifically the actors whose feeds just changed. So when a popular author posts and fans out to many followers: a follower the cursor *hasn't reached yet* this pass is trimmed later in the same pass (soon), but a follower the cursor has *already passed* waits for the next pass — and on a quiet stream the next pass only starts on the recovery interval. So the worst case for an over-cap follower is roughly one recovery interval (`FEED_PRUNE_RECOVERY_INTERVAL_MS`, 6h), not the next tick. + +This is on purpose: an author can have unboundedly many followers, and trimming every one on the spot would either overrun the per-tick request budget (one delete per follower) or overrun D1's per-query CPU limit (one big delete over all of them, which can reset the shared Durable Object). `feed_items` is just a cache, so a follower sitting a little over cap for up to an interval does no harm. Deployments with fewer than `FEED_PRUNE_SWEEP_ACTORS` (500) distinct feed actors clean the whole table on every triggered tick, so they never see this lag at all. Pruning the touched actors directly (instead of the rolling cursor) would remove the lag but trade the bounded per-tick cost for cost proportional to fan-out size; see the issue tracker for that trade-off. ## Deletes diff --git a/packages/contrail-appview/src/core/jetstream.ts b/packages/contrail-appview/src/core/jetstream.ts index ea0b479..2c03660 100644 --- a/packages/contrail-appview/src/core/jetstream.ts +++ b/packages/contrail-appview/src/core/jetstream.ts @@ -5,6 +5,7 @@ import { getDependentNsids, shortNameForNsid, buildFeedTargetCaps, + getFeedMutatingNsids, optimizeEnabled, optimizeIntervalMs, optimizeAnalysisLimit, @@ -19,6 +20,24 @@ const BATCH_SIZE = 50; * prune's per-tick CPU regardless of how large feed_items grows. */ export const FEED_PRUNE_SWEEP_ACTORS = 500; +/** How long after a completed full pass the recovery sweep becomes due again, + * even when no feed-relevant records are ingested, so over-cap rows that + * predate a config change (e.g. a lowered cap) or a bulk import still drain. + * Once due, the pass advances one slice per tick, so a full pass *completes* + * roughly every (this interval + lap time), where lap time is + * ceil(actors / FEED_PRUNE_SWEEP_ACTORS) ticks. Steady-state pruning is driven + * by ingest; this is only the safety net. */ +export const FEED_PRUNE_RECOVERY_INTERVAL_MS = 6 * 60 * 60 * 1000; // 6h + +/** `_contrail_meta` key for the wall-clock (ms) at which the rolling feed sweep + * last *completed a full pass* over every actor, so a recycled cron isolate can + * honor the recovery interval across ticks. Tracking pass completion (not the + * last slice) is what keeps the recovery interval measuring from a real drain + * rather than from one bounded slice — without it, a single slice resets the + * clock and a feed touched just after the cursor passed it could wait many + * intervals to be revisited. */ +const FEED_PRUNE_LAST_FULL_PASS_META = "feed_prune_last_full_pass_ms"; + /** `_contrail_meta` key for the persisted optimize cadence (so recycled cron * isolates don't re-run it every tick — the in-memory-state bug we hit with * the feed prune). Shared by the persistent loop. */ @@ -42,18 +61,95 @@ export async function maybeOptimize(db: Database, config: ContrailConfig, log: L } } +/** One bounded feed-prune slice: advances the persisted rolling cursor by up to + * {@link FEED_PRUNE_SWEEP_ACTORS} actors and reports whether the slice reached + * the end of the actor list (i.e. a full pass just completed and the cursor + * wrapped). Callers decide WHEN to sweep (ingest-dirty vs recovery); this owns + * the slice + cursor mechanics so the cron loop, the persistent loop, and the + * notify path all prune identically. No-op (done) when no feed caps apply. */ +export async function runFeedPruneSlice( + db: Database, + config: ContrailConfig +): Promise<{ pruned: number; done: boolean }> { + const caps = buildFeedTargetCaps(config); + if (caps.size === 0) return { pruned: 0, done: true }; + const cursor = await getFeedPruneCursor(db); + const { pruned, nextCursor, done } = await sweepFeedItems( + db, + caps, + cursor, + FEED_PRUNE_SWEEP_ACTORS + ); + await saveFeedPruneCursor(db, nextCursor); + return { pruned, done }; +} + +/** Gate and run one feed-prune slice against the *persisted* recovery clock — + * shared by the recycling cron isolate and the stateless `notifyOfUpdate` path + * (the long-lived persistent loop uses its in-memory clocks instead). Slices + * when `feedTouched` (a feed-mutating record was just ingested) or when a full + * pass is overdue, and records pass completion so the recovery clock measures + * from a real drain (one slice per tick until the cursor wraps) rather than + * resetting on a single slice. + * + * The slice advances the shared rolling cursor, which is NOT necessarily the + * actor the mutation touched: a fan-out follower the cursor has already passed + * is pruned by the next pass, up to about one recovery interval later, not + * instantly. That is + * the deliberate trade for a per-tick cost bounded by `FEED_PRUNE_SWEEP_ACTORS` + * rather than by fan-out size (a popular author has unboundedly many followers). + * feed_items is a soft cache, so a follower sitting a few rows over cap until + * the next slice is harmless. No-op when feeds are unconfigured. */ +export async function runGatedFeedPrune( + db: Database, + config: ContrailConfig, + feedTouched: boolean +): Promise { + if (!config.feeds) return; + if (buildFeedTargetCaps(config).size === 0) return; + const nowMs = Date.now(); + const lastFullPassMs = + (await getMetaNumber(db, FEED_PRUNE_LAST_FULL_PASS_META)) ?? 0; + const recoveryDue = nowMs - lastFullPassMs >= FEED_PRUNE_RECOVERY_INTERVAL_MS; + if (!feedTouched && !recoveryDue) return; + const { pruned, done } = await runFeedPruneSlice(db, config); + if (done) await setMeta(db, FEED_PRUNE_LAST_FULL_PASS_META, String(nowMs)); + if (pruned > 0) { + getLogger(config).log( + `Pruned ${pruned} feed items (sweep, reason=${feedTouched ? "ingest" : "recovery"})` + ); + } +} + /** Mutable state that persists across ingest cycles within the same process. */ export interface IngestState { cachedKnownDids?: Set; schemaInitialized: boolean; - /** 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. */ + /** Wall-clock of the last feed sweep slice — used only by the long-lived + * persistent loop to throttle ingest-driven slices; the recycling cron + * isolate persists its clocks in `_contrail_meta` instead. */ lastFeedSweepMs: number; + /** Wall-clock at which the persistent loop last *completed a full sweep pass* + * over every actor. Drives the recovery interval (a fresh pass becomes due + * {@link FEED_PRUNE_RECOVERY_INTERVAL_MS} after the last one completed, then + * laps one slice per tick), independent of the ingest-driven throttle above. + * The cron isolate persists the equivalent in + * `_contrail_meta`. */ + lastFullFeedPassMs: number; + /** Set by the persistent loop when a flushed batch ingested a feed-mutating + * record, so the next sweep window knows there may be prune work. Cleared + * when the sweep runs. The cron path makes the same decision per-tick from + * its `events` array and doesn't need the flag. */ + feedDirty: boolean; } export function createIngestState(): IngestState { - return { schemaInitialized: false, lastFeedSweepMs: 0 }; + return { + schemaInitialized: false, + lastFeedSweepMs: 0, + lastFullFeedPassMs: 0, + feedDirty: false, + }; } function getLogger(config: ContrailConfig): Logger { @@ -430,22 +526,25 @@ export async function runIngestCycle( // 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. + // the old global window+anti-join. The cron isolate recycles each tick, so the + // sweep cursor and the recovery clock both live in the DB. + // + // A feed only goes over cap right after a row is inserted, and rows are only + // inserted for feed-mutating collections (event fan-out, follow backfill). So + // we skip the sweep entirely on ticks that ingested nothing feed-relevant — + // the overwhelming majority — and otherwise advance one bounded slice. + // + // The recovery clock tracks when a *full pass* over every actor last + // completed, not the last slice: while a pass is overdue we keep slicing every + // tick (bounded cost) until the cursor wraps, then reset the clock. That bounds + // the worst-case time an over-cap feed waits to be revisited — a feed touched + // just after the cursor passed it, a lowered cap, or a bulk import all drain + // within one recovery interval plus the pass's lap time, instead of stalling + // for many intervals (one slice per interval) as a per-slice clock would. if (config.feeds) { - const caps = buildFeedTargetCaps(config); - if (caps.size > 0) { - 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)`); - } + const feedMutatingNsids = getFeedMutatingNsids(config); + const feedTouched = events.some((e) => feedMutatingNsids.has(e.collection)); + await runGatedFeedPrune(db, config, feedTouched); } // Opt-in planner-stat maintenance (gated + persisted cadence; no-op unless diff --git a/packages/contrail-appview/src/core/persistent.ts b/packages/contrail-appview/src/core/persistent.ts index 391dd55..a8234e6 100644 --- a/packages/contrail-appview/src/core/persistent.ts +++ b/packages/contrail-appview/src/core/persistent.ts @@ -4,13 +4,19 @@ import { getCollectionNsids, getDependentNsids, buildFeedTargetCaps, + getFeedMutatingNsids, resolveConfig, shortNameForNsid, } from "./types"; -import { initSchema, getLastCursor, saveCursor, applyEvents, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./db"; +import { initSchema, getLastCursor, saveCursor, applyEvents } from "./db"; import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; import { backfillFollowersFromConstellation } from "./constellation"; -import { createIngestState, FEED_PRUNE_SWEEP_ACTORS, maybeOptimize } from "./jetstream"; +import { + createIngestState, + runFeedPruneSlice, + FEED_PRUNE_RECOVERY_INTERVAL_MS, + maybeOptimize, +} from "./jetstream"; import type { IngestState } from "./jetstream"; /** How often the long-lived persistent loop runs a bounded feed sweep. The @@ -147,61 +153,87 @@ async function streamAndFlush( let flushing = false; const flush = async () => { - if (buffer.length === 0 || flushing) return; + if (flushing) return; flushing = true; - const batch = buffer.splice(0); try { - await applyEvents(db, batch, config, { pubsub: opts.pubsub }); - - const lastTimeUs = Math.max(...batch.map((e) => e.time_us)); - await saveCursor(db, lastTimeUs); - - const uniqueDids = [...new Set(batch.map((e) => e.did))]; - if (uniqueDids.length > 0) { - try { - await refreshStaleIdentities(db, uniqueDids, config); - } catch (err) { - log.warn(`Identity refresh failed: ${err}`); + if (buffer.length > 0) { + const batch = buffer.splice(0); + await applyEvents(db, batch, config, { pubsub: opts.pubsub }); + + // A feed can only go over cap right after a feed-mutating record is + // applied, so remember whether this batch had one. The sweep below uses + // it to prune promptly (see the cron path in jetstream.ts). + if (config.feeds) { + const feedMutatingNsids = getFeedMutatingNsids(config); + if (batch.some((e) => feedMutatingNsids.has(e.collection))) { + state.feedDirty = true; + } } - } - // Drain newly-known DIDs and ask Constellation for back-edges. - if (config.feeds && opts.newlyKnownDids && opts.newlyKnownDids.size > 0) { - const drained = [...opts.newlyKnownDids]; - opts.newlyKnownDids.clear(); - for (const subj of drained) { + const lastTimeUs = Math.max(...batch.map((e) => e.time_us)); + await saveCursor(db, lastTimeUs); + + const uniqueDids = [...new Set(batch.map((e) => e.did))]; + if (uniqueDids.length > 0) { try { - await backfillFollowersFromConstellation(db, config, subj); + await refreshStaleIdentities(db, uniqueDids, config); } catch (err) { - log.warn(`[constellation] subject=${subj} failed: ${err}`); + log.warn(`Identity refresh failed: ${err}`); } } + + // Drain newly-known DIDs and ask Constellation for back-edges. + if (config.feeds && opts.newlyKnownDids && opts.newlyKnownDids.size > 0) { + const drained = [...opts.newlyKnownDids]; + opts.newlyKnownDids.clear(); + for (const subj of drained) { + try { + await backfillFollowersFromConstellation(db, config, subj); + } catch (err) { + log.warn(`[constellation] subject=${subj} failed: ${err}`); + } + } + } + + // Opt-in planner-stat maintenance (gated + persisted cadence). + await maybeOptimize(db, config, log); + + log.log(`Flushed ${batch.length} events. Cursor: ${lastTimeUs}`); } - // 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) { + // Bounded, cursored feed prune (see sweepFeedItems / runFeedPruneSlice). + // Runs whether or not this tick had events: ingest-dirty windows prune + // promptly (throttled by the sweep interval), and the recovery interval + // still fires on a fully idle stream — the timer drives this flush, and + // the old "return early when the buffer is empty" path starved recovery, + // so over-cap rows from a lowered cap or a bulk import never drained while + // the stream was quiet. The recovery clock tracks the last *completed* + // full pass (not the last slice), so an overdue pass keeps slicing each + // tick until the cursor wraps rather than advancing one slice per interval. + if (config.feeds) { const caps = buildFeedTargetCaps(config); if (caps.size > 0) { - 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)`); + const now = Date.now(); + const dirtyDue = + state.feedDirty && now - state.lastFeedSweepMs > FEED_SWEEP_INTERVAL_MS; + const recoveryDue = + now - state.lastFullFeedPassMs > FEED_PRUNE_RECOVERY_INTERVAL_MS; + if (dirtyDue || recoveryDue) { + const { pruned, done } = await runFeedPruneSlice(db, config); + if (done) state.lastFullFeedPassMs = now; + if (dirtyDue) { + state.feedDirty = false; + state.lastFeedSweepMs = now; + } + if (pruned > 0) { + log.log( + `Pruned ${pruned} feed items (sweep, reason=${dirtyDue ? "ingest" : "recovery"})` + ); + } + } } - state.lastFeedSweepMs = Date.now(); } - - // Opt-in planner-stat maintenance (gated + persisted cadence). - await maybeOptimize(db, config, log); - - log.log(`Flushed ${batch.length} events. Cursor: ${lastTimeUs}`); } finally { flushing = false; } diff --git a/packages/contrail-appview/src/core/router/notify.ts b/packages/contrail-appview/src/core/router/notify.ts index 63b407b..f6c51c3 100644 --- a/packages/contrail-appview/src/core/router/notify.ts +++ b/packages/contrail-appview/src/core/router/notify.ts @@ -1,7 +1,8 @@ import type { Hono } from "hono"; import type { Database, ContrailConfig, IngestEvent } from "../types"; -import { shortNameForNsid } from "../types"; +import { shortNameForNsid, getFeedMutatingNsids } from "../types"; import { applyEvents, lookupExistingRecords } from "../db/records"; +import { runGatedFeedPrune } from "../jetstream"; import { getPDS } from "../client"; import type { Did } from "@atcute/lexicons"; import { parseCanonicalResourceUri } from "@atcute/lexicons/syntax"; @@ -136,6 +137,19 @@ export async function processNotifyUris( await applyEvents(db, events, config, { existing }); } + // applyEvents fans these records into feed_items exactly like the cron and + // persistent ingest paths, so prune here too — otherwise a notify-only + // deployment (no jetstream loop) would never sweep. Run the recovery-aware + // gate on every call, not only when records changed: a notify-only deployment + // that receives no-op notifications (a same-CID re-notify produces no events) + // must still be able to advance an overdue recovery pass. `feedTouched` is + // true only when this call actually applied a feed-mutating record. + if (config.feeds) { + const feedMutatingNsids = getFeedMutatingNsids(config); + const feedTouched = events.some((e) => feedMutatingNsids.has(e.collection)); + await runGatedFeedPrune(db, config, feedTouched); + } + return { indexed: events.filter((e) => e.operation === "create" || e.operation === "update").length, deleted: events.filter((e) => e.operation === "delete").length, diff --git a/packages/contrail-base/src/types.ts b/packages/contrail-base/src/types.ts index f6e20d3..823644e 100644 --- a/packages/contrail-base/src/types.ts +++ b/packages/contrail-base/src/types.ts @@ -482,6 +482,27 @@ export function getFeedFollowShortNames(config: ContrailConfig): string[] { /** Alias for getFeedFollowShortNames. */ export const getFeedFollowCollections = getFeedFollowShortNames; +/** + * NSIDs whose ingest can mutate `feed_items`: feed *target* collections (a + * create/update fans out to followers, a delete tears the item down) and feed + * *follow* collections (a follow backfills the follower's feed, an unfollow + * removes it). These are the only records that can push a feed over its cap, so + * a tick that ingested none of them cannot have created prune work — callers use + * this to skip the feed sweep on idle ticks. Returns an empty set when no feeds + * are configured. */ +export function getFeedMutatingNsids(config: ContrailConfig): Set { + const nsids = new Set(); + if (!config.feeds) return nsids; + for (const targetNsid of buildFeedTargetCaps(config).keys()) { + nsids.add(targetNsid); + } + for (const short of getFeedFollowShortNames(config)) { + const nsid = nsidForShortName(config, short); + if (nsid) nsids.add(nsid); + } + return nsids; +} + // Record types export interface RecordRow { diff --git a/packages/contrail/tests/feed-prune.test.ts b/packages/contrail/tests/feed-prune.test.ts index a1f6e5c..7054372 100644 --- a/packages/contrail/tests/feed-prune.test.ts +++ b/packages/contrail/tests/feed-prune.test.ts @@ -10,6 +10,7 @@ import { getFeedPruneCursor, saveFeedPruneCursor, } from "../src/core/db/records"; +import { runGatedFeedPrune } from "../src/core/jetstream"; const EVENT = "community.lexicon.calendar.event"; const RSVP = "community.lexicon.calendar.rsvp"; @@ -187,3 +188,30 @@ describe("pruneFeedItems (full recovery loop)", () => { expect(Number(remaining?.c)).toBe(25 * (2 + 3)); }); }); + +describe("runGatedFeedPrune (recovery gate)", () => { + it("sweeps when a full pass is overdue, even with feedTouched=false", async () => { + // The persisted recovery clock is unset (0), so a pass is overdue. This is + // the path a no-op notify call or an idle stream relies on to make progress. + await seed("alice", EVENT, 5); // cap 2 + await runGatedFeedPrune(db, CONFIG, false); + expect(await rows("alice", EVENT)).toHaveLength(2); + }); + + it("is a no-op when not feed-touched and recovery is not yet due", async () => { + // First call completes a pass and stamps the recovery clock to ~now. + await runGatedFeedPrune(db, CONFIG, false); + // A fresh over-cap actor is left alone: recovery isn't due and nothing was + // ingested this call. + await seed("bob", EVENT, 5); + await runGatedFeedPrune(db, CONFIG, false); + expect(await rows("bob", EVENT)).toHaveLength(5); + }); + + it("sweeps on feedTouched even when recovery is not due", async () => { + await runGatedFeedPrune(db, CONFIG, false); // stamp the recovery clock + await seed("bob", EVENT, 5); + await runGatedFeedPrune(db, CONFIG, true); + expect(await rows("bob", EVENT)).toHaveLength(2); + }); +}); diff --git a/packages/contrail/tests/persistent.test.ts b/packages/contrail/tests/persistent.test.ts index a962d93..f8ef807 100644 --- a/packages/contrail/tests/persistent.test.ts +++ b/packages/contrail/tests/persistent.test.ts @@ -412,4 +412,50 @@ describe("runPersistent", () => { "newhandle.test" ); }); + + it("runs the recovery feed sweep on a fully idle stream", async () => { + // Regression: the feed sweep used to sit behind flush()'s early return when + // the buffer was empty, so over-cap rows (a lowered cap, a bulk import) + // never drained while Jetstream was quiet. The recovery clock starts at 0, + // so the recovery sweep is due on the first idle timer flush. + const EVENT = "community.lexicon.calendar.event"; + const feedConfig = resolveConfig({ + namespace: "com.example", + collections: { event: { collection: EVENT } }, + feeds: { main: { targets: [{ collection: "event", maxItems: 2 }] } }, + }); + const feedDb = createTestDb(); + await initSchema(feedDb, feedConfig); + + // Seed five rows for one actor against a cap of two — over cap, predating + // any ingest. No events will ever flow on this run. + for (let i = 0; i < 5; i++) { + await feedDb + .prepare( + "INSERT INTO feed_items (actor, uri, collection, time_us) VALUES (?, ?, ?, ?)" + ) + .bind("did:plc:alice", `at://did:plc:alice/${EVENT}/${i}`, EVENT, 1000 + i) + .run(); + } + + const controller = new AbortController(); + const promise = runPersistent(feedDb, feedConfig, { + batchSize: 100, + flushIntervalMs: 50, + signal: controller.signal, + createSubscription: () => mockSubscription([]) as any, + }); + + await new Promise((r) => setTimeout(r, 300)); + controller.abort(); + await promise; + + const res = await feedDb + .prepare( + "SELECT COUNT(*) AS n FROM feed_items WHERE actor = ? AND collection = ?" + ) + .bind("did:plc:alice", EVENT) + .first<{ n: number }>(); + expect(Number(res?.n)).toBe(2); + }); });