feat(streams): a reply generated on one machine now streams to every machine - #2421
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Comment |
84a9991 to
c2372ad
Compare
5e643fe to
50c44bc
Compare
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5e643fe4da
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| .where(and( | ||
| eq(aiStreamFrames.messageId, messageId), | ||
| gte(aiStreamFrames.fromSeq, containing), | ||
| )) | ||
| .orderBy(asc(aiStreamFrames.fromSeq)); |
There was a problem hiding this comment.
For a long stream, this query materializes every remaining JSON frame row before the loop at line 201 applies MAX_TICK_BYTES; therefore the advertised 2 MB cap does not bound either the PostgreSQL result or this process's memory. A reconnect while the log approaches its 64 MB durable limit—and especially a fleet-wide reconnect across several streams—can allocate tens of megabytes per follower and cause severe memory pressure. Paginate or otherwise constrain the SQL query itself rather than stopping only after all rows have been fetched.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Correct, and embarrassing given that readFrames documents avoiding exactly this and I reintroduced it one module over — while claiming in a comment that the budget "bounds what gets fetched". It did not. Fixed in the latest push.
readFramesFrom is now two passes, the same shape readFrames uses:
- metadata —
from_seq, frame_count, byte_sizefrom the containing row onward, with aLIMIT. Three integers per row, so the result is negligible whatever the log holds. The contiguous prefix and the byte budget are both decided from this. - payload —
framesfor exactly the rows that survived that decision, bounded byfrom_seq <= $lastWanted.
The cursor seek stays as it was (max(from_seq) where from_seq <= $X), so the whole thing is still PK-indexed and there is no non-sargable predicate anywhere.
It also turns out to be cheaper than the version it replaces on the common path: a tick that finds nothing new — most of them, on a stream sitting in a long tool call — now stops after the metadata pass and never touches the frames jsonb at all.
The existing cases all passed against the unbounded query, which is the real lesson, so four new ones assert on what was fetched rather than what was returned: payload rows pulled vs. budget, the metadata LIMIT, the <= lastWanted upper bound, and zero payload queries on a quiet tick. Removing the upper bound turns 3 red; dropping the LIMIT turns 1 red.
| finishFollower(messageId, follower, { | ||
| aborted: state.aborted, | ||
| // `empty` is reported only for a log with no rows AT ALL — never for a failed read (see | ||
| // `FrameCursorRead.empty`), so a DB blip cannot masquerade as a released log and send | ||
| // every viewer into a needless reload. | ||
| truncated: tail.truncated || tail.empty, |
There was a problem hiding this comment.
When startFrameLogWriter disables itself after a batch insert failure or exhausting its durable budget, the table retains a nonempty contiguous prefix with no hole marker. A remote follower that already delivered that prefix eventually reaches this terminal branch, where tail.empty and tail.truncated are both false, so it sends a clean completion; because the client has delivered frames, joinFailed also remains false and the final persisted message is never reloaded. A surviving log must not be treated as complete without a durable completion/count marker, or this path should request a reload whenever completeness cannot be proven.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Agreed — a non-empty log is not a complete one, and the three early-stop paths in frame-log-writer all leave a hole-free prefix that read as a clean end. Worse than you describe, in fact: because frames HAD been delivered, joinFailed stayed false too, so nothing reloaded and there was no signal anywhere that the reply was short. Fixed in the latest push.
On the two options you offered — a durable completion marker, or requesting a reload whenever completeness cannot be proven — there turns out to be a third that needs no new schema: the stream already carries its own terminator. pumpSdkStreamToChannel appends every SDK frame verbatim, so a log containing finish (normal end), abort (stopped), or the pumps synthetic error frame (SDK stream threw, after which the pump stops reading) contains everything the generation produced. The follower now tracks whether one has arrived and requires it before ending cleanly; otherwise it ends truncated, which the route turns into reload and the client turns into joinFailed: true.
I did look at raw_parts_count first — it is the comparator materializeInterruptedStream uses for this same question — but the terminal write zeroes it, so it is gone by the time a follower needs it.
Tracked as "have we ever seen a terminator" rather than "is the last frame one": a trailing message-metadata frame would defeat the stricter test, and the two directions are not symmetric — failing to recognise completeness costs one needless reload, while falsely claiming it is the silent truncation you found.
Four new cases (surviving-but-unterminated prefix, abort as proof, error as proof, terminator seen on an earlier tick than the terminal branch). Removing the proof turns 1 red; narrowing the terminator set to finish alone turns 2 red; never remembering the terminator turns 4 red.
50c44bc to
fa8dbe0
Compare
c2372ad to
6414dd4
Compare
|
Pushed addressing both P1 findings — the unbounded payload query and the missing completeness proof. Worth calling out on the second one: there was a third option beyond the two suggested (durable completion marker / always reload). The stream already carries its own terminator — the pump appends every SDK frame verbatim, so a log containing On the first: the pre-existing cursor tests all passed against the unbounded query, which is the actual lesson. The four new cases assert on what was fetched rather than what was returned. Local verification: |
…machine
The join could classify a cross-instance stream (previous leaf) but had no
frames to serve it. Cross-instance delivery was `stream-join-poll-fallback` —
whole-array `parts` snapshot replacements at roughly 1200ms. This tails the
durable frame log instead: incremental frames at ~300ms.
THE ROUTE BODY IS NOT FORKED, and that is the design. A follower produces a real
`StreamChannel`, so `subscribe` / `onFrame` / `onEnd`, the SSE encoding, the 20s
ping, the 5s permission recheck, teardown, and the `overflow` / `resumeFromSeq`
semantics are literally the same code for both sources. The alternative — an
`if (remote)` branch in the route — meant two copies of five behaviours, one of
which nothing local exercises.
- `frame-log-cursor.ts` — `readFramesFrom({messageId, fromSeq})`. TWO CURSORS,
not the schema docblock's `from_seq + frame_count > $X`: that predicate is an
expression over two columns, so it is not sargable against the
`(message_id, from_seq)` PK and would scan the whole log on every tick, per
follower. Instead: `max(from_seq) WHERE from_seq <= $X` seeks the containing
row, then `from_seq >= that` range-scans forward, and the overshoot is sliced.
Contiguity is enforced exactly as `readFrames` does — a row whose `from_seq`
is not where the walk expected it is a HOLE, and the walk STOPS and reports
`truncated`. It never skips. Folding across a hole streams a confidently-wrong
message to a live user who cannot tell.
- `remote-frame-follower.ts` — refcounted per messageId, so co-located tabs share
one poller rather than making Postgres load scale with viewers. Starts at seq 0
so any subscriber cursor is servable (a channel starting at one reader's cursor
answers `overflow` to every reader below it, and `overflow` is a reseed, not a
resume). Polls at 250ms — matched to the writer's 64-frame / 200ms / 256KB
flush, since polling faster cannot see fresher data — backing off to 1s after 4
empty ticks and resetting on any frame. The session status is read ONLY on
ticks that found nothing. Lingers ~5s past the last reader so a reconnect does
not cold-start, plus a `STREAM_MAX_LIFETIME_MS` eviction.
- THREE HONEST ANSWERS AT TERMINAL, because frames are deleted on the terminal
write: log present → serve the rest, then a clean `done`; log released →
`{done, aborted, reload}`; row gone → the same, never a clean end. A short
bubble that looks whole is the failure being avoided.
- `reload` is a new wire field, distinct from `resumeFromSeq` and not to be
collapsed into it: `resumeFromSeq` says "ask again from here", `reload` says
there is nowhere to resume from. It plumbs `ChannelEnd.truncated` →
`consumeStreamJoin` → `endSession(joinFailed: true)`, which already triggers
the durable-message reload.
- `streamSessionRegistry`'s success path ended `joinFailed: false`
unconditionally. Now `!session.delivered || result.reload`. A join that
resolved having delivered NOTHING — it opened, the stream was already over —
is not authoritative whatever its terminal frame said, and consumers were
keeping a debounced seed snapshot that can be shorter than the finished reply.
- `X-Stream-Join-Source: local | remote | terminal`. An N=2 smoke test has no
other way to PROVE it exercised the follower rather than getting lucky with the
load balancer; in production the remote share should sit near (N-1)/N, which
makes it a free load-balancer monitor.
Out of scope and untouched, per the brief: `stream-join-poll-fallback` stays (it
remains a fallback until N=2 has baked), the `parts` / `rawPartsCount` columns
stay, and the cross-instance abort path is unchanged.
Twelve seeded mutations verified red, including skipping a hole instead of
stopping, dropping the containing-row slice, reporting a read failure as an empty
log, ending a released log cleanly, treating an unreadable status as terminal,
never resetting the poll backoff, a non-idempotent release, removing the linger,
and both `joinFailed` regressions. Two follower tests were strengthened after
their first versions failed to discriminate.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SQwbEoZweBEBgsGq9wnqXj
…ore ending a stream cleanly Two P1 review findings on the follower. 1. THE TICK BUDGET BOUNDED NOTHING. `readFramesFrom` selected every row from the cursor onward and applied `MAX_TICK_BYTES` while walking the result — so the driver had already materialized and parsed the entire remaining log, up to the writer's 64 MB per-stream ceiling, before any JavaScript ceiling could look at it. A reconnect against a long stream, and especially a fleet-wide reconnect across many streams, allocated that per follower. `readFrames` documents avoiding exactly this; this module reintroduced it and then claimed in a comment that it did not. Now two passes, the same shape `readFrames` uses: metadata only (three integers per row, `LIMIT`ed) decides the contiguous, budgeted prefix, then the payload query fetches `frames` for exactly those rows, bounded by `from_seq <= $lastWanted`. It is also CHEAPER on the common path than the version it replaces: a tick that finds nothing new — most of them, on a stream inside a tool call — now stops after the metadata pass and never touches the `frames` jsonb at all. 2. A SURVIVING LOG IS NOT A COMPLETE ONE. `frame-log-writer` stops early in three documented ways (its pre-write delete failing, a batch insert failing, exhausting the durable budget), and each leaves a valid, contiguous, HOLE-FREE prefix. A follower that delivered such a prefix and then found the row terminal saw `empty: false` and `truncated: false`, sent a clean `done`, and — because frames HAD been delivered — left `joinFailed` false too. The user kept a truncated reply that nothing ever reloaded, with no signal anywhere that it was short. There is no length marker to check, and `raw_parts_count` (the comparator `materializeInterruptedStream` uses for this exact question) is zeroed by the terminal write. But the stream carries its own terminator: the pump appends every SDK frame verbatim, so a log containing `finish`, `abort`, or the pump's synthetic `error` contains everything the generation produced. The follower now requires one before ending cleanly, and asks for a reload otherwise. Tracked as "have we ever seen one" rather than "is the last frame one" — a trailing `message-metadata` would defeat the stricter test, and the two directions are not symmetric: failing to recognise completeness costs one needless reload, falsely claiming it is the silent truncation above. The cursor's existing cases passed against the unbounded query, so four new ones assert on what was FETCHED rather than what was returned. Seven seeded mutations verified red across both fixes, including removing the `<= lastWanted` bound, dropping the metadata LIMIT, removing the completeness proof, and narrowing the terminator set. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SQwbEoZweBEBgsGq9wnqXj
fa8dbe0 to
ff6f770
Compare
6414dd4 to
20552d2
Compare
1 participant
Stack
This is one of three stacked PRs, to be reviewed and merged bottom-up:
masterpu/reap-claimpu/stream-join-contextEach PR is a strict improvement on its own, but the user-visible cross-instance join only works once all three land. Workstream A of "make AI streaming safe at more than one machine" — the goal is raising
pagespace-webfrommin_machines_running = 1to 2.The join could classify a cross-instance stream (previous leaf) but had no
frames to serve it. Cross-instance delivery was
stream-join-poll-fallback—whole-array
partssnapshot replacements at roughly 1200ms. This tails thedurable frame log instead: incremental frames at ~300ms.
THE ROUTE BODY IS NOT FORKED, and that is the design. A follower produces a real
StreamChannel, sosubscribe/onFrame/onEnd, the SSE encoding, the 20sping, the 5s permission recheck, teardown, and the
overflow/resumeFromSeqsemantics are literally the same code for both sources. The alternative — an
if (remote)branch in the route — meant two copies of five behaviours, one ofwhich nothing local exercises.
frame-log-cursor.ts—readFramesFrom({messageId, fromSeq}). TWO CURSORS,not the schema docblock's
from_seq + frame_count > $X: that predicate is anexpression over two columns, so it is not sargable against the
(message_id, from_seq)PK and would scan the whole log on every tick, perfollower. Instead:
max(from_seq) WHERE from_seq <= $Xseeks the containingrow, then
from_seq >= thatrange-scans forward, and the overshoot is sliced.Contiguity is enforced exactly as
readFramesdoes — a row whosefrom_seqis not where the walk expected it is a HOLE, and the walk STOPS and reports
truncated. It never skips. Folding across a hole streams a confidently-wrongmessage to a live user who cannot tell.
remote-frame-follower.ts— refcounted per messageId, so co-located tabs shareone poller rather than making Postgres load scale with viewers. Starts at seq 0
so any subscriber cursor is servable (a channel starting at one reader's cursor
answers
overflowto every reader below it, andoverflowis a reseed, not aresume). Polls at 250ms — matched to the writer's 64-frame / 200ms / 256KB
flush, since polling faster cannot see fresher data — backing off to 1s after 4
empty ticks and resetting on any frame. The session status is read ONLY on
ticks that found nothing. Lingers ~5s past the last reader so a reconnect does
not cold-start, plus a
STREAM_MAX_LIFETIME_MSeviction.write: log present → serve the rest, then a clean
done; log released →{done, aborted, reload}; row gone → the same, never a clean end. A shortbubble that looks whole is the failure being avoided.
reloadis a new wire field, distinct fromresumeFromSeqand not to becollapsed into it:
resumeFromSeqsays "ask again from here",reloadsaysthere is nowhere to resume from. It plumbs
ChannelEnd.truncated→consumeStreamJoin→endSession(joinFailed: true), which already triggersthe durable-message reload.
streamSessionRegistry's success path endedjoinFailed: falseunconditionally. Now
!session.delivered || result.reload. A join thatresolved having delivered NOTHING — it opened, the stream was already over —
is not authoritative whatever its terminal frame said, and consumers were
keeping a debounced seed snapshot that can be shorter than the finished reply.
X-Stream-Join-Source: local | remote | terminal. An N=2 smoke test has noother way to PROVE it exercised the follower rather than getting lucky with the
load balancer; in production the remote share should sit near (N-1)/N, which
makes it a free load-balancer monitor.
Out of scope and untouched, per the brief:
stream-join-poll-fallbackstays (itremains a fallback until N=2 has baked), the
parts/rawPartsCountcolumnsstay, and the cross-instance abort path is unchanged.
Twelve seeded mutations verified red, including skipping a hole instead of
stopping, dropping the containing-row slice, reporting a read failure as an empty
log, ending a released log cleanly, treating an unreadable status as terminal,
never resetting the poll backoff, a non-idempotent release, removing the linger,
and both
joinFailedregressions. Two follower tests were strengthened aftertheir first versions failed to discriminate.
Co-Authored-By: Claude Opus 5 (1M context) noreply@anthropic.com
Claude-Session: https://claude.ai/code/session_01SQwbEoZweBEBgsGq9wnqXj
Stacked on the join-context PR.
🤖 Generated with Claude Code
https://claude.ai/code/session_01SQwbEoZweBEBgsGq9wnqXj