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

[ericallam]

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.

[changeset-bot]

🦋 Changeset detected

Latest commit: d6d933d

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 27 packages

Name	Type
@trigger.dev/core	Patch
@trigger.dev/build	Patch
trigger.dev	Patch
@trigger.dev/python	Patch
@trigger.dev/redis-worker	Patch
@trigger.dev/schema-to-json	Patch
@trigger.dev/sdk	Patch
@internal/cache	Patch
@internal/clickhouse	Patch
@internal/redis	Patch
@internal/replication	Patch
@internal/run-engine	Patch
@internal/schedule-engine	Patch
@internal/testcontainers	Patch
@internal/tracing	Patch
@internal/tsql	Patch
@internal/zod-worker	Patch
d3-chat	Patch
references-d3-openai-agents	Patch
references-nextjs-realtime	Patch
references-realtime-hooks-test	Patch
references-realtime-streams	Patch
references-telemetry	Patch
@trigger.dev/react-hooks	Patch
@trigger.dev/rsc	Patch
@trigger.dev/database	Patch
@trigger.dev/otlp-importer	Patch

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[coderabbitai]

Walkthrough

This pull request introduces a fix for a batch trigger failure in the @trigger.dev/core package. The changes add a new safeStreamCancel utility function that wraps direct stream cancellation calls in a try/catch block to handle locked ReadableStream scenarios. The function gracefully swallows TypeError exceptions when a stream is already locked (deferring cleanup to garbage collection) while rethrowing other errors. Multiple occurrences of direct cancel() calls on streaming batch item paths are replaced with calls to this new utility, and a test case is added to validate the behavior when a connection error occurs while a stream lock is held.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

Detailed Analysis

Scope of Changes:

• A changelog entry documenting the fix
• Addition of a new utility function (safeStreamCancel) in the API client module
• Replacement of direct stream cancellation calls throughout the batch items handling logic
• A new test case validating locked stream handling during connection errors and retries

Key Considerations for Review:

• The semantics of the new error handling approach (swallowing TypeError for locked streams while rethrowing other errors)
• Verification that all direct cancel() calls have been appropriately replaced
• Test coverage adequacy for the locked stream scenario
• Understanding of ReadableStream locking behavior and its implications for garbage collection

🚥 Pre-merge checks | ✅ 1 | ❌ 2

❌ Failed checks (2 warnings)

