fix(world-postgres): ignore stream rows written after the first EOF - #3712
fix(world-postgres): ignore stream rows written after the first EOF#3712himself65 wants to merge 3 commits into
Conversation
🦋 Changeset detectedLatest commit: 6a05ac4 The changes in this PR will be included in the next version bump. This PR includes changesets to release 1 package
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 |
|
@himself65 is attempting to deploy a commit to the Vercel Labs Team on Vercel. A member of the Team first needs to authorize it. |
karthikscale3
left a comment
karthikscale3
left a comment
There was a problem hiding this comment.
See inline comment.
| controller.enqueue(new Uint8Array(msg.data)); | ||
| } | ||
| if (msg.eof) { | ||
| closed = true; |
There was a problem hiding this comment.
Thanks for submitting a fix for this. One edge case remains: because the offset > 0 branch runs before this EOF handling, a positive startIndex can consume the first EOF as if it were a data chunk. With five data rows + EOF + retried data/EOF, get(..., 6) skips the first EOF and returns the post-EOF duplicate (or hangs if no later EOF arrives). Could we handle EOF before decrementing offset—or only apply the offset branch when !msg.eof—and add a regression test for a start index beyond the valid data count?
There was a problem hiding this comment.
Good catch, thanks. Fixed in d70e0e8 by exempting EOF rows from the offset branch (if (offset > 0 && !msg.eof)) — offsets count data chunks, matching getInfo's tailIndex, so the marker is never consumed. Added a regression test for a start index at (5) and past (6) the data count with a post-EOF duplicate and no trailing EOF; the 6 case hung until timeout before the fix.
karthikscale3
left a comment
karthikscale3
left a comment
There was a problem hiding this comment.
See inline comment.
| // index at or past the data count must still close the | ||
| // stream rather than consume the marker and then hang, or | ||
| // surface rows written after it. | ||
| if (offset > 0 && !msg.eof) { |
There was a problem hiding this comment.
One remaining edge case: a negative startIndex is calculated below from chunks.length, which includes rows after the first EOF. For A…E, EOF, duplicate-E, get(..., -1) computes an offset from seven rows, then reaches the first EOF without returning the final valid chunk. Could we derive dataCount from the first EOF (for example, with findIndex) and add a startIndex = -1 regression test?
There was a problem hiding this comment.
Thanks — fixed in 4319027: dataCount now comes from chunks.findIndex((c) => c.eof) (falling back to chunks.length with no EOF), so a negative start index resolves against the same rows enqueue delivers. Regression test added for -1 (→ e) and -2 (→ d, e) on A…E + EOF + duplicate-E; -1 returned nothing before the fix.
`streams.get()`'s catch-up loop closes the controller on the EOF row and
keeps iterating. A stream with rows after its first EOF — a producer that
retried its terminal write after a lost ACK or an overlapping attempt, so
the frame was appended and the stream closed again — then hits
`controller.enqueue()` on a closed controller. Node throws
ERR_INVALID_STATE ("Invalid state: Controller is already closed") out of
`start()`, the stream errors, and every chunk still queued is discarded:
a reader sees an empty, errored stream instead of the data that was
written before the EOF.
Track the first EOF and ignore everything after it. The unit test drives
`createStreamer` against a fake pool/drizzle with several data rows before
the EOF (with a single queued chunk the consumer drains it first by
microtask order and the failure does not reproduce).
Signed-off-by: Alex Yang <himself65@outlook.com>
Offsets count data chunks (`getInfo` reports tailIndex = dataCount - 1), but the `offset > 0` branch ran before the EOF handling, so a start index at or past the data count consumed the EOF marker as if it were data — returning a post-EOF duplicate as live data, or never closing when no later EOF existed. Exempt EOF rows from the offset skip and cover the boundary (start index == data count and > data count) in the test. Signed-off-by: Alex Yang <himself65@outlook.com>
…re the first EOF The negative offset was computed from `chunks.length` (minus a trailing EOF), which counts rows written after the first EOF marker. On A…E, EOF, duplicate-E a `get(..., -1)` resolved against seven rows and closed at the first EOF without ever returning the last valid chunk. Derive the data count from the first EOF instead, matching what `enqueue` delivers. Signed-off-by: Alex Yang <himself65@outlook.com>
4319027 to
6a05ac4
Compare
2 participants
Description
streams.get()in@workflow/world-postgrescloses the controller on the EOF row during its catch-up loop and keeps iterating. If the stream has rows after its first EOF — a producer that retried its terminal write after a lost ACK or an overlapping attempt, so the frame was appended and the stream closed a second time — the next row hitscontroller.enqueue()on a closed controller. Node throwsERR_INVALID_STATE(Invalid state: Controller is already closed) out ofstart(), the stream errors, and every chunk still queued is discarded: the reader sees an empty, errored stream instead of the data written before the EOF.We hit this in production: a long-running agent run pushed its output through
writeToStream/closeStream; an ingest retry duplicated the terminal frame (data ×3, EOF ×3, data ×2, EOF ×2inworkflow_stream_chunks), and the workflow's finalize step — which reads the stream from 0 — persisted the run as failed with empty output even though all the data was there. On our instance 143 streams had rows after their EOF and 1809 had more than one EOF row. Reproduces on 4.3.3, 4.3.4 andmain.The fix tracks the first EOF and ignores every later row (data or EOF). It does not change behaviour for well-formed streams.
How did you test your changes?
packages/world-postgres/src/streamer.test.tsdrivescreateStreameragainst a fake pool/drizzle (thepgClientis stubbed so no LISTEN socket is opened). It writes five data rows, an EOF, then a duplicate data row + EOF, and expects the stream to drain exactly the five rows. Without the fix it rejects withTypeError: Invalid state: Controller is already closed.enqueueblows away. With a single queued chunk the consumer drains it first by microtask order and the failure does not reproduce.tsc --noEmitandbiome checkon the package are clean (Biome's pre-existingnoExcessiveCognitiveComplexitywarning onenqueuegoes from 16 to 20 with the extra guard;getChunksalready warns at 21).PR Checklist - Required to merge
pnpm changesetwas run to create a changelog for this PR (@workflow/world-postgres: patch)git commit --signoffon your commits)@vercel/workflowin a comment once the PR is ready, and the above checklist is complete