From 7f20db0e5d3b5b37eabd7ab8150ba509543840d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20Pacho=C5=82ek?= Date: Sun, 27 Sep 2026 23:41:36 +0200 Subject: [PATCH] Add payload_format raw: send the NATS payload as the event data (#143) An event's data was always the JSON envelope {"topic","payload","time"}, so clients that only want the payload had to unwrap it, and plain-text payloads arrived as escaped JSON strings. payload_format raw writes the NATS payload itself: one data line per line of it, which EventSource joins with "\n" again (an empty payload is one empty data line and still dispatches an event). SSE data is UTF-8 text, and a carriage return ends a line. So a payload that is not valid UTF-8, or holds a carriage return, cannot be sent: it is dropped, logged and counted as nuts_messages_dropped_total{reason="raw_not_text"}. An oversized payload keeps its raw_payload reason. The default, envelope, changes nothing. Closes #143 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 6 ++ Caddyfile.test | 16 +++++ README.md | 29 ++++++++- caddyfile.go | 6 ++ caddyfile_test.go | 9 +++ docs/CONFIGURATION.md | 1 + features/sse_streaming.feature | 8 +++ formatter_test.go | 106 +++++++++++++++++++++++++++++---- handler.go | 7 +++ metrics.go | 5 +- provision.go | 8 +++ provision_test.go | 5 ++ serve.go | 59 ++++++++++++++---- serve_integration_test.go | 56 +++++++++++++++++ website/docs/usage.md | 3 +- 15 files changed, 299 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c315a94..ee0549e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- `payload_format raw` sends the NATS payload itself as each event's data, + without the JSON envelope: a `data:` line per line of it, which + EventSource joins back (#143). A payload SSE cannot carry, not UTF-8 text + or holding a carriage return, is dropped and counted as + `nuts_messages_dropped_total{reason="raw_not_text"}`. The default, + `envelope`, changes nothing. - `event_type topic` and `event_type header ` name each message's SSE event after its topic or a header of the NATS message, so a page can add one `addEventListener` per kind of event instead of routing every diff --git a/Caddyfile.test b/Caddyfile.test index b274151..6371eda 100644 --- a/Caddyfile.test +++ b/Caddyfile.test @@ -77,6 +77,22 @@ } } + # The same stream with each event's data the raw NATS payload, for + # features/sse_streaming.feature. + handle /raw* { + uri strip_prefix /raw + nuts { + nats_url {$NATS_URL:nats://localhost:4222} + stream_name {$STREAM_NAME:EVENTS} + topic_prefix {$TOPIC_PREFIX:events.} + heartbeat_interval 1 + nats_idle_heartbeat 1 + nats_ping_interval 1 + allowed_origins * + payload_format raw + } + } + handle /example* { root * /srv/example uri strip_prefix /example diff --git a/README.md b/README.md index a7f436f..4e81d1e 100644 --- a/README.md +++ b/README.md @@ -384,6 +384,7 @@ nuts { replay_window # Time-bound replay to the last N seconds (default: 0 = all retained) event_id_format # Event ids: sequence (default) or sequence_time (sequence and message time) event_type [name] # Event names: message (default), topic, or header + payload_format # Event data: envelope (default) or raw (the NATS payload itself) health_path # Legacy readiness endpoint (empty/default: /healthz) live_path # Process liveness endpoint (empty/default: /livez) ready_path # NATS/stream readiness endpoint (empty/default: /readyz) @@ -648,6 +649,30 @@ names themselves, and an event named `error` would reach the page's message without the header, the event is named `message`. The JSON data is the same in every case. The event name counts towards `max_event_size`. +#### `payload_format` + +By default an event's data is the JSON envelope +`{"topic":…,"payload":…,"time":…}`. `payload_format raw` sends the NATS +payload itself instead: a page that only wants the payload need not unwrap +it, plain-text or CSV payloads arrive as they are, and the envelope's 120 to +150 bytes no longer count towards `max_event_size`. + +``` +event: message +data: first line of the payload +data: second line of the payload +id: 12345 +``` + +A payload with line breaks is sent as several `data:` lines, which +EventSource joins with `\n` again. SSE data is UTF-8 text and a carriage +return ends a line, so a payload that is not valid UTF-8 or holds a carriage +return cannot be sent: it is dropped, logged, and counted as +`nuts_messages_dropped_total{reason="raw_not_text"}`. The envelope's topic +and time are gone in raw mode; pair it with +[`event_type topic`](#event_type) to tell topics apart. The `connected` and +`reset` events keep their JSON data. + #### `event_id_format` `event_id_format sequence_time` makes each event's id carry the stored time @@ -953,7 +978,7 @@ Then scrape `http://localhost:8080/metrics` from Prometheus. Available metrics: |--------|------|-------------| | `nuts_active_connections` | Gauge | Currently connected SSE clients | | `nuts_messages_delivered_total` | Counter | SSE message events successfully written | -| `nuts_messages_dropped_total{reason}` | Counter (labeled) | Messages not delivered to a client. `reason` is one of `raw_payload` (inbound NATS payload exceeded `max_event_size`), `formatted_sse_message` (SSE envelope after JSON wrap exceeded `max_event_size`), `replay_window` (a replayed message older than `replay_window`) or `control_message` (a subject delete marker or schedule definition; see [Server control messages](#server-control-messages)). | +| `nuts_messages_dropped_total{reason}` | Counter (labeled) | Messages not delivered to a client. `reason` is one of `raw_payload` (inbound NATS payload exceeded `max_event_size`), `formatted_sse_message` (SSE envelope after JSON wrap exceeded `max_event_size`), `replay_window` (a replayed message older than `replay_window`), `control_message` (a subject delete marker or schedule definition; see [Server control messages](#server-control-messages)) or `raw_not_text` (a payload [`payload_format raw`](#payload_format) cannot send: not UTF-8 text, or holding a carriage return). | | `nuts_wildcard_filter_drops_total` | Counter | Deprecated, always 0: the pre-NATS-2.10 wildcard fallback was removed. | | `nuts_slow_client_disconnects_total` | Counter | Clients disconnected because a write missed `write_timeout` (the client stopped reading) | | `nuts_replay_requests_total` | Counter | Connections requesting message replay | @@ -1241,7 +1266,7 @@ data: {"topic":"my-topic","payload":{"your":"data"},"time":"2024-01-01T12:00:00Z id: 12345 ``` -The event is named `message` unless [`event_type`](#event_type) names it after its topic or a header. The `id` field contains the JetStream sequence number, which can be used with `last-id` or `Last-Event-ID` for replay. With [`event_id_format sequence_time`](#event_id_format) it also carries the stored time of the message: `id: 12345-1790513389663997004`. It is the last line of each event: SSE allows the fields in any order, and some clients (for example `@microsoft/fetch-event-source`) record an id as soon as they read it, so with the id last a frame cut off mid-write cannot make them resume after a message they never received. +The event is named `message` unless [`event_type`](#event_type) names it after its topic or a header, and its data is the JSON envelope unless [`payload_format raw`](#payload_format) sends the payload itself. The `id` field contains the JetStream sequence number, which can be used with `last-id` or `Last-Event-ID` for replay. With [`event_id_format sequence_time`](#event_id_format) it also carries the stored time of the message: `id: 12345-1790513389663997004`. It is the last line of each event: SSE allows the fields in any order, and some clients (for example `@microsoft/fetch-event-source`) record an id as soon as they read it, so with the id last a frame cut off mid-write cannot make them resume after a message they never received. The first frame of every stream is the handshake event: diff --git a/caddyfile.go b/caddyfile.go index a366275..5e0d431 100644 --- a/caddyfile.go +++ b/caddyfile.go @@ -302,6 +302,12 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error { } h.EventIDFormat = d.Val() + case "payload_format": + if !d.NextArg() { + return d.ArgErr() + } + h.PayloadFormat = d.Val() + case "event_type": if !d.NextArg() { return d.ArgErr() diff --git a/caddyfile_test.go b/caddyfile_test.go index 0cd6e27..449c384 100644 --- a/caddyfile_test.go +++ b/caddyfile_test.go @@ -46,6 +46,7 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) { hub_url https://example.com/events event_id_format sequence_time event_type header Event-Type + payload_format raw subscriber_jwt_key secret-key subscriber_jwt_cookie nuts_session allowed_origins https://example.com https://other.com @@ -81,6 +82,7 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) { EventIDFormat: "sequence_time", EventType: "header", EventTypeHeader: "Event-Type", + PayloadFormat: "raw", SubscriberJWTKey: "secret-key", SubscriberJWTCookie: "nuts_session", AllowedOrigins: []string{"https://example.com", "https://other.com"}, @@ -387,6 +389,7 @@ func TestHandler_UnmarshalCaddyfile_MissingArgs(t *testing.T) { {name: "missing nats_ping_interval arg", directive: "nats_ping_interval"}, {name: "missing event_id_format arg", directive: "event_id_format"}, {name: "missing event_type arg", directive: "event_type"}, + {name: "missing payload_format arg", directive: "payload_format"}, {name: "missing max_reconnects arg", directive: "max_reconnects"}, {name: "missing max_event_size arg", directive: "max_event_size"}, {name: "missing dispatch_timeout arg", directive: "dispatch_timeout"}, @@ -596,6 +599,12 @@ func TestHandler_UnmarshalCaddyfile_RejectsInvalidOptionalConfig(t *testing.T) { line: "event_type topic extra", wantErr: "argument", }, + { + name: "unknown payload_format", + line: "payload_format text", + wantErr: "payload_format must be", + validateErr: true, + }, { name: "unknown event_type", line: "event_type category", diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index e138d98..2d6e321 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -76,6 +76,7 @@ enforced. For public or browser-facing routes, issue short-lived tokens with | `replay_max_messages ` | `replay_max_messages` | `0` | Integer `>= 0` | `0` is unlimited. When reached during replay, NUTS closes the stream cleanly and the client reconnects from its last event ID. Only history counts: messages published after the request arrived never trip the cap. | | `event_type [header name]` | `event_type` | `message` | `message`, `topic`, or `header` followed by a header name | The SSE event name of each message: `message`, its topic without `topic_prefix`, or the value of a NATS message header. A name outside `[A-Za-z0-9._:-]{1,100}`, a reserved name (`connected`, `reset`, `open`, `error`) or a missing header falls back to `message`. | | (the name after `event_type header`) | `event_type_header` | Empty | Printable ASCII without spaces or colons | The header `event_type header` reads; case-sensitive, as NATS headers are. Rejected unless `event_type` is `header`. | +| `payload_format ` | `payload_format` | `envelope` | `envelope` or `raw` | What an event's data holds: the JSON envelope `{"topic","payload","time"}`, or the NATS payload itself, a `data:` line per line of it. In raw mode a payload that is not UTF-8 text, or holds a carriage return, is dropped and counted as `nuts_messages_dropped_total{reason="raw_not_text"}`. | | `event_id_format ` | `event_id_format` | `sequence` | `sequence` or `sequence_time` | What an event's id holds: the message's stream sequence, or the sequence and the stored time of the message, `-`. With `sequence_time`, a client that resumes with a cursor from a stream that was since recreated or restored replays the stream from its start instead of skipping its first messages; both forms are accepted as cursors either way. Enable it only for clients that treat ids as opaque strings. | | `replay_window ` | `replay_window` | `0` | Integer `>= 0` | `0` preserves retained replay. Positive values bound old replay cursors to `StartTime(now - replay_window)` while preserving exact sequence replay inside the window. Also applies when stream info cannot be read or the cursor is ahead of the stream. Live messages are never filtered. | diff --git a/features/sse_streaming.feature b/features/sse_streaming.feature index c348260..1471a38 100644 --- a/features/sse_streaming.feature +++ b/features/sse_streaming.feature @@ -22,6 +22,14 @@ Feature: SSE Streaming with JetStream Then I should receive a "orders" event And I should not receive an SSE event with topic "notifications" + # payload_format raw sends the NATS payload itself as the event's data, + # without the JSON envelope (#143). + Scenario: Events carry the raw payload + Given I am connected to SSE endpoint "/raw?topic=plain" + When I publish message 'just some text' to subject "events.plain" + Then I should receive an SSE event containing 'just some text' + And I should not receive an SSE event containing '"payload"' + Scenario: Receive messages from multiple topics Given I am connected to SSE endpoint "/events?topic=alerts&topic=updates" When I publish message '{"type": "alert"}' to subject "events.alerts" diff --git a/formatter_test.go b/formatter_test.go index 3c56803..19ac837 100644 --- a/formatter_test.go +++ b/formatter_test.go @@ -8,8 +8,11 @@ import ( "strings" "testing" "time" + "unicode/utf8" "github.com/nats-io/nats.go" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" ) // referenceMessageFrame renders a message frame the way formatMessageEvent @@ -29,9 +32,19 @@ func referenceMessageFrame(h *Handler, msg streamMessage, now time.Time) string } var event strings.Builder event.WriteString("event: " + referenceEventName(h, msg) + "\n") - event.WriteString("data: ") - event.WriteString(toJSON(payload)) - event.WriteString("\n") + if h.PayloadFormat == "raw" { + text := string(msg.Data) + if !utf8.ValidString(text) || strings.ContainsRune(text, '\r') { + return "" + } + for _, line := range strings.Split(text, "\n") { + event.WriteString("data: " + line + "\n") + } + } else { + event.WriteString("data: ") + event.WriteString(toJSON(payload)) + event.WriteString("\n") + } if msg.HasMetadata { event.WriteString("id: ") event.WriteString(strconv.FormatUint(msg.StreamSequence, 10)) @@ -124,18 +137,24 @@ func TestFormatMessageEvent_MatchesTheReferenceFormatter(t *testing.T) { func FuzzFormatMessageEventMatchesReference(f *testing.F) { for _, data := range formatterEdgeCases { - f.Add([]byte(data), "orders", uint64(42), true, false, uint8(0), "") - f.Add([]byte(data), "orders", uint64(42), true, true, uint8(1), "") - f.Add([]byte(data), "orders", uint64(42), true, false, uint8(2), "order.created") + f.Add([]byte(data), "orders", uint64(42), true, false, uint8(0), "", false) + f.Add([]byte(data), "orders", uint64(42), true, true, uint8(1), "", false) + f.Add([]byte(data), "orders", uint64(42), true, false, uint8(2), "order.created", false) + f.Add([]byte(data), "orders", uint64(42), true, false, uint8(0), "", true) } - f.Add([]byte(`{}`), "reset", uint64(1), true, false, uint8(1), "") - f.Add([]byte(`{}`), "orders", uint64(1), true, false, uint8(2), "has space") + f.Add([]byte(`{}`), "reset", uint64(1), true, false, uint8(1), "", false) + f.Add([]byte(`{}`), "orders", uint64(1), true, false, uint8(2), "has space", false) + f.Add([]byte("line one\nline two\n"), "orders", uint64(1), true, false, uint8(0), "", true) + f.Add([]byte("a\r\nb"), "orders", uint64(1), true, false, uint8(0), "", true) now := time.Date(2026, 9, 26, 12, 0, 0, 0, time.UTC) - f.Fuzz(func(t *testing.T, data []byte, topic string, seq uint64, hasMeta, timedIDs bool, eventType uint8, header string) { + f.Fuzz(func(t *testing.T, data []byte, topic string, seq uint64, hasMeta, timedIDs bool, eventType uint8, header string, raw bool) { h := &Handler{TopicPrefix: "events.", MaxEventSize: -1} if timedIDs { h.EventIDFormat = eventIDSequenceTime } + if raw { + h.PayloadFormat = payloadFormatRaw + } switch eventType % 3 { case 1: h.EventType = eventTypeTopic @@ -144,12 +163,79 @@ func FuzzFormatMessageEventMatchesReference(f *testing.F) { } msg := streamMessage{Subject: "events." + topic, Data: data, Header: nats.Header{"Event-Type": []string{header}}, HasMetadata: hasMeta, StreamSequence: seq, Timestamp: now.Add(-time.Minute)} want := referenceMessageFrame(h, msg, now) - if got := h.formatMessageEvent(msg, now).Frame; got != want { + got := h.formatMessageEvent(msg, now).Frame + if got != want { t.Fatalf("frame differs for data=%q topic=%q:\n got %q\nwant %q", data, topic, got, want) } + // #143: what EventSource makes of a raw frame is the payload itself. + if raw && got != "" { + if back := eventSourceData(got); back != string(data) { + t.Fatalf("raw payload %q reads back as %q from %q", data, back, got) + } + } }) } +// eventSourceData is the data EventSource dispatches for one frame: the +// values of its data lines, joined with "\n". +func eventSourceData(frame string) string { + var lines []string + for _, line := range strings.Split(strings.TrimSuffix(frame, "\n\n"), "\n") { + if value, ok := strings.CutPrefix(line, "data: "); ok { + lines = append(lines, value) + } + } + return strings.Join(lines, "\n") +} + +// TestFormatMessageEvent_RawPayload covers #143: in payload_format raw the +// event's data is the NATS payload, a data line per line of it, which +// EventSource joins back. A payload SSE cannot carry is dropped and counted; +// an oversized one keeps its own reason. +func TestFormatMessageEvent_RawPayload(t *testing.T) { + h := &Handler{TopicPrefix: "events.", MaxEventSize: -1, PayloadFormat: payloadFormatRaw} + now := time.Date(2026, 9, 27, 12, 0, 0, 0, time.UTC) + frame := func(data string) formattedMessageEvent { + return h.formatMessageEvent(streamMessage{Subject: "events.orders", Data: []byte(data), HasMetadata: true, StreamSequence: 7, Timestamp: now}, now) + } + for data, want := range map[string]string{ + `{"id":1}`: "event: message\ndata: {\"id\":1}\nid: 7\n\n", + "line one\nline two": "event: message\ndata: line one\ndata: line two\nid: 7\n\n", + "ends with a break\n": "event: message\ndata: ends with a break\ndata: \nid: 7\n\n", + "": "event: message\ndata: \nid: 7\n\n", + "zażółć &": "event: message\ndata: zażółć &\nid: 7\n\n", + } { + got := frame(data) + if got.Dropped || got.Frame != want { + t.Errorf("raw %q: frame %q (dropped %v), want %q", data, got.Frame, got.Dropped, want) + } + if back := eventSourceData(got.Frame); back != data { + t.Errorf("raw %q reads back as %q", data, back) + } + } + for _, data := range []string{"a\r\nb", "carriage\rreturn", "\xff\xfe not UTF-8"} { + if got := frame(data); !got.Dropped || got.DropReason != dropReasonRawNotText || got.DropSize != len(data) || got.Frame != "" { + t.Errorf("raw %q: dropped %v for %q (size %d), want a raw_not_text drop", data, got.Dropped, got.DropReason, got.DropSize) + } + } + h.MaxEventSize = 4 + if got := frame("\xff too big"); got.DropReason != dropReasonRawPayload { + t.Errorf("an oversized payload was dropped for %q, want %q", got.DropReason, dropReasonRawPayload) + } + + core, obs := observer.New(zap.WarnLevel) + h.logger = zap.New(core) + before := metricValue(t, metricsMessagesDropped.WithLabelValues(dropReasonRawNotText)) + h.MaxEventSize = -1 + h.recordDroppedMessage(frame("a\rb")) + if got := metricValue(t, metricsMessagesDropped.WithLabelValues(dropReasonRawNotText)); got != before+1 { + t.Fatalf("messages_dropped_total{raw_not_text} = %v, want %v", got, before+1) + } + if !hasLogField(obs, "topic", "events.orders") || obs.FilterMessageSnippet("payload_format raw cannot send").Len() != 1 { + t.Fatalf("the drop was not logged: %+v", obs.All()) + } +} + func BenchmarkFormatMessageEventLarge(b *testing.B) { h := &Handler{TopicPrefix: "events.", MaxEventSize: -1} now := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC) diff --git a/handler.go b/handler.go index b960d03..3cb96b1 100644 --- a/handler.go +++ b/handler.go @@ -107,6 +107,13 @@ type Handler struct { // EventTypeHeader is the header EventType "header" reads. EventTypeHeader string `json:"event_type_header,omitempty"` + // PayloadFormat is what an event's data holds: "envelope" (the default), + // the JSON {"topic","payload","time"}; or "raw", the NATS payload itself, + // one data line per line of it. In raw mode a payload that is not UTF-8 + // text, or holds a carriage return, cannot be sent as SSE data and is + // dropped (#143). + PayloadFormat string `json:"payload_format,omitempty"` + // HubURL is the URL advertised in the Link header for hub discovery. // When set, SSE responses include a Link: ; rel="nuts" header // so that clients and upstream APIs can discover the event hub automatically. diff --git a/metrics.go b/metrics.go index 7ba904f..4afdc09 100644 --- a/metrics.go +++ b/metrics.go @@ -39,11 +39,12 @@ var ( // producer, the second at envelope overhead on small but pathological // payloads. replay_window counts replayed messages older than the window, // control_message the server's subject delete markers and schedule - // definitions. + // definitions, raw_not_text the payloads payload_format raw cannot send + // (#143). metricsMessagesDropped = promauto.NewCounterVec(prometheus.CounterOpts{ Namespace: "nuts", Name: "messages_dropped_total", - Help: "Total number of messages not delivered to a client, labelled by reason (raw_payload, formatted_sse_message, replay_window, control_message).", + Help: "Total number of messages not delivered to a client, labelled by reason (raw_payload, formatted_sse_message, replay_window, control_message, raw_not_text).", }, []string{"reason"}) // nuts_wildcard_filter_drops_total is deprecated: it counted messages the diff --git a/provision.go b/provision.go index 2947dd1..f3c38e2 100644 --- a/provision.go +++ b/provision.go @@ -213,6 +213,9 @@ func (h *Handler) Provision(ctx caddy.Context) error { if h.EventType == "" { h.EventType = eventTypeMessage } + if h.PayloadFormat == "" { + h.PayloadFormat = payloadFormatEnvelope + } if h.HealthPath == "" { h.HealthPath = defaultHealthPath } @@ -867,6 +870,11 @@ func (h *Handler) validateConfigValues() error { default: return fmt.Errorf("event_id_format must be %q or %q, not %q", eventIDSequence, eventIDSequenceTime, h.EventIDFormat) } + switch h.PayloadFormat { + case "", payloadFormatEnvelope, payloadFormatRaw: + default: + return fmt.Errorf("payload_format must be %q or %q, not %q", payloadFormatEnvelope, payloadFormatRaw, h.PayloadFormat) + } switch h.EventType { case "", eventTypeMessage, eventTypeTopic: if h.EventTypeHeader != "" { diff --git a/provision_test.go b/provision_test.go index 545bdfa..7b71720 100644 --- a/provision_test.go +++ b/provision_test.go @@ -1194,6 +1194,11 @@ func TestHandler_Provision_RejectsInvalidOptionalJSONConfigBeforeDialing(t *test fragment: `"event_type": "category"`, wantErr: "event_type must be", }, + { + name: "unknown payload format", + fragment: `"payload_format": "binary"`, + wantErr: "payload_format must be", + }, { name: "event type header without header mode", fragment: `"event_type": "topic", "event_type_header": "Event-Type"`, diff --git a/serve.go b/serve.go index fe987d7..63f3604 100644 --- a/serve.go +++ b/serve.go @@ -19,6 +19,7 @@ package nuts import ( + "bytes" "context" "encoding/json" "errors" @@ -31,6 +32,7 @@ import ( "sync" "sync/atomic" "time" + "unicode/utf8" "github.com/caddyserver/caddy/v2/modules/caddyhttp" "github.com/nats-io/nats.go" @@ -78,6 +80,11 @@ const ( eventTypeMessage = "message" eventTypeTopic = "topic" eventTypeHeader = "header" + + // Values of payload_format: an event's data is the JSON envelope, or + // the NATS payload itself (#143). + payloadFormatEnvelope = "envelope" + payloadFormatRaw = "raw" ) // replayMode describes how a JetStream subscription should position itself @@ -116,6 +123,9 @@ const ( // bookkeeping (a subject delete marker or a schedule definition), which // is not an application event. dropReasonControlMessage string = "control_message" + // dropReasonRawNotText tags a payload that payload_format raw cannot + // send: SSE data is UTF-8 text, and a carriage return ends a line (#143). + dropReasonRawNotText string = "raw_not_text" ) // isFallback reports whether the mode was selected by the fallback path @@ -1339,6 +1349,13 @@ func (h *Handler) formatMessageEvent(msg streamMessage, now time.Time) formatted formatted.DropSize = len(msg.Data) return formatted } + raw := h.PayloadFormat == payloadFormatRaw + if raw && (!utf8.Valid(msg.Data) || bytes.IndexByte(msg.Data, '\r') >= 0) { + formatted.Dropped = true + formatted.DropReason = dropReasonRawNotText + formatted.DropSize = len(msg.Data) + return formatted + } // Build the frame in one pass. The data line is the JSON encoding of // messageEventPayload, written by hand: the payload is validated once @@ -1350,17 +1367,33 @@ func (h *Handler) formatMessageEvent(msg streamMessage, now time.Time) formatted var scratch [64]byte event.WriteString("event: ") event.WriteString(h.eventName(msg)) - event.WriteString("\ndata: {\"topic\":") - writeJSONString(&event, strings.TrimPrefix(msg.Subject, h.TopicPrefix)) - event.WriteString(`,"payload":`) - writeJSONPayload(&event, msg.Data) - event.WriteString(`,"time":"`) - timestamp := now - if msg.HasMetadata { - timestamp = msg.Timestamp + if raw { + // The payload itself, a data line per line of it: EventSource joins + // them with "\n" again. An empty payload is one empty data line, + // which still dispatches an event. + for rest := msg.Data; ; { + line, after, more := bytes.Cut(rest, []byte{'\n'}) + event.WriteString("\ndata: ") + event.Write(line) + if !more { + break + } + rest = after + } + event.WriteByte('\n') + } else { + event.WriteString("\ndata: {\"topic\":") + writeJSONString(&event, strings.TrimPrefix(msg.Subject, h.TopicPrefix)) + event.WriteString(`,"payload":`) + writeJSONPayload(&event, msg.Data) + event.WriteString(`,"time":"`) + timestamp := now + if msg.HasMetadata { + timestamp = msg.Timestamp + } + event.Write(timestamp.UTC().AppendFormat(scratch[:0], time.RFC3339)) + event.WriteString("\"}\n") } - event.Write(timestamp.UTC().AppendFormat(scratch[:0], time.RFC3339)) - event.WriteString("\"}\n") // The id goes last (#107). EventSource applies it only when the event // is dispatched, but clients such as fetch-event-source store it as soon // as they parse the line: with the id first, a frame cut off by a write @@ -1425,6 +1458,12 @@ func (h *Handler) recordDroppedMessage(formatted formattedMessageEvent) { zap.Int("payload_size", formatted.DropSize), zap.Int("max_event_size", h.MaxEventSize), ) + case dropReasonRawNotText: + h.log().Warn("dropping a NATS payload payload_format raw cannot send: not UTF-8 text, or holds a carriage return", + zap.String("topic", formatted.Subject), + zap.Uint64("stream_sequence", formatted.StreamSequence), + zap.Int("payload_size", formatted.DropSize), + ) case dropReasonFormattedSSEMessage: h.log().Warn("dropping oversized SSE event", zap.String("topic", formatted.Subject), diff --git a/serve_integration_test.go b/serve_integration_test.go index 36d4dae..f3b8e79 100644 --- a/serve_integration_test.go +++ b/serve_integration_test.go @@ -1568,3 +1568,59 @@ func TestHandler_EventTypeFromTheTopicOrAHeader(t *testing.T) { }) } } + +// TestHandler_RawPayloadFormat: with payload_format raw, an event's data is +// the published payload, a data line per line of it; a payload SSE cannot +// carry is dropped, and the stream goes on (#143). +func TestHandler_RawPayloadFormat(t *testing.T) { + ns := startJetStreamServer(t) + nc, err := nats.Connect(ns.ClientURL()) + if err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(nc.Close) + createTestStream(t, nc, "EVENTS", []string{"events.>"}) + _, srv := newContractServer(t, ns.ClientURL(), func(h *Handler) { h.PayloadFormat = payloadFormatRaw }) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/events?topic=text", nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("request: %v", err) + } + defer resp.Body.Close() + lines := bufio.NewScanner(resp.Body) + nextData := func() string { + t.Helper() + var data []string + event := "" + for lines.Scan() { + line := lines.Text() + switch { + case strings.HasPrefix(line, "event: "): + event = strings.TrimPrefix(line, "event: ") + case strings.HasPrefix(line, "data: "): + data = append(data, strings.TrimPrefix(line, "data: ")) + case line == "" && event == "message": + return strings.Join(data, "\n") + case line == "": + event, data = "", nil + } + } + t.Fatalf("stream ended: %v", lines.Err()) + return "" + } + js, _ := nc.JetStream() + for _, payload := range []string{"first line\nsecond line", "carriage\rreturn", `{"after":"the drop"}`} { + if _, err := js.Publish("events.text", []byte(payload)); err != nil { + t.Fatalf("publish: %v", err) + } + } + if got := nextData(); got != "first line\nsecond line" { + t.Fatalf("first event data = %q", got) + } + if got := nextData(); got != `{"after":"the drop"}` { + t.Fatalf("the event after the unsendable payload = %q", got) + } +} diff --git a/website/docs/usage.md b/website/docs/usage.md index dc73daa..5c7a534 100644 --- a/website/docs/usage.md +++ b/website/docs/usage.md @@ -64,6 +64,7 @@ nuts { replay_window # Time-bound replay window (default: 0 = all retained) event_id_format # Event ids: sequence (default) or sequence_time event_type [name] # Event names: message (default), topic, or header + payload_format # Event data: envelope (default) or raw # Health probes live_path # Liveness probe (default: /livez) @@ -286,7 +287,7 @@ NUTS registers the following metrics; expose them via Caddy's `metrics` handler: |--------|------|-------------| | `nuts_active_connections` | Gauge | Currently connected SSE clients | | `nuts_messages_delivered_total` | Counter | SSE message events successfully written to clients | -| `nuts_messages_dropped_total{reason}` | Counter | Messages not delivered (`reason`: `raw_payload`, `formatted_sse_message` — exceeded `max_event_size`; `replay_window` — a replayed message older than `replay_window`; `control_message` — a subject delete marker or schedule definition) | +| `nuts_messages_dropped_total{reason}` | Counter | Messages not delivered (`reason`: `raw_payload`, `formatted_sse_message` — exceeded `max_event_size`; `replay_window` — a replayed message older than `replay_window`; `control_message` — a subject delete marker or schedule definition; `raw_not_text` — a payload `payload_format raw` cannot send) | | `nuts_wildcard_filter_drops_total` | Counter | Deprecated, always `0`: the pre-2.10 wildcard fallback was removed | | `nuts_slow_client_disconnects_total` | Counter | Clients disconnected because a write missed `write_timeout` | | `nuts_replay_requests_total` | Counter | Connections requesting message replay |