Fix streaming splits in realtime streams v2 #1549
Conversation
🦋 Changeset detectedLatest commit: d4f533f 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 |
|
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
apps/webapp/server.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. WalkthroughThe pull request introduces a patch for the Changes
Possibly related PRs
Suggested reviewers
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 (3)
apps/webapp/app/services/realtime/databaseRealtimeStreams.server.ts (1)
57-63: Consider batch processing for better performanceThe switch from batch processing to individual inserts might impact performance when dealing with high-volume streams. Consider using a buffering strategy to batch inserts while maintaining real-time characteristics.
-await this.options.prisma.realtimeStreamChunk.create({ - data: { - runId, - key: streamId, - sequence: sequence++, - value, - }, -}); +const BATCH_SIZE = 100; +let buffer = []; +buffer.push({ + runId, + key: streamId, + sequence: sequence++, + value, +}); + +if (buffer.length >= BATCH_SIZE) { + await this.options.prisma.realtimeStreamChunk.createMany({ + data: buffer, + }); + buffer = []; +}packages/core/src/v3/apiClient/stream.ts (1)
225-231: Consider reducing debug logging in productionThe extensive console logging might impact performance in production. Consider using a debug flag or environment variable to control logging.
-console.log("LineTransformStream", { - chunk, - lines, - fullLines, - buffer: this.buffer, - streamId, -}); +if (process.env.DEBUG_STREAMS === 'true') { + console.log("LineTransformStream", { + chunk, + lines, + fullLines, + buffer: this.buffer, + streamId, + }); +}packages/core/src/v3/apiClient/runStream.ts (1)
217-238: Consider optimizing the transform pipelineThe current implementation has multiple transform stages with logging, which might impact performance. Consider:
- Combining transforms where possible
- Making logging conditional
return zodShapeStream(SubscribeRealtimeStreamChunkRawShape, this.url, this.options) .pipeThrough( new TransformStream({ transform(chunk, controller) { - console.log("ElectricStreamSubscription chunk.value", chunk.value); + if (process.env.DEBUG_STREAMS === 'true') { + console.log("ElectricStreamSubscription chunk.value", chunk.value); + } controller.enqueue(chunk.value); }, }) ) .pipeThrough(new LineTransformStream(this.url)) .pipeThrough( new TransformStream({ transform(chunk, controller) { for (const line of chunk) { - console.log("ElectricStreamSubscription line", line); + if (process.env.DEBUG_STREAMS === 'true') { + console.log("ElectricStreamSubscription line", line); + } controller.enqueue(safeParseJSON(line)); } }, }) );
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (4)
.changeset/rude-walls-help.md(1 hunks)apps/webapp/app/services/realtime/databaseRealtimeStreams.server.ts(2 hunks)packages/core/src/v3/apiClient/runStream.ts(2 hunks)packages/core/src/v3/apiClient/stream.ts(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- .changeset/rude-walls-help.md
🔇 Additional comments (2)
apps/webapp/app/services/realtime/databaseRealtimeStreams.server.ts (1)
47-49: LGTM: Stream termination condition is now more explicit
The simplified condition if (done || !value) properly handles both stream completion and empty chunks.
packages/core/src/v3/apiClient/stream.ts (1)
238-244: LGTM: Proper handling of incomplete data
The flush implementation correctly handles any remaining data in the buffer when the stream ends.
| for (const line of chunk) { | ||
| console.log("ElectricStreamSubscription line", line); | ||
|
|
||
| controller.enqueue(safeParseJSON(line)); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Consider handling JSON parse errors explicitly
The safeParseJSON function silently returns the raw string on parse errors. Consider adding explicit error handling for malformed JSON.
-controller.enqueue(safeParseJSON(line));
+try {
+ const parsed = safeParseJSON(line);
+ if (typeof parsed === 'string' && line !== parsed) {
+ console.warn(`Failed to parse JSON for line: ${line}`);
+ }
+ controller.enqueue(parsed);
+} catch (error) {
+ console.error(`Error processing line: ${line}`, error);
+}📝 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.
| controller.enqueue(safeParseJSON(line)); | |
| try { | |
| const parsed = safeParseJSON(line); | |
| if (typeof parsed === 'string' && line !== parsed) { | |
| console.warn(`Failed to parse JSON for line: ${line}`); | |
| } | |
| controller.enqueue(parsed); | |
| } catch (error) { | |
| console.error(`Error processing line: ${line}`, error); | |
| } |
@trigger.dev/build
@trigger.dev/core
trigger.dev
@trigger.dev/react-hooks
@trigger.dev/rsc
@trigger.dev/sdk
commit: |
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Outside diff range and nitpick comments (7)
apps/webapp/server.ts (1)
84-95: Consider adding request logging for blocked endpoints.The middleware implementation correctly restricts access to realtime endpoints while maintaining the healthcheck endpoint. However, adding logging for blocked requests would help with monitoring and debugging.
Consider adding logging like this:
if (!req.url.startsWith("/realtime") && req.url !== "/healthcheck") { + console.log(`Blocked request to: ${req.method} ${req.url}`); res.status(404).send("Not Found"); return; }apps/webapp/test/authorizationRateLimitMiddleware.test.ts (2)
Line range hint
3-5: Consider documenting test timeout rationaleThe 30-second timeout is set globally for all tests. Consider:
- Adding a comment explaining why 30 seconds was chosen
- Using per-test timeouts for long-running tests
- Documenting which tests might approach this timeout
Example documentation:
+ // Set a longer timeout for rate limiting tests as they include + // multiple sleep operations to test time-based behaviors vi.setConfig({ testTimeout: 30_000 }); // 30 seconds timeout
Line range hint
171-315: Consider adding tests for concurrent scenariosWhile the test suite is comprehensive, consider adding these scenarios:
- Concurrent requests hitting the rate limiter simultaneously
- Redis connection errors/timeouts
- Race conditions in token bucket refill
Example test structure:
redisTest("should handle concurrent requests correctly", async ({ redis }) => { // Setup rate limiter with small window // Fire multiple requests simultaneously using Promise.all // Verify correct number of requests succeeded/failed }); redisTest("should handle Redis errors gracefully", async ({ redis }) => { // Setup rate limiter // Simulate Redis connection issues // Verify fallback behavior });packages/core/src/v3/runMetadata/metadataStream.ts (2)
68-69: Handle errors in async iterator returned by[Symbol.asyncIterator].Errors occurring in the
consumerStreammay not be propagated to the consumer of the iterator.Ensure that errors from the stream are properly caught and forwarded.
72-81: Manage cancellation instreamToAsyncIterator.When the consumer stops iteration early, the underlying stream reader should be canceled to prevent resource leaks.
Modify the function to handle cancellation:
async function* streamToAsyncIterator<T>(stream: ReadableStream<T>): AsyncIterableIterator<T> { const reader = stream.getReader(); try { while (true) { const { done, value } = await reader.read(); if (done) return; yield value; } } catch (error) { // Optionally handle errors from reader throw error; } finally { + await reader.cancel(); reader.releaseLock(); } }packages/core/src/v3/runMetadata/manager.ts (2)
Line range hint
232-243: Inconsistent parameter naming instreammethod.The parameter
valuewas renamed tosourcein the method signature butvalueis still used within the method. This inconsistency can lead to errors.Apply this diff to ensure consistent naming:
public async stream<T>( key: string, - value: AsyncIterable<T> | ReadableStream<T>, + source: AsyncIterable<T> | ReadableStream<T>, signal?: AbortSignal ): Promise<AsyncIterable<T>> { - const $value = value as AsyncIterable<T>; + const $source = source as AsyncIterable<T>; if (!this.runId) { - return $value; + return $source; } // ... existing code ... const streamInstance = new MetadataStream({ key, runId: this.runId, - source: $value, + source: $source, baseUrl: this.streamsBaseUrl, headers: this.apiClient.getHeaders(), signal, version: this.streamsVersion, });
236-236: Redundant setting of$$streamsBaseUrlin metadata.Setting
$$streamsBaseUrlevery time a new stream is created might be unnecessary if it doesn't change between streams.Consider setting
$$streamsBaseUrlonce during initialization rather than on every stream creation.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (7)
apps/webapp/server.ts(1 hunks)apps/webapp/test/authorizationRateLimitMiddleware.test.ts(1 hunks)packages/core/src/v3/runMetadata/manager.ts(1 hunks)packages/core/src/v3/runMetadata/metadataStream.ts(2 hunks)references/nextjs-realtime/package.json(1 hunks)references/nextjs-realtime/src/components/TriggerButtonWithStreaming.tsx(0 hunks)references/nextjs-realtime/src/trigger/ai.ts(0 hunks)
💤 Files with no reviewable changes (2)
- references/nextjs-realtime/src/trigger/ai.ts
- references/nextjs-realtime/src/components/TriggerButtonWithStreaming.tsx
🧰 Additional context used
🪛 Biome (1.9.4)
packages/core/src/v3/runMetadata/metadataStream.ts
[error] 15-15: void is confusing inside a union type.
Unsafe fix: Use undefined instead.
(lint/suspicious/noConfusingVoidType)
🔇 Additional comments (2)
references/nextjs-realtime/package.json (1)
10-11: LGTM! Deployment script added correctly.
The new deploy script is properly added and aligns with the existing trigger.dev CLI usage pattern.
Let's verify the trigger.dev CLI installation and version compatibility:
✅ Verification successful
Trigger.dev CLI and SDK versions are properly aligned
The verification confirms that both the CLI (trigger.dev) and SDK (@trigger.dev/sdk) packages are using the same workspace version ^3, ensuring compatibility for the deployment script to work correctly.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Verify trigger.dev CLI installation and version compatibility
# Expected: The CLI package should be listed in devDependencies with a matching version
# Check if trigger.dev CLI package exists in package.json
jq '.devDependencies["trigger.dev"]' references/nextjs-realtime/package.json
# Check for any potential version mismatches between SDK and CLI
jq '[.dependencies["@trigger.dev/sdk"], .devDependencies["trigger.dev"]]' references/nextjs-realtime/package.json
Length of output: 248
apps/webapp/test/authorizationRateLimitMiddleware.test.ts (1)
18-18: Consider alternatives to skipping tests in CI
Skipping these tests in GitHub Actions could lead to reduced test coverage in CI. Consider these alternatives:
- Set up Redis in GitHub Actions using service containers
- Mock Redis for CI environment
- Create a separate test suite for Redis-dependent tests
Let's check if other test files have similar patterns:
✅ Verification successful
Tests are intentionally skipped in CI due to Redis dependency
The skip pattern is consistent across Redis-dependent test files. The test file shows comprehensive Redis integration tests that:
- Use
@internal/testcontainersfor Redis setup - Test various rate limiting scenarios including token bucket, fixed window, and sliding window
- Verify edge cases and caching behavior
Since these tests require a Redis instance and are properly tested in a containerized environment locally, skipping them in CI is a valid approach. No changes needed.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for other test files that might be skipped in CI
rg -l "skipIf.*GITHUB_ACTIONS" "test/"
# Check if Redis is used in other test files
rg -l "redisTest" "test/"
Length of output: 154
Script:
#!/bin/bash
# Search for test files with skipIf and GITHUB_ACTIONS from repository root
fd -e test.ts -e test.js | xargs rg -l "skipIf.*GITHUB_ACTIONS"
# Check Redis usage in test files
fd -e test.ts -e test.js | xargs rg -l "redis"
# Look at the actual test file to understand Redis setup
fd "authorizationRateLimitMiddleware.test.ts" -x cat
Length of output: 13980
| if (process.env.ALLOW_ONLY_REALTIME_API === "true") { | ||
| // Block all requests that do not start with /realtime | ||
| app.use((req, res, next) => { | ||
| // Make sure /healthcheck is still accessible | ||
| if (!req.url.startsWith("/realtime") && req.url !== "/healthcheck") { | ||
| res.status(404).send("Not Found"); | ||
| return; | ||
| } | ||
|
|
||
| next(); | ||
| }); | ||
| } |
There was a problem hiding this comment.
💡 Codebase verification
🛠️ Refactor suggestion
The review comment is valid. Static assets need to be allowed.
The verification confirms that the application serves static assets from two middleware configurations:
/buildpath serving frompublic/buildwith long-term caching- Root path
/serving frompublicdirectory with 1-hour caching
The current REALTIME_API restriction would indeed block access to these essential static assets. The suggested improvement in the review to allow /build and /assets paths is correct, though it should also ensure the root public directory assets remain accessible.
Recommended paths to allow:
/realtime/*/healthcheck/build/*/assets/*- Static files from
public/*
🔗 Analysis chain
Verify impact on static assets and consider improvements.
The current implementation might block access to necessary static assets under /build and public directories. Additionally, the error response could be more informative for API consumers.
Let's verify the static asset paths that might be affected:
Consider these improvements:
if (process.env.ALLOW_ONLY_REALTIME_API === "true") {
app.use((req, res, next) => {
// Make sure /healthcheck is still accessible
- if (!req.url.startsWith("/realtime") && req.url !== "/healthcheck") {
+ if (!req.url.startsWith("/realtime") &&
+ !req.url.startsWith("/build") &&
+ !req.url.startsWith("/assets") &&
+ req.url !== "/healthcheck") {
res.status(404).send("Not Found");
return;
}
next();
});
}🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Find all static asset paths that might be affected
fd -t f . public/build -x echo "Static asset path:" {}
Length of output: 158
Script:
#!/bin/bash
# Let's first check if there are any static asset directories
fd -t d "^(public|build|assets|dist)$"
# Check for any static file serving middleware or route configurations
rg -A 5 "express\.static|app\.use.*static|serveStatic" apps/webapp/
Length of output: 769
| private initializeServerStream(): Promise<Response> { | ||
| const serverStream = this.serverStream.pipeThrough( | ||
| new TransformStream<T, string>({ | ||
| async transform(chunk, controller) { | ||
| controller.enqueue(JSON.stringify(chunk) + "\n"); | ||
| }, | ||
| }) | ||
| ); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Ensure proper error handling in initializeServerStream.
The fetch request may fail or throw an exception, which isn't currently handled. This could lead to unhandled promise rejections.
Consider wrapping the fetch call in a try-catch block or handling errors appropriately.
+try {
return fetch(
`${this.options.baseUrl}/realtime/${this.options.version ?? "v1"}/streams/${
this.options.runId
}/${this.options.key}`,
{
method: "POST",
headers: this.options.headers ?? {},
body: serverStream,
// @ts-expect-error
duplex: "half",
signal: this.controller.signal,
}
);
+} catch (error) {
+ // Handle error, possibly by rejecting the promise or logging
+ console.error('Failed to initialize server stream:', error);
+ throw error;
+}Committable suggestion skipped: line range outside the PR's diff.
| private serverIterator: AsyncIterator<T>; | ||
| private consumerIterator: AsyncIterator<T>; | ||
| private serverStream: ReadableStream<T>; | ||
| private consumerStream: ReadableStream<T>; | ||
| private streamPromise: Promise<void | Response>; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Avoid using 'void' in union types for clarity.
Using void in a union type can be confusing. It is better to use undefined instead to represent the absence of a value.
Apply this diff to improve type clarity:
-private streamPromise: Promise<void | Response>;
+private streamPromise: Promise<Response | undefined>;📝 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.
| private streamPromise: Promise<void | Response>; | |
| private streamPromise: Promise<Response | undefined>; |
🧰 Tools
🪛 Biome (1.9.4)
[error] 15-15: void is confusing inside a union type.
Unsafe fix: Use undefined instead.
(lint/suspicious/noConfusingVoidType)
| private createTeeStreams() { | ||
| const readableSource = new ReadableStream<T>({ | ||
| start: async (controller) => { | ||
| for await (const value of this.options.source) { | ||
| controller.enqueue(value); | ||
| } | ||
| return queue.shift()!; | ||
|
|
||
| controller.close(); | ||
| }, | ||
| }); | ||
|
|
||
| return { | ||
| serverIterator: teeIterator(this.serverQueue), | ||
| consumerIterator: teeIterator(this.consumerQueue), | ||
| }; | ||
| return readableSource.tee(); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Handle errors during stream creation in createTeeStreams.
If an error occurs while iterating over this.options.source, it may not be properly propagated, leading to unhandled exceptions. Consider adding error handling to manage exceptions during iteration.
You can modify the start method to catch and forward errors:
const readableSource = new ReadableStream<T>({
start: async (controller) => {
+ try {
for await (const value of this.options.source) {
controller.enqueue(value);
}
-
controller.close();
+ } catch (error) {
+ controller.error(error);
+ } finally {
+ controller.close();
+ }
},
});Committable suggestion skipped: line range outside the PR's diff.
Summary by CodeRabbit
New Features
LineTransformStreamclass for improved line-by-line processing of incoming data.ElectricStreamSubscriptionclass with better logging and error handling capabilities.Bug Fixes
@trigger.dev/sdkpackage.Improvements