diff --git a/CHANGELOG.md b/CHANGELOG.md index 8315f54e..ab5332fd 100644 --- a/CHANGELOG.md +++ b/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 `
` 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 `` 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 diff --git a/app/assets/javascripts/pgbus/stream_source_element.js b/app/assets/javascripts/pgbus/stream_source_element.js index 1ffdf029..877a07b3 100644 --- a/app/assets/javascripts/pgbus/stream_source_element.js +++ b/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 diff --git a/lib/pgbus/streams/envelope.rb b/lib/pgbus/streams/envelope.rb index f8e7b8d4..ac2c7acb 100644 --- a/lib/pgbus/streams/envelope.rb +++ b/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: \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 diff --git a/spec/pgbus/streams/envelope_spec.rb b/spec/pgbus/streams/envelope_spec.rb index 1fa0dfc0..6d6e017f 100644 --- a/spec/pgbus/streams/envelope_spec.rb +++ b/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 = " \n x\n " result = described_class.message(id: 1, event: "turbo-stream", data: data) - # Newlines collapsed; surrounding spaces preserved - expect(result).not_to include("\n ") - expect(result).to include("data:x ") + expect(result).to eq( + "id: 1\n" \ + "event: turbo-stream\n" \ + "data:\n" \ + "data: x\n" \ + "data: \n" \ + "\n" + ) + end + + it "round-trips a multiline payload through an EventSource-style rejoin" do + data = "\n significant\n\n whitespace\n" + 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