diff --git a/CHANGELOG.md b/CHANGELOG.md index d1681e5..3f44242 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- `sse_retry ` sends `retry:` with every stream's handshake, + setting EventSource's reconnection delay after ordinary disconnects, and + `transient_retry ` 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 diff --git a/Caddyfile.test b/Caddyfile.test index 5cd6f38..a8b4e25 100644 --- a/Caddyfile.test +++ b/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 } } diff --git a/README.md b/README.md index a6c6af5..645a0b0 100644 --- a/README.md +++ b/README.md @@ -386,6 +386,8 @@ nuts { event_type [name] # Event names: message (default), topic, or header payload_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 # retry: sent with every stream's handshake (default: 0 = none) + transient_retry # Average retry delay after transient failures, jittered ±50% (0=default 5) 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) @@ -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 ` 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 ` 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 diff --git a/caddyfile.go b/caddyfile.go index dcda788..125284c 100644 --- a/caddyfile.go +++ b/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()) diff --git a/caddyfile_test.go b/caddyfile_test.go index 6460bc8..5d3db63 100644 --- a/caddyfile_test.go +++ b/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) + } +} diff --git a/consumer_test.go b/consumer_test.go index db7e8ee..ab641f7 100644 --- a/consumer_test.go +++ b/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) diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 0a76823..0c6b316 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -67,7 +67,7 @@ enforced. For public or browser-facing routes, issue short-lived tokens with | `nats_idle_heartbeat ` | `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 ` | `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 ` | `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 ` | `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 ` | `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 ` | `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 ` | `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 ` | `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`. | +| `sse_retry ` | `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 ` | `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 ` | `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. | diff --git a/docs/TROUBLESHOOTING.md b/docs/TROUBLESHOOTING.md index c0da989..827ba21 100644 --- a/docs/TROUBLESHOOTING.md +++ b/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: diff --git a/features/sse_streaming.feature b/features/sse_streaming.feature index 1669fc9..91ed576 100644 --- a/features/sse_streaming.feature +++ b/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. diff --git a/functional_test/steps_test.go b/functional_test/steps_test.go index b3a8057..de0ce0c 100644 --- a/functional_test/steps_test.go +++ b/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) diff --git a/handler.go b/handler.go index 6519c81..9c9acbe 100644 --- a/handler.go +++ b/handler.go @@ -121,6 +121,17 @@ type Handler struct { // servers, clusters and limits (#144). HealthDetails bool `json:"health_details,omitempty"` + // SSERetry is the reconnection delay, in milliseconds, each stream sends + // as `retry:` with its connected event. EventSource waits that long + // after an ordinary disconnect (a dropped network, a reload, a write + // timeout). Zero sends none, leaving the browser's default (#145). + SSERetry int `json:"sse_retry,omitempty"` + // TransientRetry is the average delay, in seconds, a client is asked to + // wait after a transient failure or a stream closed because its + // JetStream stream was recreated or rewound; the delay sent is jittered + // by ±50%. Zero uses 5 (#145). + TransientRetry int `json:"transient_retry,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/provision.go b/provision.go index f3c38e2..26e4490 100644 --- a/provision.go +++ b/provision.go @@ -216,6 +216,9 @@ func (h *Handler) Provision(ctx caddy.Context) error { if h.PayloadFormat == "" { h.PayloadFormat = payloadFormatEnvelope } + if h.TransientRetry == 0 { + h.TransientRetry = int(transientRetryBase / time.Second) + } if h.HealthPath == "" { h.HealthPath = defaultHealthPath } @@ -870,6 +873,12 @@ func (h *Handler) validateConfigValues() error { default: return fmt.Errorf("event_id_format must be %q or %q, not %q", eventIDSequence, eventIDSequenceTime, h.EventIDFormat) } + if h.SSERetry < 0 || h.SSERetry > maxSSERetryMillis { + return fmt.Errorf("sse_retry must be between 0 and %d milliseconds", maxSSERetryMillis) + } + if h.TransientRetry < 0 || h.TransientRetry > maxTransientRetrySeconds { + return fmt.Errorf("transient_retry must be between 0 and %d seconds", maxTransientRetrySeconds) + } switch h.PayloadFormat { case "", payloadFormatEnvelope, payloadFormatRaw: default: diff --git a/provision_test.go b/provision_test.go index 7b71720..4075dfe 100644 --- a/provision_test.go +++ b/provision_test.go @@ -1194,6 +1194,16 @@ func TestHandler_Provision_RejectsInvalidOptionalJSONConfigBeforeDialing(t *test fragment: `"event_type": "category"`, wantErr: "event_type must be", }, + { + name: "negative sse_retry", + fragment: `"sse_retry": -1`, + wantErr: "sse_retry", + }, + { + name: "negative transient_retry", + fragment: `"transient_retry": -5`, + wantErr: "transient_retry", + }, { name: "unknown payload format", fragment: `"payload_format": "binary"`, diff --git a/serve.go b/serve.go index 287cd06..a6a9724 100644 --- a/serve.go +++ b/serve.go @@ -441,15 +441,27 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyht return h.serveStream(w, r, plan, feed, runtime.shutdown) } -// transientRetryBase is the average delay a client is asked to wait before -// retrying a stream request that failed for a transient reason. The delay -// actually sent is jittered by ±50%, so clients rejected together do not all -// come back together. +// transientRetryBase is the default average delay a client is asked to wait +// before retrying a stream request that failed for a transient reason +// (transient_retry). The delay actually sent is jittered by ±50%, so clients +// rejected together do not all come back together. const transientRetryBase = 5 * time.Second -// retryDelay returns a delay in [transientRetryBase/2, 3*transientRetryBase/2). -func retryDelay() time.Duration { - return transientRetryBase/2 + rand.N(transientRetryBase) +// maxSSERetryMillis and maxTransientRetrySeconds bound sse_retry and +// transient_retry to an hour, past which a client would look dead. +const ( + maxSSERetryMillis = 3600 * 1000 + maxTransientRetrySeconds = 3600 +) + +// retryDelay returns a delay jittered by ±50% around transient_retry: in +// [base/2, 3*base/2). +func (h *Handler) retryDelay() time.Duration { + base := time.Duration(h.TransientRetry) * time.Second + if base <= 0 { + base = transientRetryBase + } + return base/2 + rand.N(base) } // acceptsEventStream reports whether the client asked for an event stream. @@ -472,7 +484,7 @@ func acceptsEventStream(r *http.Request) bool { // reconnects after that delay and keeps its Last-Event-ID. Other clients get // the status code with a Retry-After header. func (h *Handler) rejectTransient(w http.ResponseWriter, r *http.Request, status int, message string) { - delay := retryDelay() + delay := h.retryDelay() if acceptsEventStream(r) { h.setSSEHeaders(w) w.WriteHeader(http.StatusOK) @@ -1123,7 +1135,7 @@ func (h *Handler) endGeneration(w http.ResponseWriter, rc *http.ResponseControll h.log().Info("closing SSE stream: its JetStream stream was recreated or rewound", appendStreamLogFields(plan, zap.String("disconnect_reason", reason))..., ) - frame := fmt.Sprintf(": %s\nretry: %d\nevent: reset\ndata: {\"reason\":%q}\nid: 0\n\n", reason, retryDelay().Milliseconds(), reason) + frame := fmt.Sprintf(": %s\nretry: %d\nevent: reset\ndata: {\"reason\":%q}\nid: 0\n\n", reason, h.retryDelay().Milliseconds(), reason) // Best effort: the stream closes either way, and a client that misses // the frame reconnects with its old cursor. _ = writeSSEChunkWithTimeout(w, rc, frame, writeTimeout) @@ -1285,6 +1297,12 @@ func (h *Handler) eventName(msg streamMessage) string { // first message resumes from there instead of from "now". func (h *Handler) formatConnectedEvent(plan streamPlan) string { var event strings.Builder + if h.SSERetry > 0 { + // Applied by EventSource as soon as it reads the line (#145). + event.WriteString("retry: ") + event.WriteString(strconv.Itoa(h.SSERetry)) + event.WriteByte('\n') + } event.WriteString("event: connected\ndata: {\"topics\":") event.WriteString(toJSON(plan.Topics)) event.WriteString("}\n") diff --git a/serve_test.go b/serve_test.go index d1ba86e..71d746d 100644 --- a/serve_test.go +++ b/serve_test.go @@ -509,6 +509,13 @@ func TestFormatConnectedEvent(t *testing.T) { if got := h.formatConnectedEvent(streamPlan{Topics: []string{"a"}, Replay: start}); got != "event: connected\ndata: {\"topics\":[\"a\"]}\nid: 0\n\n" { t.Fatalf("connected event at position 0 = %q", got) } + + // #145: sse_retry sets EventSource's reconnection delay with the + // handshake. + h = &Handler{SSERetry: 2500} + if got := h.formatConnectedEvent(streamPlan{Topics: []string{"a"}, Replay: replayPlan{Mode: replayModeDeliverNew, StartSequence: 43}}); got != "retry: 2500\nevent: connected\ndata: {\"topics\":[\"a\"]}\nid: 42\n\n" { + t.Fatalf("connected event with sse_retry = %q", got) + } } // TestHandler_PlanSubscriptionChecksTheCursorsMessage covers #138: a cursor @@ -1288,17 +1295,62 @@ func TestHandler_ParseStreamRequest_CursorPrecedence(t *testing.T) { } } +// TestRetryDelay_StaysWithinTheJitterRange: the retry delay is jittered by +// ±50% around transient_retry, 5 s by default (#145). func TestRetryDelay_StaysWithinTheJitterRange(t *testing.T) { - seen := map[time.Duration]bool{} - for i := 0; i < 1000; i++ { - d := retryDelay() - if d < transientRetryBase/2 || d >= transientRetryBase*3/2 { - t.Fatalf("retryDelay() = %v, want [%v, %v)", d, transientRetryBase/2, transientRetryBase*3/2) + for _, c := range []struct { + retry int + base time.Duration + }{ + {0, transientRetryBase}, + {2, 2 * time.Second}, + {30, 30 * time.Second}, + } { + h := &Handler{TransientRetry: c.retry} + seen := map[time.Duration]bool{} + below, above := 0, 0 + for i := 0; i < 1000; i++ { + d := h.retryDelay() + if d < c.base/2 || d >= c.base*3/2 { + t.Fatalf("transient_retry %d: retryDelay() = %v, want [%v, %v)", c.retry, d, c.base/2, c.base*3/2) + } + seen[d] = true + if d < c.base { + below++ + } else { + above++ + } + } + if len(seen) < 100 { + t.Fatalf("transient_retry %d: %d distinct values in 1000 calls; want jitter", c.retry, len(seen)) + } + // Both halves of the range: the average is the base itself. + if below < 300 || above < 300 { + t.Fatalf("transient_retry %d: %d delays below %v and %d above; want both around 500", c.retry, below, c.base, above) } - seen[d] = true } - if len(seen) < 100 { - t.Fatalf("retryDelay() produced %d distinct values in 1000 calls; want jitter", len(seen)) +} + +// TestHandler_RejectTransientFollowsTransientRetry: both the retry stream's +// delay and Retry-After follow transient_retry (#145). +func TestHandler_RejectTransientFollowsTransientRetry(t *testing.T) { + h := &Handler{TransientRetry: 2, logger: zap.NewNop()} + for range 50 { + rr := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/events?topic=a", nil) + req.Header.Set("Accept", "text/event-stream") + h.rejectTransient(rr, req, http.StatusServiceUnavailable, "JetStream not available") + ms, ok := strings.CutPrefix(strings.Split(rr.Body.String(), "\n")[1], "retry: ") + delay, err := strconv.Atoi(ms) + if !ok || err != nil || delay < 1000 || delay >= 3000 { + t.Fatalf("retry stream %q, want a retry in [1000, 3000) ms", rr.Body.String()) + } + + rr = httptest.NewRecorder() + h.rejectTransient(rr, httptest.NewRequest(http.MethodGet, "/events?topic=a", nil), http.StatusServiceUnavailable, "JetStream not available") + if after, err := strconv.Atoi(rr.Header().Get("Retry-After")); err != nil || after < 1 || after > 3 { + t.Fatalf("Retry-After = %q, want 1 to 3 s", rr.Header().Get("Retry-After")) + } } } diff --git a/website/docs/usage.md b/website/docs/usage.md index e5c15de..7f3c5d1 100644 --- a/website/docs/usage.md +++ b/website/docs/usage.md @@ -66,6 +66,8 @@ nuts { event_type [name] # Event names: message (default), topic, or header payload_format # Event data: envelope (default) or raw health_details [true|false] # Readiness probes add NATS server and stream details + sse_retry # retry: sent with every stream's handshake (default: none) + transient_retry # Average retry delay after transient failures (default: 5) # Health probes live_path # Liveness probe (default: /livez) @@ -135,7 +137,7 @@ default, or a negative value to disable the limit entirely. 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 +jittered `Retry-After` of 3–8 seconds by default (`transient_retry`), and `nuts_connections_rejected_total{reason="max_connections"}` increments. Browser `EventSource` clients get a `200` stream with a `retry:` delay instead, so they keep reconnecting. A client-side concurrency cap is