fix(streams): ephemeral broadcasts over the NOTIFY cap auto-degrade to durable; coalescer flush errors reach ErrorReporter - #394
Merged
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI (base), Organization UI (inherited) Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughEphemeral broadcasts validate serialized NOTIFY payloads, use durable PGMQ delivery when oversized, and return durable message IDs. Direct oversized notifications raise a typed error. Coalescer flush failures are reported through ChangesEphemeral delivery safeguards
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant Stream
participant NotifyStream
participant PGMQ
participant ErrorReporter
Stream->>NotifyStream: Send serialized ephemeral payload
NotifyStream-->>Stream: Return success or PayloadTooLarge
Stream->>PGMQ: Ensure queue and publish oversized payload
PGMQ-->>Stream: Return durable message ID
Stream->>ErrorReporter: Report coalescer flush failures with context
Possibly related PRs
Suggested labels: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@CHANGELOG.md`:
- Line 5: Correct the CHANGELOG description of the PayloadTooLarge threshold to
state that 7,999 bytes is the maximum accepted payload, with the exception
raised only when the payload exceeds that limit.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 14d1a597-0ac5-48f0-b2a5-8ec006b88013
📒 Files selected for processing (8)
CHANGELOG.mdlib/pgbus/client/notify_stream.rblib/pgbus/streams.rblib/pgbus/streams/coalescer.rbspec/pgbus/client/notify_stream_spec.rbspec/pgbus/streams/coalescer_spec.rbspec/pgbus/streams/ephemeral_broadcast_spec.rbspec/pgbus/streams/ephemeral_overflow_spec.rb
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@CHANGELOG.md`:
- Line 5: Update the upgrade note in the changelog to limit the durable-mode
recommendation to Turbo Stream UI that requires since-id replay or archived
messages. Remove the unconditional claim that durable mode is appropriate for
all Turbo Stream UI, while preserving the existing configuration examples and
auto-fallback guidance.
- Line 5: Update the changelog’s fallback observability wording to state that
oversized ephemeral frames use durable fallback on both synchronous and
coalesced broadcast paths, with warning logs and instrumentation; only direct
Client#notify_stream calls raise PayloadTooLarge. Also retain that coalescer
flush failures are reported through ErrorReporter, rather than describing the
coalescer fallback as silent.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: eefa0818-5dee-4338-afdd-877a72c8a9b0
📒 Files selected for processing (1)
CHANGELOG.md
mhenrixon
added
streaming
…o durable; coalescer flush errors reach ErrorReporter Ephemeral frames ride the PG NOTIFY payload, capped below 8000 bytes. An oversized frame raised a misleading PGMQ ConnectionError on the sync path and vanished silently inside the coalescer's flush thread. Now: Stream#broadcast measures the wrapped JSON and publishes over-budget frames durably (PGMQ payload, trigger NOTIFY as wake — same channel, delivery preserved); Client#notify_stream raises a typed Pgbus::Streams::PayloadTooLarge for direct callers; the coalescer routes flush errors through ErrorReporter so APM sees them. - notify_stream_spec: typed raise, byte-not-char measurement, at-limit boundary, no NOTIFY attempted - ephemeral_overflow_spec: durable fallback (queue ensured, msg_id returned, warn log, instrumentation), boundary incl. metadata overflow, small-frame NOTIFY path - coalescer_spec: raising flush reports via ErrorReporter with stream/target context, no re-raise, key reusable next window - [x] bundle exec rubocop passes - [x] bundle exec rspec passes (only pre-existing i18n baseline failures) Refs #391
…sced frames The oversized-frame durable fallback runs at flush, on the coalescer thread, where the request's transaction is structurally invisible — so gating must be (and is) decided at submit time by the requested mode. Pins the ephemeral half: fire-and-forget submits immediately even inside an open transaction, so the fallback inherits the ephemeral contract the caller chose. Refs #391.
…rejection threshold
mhenrixon
force-pushed
the
issue-391-ephemeral-payload-cap
branch
from
August 4, 2026 20:49
ef12b0e to
2286f3f
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. |
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@lib/pgbus/streams/coalescer.rb`:
- Around line 76-81: Update the rescue path around the coalescer flush to
persist failed durable frames in pgbus_failed_events before or alongside
ErrorReporter.report. Include sufficient stream identity and buffered payload
context for retry or replay, while preserving ErrorReporter reporting; if an
existing durable failure helper is available, reuse it rather than adding a
parallel mechanism.
In `@spec/pgbus/streams_spec.rb`:
- Around line 221-225: Update the expectation in the “submits an ephemeral
coalesced frame immediately, even inside the transaction” example to require
that coalescer.submit receives durable: false in addition to target: "t". Keep
the existing immediate-submission assertion unchanged.
In `@spec/pgbus/streams/coalescer_spec.rb`:
- Around line 124-132: Update the ErrorReporter expectation in the “routes a
raising flush through ErrorReporter with stream/target context” example to
assert component: "streams.coalescer" alongside the existing stream and target
context, ensuring the complete reporting context is verified.
In `@spec/pgbus/streams/ephemeral_overflow_spec.rb`:
- Around line 52-56: Update the warning assertion in the “warn-logs the fallback
with stream and byte count” example to execute the supplied block and verify the
warning message includes the stream name and serialized byte count, not merely
that Pgbus.logger.warn was called.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: e207e607-8ca0-4d0b-b297-302b6b7a7df3
📒 Files selected for processing (9)
CHANGELOG.mdlib/pgbus/client/notify_stream.rblib/pgbus/streams.rblib/pgbus/streams/coalescer.rbspec/pgbus/client/notify_stream_spec.rbspec/pgbus/streams/coalescer_spec.rbspec/pgbus/streams/ephemeral_broadcast_spec.rbspec/pgbus/streams/ephemeral_overflow_spec.rbspec/pgbus/streams_spec.rb
- assert durable: false rides the ephemeral coalesced submit - assert component: "streams.coalescer" in ErrorReporter context - assert the fallback warning carries stream name and byte count
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Production repro (pgbus 0.13.x,
streams_default_broadcast_modeleft at its:ephemeraldefault): any rendered-component broadcast over the ~8KB PG NOTIFY payload cap raised a misleadingPGMQ::Errors::ConnectionErroron the sync path — and on thecoalesce:path the raise happened in the coalescer's flush thread, reaching no caller, no ErrorReporter, no log. Small frames delivered, big frames vanished.Three changes:
Stream#broadcast(lib/pgbus/streams.rb) measures the wrapped JSON before the NOTIFY. An over-budget frame publishes durably instead: payload stored in PGMQ, the queue's insert trigger fires the NOTIFY as a bare wake on the samepgmq.q_<name>.INSERTchannel the subscriber already LISTENs on — delivery preserved on both the sync and coalesced paths, zero dispatcher changes needed. Warn-logged and instrumented (pgbus.stream.broadcastwithephemeral_fallback: true). The JSON is generated once and passed pre-serialized tonotify_stream, so the size check adds no allocation to the hot path.Client#notify_stream(lib/pgbus/client/notify_stream.rb) raisesPgbus::Streams::PayloadTooLarge(aPgbus::Error) at the call site for direct callers, naming the stream, byte count, cap (NOTIFY_PAYLOAD_LIMIT_BYTES = 7999), and the durable-mode escape hatch — instead of the driver's "connection error" that sent diagnosis the wrong way.Coalescer#flush_key(lib/pgbus/streams/coalescer.rb) routes any flush error throughErrorReporterwith stream/target context (same report-don't-log reasoning as Dedicated LISTEN connections (streamer, worker NotifyListener) fail under connection_guc_mode = :session —invalid connection option "variables"#352); the key stays usable for the next window.CHANGELOG gains the fix entry plus a⚠️ upgrade callout: the
:ephemeraldefault is a behavior change for installs broadcasting rendered components — pin:durable(right for turbo-stream UI anyway; since-id replay needs the archive) or rely on the fallback for what stays ephemeral.Closes #391
Test plan
bundle exec rspec spec/pgbus/streams/ spec/pgbus/client/ spec/pgbus/web/streamer/— 742 examples, 0 failuresbundle exec rubocopclean on all changed filesPgbus.stream("probe").broadcast("<div>#{"x" * 9000}</div>", target: "t")now delivers (durable fallback) on both sync andcoalesce: truepathsDeviations & judgment calls
PayloadTooLargeis kept as a guard for directClient#notify_streamcallers.after_commitdeferral — matching the fire-and-forget timing of the ephemeral path it replaces (pg_notify runs on the PGMQ pool connection, outside the request's AR transaction, so ephemeral never deferred either).msg_idwhere ephemeral broadcasts return nil — strictly more information, matchesbroadcast's documented contract.Stream#broadcast_ephemeralnow passes the pre-serialized JSON string tonotify_stream(which already accepted strings) — oneJSON.generateper broadcast instead of two; two existing spec expectations updated from hash to JSON-string args.ErrorReporter.reportlives inCoalescer#flush_key(the background-thread boundary), not only the streams flush lambda, so any injected flush gets APM visibility.[Unreleased]block (no## [0.13.x]headers exist), so thedocs/) not touched: it never documented the payload cap or mode default; the issue asked for CHANGELOG/upgrade-guide coverage, which the CHANGELOG entry carries.NOTIFY_PAYLOAD_LIMIT_BYTES = 7999: Postgres rejects payloads of 8000 bytes or more, so 7999 is the largest deliverable payload.Summary by CodeRabbit
Bug Fixes
Reliability