diff --git a/apps/worker/src/jobs/archive-recordings.ts b/apps/worker/src/jobs/archive-recordings.ts index 183b141..d8b7f0a 100644 --- a/apps/worker/src/jobs/archive-recordings.ts +++ b/apps/worker/src/jobs/archive-recordings.ts @@ -7,18 +7,30 @@ import { putBuffer } from "../lib/minio"; // ALL recordings (incl. discarded ones) at full fidelity — unlike compress, which // only handles scored sessions and caps blobs for token budget. const RRWEB_BUCKET = process.env.RRWEB_ARCHIVE_BUCKET ?? "rrweb-archive"; -const BATCH = Number(process.env.INSIGHT_RECORDING_ARCHIVE_BATCH ?? "100"); +const BATCH = Number(process.env.INSIGHT_RECORDING_ARCHIVE_BATCH ?? "60"); const MAX_AGE_DAYS = Number(process.env.INSIGHT_RECORDING_ARCHIVE_MAX_AGE_DAYS ?? "25"); +// PostHog's snapshot API is aggressively rate-limited. Pace requests and bail out +// of the run on sustained 429s — unarchived rows keep rrwebArchivedAt=null and are +// retried next cycle (the 25d window vs */6h cadence gives ample slack). +const RATE_DELAY_MS = Number(process.env.INSIGHT_RECORDING_ARCHIVE_DELAY_MS ?? "500"); +// blob_v2 rejects wide blob-key ranges (a 23-key span 400s, a 5-key span is fine), +// so fetch in small contiguous chunks and concatenate. +const BLOB_CHUNK = Number(process.env.INSIGHT_RECORDING_BLOB_CHUNK ?? "10"); +const CHUNK_DELAY_MS = Number(process.env.INSIGHT_RECORDING_CHUNK_DELAY_MS ?? "150"); export type RecordingArchiveResult = { scanned: number; archived: number; failed: number; skipped: number; + rateLimited: boolean; }; +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +const isRateLimit = (e: unknown) => (e as Error)?.message?.includes(" 429"); + export async function runArchiveRecordings(): Promise { - if (!isConfigured()) return { scanned: 0, archived: 0, failed: 0, skipped: 0 }; + if (!isConfigured()) return { scanned: 0, archived: 0, failed: 0, skipped: 0, rateLimited: false }; const floor = new Date(Date.now() - MAX_AGE_DAYS * 24 * 3600_000); const pending = await prisma.sessionMeta.findMany({ @@ -26,13 +38,17 @@ export async function runArchiveRecordings(): Promise { orderBy: { startedAt: "asc" }, take: BATCH, }); - if (pending.length === 0) return { scanned: 0, archived: 0, failed: 0, skipped: 0 }; + if (pending.length === 0) return { scanned: 0, archived: 0, failed: 0, skipped: 0, rateLimited: false }; let archived = 0; let failed = 0; let skipped = 0; + let rateLimited = false; + let processed = 0; for (const s of pending) { + if (processed > 0) await sleep(RATE_DELAY_MS); + processed++; try { const sources = await getSnapshotSources(s.id); if (sources.length === 0) { @@ -42,7 +58,7 @@ export async function runArchiveRecordings(): Promise { continue; } - // Fetch the FULL blob range per source (no 12-cap) and concatenate raw JSONL. + // Fetch ALL blobs per source (no 12-cap), chunked to dodge blob_v2's range limit. const bySource = new Map(); for (const src of sources) { if (!bySource.has(src.source)) bySource.set(src.source, []); @@ -51,9 +67,13 @@ export async function runArchiveRecordings(): Promise { const parts: string[] = []; for (const [source, keys] of bySource) { const sorted = [...keys].sort((a, b) => Number(a) - Number(b)); - if (!sorted.length) continue; - const blob = await getSnapshotBlob(s.id, source, sorted[0], sorted[sorted.length - 1]); - if (blob) parts.push(blob); + for (let i = 0; i < sorted.length; i += BLOB_CHUNK) { + if (i > 0) await sleep(CHUNK_DELAY_MS); + const lo = sorted[i]; + const hi = sorted[Math.min(i + BLOB_CHUNK - 1, sorted.length - 1)]; + const blob = await getSnapshotBlob(s.id, source, lo, hi); + if (blob) parts.push(blob); + } } const raw = parts.join("\n").trim(); if (!raw) { @@ -72,13 +92,18 @@ export async function runArchiveRecordings(): Promise { }); archived++; } catch (e) { - // Transient — leave rrwebArchivedAt null so it retries next run. + if (isRateLimit(e)) { + // Rate limited — stop hammering. Remaining rows retry next cycle. + rateLimited = true; + break; + } + // Other transient errors — leave rrwebArchivedAt null so it retries next run. console.warn("[archive-recordings] failed", s.id, (e as Error).message); failed++; } } - return { scanned: pending.length, archived, failed, skipped }; + return { scanned: pending.length, archived, failed, skipped, rateLimited }; } function dateFolder(d: Date): string { diff --git a/apps/worker/src/schedulers/pipeline.ts b/apps/worker/src/schedulers/pipeline.ts index 2d8b154..2f808ca 100644 --- a/apps/worker/src/schedulers/pipeline.ts +++ b/apps/worker/src/schedulers/pipeline.ts @@ -102,7 +102,7 @@ async function runJob(job: Job) { const res = await runArchiveRecordings(); if (res.scanned > 0) { console.log( - `[pipeline] archive-recordings scanned=${res.scanned} archived=${res.archived} skipped=${res.skipped} failed=${res.failed}`, + `[pipeline] archive-recordings scanned=${res.scanned} archived=${res.archived} skipped=${res.skipped} failed=${res.failed}${res.rateLimited ? " rate_limited" : ""}`, ); } return res;