From 43158710a59ae421fa66b4eed345da55469d7b10 Mon Sep 17 00:00:00 2001 From: Rikuo Takahama Date: Thu, 6 Aug 2026 10:54:40 +0900 Subject: [PATCH] fix(http_server): register disabled endpoint synchronously to avoid stuck 503 When an http_server input uses the service-wide HTTP server (no dedicated address), loop()'s teardown registered a disabled 503 "Endpoint disabled." handler from an asynchronous goroutine that waited on the same HasStopped signal as WaitForClose. TriggerHasStopped released both concurrently, so the 503 registration raced with the new instance's normal-handler registration on restart (streams manager Update = Delete + Create). Since RegisterEndpoint is last-writer-wins per path, the 503 handler could occasionally win, leaving the endpoint permanently returning 503. Register the disabled handler synchronously, after handlerWG.Wait() drains in-flight requests and before TriggerHasStopped(). This guarantees the new instance's registration always happens after ours and wins, so the endpoint recovers. Hard-stop no longer needs a separate goroutine because in-flight handlers already observe HardStopChan and unblock handlerWG.Wait(), and new requests still receive 503 from the existing handler's soft-stop check during draining. Applies the same fix to the wasm build. Adds a regression test exercising repeated create/use/stop cycles on a shared service-wide server. Fixes #469 Co-Authored-By: Claude Opus 4.8 --- internal/impl/io/input_http_server.go | 57 +++++++++------- internal/impl/io/input_http_server_test.go | 75 ++++++++++++++++++++++ internal/impl/io/input_http_server_wasm.go | 57 +++++++++------- 3 files changed, 141 insertions(+), 48 deletions(-) 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() }()