fix(epoll, iouring, std): graceful shutdown waits for the HTTP/2 streams on the shared worker pool, and std for its h2c streams (celeris#759) (#808)

epoll, io_uring, adaptive: once shutdown begins, a loop or worker sends every HTTP/2 connection GOAWAY and keeps serving it until no stream on the shared HTTP/2 worker pool (an async route) has a handler running, a response in its write queue, or DATA waiting for WINDOW_UPDATE; the wait is bounded like epoll's send drain (the last Shutdown's budget while it is live, WriteTimeout while it is set, never less than 250 ms). Before, those streams' connections were closed under their handlers (unexpected EOF) and the OnShutdown hooks ran first.
During that wait the native engines accept no new connection (epoll closes its listener when shutdown begins, io_uring when the wait begins), and a stream the client opens above the GOAWAY's last-stream-id is refused with REFUSED_STREAM, never served; a refused stream is not touched after its RST_STREAM has put it back in the stream pool.
std: the Bridge counts the HTTP/2 requests in their handler, and the drain waits for them after http.Server.Shutdown, bounded by the drain's context, so the hooks no longer run while an h2c handler is still running.
Engine.Shutdown's and Server.Shutdown's docs say what the engines do now. Follow-ups: #820.

Fixes #759

Author: FumingPower3925

engine/engine.go

