fix(insights): chunk rrweb blob fetch + rate-limit handling

Live verification surfaced two issues in archive-recordings:
- blob_v2 rejects wide blob-key ranges (a 23-key span 400s; ~5 is fine).
  Fetch in contiguous chunks of BLOB_CHUNK (default 10) and concatenate.
- PostHog's snapshot API is aggressively rate-limited; back-to-back
  sessions tripped 429. Pace requests (per-session + per-chunk delays)
  and stop the run on sustained 429 — unarchived rows keep
  rrwebArchivedAt=null and retry next cycle (oldest-first ordering
  protects soon-to-be-deleted recordings first). Result now carries
  rateLimited.

Verified against prod: 87- and 23-blob recordings reconstruct fully
(no 400), failed=0, graceful 429 stop.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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<RecordingArchiveResult> {
-  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<RecordingArchiveResult> {
     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<RecordingArchiveResult> {
         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<string, string[]>();
       for (const src of sources) {
         if (!bySource.has(src.source)) bySource.set(src.source, []);
@@ -51,9 +67,13 @@ export async function runArchiveRecordings(): Promise<RecordingArchiveResult> {
       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<RecordingArchiveResult> {
       });
       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 {

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;