Check name	Status	Explanation	Resolution
Description check	⚠️ Warning	The description explains the problem, the solution, and references a customer issue, but does not follow the provided template structure with required sections like Closes #, Checklist, Testing, Changelog, and Screenshots.	Update the description to follow the template: add issue reference, checklist items, testing steps, and structured changelog/screenshots sections.
Docstring Coverage	⚠️ Warning	Docstring coverage is 50.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (1 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and specifically describes the main fix: handling locked ReadableStream errors when retrying batch triggers, which directly matches the changeset's primary objective.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

---

📜 Recent review details

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between e770b4b and d6d933d.

📒 Files selected for processing (3)

• .changeset/gentle-streams-flow.md
• packages/core/src/v3/apiClient/index.ts
• packages/core/src/v3/apiClient/streamBatchItems.test.ts

✅ Files skipped from review due to trivial changes (1)

• .changeset/gentle-streams-flow.md

🚧 Files skipped from review as they are similar to previous changes (1)

• packages/core/src/v3/apiClient/index.ts

🧰 Additional context used

📓 Path-based instructions (9)

**/*.{ts,tsx}

📄 CodeRabbit inference engine (.github/copilot-instructions.md)

**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead

**/*.{ts,tsx}: Always import tasks from @trigger.dev/sdk, never use @trigger.dev/sdk/v3 or deprecated client.defineJob pattern
Every Trigger.dev task must be exported and have a unique id property with no timeouts in the run function

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

{packages/core,apps/webapp}/**/*.{ts,tsx}

📄 CodeRabbit inference engine (.github/copilot-instructions.md)

Use zod for validation in packages/core and apps/webapp

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

**/*.{ts,tsx,js,jsx}

📄 CodeRabbit inference engine (.github/copilot-instructions.md)

Use function declarations instead of default exports

Import from @trigger.dev/core using subpaths only, never import from root

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

**/*.{test,spec}.{ts,tsx}

📄 CodeRabbit inference engine (.github/copilot-instructions.md)

Use vitest for all tests in the Trigger.dev repository

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

**/*.ts

📄 CodeRabbit inference engine (.cursor/rules/otel-metrics.mdc)

**/*.ts: When creating or editing OTEL metrics (counters, histograms, gauges), ensure metric attributes have low cardinality by using only enums, booleans, bounded error codes, or bounded shard IDs
Do not use high-cardinality attributes in OTEL metrics such as UUIDs/IDs (envId, userId, runId, projectId, organizationId), unbounded integers (itemCount, batchSize, retryCount), timestamps (createdAt, startTime), or free-form strings (errorMessage, taskName, queueName)
When exporting OTEL metrics via OTLP to Prometheus, be aware that the exporter automatically adds unit suffixes to metric names (e.g., 'my_duration_ms' becomes 'my_duration_ms_milliseconds', 'my_counter' becomes 'my_counter_total'). Account for these transformations when writing Grafana dashboards or Prometheus queries

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

**/*.{js,ts,jsx,tsx,json,md,yaml,yml}

📄 CodeRabbit inference engine (AGENTS.md)

Format code using Prettier before committing

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

**/*.test.{ts,tsx,js,jsx}

📄 CodeRabbit inference engine (AGENTS.md)

**/*.test.{ts,tsx,js,jsx}: Test files should live beside the files under test and use descriptive describe and it blocks
Tests should avoid mocks or stubs and use the helpers from @internal/testcontainers when Redis or Postgres are needed
Use vitest for running unit tests

**/*.test.{ts,tsx,js,jsx}: Use vitest exclusively for testing and never mock anything - use testcontainers instead
Place test files next to source files with naming pattern: source file (e.g., MyService.ts) → MyService.test.ts

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

**/*.test.{ts,tsx}

📄 CodeRabbit inference engine (CLAUDE.md)

Use testcontainers helpers (redisTest, postgresTest, containerTest) from @internal/testcontainers for Redis/PostgreSQL testing instead of mocks

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

{packages,integrations}/**/*

📄 CodeRabbit inference engine (CLAUDE.md)

Add a changeset when modifying any public package in packages/* or integrations/* using pnpm run changeset:add

Files:

• packages/core/src/v3/apiClient/streamBatchItems.test.ts

🧠 Learnings (1)

📓 Common learnings

Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `batch.triggerAndWait()` to batch trigger multiple different tasks and wait for results

Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .github/copilot-instructions.md:0-0
Timestamp: 2025-11-27T16:26:37.432Z
Learning: Applies to packages/trigger-sdk/**/*.{ts,tsx} : In the Trigger.dev SDK (packages/trigger-sdk), prefer isomorphic code like fetch and ReadableStream instead of Node.js-specific code

Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `runs.subscribeToBatch()` to subscribe to changes for all runs in a batch

Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `.withStreams()` to subscribe to realtime streams from task metadata in addition to run changes

Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `batch.triggerByTaskAndWait()` to batch trigger tasks by passing task instances and wait for results

Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `batch.trigger()` to trigger multiple different tasks at once from backend code

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (23)

• GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
• GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
• GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
• GitHub Check: typecheck / typecheck
• GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
• GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
• GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
• GitHub Check: Analyze (javascript-typescript)

🔇 Additional comments (1)

packages/core/src/v3/apiClient/streamBatchItems.test.ts (1)

401-451: Well-structured test for the locked stream scenario.

The test correctly simulates the real-world failure mode where fetch crashes mid-read while holding the reader lock. The intentional omission of reader.releaseLock() on line 417-418 accurately reproduces the problematic state.

One minor suggestion: consider adding an explicit assertion that no "ReadableStream is locked" error propagates, which would make the test's intent even clearer:

💡 Optional: Add explicit error message check

     // This should NOT throw "ReadableStream is locked" error
     // Instead it should gracefully handle the locked stream and retry
-    const result = await client.streamBatchItems(
+    let caughtError: Error | undefined;
+    const result = await client.streamBatchItems(
       "batch_test123",
       [{ index: 0, task: "test-task", payload: "{}" }],
       { retry: { maxAttempts: 3, minTimeoutInMs: 10, maxTimeoutInMs: 50 } }
-    );
+    ).catch((e) => {
+      caughtError = e;
+      throw e;
+    });

+    // Ensure we didn't get the locked stream error
+    expect(caughtError).toBeUndefined();
     expect(result.sealed).toBe(true);
     expect(mockFetch).toHaveBeenCalledTimes(2);

_{✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.}

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[vibe-kanban-cloud]

Review Complete

Your review story is ready!

View Story

Comment !reviewfast on this PR to re-generate the story.