From 23b57c019b30acfcf9c6ed409f4624a8e7120309 Mon Sep 17 00:00:00 2001 From: Matt Wisner Date: Thu, 20 Aug 2026 06:24:37 -0400 Subject: [PATCH 1/3] fix(router): add SSE server write timeout --- .../events/kafka_sse_write_timeout_test.go | 249 ++++++++++++++++++ router/core/graph_server.go | 1 + router/core/graphql_handler.go | 21 +- router/core/subscription_response_writer.go | 100 +++++-- .../core/subscription_response_writer_test.go | 101 ++++++- router/pkg/config/config.go | 1 + router/pkg/config/config.schema.json | 6 + router/pkg/config/fixtures/full.yaml | 1 + .../pkg/config/testdata/config_defaults.json | 1 + router/pkg/config/testdata/config_full.json | 1 + 10 files changed, 446 insertions(+), 36 deletions(-) create mode 100644 router-tests/events/kafka_sse_write_timeout_test.go diff --git a/router-tests/events/kafka_sse_write_timeout_test.go b/router-tests/events/kafka_sse_write_timeout_test.go new file mode 100644 index 0000000000..e8148a7557 --- /dev/null +++ b/router-tests/events/kafka_sse_write_timeout_test.go @@ -0,0 +1,249 @@ +package events_test + +import ( + "bufio" + "context" + "errors" + "net/http" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/wundergraph/cosmo/router-tests/events" + "github.com/wundergraph/cosmo/router-tests/testenv" + "github.com/wundergraph/cosmo/router/core" + nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" + "github.com/wundergraph/cosmo/router/pkg/config" +) + +const blockSSEWriteHeader = "X-Test-Block-SSE-Write" + +var ( + _ core.Module = (*blockingSSEWriterModule)(nil) + _ core.RouterOnRequestHandler = (*blockingSSEWriterModule)(nil) +) + +// blockingSSEWriterModule simulates a client that stops draining its SSE +// connection without closing it. The wrapped writer only returns when the +// router sets a write deadline or the test releases it during cleanup. +type blockingSSEWriterModule struct { + armed *atomic.Bool + writeStarted chan struct{} + startedOnce *sync.Once + release chan struct{} +} + +func (m *blockingSSEWriterModule) Module() core.ModuleInfo { + return core.ModuleInfo{ + ID: "blockingSSEWriterModule", + Priority: 1, + New: func() core.Module { + return &blockingSSEWriterModule{ + armed: m.armed, + writeStarted: m.writeStarted, + startedOnce: m.startedOnce, + release: m.release, + } + }, + } +} + +func (m *blockingSSEWriterModule) RouterOnRequest(ctx core.RequestContext, next http.Handler) { + if ctx.Request().Header.Get(blockSSEWriteHeader) != "true" { + next.ServeHTTP(ctx.ResponseWriter(), ctx.Request()) + return + } + + next.ServeHTTP(&deadlineBlockingResponseWriter{ + ResponseWriter: ctx.ResponseWriter(), + armed: m.armed, + writeStarted: m.writeStarted, + startedOnce: m.startedOnce, + release: m.release, + }, ctx.Request()) +} + +type deadlineBlockingResponseWriter struct { + http.ResponseWriter + armed *atomic.Bool + writeStarted chan struct{} + startedOnce *sync.Once + release chan struct{} + deadlineNanos atomic.Int64 +} + +func (w *deadlineBlockingResponseWriter) Write(data []byte) (int, error) { + if !w.armed.CompareAndSwap(true, false) { + return w.ResponseWriter.Write(data) + } + + w.startedOnce.Do(func() { close(w.writeStarted) }) + deadlineNanos := w.deadlineNanos.Load() + if deadlineNanos == 0 { + <-w.release + return 0, os.ErrDeadlineExceeded + } + + wait := time.Until(time.Unix(0, deadlineNanos)) + if wait <= 0 { + return 0, os.ErrDeadlineExceeded + } + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case <-w.release: + return 0, os.ErrDeadlineExceeded + case <-timer.C: + return 0, os.ErrDeadlineExceeded + } +} + +func (w *deadlineBlockingResponseWriter) Flush() { + if flusher, ok := w.ResponseWriter.(http.Flusher); ok { + flusher.Flush() + } +} + +func (w *deadlineBlockingResponseWriter) FlushError() error { + if flusher, ok := w.ResponseWriter.(interface{ FlushError() error }); ok { + return flusher.FlushError() + } + w.Flush() + return nil +} + +func (w *deadlineBlockingResponseWriter) SetWriteDeadline(deadline time.Time) error { + w.deadlineNanos.Store(deadline.UnixNano()) + return nil +} + +func (w *deadlineBlockingResponseWriter) Unwrap() http.ResponseWriter { + return w.ResponseWriter +} + +func TestKafkaSubscriptionRecoversAfterSSEWriteTimeout(t *testing.T) { + if testing.Short() { + t.Skip("skipping Kafka integration test in short mode") + } + + const topic = "employeeUpdated-sse-write-timeout" + armed := &atomic.Bool{} + writeStarted := make(chan struct{}) + release := make(chan struct{}) + var releaseOnce sync.Once + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) }) + + module := &blockingSSEWriterModule{ + armed: armed, + writeStarted: writeStarted, + startedOnce: &sync.Once{}, + release: release, + } + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate, + EnableKafka: true, + RouterOptions: []core.Option{core.WithCustomModules(module)}, + ModifyRouterConfig: func(routerConfig *nodev1.RouterConfig) { + overrideKafkaTopicsForField(t, routerConfig, "employeeUpdatedMyKafka", + []string{"employeeUpdated", "employeeUpdatedTwo"}, topic) + }, + ModifyEngineExecutionConfiguration: func(cfg *config.EngineExecutionConfiguration) { + cfg.SSEServerWriteTimeout = 100 * time.Millisecond + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + events.KafkaEnsureTopicExists(t, xEnv, EventWaitTimeout, topic) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + client := &http.Client{} + blockedResp := openSSESubscription(t, ctx, client, xEnv.GraphQLRequestURL(), true) + defer blockedResp.Body.Close() + healthyResp := openSSESubscription(t, ctx, client, xEnv.GraphQLRequestURL(), false) + defer healthyResp.Body.Close() + healthyReader := bufio.NewReader(healthyResp.Body) + + xEnv.WaitForSubscriptionCount(2, EventWaitTimeout) + xEnv.WaitForTriggerCount(1, EventWaitTimeout) + + armed.Store(true) + xEnv.KafkaPublishUntilReceived(topic, + `{"__typename":"Employee","id":1,"update":{"name":"blocked"}}`, 1, EventWaitTimeout) + + select { + case <-writeStarted: + case <-time.After(EventWaitTimeout): + t.Fatal("timed out waiting for the SSE write to block") + } + + require.Contains(t, readSSEData(t, healthyReader), `"id":1`) + + events.ProduceKafkaMessage(t, xEnv, EventWaitTimeout, topic, + `{"__typename":"Employee","id":2,"update":{"name":"recovery"}}`) + + recovery := make(chan string, 1) + go func() { + data, err := readSSEDataLine(healthyReader) + if err != nil { + recovery <- "error: " + err.Error() + return + } + recovery <- data + }() + + select { + case data := <-recovery: + require.Contains(t, data, `"id":2`) + case <-time.After(EventWaitTimeout): + t.Fatal("healthy subscription did not receive the queued event after the SSE write deadline") + } + + xEnv.WaitForSubscriptionCount(1, EventWaitTimeout) + }) +} + +func openSSESubscription(t *testing.T, ctx context.Context, client *http.Client, url string, blocked bool) *http.Response { + t.Helper() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, + strings.NewReader(`{"query":"subscription { employeeUpdatedMyKafka(employeeID: 3) { id } }"}`)) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + if blocked { + req.Header.Set(blockSSEWriteHeader, "true") + } + + resp, err := client.Do(req) + require.NoError(t, err) + require.Equal(t, http.StatusOK, resp.StatusCode) + require.Equal(t, "text/event-stream", resp.Header.Get("Content-Type")) + return resp +} + +func readSSEData(t *testing.T, reader *bufio.Reader) string { + t.Helper() + data, err := readSSEDataLine(reader) + require.NoError(t, err) + return data +} + +func readSSEDataLine(reader *bufio.Reader) (string, error) { + for { + line, err := reader.ReadString('\n') + if err != nil { + return "", err + } + line = strings.TrimSpace(line) + if strings.HasPrefix(line, "data: ") { + return strings.TrimPrefix(line, "data: "), nil + } + if strings.HasPrefix(line, "event: complete") { + return "", errors.New("subscription completed before receiving data") + } + } +} diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 3cc45d5f9e..e4876a514d 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -1806,6 +1806,7 @@ func (s *graphServer) buildGraphMux( SubgraphErrorPropagation: s.subgraphErrorPropagation, EngineLoaderHooks: loaderHooks, HeaderPropagation: s.headerPropagation, + SSEServerWriteTimeout: s.engineExecutionConfiguration.SSEServerWriteTimeout, } if s.redisClient != nil { diff --git a/router/core/graphql_handler.go b/router/core/graphql_handler.go index 4ef92da46b..34ae0f8a3c 100644 --- a/router/core/graphql_handler.go +++ b/router/core/graphql_handler.go @@ -10,6 +10,7 @@ import ( "net/http" "strconv" "strings" + "time" otelmetric "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/trace" @@ -87,6 +88,7 @@ type HandlerOptions struct { EnableCostResponseHeaders bool ApolloSubscriptionMultipartPrintBoundary bool + SSEServerWriteTimeout time.Duration HeaderPropagation *HeaderPropagation } @@ -109,6 +111,7 @@ func NewGraphQLHandler(opts HandlerOptions) *GraphQLHandler { subgraphErrorPropagation: opts.SubgraphErrorPropagation, engineLoaderHooks: opts.EngineLoaderHooks, apolloSubscriptionMultipartPrintBoundary: opts.ApolloSubscriptionMultipartPrintBoundary, + sseServerWriteTimeout: opts.SSEServerWriteTimeout, headerPropagation: opts.HeaderPropagation, } return graphQLHandler @@ -143,6 +146,7 @@ type GraphQLHandler struct { enableCostResponseHeaders bool apolloSubscriptionMultipartPrintBoundary bool + sseServerWriteTimeout time.Duration } func (h *GraphQLHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -284,21 +288,24 @@ func (h *GraphQLHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } case *plan.SubscriptionResponsePlan: var ( - writer resolve.SubscriptionResponseWriter - ok bool + writer resolve.SubscriptionResponseWriter + writerErr error ) h.setDebugCacheHeaders(w, reqCtx.operation) defer propagateSubgraphErrors(resolveCtx) - resolveCtx, writer, ok = GetSubscriptionResponseWriter(resolveCtx, r, w, h.apolloSubscriptionMultipartPrintBoundary) - if !ok { - reqCtx.logger.Error("unable to get subscription response writer", zap.Error(errCouldNotFlushResponse)) - trackFinalResponseError(r.Context(), errCouldNotFlushResponse) + resolveCtx, writer, writerErr = GetSubscriptionResponseWriter(resolveCtx, r, w, SubscriptionResponseWriterOptions{ + ApolloSubscriptionMultipartPrintBoundary: h.apolloSubscriptionMultipartPrintBoundary, + SSEWriteTimeout: h.sseServerWriteTimeout, + }) + if writerErr != nil { + reqCtx.logger.Error("unable to get subscription response writer", zap.Error(writerErr)) + trackFinalResponseError(r.Context(), writerErr) writeRequestErrors(writeRequestErrorsParams{ request: r, writer: w, statusCode: http.StatusInternalServerError, - requestErrors: graphqlerrors.RequestErrorsFromError(errCouldNotFlushResponse), + requestErrors: graphqlerrors.RequestErrorsFromError(writerErr), logger: reqCtx.logger, headerPropagation: h.headerPropagation, }) diff --git a/router/core/subscription_response_writer.go b/router/core/subscription_response_writer.go index abe951a380..5b87b2043a 100644 --- a/router/core/subscription_response_writer.go +++ b/router/core/subscription_response_writer.go @@ -3,11 +3,14 @@ package core import ( "bytes" "context" + "errors" + "fmt" "io" "mime" "net/http" "strconv" "strings" + "time" "github.com/wundergraph/astjson" "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve" @@ -31,16 +34,23 @@ type withFlushWriter interface { SubscriptionResponseWriter() resolve.SubscriptionResponseWriter } +type SubscriptionResponseWriterOptions struct { + ApolloSubscriptionMultipartPrintBoundary bool + SSEWriteTimeout time.Duration +} + type HttpFlushWriter struct { - ctx context.Context - cancel context.CancelFunc - writer io.Writer - flusher http.Flusher - subscribeOnce bool - sse bool - multipart bool - buf *bytes.Buffer - firstMessage bool + ctx context.Context + cancel context.CancelFunc + writer io.Writer + flusher http.Flusher + responseControl *http.ResponseController + subscribeOnce bool + sse bool + multipart bool + buf *bytes.Buffer + firstMessage bool + sseWriteTimeout time.Duration // apolloSubscriptionMultipartPrintBoundary if set to true will send the multipart boundary at the end of the message to allow // misbehaving client (like apollo client) to read the message just sent before the next one or the heartbeat apolloSubscriptionMultipartPrintBoundary bool @@ -53,7 +63,10 @@ func (f *HttpFlushWriter) Complete() { return } if f.sse { - _, _ = f.writer.Write([]byte("event: complete\ndata: \n\n")) + _ = f.writeAndFlushSSE(func() error { + _, err := f.writer.Write([]byte("event: complete\ndata: \n\n")) + return err + }) } else if f.multipart { // Write the final boundary in the multipart response if f.apolloSubscriptionMultipartPrintBoundary { @@ -63,8 +76,10 @@ func (f *HttpFlushWriter) Complete() { } } - // Flush before closing the writer to ensure all data is sent - f.flusher.Flush() + if !f.sse { + // Flush before closing the writer to ensure all data is sent. + f.flusher.Flush() + } f.cancel() } @@ -85,12 +100,10 @@ func (f *HttpFlushWriter) Heartbeat() error { var heartbeat []byte if f.sse { heartbeat = []byte(":heartbeat\n\n") - - if _, err := f.writer.Write(heartbeat); err != nil { + return f.writeAndFlushSSE(func() error { + _, err := f.writer.Write(heartbeat) return err - } - - f.flusher.Flush() + }) } else if f.multipart { if _, err := f.Write([]byte("{}")); err != nil { return err @@ -151,14 +164,22 @@ func (f *HttpFlushWriter) Flush() (err error) { } full := flushBreak + string(resp) + separation - _, err = f.writer.Write([]byte(full)) + if f.sse { + err = f.writeAndFlushSSE(func() error { + _, writeErr := f.writer.Write([]byte(full)) + return writeErr + }) + } else { + _, err = f.writer.Write([]byte(full)) + if err == nil { + // Flush before closing the writer to ensure all data is sent. + f.flusher.Flush() + } + } if err != nil { return err } - // Flush before closing the writer to ensure all data is sent - f.flusher.Flush() - if f.subscribeOnce { defer f.cancel() } @@ -166,15 +187,31 @@ func (f *HttpFlushWriter) Flush() (err error) { return nil } -func GetSubscriptionResponseWriter(ctx *resolve.Context, r *http.Request, w http.ResponseWriter, apolloSubscriptionMultipartPrintBoundary bool) (*resolve.Context, resolve.SubscriptionResponseWriter, bool) { +func (f *HttpFlushWriter) writeAndFlushSSE(write func() error) error { + if f.sseWriteTimeout > 0 { + if err := f.responseControl.SetWriteDeadline(time.Now().Add(f.sseWriteTimeout)); err != nil { + // Failing closed prevents a response writer without deadline support from + // reintroducing an unbounded shared-trigger stall. + return fmt.Errorf("set SSE write deadline: %w", err) + } + } + + if err := write(); err != nil { + return err + } + + return f.responseControl.Flush() +} + +func GetSubscriptionResponseWriter(ctx *resolve.Context, r *http.Request, w http.ResponseWriter, opts SubscriptionResponseWriterOptions) (*resolve.Context, resolve.SubscriptionResponseWriter, error) { if wfw, ok := w.(withFlushWriter); ok { - return ctx, wfw.SubscriptionResponseWriter(), true + return ctx, wfw.SubscriptionResponseWriter(), nil } wgParams := NegotiateSubscriptionParams(r, false) flusher, ok := w.(http.Flusher) if !ok { - return ctx, nil, false + return ctx, nil, errors.New("subscription response writer does not support flushing") } setSubscriptionHeaders(wgParams, r, w) @@ -182,12 +219,14 @@ func GetSubscriptionResponseWriter(ctx *resolve.Context, r *http.Request, w http flushWriter := &HttpFlushWriter{ writer: w, flusher: flusher, + responseControl: http.NewResponseController(w), sse: wgParams.UseSse, multipart: wgParams.UseMultipart, subscribeOnce: wgParams.SubscribeOnce, buf: &bytes.Buffer{}, firstMessage: true, - apolloSubscriptionMultipartPrintBoundary: apolloSubscriptionMultipartPrintBoundary, + sseWriteTimeout: opts.SSEWriteTimeout, + apolloSubscriptionMultipartPrintBoundary: opts.ApolloSubscriptionMultipartPrintBoundary, } flushWriter.ctx, flushWriter.cancel = context.WithCancel(ctx.Context()) @@ -197,10 +236,17 @@ func GetSubscriptionResponseWriter(ctx *resolve.Context, r *http.Request, w http ctx.ExecutionOptions.SendHeartbeat = true // Flush the response head immediately so the client establishes the connection // before the first message, instead of blocking until one is streamed. - flusher.Flush() + if wgParams.UseSse { + if err := flushWriter.writeAndFlushSSE(func() error { return nil }); err != nil { + flushWriter.cancel() + return ctx, nil, fmt.Errorf("flush initial SSE response headers: %w", err) + } + } else { + flusher.Flush() + } } - return ctx, flushWriter, true + return ctx, flushWriter, nil } func wrapMultipartMessage(resp []byte, wrapPayload bool) ([]byte, error) { diff --git a/router/core/subscription_response_writer_test.go b/router/core/subscription_response_writer_test.go index 02db6b7400..f219412cfd 100644 --- a/router/core/subscription_response_writer_test.go +++ b/router/core/subscription_response_writer_test.go @@ -2,16 +2,41 @@ package core import ( "context" + "errors" "net/http" "net/http/httptest" "net/url" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve" ) +type deadlineRecorder struct { + *httptest.ResponseRecorder + deadlines []time.Time + deadlineErr error + flushErr error +} + +func (r *deadlineRecorder) SetWriteDeadline(deadline time.Time) error { + if r.deadlineErr != nil { + return r.deadlineErr + } + r.deadlines = append(r.deadlines, deadline) + return nil +} + +func (r *deadlineRecorder) FlushError() error { + if r.flushErr != nil { + return r.flushErr + } + r.Flush() + return nil +} + func TestNegotiateSubscriptionParams(t *testing.T) { type args struct { r *http.Request @@ -137,10 +162,82 @@ func TestGetSubscriptionResponseWriter(t *testing.T) { req := httptest.NewRequest(http.MethodPost, "/graphql", nil) req.Header.Set("Accept", sseMimeType) - _, _, ok := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, false) - require.True(t, ok) + _, _, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{}) + require.NoError(t, err) assert.Equal(t, sseMimeType, recorder.Header().Get("Content-Type")) assert.True(t, recorder.Flushed, "expected the SSE response head to be flushed before any message is written") }) + + t.Run("sets a fresh deadline for every SSE write and flush", func(t *testing.T) { + recorder := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()} + req := httptest.NewRequest(http.MethodPost, "/graphql", nil) + req.Header.Set("Accept", sseMimeType) + + _, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{SSEWriteTimeout: time.Second}) + require.NoError(t, err) + require.Len(t, recorder.deadlines, 1, "expected the initial header flush to have a deadline") + + _, err = writer.Write([]byte(`{"data":{"id":1}}`)) + require.NoError(t, err) + require.NoError(t, writer.Flush()) + require.Len(t, recorder.deadlines, 2, "expected the data frame to refresh the deadline") + assert.False(t, recorder.deadlines[1].Before(recorder.deadlines[0])) + + require.NoError(t, writer.Heartbeat()) + require.Len(t, recorder.deadlines, 3, "expected the heartbeat to refresh the deadline") + + writer.Complete() + require.Len(t, recorder.deadlines, 4, "expected the completion frame to refresh the deadline") + }) + + t.Run("propagates an SSE flush error", func(t *testing.T) { + recorder := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()} + req := httptest.NewRequest(http.MethodPost, "/graphql", nil) + req.Header.Set("Accept", sseMimeType) + + _, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{}) + require.NoError(t, err) + + flushErr := errors.New("flush failed") + recorder.flushErr = flushErr + require.ErrorIs(t, writer.Heartbeat(), flushErr) + }) + + t.Run("propagates an SSE deadline error", func(t *testing.T) { + recorder := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()} + req := httptest.NewRequest(http.MethodPost, "/graphql", nil) + req.Header.Set("Accept", sseMimeType) + + _, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{SSEWriteTimeout: time.Second}) + require.NoError(t, err) + + deadlineErr := errors.New("deadline failed") + recorder.deadlineErr = deadlineErr + err = writer.Heartbeat() + assert.ErrorIs(t, err, deadlineErr) + assert.ErrorContains(t, err, "set SSE write deadline") + }) + + t.Run("fails closed when an SSE deadline is configured but unsupported", func(t *testing.T) { + recorder := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/graphql", nil) + req.Header.Set("Accept", sseMimeType) + + _, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{SSEWriteTimeout: time.Second}) + require.Error(t, err) + assert.ErrorIs(t, err, http.ErrNotSupported) + assert.ErrorContains(t, err, "set SSE write deadline") + assert.Nil(t, writer) + }) + + t.Run("does not require deadline support when the timeout is disabled", func(t *testing.T) { + recorder := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/graphql", nil) + req.Header.Set("Accept", sseMimeType) + + _, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{}) + require.NoError(t, err) + assert.NotNil(t, writer) + }) } diff --git a/router/pkg/config/config.go b/router/pkg/config/config.go index 90d99e551c..3fa09b9bca 100644 --- a/router/pkg/config/config.go +++ b/router/pkg/config/config.go @@ -500,6 +500,7 @@ type EngineExecutionConfiguration struct { DisableVariablesRemapping bool `envDefault:"false" env:"ENGINE_DISABLE_VARIABLES_REMAPPING" yaml:"disable_variables_remapping"` EnableRequireFetchReasons bool `envDefault:"false" env:"ENGINE_ENABLE_REQUIRE_FETCH_REASONS" yaml:"enable_require_fetch_reasons"` SubscriptionFetchTimeout time.Duration `envDefault:"30s" env:"ENGINE_SUBSCRIPTION_FETCH_TIMEOUT" yaml:"subscription_fetch_timeout,omitempty"` + SSEServerWriteTimeout time.Duration `envDefault:"0s" env:"ENGINE_SSE_SERVER_WRITE_TIMEOUT" yaml:"sse_server_write_timeout,omitempty"` EnableDefer bool `envDefault:"false" env:"ENGINE_ENABLE_DEFER" yaml:"enable_defer"` // EnableMultiFetch merges entity fetches to the same subgraph that execute diff --git a/router/pkg/config/config.schema.json b/router/pkg/config/config.schema.json index 069292f78e..1aa18493c5 100644 --- a/router/pkg/config/config.schema.json +++ b/router/pkg/config/config.schema.json @@ -4151,6 +4151,12 @@ "default": "30s", "description": "The maximum time a subscription fetch can take before it is considered timed out. The period is specified as a string with a number and a unit, e.g. 10ms, 1s, 1m, 1h. The supported units are 'ms', 's', 'm', 'h'." }, + "sse_server_write_timeout": { + "type": "string", + "format": "go-duration", + "default": "0s", + "description": "The maximum time allowed for each downstream SSE write and flush. When exceeded, the affected SSE subscription is terminated so it cannot indefinitely block other subscriptions sharing a trigger. A value of 0s disables the deadline." + }, "enable_defer": { "type": "boolean", "default": false, diff --git a/router/pkg/config/fixtures/full.yaml b/router/pkg/config/fixtures/full.yaml index b953f28ae4..46afdb6c24 100644 --- a/router/pkg/config/fixtures/full.yaml +++ b/router/pkg/config/fixtures/full.yaml @@ -450,6 +450,7 @@ engine: websocket_client_write_timeout: 10s websocket_server_read_timeout: 5s websocket_server_write_timeout: 10s + sse_server_write_timeout: 10s websocket_server_poll_timeout: 1s websocket_server_conn_buffer_size: 128 websocket_client_read_limit: 1MB diff --git a/router/pkg/config/testdata/config_defaults.json b/router/pkg/config/testdata/config_defaults.json index ad7e252cf8..36751bb81a 100644 --- a/router/pkg/config/testdata/config_defaults.json +++ b/router/pkg/config/testdata/config_defaults.json @@ -519,6 +519,7 @@ "DisableVariablesRemapping": false, "EnableRequireFetchReasons": false, "SubscriptionFetchTimeout": 30000000000, + "SSEServerWriteTimeout": 0, "EnableDefer": false, "EnableMultiFetch": false, "EnableScheduleFetches": false, diff --git a/router/pkg/config/testdata/config_full.json b/router/pkg/config/testdata/config_full.json index f610b51c11..189a37e0eb 100644 --- a/router/pkg/config/testdata/config_full.json +++ b/router/pkg/config/testdata/config_full.json @@ -987,6 +987,7 @@ "DisableVariablesRemapping": false, "EnableRequireFetchReasons": false, "SubscriptionFetchTimeout": 30000000000, + "SSEServerWriteTimeout": 10000000000, "EnableDefer": false, "EnableMultiFetch": false, "EnableScheduleFetches": false, From 90ad6b54cbd8d2fc0e6867e27870e5dabb8ecfab Mon Sep 17 00:00:00 2001 From: Matt Wisner Date: Thu, 20 Aug 2026 06:40:11 -0400 Subject: [PATCH 2/3] fix(router): address SSE timeout review feedback --- .../events/kafka_sse_write_timeout_test.go | 7 ++--- router/pkg/config/config.go | 4 +++ router/pkg/config/config.schema.json | 3 ++ router/pkg/config/config_test.go | 30 +++++++++++++++++++ router/pkg/config/json_schema.go | 24 ++++++++------- 5 files changed, 54 insertions(+), 14 deletions(-) diff --git a/router-tests/events/kafka_sse_write_timeout_test.go b/router-tests/events/kafka_sse_write_timeout_test.go index e8148a7557..a7c602a294 100644 --- a/router-tests/events/kafka_sse_write_timeout_test.go +++ b/router-tests/events/kafka_sse_write_timeout_test.go @@ -183,8 +183,9 @@ func TestKafkaSubscriptionRecoversAfterSSEWriteTimeout(t *testing.T) { require.Contains(t, readSSEData(t, healthyReader), `"id":1`) - events.ProduceKafkaMessage(t, xEnv, EventWaitTimeout, topic, - `{"__typename":"Employee","id":2,"update":{"name":"recovery"}}`) + xEnv.WaitForSubscriptionCount(1, EventWaitTimeout) + xEnv.KafkaPublishUntilReceived(topic, + `{"__typename":"Employee","id":2,"update":{"name":"recovery"}}`, 1, EventWaitTimeout) recovery := make(chan string, 1) go func() { @@ -202,8 +203,6 @@ func TestKafkaSubscriptionRecoversAfterSSEWriteTimeout(t *testing.T) { case <-time.After(EventWaitTimeout): t.Fatal("healthy subscription did not receive the queued event after the SSE write deadline") } - - xEnv.WaitForSubscriptionCount(1, EventWaitTimeout) }) } diff --git a/router/pkg/config/config.go b/router/pkg/config/config.go index 3fa09b9bca..60aa2256f2 100644 --- a/router/pkg/config/config.go +++ b/router/pkg/config/config.go @@ -1722,6 +1722,10 @@ func LoadConfig(configFilePaths []string) (*LoadResult, error) { } } + if cfg.Config.EngineExecutionConfiguration.SSEServerWriteTimeout < 0 { + return nil, errors.New("engine.sse_server_write_timeout must be greater or equal to 0s") + } + // Post-process the config if cfg.Config.DevelopmentMode { cfg.Config.JSONLog = false diff --git a/router/pkg/config/config.schema.json b/router/pkg/config/config.schema.json index 1aa18493c5..15d8b15f1e 100644 --- a/router/pkg/config/config.schema.json +++ b/router/pkg/config/config.schema.json @@ -4154,6 +4154,9 @@ "sse_server_write_timeout": { "type": "string", "format": "go-duration", + "duration": { + "minimum": "0s" + }, "default": "0s", "description": "The maximum time allowed for each downstream SSE write and flush. When exceeded, the affected SSE subscription is terminated so it cannot indefinitely block other subscriptions sharing a trigger. A value of 0s disables the deadline." }, diff --git a/router/pkg/config/config_test.go b/router/pkg/config/config_test.go index e05dca1228..da7db70f1d 100644 --- a/router/pkg/config/config_test.go +++ b/router/pkg/config/config_test.go @@ -478,6 +478,36 @@ telemetry: require.Equal(t, "at '/telemetry/tracing/exporters/0/export_timeout': duration must be less or equal than 2m0s", js.Causes[0].Error()) } +func TestSSEServerWriteTimeoutRejectsNegativeValues(t *testing.T) { + t.Run("config file", func(t *testing.T) { + f := createTempFileFromFixture(t, ` +version: "1" + +graph: + token: "token" + +engine: + sse_server_write_timeout: -1s +`) + + _, err := LoadConfig([]string{f}) + require.ErrorContains(t, err, "duration must be greater or equal than 0s") + }) + + t.Run("environment variable", func(t *testing.T) { + t.Setenv("ENGINE_SSE_SERVER_WRITE_TIMEOUT", "-1s") + f := createTempFileFromFixture(t, ` +version: "1" + +graph: + token: "token" +`) + + _, err := LoadConfig([]string{f}) + require.EqualError(t, err, "engine.sse_server_write_timeout must be greater or equal to 0s") + }) +} + func TestLoadFullConfig(t *testing.T) { t.Parallel() diff --git a/router/pkg/config/json_schema.go b/router/pkg/config/json_schema.go index 46bc4432a9..dc0675f6fe 100644 --- a/router/pkg/config/json_schema.go +++ b/router/pkg/config/json_schema.go @@ -27,8 +27,10 @@ const ( ) type duration struct { - min time.Duration - max time.Duration + min time.Duration + max time.Duration + hasMin bool + hasMax bool } func (d duration) Validate(ctx *jsonschema.ValidatorContext, v any) { @@ -51,7 +53,7 @@ func (d duration) Validate(ctx *jsonschema.ValidatorContext, v any) { return } - if d.min > 0 { + if d.hasMin { if duration < d.min { ctx.AddError(&validationErrorKind{ fmt.Sprintf("duration must be greater or equal than %s", d.min), @@ -61,7 +63,7 @@ func (d duration) Validate(ctx *jsonschema.ValidatorContext, v any) { } } - if d.max > 0 { + if d.hasMax { if duration > d.max { ctx.AddError(&validationErrorKind{ fmt.Sprintf("duration must be less or equal than %s", d.max), @@ -118,23 +120,25 @@ func compileDuration(ctx *jsonschema.CompilerContext, m map[string]any) (jsonsch var minDuration, maxDuration time.Duration var err error - minDurationString, ok := mapVal["minimum"].(string) - if ok { + minDurationString, hasMin := mapVal["minimum"].(string) + if hasMin { minDuration, err = time.ParseDuration(minDurationString) if err != nil { return nil, err } } - maxDurationString, ok := mapVal["maximum"].(string) - if ok { + maxDurationString, hasMax := mapVal["maximum"].(string) + if hasMax { maxDuration, err = time.ParseDuration(maxDurationString) if err != nil { return nil, err } } return duration{ - min: minDuration, - max: maxDuration, + min: minDuration, + max: maxDuration, + hasMin: hasMin, + hasMax: hasMax, }, nil } From 75bc980627a04e98087fbd29d09ca9a4d1f156e7 Mon Sep 17 00:00:00 2001 From: Matt Wisner Date: Fri, 21 Aug 2026 08:05:00 -0400 Subject: [PATCH 3/3] test(router): verify SSE timeout unblocks shared trigger --- .../events/kafka_sse_write_timeout_test.go | 123 +++++++++++++----- 1 file changed, 87 insertions(+), 36 deletions(-) diff --git a/router-tests/events/kafka_sse_write_timeout_test.go b/router-tests/events/kafka_sse_write_timeout_test.go index a7c602a294..b0fd75b98d 100644 --- a/router-tests/events/kafka_sse_write_timeout_test.go +++ b/router-tests/events/kafka_sse_write_timeout_test.go @@ -32,10 +32,13 @@ var ( // connection without closing it. The wrapped writer only returns when the // router sets a write deadline or the test releases it during cleanup. type blockingSSEWriterModule struct { - armed *atomic.Bool - writeStarted chan struct{} - startedOnce *sync.Once - release chan struct{} + armed *atomic.Bool + writeStarted chan struct{} + startedOnce *sync.Once + writeReturned *atomic.Bool + returned chan struct{} + returnedOnce *sync.Once + release chan struct{} } func (m *blockingSSEWriterModule) Module() core.ModuleInfo { @@ -44,10 +47,13 @@ func (m *blockingSSEWriterModule) Module() core.ModuleInfo { Priority: 1, New: func() core.Module { return &blockingSSEWriterModule{ - armed: m.armed, - writeStarted: m.writeStarted, - startedOnce: m.startedOnce, - release: m.release, + armed: m.armed, + writeStarted: m.writeStarted, + startedOnce: m.startedOnce, + writeReturned: m.writeReturned, + returned: m.returned, + returnedOnce: m.returnedOnce, + release: m.release, } }, } @@ -64,6 +70,9 @@ func (m *blockingSSEWriterModule) RouterOnRequest(ctx core.RequestContext, next armed: m.armed, writeStarted: m.writeStarted, startedOnce: m.startedOnce, + writeReturned: m.writeReturned, + returned: m.returned, + returnedOnce: m.returnedOnce, release: m.release, }, ctx.Request()) } @@ -73,6 +82,9 @@ type deadlineBlockingResponseWriter struct { armed *atomic.Bool writeStarted chan struct{} startedOnce *sync.Once + writeReturned *atomic.Bool + returned chan struct{} + returnedOnce *sync.Once release chan struct{} deadlineNanos atomic.Int64 } @@ -83,6 +95,10 @@ func (w *deadlineBlockingResponseWriter) Write(data []byte) (int, error) { } w.startedOnce.Do(func() { close(w.writeStarted) }) + defer func() { + w.writeReturned.Store(true) + w.returnedOnce.Do(func() { close(w.returned) }) + }() deadlineNanos := w.deadlineNanos.Load() if deadlineNanos == 0 { <-w.release @@ -131,18 +147,27 @@ func TestKafkaSubscriptionRecoversAfterSSEWriteTimeout(t *testing.T) { t.Skip("skipping Kafka integration test in short mode") } - const topic = "employeeUpdated-sse-write-timeout" + const ( + topic = "employeeUpdated-sse-write-timeout" + sseWriteTimeout = 3 * time.Second + healthyClients = 2 + ) armed := &atomic.Bool{} writeStarted := make(chan struct{}) + writeReturned := &atomic.Bool{} + returned := make(chan struct{}) release := make(chan struct{}) var releaseOnce sync.Once t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) }) module := &blockingSSEWriterModule{ - armed: armed, - writeStarted: writeStarted, - startedOnce: &sync.Once{}, - release: release, + armed: armed, + writeStarted: writeStarted, + startedOnce: &sync.Once{}, + writeReturned: writeReturned, + returned: returned, + returnedOnce: &sync.Once{}, + release: release, } testenv.Run(t, &testenv.Config{ @@ -154,7 +179,7 @@ func TestKafkaSubscriptionRecoversAfterSSEWriteTimeout(t *testing.T) { []string{"employeeUpdated", "employeeUpdatedTwo"}, topic) }, ModifyEngineExecutionConfiguration: func(cfg *config.EngineExecutionConfiguration) { - cfg.SSEServerWriteTimeout = 100 * time.Millisecond + cfg.SSEServerWriteTimeout = sseWriteTimeout }, }, func(t *testing.T, xEnv *testenv.Environment) { events.KafkaEnsureTopicExists(t, xEnv, EventWaitTimeout, topic) @@ -164,11 +189,14 @@ func TestKafkaSubscriptionRecoversAfterSSEWriteTimeout(t *testing.T) { client := &http.Client{} blockedResp := openSSESubscription(t, ctx, client, xEnv.GraphQLRequestURL(), true) defer blockedResp.Body.Close() - healthyResp := openSSESubscription(t, ctx, client, xEnv.GraphQLRequestURL(), false) - defer healthyResp.Body.Close() - healthyReader := bufio.NewReader(healthyResp.Body) + healthyReaders := make([]*bufio.Reader, 0, healthyClients) + for range healthyClients { + healthyResp := openSSESubscription(t, ctx, client, xEnv.GraphQLRequestURL(), false) + defer healthyResp.Body.Close() + healthyReaders = append(healthyReaders, bufio.NewReader(healthyResp.Body)) + } - xEnv.WaitForSubscriptionCount(2, EventWaitTimeout) + xEnv.WaitForSubscriptionCount(healthyClients+1, EventWaitTimeout) xEnv.WaitForTriggerCount(1, EventWaitTimeout) armed.Store(true) @@ -181,27 +209,50 @@ func TestKafkaSubscriptionRecoversAfterSSEWriteTimeout(t *testing.T) { t.Fatal("timed out waiting for the SSE write to block") } - require.Contains(t, readSSEData(t, healthyReader), `"id":1`) + for _, reader := range healthyReaders { + require.Contains(t, readSSEData(t, reader), `"id":1`) + } - xEnv.WaitForSubscriptionCount(1, EventWaitTimeout) - xEnv.KafkaPublishUntilReceived(topic, - `{"__typename":"Employee","id":2,"update":{"name":"recovery"}}`, 1, EventWaitTimeout) - - recovery := make(chan string, 1) - go func() { - data, err := readSSEDataLine(healthyReader) - if err != nil { - recovery <- "error: " + err.Error() - return - } - recovery <- data - }() + type readResult struct { + data string + err error + blockedWriteHadReturned bool + } + recovery := make(chan readResult, healthyClients) + for _, reader := range healthyReaders { + go func() { + data, err := readSSEDataLine(reader) + recovery <- readResult{ + data: data, + err: err, + blockedWriteHadReturned: writeReturned.Load(), + } + }() + } + + // Queue the next provider event while the first event's shared-trigger + // dispatch is still blocked on one subscriber's SSE write. + events.ProduceKafkaMessage(t, xEnv, EventWaitTimeout, topic, + `{"__typename":"Employee","id":2,"update":{"name":"recovery"}}`) + require.False(t, writeReturned.Load(), "blocked SSE write returned before the second event was queued") select { - case data := <-recovery: - require.Contains(t, data, `"id":2`) - case <-time.After(EventWaitTimeout): - t.Fatal("healthy subscription did not receive the queued event after the SSE write deadline") + case <-returned: + case <-time.After(sseWriteTimeout + time.Second): + t.Fatal("blocked SSE write did not return after its deadline") + } + + xEnv.WaitForSubscriptionCount(healthyClients, EventWaitTimeout) + for range healthyClients { + select { + case result := <-recovery: + require.NoError(t, result.err) + require.True(t, result.blockedWriteHadReturned, + "healthy subscription received the queued event while shared-trigger dispatch was blocked") + require.Contains(t, result.data, `"id":2`) + case <-time.After(EventWaitTimeout): + t.Fatal("healthy subscription did not receive the queued event after the SSE write deadline") + } } }) }