Uh oh!
There was an error while loading. Please reload this page.
Fix stream serialization to resolve when user releases lock instead of waiting for stream to close - #661
Conversation
🦋 Changeset detectedLatest commit: 579162a The changes in this PR will be included in the next version bump. This PR includes changesets to release 12 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 |
📊 Benchmark Results
workflow with no steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Next.js (Turbopack) | Nitro workflow with 1 step💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) | Express | Nitro workflow with 10 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) Promise.all with 10 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) | Express Promise.all with 25 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Next.js (Turbopack) | Nitro Promise.race with 10 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) Promise.race with 25 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Express | Next.js (Turbopack) Stream Benchmarks(includes TTFB metrics)workflow with stream💻 Local Development
▲ Production (Vercel)
🔍 Observability: Express | Nitro | Next.js (Turbopack) SummaryFastest Framework by WorldWinner determined by most benchmark wins
Fastest World by FrameworkWinner determined by most benchmark wins
Column Definitions
Worlds:
|
🧪 E2E Test Results❌ Some tests failed Summary
❌ Failed Tests🌍 Community Worlds (11 failed)mongodb (1 failed):
redis (1 failed):
starter (8 failed):
turso (1 failed):
Details by Category✅ ▲ Vercel Production
✅ 💻 Local Development
✅ 📦 Local Production
✅ 🐘 Local Postgres
✅ 🪟 Windows
❌ 🌍 Community Worlds
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
This stack of pull requests is managed by Graphite. Learn more about stacking. |
| } catch (err) { | ||
| state.streamEnded = true; | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.reject(err); | ||
| } | ||
| throw err; |
There was a problem hiding this comment.
Stream locks are not released when an error occurs in the pipe. The error is thrown without releasing the locks acquired at the start of the function.
View Details
📝 Patch Details
diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts
index c8ebc73..23e2ae6 100644
--- a/packages/core/src/serialization.ts+++ b/packages/core/src/serialization.ts@@ -359,6 +359,8 @@ async function flushablePipe(
state.doneResolved = true;
state.reject(err);
}
+ reader.releaseLock();+ writer.releaseLock();
throw err;
}
}
Analysis
Stream locks not released when error occurs in flushablePipe
What fails: The flushablePipe() function in packages/core/src/serialization.ts does not release the reader and writer locks when an error is thrown in the catch block (lines 356-362), leaving the streams in a permanently locked state.
How to reproduce: Create a test scenario where an error occurs during pipe operation:
// Create a readable streamconstsource=newReadableStream({start(controller){controller.enqueue('chunk1');controller.enqueue('chunk2');controller.close();}});// Create a writable stream that errors on second writeletwriteCount=0;constsink=newWritableStream({asyncwrite(chunk){writeCount++;if(writeCount===2){thrownewError('Write failed!');}}});// Call flushablePipe with error stateconststate={pendingOps: 0,doneResolved: false,streamEnded: false,resolve: ()=>{},reject: ()=>{}};try{awaitflushablePipe(source,sink,state);}catch(err){// Error caught}// Attempt to acquire new reader/writersource.getReader();// Throws: "InvalidStateError: ReadableStream is locked"sink.getWriter();// Throws: "InvalidStateError: WritableStream is locked"Result: Streams remain locked (source.locked === true, sink.locked === true), preventing new readers/writers from being obtained. Cannot acquire new reader: "InvalidStateError: ReadableStream is locked". Cannot acquire new writer: "InvalidStateError: WritableStream is locked".
Expected behavior: The locks should be released before throwing the error, as they are in normal code paths (lines 326, 339, 346). Per WHATWG Streams API specification, a lock must be explicitly released, and a locked stream prevents other readers/writers from being obtained. Leaving locks unreleased creates a resource leak that prevents stream reuse.
Root cause: The catch block at lines 356-362 sets error state and throws without calling reader.releaseLock() and writer.releaseLock(), unlike all other exit paths from the function which properly release locks.
Fix applied: Added reader.releaseLock() and writer.releaseLock() before the throw err statement in the catch block to match the cleanup pattern used in normal code paths.
| if (readResult.done) { | ||
| // Source stream completed - close sink and resolve | ||
| state.streamEnded = true; | ||
| await writer.close(); | ||
| // Resolve done promise if not already resolved | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.resolve(); | ||
| } | ||
| return; |
There was a problem hiding this comment.
Stream locks are not released when the source stream ends. The flushablePipe function acquires locks on the reader and writer but doesn't release them before returning when the readable stream completes.
View Details
📝 Patch Details
diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts
index c8ebc73..57e4750 100644
--- a/packages/core/src/serialization.ts+++ b/packages/core/src/serialization.ts@@ -335,6 +335,8 @@ async function flushablePipe(
state.doneResolved = true;
state.resolve();
}
+ reader.releaseLock();+ writer.releaseLock();
return;
}
@@ -359,6 +361,8 @@ async function flushablePipe(
state.doneResolved = true;
state.reject(err);
}
+ reader.releaseLock();+ writer.releaseLock();
throw err;
}
}
Analysis
Stream locks not released in flushablePipe() when source completes or errors
What fails: The flushablePipe() function in packages/core/src/serialization.ts (lines 333-341 and 365-371) acquires locks on ReadableStream and WritableStream but fails to release them before returning when the stream completes or when an error occurs.
How to reproduce:
// When readResult.done is true (source stream completed):constreadResult=awaitreader.read();if(readResult.done){state.streamEnded=true;awaitwriter.close();// ... resolve promise ...return;// ❌ Returns without releasing locks}// Or when error occurs:}catch(err){state.streamEnded=true;// ... reject promise ...throwerr;// ❌ Throws without releasing locks}Result: Unreleased locks prevent any other code from acquiring new readers or writers on these streams. Subsequent attempts to call getReader() or getWriter() will throw TypeError: ReadableStreamDefaultReader constructor can only accept readable streams that are not yet locked to a reader. The stream becomes permanently inaccessible.
Expected behavior: According to MDN Web JavaScript resource management, unreleased locks cause the stream to remain locked indefinitely and "does not permit another reader to consume it." The locks must be explicitly released via reader.releaseLock() and writer.releaseLock() before the function exits.
Fix: Added reader.releaseLock() and writer.releaseLock() calls before all return/throw statements in the flushablePipe function:
- Before returning when
readResult.done(normal stream completion) - Before throwing in the catch block (error case)
There was a problem hiding this comment.
Pull request overview
This PR implements a polling mechanism to detect when stream locks are released, allowing workflow steps to complete without waiting for streams to fully close. This fixes an issue where Vercel functions would hang when users write to streams and release locks without explicitly closing them.
Key Changes:
- Added
flushablePipefunction that resolves when either the stream completes or when the user releases their lock with all pending writes flushed - Implemented polling functions (
pollWritableLockandpollReadableLock) that check stream lock state every 100ms - Created state tracking system (
FlushableStreamState) to monitor pending operations and resolution status
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 15 comments.
| File | Description |
|---|---|
| packages/core/src/serialization.ts | Implements the core flushable pipe mechanism with polling-based lock detection and state tracking for both readable and writable streams |
| packages/core/src/serialization.test.ts | Adds tests verifying the new behavior for both lock release and natural stream closure scenarios |
| .changeset/stream-lock-polling.md | Documents the patch-level change for the @workflow/core package |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| flushablePipe(readable, transform.writable, state).catch(() => { | ||
| // Errors are handled via state.reject | ||
| }); |
There was a problem hiding this comment.
The flushablePipe function throws errors after calling state.reject() (line 405), but all call sites catch and ignore these thrown errors with empty catch blocks (e.g., line 976, 992, 1016). This pattern is confusing because the error is both rejected through the state AND thrown. Since the error is already propagated via state.reject() which rejects the 'done' promise, the re-throw serves no purpose and the empty catch blocks make this unclear. Consider either removing the throw from flushablePipe or documenting why the error needs to be both rejected and thrown.
| } catch (err) { | ||
| state.streamEnded = true; | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.reject(err); | ||
| } | ||
| throw err; | ||
| } |
There was a problem hiding this comment.
The error handling in the flushablePipe function could potentially leak resources. When an exception occurs in the catch block at line 399, the reader and writer locks are not released before the error is re-thrown. This could leave the streams in a locked state. Consider releasing the locks in the catch block before re-throwing.
| /** | ||
| * Checks if a WritableStream is unlocked (user released lock) vs closed. | ||
| * When a stream is closed, .locked is false but getWriter() throws. | ||
| * We only want to resolve via polling when the stream is unlocked, not closed. | ||
| * If closed, the pump will handle resolution via the stream ending naturally. | ||
| */ | ||
| function isWritableUnlockedNotClosed(writable: WritableStream): boolean { | ||
| if (writable.locked) return false; | ||
| try { | ||
| // Try to acquire writer - if successful, stream is unlocked (not closed) | ||
| const writer = writable.getWriter(); | ||
| writer.releaseLock(); | ||
| return true; | ||
| } catch { | ||
| // getWriter() throws if stream is closed/errored - let pump handle it | ||
| return false; | ||
| } | ||
| } | ||
| /** | ||
| * Checks if a ReadableStream is unlocked (user released lock) vs closed. | ||
| */ | ||
| function isReadableUnlockedNotClosed(readable: ReadableStream): boolean { | ||
| if (readable.locked) return false; | ||
| try { | ||
| // Try to acquire reader - if successful, stream is unlocked (not closed) | ||
| const reader = readable.getReader(); | ||
| reader.releaseLock(); | ||
| return true; | ||
| } catch { | ||
| // getReader() throws if stream is closed/errored - let pump handle it |
There was a problem hiding this comment.
The polling functions acquire and release stream locks every 100ms to check if a stream is unlocked vs closed. This repeated lock acquisition could have performance implications and might interfere with stream operations. Consider using a different approach to detect stream closure, such as checking for specific error messages or using a flag that's set when the stream closes, to avoid repeatedly acquiring locks during polling.
| /** | |
| *ChecksifaWritableStreamisunlocked(userreleasedlock)vsclosed. | |
| *Whenastreamisclosed,.lockedisfalsebutgetWriter()throws. | |
| *Weonlywanttoresolveviapollingwhenthestreamisunlocked,notclosed. | |
| *Ifclosed,thepumpwillhandleresolutionviathestreamendingnaturally. | |
| */ | |
| functionisWritableUnlockedNotClosed(writable: WritableStream): boolean{ | |
| if(writable.locked)returnfalse; | |
| try{ | |
| // Try to acquire writer - if successful, stream is unlocked (not closed) | |
| constwriter=writable.getWriter(); | |
| writer.releaseLock(); | |
| returntrue; | |
| }catch{ | |
| // getWriter() throws if stream is closed/errored - let pump handle it | |
| returnfalse; | |
| } | |
| } | |
| /** | |
| *ChecksifaReadableStreamisunlocked(userreleasedlock)vsclosed. | |
| */ | |
| functionisReadableUnlockedNotClosed(readable: ReadableStream): boolean{ | |
| if(readable.locked)returnfalse; | |
| try{ | |
| // Try to acquire reader - if successful, stream is unlocked (not closed) | |
| constreader=readable.getReader(); | |
| reader.releaseLock(); | |
| returntrue; | |
| }catch{ | |
| // getReader() throws if stream is closed/errored - let pump handle it | |
| /** | |
| *Cacheofper-stream"unlocked but not closed"statusforWritableStreams. | |
| *Thisavoidsrepeatedlyacquiringandreleasinglocksduringpolling;we | |
| *probethestreamatmostonceperinstancewhenittransitionstounlocked. | |
| */ | |
| constwritableUnlockedNotClosedCache=newWeakMap<WritableStream,boolean>(); | |
| /** | |
| *ChecksifaWritableStreamisunlocked(userreleasedlock)vsclosed. | |
| *Whenastreamisclosed,.lockedisfalsebutgetWriter()throws. | |
| *Weonlywanttoresolveviapollingwhenthestreamisunlocked,notclosed. | |
| *Ifclosed,thepumpwillhandleresolutionviathestreamendingnaturally. | |
| * | |
| *Toavoidrepeatedlockacquisitionduringpolling,wecachetheresultof | |
| *theunlocked-vs-closedprobeperstream.Oncedetermined,thestatus is | |
| *reusedonsubsequentcallsforthesamestream. | |
| */ | |
| functionisWritableUnlockedNotClosed(writable: WritableStream): boolean{ | |
| if(writable.locked)returnfalse; | |
| constcached=writableUnlockedNotClosedCache.get(writable); | |
| if(cached!==undefined){ | |
| returncached; | |
| } | |
| try{ | |
| // Try to acquire writer - if successful, stream is unlocked (not closed) | |
| constwriter=writable.getWriter(); | |
| writer.releaseLock(); | |
| writableUnlockedNotClosedCache.set(writable,true); | |
| returntrue; | |
| }catch{ | |
| // getWriter() throws if stream is closed/errored - let pump handle it | |
| writableUnlockedNotClosedCache.set(writable,false); | |
| returnfalse; | |
| } | |
| } | |
| /** | |
| *Cacheofper-stream"unlocked but not closed"statusforReadableStreams. | |
| *Likethewritablecache,thispreventsrepeatedreaderacquisitionswhile | |
| *pollingforlockrelease. | |
| */ | |
| constreadableUnlockedNotClosedCache=newWeakMap<ReadableStream,boolean>(); | |
| /** | |
| *ChecksifaReadableStreamisunlocked(userreleasedlock)vsclosed. | |
| *Usesaper-streamcachetoavoidrepeatedlyacquiringandreleasinglocks | |
| *duringpolling. | |
| */ | |
| functionisReadableUnlockedNotClosed(readable: ReadableStream): boolean{ | |
| if(readable.locked)returnfalse; | |
| constcached=readableUnlockedNotClosedCache.get(readable); | |
| if(cached!==undefined){ | |
| returncached; | |
| } | |
| try{ | |
| // Try to acquire reader - if successful, stream is unlocked (not closed) | |
| constreader=readable.getReader(); | |
| reader.releaseLock(); | |
| readableUnlockedNotClosedCache.set(readable,true); | |
| returntrue; | |
| }catch{ | |
| // getReader() throws if stream is closed/errored - let pump handle it | |
| readableUnlockedNotClosedCache.set(readable,false); |
| } catch (err) { | ||
| state.streamEnded = true; | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.reject(err as Error); | ||
| } | ||
| } | ||
| })(); |
There was a problem hiding this comment.
The error handling in the test's pipe implementation (line 1179-1185) doesn't release the reader and writer locks before exiting, similar to the issue in the actual flushablePipe implementation. While this mirrors the production code, it would be better to release the locks in the catch block to properly clean up resources. Consider updating the test to match the corrected implementation once the production code is fixed.
| flushablePipe(serialize.readable, serverWritable, state).catch(() => { | ||
| // Errors are handled via state.reject | ||
| }); |
There was a problem hiding this comment.
The flushablePipe function throws errors after calling state.reject() (line 405), but all call sites catch and ignore these thrown errors with empty catch blocks (e.g., line 976, 992, 1016). This pattern is confusing because the error is both rejected through the state AND thrown. Since the error is already propagated via state.reject() which rejects the 'done' promise, the re-throw serves no purpose and the empty catch blocks make this unclear. Consider either removing the throw from flushablePipe or documenting why the error needs to be both rejected and thrown.
| flushablePipe(serialize.readable,serverWritable,state).catch(()=>{ | |
| // Errors are handled via state.reject | |
| }); | |
| flushablePipe(serialize.readable,serverWritable,state).catch( | |
| (error)=>{ | |
| // flushablePipe already calls state.reject(error), which rejects the | |
| // `done` promise added to `ops`. We catch and intentionally ignore | |
| // this rejection here only to avoid an unhandled promise rejection; | |
| // all real error handling flows through state.done. | |
| } | |
| ); |
| describe('flushable stream behavior', () => { | ||
| const POLL_INTERVAL = 100; // Match the actual implementation | ||
| const STABLE_POLL_COUNT = 2; // Match the actual implementation | ||
| it('done promise should resolve when writable stream lock is released (polling)', async () => { | ||
| // Test the pattern: user writes, releases lock, polling detects it, done resolves | ||
| const chunks: string[] = []; | ||
| let streamClosed = false; | ||
| // Create a simple mock for the sink | ||
| const mockSink = new WritableStream<string>({ | ||
| write(chunk) { | ||
| chunks.push(chunk); | ||
| }, | ||
| close() { | ||
| streamClosed = true; | ||
| }, | ||
| }); | ||
| // Create a TransformStream like we do in getStepRevivers | ||
| const { readable, writable } = new TransformStream<string, string>(); | ||
| // Track flushable state - this mirrors the actual implementation | ||
| const state = { | ||
| pendingOps: 0, // Only counts writes to server | ||
| doneResolved: false, | ||
| streamEnded: false, | ||
| resolve: () => {}, | ||
| reject: (_err: Error) => {}, | ||
| }; | ||
| const done = new Promise<void>((res, rej) => { | ||
| state.resolve = res; | ||
| state.reject = rej; | ||
| }); | ||
| // Start piping in background (mirrors flushablePipe implementation) | ||
| (async () => { | ||
| const reader = readable.getReader(); | ||
| const writer = mockSink.getWriter(); | ||
| try { | ||
| while (!state.streamEnded) { | ||
| const result = await reader.read(); | ||
| if (result.done) { | ||
| state.streamEnded = true; | ||
| await writer.close(); | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.resolve(); | ||
| } | ||
| return; | ||
| } | ||
| // Only writes count as pending ops | ||
| state.pendingOps++; | ||
| await writer.write(result.value); | ||
| state.pendingOps--; | ||
| if (state.streamEnded) { | ||
| reader.releaseLock(); | ||
| writer.releaseLock(); | ||
| return; | ||
| } | ||
| } | ||
| } catch (err) { | ||
| state.streamEnded = true; | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.reject(err as Error); | ||
| } | ||
| } | ||
| })(); | ||
| // Start polling (mirrors pollWritableLock implementation) | ||
| let stableCount = 0; | ||
| const intervalId = setInterval(() => { | ||
| if (state.doneResolved || state.streamEnded) { | ||
| clearInterval(intervalId); | ||
| return; | ||
| } | ||
| // Check if lock is released by checking .locked property | ||
| if (!writable.locked && state.pendingOps === 0) { | ||
| stableCount++; | ||
| if (stableCount >= STABLE_POLL_COUNT) { | ||
| state.doneResolved = true; | ||
| state.resolve(); | ||
| clearInterval(intervalId); | ||
| } | ||
| } else { | ||
| stableCount = 0; | ||
| } | ||
| }, POLL_INTERVAL); | ||
| // Simulate user interaction - write and release lock | ||
| const userWriter = writable.getWriter(); | ||
| await userWriter.write('chunk1'); | ||
| await userWriter.write('chunk2'); | ||
| // Release lock without closing stream | ||
| userWriter.releaseLock(); | ||
| // Wait for pipe to process + polling intervals (need STABLE_POLL_COUNT consecutive polls) | ||
| await new Promise((r) => setTimeout(r, 250)); | ||
| // The done promise should resolve | ||
| await expect( | ||
| Promise.race([ | ||
| done, | ||
| new Promise((_, r) => setTimeout(() => r(new Error('timeout')), 400)), | ||
| ]) | ||
| ).resolves.toBeUndefined(); | ||
| // Chunks should have been written | ||
| expect(chunks).toContain('chunk1'); | ||
| expect(chunks).toContain('chunk2'); | ||
| // Stream should NOT be closed (user only released lock) | ||
| expect(streamClosed).toBe(false); | ||
| }); | ||
| it('done promise should resolve when writable stream closes naturally', async () => { | ||
| const chunks: string[] = []; | ||
| let streamClosed = false; | ||
| const mockSink = new WritableStream<string>({ | ||
| write(chunk) { | ||
| chunks.push(chunk); | ||
| }, | ||
| close() { | ||
| streamClosed = true; | ||
| }, | ||
| }); | ||
| const { readable, writable } = new TransformStream<string, string>(); | ||
| const state = { | ||
| pendingOps: 0, | ||
| doneResolved: false, | ||
| streamEnded: false, | ||
| resolve: () => {}, | ||
| reject: (_err: Error) => {}, | ||
| }; | ||
| const done = new Promise<void>((res, rej) => { | ||
| state.resolve = res; | ||
| state.reject = rej; | ||
| }); | ||
| // Start piping in background | ||
| (async () => { | ||
| const reader = readable.getReader(); | ||
| const writer = mockSink.getWriter(); | ||
| try { | ||
| while (!state.streamEnded) { | ||
| const result = await reader.read(); | ||
| if (result.done) { | ||
| state.streamEnded = true; | ||
| await writer.close(); | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.resolve(); | ||
| } | ||
| return; | ||
| } | ||
| state.pendingOps++; | ||
| await writer.write(result.value); | ||
| state.pendingOps--; | ||
| if (state.streamEnded) { | ||
| reader.releaseLock(); | ||
| writer.releaseLock(); | ||
| return; | ||
| } | ||
| } | ||
| } catch (err) { | ||
| state.streamEnded = true; | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.reject(err as Error); | ||
| } | ||
| } | ||
| })(); | ||
| // Start polling (won't trigger since stream will close first) | ||
| let stableCount2 = 0; | ||
| const intervalId = setInterval(() => { | ||
| if (state.doneResolved || state.streamEnded) { | ||
| clearInterval(intervalId); | ||
| return; | ||
| } | ||
| if (!writable.locked && state.pendingOps === 0) { | ||
| stableCount2++; | ||
| if (stableCount2 >= STABLE_POLL_COUNT) { | ||
| state.doneResolved = true; | ||
| state.resolve(); | ||
| clearInterval(intervalId); | ||
| } | ||
| } else { | ||
| stableCount2 = 0; | ||
| } | ||
| }, POLL_INTERVAL); | ||
| // User writes and then closes the stream | ||
| const userWriter = writable.getWriter(); | ||
| await userWriter.write('data'); | ||
| await userWriter.close(); | ||
| // Wait a tick for the pipe to process | ||
| await new Promise((r) => setTimeout(r, 50)); | ||
| // The done promise should resolve | ||
| await expect( | ||
| Promise.race([ | ||
| done, | ||
| new Promise((_, r) => setTimeout(() => r(new Error('timeout')), 200)), | ||
| ]) | ||
| ).resolves.toBeUndefined(); | ||
| // Chunks should have been written | ||
| expect(chunks).toContain('data'); | ||
| // Stream should be closed (user closed it) | ||
| expect(streamClosed).toBe(true); | ||
| }); | ||
| }); |
There was a problem hiding this comment.
The new flushable stream behavior tests only cover the happy paths (lock release and normal closure). Consider adding test cases for error scenarios such as: stream errors during write operations, reader errors, and what happens when the stream is aborted. This would help ensure the error handling in flushablePipe and the polling mechanisms work correctly under failure conditions.
| flushablePipe(readable, writable, state).catch(() => { | ||
| // Errors are handled via state.reject | ||
| }); |
There was a problem hiding this comment.
The flushablePipe function throws errors after calling state.reject() (line 405), but all call sites catch and ignore these thrown errors with empty catch blocks (e.g., line 976, 992, 1016). This pattern is confusing because the error is both rejected through the state AND thrown. Since the error is already propagated via state.reject() which rejects the 'done' promise, the re-throw serves no purpose and the empty catch blocks make this unclear. Consider either removing the throw from flushablePipe or documenting why the error needs to be both rejected and thrown.
| * - `doneResolved`: The `done` promise has been resolved (step can complete) | ||
| * - `streamEnded`: The underlying stream has actually closed/errored | ||
| * | ||
| * The pump continues running even after `doneResolved=true` to handle | ||
| * any future writes if the user acquires a new lock. |
There was a problem hiding this comment.
The comment states that "The pump continues running even after doneResolved=true to handle any future writes if the user acquires a new lock." However, this design could be problematic. Once the done promise resolves (pushed to ops), the workflow step will complete. If the user then acquires a new lock and writes more data after the step has completed, it could lead to unexpected behavior where data is being written to a stream associated with an already-completed step. Consider whether this is the intended behavior or if the stream should be closed/locked after the done promise resolves.
| *-`doneResolved`: The`done`promisehasbeenresolved(stepcancomplete) | |
| *-`streamEnded`: Theunderlyingstreamhasactuallyclosed/errored | |
| * | |
| *Thepumpcontinuesrunningevenafter`doneResolved=true`tohandle | |
| * any futurewritesiftheuseracquiresanewlock. | |
| *-`doneResolved`: The`done`promisehasbeenresolved(stepcancomplete), | |
| *andnofurtherwritesshouldbeacceptedforthisstream. | |
| *-`streamEnded`: Theunderlyingstreamhasactuallyclosed/errored. | |
| * | |
| *Once`doneResolved`becomes`true`,thepumpshouldonlydrainany | |
| *already-pendingoperationsandthenstop;itmustnothandlenewwrites | |
| *fromsubsequentlyacquiredlocks. |
| } catch (err) { | ||
| state.streamEnded = true; | ||
| if (!state.doneResolved) { | ||
| state.doneResolved = true; | ||
| state.reject(err as Error); | ||
| } | ||
| } | ||
| })(); |
There was a problem hiding this comment.
The error handling in the test's pipe implementation (line 1066-1072) doesn't release the reader and writer locks before exiting, similar to the issue in the actual flushablePipe implementation. While this mirrors the production code, it would be better to release the locks in the catch block to properly clean up resources. Consider updating the test to match the corrected implementation once the production code is fixed.
| let stableCount2 = 0; | ||
| const intervalId = setInterval(() => { | ||
| if (state.doneResolved || state.streamEnded) { | ||
| clearInterval(intervalId); | ||
| return; | ||
| } | ||
| if (!writable.locked && state.pendingOps === 0) { | ||
| stableCount2++; | ||
| if (stableCount2 >= STABLE_POLL_COUNT) { | ||
| state.doneResolved = true; | ||
| state.resolve(); | ||
| clearInterval(intervalId); | ||
| } | ||
| } else { | ||
| stableCount2 = 0; |
There was a problem hiding this comment.
The variable name stableCount2 is unclear and suggests copy-paste from the first test. Consider renaming to stableCount for consistency, as the numbering doesn't add clarity and makes the code harder to read.
| letstableCount2=0; | |
| constintervalId=setInterval(()=>{ | |
| if(state.doneResolved||state.streamEnded){ | |
| clearInterval(intervalId); | |
| return; | |
| } | |
| if(!writable.locked&&state.pendingOps===0){ | |
| stableCount2++; | |
| if(stableCount2>=STABLE_POLL_COUNT){ | |
| state.doneResolved=true; | |
| state.resolve(); | |
| clearInterval(intervalId); | |
| } | |
| }else{ | |
| stableCount2=0; | |
| letstableCount=0; | |
| constintervalId=setInterval(()=>{ | |
| if(state.doneResolved||state.streamEnded){ | |
| clearInterval(intervalId); | |
| return; | |
| } | |
| if(!writable.locked&&state.pendingOps===0){ | |
| stableCount++; | |
| if(stableCount>=STABLE_POLL_COUNT){ | |
| state.doneResolved=true; | |
| state.resolve(); | |
| clearInterval(intervalId); | |
| } | |
| }else{ | |
| stableCount=0; |
VaguelySerious
left a comment
There was a problem hiding this comment.
Before I review - there's a ton of AI comments on this PR and I want to give you a chance to resolve those, so I'm subscribing and re-reviewing after the next commit/message
TooTallNate
commented
Dec 22, 2025
@TooTallNate I've opened a new pull request, #677, to work on those changes. Once the pull request is ready, I'll request review from you. |
…f waiting for stream to close This prevents Vercel functions from hanging when users incrementally write to streams within steps (e.g., `await writer.write(data); writer.releaseLock()`). Uses a polling approach to detect when the stream lock is released and all pending writes are flushed.
b58f72a to
579162aCompare| function pollWritableLock( | ||
| writable: WritableStream, | ||
| state: FlushableStreamState | ||
| ): void { | ||
| const intervalId = setInterval(() => { | ||
| // Stop polling if already resolved or stream ended | ||
| if (state.doneResolved || state.streamEnded) { | ||
| clearInterval(intervalId); | ||
| return; | ||
| } | ||
| // Check if lock is released (not closed) and no pending ops | ||
| if (isWritableUnlockedNotClosed(writable) && state.pendingOps === 0) { | ||
| state.doneResolved = true; | ||
| state.resolve(); | ||
| clearInterval(intervalId); | ||
| } | ||
| }, LOCK_POLL_INTERVAL_MS); | ||
| } | ||
| /** | ||
| * Polls a ReadableStream to check if the user has released their lock. | ||
| * Resolves the done promise when lock is released and no pending ops remain. | ||
| * | ||
| * Note: Only resolves if stream is unlocked but NOT closed. If the user closes | ||
| * the stream, the pump will handle resolution via the stream ending naturally. | ||
| */ | ||
| function pollReadableLock( | ||
| readable: ReadableStream, | ||
| state: FlushableStreamState | ||
| ): void { | ||
| const intervalId = setInterval(() => { | ||
| // Stop polling if already resolved or stream ended | ||
| if (state.doneResolved || state.streamEnded) { | ||
| clearInterval(intervalId); | ||
| return; | ||
| } | ||
| // Check if lock is released (not closed) and no pending ops | ||
| if (isReadableUnlockedNotClosed(readable) && state.pendingOps === 0) { | ||
| state.doneResolved = true; | ||
| state.resolve(); | ||
| clearInterval(intervalId); | ||
| } | ||
| }, LOCK_POLL_INTERVAL_MS); | ||
| } |
There was a problem hiding this comment.
The polling functions don't implement the "multiple consecutive polls" behavior mentioned in the docstring, creating a potential race condition where the done promise resolves before the pump finishes writing queued data.
View Details
📝 Patch Details
diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts
index 69c391b..376afe4 100644
--- a/packages/core/src/serialization.ts+++ b/packages/core/src/serialization.ts@@ -182,6 +182,9 @@ export class WorkflowServerWritableStream extends WritableStream<Uint8Array> {
/** Polling interval for checking stream lock state (in milliseconds) */
const LOCK_POLL_INTERVAL_MS = 100;
+/** Number of consecutive stable polls before resolving done promise */+const STABLE_POLL_COUNT = 2;+
/**
* State tracker for flushable stream operations.
* Resolves when either:
@@ -272,6 +275,7 @@ function pollWritableLock(
writable: WritableStream,
state: FlushableStreamState
): void {
+ let stableCount = 0;
const intervalId = setInterval(() => {
// Stop polling if already resolved or stream ended
if (state.doneResolved || state.streamEnded) {
@@ -281,16 +285,22 @@ function pollWritableLock(
// Check if lock is released (not closed) and no pending ops
if (isWritableUnlockedNotClosed(writable) && state.pendingOps === 0) {
- state.doneResolved = true;- state.resolve();- clearInterval(intervalId);+ stableCount++;+ if (stableCount >= STABLE_POLL_COUNT) {+ state.doneResolved = true;+ state.resolve();+ clearInterval(intervalId);+ }+ } else {+ stableCount = 0;
}
}, LOCK_POLL_INTERVAL_MS);
}
/**
* Polls a ReadableStream to check if the user has released their lock.
- * Resolves the done promise when lock is released and no pending ops remain.+ * Resolves the done promise when lock is released and no pending ops remain+ * for multiple consecutive polls (to avoid race conditions with the pump).
*
* Note: Only resolves if stream is unlocked but NOT closed. If the user closes
* the stream, the pump will handle resolution via the stream ending naturally.
@@ -299,6 +309,7 @@ function pollReadableLock(
readable: ReadableStream,
state: FlushableStreamState
): void {
+ let stableCount = 0;
const intervalId = setInterval(() => {
// Stop polling if already resolved or stream ended
if (state.doneResolved || state.streamEnded) {
@@ -308,9 +319,14 @@ function pollReadableLock(
// Check if lock is released (not closed) and no pending ops
if (isReadableUnlockedNotClosed(readable) && state.pendingOps === 0) {
- state.doneResolved = true;- state.resolve();- clearInterval(intervalId);+ stableCount++;+ if (stableCount >= STABLE_POLL_COUNT) {+ state.doneResolved = true;+ state.resolve();+ clearInterval(intervalId);+ }+ } else {+ stableCount = 0;
}
}, LOCK_POLL_INTERVAL_MS);
}
Analysis
Race condition in polling functions: done promise resolves before pump finishes flushing
What fails: The pollWritableLock() and pollReadableLock() functions in packages/core/src/serialization.ts resolve the done promise immediately when the stream lock is released and pendingOps === 0, without waiting for multiple consecutive stable polls. This creates a race condition where the done promise can resolve while the pump is still reading/writing data.
How to reproduce:
- A stream is being piped asynchronously via
flushablePipe() - User releases the lock on the stream (but doesn't close it)
- Pump is momentarily between reads (at
await reader.read()) withpendingOps === 0 - Polling fires and sees: lock released +
pendingOps === 0→ resolves done promise immediately - Runtime awaits
Promise.all(ops)which includes the done promise and proceeds - If pump reads more data and tries to write, the step is already marked complete
- This violates the contract that all writes must be flushed before the step completes
Expected behavior: Per the docstring at lines 265-266 and 293-294, polling should only resolve after the condition is stable for "multiple consecutive polls (to avoid race conditions with the pump)." The test file at lines 1002-1003 and 1084-1090 confirms this behavior by using STABLE_POLL_COUNT = 2 - only resolving after the unlock condition is seen 2 consecutive times.
Root cause: The actual implementation was missing the stability counter logic that the docstring promised and that the test expected. Added STABLE_POLL_COUNT constant and stability tracking to both polling functions.
Evidence:
- Docstring promises: "for multiple consecutive polls (to avoid race conditions with the pump)"
- Test implementation (lines 1084-1090): Implements stability counter that resets when condition changes
- Test comment (line 1001): "mirrors the actual implementation"
- Git history: Commit e8c3800 added polling with docstring but without stability logic
Fix stream serialization to resolve when users release locks instead of waiting for streams to close, preventing Vercel functions from hanging.
What changed?
flushablePipefunction that resolves in two scenarios:How to test?
Create a workflow step that incrementally writes to a stream:
Verify the step completes immediately after lock release rather than hanging
Run the new test cases that verify both lock release and normal stream closure behaviors
Why make this change?
This fixes an issue where Vercel functions would hang when users incrementally write to streams within steps. Previously, the system would wait for the stream to fully close before resolving, but many users follow a pattern where they write data and release the lock without explicitly closing the stream. This change allows steps to complete as soon as the user releases the lock and all pending writes are flushed, which is the expected behavior in most streaming scenarios.