From b0ea183024193f31fab7fb44b101522af7ae9d1a Mon Sep 17 00:00:00 2001 From: Andrew Lisowski Date: Sun, 4 Oct 2026 20:30:42 -0700 Subject: [PATCH] Fix Mentions Watcher OOM and Tap Sync at_uri conflicts Jetstream consumer: process messages serially through a bounded queue, pausing the socket above 1000 queued messages, and throttle cursor saves to once per second. Previously every message spawned a concurrent DB write, so replaying a saved cursor piled up handlers until the 4 GB heap ran out, leaving the service in a crash loop. Tap listing ingest: reconcile the record onto the row that owns its at_uri before the slug-keyed upsert. A record whose slug didn't match its existing row failed on store_listings_at_uri_idx and was retried forever. Co-Authored-By: Claude Opus 5.5 --- scripts/jetstream-consumer.ts | 111 ++++++++++++++++++++-------- src/lib/atproto/tap-listing-sync.ts | 56 ++++++++++++-- 2 files changed, 130 insertions(+), 37 deletions(-) diff --git a/scripts/jetstream-consumer.ts b/scripts/jetstream-consumer.ts index 0e6859b..8156542 100644 --- a/scripts/jetstream-consumer.ts +++ b/scripts/jetstream-consumer.ts @@ -64,6 +64,12 @@ function parseJetstreamTimeUs(line: string): number | null { } } +/** Pause the socket above this many queued messages; resume once drained to the low-water mark. */ +const QUEUE_HIGH_WATER = 1000; +const QUEUE_LOW_WATER = 100; +/** The cursor only needs to be roughly current — the reconnect overlap buffer covers the gap. */ +const CURSOR_SAVE_INTERVAL_MS = 1000; + function buildJetstreamUrl(cursor?: number): string { const raw = process.env.JETSTREAM_URL?.trim() || @@ -236,6 +242,51 @@ async function main() { statsTimer.unref?.(); } + let lastCursorSavedAt = 0; + const saveCursor = async (force = false) => { + if (lastCursor == null) return; + const now = Date.now(); + if (!force && now - lastCursorSavedAt < CURSOR_SAVE_INTERVAL_MS) return; + lastCursorSavedAt = now; + await saveJetstreamCursor(db, lastCursor); + stats.cursorSaves += 1; + }; + + const processLine = async (line: string) => { + const timeUs = parseJetstreamTimeUs(line); + let result: JetstreamIngestResult | null = null; + try { + result = await ingestJetstreamCommitLine(db, line, index); + applyIngestStats(stats, result); + if (result) { + logDirectoryListingMatch(result); + logVerboseIngest(result); + } + } catch (error) { + stats.ingestErrors += 1; + log("error", "ingest_threw", { + err: error instanceof Error ? error.message : String(error), + stack: error instanceof Error ? error.stack : undefined, + linePreview: line.length > 200 ? `${line.slice(0, 200)}…` : line, + }); + } + const cursorToSave = timeUs ?? result?.time_us ?? null; + if (cursorToSave != null) { + lastCursor = cursorToSave; + try { + await saveCursor(); + } catch (error) { + log("error", "cursor_save_failed", { + err: error instanceof Error ? error.message : String(error), + }); + } + } else if (result) { + log("warn", "no_cursor_in_event", { + processed: result.processed, + }); + } + }; + const connect = () => { const url = buildJetstreamUrl(lastCursor); log("info", "connecting", { @@ -252,39 +303,37 @@ async function main() { }); }); - ws.on("message", async (data: WebSocket.RawData) => { - stats.messages += 1; - const line = - typeof data === "string" - ? data - : Buffer.from(data as Buffer).toString("utf8"); - const timeUs = parseJetstreamTimeUs(line); - let result: JetstreamIngestResult | null = null; + /** + * Process messages one at a time and pause the socket while the queue is backed up. Without + * backpressure every message spawned its own pending DB write, and replaying a saved cursor + * (~3.5k msgs/sec) piled up handlers until the heap ran out. + */ + const queue: Array = []; + let draining = false; + + const drain = async () => { + if (draining) return; + draining = true; try { - result = await ingestJetstreamCommitLine(db, line, index); - applyIngestStats(stats, result); - if (result) { - logDirectoryListingMatch(result); - logVerboseIngest(result); + let line: string | undefined; + while ((line = queue.shift()) !== undefined) { + await processLine(line); + if (queue.length <= QUEUE_LOW_WATER && ws.isPaused) ws.resume(); } - } catch (error) { - stats.ingestErrors += 1; - log("error", "ingest_threw", { - err: error instanceof Error ? error.message : String(error), - stack: error instanceof Error ? error.stack : undefined, - linePreview: line.length > 200 ? `${line.slice(0, 200)}…` : line, - }); - } - const cursorToSave = timeUs ?? result?.time_us ?? null; - if (cursorToSave != null) { - lastCursor = cursorToSave; - await saveJetstreamCursor(db, cursorToSave); - stats.cursorSaves += 1; - } else if (result) { - log("warn", "no_cursor_in_event", { - processed: result.processed, - }); + } finally { + draining = false; } + }; + + ws.on("message", (data: WebSocket.RawData) => { + stats.messages += 1; + queue.push( + typeof data === "string" + ? data + : Buffer.from(data as Buffer).toString("utf8"), + ); + if (queue.length >= QUEUE_HIGH_WATER && !ws.isPaused) ws.pause(); + void drain(); }); ws.on("close", (code, reason) => { @@ -315,7 +364,7 @@ async function main() { shouldRun = false; if (statsTimer) clearInterval(statsTimer); log("info", "shutdown", { finalStats: { ...stats } }); - process.exit(0); + void saveCursor(true).finally(() => process.exit(0)); }; process.on("SIGINT", shutdown); process.on("SIGTERM", shutdown); diff --git a/src/lib/atproto/tap-listing-sync.ts b/src/lib/atproto/tap-listing-sync.ts index e1c4199..5368545 100644 --- a/src/lib/atproto/tap-listing-sync.ts +++ b/src/lib/atproto/tap-listing-sync.ts @@ -359,6 +359,49 @@ function resolveListingVerificationStatus(input: { return "verified"; } +/** + * The upsert below is keyed on `slug`, but `at_uri` is also unique. When the row already mirroring + * this record has a different slug (slug edited on the PDS outside our API, or a record reusing + * another listing's slug), a plain slug upsert either inserts a duplicate `at_uri` or repoints + * another row onto it — both fail with `store_listings_at_uri_idx`, and Tap retries the event forever. + * + * Reconcile onto the row that owns `atUri`: adopt the record's slug when it is free, otherwise keep + * the row's current slug. + */ +async function resolveTapIngestSlug( + db: Database, + atUri: string, + recordSlug: string, +): Promise { + const [rowByAtUri] = await db + .select({ id: schema.storeListings.id, slug: schema.storeListings.slug }) + .from(schema.storeListings) + .where(eq(schema.storeListings.atUri, atUri)) + .limit(1); + if (!rowByAtUri || rowByAtUri.slug === recordSlug) return recordSlug; + + const [slugOwner] = await db + .select({ atUri: schema.storeListings.atUri }) + .from(schema.storeListings) + .where(eq(schema.storeListings.slug, recordSlug)) + .limit(1); + if (slugOwner) { + console.warn( + `[tap-ingest] ${atUri} record slug=${recordSlug} already owned by ${slugOwner.atUri ?? "(no at_uri)"}; keeping slug=${rowByAtUri.slug}`, + ); + return rowByAtUri.slug; + } + + console.log( + `[tap-ingest] ${atUri} slug changed ${rowByAtUri.slug} -> ${recordSlug}`, + ); + await db + .update(schema.storeListings) + .set({ slug: recordSlug }) + .where(eq(schema.storeListings.id, rowByAtUri.id)); + return recordSlug; +} + /** * Upsert `store_listings` from Tap (`fyi.atstore.listing.detail` only — does not touch `directory_listings`). * Resolves icon / hero / screenshots to Bluesky CDN URLs from blob refs (Kitchen-style). @@ -375,6 +418,7 @@ export async function upsertDirectoryListingFromTap(input: { const sourceUrl = record.externalUrl.trim(); const atstoreDidRaw = process.env.ATSTORE_REPO_DID?.trim() ?? null; const atstoreDid = atstoreDidRaw?.startsWith("did:") ? atstoreDidRaw : null; + const slug = await resolveTapIngestSlug(db, atUri, record.slug); const [existingRow] = await db .select({ @@ -385,7 +429,7 @@ export async function upsertDirectoryListingFromTap(input: { verificationStatus: schema.storeListings.verificationStatus, }) .from(schema.storeListings) - .where(eq(schema.storeListings.slug, record.slug)) + .where(eq(schema.storeListings.slug, slug)) .limit(1); const existingClaimPendingForDid = existingRow?.claimPendingForDid?.trim() ?? null; @@ -434,7 +478,7 @@ export async function upsertDirectoryListingFromTap(input: { const heroCid = getBlobCidString(record.heroImage); const shotsParsed = (record.screenshots ?? []).length; console.log( - `[tap-ingest] slug=${record.slug} rkey=${rkey} images: icon=${iconUrl ? `cdn(${iconCid?.slice(0, 12)}…)` : `MISS:${explainMissingBlobUrl(record.icon)}`} hero=${heroImageUrl ? `cdn(${heroCid?.slice(0, 12)}…)` : `MISS:${explainMissingBlobUrl(record.heroImage)}`} screenshots=${screenshotUrls.length}/${shotsParsed}`, + `[tap-ingest] slug=${slug} rkey=${rkey} images: icon=${iconUrl ? `cdn(${iconCid?.slice(0, 12)}…)` : `MISS:${explainMissingBlobUrl(record.icon)}`} hero=${heroImageUrl ? `cdn(${heroCid?.slice(0, 12)}…)` : `MISS:${explainMissingBlobUrl(record.heroImage)}`} screenshots=${screenshotUrls.length}/${shotsParsed}`, ); const productDid = record.productAccountDid?.trim() ?? null; @@ -454,7 +498,7 @@ export async function upsertDirectoryListingFromTap(input: { .values({ sourceUrl, name: record.name, - slug: record.slug, + slug, externalUrl: record.externalUrl, iconUrl, screenshotUrls, @@ -522,7 +566,7 @@ export async function upsertDirectoryListingFromTap(input: { .where(eq(schema.storeListings.sourceUrl, sourceUrl)) .limit(1); console.warn( - `[tap-ingest] skip listing ${atUri} slug=${record.slug} — sourceUrl=${sourceUrl} already owned by ${conflictRow?.atUri ?? "(unknown row)"} slug=${conflictRow?.slug ?? "?"} repoDid=${conflictRow?.repoDid ?? "?"}`, + `[tap-ingest] skip listing ${atUri} slug=${slug} — sourceUrl=${sourceUrl} already owned by ${conflictRow?.atUri ?? "(unknown row)"} slug=${conflictRow?.slug ?? "?"} repoDid=${conflictRow?.repoDid ?? "?"}`, ); return; } @@ -532,7 +576,7 @@ export async function upsertDirectoryListingFromTap(input: { const [saved] = await db .select({ id: schema.storeListings.id }) .from(schema.storeListings) - .where(eq(schema.storeListings.slug, record.slug)) + .where(eq(schema.storeListings.slug, slug)) .limit(1); if (saved?.id) { @@ -540,7 +584,7 @@ export async function upsertDirectoryListingFromTap(input: { await refreshListingPageSnapshot(db, saved.id); } catch (refreshError) { console.warn( - `[tap-ingest] page snapshot refresh failed slug=${record.slug} id=${saved.id}`, + `[tap-ingest] page snapshot refresh failed slug=${slug} id=${saved.id}`, refreshError, ); }