fix(http_server): register disabled endpoint synchronously to avoid stuck 503 on restart

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()
 	}()

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))
+	}
+}

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()
 	}()