Add sse_retry and transient_retry: set the retry hints NUTS sends (#145)

CHANGELOG.md

@@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- `sse_retry <milliseconds>` sends `retry:` with every stream's handshake,
+  setting EventSource's reconnection delay after ordinary disconnects, and
+  `transient_retry <seconds>` sets the average of the jittered delay NUTS
+  asks for after transient failures and closed streams, 5 seconds as
+  before by default (#145).
 - `health_details` adds the NATS server the handler is connected to
   (version, name, cluster, domain) and the stream as the probe read it
   (subjects, storage, replicas, message and byte counts, first and last

Caddyfile.test

@@ -94,7 +94,8 @@
 	}
 
 	# The same stream whose readiness probes add the NATS server and the
-	# stream's details, for features/sse_streaming.feature.
+	# stream's details, and whose streams set EventSource's reconnection
+	# delay, for features/sse_streaming.feature.
 	handle /details* {
 		uri strip_prefix /details
 		nuts {
@@ -106,6 +107,7 @@
 			nats_ping_interval 1
 			allowed_origins *
 			health_details
+			sse_retry 2500
 		}
 	}
 

README.md

@@ -386,6 +386,8 @@ nuts {
     event_type <source> [name]   # Event names: message (default), topic, or header <name>
     payload_format <format>      # Event data: envelope (default) or raw (the NATS payload itself)
     health_details [true|false]  # Readiness probes add the NATS server and stream details (default: off)
+    sse_retry <milliseconds>     # retry: sent with every stream's handshake (default: 0 = none)
+    transient_retry <seconds>    # Average retry delay after transient failures, jittered ±50% (0=default 5)
     health_path <path>           # Legacy readiness endpoint (empty/default: /healthz)
     live_path <path>             # Process liveness endpoint (empty/default: /livez)
     ready_path <path>            # NATS/stream readiness endpoint (empty/default: /readyz)
@@ -460,7 +462,7 @@ For example, setting `max_event_size 1000` means that if a NATS message produces
 
 Caps the number of concurrent SSE streams per NUTS instance. When the cap
 is reached, new clients receive `429 Too Many Requests` (RFC 6585) with a
-jittered `Retry-After` of 3–8 seconds, and the
+jittered `Retry-After` of 3–8 seconds by default (see `transient_retry`), and the
 `nuts_connections_rejected_total{reason="max_connections"}` counter is
 incremented. Browser `EventSource` clients get a `retry:` stream instead; see
 [Transient failures and EventSource](#transient-failures-and-eventsource).
@@ -1180,9 +1182,9 @@ that send `Accept: text/event-stream`, as every `EventSource` does:
 | Stream consumer limit reached | `200` retry stream | `503` + `Retry-After` |
 | Consumer could not be created | `200` retry stream | `503` + `Retry-After` |
 
-The retry stream holds a comment naming the reason and a `retry:` delay of
-2.5–7.5 seconds (jittered so rejected clients do not return together), then
-closes:
+The retry stream holds a comment naming the reason and a `retry:` delay
+jittered by ±50% around `transient_retry`, 2.5–7.5 seconds by default (so
+rejected clients do not return together), then closes:
 
 ```
 : JetStream not available
@@ -1198,6 +1200,23 @@ Client errors (`400`, `401`, `403`) and topics outside the stream
 (`503 Failed to subscribe to requested topics`) are not retryable: they will
 not succeed on their own.
 
+#### Retry hints: `sse_retry` and `transient_retry`
+
+- `transient_retry <seconds>` sets the average of the delay above, 5 by
+  default: the retry stream's `retry:`, `Retry-After`, and the `retry:` of a
+  stream closed because its JetStream stream was recreated or rewound. The
+  delay sent is jittered by ±50%, so clients sent away together do not come
+  back together. A large deployment may want longer, more spread-out
+  reconnects after an outage.
+- `sse_retry <milliseconds>` sends `retry:` with every stream's handshake,
+  so EventSource waits that long after an ordinary disconnect too (a dropped
+  network, a reload, a write timeout). Unset, browsers use their own default,
+  a few seconds. A `retry:` sets EventSource's delay for the rest of its
+  life, so a client that received a retry stream keeps its delay until the
+  next one.
+
+Both accept up to an hour.
+
 ### Slow Clients And Replay
 
 Each stream pulls from JetStream only as fast as its client reads. A client

caddyfile.go

@@ -302,6 +302,26 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
 				}
 				h.EventIDFormat = d.Val()
 
+			case "sse_retry":
+				v, err := parseInt("sse_retry")
+				if err != nil {
+					return err
+				}
+				if v < 0 {
+					return d.Errf("sse_retry must be >= 0 (0 sends none)")
+				}
+				h.SSERetry = v
+
+			case "transient_retry":
+				v, err := parseInt("transient_retry")
+				if err != nil {
+					return err
+				}
+				if v < 0 {
+					return d.Errf("transient_retry must be >= 0 (0 uses the default)")
+				}
+				h.TransientRetry = v
+
 			case "health_details":
 				if d.NextArg() {
 					b, err := strconv.ParseBool(d.Val())

caddyfile_test.go

@@ -55,6 +55,8 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) {
 				event_type header Event-Type
 				payload_format raw
 				health_details
+				sse_retry 2500
+				transient_retry 8
 				subscriber_jwt_key secret-key
 				subscriber_jwt_cookie nuts_session
 				allowed_origins https://example.com https://other.com
@@ -92,6 +94,8 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) {
 				EventTypeHeader:           "Event-Type",
 				PayloadFormat:             "raw",
 				HealthDetails:             true,
+				SSERetry:                  2500,
+				TransientRetry:            8,
 				SubscriberJWTKey:          "secret-key",
 				SubscriberJWTCookie:       "nuts_session",
 				AllowedOrigins:            []string{"https://example.com", "https://other.com"},
@@ -399,6 +403,8 @@ func TestHandler_UnmarshalCaddyfile_MissingArgs(t *testing.T) {
 		{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 sse_retry arg", directive: "sse_retry"},
+		{name: "missing transient_retry arg", directive: "transient_retry"},
 		{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"},
@@ -608,6 +614,38 @@ func TestHandler_UnmarshalCaddyfile_RejectsInvalidOptionalConfig(t *testing.T) {
 			line:    "event_type topic extra",
 			wantErr: "argument",
 		},
+		{
+			name:    "negative sse_retry at parse time",
+			line:    "sse_retry -1",
+			wantErr: "sse_retry must be >= 0",
+		},
+		{
+			name:    "negative transient_retry at parse time",
+			line:    "transient_retry -1",
+			wantErr: "transient_retry must be >= 0",
+		},
+		{
+			name:        "sse_retry over an hour",
+			line:        "sse_retry 3600001",
+			wantErr:     "sse_retry must be between",
+			validateErr: true,
+		},
+		{
+			name:        "transient_retry over an hour",
+			line:        "transient_retry 3601",
+			wantErr:     "transient_retry must be between",
+			validateErr: true,
+		},
+		{
+			name:    "sse_retry that is not a number",
+			line:    "sse_retry soon",
+			wantErr: "invalid sse_retry",
+		},
+		{
+			name:    "transient_retry that is not a number",
+			line:    "transient_retry 2.5",
+			wantErr: "invalid transient_retry",
+		},
 		{
 			name:    "health_details that is not a boolean",
 			line:    "health_details sometimes",
@@ -726,3 +764,19 @@ func TestHandler_UnmarshalCaddyfile_SharedSubscriptions(t *testing.T) {
 		}
 	}
 }
+
+// TestHandler_UnmarshalCaddyfile_RetryHintsGoUpToAnHour: sse_retry and
+// transient_retry accept an hour, their limit (#145).
+func TestHandler_UnmarshalCaddyfile_RetryHintsGoUpToAnHour(t *testing.T) {
+	d := caddyfile.NewTestDispenser("nuts {\n    nats_url nats://localhost:4222\n    stream_name EVENTS\n    sse_retry 3600000\n    transient_retry 3600\n}")
+	h := Handler{}
+	if err := h.UnmarshalCaddyfile(d); err != nil {
+		t.Fatalf("UnmarshalCaddyfile: %v", err)
+	}
+	if err := h.Validate(); err != nil {
+		t.Fatalf("Validate() = %v, want an hour accepted", err)
+	}
+	if h.SSERetry != 3600000 || h.TransientRetry != 3600 {
+		t.Fatalf("SSERetry = %d, TransientRetry = %d; want 3600000 and 3600", h.SSERetry, h.TransientRetry)
+	}
+}

consumer_test.go

@@ -709,20 +709,21 @@ func TestHandler_WatchStreamReadsEveryIntervalUntilShutdown(t *testing.T) {
 // JetStream stream was recreated or rewound ends with a reset event whose id,
 // 0, makes EventSource replay the stream from its start (#133, #137). It is
 // an event so that pages keeping the cursor see it, and it spreads the
-// reconnects with a jittered retry.
+// reconnects with a retry jittered around transient_retry (#145).
 func TestHandler_EndGenerationResetsTheCursor(t *testing.T) {
 	for _, tt := range []struct {
-		reason string
-		frame  string
+		reason    string
+		transient int
+		frame     string
 	}{
-		{streamRecreated, "event: reset\ndata: {\"reason\":\"stream_recreated\"}\nid: 0\n\n"},
-		{streamRewound, "event: reset\ndata: {\"reason\":\"stream_rewound\"}\nid: 0\n\n"},
+		{streamRecreated, 0, "event: reset\ndata: {\"reason\":\"stream_recreated\"}\nid: 0\n\n"},
+		{streamRewound, 30, "event: reset\ndata: {\"reason\":\"stream_rewound\"}\nid: 0\n\n"},
 	} {
 		t.Run(tt.reason, func(t *testing.T) {
 			gen := newStreamGeneration()
 			gen.reason = tt.reason
 			core, obs := observer.New(zap.InfoLevel)
-			h := &Handler{logger: zap.New(core)}
+			h := &Handler{logger: zap.New(core), TransientRetry: tt.transient}
 			before := metricValue(t, metricsConsumerInvalidated.WithLabelValues(tt.reason))
 			w := newSafeRecorder()
 			h.endGeneration(w, http.NewResponseController(w), streamPlan{Generation: gen}, time.Second)
@@ -733,8 +734,12 @@ func TestHandler_EndGenerationResetsTheCursor(t *testing.T) {
 			if !ok || err != nil || end != tt.frame {
 				t.Fatalf("frame %q, want a %s comment, a retry, then %q", w.Body(), tt.reason, tt.frame)
 			}
-			if d := time.Duration(delay) * time.Millisecond; d < transientRetryBase/2 || d >= 3*transientRetryBase/2 {
-				t.Fatalf("retry %v outside [%v, %v)", d, transientRetryBase/2, 3*transientRetryBase/2)
+			base := transientRetryBase
+			if tt.transient > 0 {
+				base = time.Duration(tt.transient) * time.Second
+			}
+			if d := time.Duration(delay) * time.Millisecond; d < base/2 || d >= 3*base/2 {
+				t.Fatalf("retry %v outside [%v, %v)", d, base/2, 3*base/2)
 			}
 			if got := metricValue(t, metricsConsumerInvalidated.WithLabelValues(tt.reason)); got != before+1 {
 				t.Fatalf("consumer_invalidated{reason=%s} = %v, want %v", tt.reason, got, before+1)

docs/CONFIGURATION.md

@@ -67,7 +67,7 @@ enforced. For public or browser-facing routes, issue short-lived tokens with
 | `nats_idle_heartbeat <seconds>` | `nats_idle_heartbeat` | `10` | Integer in `(0, 15)`; `0` uses the default; `-1` accepted for compatibility | Heartbeat interval of every pull request the stream's ordered consumer sends. When heartbeats stop (a dead link, a consumer reaped or deleted on the server), the consumer recreates itself from the last delivered sequence; see README "JetStream consumers". Must stay below 15 s, half the pull request expiry. `-1` no longer disables anything: the library default of 5 s applies and a warning is logged. |
 | `max_reconnects <count>` | `max_reconnects` | `-1` when omitted | Integer `>= -1`; `0` means no reconnects, `-1` means unlimited | JSON uses a pointer internally so explicit `0` is preserved. |
 | `max_event_size <bytes>` | `max_event_size` | `1048576` when `0` or omitted | Positive cap, `0` for default, negative for unlimited | Caps the formatted SSE frame. Oversized events are dropped and counted. |
-| `max_connections <count>` | `max_connections` | `0` | Integer `>= 0` | `0` disables the cap. Rejected clients receive `429` (Too Many Requests, RFC 6585) with a jittered `Retry-After` of 3–8 s; `EventSource` clients get a `200` stream with a `retry:` delay instead (README "Transient failures and EventSource"). Keep it at or below the stream's `max_consumers`: every connection needs its own consumer. |
+| `max_connections <count>` | `max_connections` | `0` | Integer `>= 0` | `0` disables the cap. Rejected clients receive `429` (Too Many Requests, RFC 6585) with a jittered `Retry-After` of 3–8 s by default (`transient_retry`); `EventSource` clients get a `200` stream with a `retry:` delay instead (README "Transient failures and EventSource"). Keep it at or below the stream's `max_consumers`: every connection needs its own consumer. |
 | `max_topics_per_subscription <count>` | `max_topics_per_subscription` | `32` when `0` or omitted | Positive cap, `0` for the default, `-1` for no limit | Caps the distinct `?topic=` filters allowed per SSE request after deduplication. Requests over the cap receive `400`. Other negative values are rejected as typos; `-1` is logged at startup. |
 | `client_buffer_size <count>` | `client_buffer_size` | `64` when `0` or omitted | Integer `>= 0` | How many messages each connection prefetches from JetStream. Pulling pauses while the client is behind, so a full prefetch never disconnects anyone. Per-connection memory is about `client_buffer_size` × the largest message the stream accepts, plus 18 formatted frames on their way ([PERFORMANCE.md](PERFORMANCE.md)). |
 | `shared_subscriptions [bool]` | `shared_subscriptions` | `false` | No argument means `true`; optional boolean accepted | Connections that are caught up with the live stream share one JetStream consumer per topic set, and each message is formatted once. Connections with history to replay, or more than `client_buffer_size` frames behind, use their own consumer until they catch up. See README "`shared_subscriptions`". |
@@ -76,6 +76,8 @@ enforced. For public or browser-facing routes, issue short-lived tokens with
 | `replay_max_messages <count>` | `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 <source> [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`. |
+| `sse_retry <milliseconds>` | `sse_retry` | `0` | Integer from `0` to `3600000`; `0` sends none | Sent as `retry:` with every stream's handshake: EventSource's reconnection delay after an ordinary disconnect. Unset, browsers use their own default. Negative values are rejected when the Caddyfile is parsed. |
+| `transient_retry <seconds>` | `transient_retry` | `5` when `0` or omitted | Integer from `0` to `3600`; `0` uses the default | Average delay a client is asked to wait after a transient failure (the retry stream's `retry:` and `Retry-After`) or after its stream closed because the JetStream stream was recreated or rewound. Jittered by ±50%. Negative values are rejected when the Caddyfile is parsed. |
 | `health_details [bool]` | `health_details` | `false` | No argument means `true`; optional boolean accepted | The readiness probes (`ready_path`, `health_path`) add `nats_server` (version, name, cluster, domain) and `stream_info` (subjects, storage, replicas, message and byte counts, first and last sequence, consumers, limits, creation time). Off by default because probe paths are often public and the details name servers, clusters and limits. |
 | `payload_format <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 <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, `<sequence>-<unix ns>`. 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. |

docs/TROUBLESHOOTING.md

@@ -189,10 +189,10 @@ Each is logged once as a warning: `JetStream stream was recreated`,
 `JetStream stream went back to an earlier sequence`, or, for a rewound stream
 that already caught up, `… and has since passed it`. Each closed SSE stream
 counts in `nuts_consumer_invalidated_total{reason}`. The reconnects are spread
-over 2.5 to 7.5 seconds, and each replay is bounded by `replay_max_messages`
-and `replay_window`. Operations that keep the stream, such as
-`nats stream purge` or a config update, do not reset its sequence numbers,
-and NUTS leaves its streams alone.
+over 2.5 to 7.5 seconds by default (`transient_retry`), and each replay is
+bounded by `replay_max_messages` and `replay_window`. Operations that keep
+the stream, such as `nats stream purge` or a config update, do not reset its
+sequence numbers, and NUTS leaves its streams alone.
 
 Limits:
 

features/sse_streaming.feature

@@ -82,6 +82,12 @@ Feature: SSE Streaming with JetStream
     Then I should receive HTTP status 200
     And the response should not contain "nats_server"
 
+  # sse_retry sets EventSource's reconnection delay with every stream's
+  # handshake (#145); the /details route sets 2500 ms.
+  Scenario: Streams set the reconnection delay
+    Given I am connected to SSE endpoint "/details?topic=retry"
+    Then the stream should set a reconnection delay of 2500 ms
+
   Scenario: A backlog larger than the prefetch replays on one connection
     # client_buffer_size is 64 by default; the stream stops pulling while the
     # client catches up instead of disconnecting it.

functional_test/steps_test.go

@@ -51,6 +51,8 @@ type testContext struct {
 	streamNames    map[string]struct{}
 	sseReadDone    chan struct{}
 	natsPaused     bool
+	// sseRetry is the value of the last retry: line of the scenario's stream.
+	sseRetry string
 }
 
 type sseEvent struct {
@@ -458,6 +460,10 @@ func readSSEEvents(body io.Reader, done chan<- struct{}) {
 			tc.sseEventsMutex.Lock()
 			tc.heartbeats = append(tc.heartbeats, line)
 			tc.sseEventsMutex.Unlock()
+		} else if retry, ok := strings.CutPrefix(line, "retry: "); ok {
+			tc.sseEventsMutex.Lock()
+			tc.sseRetry = retry
+			tc.sseEventsMutex.Unlock()
 		}
 	}
 }
@@ -706,6 +712,14 @@ func theResponseShouldContain(text string) error {
 	return nil
 }
 
+func theStreamShouldSetAReconnectionDelayOf(ms string) error {
+	return waitUntil("retry: "+ms, functionalWaitTimeout, func() (bool, string) {
+		tc.sseEventsMutex.Lock()
+		defer tc.sseEventsMutex.Unlock()
+		return tc.sseRetry == ms, "retry: " + tc.sseRetry
+	})
+}
+
 func theResponseShouldNotContain(text string) error {
 	if strings.Contains(tc.httpBody, text) {
 		return fmt.Errorf("response contains %q: %s", text, tc.httpBody)
@@ -1202,6 +1216,7 @@ func InitializeScenario(ctx *godog.ScenarioContext) {
 	ctx.Step(`^I should receive HTTP status (\d+)$`, iShouldReceiveHTTPStatus)
 	ctx.Step(`^the response should contain "([^"]*)"$`, theResponseShouldContain)
 	ctx.Step(`^the response should not contain "([^"]*)"$`, theResponseShouldNotContain)
+	ctx.Step(`^the stream should set a reconnection delay of (\d+) ms$`, theStreamShouldSetAReconnectionDelayOf)
 	ctx.Step(`^the response header "([^"]*)" should be "([^"]*)"$`, theResponseHeaderShouldBe)
 	ctx.Step(`^the SSE response header "([^"]*)" should be "([^"]*)"$`, theSSEResponseHeaderShouldBe)
 	ctx.Step(`^I should receive a heartbeat comment$`, iShouldReceiveAHeartbeatComment)