@@ -15,11 +15,16 @@ type Engine interface {
 	// Listen starts the engine and blocks until ctx is canceled or a fatal
 	// error occurs. The engine begins accepting connections on the configured address.
 	Listen(ctx context.Context) error
-	// Shutdown gracefully drains in-flight connections, bounded by ctx: when
-	// ctx expires, Shutdown returns its error. On epoll and io_uring the
-	// drain runs in Listen once Listen's ctx is cancelled, and Shutdown itself
-	// does nothing. No engine closes a connection whose handler is still
-	// running when ctx expires; the handler runs to completion (celeris#753).
+	// Shutdown gracefully drains in-flight connections, bounded by ctx. On
+	// std Shutdown is the drain: when ctx expires first, it returns ctx's
+	// error. On epoll and io_uring the drain runs in Listen once Listen's
+	// ctx is cancelled, and Shutdown hands it ctx as its budget and returns
+	// nil at once (celeris#759, celeris#760); adaptive hands ctx to its
+	// sub-engines, then cancels its own Listen and waits for it, bounded by
+	// ctx. An HTTP/1 handler runs to completion on every engine whatever
+	// ctx (celeris#753); a handler of an HTTP/2 stream on the shared worker
+	// pool can still be running when the native engines close its
+	// connection at the end of the budget.
 	Shutdown(ctx context.Context) error
 	// Metrics returns a point-in-time snapshot of engine performance counters.
 	Metrics() EngineMetrics

engine/epoll/conn.go

@@ -273,6 +273,11 @@ type connState struct {
 	// this set and does nothing (celeris#668).
 	hijackSettled bool
 
+	// h2GoAwaySent records that a graceful shutdown has sent this HTTP/2
+	// conn its GOAWAY (celeris#759; Loop.h2PoolSettled). Loop thread; reset
+	// on release.
+	h2GoAwaySent bool
+
 	// relinkOwed (guarded by asyncInMu) is set by the dirty pass or the
 	// EPOLLOUT resume when they give the conn up because its dispatch
 	// goroutine holds detachMu across a handler (celeris#669). The goroutine
@@ -377,6 +382,7 @@ func releaseConnState(cs *connState) {
 	cs.liveIdx = -1
 	cs.hijacked.Store(false)
 	cs.hijackSettled = false
+	cs.h2GoAwaySent = false
 	cs.closeOwed = false
 	cs.closeErr = nil
 	cs.relinkOwed = false

engine/epoll/loop.go

@@ -80,6 +80,10 @@ const shutdownSendDrainFloor = 250 * time.Millisecond
 // the drain notices a budget that ends early (a cancelled Shutdown ctx).
 const shutdownSendDrainPoll = 20 * time.Millisecond
 
+// h2PoolDrainPollMs caps one epoll_wait while the loop waits out the HTTP/2
+// pool handlers at shutdown (celeris#759).
+const h2PoolDrainPollMs = 10
+
 // Loop is an epoll-based event loop worker.
 type Loop struct {
 	id       int
@@ -197,6 +201,11 @@ type Loop struct {
 	// floor, shutdownSendDrainFloor.
 	drainBudget *atomic.Pointer[context.Context]
 
+	// h2DrainStart is when the loop began waiting, its context cancelled,
+	// for the HTTP/2 stream handlers running on the shared worker pool
+	// (celeris#759; h2PoolSettled). Zero until then. Loop thread.
+	h2DrainStart time.Time
+
 	// transplantInFlight counts connections this loop has detached for a
 	// transplant whose hand-off is not finished yet — the deferred async
 	// path, where tryTransplant detaches and drainDetachQueue completes.
@@ -474,8 +483,14 @@ func (l *Loop) run(ctx context.Context) {
 
 	for {
 		if ctx.Err() != nil {
-			l.shutdown()
-			return
+			// Accept nothing more (celeris#759): the loop may keep turning
+			// below, for the HTTP/2 conns it has, and a conn accepted now
+			// would be served and then cut at the budget.
+			l.stopAccepting()
+			if l.h2PoolSettled() {
+				l.shutdown()
+				return
+			}
 		}
 
 		// Cache the atomic load: ACTIVE→LINGERING→DRAINING and
@@ -498,8 +513,9 @@ func (l *Loop) run(ctx context.Context) {
 		// false so a subsequent Pause observes a fresh signal.
 		l.listenFDClosed.Store(paused && l.listenFD < 0)
 
-		// SUSPENDED → ACTIVE: re-create listen socket after ResumeAccept.
-		if l.listenFD < 0 && !paused {
+		// SUSPENDED → ACTIVE: re-create listen socket after ResumeAccept;
+		// never once shutdown has begun (stopAccepting).
+		if l.listenFD < 0 && !paused && ctx.Err() == nil {
 			fd, err := createListenSocket(l.cfg.Addr, !l.cfg.DisableDeferAccept)
 			l.deferCapable = !l.cfg.DisableDeferAccept
 			if err != nil {
@@ -548,6 +564,12 @@ func (l *Loop) run(ctx context.Context) {
 		if l.listenHot {
 			timeoutMs = 0
 		}
+		// Waiting out HTTP/2 pool handlers at shutdown (celeris#759): a
+		// handler that ends without a write the loop hears of must not
+		// leave the loop blocked.
+		if !l.h2DrainStart.IsZero() && (timeoutMs < 0 || timeoutMs > h2PoolDrainPollMs) {
+			timeoutMs = h2PoolDrainPollMs
+		}
 
 		n, err := unix.EpollWait(l.epollFD, l.events, timeoutMs)
 		if err != nil {
@@ -3670,6 +3692,76 @@ func (l *Loop) shutdown() {
 	l.closeEpollFD()
 }
 
+// h2PoolSettled reports whether the loop, its context cancelled, may shut
+// down as far as HTTP/2 is concerned (celeris#759). A stream on an async
+// route runs its handler on the shared worker pool, off this loop, and its
+// response comes back through the conn's write queue, which only this loop
+// drains; shutdown cancelled such streams and closed their conns under their
+// handlers, so the client got unexpected EOF, and the hooks ran before the
+// handlers had finished. So the loop keeps turning, reading and writing as
+// usual on the conns it has (it accepts no new one: stopAccepting), until no
+// HTTP/2 conn has a pool handler running, a response still in its write
+// queue, or response DATA waiting for the client's WINDOW_UPDATE. Every
+// HTTP/2 conn is sent GOAWAY first, so its client opens no new stream on it,
+// as net/http's graceful shutdown does, and a stream it opens anyway is
+// refused. The wait is bounded like the send drain (sendDrainWait): the
+// budget of the last Engine.Shutdown, and never less than
+// shutdownSendDrainFloor. Loop thread.
+func (l *Loop) h2PoolSettled() bool {
+	if len(l.h2Conns) == 0 {
+		return true
+	}
+	if l.h2DrainStart.IsZero() {
+		l.h2DrainStart = time.Now()
+	}
+	busy := false
+	for _, fd := range l.h2Conns {
+		cs := l.conns[fd]
+		if cs == nil || cs.h2State == nil {
+			continue
+		}
+		if !cs.h2GoAwaySent {
+			mu := cs.detachMu
+			if mu != nil {
+				mu.Lock()
+			}
+			cs.h2GoAwaySent = cs.h2State.GoAway(cs.writeFn)
+			if mu != nil {
+				mu.Unlock()
+			}
+			if cs.h2GoAwaySent {
+				l.markDirty(cs) // the dirty pass flushes the GOAWAY
+			}
+		}
+		if cs.h2State.PoolHandlersRunning() || cs.h2State.WriteQueuePending() || cs.h2State.OutboundPending() {
+			busy = true
+		}
+	}
+	if !busy {
+		return true
+	}
+	_, more := l.sendDrainWait(l.h2DrainStart)
+	return !more
+}
+
+// stopAccepting takes the listener out of the epoll set and closes it, once
+// the loop's context is cancelled (celeris#759). The loop may go on turning
+// after that, for as long as its HTTP/2 conns keep it (h2PoolSettled), and it
+// used to accept and serve new connections meanwhile, which were then cut at
+// the budget; net/http's Shutdown closes its listeners first. What is still
+// in the kernel's accept queue is reset, as the close in shutdown did.
+// Idempotent. Loop thread.
+func (l *Loop) stopAccepting() {
+	if l.listenFD < 0 {
+		return
+	}
+	_ = unix.EpollCtl(l.epollFD, unix.EPOLL_CTL_DEL, l.listenFD, nil)
+	_ = unix.Close(l.listenFD)
+	l.listenFD = -1
+	l.listenHot = false
+	l.lingerUntil = 0
+}
+
 // drainSends is shutdown's send drain (celeris#760). It flushes every live
 // conn with response bytes still queued, and waits for their sockets to take
 // more, until nothing is queued or the drain's time is up (sendDrainWait). A

engine/iouring/conn.go

@@ -122,6 +122,7 @@ type connState struct {
 	needsRecv      bool         // 1: recv arm was dropped (SQ ring full); retry on next opportunity
 	recvIntoBody   bool         // 1: next recv CQE fills h1State.bodyBuf directly (skips ProcessH1 + cs.buf memcpy)
 	zcNotifPending bool         // 1: waiting for SEND_ZC notification CQE
+	h2GoAwaySent   bool         // 1: a graceful shutdown sent this H2 conn its GOAWAY (celeris#759)
 	// sendIsZC records how the send SQE currently in flight for this
 	// connection was ARMED: true for IORING_OP_SEND_ZC, false for a plain
 	// SEND / WRITEV / linked SEND. It is the provenance flag the error
@@ -560,6 +561,7 @@ func releaseConnState(cs *connState) {
 	cs.needsRecv = false
 	cs.recvIntoBody = false
 	cs.zcNotifPending = false
+	cs.h2GoAwaySent = false
 	cs.sendIsZC = false
 	cs.zcSentBytes = 0
 	cs.lastActivity = 0

engine/iouring/engine.go

@@ -109,6 +109,11 @@ type Engine struct {
 	// and when the probe got no answer. Every worker gets a copy; the
 	// hand-off's REAP needs them (celeris#657).
 	asyncCancelFlags bool
+
+	// drainBudget is the ctx of the last Shutdown call: the budget of the
+	// wait for HTTP/2 pool handlers the workers run when Listen's context is
+	// cancelled (celeris#759; Worker.h2PoolSettled).
+	drainBudget atomic.Pointer[context.Context]
 }
 
 // New creates a new io_uring engine.
@@ -465,6 +470,7 @@ func (e *Engine) createWorkers(tier TierStrategy, cpus []int,
 		w.sweepCnt = &e.metrics.sweep          // celeris#657 P9 sweep witnesses
 		w.asyncCancelFlags = e.asyncCancelFlags
 		w.pause = &e.pause // celeris#662 pause linger
+		w.drainBudget = &e.drainBudget
 		workers[i] = w
 	}
 	return workers, nil
@@ -491,7 +497,7 @@ func fallbackTier(current TierStrategy) TierStrategy {
 	}
 }
 
-// Shutdown is a no-op for the io_uring engine — graceful shutdown is
+// Shutdown does not stop the io_uring engine itself — graceful shutdown is
 // driven by context cancellation on Listen's parent context. Workers
 // exit their run loops on ctx.Done, drain the responses still queued for
 // the ring (Worker.hasPendingSends, celeris#595) and call Worker.shutdown,
@@ -503,7 +509,13 @@ func fallbackTier(current TierStrategy) TierStrategy {
 // point owns one and Server.Shutdown cancels it after the graceful phase.
 // Handing Listen a context.Background() is what made Start hang here
 // (celeris#595), since this method cannot wake it.
-func (e *Engine) Shutdown(_ context.Context) error {
+//
+// What Shutdown does is hand ctx to the workers as the budget of their wait
+// for the HTTP/2 stream handlers still running on the shared worker pool
+// when Listen's context is cancelled (celeris#759): Server.Shutdown calls it
+// before it cancels that context.
+func (e *Engine) Shutdown(ctx context.Context) error {
+	e.drainBudget.Store(&ctx)
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	return nil