fix: WET processor OOM — process records inline, increase memory to 2Gi

Node.js heap exhausted at 512MB buffering 21K WARC records.
Fix: process each record immediately instead of accumulating in
pendingRecords array. Also cap per-record content length and
increase Cloud Run Job memory from 1Gi to 2Gi with --max-old-space-size=1536.

Co-Authored-By: claude-flow <ruv@ruv.net>

Author: ruvnet

scripts/deploy-wet-job.sh

@@ -62,7 +62,7 @@ echo "Crawl index: $CRAWL_INDEX"
 
 curl -sL "https://data.commoncrawl.org/$WET_PATH" \
   | gunzip \
-  | node /app/filter.js \
+  | node --max-old-space-size=1536 /app/filter.js \
     --brain-url "$BRAIN_URL" \
     --auth "$AUTH_HEADER" \
     --batch-size "$BATCH_SIZE" \
@@ -92,7 +92,7 @@ gcloud run jobs deploy "$JOB_NAME" \
   --parallelism=10 \
   --max-retries=1 \
   --cpu=1 \
-  --memory=1Gi \
+  --memory=2Gi \
   --task-timeout=3600s \
   --env-vars-file="$BUILD_DIR/env.yaml" \
   2>&1

scripts/wet-filter-inject.js

@@ -187,12 +187,16 @@ const rl = readline.createInterface({ input: process.stdin, crlfDelay: Infinity
 let recordUrl = '';
 let recordContent = '';
 let inRecord = false;
-const pendingRecords = [];
+// Process records inline to avoid OOM — never buffer all records
+let processQueue = Promise.resolve();
 
 rl.on('line', (line) => {
   if (line.startsWith('WARC/1.0')) {
     if (recordUrl && recordContent) {
-      pendingRecords.push({ url: recordUrl, content: recordContent });
+      // Process immediately, don't accumulate
+      const url = recordUrl;
+      const content = recordContent;
+      processQueue = processQueue.then(() => processRecord(url, content));
     }
     recordUrl = '';
     recordContent = '';
@@ -202,19 +206,20 @@ rl.on('line', (line) => {
   } else if (line.startsWith('Content-Length:')) {
     inRecord = true;
   } else if (inRecord) {
-    recordContent += line + '\n';
+    // Limit content accumulation per record to prevent single-record bloat
+    if (recordContent.length < MAX_CONTENT_LENGTH * 2) {
+      recordContent += line + '\n';
+    }
   }
 });
 
 rl.on('close', async () => {
   // Process last record
   if (recordUrl && recordContent) {
-    pendingRecords.push({ url: recordUrl, content: recordContent });
-  }
-
-  // Process all records sequentially
-  for (const rec of pendingRecords) {
-    await processRecord(rec.url, rec.content);
+    await processQueue;
+    await processRecord(recordUrl, recordContent);
+  } else {
+    await processQueue;
   }
 
   // Flush remaining batch