Skip to content

Fix Mentions Watcher OOM and Tap Sync at_uri conflicts #56

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hipstersmoothie wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fix Mentions Watcher OOM and Tap Sync at_uri conflicts #56

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 80 additions & 31 deletions scripts/jetstream-consumer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() ||
Expand Down Expand Up @@ -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", {
Expand All @@ -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<string> = [];
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) => {
Expand Down Expand Up @@ -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);
Expand Down
56 changes: 50 additions & 6 deletions src/lib/atproto/tap-listing-sync.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<string> {
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).
Expand All @@ -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({
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -454,7 +498,7 @@ export async function upsertDirectoryListingFromTap(input: {
.values({
sourceUrl,
name: record.name,
slug: record.slug,
slug,
externalUrl: record.externalUrl,
iconUrl,
screenshotUrls,
Expand Down Expand Up @@ -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;
}
Expand All @@ -532,15 +576,15 @@ 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) {
try {
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,
);
}
Expand Down
Loading