Multiple streams can now be consumed at the same time #1522
Conversation
🦋 Changeset detectedLatest commit: 52b7627 The changes in this PR will be included in the next version bump. This PR includes changesets to release 11 packages
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 |
WalkthroughThe pull request introduces significant enhancements to the Changes
Possibly related PRs
Warning There were issues while running some tools. Please review the errors and either fix the tool’s configuration or disable the tool if it’s a critical failure. 🔧 eslint
packages/core/src/v3/apiClient/runStream.tsOops! Something went wrong! :( ESLint: 8.45.0 ESLint couldn't find the config "custom" to extend from. Please check that the name of the config is correct. The config "custom" was referenced from the config file in "/.eslintrc.js". If you still have problems, please stop by https://eslint.org/chat/help to chat with the team. Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Outside diff range and nitpick comments (4)
packages/core/src/v3/apiClient/runStream.ts (1)
253-277: Handle Errors in Stream Processing GracefullyIn the
withStreamsmethod, errors caught in the stream pipeline are currently only logged to the console. Consider enhancing this by propagating the errors to upstream consumers or adding retry logic if appropriate to ensure robust error handling.For example, you can modify the error handling as follows:
.catch((error) => { - console.error(`Error in stream ${streamKey}:`, error); + controller.error(new Error(`Error in stream ${streamKey}: ${error.message}`)); });This change will signal the error to the consumer of the stream rather than just logging it.
packages/core/test/runStream.test.ts (1)
16-24: Simplify ReadableStream Creation in Test SubscriptionConsider simplifying the
subscribemethod by using theReadableStreamconstructor's default behavior or by utilizingasyncgenerators for better readability.Here's an example using an async generator:
async subscribe(): Promise<ReadableStream<unknown>> { - return new ReadableStream({ - start: async (controller) => { - for (const chunk of this.chunks) { - controller.enqueue(chunk); - } - controller.close(); - }, - }); + const generator = async function* () { + for (const chunk of this.chunks) { + yield chunk; + } + }.bind(this); + return new ReadableStream({ + async pull(controller) { + const { value, done } = await generator().next(); + if (done) { + controller.close(); + } else { + controller.enqueue(value); + } + }, + }); }.changeset/modern-nails-refuse.md (1)
5-5: Enhance Change Description for ClarityConsider providing a more detailed description of the changes in the changeset to help users understand the impact and benefits of the update.
For example:
Multiple streams can now be consumed simultaneously. This enhancement allows for concurrent processing of data streams, improving performance and efficiency when handling multiple asynchronous tasks.scripts/publish-prerelease.sh (1)
41-49: Improve Error Handling for Changeset Version CommandThe current script may not handle all failure cases of the
changeset versioncommand. If the command fails for reasons other than "No unreleased changesets found," the script might not exit appropriately.Consider updating the error handling to accurately capture and handle different failure scenarios:
echo "Running: pnpm exec changeset version --snapshot $version" -if output=$(pnpm exec changeset version --snapshot $version 2>&1); then +if output=$(pnpm exec changeset version --snapshot $version 2>&1); then + if echo "$output" | grep -q "No unreleased changesets found"; then + echo "No unreleased changesets found. Exiting." + exit 0 + fi +elif echo "$output" | grep -q "Some specific known error"; then + echo "Known error occurred: $output" + exit 1 else echo "Error running changeset version command" + echo "$output" exit 1 fiThis adjustment ensures that unexpected errors cause the script to exit and provides error output for debugging.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (4)
.changeset/modern-nails-refuse.md(1 hunks)packages/core/src/v3/apiClient/runStream.ts(3 hunks)packages/core/test/runStream.test.ts(1 hunks)scripts/publish-prerelease.sh(1 hunks)
🔇 Additional comments (2)
packages/core/src/v3/apiClient/runStream.ts (1)
100-100: Update Implementations and Usages of subscribe Method
The subscribe method signature in the StreamSubscription interface has changed to return a Promise<ReadableStream<unknown>>. Ensure that all implementations and usages of this method throughout the codebase are updated to match the new signature to prevent any runtime errors.
Run the following script to find all implementations and usages of the subscribe method:
✅ Verification successful
Let's also search for any class that might implement this interface with a different pattern:
Based on the search results, I can now provide a final response:
The subscribe Method Implementation is Already Consistent
The search results show that there are only two implementations of the StreamSubscription interface:
SSEStreamSubscriptioninpackages/core/src/v3/apiClient/runStream.tswhich already implements the correct signature:
async subscribe(): Promise<ReadableStream<unknown>>- A test implementation in
packages/core/test/runStream.test.tswhich also matches the correct signature:
async subscribe(): Promise<ReadableStream<unknown>>All implementations already return Promise<ReadableStream<unknown>> as required. Other subscribe method occurrences in the codebase are unrelated to this interface (they belong to Redis subscriptions and other messaging systems).
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Find all implementations and usages of the `subscribe` method.
# Find implementations of the `subscribe` method
ast-grep --pattern $'class $_ implements StreamSubscription {
$$$
subscribe() {
$$$
}
$$$
}'
# Find usages of the `subscribe` method
rg 'subscribe\(' -t js -A 2
Length of output: 135
Script:
#!/bin/bash
# Let's try a different approach to find implementations and usages
# Find files containing StreamSubscription interface or implementations
rg -l "StreamSubscription" --type ts
# Find all subscribe method declarations and calls
rg "subscribe\(" --type ts -A 2
# Find interface definition
rg "interface StreamSubscription" --type ts -A 5
Length of output: 5532
Script:
#!/bin/bash
# Search for class implementations with 'subscribe' method
ast-grep --pattern 'class $_ {
$$$
subscribe() {
$$$
}
}'
Length of output: 69
packages/core/test/runStream.test.ts (1)
16-24: Ensure subscribe Method in Tests Aligns with Updated Interface
The subscribe method in TestStreamSubscription now returns a Promise<ReadableStream<unknown>>, aligning it with the updated StreamSubscription interface. Verify that all test cases using this method are updated to handle the ReadableStream appropriately.
| async subscribe(): Promise<ReadableStream<unknown>> { | ||
| return fetch(this.url, { | ||
| headers: { | ||
| Accept: "text/event-stream", | ||
| ...this.options.headers, | ||
| }, | ||
| signal: this.options.signal, | ||
| }); | ||
|
|
||
| if (!response.body) { | ||
| throw new Error("No response body"); | ||
| } | ||
|
|
||
| const reader = response.body | ||
| .pipeThrough(new TextDecoderStream()) | ||
| .pipeThrough(new EventSourceParserStream()) | ||
| .getReader(); | ||
|
|
||
| while (true) { | ||
| const { done, value } = await reader.read(); | ||
|
|
||
| if (done) break; | ||
|
|
||
| await onChunk(safeParseJSON(value.data)); | ||
| } | ||
| }).then((response) => { | ||
| if (!response.body) { | ||
| throw new Error("No response body"); | ||
| } | ||
|
|
||
| return () => reader.cancel(); | ||
| return response.body | ||
| .pipeThrough(new TextDecoderStream()) | ||
| .pipeThrough(new EventSourceParserStream()) | ||
| .pipeThrough( | ||
| new TransformStream({ | ||
| transform(chunk, controller) { | ||
| controller.enqueue(safeParseJSON(chunk.data)); | ||
| }, | ||
| }) | ||
| ); | ||
| }); |
There was a problem hiding this comment.
Add Error Handling for Fetch Response
In the SSEStreamSubscription class, the subscribe method performs a fetch request but doesn't check if the response is successful (e.g., HTTP status 200). If the response is an error (like 4xx or 5xx), this could lead to unexpected behavior.
Apply this diff to handle HTTP errors appropriately:
}).then((response) => {
+ if (!response.ok) {
+ throw new Error(`Network response was not ok: ${response.status} ${response.statusText}`);
+ }
if (!response.body) {
throw new Error("No response body");
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async subscribe(): Promise<ReadableStream<unknown>> { | |
| return fetch(this.url, { | |
| headers: { | |
| Accept: "text/event-stream", | |
| ...this.options.headers, | |
| }, | |
| signal: this.options.signal, | |
| }); | |
| if (!response.body) { | |
| throw new Error("No response body"); | |
| } | |
| const reader = response.body | |
| .pipeThrough(new TextDecoderStream()) | |
| .pipeThrough(new EventSourceParserStream()) | |
| .getReader(); | |
| while (true) { | |
| const { done, value } = await reader.read(); | |
| if (done) break; | |
| await onChunk(safeParseJSON(value.data)); | |
| } | |
| }).then((response) => { | |
| if (!response.body) { | |
| throw new Error("No response body"); | |
| } | |
| return () => reader.cancel(); | |
| return response.body | |
| .pipeThrough(new TextDecoderStream()) | |
| .pipeThrough(new EventSourceParserStream()) | |
| .pipeThrough( | |
| new TransformStream({ | |
| transform(chunk, controller) { | |
| controller.enqueue(safeParseJSON(chunk.data)); | |
| }, | |
| }) | |
| ); | |
| }); | |
| async subscribe(): Promise<ReadableStream<unknown>> { | |
| return fetch(this.url, { | |
| headers: { | |
| Accept: "text/event-stream", | |
| ...this.options.headers, | |
| }, | |
| signal: this.options.signal, | |
| }).then((response) => { | |
| if (!response.ok) { | |
| throw new Error(`Network response was not ok: ${response.status} ${response.statusText}`); | |
| } | |
| if (!response.body) { | |
| throw new Error("No response body"); | |
| } | |
| return response.body | |
| .pipeThrough(new TextDecoderStream()) | |
| .pipeThrough(new EventSourceParserStream()) | |
| .pipeThrough( | |
| new TransformStream({ | |
| transform(chunk, controller) { | |
| controller.enqueue(safeParseJSON(chunk.data)); | |
| }, | |
| }) | |
| ); | |
| }); |
@trigger.dev/build
@trigger.dev/core
@trigger.dev/react-hooks
trigger.dev
@trigger.dev/rsc
@trigger.dev/sdk
commit: |
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Outside diff range and nitpick comments (2)
packages/core/src/v3/apiClient/runStream.ts (2)
101-101: Excellent architectural improvement!The simplified
StreamSubscriptioninterface returningPromise<ReadableStream<unknown>>is a significant improvement that:
- Enables better composition with other stream processing tools
- Provides native backpressure handling
- Allows for easier error propagation
- Simplifies resource cleanup
115-146: Consider adding timeout and retry mechanismsWhile the stream implementation is solid, consider enhancing reliability with:
- Timeout handling for the fetch request
- Retry mechanism for temporary failures
async subscribe(): Promise<ReadableStream<unknown>> { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 30000); // 30s timeout + return fetch(this.url, { headers: { Accept: "text/event-stream", ...this.options.headers, }, - signal: this.options.signal, + signal: this.options.signal + ? composeAbortSignals(this.options.signal, controller.signal) + : controller.signal, }).then((response) => { + clearTimeout(timeout); if (!response.ok) { + if (response.status >= 500 || response.status === 429) { + // Retry after 1s for server errors or rate limiting + return new Promise((resolve) => setTimeout(resolve, 1000)) + .then(() => this.subscribe()); + } throw ApiError.generate( response.status, {},
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (1)
packages/core/src/v3/apiClient/runStream.ts(4 hunks)
🔇 Additional comments (1)
packages/core/src/v3/apiClient/runStream.ts (1)
123-130: Robust error handling implementation!
The error handling has been properly implemented using ApiError with detailed status and headers information.
| const stream = await subscription.subscribe(); | ||
|
|
||
| // Create the pipeline and start it | ||
| stream | ||
| .pipeThrough( | ||
| new TransformStream({ | ||
| transform(chunk, controller) { | ||
| controller.enqueue({ | ||
| type: streamKey, | ||
| chunk: chunk as TStreams[typeof streamKey], | ||
| run, | ||
| } as StreamPartResult<RunShape<TRunTypes>, TStreams>); | ||
| }, | ||
| }) | ||
| ) | ||
| .pipeTo( | ||
| new WritableStream({ | ||
| write(chunk) { | ||
| controller.enqueue(chunk); | ||
| }, | ||
| }) | ||
| ) | ||
| .catch((error) => { | ||
| console.error(`Error in stream ${streamKey}:`, error); | ||
| }); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Improve error handling in stream pipeline
The current error handling only logs to console, which could lead to silent failures. Consider:
- Propagating errors to the caller
- Adding error recovery mechanisms
- Providing error events that consumers can listen to
stream
.pipeThrough(
new TransformStream({
transform(chunk, controller) {
controller.enqueue({
type: streamKey,
chunk: chunk as TStreams[typeof streamKey],
run,
} as StreamPartResult<RunShape<TRunTypes>, TStreams>);
},
})
)
.pipeTo(
new WritableStream({
write(chunk) {
controller.enqueue(chunk);
},
+ abort(reason) {
+ controller.error(new Error(`Stream ${streamKey} aborted: ${reason}`));
+ }
})
)
.catch((error) => {
- console.error(`Error in stream ${streamKey}:`, error);
+ // Propagate error to consumer
+ controller.error(new Error(`Error in stream ${streamKey}: ${error.message}`));
+ // Attempt recovery by resubscribing after a delay
+ setTimeout(() => {
+ if (!activeStreams.has(streamKey)) return;
+ subscription.subscribe()
+ .then(/* ... handle resubscription ... */)
+ .catch(/* ... handle resubscription failure ... */);
+ }, 1000);
});Committable suggestion skipped: line range outside the PR's diff.
Previously subscribing to a single realtime stream would block other run updates and streams from being consumed, this update fixes it so multiple streams can be consumed at once and they don't block the run updates from being consumed.