fix: resume open streams from tail cursors #3430
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| --- | ||
| '@workflow/web': patch | ||
| --- | ||
|
|
||
| Resume dashboard stream readers from the last delivered chunk without replaying the current tail. |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -8,7 +8,6 @@ import { | |
| import { getWebRevivers } from '@workflow/web-shared'; | ||
| import type { WorkflowRunStatus } from '@workflow/world'; | ||
| import { useCallback, useEffect, useRef, useState } from 'react'; | ||
| import { readStream } from '~/lib/workflow-api-client'; | ||
|
|
||
| export interface StreamChunk { | ||
|
|
@@ -21,7 +20,7 @@ const FRAME_HEADER_SIZE = 4; | |
| const ENCRYPTED_PLACEHOLDER = '[Encrypted]'; | ||
| const POLL_INTERVAL_MS = 3000; | ||
|
|
||
| function isRunActive(status: WorkflowRunStatus): boolean { | ||
| return status === 'pending' || status === 'running'; | ||
| } | ||
|
|
||
|
|
@@ -49,21 +48,16 @@ function detectEncoding(data: Uint8Array): StreamEncoding { | |
| } | ||
|
|
||
| export function useStreamReader( | ||
| streamId: string | null, | ||
| runId: string, | ||
| encryptionKey: Uint8Array | null, | ||
| runStatus: WorkflowRunStatus | ||
| ) { | ||
| const [chunks, setChunks] = useState<StreamChunk[]>([]); | ||
| const [isLive, setIsLive] = useState(false); | ||
| const [isInitialLoading, setIsInitialLoading] = useState(Boolean(streamId)); | ||
| const [error, setError] = useState<string | null>(null); | ||
| const chunkIdRef = useRef(0); | ||
| const pollTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null); | ||
| const runStatusRef = useRef(runStatus); | ||
| runStatusRef.current = runStatus; | ||
|
|
@@ -132,24 +126,22 @@ export function useStreamReader( | |
| setChunks([]); | ||
| setError(null); | ||
| setIsLive(false); | ||
| setIsInitialLoading(Boolean(streamId)); | ||
| chunkIdRef.current = 0; | ||
| serverCursorRef.current = null; | ||
|
|
||
| if (pollTimerRef.current) { | ||
| clearTimeout(pollTimerRef.current); | ||
| pollTimerRef.current = null; | ||
| } | ||
|
|
||
| if (!streamId) { | ||
| setIsInitialLoading(false); | ||
| return; | ||
| } | ||
|
|
||
| let mounted = true; | ||
| const abortController = new AbortController(); | ||
|
|
||
| const revivers = getWebRevivers(); | ||
|
|
||
|
|
@@ -168,37 +160,33 @@ export function useStreamReader( | |
| * Fetch stream data and parse frames. | ||
| * | ||
| * When `cursor` is provided, the server only returns chunks after that | ||
| * position (incremental fetch). | ||
| */ | ||
| const fetchAndParse = async ( | ||
| cryptoKey: PayloadKey | undefined, | ||
| cursor: string | null | ||
| ): Promise< | ||
| | { encrypted: true } | ||
| | { | ||
| encrypted: false; | ||
| chunks: StreamChunk[]; | ||
| cursor: string | null; | ||
| done: boolean; | ||
| } | ||
| > => { | ||
| const streamResponse = await readStream( | ||
| streamId, | ||
| runId, | ||
| cursor, | ||
| abortController.signal | ||
| ); | ||
|
|
||
| const chunks: StreamChunk[] = []; | ||
| const reader = streamResponse.body.getReader(); | ||
| const decoder = new TextDecoder(); | ||
| let buffer = new Uint8Array(0); | ||
| let encoding: StreamEncoding | null = null; | ||
| let textRemainder = ''; | ||
|
|
||
| const appendToBuffer = (data: Uint8Array) => { | ||
| const newBuffer = new Uint8Array(buffer.length + data.length); | ||
|
|
@@ -213,10 +201,7 @@ export function useStreamReader( | |
| const { value, done } = await reader.read(); | ||
| if (done) { | ||
| if (encoding === 'legacy' && textRemainder.trim()) { | ||
| chunks.push(parseLegacyLine(textRemainder.trim())); | ||
| textRemainder = ''; | ||
| } | ||
| break; | ||
|
|
@@ -243,10 +228,7 @@ export function useStreamReader( | |
| for (const line of lines) { | ||
| const trimmed = line.trim(); | ||
| if (trimmed) { | ||
| chunks.push(parseLegacyLine(trimmed)); | ||
| } | ||
| } | ||
| continue; | ||
|
|
@@ -277,17 +259,12 @@ export function useStreamReader( | |
| ); | ||
| offset += FRAME_HEADER_SIZE + frameLength; | ||
|
|
||
| const result = await processFrame(frameData, cryptoKey, revivers); | ||
| if (result.encrypted) { | ||
| reader.cancel().catch(() => {}); | ||
| return { encrypted: true }; | ||
| } | ||
| chunks.push(result.chunk); | ||
|
|
||
| framesInBatch++; | ||
| if (framesInBatch % YIELD_EVERY_N_FRAMES === 0) { | ||
|
|
@@ -301,7 +278,7 @@ export function useStreamReader( | |
|
|
||
| return { | ||
| encrypted: false, | ||
| chunks, | ||
| cursor: streamResponse.cursor, | ||
| done: streamResponse.done, | ||
| }; | ||
|
|
@@ -315,8 +292,7 @@ export function useStreamReader( | |
| ? await deriveRunPayloadKeys(encryptionKey) | ||
| : undefined; | ||
|
|
||
| const result = await fetchAndParse(cryptoKey, null); | ||
|
|
||
| if (result.encrypted) { | ||
| if (mounted) { | ||
|
|
@@ -327,12 +303,11 @@ export function useStreamReader( | |
| return; | ||
| } | ||
|
|
||
| serverCursorRef.current = result.cursor; | ||
|
|
||
| if (!mounted || abortController.signal.aborted) return; | ||
|
|
||
| setChunks(result.chunks); | ||
| setIsInitialLoading(false); | ||
|
|
||
| // If the stream itself is done, no need to poll regardless of run status | ||
|
|
@@ -346,18 +321,15 @@ export function useStreamReader( | |
| const poll = async () => { | ||
| if (!mounted || abortController.signal.aborted) return; | ||
| try { | ||
| const pollResult = await fetchAndParse( | ||
| cryptoKey, | ||
| serverCursorRef.current | ||
| ); | ||
| if (!pollResult.encrypted) { | ||
| serverCursorRef.current = | ||
| pollResult.cursor ?? serverCursorRef.current; | ||
|
Member
There was a problem hiding this comment. AI Review: BlockingDropping Against a World still on the old contract, the open tail returns "Old contract" is not hypothetical here. It is any pinned Two ways out. Make the cursor change additive in #3475, adding a new Worth noting the reverse skew is already handled well:
Contributor
There was a problem hiding this comment. A |
||
| if (pollResult.chunks.length > 0 && mounted) { | ||
| setChunks((prev) => [...prev, ...pollResult.chunks]); | ||
| } | ||
| if (pollResult.done) { | ||
| setIsLive(false); | ||
|
|
@@ -396,7 +368,7 @@ export function useStreamReader( | |
| } | ||
| }; | ||
| // eslint-disable-next-line react-hooks/exhaustive-deps | ||
| }, [streamId, runId, encryptionKey, processFrame]); | ||
|
|
||
| // When run finishes, stop polling | ||
| useEffect(() => { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
AI Review: Note
if (!hasMore) expect(page.done).toBe(true)asserts an implication the contract explicitly disclaims. ThegetChunksdocs already say new chunks may appear in future requests even afterhasMoreisfalse, and this PR's premise is thathasMore: falseon an open stream is a normal terminal state for a page loop. The assertion only passes here because the stream is already closed before the loop runs.That makes it a trap for whoever copies this loop against a live stream, and it does not add coverage beyond the
expect(paginatedChunks).toHaveLength(streamChunks.length)below it. I would drop it, or hoist it to a single assertion after the loop with a comment that the stream was closed first.Since this is the loop shape users will copy, it would also be worth asserting the open-stream case somewhere: read a stream that is still open, confirm
hasMoregoesfalsewith a non-nullcursor, write another chunk, and confirm the cursor resumes after the delivered chunks. That is the behavior the whole stack turns on and the e2e suite does not cover it yet.