diff --git a/internal/impl/io/input_http_server.go b/internal/impl/io/input_http_server.go index 360056217..7d10f448e 100644 --- a/internal/impl/io/input_http_server.go +++ b/internal/impl/io/input_http_server.go @@ -862,40 +862,49 @@ func (h *httpServerInput) wsHandler(w http.ResponseWriter, r *http.Request) { func (h *httpServerInput) loop() { defer func() { + // Whether we're using the service-wide HTTP server (rather than a + // dedicated listener) must be captured before we potentially nil the + // server reference below. + usingServiceWideServer := h.server == nil + if h.server != nil { if err := h.server.Shutdown(context.Background()); err != nil { h.log.Error("Failed to gracefully terminate http_server: %v\n", err) } h.server = nil h.listener = nil - } else { - // We are using the service-wide HTTP server. In order to prevent - // situations where a slow shutdown results in serving an abundance - // of 503 responses we wait until either the current requests are - // handled and shutdown can commence, or we've been instructed to - // close immediately, which prevents these requests from - // indefinitely blocking shutdown. - go func() { - select { - case <-h.shutSig.HasStoppedChan(): - case <-h.shutSig.HardStopChan(): - } - - if h.conf.Path != "" { - h.mgr.RegisterEndpoint(h.conf.Path, "Endpoint disabled.", func(w http.ResponseWriter, r *http.Request) { - http.Error(w, "Service unavailable", http.StatusServiceUnavailable) - }) - } - if h.conf.WSPath != "" { - h.mgr.RegisterEndpoint(h.conf.WSPath, "Endpoint disabled.", func(w http.ResponseWriter, r *http.Request) { - http.Error(w, "Service unavailable", http.StatusServiceUnavailable) - }) - } - }() } h.handlerWG.Wait() + if usingServiceWideServer { + // We are using the service-wide HTTP server, so once all in-flight + // requests have drained we replace our endpoints with disabled + // handlers that return a 503. In-flight requests are served by the + // existing handlers (which already return a 503 once soft stop is + // signalled), so waiting for them to drain first avoids serving an + // abundance of 503 responses during a slow shutdown. + // + // This registration is performed synchronously, before signalling + // that we've stopped (TriggerHasStopped) which is what WaitForClose + // and therefore Stop block on. That ordering guarantees that when + // this input is being replaced on the same path (e.g. a stream + // update, which stops the old stream then creates a new one) the new + // instance's endpoint registration always happens after ours and + // wins, allowing the endpoint to recover instead of being left stuck + // returning 503. + if h.conf.Path != "" { + h.mgr.RegisterEndpoint(h.conf.Path, "Endpoint disabled.", func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Service unavailable", http.StatusServiceUnavailable) + }) + } + if h.conf.WSPath != "" { + h.mgr.RegisterEndpoint(h.conf.WSPath, "Endpoint disabled.", func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Service unavailable", http.StatusServiceUnavailable) + }) + } + } + close(h.transactions) h.shutSig.TriggerHasStopped() }() diff --git a/internal/impl/io/input_http_server_test.go b/internal/impl/io/input_http_server_test.go index 85885c16c..38c05813c 100644 --- a/internal/impl/io/input_http_server_test.go +++ b/internal/impl/io/input_http_server_test.go @@ -1471,3 +1471,78 @@ http_server: h2.TriggerStopConsuming() require.NoError(t, h2.WaitForClose(tCtx)) } + +// TestHTTPServerSharedServerRestartRecovers exercises repeatedly creating, +// using and stopping an http_server input that shares the service-wide HTTP +// server (no dedicated address) on the same path, mimicking what a streams +// manager does on every stream update (stop old stream, create new one). +// +// Previously the stopping input registered a disabled 503 handler from an +// asynchronous goroutine that raced with the new instance's registration, +// which could leave the endpoint stuck returning 503 forever. The disabled +// handler is now registered synchronously before the input signals that it has +// stopped, so the subsequent registration always wins and the endpoint +// recovers. +func TestHTTPServerSharedServerRestartRecovers(t *testing.T) { + tCtx, done := context.WithTimeout(t.Context(), time.Minute) + defer done() + + t.Parallel() + + // Use the real api.Type registry so that repeated RegisterEndpoint calls on + // the same path swap the handler (its dynamic handler map), matching the + // behaviour of the service-wide HTTP server in production. The gorilla mux + // test wrapper used elsewhere adds a new route per call instead. + apiConf := api.NewConfig() + apiImpl, err := api.New("", "", apiConf, nil, log.Noop(), metrics.Noop()) + require.NoError(t, err) + + mgr, err := manager.New(manager.ResourceConfig{}, manager.OptSetAPIReg(apiImpl)) + require.NoError(t, err) + + conf := parseYAMLInputConf(t, ` +http_server: + path: /testpost +`) + + server := httptest.NewServer(apiImpl.Handler()) + defer server.Close() + + for i := range 20 { + h, err := mgr.NewInput(conf) + require.NoError(t, err) + + h.TriggerStartConsuming() + + var wg sync.WaitGroup + wg.Add(1) + var statusCode int + go func() { + defer wg.Done() + res, cerr := http.Post( + server.URL+"/testpost", + "application/octet-stream", + bytes.NewBufferString("hello"), + ) + if cerr != nil { + t.Errorf("iteration %v: request failed: %v", i, cerr) + return + } + defer res.Body.Close() + statusCode = res.StatusCode + }() + + select { + case ts := <-h.TransactionChan(): + require.NoError(t, ts.Ack(tCtx, nil)) + case <-time.After(5 * time.Second): + t.Fatalf("iteration %v: timed out waiting for message", i) + } + + wg.Wait() + assert.Equalf(t, 200, statusCode, "iteration %v: endpoint should have recovered instead of returning 503", i) + + h.TriggerStopConsuming() + require.NoError(t, h.WaitForClose(tCtx)) + } +} diff --git a/internal/impl/io/input_http_server_wasm.go b/internal/impl/io/input_http_server_wasm.go index 812c9ef7b..8fd33ff5a 100644 --- a/internal/impl/io/input_http_server_wasm.go +++ b/internal/impl/io/input_http_server_wasm.go @@ -840,38 +840,47 @@ func (h *httpServerInput) wsHandler(w http.ResponseWriter, r *http.Request) { func (h *httpServerInput) loop() { defer func() { + // Whether we're using the service-wide HTTP server (rather than a + // dedicated listener) must be captured before any changes to the server + // reference below. + usingServiceWideServer := h.server == nil + if h.server != nil { if err := h.server.Shutdown(context.Background()); err != nil { h.log.Error("Failed to gracefully terminate http_server: %v\n", err) } - } else { - // We are using the service-wide HTTP server. In order to prevent - // situations where a slow shutdown results in serving an abundance - // of 503 responses we wait until either the current requests are - // handled and shutdown can commence, or we've been instructed to - // close immediately, which prevents these requests from - // indefinitely blocking shutdown. - go func() { - select { - case <-h.shutSig.HasStoppedChan(): - case <-h.shutSig.HardStopChan(): - } - - if h.conf.Path != "" { - h.mgr.RegisterEndpoint(h.conf.Path, "Endpoint disabled.", func(w http.ResponseWriter, r *http.Request) { - http.Error(w, "Service unavailable", http.StatusServiceUnavailable) - }) - } - if h.conf.WSPath != "" { - h.mgr.RegisterEndpoint(h.conf.WSPath, "Endpoint disabled.", func(w http.ResponseWriter, r *http.Request) { - http.Error(w, "Service unavailable", http.StatusServiceUnavailable) - }) - } - }() } h.handlerWG.Wait() + if usingServiceWideServer { + // We are using the service-wide HTTP server, so once all in-flight + // requests have drained we replace our endpoints with disabled + // handlers that return a 503. In-flight requests are served by the + // existing handlers (which already return a 503 once soft stop is + // signalled), so waiting for them to drain first avoids serving an + // abundance of 503 responses during a slow shutdown. + // + // This registration is performed synchronously, before signalling + // that we've stopped (TriggerHasStopped) which is what WaitForClose + // and therefore Stop block on. That ordering guarantees that when + // this input is being replaced on the same path (e.g. a stream + // update, which stops the old stream then creates a new one) the new + // instance's endpoint registration always happens after ours and + // wins, allowing the endpoint to recover instead of being left stuck + // returning 503. + if h.conf.Path != "" { + h.mgr.RegisterEndpoint(h.conf.Path, "Endpoint disabled.", func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Service unavailable", http.StatusServiceUnavailable) + }) + } + if h.conf.WSPath != "" { + h.mgr.RegisterEndpoint(h.conf.WSPath, "Endpoint disabled.", func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Service unavailable", http.StatusServiceUnavailable) + }) + } + } + close(h.transactions) h.shutSig.TriggerHasStopped() }()