fix(batch): handle locked ReadableStream when retrying batch trigger

ericallam · ericallam · commit fa43e8e05d4e · 2026-01-20T15:23:41.000Z
When fetch crashes mid-stream during batch item upload (e.g., connection
reset, timeout), the request stream may remain locked by fetch's internal
reader. Attempting to cancel a locked stream throws 'Invalid state:
ReadableStream is locked', causing the batch operation to fail.

Added safeStreamCancel() helper that gracefully handles locked streams by
catching and ignoring the locked error. The stream will be cleaned up by
garbage collection when fetch eventually releases the reader.

Fixes customer issue where batchTrigger failed with ReadableStream locked
error during network instability.
diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts
@@ -444,14 +444,14 @@ export class ApiClient {
 
         if (retryResult.retry) {
           // Cancel the request stream before retry to prevent tee() from buffering
-          await forRequest.cancel();
+          await safeStreamCancel(forRequest);
           await sleep(retryResult.delay);
           // Use the backup stream for retry
           return this.#streamBatchItemsWithRetry(batchId, forRetry, retryOptions, attempt + 1);
         }
 
         // Not retrying - cancel the backup stream
-        await forRetry.cancel();
+        await safeStreamCancel(forRetry);
 
         const errText = await response.text().catch((e) => (e as Error).message);
         let errJSON: Object | undefined;
@@ -471,7 +471,7 @@ export class ApiClient {
 
       if (!parsed.success) {
         // Cancel backup stream since we're throwing
-        await forRetry.cancel();
+        await safeStreamCancel(forRetry);
         throw new Error(
           `Invalid response from server for batch ${batchId}: ${parsed.error.message}`
         );
@@ -484,14 +484,14 @@ export class ApiClient {
 
         if (delay) {
           // Cancel the request stream before retry to prevent tee() from buffering
-          await forRequest.cancel();
+          await safeStreamCancel(forRequest);
           // Retry with the backup stream
           await sleep(delay);
           return this.#streamBatchItemsWithRetry(batchId, forRetry, retryOptions, attempt + 1);
         }
 
         // No more retries - cancel backup stream and throw descriptive error
-        await forRetry.cancel();
+        await safeStreamCancel(forRetry);
         throw new BatchNotSealedError({
           batchId,
           enqueuedCount: parsed.data.enqueuedCount ?? 0,
@@ -502,7 +502,7 @@ export class ApiClient {
       }
 
       // Success - cancel the backup stream to release resources
-      await forRetry.cancel();
+      await safeStreamCancel(forRetry);
 
       return parsed.data;
     } catch (error) {
@@ -519,13 +519,13 @@ export class ApiClient {
       const delay = calculateNextRetryDelay(retryOptions, attempt);
       if (delay) {
         // Cancel the request stream before retry to prevent tee() from buffering
-        await forRequest.cancel();
+        await safeStreamCancel(forRequest);
         await sleep(delay);
         return this.#streamBatchItemsWithRetry(batchId, forRetry, retryOptions, attempt + 1);
       }
 
       // No more retries - cancel the backup stream
-      await forRetry.cancel();
+      await safeStreamCancel(forRetry);
 
       // Wrap in a more descriptive error
       const cause = error instanceof Error ? error : new Error(String(error));
@@ -1731,6 +1731,30 @@ function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+/**
+ * Safely cancels a ReadableStream, handling the case where it might be locked.
+ *
+ * When fetch uses a ReadableStream as a request body and an error occurs mid-transfer
+ * (connection reset, timeout, etc.), the stream may remain locked by fetch's internal reader.
+ * Attempting to cancel a locked stream throws "Invalid state: ReadableStream is locked".
+ *
+ * This function gracefully handles that case by catching the error and doing nothing -
+ * the stream will be cleaned up by garbage collection when the reader is released.
+ */
+async function safeStreamCancel(stream: ReadableStream<unknown>): Promise<void> {
+  try {
+    await stream.cancel();
+  } catch (error) {
+    // Ignore "locked" errors - the stream will be cleaned up when the reader is released.
+    // This happens when fetch crashes mid-read and doesn't release the reader lock.
+    if (error instanceof TypeError && String(error).includes("locked")) {
+      return;
+    }
+    // Re-throw unexpected errors
+    throw error;
+  }
+}
+
 // ============================================================================
 // NDJSON Stream Helpers
 // ============================================================================
diff --git a/packages/core/src/v3/apiClient/streamBatchItems.test.ts b/packages/core/src/v3/apiClient/streamBatchItems.test.ts
@@ -398,6 +398,58 @@ describe("streamBatchItems stream cancellation on retry", () => {
     expect(cancelCallCount).toBeGreaterThanOrEqual(1);
   });
 
+  it("handles locked stream when connection error occurs mid-read", async () => {
+    // This test simulates the real-world scenario where fetch throws an error
+    // while still holding the reader lock on the request body stream.
+    // This can happen with connection resets, timeouts, or network failures.
+    let callIndex = 0;
+
+    const mockFetch = vi.fn().mockImplementation(async (_url: string, init?: RequestInit) => {
+      const currentAttempt = callIndex;
+      callIndex++;
+
+      if (init?.body && init.body instanceof ReadableStream) {
+        if (currentAttempt === 0) {
+          // First attempt: Get a reader and start reading, but throw while still holding the lock.
+          // This simulates a connection error that happens mid-transfer.
+          const reader = init.body.getReader();
+          await reader.read(); // Start reading
+          // DON'T release the lock - this simulates fetch crashing mid-read
+          throw new TypeError("Connection reset by peer");
+        }
+
+        // Subsequent attempts: consume and release normally
+        await consumeAndRelease(init.body);
+      }
+
+      // Second attempt: success
+      return {
+        ok: true,
+        json: () =>
+          Promise.resolve({
+            id: "batch_test123",
+            itemsAccepted: 10,
+            itemsDeduplicated: 0,
+            sealed: true,
+          }),
+      };
+    });
+    globalThis.fetch = mockFetch;
+
+    const client = new ApiClient("http://localhost:3030", "tr_test_key");
+
+    // This should NOT throw "ReadableStream is locked" error
+    // Instead it should gracefully handle the locked stream and retry
+    const result = await client.streamBatchItems(
+      "batch_test123",
+      [{ index: 0, task: "test-task", payload: "{}" }],
+      { retry: { maxAttempts: 3, minTimeoutInMs: 10, maxTimeoutInMs: 50 } }
+    );
+
+    expect(result.sealed).toBe(true);
+    expect(mockFetch).toHaveBeenCalledTimes(2);
+  });
+
   it("does not leak memory by leaving tee branches unconsumed during multiple retries", async () => {
     let cancelCallCount = 0;
     let callIndex = 0;
diff --git a/scripts/batch-concurrency-cleaner.sh b/scripts/batch-concurrency-cleaner.sh
@@ -0,0 +1,180 @@
+#!/bin/bash
+#
+# Batch Queue Concurrency Cleaner
+#
+# Detects and cleans up stale concurrency entries that block batch processing.
+# This is a workaround for a bug where visibility timeout reclaims don't release concurrency.
+#
+# Uses a Lua script for ATOMIC detection of stale entries - no race conditions.
+#
+# Usage:
+#   ./batch-concurrency-cleaner.sh --read-redis <url> --write-redis <url> [--delay <seconds>] [--dry-run]
+#
+
+set -e
+
+# Defaults
+DELAY=10
+DRY_RUN=false
+READ_REDIS=""
+WRITE_REDIS=""
+
+# Parse arguments
+while [[ $# -gt 0 ]]; do
+  case $1 in
+    --read-redis)
+      READ_REDIS="$2"
+      shift 2
+      ;;
+    --write-redis)
+      WRITE_REDIS="$2"
+      shift 2
+      ;;
+    --delay)
+      DELAY="$2"
+      shift 2
+      ;;
+    --dry-run)
+      DRY_RUN=true
+      shift
+      ;;
+    -h|--help)
+      echo "Usage: $0 --read-redis <url> --write-redis <url> [--delay <seconds>] [--dry-run]"
+      exit 0
+      ;;
+    *)
+      echo "Unknown option: $1"
+      exit 1
+      ;;
+  esac
+done
+
+if [[ -z "$READ_REDIS" ]] || [[ -z "$WRITE_REDIS" ]]; then
+  echo "Error: --read-redis and --write-redis are required"
+  exit 1
+fi
+
+echo "Batch Queue Concurrency Cleaner (Atomic Version)"
+echo "================================================="
+echo "Read Redis:  ${READ_REDIS:0:30}..."
+echo "Write Redis: ${WRITE_REDIS:0:30}..."
+echo "Delay:       ${DELAY}s"
+echo "Dry run:     $DRY_RUN"
+echo ""
+
+rcli_read() {
+  redis-cli -u "$READ_REDIS" --no-auth-warning "$@" 2>/dev/null
+}
+
+rcli_write() {
+  redis-cli -u "$WRITE_REDIS" --no-auth-warning "$@" 2>/dev/null
+}
+
+# Lua script that ATOMICALLY checks for stale concurrency entries
+# KEYS[1] = concurrency key to check
+# KEYS[2-13] = in-flight data hash keys for shards 0-11
+# Returns: list of stale messageIds (not in any in-flight hash)
+FIND_STALE_LUA='
+local concurrency_key = KEYS[1]
+local stale = {}
+
+-- Get all members of the concurrency set
+local members = redis.call("SMEMBERS", concurrency_key)
+
+for _, msg_id in ipairs(members) do
+  local found = false
+  -- Check each in-flight shard (KEYS[2] through KEYS[13])
+  for i = 2, 13 do
+    if redis.call("HEXISTS", KEYS[i], msg_id) == 1 then
+      found = true
+      break
+    end
+  end
+  if not found then
+    table.insert(stale, msg_id)
+  end
+end
+
+return stale
+'
+
+# Build the in-flight keys array (used in every Lua call)
+INFLIGHT_KEYS="engine:batch:inflight:0:data"
+for shard in 1 2 3 4 5 6 7 8 9 10 11; do
+  INFLIGHT_KEYS="$INFLIGHT_KEYS engine:batch:inflight:$shard:data"
+done
+
+# Main loop
+while true; do
+  ts=$(date '+%H:%M:%S')
+
+  # Get master queue total and in-flight count for status display
+  master_total=0
+  for i in 0 1 2 3 4 5 6 7 8 9 10 11; do
+    count=$(rcli_read ZCARD "engine:batch:master:$i")
+    master_total=$((master_total + count))
+  done
+
+  inflight_total=0
+  for i in 0 1 2 3 4 5 6 7 8 9 10 11; do
+    count=$(rcli_read HLEN "engine:batch:inflight:$i:data")
+    inflight_total=$((inflight_total + count))
+  done
+
+  # Scan for concurrency keys
+  cursor=0
+  total_stale=0
+  cleaned_tenants=0
+
+  while true; do
+    scan_output=$(rcli_read SCAN $cursor MATCH 'engine:batch:concurrency:tenant:*' COUNT 1000)
+    cursor=$(echo "$scan_output" | head -1)
+    keys=$(echo "$scan_output" | tail -n +2)
+
+    while IFS= read -r conc_key; do
+      [[ -z "$conc_key" ]] && continue
+
+      # ATOMIC check: Run Lua script to find stale entries
+      # 13 keys total: 1 concurrency key + 12 in-flight keys
+      stale_ids=$(rcli_read EVAL "$FIND_STALE_LUA" 13 "$conc_key" $INFLIGHT_KEYS)
+
+      # Count stale entries
+      stale_count=0
+      stale_array=()
+      while IFS= read -r stale_id; do
+        [[ -z "$stale_id" ]] && continue
+        stale_array+=("$stale_id")
+        stale_count=$((stale_count + 1))
+      done <<< "$stale_ids"
+
+      if [[ $stale_count -gt 0 ]]; then
+        tenant="${conc_key#engine:batch:concurrency:tenant:}"
+        total_stale=$((total_stale + stale_count))
+
+        if [[ "$DRY_RUN" == "true" ]]; then
+          echo "[$ts] STALE (dry-run): $tenant ($stale_count entries)"
+          for sid in "${stale_array[@]}"; do
+            echo "       - $sid"
+          done
+        else
+          # Remove each stale entry individually with SREM (idempotent, safe)
+          for sid in "${stale_array[@]}"; do
+            rcli_write SREM "$conc_key" "$sid" >/dev/null
+          done
+          echo "[$ts] CLEANED: $tenant ($stale_count stale entries removed)"
+          cleaned_tenants=$((cleaned_tenants + 1))
+        fi
+      fi
+    done <<< "$keys"
+
+    [[ "$cursor" == "0" ]] && break
+  done
+
+  if [[ "$DRY_RUN" == "true" ]]; then
+    echo "[$ts] in-flight=$inflight_total master-queue=$master_total stale-found=$total_stale"
+  else
+    echo "[$ts] in-flight=$inflight_total master-queue=$master_total cleaned=$cleaned_tenants"
+  fi
+
+  sleep "$DELAY"
+done
