fix(web): stop AsyncQueue.fail from ending a parked consumer normally - #721
fix(web): stop AsyncQueue.fail from ending a parked consumer normally#721ayaangazali wants to merge 1 commit into
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
Included review availability: Your plan provides up to 10 included reviews per hour; 8 remain after this review. 📝 WalkthroughWalkthrough
ChangesAsyncQueue behavior
Estimated code review effort: 3 (Moderate) | ~20 minutes Merge Risk: ⚪ Minimal · up to This change makes parked streaming consumers receive mid-stream failures instead of silently treating truncated output as normal completion. No actionable merge-blocking risk remains after normal checks and review. Sequence Diagram(s)sequenceDiagram
participant Producer
participant AsyncQueue
participant Consumer
Producer->>AsyncQueue: push, complete, or fail
AsyncQueue->>Consumer: signal parked consumer
Consumer->>AsyncQueue: next()
AsyncQueue-->>Consumer: buffered value, error, or completion
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Full details: Description checkExplanation The description clearly explains the defect, impact, implementation, compatibility behavior, and verification results. It does not use every template heading or checklist item, but the missing items are non-critical for this change. Full details: Docstring CoverageExplanation No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check. Docstring coverage is scoped to functions touched by this diff. Analyzed 0 functions across 2 files. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
037d2ef to
d2ff1ab
Compare
|
Rebased onto Still reproduces on fail(error: Error): void {
if (this.done) return;
this.done = true;
this.error = error;
if (this.resolveNext) {
const r = this.resolveNext;
this.resolveNext = null;
r({ value: undefined as unknown as T, done: true }); // <- resolves, never throws
}
}A consumer already parked in One thing worth raising rather than sitting on, since it may change what you want here: this file duplicates a class that already exists and already gets this right. So there is a smaller change available than this one: delete the 79-line web copy and point the four importers ( I did not do that here because it is a different change from the one you have been sent, and I did not want to swap it under you. Say the word and I will convert this PR to the deletion, or keep this as the minimal fix. |
d2ff1ab to
4bb9dee
Compare
`fail()` resolved a parked consumer with `{ done: true }` and only stored the
error, so the error was reachable solely from a later `next()` that a finished
`for await` loop never makes. Whether a caller saw the failure or a clean
end-of-stream depended on whether it happened to be awaiting at that instant,
and during streaming it almost always is: the consumer drains tokens faster
than inference produces them, so it is parked nearly all the time.
The visible effect on `streamGenerate` (RunAnywhere+TextGeneration.ts:183,
which fails both the token queue and the event queue on a mid-stream error) is
that iterating `stream` or `events` ends as though generation completed. The
caller keeps a truncated answer and is told nothing. The error survives only on
the separate `result` promise, which a streaming caller need not await.
Producers now just signal and `next()` re-reads the queue state, so the error
is raised on the same check that a late consumer already hit. That is how
`AsyncQueue` in `@runanywhere/proto-ts/streams/push` already works, and this
file is a hand-rolled second copy of it that drifted; matching it removes the
divergence rather than adding a second way to be right.
Buffered values still drain before the error, and `complete()` still ends the
stream normally.
4bb9dee to
78ec3a5
Compare
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
What is wrong
AsyncQueue.fail()resolved a parked consumer with{ done: true }and only stored the error:The stored
this.erroris only ever read bynext(). A consumer that was parked has already been handed a normal end-of-stream, itsfor awaitloop has exited, and it never callsnext()again. So whether a caller sees the failure or a clean completion depends on whether it happened to be awaiting at that instant.During streaming it almost always is. The consumer drains tokens faster than inference produces them, so parked is the normal state, and the silent path is the common one rather than the edge case.
Why it matters
streamGeneratefails both queues on a mid-stream error (RunAnywhere+TextGeneration.ts:183-184). With the queue swallowing it, iteratingstreamoreventsends as if generation had completed: the caller keeps a truncated answer and is told nothing went wrong.The error does survive on the separate
resultpromise, butLLMStreamingResultexposesstream,eventsandresultindependently, and a caller driving the token stream has no obligation to awaitresult.What this changes
Producers now only signal, and
next()re-reads the queue state, so a parked consumer raises the error on the same check a late consumer already hit.That is exactly how
AsyncQueuein@runanywhere/proto-ts/streams/push(the shared implementation used by React Native and Electron) already behaves. This file is a hand-rolled second copy that drifted from it, so this brings the clone back in line rather than inventing a third behaviour.Buffered values still drain before the error surfaces, and
complete()still ends the stream normally. Net 25 insertions, 24 deletions.Verification
Ran in
bindings/web/packages/core:vitest run: 57 files, 248 tests, all pass (including the 4 new ones)tsc --noEmit: cleaneslint src/Foundation/AsyncQueue.ts tests/unit/Foundation/AsyncQueue.test.ts --max-warnings 0: cleanThe new test is in
tests/unit/Foundation/AsyncQueue.test.ts. I confirmed it is a real regression test by stashing onlyAsyncQueue.tsand re-running: the parked-consumer case fails on unfixed source withand the other three pass on unfixed source, so they pin behaviour this change had to preserve rather than describing the fix.
The suite and typecheck need the generated proto tree, so I ran
idl/codegen/generate_ts.sh,generate_ts_convenience.py,generate_defaults_pool.pyandgenerate_streams.shfirst. None of that output is committed;git statusshows only the two files.Summary by CodeRabbit
Bug Fixes
Tests