fix(streams): frame multiline SSE payloads as consecutive data: lines

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ### Fixed
 
+- **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.
+
 - **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

app/assets/javascripts/pgbus/stream_source_element.js

@@ -225,13 +225,17 @@ class PgbusStreamSourceElement extends HTMLElement {
 
     let id = null
     let event = "message"
-    let data = ""
+    const dataLines = []
 
     for (const line of block.split("\n")) {
       if (line.startsWith("id:")) id = line.slice(3).trim()
       else if (line.startsWith("event:")) event = line.slice(6).trim()
-      else if (line.startsWith("data:")) data += line.slice(5).trim()
+      // Per the SSE spec: strip only a single leading space after the colon
+      // (never trim — payload whitespace is significant, issue #392) and
+      // rejoin consecutive data: lines with \n, matching native EventSource.
+      else if (line.startsWith("data:")) dataLines.push(line.slice(5).replace(/^ /, ""))
     }
+    const data = dataLines.join("\n")
 
     if (id !== null) this.lastEventId = id
 

lib/pgbus/streams/envelope.rb

@@ -12,13 +12,17 @@ module Streams
     #   - `comment(text)` — a heartbeat or sentinel that the SSE parser ignores
     #   - `retry_directive(ms)` — tells `EventSource` how long to wait before reconnecting
     #
-    # All frames end with `\n\n` (the SSE event terminator). `data:` lines must not
-    # contain newlines — the SSE spec uses `\n` as the field terminator, so a multi-line
-    # payload would arrive as multiple events. We strip `\r` and `\n` from data and
-    # comment text rather than splitting into multiple `data:` lines, because Turbo
-    # Stream HTML is already flat and the simpler encoding is easier to debug.
+    # All frames end with `\n\n` (the SSE event terminator). A multiline payload is
+    # framed as consecutive `data:` lines (issue #392) — the client rejoins them with
+    # `\n`, so delivery is lossless. `\r\n` and lone `\r` are also SSE line terminators,
+    # so they become `data:` line breaks too (rejoined as `\n`; SSE cannot represent a
+    # raw `\r`). Every payload line carries the `data: ` prefix, so a crafted payload
+    # cannot inject forged id:/event: fields. Single-line fields (`event:`, comments)
+    # still strip newlines — there a `\r`/`\n` would terminate the field early and
+    # permit SSE field injection.
     module Envelope
       NEWLINES = /[\r\n]+/
+      DATA_LINE_BREAK = /\r\n|\r|\n/
 
       RESPONSE_HEADERS = "HTTP/1.1 200 OK\r\n" \
                          "content-type: text/event-stream\r\n" \
@@ -31,11 +35,13 @@ def self.message(id:, event:, data:)
         raise ArgumentError, "id is required" if id.nil?
         raise ArgumentError, "event is required" if event.nil? || event.to_s.empty?
 
-        # Strip newlines from BOTH event and data, not just data: each is
-        # interpolated into its own SSE field line, so an unescaped \r/\n in
-        # either would terminate the field early and let a crafted value
-        # inject extra SSE fields (a forged id:/data:) into the frame.
-        "id: #{id}\nevent: #{strip_newlines(event.to_s)}\ndata: #{strip_newlines(data.to_s)}\n\n"
+        # The event name is a single SSE field line, so newlines are stripped —
+        # an unescaped \r/\n would terminate the field early and let a crafted
+        # value inject extra SSE fields (a forged id:/data:) into the frame.
+        # The payload is framed as one `data:` line per payload line instead:
+        # every line carries the `data: ` prefix, which is both spec-correct
+        # (the client rejoins with \n) and injection-safe.
+        "id: #{id}\nevent: #{strip_newlines(event.to_s)}\n#{data_lines(data.to_s)}\n"
       end
 
       def self.comment(text)
@@ -71,7 +77,16 @@ def self.strip_newlines(str)
         str.gsub(NEWLINES, "")
       end
 
-      private_class_method :strip_newlines
+      # One `data: <line>\n` per payload line. The -1 limit keeps trailing
+      # empty strings, so a payload ending in \n round-trips as an empty
+      # final `data:` line (the client's rejoin restores the newline).
+      def self.data_lines(str)
+        lines = str.split(DATA_LINE_BREAK, -1)
+        lines = [""] if lines.empty? # "".split → [] — an empty payload still gets its data: line
+        lines.map { |line| "data: #{line}\n" }.join
+      end
+
+      private_class_method :strip_newlines, :data_lines
     end
   end
 end

spec/pgbus/streams/envelope_spec.rb

@@ -15,18 +15,50 @@
       )
     end
 
-    it "strips newlines from data so the SSE parser sees one logical event" do
+    it "frames a multiline payload as consecutive data: lines (SSE spec, issue #392)" do
       data = "<turbo-stream>\n  <template>x</template>\n</turbo-stream>"
       result = described_class.message(id: 1, event: "turbo-stream", data: data)
-      # Newlines collapsed; surrounding spaces preserved
-      expect(result).not_to include("\n  <template>")
-      expect(result).to include("data: <turbo-stream>  <template>x</template></turbo-stream>")
+      expect(result).to eq(
+        "id: 1\n" \
+        "event: turbo-stream\n" \
+        "data: <turbo-stream>\n" \
+        "data:   <template>x</template>\n" \
+        "data: </turbo-stream>\n" \
+        "\n"
+      )
+    end
+
+    it "round-trips a multiline payload through an EventSource-style rejoin" do
+      data = "<pre>\n  significant\n\n  whitespace\n</pre>"
+      result = described_class.message(id: 1, event: "msg", data: data)
+      rejoined = result.lines.grep(/\Adata:/).map { |l| l.chomp.delete_prefix("data:").delete_prefix(" ") }.join("\n")
+      expect(rejoined).to eq(data)
     end
 
-    it "strips carriage returns from data" do
+    it "treats \\r\\n and lone \\r as line breaks (SSE cannot carry a raw \\r)" do
       data = "a\r\nb\rc"
       result = described_class.message(id: 1, event: "msg", data: data)
-      expect(result).to include("data: abc\n")
+      expect(result).to include("data: a\ndata: b\ndata: c\n")
+      expect(result).not_to include("\r")
+    end
+
+    it "preserves a trailing newline in the payload (empty final data: line)" do
+      result = described_class.message(id: 1, event: "msg", data: "a\n")
+      expect(result).to include("data: a\ndata: \n\n")
+      rejoined = result.lines.grep(/\Adata:/).map { |l| l.chomp.delete_prefix("data:").delete_prefix(" ") }.join("\n")
+      expect(rejoined).to eq("a\n")
+    end
+
+    it "cannot be used to inject SSE fields via newlines in data (every line is data:-prefixed)" do
+      result = described_class.message(id: 1, event: "msg", data: "x\nid: 999\nevent: evil")
+      expect(result.scan(/^id:/).size).to eq(1)
+      expect(result.scan(/^event:/).size).to eq(1)
+      expect(result).to include("data: x\ndata: id: 999\ndata: event: evil\n")
+    end
+
+    it "emits a single empty data: line for an empty payload" do
+      result = described_class.message(id: 1, event: "msg", data: "")
+      expect(result).to eq("id: 1\nevent: msg\ndata: \n\n")
     end
 
     it "preserves UTF-8 characters in data" do