fix(streams): ephemeral broadcasts over the NOTIFY cap auto-degrade to durable; coalescer flush errors reach ErrorReporter

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 - **SSE delivery no longer strips newlines from broadcast payloads — multiline payloads are framed as consecutive `data:` lines per the SSE spec (issue #392).** `Streams::Envelope.message` collapsed `\r`/`\n` in the payload to nothing before writing the single `data:` line, silently corrupting any whitespace-significant broadcast (pre-formatted `<pre>` content, textarea seeds, JSON-in-data frames) on **both** the ephemeral and durable delivery paths — HTML's whitespace tolerance is why it went unnoticed. A multiline payload is now split on `\r\n`/`\r`/`\n` into consecutive `data:` lines, which EventSource clients rejoin with `\n`, making delivery lossless (a trailing newline survives via an empty final `data:` line; `\r` variants normalize to `\n` — SSE line terminators cannot be carried raw). The original injection defense is preserved: every payload line carries the `data:` prefix followed by one space, so a crafted payload still cannot forge `id:`/`event:` fields, and single-line fields (event names, comments) still strip newlines. The `<pgbus-stream-source>` element's fetch-path parser had the matching client-side bug — it joined `data:` lines without `\n` *and* `trim()`ed payload whitespace — and now follows EventSource semantics (join with `\n`, strip only the single leading space). Refs #392.
 
+- **Ephemeral broadcasts over the PG NOTIFY payload cap no longer fail — loudly on the sync path, silently in the coalescer — they auto-degrade to a durable publish (issue #391).** Ephemeral frames ride the NOTIFY payload itself, which PostgreSQL caps below 8000 bytes. Any rendered-component broadcast (a progress card with Tailwind classes easily exceeds it) previously raised `PGMQ::Errors::ConnectionError: … payload string too long` — an error class that sent diagnosis toward the connection, not the payload — and on the `coalesce:` path that raise happened inside the coalescer's flush thread, reaching no caller, no ErrorReporter, no log: small frames delivered, big frames vanished, and the operator saw "SSE works but updates don't arrive". Three changes: **(1)** `Stream#broadcast` now measures the wrapped JSON before the NOTIFY and publishes an over-budget frame durably instead (payload stored in PGMQ, the queue's insert trigger fires the NOTIFY as a bare wake on the same channel the subscriber already LISTENs on) — delivery semantics preserved on both the sync and coalesced paths, warn-logged and instrumented (`pgbus.stream.broadcast` with `ephemeral_fallback: true`). **(2)** Direct `Client#notify_stream` callers get publish-time validation: a typed `Pgbus::Streams::PayloadTooLarge` raised at the call site for payloads exceeding `Pgbus::Client::NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES` (7999 bytes, the largest accepted payload), naming the stream, the byte count, and the durable-mode escape hatch. **(3)** The coalescer's flush thread routes every flush error through `ErrorReporter` (same report-don't-log reasoning as #352) — a background thread swallowing delivery failures is invisible to APM by construction. ⚠️ **Upgrade note for 0.13 installs:** `streams_default_broadcast_mode` defaults to `:ephemeral`, and that default is a **behavior change** for apps broadcasting rendered components (what `broadcast_render`-style usage produces) — before this fix, any frame over ~8KB was silently lost or misdiagnosed. Durable is the right mode for turbo-stream UI regardless (since-id replay needs the archive): pin `config.streams_default_broadcast_mode = :durable`, or use `streams_durable_patterns` for the streams that need it; the auto-fallback now covers whatever stays ephemeral. Refs #391.
+
 - **Queue-age metrics no longer count vt-parked (scheduled/retrying) messages — one delayed job stops reading as a degraded queue (issue #389).** ⚠️ **Behavior change on the AppSignal `pgbus_queue_latency` gauge.** pgmq's `oldest_msg_age_sec` is computed from `enqueued_at` and ignores `vt`, but a job enqueued with `wait:` or parked on a long retry backoff lives in the queue table with a future `vt` — that *is* the delayed-delivery mechanism. So a single parked message made the age metric grow at wall-clock rate for hours on an otherwise drained queue, and any latency alert thresholding on it fired continuously ("oldest message is 17045s old" on a healthy queue with depth 1, `read_ct` 0). Every metrics surface now also exposes **`oldest_claimable_age_sec`** — `now() - min(vt)` over rows with `vt <= now()`, i.e. the age of the oldest message actually *eligible for pickup*: an immediately-enqueued message contributes from enqueue time (matching the old number on a plain backlog), a scheduled/backoff-parked message contributes nothing until due, an in-flight message (vt pushed forward) is excluded, and nil means "no claimable backlog" even when the table is non-empty. Surfaces: `Web::DataSource` (dashboard, JSON API, MCP `pgbus_queues` tool), a new Prometheus gauge `pgbus_queue_oldest_claimable_age_seconds`, a new AppSignal gauge `pgbus_queue_oldest_claimable_age_seconds`, `Pgbus::Client#oldest_claimable_ages` (raw-SQL reader, since pgmq's `metrics_result` type is frozen upstream), and a CLAIMABLE column in `pgbus queues`. The AppSignal **`pgbus_queue_latency` gauge now derives from the claimable age** and always emits — `(claimable_age || 0) * 1000`, 0 = no claimable backlog — so existing latency alerts stop false-firing with no dashboard changes; the raw `pgbus_queue_oldest_message_age_seconds` gauge keeps its enqueue-time semantics everywhere. The dashboard queue tables additionally split depth into **Parked** (`depth − visible`) and show the claimable age in place of the raw age, so a queue holding only backoff retries reads visibly healthy. Refs #389.
 
 ### Added

lib/pgbus/client/notify_stream.rb

@@ -14,15 +14,24 @@ class Client
     # no orphan tables.
     #
     # The payload is JSON-serialized into the NOTIFY's optional payload
-    # parameter (max 8000 bytes in Postgres). Broadcasts exceeding this
-    # limit will raise a PG::ProgramLimitExceeded error — callers needing
-    # large payloads should use durable mode (which inserts into PGMQ).
+    # parameter. Postgres caps NOTIFY payloads at < 8000 bytes; oversized
+    # payloads raise a typed Pgbus::Streams::PayloadTooLarge here, at the
+    # call site, instead of surfacing as a misleading
+    # PGMQ::Errors::ConnectionError ("payload string too long") from deep
+    # inside the driver (issue #391). Callers needing large payloads should
+    # use durable mode (which inserts into PGMQ).
     module NotifyStream
+      # PostgreSQL rejects NOTIFY payloads of 8000 bytes or more
+      # ("payload string too long"), so 7999 is the largest deliverable
+      # payload.
+      NOTIFY_PAYLOAD_LIMIT_BYTES = 7999
+
       def notify_stream(stream_name, payload)
         full_name = config.queue_name(stream_name)
         sanitized = QueueNameValidator.sanitize!(full_name)
         channel = "pgmq.q_#{sanitized}.INSERT"
         json = payload.is_a?(String) ? payload : JSON.generate(payload)
+        validate_notify_payload_size!(stream_name, json)
 
         Instrumentation.instrument("pgbus.stream.notify", stream: stream_name, bytes: json.bytesize) do
           with_stale_connection_retry do
@@ -34,6 +43,19 @@ def notify_stream(stream_name, payload)
           end
         end
       end
+
+      private
+
+      def validate_notify_payload_size!(stream_name, json)
+        return if json.bytesize <= NOTIFY_PAYLOAD_LIMIT_BYTES
+
+        raise Pgbus::Streams::PayloadTooLarge,
+              "Ephemeral broadcast on stream #{stream_name.inspect} is #{json.bytesize} bytes; " \
+              "PostgreSQL caps NOTIFY payloads at #{NOTIFY_PAYLOAD_LIMIT_BYTES} bytes. " \
+              "Use durable mode for large payloads (payload stored in PGMQ, NOTIFY as wake) — " \
+              "e.g. broadcast(..., durable: true), a streams_durable_patterns match, or " \
+              "streams_default_broadcast_mode = :durable."
+      end
     end
   end
 end

lib/pgbus/streams.rb

@@ -15,6 +15,18 @@ module Streams
     # this specifically can rescue Pgbus::Streams::StreamNameTooLong.
     class StreamNameTooLong < ArgumentError; end
 
+    # Raised when an ephemeral broadcast's JSON payload exceeds PostgreSQL's
+    # NOTIFY payload budget (< 8000 bytes). Ephemeral frames ride the NOTIFY
+    # itself, so the cap is a hard PostgreSQL limit — durable mode (payload
+    # stored in PGMQ, NOTIFY as a bare wake) has no such cap.
+    #
+    # Stream#broadcast never raises this: an oversized ephemeral frame
+    # auto-degrades to a durable publish (issue #391). The error exists for
+    # direct Client#notify_stream callers, where the previous failure mode
+    # was a misleading PGMQ::Errors::ConnectionError ("payload string too
+    # long") that pointed diagnosis at the connection instead of the payload.
+    class PayloadTooLarge < Pgbus::Error; end
+
     # The default SSE `event:` name for a broadcast frame. Turbo's
     # StreamObserver consumes frames the client re-dispatches as the
     # `message` DOM event; the client maps this SSE event name to
@@ -261,11 +273,47 @@ def self.validate_name_length!(name, streamables)
 
       private
 
+      # Ephemeral frames ride the NOTIFY payload itself, which PostgreSQL
+      # caps below 8000 bytes. A frame over the cap auto-degrades to a
+      # durable publish (issue #391): payload stored in PGMQ, the queue's
+      # insert trigger fires the NOTIFY as a bare wake on the same channel
+      # the subscriber already LISTENs on — delivery semantics preserved,
+      # cap irrelevant. The JSON is generated once here and passed
+      # pre-serialized to notify_stream so the size check costs no extra
+      # allocation on the hot path.
       def broadcast_ephemeral(wrapped)
-        @client.notify_stream(@name, wrapped)
+        json = JSON.generate(wrapped)
+        return durable_fallback(wrapped, json.bytesize) if json.bytesize > Client::NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES
+
+        @client.notify_stream(@name, json)
         nil
       end
 
+      # The durable degrade path for an oversized ephemeral frame. Stays
+      # fire-and-forget like the ephemeral path it replaces: no after_commit
+      # deferral (pg_notify runs on the PGMQ pool connection, outside the
+      # request's AR transaction, so the ephemeral path never deferred
+      # either). Returns the msg_id like any durable publish.
+      def durable_fallback(wrapped, bytes)
+        Pgbus.logger.warn do
+          "[Pgbus::Streams] ephemeral broadcast on #{@name.inspect} is #{bytes} bytes " \
+            "(NOTIFY cap is #{Client::NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES}); " \
+            "publishing durably instead. Consider durable mode for this stream " \
+            "(streams_durable_patterns or durable: true) to skip this check."
+        end
+        ensure_queue!
+        instrument_payload = {
+          stream: @name,
+          visible_to: wrapped["visible_to"],
+          deferred: false,
+          bytes: wrapped["html"].bytesize,
+          ephemeral_fallback: true
+        }
+        Instrumentation.instrument("pgbus.stream.broadcast", instrument_payload) do
+          @client.send_stream_message(@name, wrapped)
+        end
+      end
+
       # Submits a frame to the process-wide coalescer instead of
       # broadcasting now. Requires a target (the dedupe key — there's no
       # way to last-write-win without one). The window is `coalesce` in ms

lib/pgbus/streams/coalescer.rb

@@ -73,6 +73,12 @@ def flush_key(key, stream_name, target)
         return unless entry
 
         @flush.call(stream_name: stream_name, target: target, payload: entry.payload, opts: entry.opts)
+      rescue StandardError => e
+        # The flush runs on the scheduler's thread — a raise here reaches no
+        # caller, so a swallowed error is invisible to APM by construction
+        # (issue #391: an oversized ephemeral frame died here without a
+        # trace). Route through ErrorReporter so configured reporters see it.
+        ErrorReporter.report(e, { component: "streams.coalescer", stream: stream_name, target: target })
       end
 
       # Default scheduler backed by Concurrent::ScheduledTask. Kept as a

spec/pgbus/client/notify_stream_spec.rb

@@ -69,6 +69,50 @@ def initialize(*args, **kwargs); end
       )
     end
 
+    context "when the payload exceeds the PG NOTIFY budget" do
+      let(:oversized) { { "html" => "<div>#{"x" * 9000}</div>" } }
+
+      it "raises a typed Pgbus::Streams::PayloadTooLarge at the call site" do
+        expect do
+          client.notify_stream("chat", oversized)
+        end.to raise_error(Pgbus::Streams::PayloadTooLarge, /\d+ bytes.*7999/m)
+      end
+
+      it "names the stream and suggests durable mode in the message" do
+        expect do
+          client.notify_stream("chat", oversized)
+        end.to raise_error(Pgbus::Streams::PayloadTooLarge, /chat.*durable/m)
+      end
+
+      it "does not attempt the NOTIFY" do
+        begin
+          client.notify_stream("chat", oversized)
+        rescue Pgbus::Streams::PayloadTooLarge
+          nil
+        end
+        expect(raw_conn).not_to have_received(:exec_params)
+      end
+
+      it "is a Pgbus::Error so blanket pgbus rescues keep working" do
+        expect(Pgbus::Streams::PayloadTooLarge.ancestors).to include(Pgbus::Error)
+      end
+
+      it "measures bytes, not characters (multibyte payloads)" do
+        # 4500 two-byte chars = 4500 chars but 9000 bytes of HTML.
+        multibyte = { "html" => "é" * 4500 }
+        expect do
+          client.notify_stream("chat", multibyte)
+        end.to raise_error(Pgbus::Streams::PayloadTooLarge)
+      end
+
+      it "accepts a payload exactly at the budget" do
+        # {"html":""} wrapper is 11 bytes; fill to exactly 7999.
+        at_limit = { "html" => "x" * (7999 - 11) }
+        expect { client.notify_stream("chat", at_limit) }.not_to raise_error
+        expect(raw_conn).to have_received(:exec_params)
+      end
+    end
+
     context "with stale pgmq connection recovery" do
       before do
         real_pgmq_connection_error

spec/pgbus/streams/coalescer_spec.rb

@@ -112,4 +112,42 @@ def submit(stream: "chat", target: "cursor", payload: "x", window_ms: 50, **opts
     scheduler.run_all
     expect(flushed.map { |f| f[:target] }).to eq(["cursor"])
   end
+
+  # Issue #391: the flush runs on the scheduler's thread — a raise there
+  # never reaches any caller, so a swallowed error is invisible to APM by
+  # construction. Errors must route through ErrorReporter (same reasoning
+  # as the StreamApp report-don't-log fix in #352).
+  describe "flush errors" do
+    let(:boom) { StandardError.new("payload string too long") }
+    let(:flush) { ->(**) { raise boom } }
+
+    it "routes a raising flush through ErrorReporter with stream/target context" do
+      allow(Pgbus::ErrorReporter).to receive(:report)
+
+      submit(stream: "chat", target: "cursor", payload: "a")
+      scheduler.run_all
+
+      expect(Pgbus::ErrorReporter).to have_received(:report).with(
+        boom, hash_including(
+                component: "streams.coalescer", stream: "chat", target: "cursor"
+              )
+      )
+    end
+
+    it "does not re-raise out of the flush" do
+      allow(Pgbus::ErrorReporter).to receive(:report)
+      submit(payload: "a")
+      expect { scheduler.run_all }.not_to raise_error
+    end
+
+    it "keeps the key usable for the next window after a failed flush" do
+      allow(Pgbus::ErrorReporter).to receive(:report)
+      submit(payload: "a")
+      scheduler.run_all
+
+      # A new submit after the failed flush schedules a fresh window.
+      submit(payload: "b")
+      expect(scheduler.scheduled.size).to eq(1)
+    end
+  end
 end

spec/pgbus/streams/ephemeral_broadcast_spec.rb

@@ -34,19 +34,19 @@
       expect(client).not_to have_received(:send_stream_message)
     end
 
-    it "sends a PG NOTIFY with the payload" do
+    it "sends a PG NOTIFY with the pre-serialized payload" do
       stream.broadcast("<turbo-stream>X</turbo-stream>")
       expect(client).to have_received(:notify_stream).with(
         "chat",
-        { "html" => "<turbo-stream>X</turbo-stream>" }
+        JSON.generate({ "html" => "<turbo-stream>X</turbo-stream>" })
       )
     end
 
     it "includes visible_to in the NOTIFY payload when specified" do
       stream.broadcast("<turbo-stream>X</turbo-stream>", visible_to: :admin_only)
       expect(client).to have_received(:notify_stream).with(
         "chat",
-        { "html" => "<turbo-stream>X</turbo-stream>", "visible_to" => "admin_only" }
+        JSON.generate({ "html" => "<turbo-stream>X</turbo-stream>", "visible_to" => "admin_only" })
       )
     end