http_server input can get permanently stuck returning 503 after a stream restart (stream mode)

Summary

When an http_server input runs on the service-wide HTTP server (i.e. no dedicated address configured) and its stream is restarted/updated repeatedly in stream mode, the registered endpoint can, very rarely, get stuck returning 503 Service Unavailable forever and never recover — even though the new stream is otherwise running fine.

The root cause is a race condition between the old input's teardown (which registers a disabled 503 handler) and the new input's startup (which registers the normal handler) on the same shared endpoint path.

Environment

• Component: input / http_server
• Mode: stream mode (streams manager Update = Delete + Create), http_server input without a dedicated address (uses the service-wide HTTP server)

Steps to reproduce

1. Run Benthos in stream mode with a stream containing an http_server input that has no address set, e.g.:

   input:
     http_server:
       path: /testpost
   

2. Repeatedly update/restart that stream (e.g. via the streams manager REST API PUT /streams/{id}, or a config watcher), while traffic hits /testpost.
3. Occasionally (timing-dependent), after a restart the /testpost endpoint keeps returning 503 and never recovers.

Root cause

In internal/impl/io/input_http_server.go, loop()'s defer handles teardown. For the service-wide server case it launches an asynchronous goroutine that waits on HasStoppedChan() / HardStopChan() and then re-registers the path with a static 503 "Endpoint disabled." handler:

go func() {
    select {
    case <-h.shutSig.HasStoppedChan():
    case <-h.shutSig.HardStopChan():
    }
    // RegisterEndpoint(path, "Endpoint disabled.", -> 503)
}()
...
h.handlerWG.Wait()
close(h.transactions)
h.shutSig.TriggerHasStopped()   // fires HasStoppedChan()

WaitForClose() waits on the same HasStoppedChan():

func (h *httpServerInput) WaitForClose(ctx context.Context) error {
    select {
    case <-h.shutSig.HasStoppedChan():
    case <-ctx.Done():
        return ctx.Err()
    }
    return nil
}

So TriggerHasStopped() releases both the 503-registering goroutine and WaitForClose() at the same time, with no ordering guarantee between:

• goroutine G registering the 503 handler, and
• the streams manager proceeding WaitForClose → Stop → Delete → Create → newHTTPServerInput() registering the normal (200) handler.

RegisterEndpoint (internal/api/api.go) is last-writer-wins for a given path. Usually G (a trivial map write) runs first and the subsequent Create overwrites it with the good handler → 200. But occasionally G is scheduled after Create's registration, overwriting the good handler with the static 503 → the endpoint is stuck at 503 with no way to recover.

This is why it happens only "very rarely" — it depends on goroutine scheduling.

Affected code

• internal/impl/io/input_http_server.go — loop() teardown (service-wide server branch)
• internal/impl/io/input_http_server_wasm.go — same pattern

output_http_server is not affected: it does not re-register a disabled 503 handler on stop; its handler is simply overwritten by the new instance on restart.

Proposed fix

Register the disabled 503 handler synchronously, after handlerWG.Wait() (all in-flight requests drained) and before TriggerHasStopped(). This guarantees the ordering:

old input: drain -> register 503 -> TriggerHasStopped
        -> WaitForClose returns -> Stop returns -> Delete returns
        -> Create -> new input registers normal handler   (always wins)

so the new instance's registration always happens last and the endpoint recovers. Hard-stop no longer needs the separate goroutine because in-flight handlers already observe HardStopChan() and unblock handlerWG.Wait(); new requests already receive 503 from the existing handler's soft-stop check during draining, so behavior is preserved for the full-shutdown case.

Start in internal/impl/io/input_http_server.go, focusing on loop() teardown and WaitForClose(), then compare the corresponding pattern in internal/impl/io/input_http_server_wasm.go. Reproduce repeated stream updates against an endpoint without a dedicated address and verify that the endpoint returns normally after restart rather than remaining at 503.