fix(middleware): enforce transfer inactivity with a request-scoped timer

Replace per-call ResponseController deadline renewal with a request-scoped timer and shared lifetime state. The controller is never used after the handler returns or the connection is hijacked, fixing a race and nil-pointer panic when the HTTP/2 transport closes the outbound body asynchronously. Idle expiry cancels the request context and expires connection deadlines, so silent handlers and upstreams are cut. Response writes use 8 KiB progress units so steadily reading, backpressured clients are not cut.

commit: 98315bde309289803c17e0d596e13b23c84c7f96
author: brice · 6d ago · 2026-10-16 19:19 UTC
parent: e5ab890
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +450 -94

M docs/config/traffic.md +3 -2


@@ -148,11 +148,12 @@ Values are Go duration strings; malformed, zero, or negative values are rejected
148	148
149	149		### Inactivity timeout
150	150
151		-	Gordon renews the response write deadline before every write and flush, and the request read deadline before every request body read. The deadline is also armed when a request starts, so the wait for the first response byte from a slow backend is bounded by `transfer_idle_timeout` too. Consequences:
	151	+	Each request has an inactivity timer, armed when the request starts. Progress resets it: a response chunk accepted by the client connection (8 KiB granularity), a flush, or request body bytes read. If `transfer_idle_timeout` passes without progress, Gordon cancels the request context, which aborts the upstream request of a reverse-proxied app, and closes the client connection. Consequences:
152	152
153		-	- a response that streams data at least once per `transfer_idle_timeout` can last arbitrarily long;
	153	+	- a response that streams data at least once per `transfer_idle_timeout` can last arbitrarily long, and a client that keeps reading is never cut (it must accept 8 KiB per period at minimum);
154	154		- an upload whose body arrives at least once per `transfer_idle_timeout` can last arbitrarily long;
155	155		- a backend that stays silent longer than `transfer_idle_timeout` before or between response chunks has its connection cut, so set it above the slowest expected backend response time;
	156	+	- the timer also fires when a handler or backend is silent and no write is pending;
156	157		- WebSocket and other upgraded connections are not subject to it, and HTTP/2 streams are handled the same way as HTTP/1.
157	158
158	159		`transfer_idle_timeout` does not apply to `[traffic.tcp]` or `[traffic.udp]` routers; those use their own `idle_timeout` (see below).

 ### Inactivity timeout
-Gordon renews the response write deadline before every write and flush, and the request read deadline before every request body read. The deadline is also armed when a request starts, so the wait for the first response byte from a slow backend is bounded by `transfer_idle_timeout` too. Consequences:
+Each request has an inactivity timer, armed when the request starts. Progress resets it: a response chunk accepted by the client connection (8 KiB granularity), a flush, or request body bytes read. If `transfer_idle_timeout` passes without progress, Gordon cancels the request context, which aborts the upstream request of a reverse-proxied app, and closes the client connection. Consequences:
-- a response that streams data at least once per `transfer_idle_timeout` can last arbitrarily long;
+- a response that streams data at least once per `transfer_idle_timeout` can last arbitrarily long, and a client that keeps reading is never cut (it must accept 8 KiB per period at minimum);
 - an upload whose body arrives at least once per `transfer_idle_timeout` can last arbitrarily long;
 - a backend that stays silent longer than `transfer_idle_timeout` before or between response chunks has its connection cut, so set it above the slowest expected backend response time;
+- the timer also fires when a handler or backend is silent and no write is pending;
 - WebSocket and other upgraded connections are not subject to it, and HTTP/2 streams are handled the same way as HTTP/1.
 `transfer_idle_timeout` does not apply to `[traffic.tcp]` or `[traffic.udp]` routers; those use their own `idle_timeout` (see below).

M internal/adapters/in/http/middleware/transfer_idle.go +159 -88


@@ -2,6 +2,7 @@ package middleware
2	2
3	3		import (
4	4		"bufio"
	5	+	"context"
5	6		"errors"
6	7		"io"
7	8		"net"

@@ -10,95 +11,178 @@ import (
10	11		"time"
11	12		)
12	13
	14	+	// errTransferFinished is returned by writer calls made after the request
	15	+	// finished (or was hijacked). The controller is never touched in that state.
	16	+	var errTransferFinished = errors.New("transfer idle: request already finished")
	17	+
	18	+	// idleWriteChunk is the progress unit of response writes. Each chunk is handed
	19	+	// to the connection separately and counts as progress once it has been fully
	20	+	// accepted, so a client that keeps consuming data is never cut while a client
	21	+	// that stops reading is cut after the idle timeout. The residual granularity is
	22	+	// one chunk: a client must accept idleWriteChunk bytes within one idle period,
	23	+	// which is a floor of idleWriteChunk/idle bytes per second (about 27 B/s for
	24	+	// the 2 minute default).
	25	+	const idleWriteChunk = 8 * 1024
	26	+
13	27		// TransferIdleTimeout returns a middleware that replaces fixed per-request
14		-	// read/write timeouts with an inactivity timeout. The server it wraps must not
15		-	// set ReadTimeout or WriteTimeout.
	28	+	// read/write timeouts with a request-scoped inactivity timeout. The server it
	29	+	// wraps must not set ReadTimeout or WriteTimeout.
	30	+	//
	31	+	// A single timer covers the request. Progress renews it: a response chunk
	32	+	// accepted by the connection, a flush, or request body bytes read. When idle
	33	+	// elapses without progress, the middleware cancels the request context (so a
	34	+	// ReverseProxy aborts its upstream request) and expires the connection's read
	35	+	// and write deadlines, which unblocks pending I/O and makes the server drop the
	36	+	// connection once the handler returns. The timer also covers a handler that is
	37	+	// silent and not blocked in any I/O, which a plain write deadline cannot.
16	38		//
17		-	// - The write deadline is set to now+idle when the request starts and renewed
18		-	// before every Write and Flush. A response is cut when no byte is written
19		-	// for idle (the next Write fails because the deadline has expired, or the
20		-	// pending Write fails when the client stops reading).
21		-	// - The read deadline is renewed before every request body Read, so a slow
22		-	// but steady upload is never cut, while a stalled one is.
	39	+	// The timer is armed when the request starts, so the wait for the first
	40	+	// response byte is bounded too. It is stopped when the handler returns and when
	41	+	// the connection is hijacked: upgraded streams (WebSocket) are exempt.
	42	+	// Deadlines go through http.ResponseController, so HTTP/1.x and HTTP/2 both
	43	+	// work; Flush, Hijack and Unwrap are preserved.
23	44		//
24		-	// Deadlines are applied with http.ResponseController, so HTTP/1.x and HTTP/2
25		-	// are both supported. Flush, Hijack (protocol upgrades) and Unwrap keep
26		-	// working. A non-positive idle disables the middleware.
	45	+	// Limits: a handler that ignores the canceled context and never touches the
	46	+	// connection again keeps its HTTP/1 connection until it returns. A non-positive
	47	+	// idle disables the middleware.
27	48		func TransferIdleTimeout(idle time.Duration) func(http.Handler) http.Handler {
28	49		return func(next http.Handler) http.Handler {
29	50		if idle <= 0 {
30	51		return next
31	52		}
32	53		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
33		-	rc := http.NewResponseController(w)
34		-	iw := &idleResponseWriter{ResponseWriter: w, rc: rc, idle: idle}
35		-	iw.renew()
	54	+	ctx, cancel := context.WithCancel(r.Context())
	55	+	state := newIdleState(idle, http.NewResponseController(w), cancel)
	56	+	// finish marks the state done under the mutex before this
	57	+	// function returns, so nothing touches the controller afterwards.
	58	+	defer state.finish()
	59	+
	60	+	req := r.WithContext(ctx)
36	61		if r.Body != nil && r.Body != http.NoBody {
37		-	ib := &idleBody{ReadCloser: r.Body, rc: rc, idle: idle, writer: iw}
38		-	ib.renew()
39		-	r.Body = ib
	62	+	req.Body = &idleBody{ReadCloser: r.Body, state: state}
40	63		}
41		-	next.ServeHTTP(iw, r)
	64	+	next.ServeHTTP(&idleResponseWriter{ResponseWriter: w, state: state}, req)
42	65		})
43	66		}
44	67		}
45	68
46		-	// idleResponseWriter renews the write deadline on every Write/Flush.
47		-	type idleResponseWriter struct {
48		-	http.ResponseWriter
49		-	rc *http.ResponseController
50		-	idle time.Duration
	69	+	// idleState is the request-lifetime state shared by the writer, the body and
	70	+	// the timer. The controller is only used while the state is live, under mu.
	71	+	type idleState struct {
	72	+	idle time.Duration
	73	+	rc *http.ResponseController
	74	+	cancel context.CancelFunc
51	75
52		-	mu sync.Mutex
53		-	deadline time.Time
	76	+	mu sync.Mutex
	77	+	timer *time.Timer
	78	+	last time.Time
	79	+	done bool // handler returned or connection hijacked
54	80		}
55	81
56		-	// renew extends the write deadline unless it already expired. An expired
57		-	// deadline is left in place so that the write fails: the response was silent
58		-	// for longer than the idle timeout.
59		-	func (w *idleResponseWriter) renew() {
60		-	w.mu.Lock()
61		-	defer w.mu.Unlock()
62		-	now := time.Now()
63		-	if !w.deadline.IsZero() && now.After(w.deadline) {
	82	+	func newIdleState(idle time.Duration, rc http.ResponseController, cancel context.CancelFunc) idleState {
	83	+	s := &idleState{idle: idle, rc: rc, cancel: cancel, last: time.Now()}
	84	+	s.mu.Lock()
	85	+	s.timer = time.AfterFunc(idle, s.onTimer)
	86	+	s.mu.Unlock()
	87	+	return s
	88	+	}
	89	+
	90	+	// progress records activity. It never touches the controller.
	91	+	func (s *idleState) progress() {
	92	+	s.mu.Lock()
	93	+	if !s.done {
	94	+	s.last = time.Now()
	95	+	}
	96	+	s.mu.Unlock()
	97	+	}
	98	+
	99	+	// live reports whether the request is still being served.
	100	+	func (s *idleState) live() bool {
	101	+	s.mu.Lock()
	102	+	defer s.mu.Unlock()
	103	+	return !s.done
	104	+	}
	105	+
	106	+	// onTimer runs on timer expiry. Progress only stamps s.last, so the timer is
	107	+	// re-armed for the remainder instead of being reset on every event; this also
	108	+	// makes progress racing with expiry safe.
	109	+	func (s *idleState) onTimer() {
	110	+	s.mu.Lock()
	111	+	defer s.mu.Unlock()
	112	+	if s.done {
	113	+	return
	114	+	}
	115	+	if remaining := s.idle - time.Since(s.last); remaining > 0 {
	116	+	s.timer.Reset(remaining)
64	117		return
65	118		}
66		-	next := now.Add(w.idle)
67		-	if err := w.rc.SetWriteDeadline(next); err == nil {
68		-	w.deadline = next
	119	+	s.cancel()
	120	+	// Held under mu so finish() cannot complete while these run.
	121	+	now := time.Now()
	122	+	_ = s.rc.SetWriteDeadline(now)
	123	+	_ = s.rc.SetReadDeadline(now)
	124	+	}
	125	+
	126	+	// finish ends the request lifetime. After it returns, no code path calls the
	127	+	// controller. It is safe to call more than once.
	128	+	func (s *idleState) finish() {
	129	+	s.mu.Lock()
	130	+	s.done = true
	131	+	s.timer.Stop()
	132	+	s.mu.Unlock()
	133	+	s.cancel()
	134	+	}
	135	+
	136	+	// flush flushes through the controller unless the request finished. The mutex
	137	+	// is not held during the flush: it may block on a slow client, and the timer
	138	+	// callback needs the mutex to expire the deadlines that unblock it. net/http
	139	+	// forbids using the writer after the handler returned or concurrently with its
	140	+	// return, so the live check is a guard against misuse, not a synchronisation
	141	+	// point for the flush itself.
	142	+	func (s *idleState) flush() error {
	143	+	if !s.live() {
	144	+	return errTransferFinished
69	145		}
	146	+	return s.rc.Flush()
70	147		}
71	148
72		-	// extend pushes the write deadline forward unconditionally. It is called when
73		-	// the request body makes progress: a client that is still uploading has not
74		-	// left the response idle, so the wait for the first response byte starts from
75		-	// the last upload activity.
76		-	func (w *idleResponseWriter) extend() {
77		-	w.mu.Lock()
78		-	defer w.mu.Unlock()
79		-	next := time.Now().Add(w.idle)
80		-	if err := w.rc.SetWriteDeadline(next); err == nil {
81		-	w.deadline = next
	149	+	func (s idleState) hijack() (net.Conn, bufio.ReadWriter, error) {
	150	+	s.mu.Lock()
	151	+	defer s.mu.Unlock()
	152	+	if s.done {
	153	+	return nil, nil, errTransferFinished
	154	+	}
	155	+	conn, rw, err := s.rc.Hijack()
	156	+	if err == nil {
	157	+	s.done = true
	158	+	s.timer.Stop()
82	159		}
	160	+	return conn, rw, err
83	161		}
84	162
85		-	// maxIdleWriteChunk bounds how much is handed to the connection per deadline
86		-	// renewal, so one very large Write to a slow but steady client is not cut as if
87		-	// it were stalled.
88		-	const maxIdleWriteChunk = 64 * 1024
	163	+	// idleResponseWriter reports response progress.
	164	+	type idleResponseWriter struct {
	165	+	http.ResponseWriter
	166	+	state *idleState
	167	+	}
89	168
90		-	// Write renews the write deadline before each chunk, then writes it.
	169	+	// Write hands p to the connection in idleWriteChunk pieces and records progress
	170	+	// after each piece.
91	171		func (w *idleResponseWriter) Write(p []byte) (int, error) {
	172	+	if !w.state.live() {
	173	+	return 0, errTransferFinished
	174	+	}
92	175		if len(p) == 0 {
93		-	w.renew()
94	176		return w.ResponseWriter.Write(p)
95	177		}
96	178		written := 0
97	179		for written < len(p) {
98		-	chunk := p[written:min(written+maxIdleWriteChunk, len(p))]
99		-	w.renew()
	180	+	chunk := p[written:min(written+idleWriteChunk, len(p))]
100	181		n, err := w.ResponseWriter.Write(chunk)
101	182		written += n
	183	+	if n > 0 {
	184	+	w.state.progress()
	185	+	}
102	186		if err != nil {
103	187		return written, err
104	188		}

@@ -106,11 +190,14 @@ func (w *idleResponseWriter) Write(p []byte) (int, error) {
106	190		return written, nil
107	191		}
108	192
109		-	// FlushError renews the write deadline, then flushes. It lets
110		-	// http.ResponseController report flush failures.
	193	+	// FlushError flushes and records progress. It lets http.ResponseController
	194	+	// report flush failures.
111	195		func (w *idleResponseWriter) FlushError() error {
112		-	w.renew()
113		-	return w.rc.Flush()
	196	+	err := w.state.flush()
	197	+	if err == nil {
	198	+	w.state.progress()
	199	+	}
	200	+	return err
114	201		}
115	202
116	203		// Flush implements http.Flusher.

@@ -118,11 +205,10 @@ func (w *idleResponseWriter) Flush() {
118	205		_ = w.FlushError()
119	206		}
120	207
121		-	// Hijack implements http.Hijacker. The server clears connection deadlines when
122		-	// a connection is hijacked, so upgraded streams (e.g. WebSocket) are not
123		-	// subject to the inactivity timeout.
	208	+	// Hijack implements http.Hijacker and stops the inactivity timer: upgraded
	209	+	// streams (e.g. WebSocket) are exempt.
124	210		func (w idleResponseWriter) Hijack() (net.Conn, bufio.ReadWriter, error) {
125		-	return w.rc.Hijack()
	211	+	return w.state.hijack()
126	212		}
127	213
128	214		// Unwrap exposes the underlying writer to http.ResponseController.

@@ -130,42 +216,27 @@ func (w *idleResponseWriter) Unwrap() http.ResponseWriter {
130	216		return w.ResponseWriter
131	217		}
132	218
133		-	// idleBody renews the read deadline before every request body Read.
	219	+	// idleBody reports request body progress. It only records timestamps, so it is
	220	+	// safe to Read or Close after the handler returned (the HTTP/2 transport closes
	221	+	// the outbound body asynchronously).
134	222		type idleBody struct {
135	223		io.ReadCloser
136		-	rc *http.ResponseController
137		-	idle time.Duration
138		-	writer *idleResponseWriter
139		-	}
	224	+	state *idleState
140	225
141		-	func (b *idleBody) renew() {
142		-	_ = b.rc.SetReadDeadline(time.Now().Add(b.idle))
	226	+	closeOnce sync.Once
	227	+	closeErr error
143	228		}
144	229
145		-	// clear removes the read deadline. It must not stay armed once the body is
146		-	// fully read: net/http then watches the connection for client disconnects and
147		-	// treats a read timeout as an abort.
148		-	func (b *idleBody) clear() {
149		-	_ = b.rc.SetReadDeadline(time.Time{})
150		-	}
151		-
152		-	// Read renews the read deadline, reads, and clears the deadline at EOF.
153	230		func (b *idleBody) Read(p []byte) (int, error) {
154		-	b.renew()
155	231		n, err := b.ReadCloser.Read(p)
156		-	if n > 0 \|\| errors.Is(err, io.EOF) {
157		-	b.writer.extend()
158		-	}
159		-	if errors.Is(err, io.EOF) {
160		-	b.clear()
	232	+	if n > 0 {
	233	+	b.state.progress()
161	234		}
162	235		return n, err
163	236		}
164	237
165		-	// Close drains within a fresh deadline, then clears it.
	238	+	// Close is idempotent.
166	239		func (b *idleBody) Close() error {
167		-	b.renew()
168		-	err := b.ReadCloser.Close()
169		-	b.clear()
170		-	return err
	240	+	b.closeOnce.Do(func() { b.closeErr = b.ReadCloser.Close() })
	241	+	return b.closeErr
171	242		}


@@ -2,6 +2,7 @@ package middleware
2			2
3		import (	3		import (
4		"bufio"	4		"bufio"
			5	+	"context"
5		"errors"	6		"errors"
6		"io"	7		"io"
7		"net"	8		"net"

@@ -10,95 +11,178 @@ import (
10		"time"	11		"time"
11		)	12		)
12			13
			14	+	// errTransferFinished is returned by writer calls made after the request
			15	+	// finished (or was hijacked). The controller is never touched in that state.
			16	+	var errTransferFinished = errors.New("transfer idle: request already finished")
			17	+
			18	+	// idleWriteChunk is the progress unit of response writes. Each chunk is handed
			19	+	// to the connection separately and counts as progress once it has been fully
			20	+	// accepted, so a client that keeps consuming data is never cut while a client
			21	+	// that stops reading is cut after the idle timeout. The residual granularity is
			22	+	// one chunk: a client must accept idleWriteChunk bytes within one idle period,
			23	+	// which is a floor of idleWriteChunk/idle bytes per second (about 27 B/s for
			24	+	// the 2 minute default).
			25	+	const idleWriteChunk = 8 * 1024
			26	+
13		// TransferIdleTimeout returns a middleware that replaces fixed per-request	27		// TransferIdleTimeout returns a middleware that replaces fixed per-request
14	-	// read/write timeouts with an inactivity timeout. The server it wraps must not	28	+	// read/write timeouts with a request-scoped inactivity timeout. The server it
15	-	// set ReadTimeout or WriteTimeout.	29	+	// wraps must not set ReadTimeout or WriteTimeout.
			30	+	//
			31	+	// A single timer covers the request. Progress renews it: a response chunk
			32	+	// accepted by the connection, a flush, or request body bytes read. When idle
			33	+	// elapses without progress, the middleware cancels the request context (so a
			34	+	// ReverseProxy aborts its upstream request) and expires the connection's read
			35	+	// and write deadlines, which unblocks pending I/O and makes the server drop the
			36	+	// connection once the handler returns. The timer also covers a handler that is
			37	+	// silent and not blocked in any I/O, which a plain write deadline cannot.
16		//	38		//
17	-	// - The write deadline is set to now+idle when the request starts and renewed	39	+	// The timer is armed when the request starts, so the wait for the first
18	-	// before every Write and Flush. A response is cut when no byte is written	40	+	// response byte is bounded too. It is stopped when the handler returns and when
19	-	// for idle (the next Write fails because the deadline has expired, or the	41	+	// the connection is hijacked: upgraded streams (WebSocket) are exempt.
20	-	// pending Write fails when the client stops reading).	42	+	// Deadlines go through http.ResponseController, so HTTP/1.x and HTTP/2 both
21	-	// - The read deadline is renewed before every request body Read, so a slow	43	+	// work; Flush, Hijack and Unwrap are preserved.
22	-	// but steady upload is never cut, while a stalled one is.
23		//	44		//
24	-	// Deadlines are applied with http.ResponseController, so HTTP/1.x and HTTP/2	45	+	// Limits: a handler that ignores the canceled context and never touches the
25	-	// are both supported. Flush, Hijack (protocol upgrades) and Unwrap keep	46	+	// connection again keeps its HTTP/1 connection until it returns. A non-positive
26	-	// working. A non-positive idle disables the middleware.	47	+	// idle disables the middleware.
27		func TransferIdleTimeout(idle time.Duration) func(http.Handler) http.Handler {	48		func TransferIdleTimeout(idle time.Duration) func(http.Handler) http.Handler {
28		return func(next http.Handler) http.Handler {	49		return func(next http.Handler) http.Handler {
29		if idle <= 0 {	50		if idle <= 0 {
30		return next	51		return next
31		}	52		}
32		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {	53		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
33	-	rc := http.NewResponseController(w)	54	+	ctx, cancel := context.WithCancel(r.Context())
34	-	iw := &idleResponseWriter{ResponseWriter: w, rc: rc, idle: idle}	55	+	state := newIdleState(idle, http.NewResponseController(w), cancel)
35	-	iw.renew()	56	+	// finish marks the state done under the mutex before this
			57	+	// function returns, so nothing touches the controller afterwards.
			58	+	defer state.finish()
			59	+
			60	+	req := r.WithContext(ctx)
36		if r.Body != nil && r.Body != http.NoBody {	61		if r.Body != nil && r.Body != http.NoBody {
37	-	ib := &idleBody{ReadCloser: r.Body, rc: rc, idle: idle, writer: iw}	62	+	req.Body = &idleBody{ReadCloser: r.Body, state: state}
38	-	ib.renew()
39	-	r.Body = ib
40		}	63		}
41	-	next.ServeHTTP(iw, r)	64	+	next.ServeHTTP(&idleResponseWriter{ResponseWriter: w, state: state}, req)
42		})	65		})
43		}	66		}
44		}	67		}
45			68
46	-	// idleResponseWriter renews the write deadline on every Write/Flush.	69	+	// idleState is the request-lifetime state shared by the writer, the body and
47	-	type idleResponseWriter struct {	70	+	// the timer. The controller is only used while the state is live, under mu.
48	-	http.ResponseWriter	71	+	type idleState struct {
49	-	rc *http.ResponseController	72	+	idle time.Duration
50	-	idle time.Duration	73	+	rc *http.ResponseController
			74	+	cancel context.CancelFunc
51			75
52	-	mu sync.Mutex	76	+	mu sync.Mutex
53	-	deadline time.Time	77	+	timer *time.Timer
			78	+	last time.Time
			79	+	done bool // handler returned or connection hijacked
54		}	80		}
55			81
56	-	// renew extends the write deadline unless it already expired. An expired	82	+	func newIdleState(idle time.Duration, rc http.ResponseController, cancel context.CancelFunc) idleState {
57	-	// deadline is left in place so that the write fails: the response was silent	83	+	s := &idleState{idle: idle, rc: rc, cancel: cancel, last: time.Now()}
58	-	// for longer than the idle timeout.	84	+	s.mu.Lock()
59	-	func (w *idleResponseWriter) renew() {	85	+	s.timer = time.AfterFunc(idle, s.onTimer)
60	-	w.mu.Lock()	86	+	s.mu.Unlock()
61	-	defer w.mu.Unlock()	87	+	return s
62	-	now := time.Now()	88	+	}
63	-	if !w.deadline.IsZero() && now.After(w.deadline) {	89	+
			90	+	// progress records activity. It never touches the controller.
			91	+	func (s *idleState) progress() {
			92	+	s.mu.Lock()
			93	+	if !s.done {
			94	+	s.last = time.Now()
			95	+	}
			96	+	s.mu.Unlock()
			97	+	}
			98	+
			99	+	// live reports whether the request is still being served.
			100	+	func (s *idleState) live() bool {
			101	+	s.mu.Lock()
			102	+	defer s.mu.Unlock()
			103	+	return !s.done
			104	+	}
			105	+
			106	+	// onTimer runs on timer expiry. Progress only stamps s.last, so the timer is
			107	+	// re-armed for the remainder instead of being reset on every event; this also
			108	+	// makes progress racing with expiry safe.
			109	+	func (s *idleState) onTimer() {
			110	+	s.mu.Lock()
			111	+	defer s.mu.Unlock()
			112	+	if s.done {
			113	+	return
			114	+	}
			115	+	if remaining := s.idle - time.Since(s.last); remaining > 0 {
			116	+	s.timer.Reset(remaining)
64		return	117		return
65		}	118		}
66	-	next := now.Add(w.idle)	119	+	s.cancel()
67	-	if err := w.rc.SetWriteDeadline(next); err == nil {	120	+	// Held under mu so finish() cannot complete while these run.
68	-	w.deadline = next	121	+	now := time.Now()
			122	+	_ = s.rc.SetWriteDeadline(now)
			123	+	_ = s.rc.SetReadDeadline(now)
			124	+	}
			125	+
			126	+	// finish ends the request lifetime. After it returns, no code path calls the
			127	+	// controller. It is safe to call more than once.
			128	+	func (s *idleState) finish() {
			129	+	s.mu.Lock()
			130	+	s.done = true
			131	+	s.timer.Stop()
			132	+	s.mu.Unlock()
			133	+	s.cancel()
			134	+	}
			135	+
			136	+	// flush flushes through the controller unless the request finished. The mutex
			137	+	// is not held during the flush: it may block on a slow client, and the timer
			138	+	// callback needs the mutex to expire the deadlines that unblock it. net/http
			139	+	// forbids using the writer after the handler returned or concurrently with its
			140	+	// return, so the live check is a guard against misuse, not a synchronisation
			141	+	// point for the flush itself.
			142	+	func (s *idleState) flush() error {
			143	+	if !s.live() {
			144	+	return errTransferFinished
69		}	145		}
			146	+	return s.rc.Flush()
70		}	147		}
71			148
72	-	// extend pushes the write deadline forward unconditionally. It is called when	149	+	func (s idleState) hijack() (net.Conn, bufio.ReadWriter, error) {
73	-	// the request body makes progress: a client that is still uploading has not	150	+	s.mu.Lock()
74	-	// left the response idle, so the wait for the first response byte starts from	151	+	defer s.mu.Unlock()
75	-	// the last upload activity.	152	+	if s.done {
76	-	func (w *idleResponseWriter) extend() {	153	+	return nil, nil, errTransferFinished
77	-	w.mu.Lock()	154	+	}
78	-	defer w.mu.Unlock()	155	+	conn, rw, err := s.rc.Hijack()
79	-	next := time.Now().Add(w.idle)	156	+	if err == nil {
80	-	if err := w.rc.SetWriteDeadline(next); err == nil {	157	+	s.done = true
81	-	w.deadline = next	158	+	s.timer.Stop()
82		}	159		}
			160	+	return conn, rw, err
83		}	161		}
84			162
85	-	// maxIdleWriteChunk bounds how much is handed to the connection per deadline	163	+	// idleResponseWriter reports response progress.
86	-	// renewal, so one very large Write to a slow but steady client is not cut as if	164	+	type idleResponseWriter struct {
87	-	// it were stalled.	165	+	http.ResponseWriter
88	-	const maxIdleWriteChunk = 64 * 1024	166	+	state *idleState
			167	+	}
89			168
90	-	// Write renews the write deadline before each chunk, then writes it.	169	+	// Write hands p to the connection in idleWriteChunk pieces and records progress
			170	+	// after each piece.
91		func (w *idleResponseWriter) Write(p []byte) (int, error) {	171		func (w *idleResponseWriter) Write(p []byte) (int, error) {
			172	+	if !w.state.live() {
			173	+	return 0, errTransferFinished
			174	+	}
92		if len(p) == 0 {	175		if len(p) == 0 {
93	-	w.renew()
94		return w.ResponseWriter.Write(p)	176		return w.ResponseWriter.Write(p)
95		}	177		}
96		written := 0	178		written := 0
97		for written < len(p) {	179		for written < len(p) {
98	-	chunk := p[written:min(written+maxIdleWriteChunk, len(p))]	180	+	chunk := p[written:min(written+idleWriteChunk, len(p))]
99	-	w.renew()
100		n, err := w.ResponseWriter.Write(chunk)	181		n, err := w.ResponseWriter.Write(chunk)
101		written += n	182		written += n
			183	+	if n > 0 {
			184	+	w.state.progress()
			185	+	}
102		if err != nil {	186		if err != nil {
103		return written, err	187		return written, err
104		}	188		}

@@ -106,11 +190,14 @@ func (w *idleResponseWriter) Write(p []byte) (int, error) {
106		return written, nil	190		return written, nil
107		}	191		}
108			192
109	-	// FlushError renews the write deadline, then flushes. It lets	193	+	// FlushError flushes and records progress. It lets http.ResponseController
110	-	// http.ResponseController report flush failures.	194	+	// report flush failures.
111		func (w *idleResponseWriter) FlushError() error {	195		func (w *idleResponseWriter) FlushError() error {
112	-	w.renew()	196	+	err := w.state.flush()
113	-	return w.rc.Flush()	197	+	if err == nil {
			198	+	w.state.progress()
			199	+	}
			200	+	return err
114		}	201		}
115			202
116		// Flush implements http.Flusher.	203		// Flush implements http.Flusher.

@@ -118,11 +205,10 @@ func (w *idleResponseWriter) Flush() {
118		_ = w.FlushError()	205		_ = w.FlushError()
119		}	206		}
120			207
121	-	// Hijack implements http.Hijacker. The server clears connection deadlines when	208	+	// Hijack implements http.Hijacker and stops the inactivity timer: upgraded
122	-	// a connection is hijacked, so upgraded streams (e.g. WebSocket) are not	209	+	// streams (e.g. WebSocket) are exempt.
123	-	// subject to the inactivity timeout.
124		func (w idleResponseWriter) Hijack() (net.Conn, bufio.ReadWriter, error) {	210		func (w idleResponseWriter) Hijack() (net.Conn, bufio.ReadWriter, error) {
125	-	return w.rc.Hijack()	211	+	return w.state.hijack()
126		}	212		}
127			213
128		// Unwrap exposes the underlying writer to http.ResponseController.	214		// Unwrap exposes the underlying writer to http.ResponseController.

@@ -130,42 +216,27 @@ func (w *idleResponseWriter) Unwrap() http.ResponseWriter {
130		return w.ResponseWriter	216		return w.ResponseWriter
131		}	217		}
132			218
133	-	// idleBody renews the read deadline before every request body Read.	219	+	// idleBody reports request body progress. It only records timestamps, so it is
			220	+	// safe to Read or Close after the handler returned (the HTTP/2 transport closes
			221	+	// the outbound body asynchronously).
134		type idleBody struct {	222		type idleBody struct {
135		io.ReadCloser	223		io.ReadCloser
136	-	rc *http.ResponseController	224	+	state *idleState
137	-	idle time.Duration
138	-	writer *idleResponseWriter
139	-	}
140			225
141	-	func (b *idleBody) renew() {	226	+	closeOnce sync.Once
142	-	_ = b.rc.SetReadDeadline(time.Now().Add(b.idle))	227	+	closeErr error
143		}	228		}
144			229
145	-	// clear removes the read deadline. It must not stay armed once the body is
146	-	// fully read: net/http then watches the connection for client disconnects and
147	-	// treats a read timeout as an abort.
148	-	func (b *idleBody) clear() {
149	-	_ = b.rc.SetReadDeadline(time.Time{})
150	-	}
151	-
152	-	// Read renews the read deadline, reads, and clears the deadline at EOF.
153		func (b *idleBody) Read(p []byte) (int, error) {	230		func (b *idleBody) Read(p []byte) (int, error) {
154	-	b.renew()
155		n, err := b.ReadCloser.Read(p)	231		n, err := b.ReadCloser.Read(p)
156	-	if n > 0 \|\| errors.Is(err, io.EOF) {	232	+	if n > 0 {
157	-	b.writer.extend()	233	+	b.state.progress()
158	-	}
159	-	if errors.Is(err, io.EOF) {
160	-	b.clear()
161		}	234		}
162		return n, err	235		return n, err
163		}	236		}
164			237
165	-	// Close drains within a fresh deadline, then clears it.	238	+	// Close is idempotent.
166		func (b *idleBody) Close() error {	239		func (b *idleBody) Close() error {
167	-	b.renew()	240	+	b.closeOnce.Do(func() { b.closeErr = b.ReadCloser.Close() })
168	-	err := b.ReadCloser.Close()	241	+	return b.closeErr
169	-	b.clear()
170	-	return err
171		}	242		}

M internal/adapters/in/http/middleware/transfer_idle_test.go +277 -0


@@ -2,11 +2,16 @@ package middleware
2	2
3	3		import (
4	4		"bufio"
	5	+	"context"
	6	+	"crypto/tls"
5	7		"io"
6	8		"net"
7	9		"net/http"
8	10		"net/http/httptest"
	11	+	"net/http/httputil"
	12	+	"net/url"
9	13		"strings"
	14	+	"sync"
10	15		"sync/atomic"
11	16		"testing"
12	17		"time"

@@ -227,3 +232,275 @@ func TestTransferIdleTimeout_HTTP2(t *testing.T) {
227	232		require.NoError(t, err)
228	233		assert.Len(t, body, chunks)
229	234		}
	235	+
	236	+	// pipeListener serves connections created with dial over net.Pipe, which has no
	237	+	// kernel buffering: a write completes only when the peer reads it.
	238	+	type pipeListener struct {
	239	+	conns chan net.Conn
	240	+	done chan struct{}
	241	+	}
	242	+
	243	+	func newPipeListener() *pipeListener {
	244	+	return &pipeListener{conns: make(chan net.Conn), done: make(chan struct{})}
	245	+	}
	246	+
	247	+	func (l *pipeListener) Accept() (net.Conn, error) {
	248	+	select {
	249	+	case c := <-l.conns:
	250	+	return c, nil
	251	+	case <-l.done:
	252	+	return nil, net.ErrClosed
	253	+	}
	254	+	}
	255	+
	256	+	func (l *pipeListener) Close() error {
	257	+	select {
	258	+	case <-l.done:
	259	+	default:
	260	+	close(l.done)
	261	+	}
	262	+	return nil
	263	+	}
	264	+
	265	+	func (l *pipeListener) Addr() net.Addr { return pipeAddr{} }
	266	+
	267	+	func (l *pipeListener) dial() net.Conn {
	268	+	client, server := net.Pipe()
	269	+	l.conns <- server
	270	+	return client
	271	+	}
	272	+
	273	+	type pipeAddr struct{}
	274	+
	275	+	func (pipeAddr) Network() string { return "pipe" }
	276	+	func (pipeAddr) String() string { return "pipe" }
	277	+
	278	+	// slowReadConn reads at most 4 KiB every pause, like a congested client.
	279	+	type slowReadConn struct {
	280	+	net.Conn
	281	+	pause time.Duration
	282	+	}
	283	+
	284	+	func (c *slowReadConn) Read(p []byte) (int, error) {
	285	+	time.Sleep(c.pause)
	286	+	return c.Conn.Read(p[:min(len(p), 4096)])
	287	+	}
	288	+
	289	+	func servePipe(t testing.T, handler http.Handler) pipeListener {
	290	+	t.Helper()
	291	+	ln := newPipeListener()
	292	+	srv := &http.Server{Handler: TransferIdleTimeout(testIdle)(handler), ReadHeaderTimeout: 5 * time.Second}
	293	+	go func() { _ = srv.Serve(ln) }()
	294	+	t.Cleanup(func() { _ = srv.Close() })
	295	+	return ln
	296	+	}
	297	+
	298	+	func TestTransferIdleTimeout_BackpressuredButReadingClientIsNotCut(t *testing.T) {
	299	+	const size = 64 * 1024
	300	+	ln := servePipe(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
	301	+	_, _ = w.Write([]byte(strings.Repeat("d", size)))
	302	+	}))
	303	+	client := &http.Client{Transport: &http.Transport{
	304	+	DisableKeepAlives: true,
	305	+	DialContext: func(context.Context, string, string) (net.Conn, error) {
	306	+	return &slowReadConn{Conn: ln.dial(), pause: 25 * time.Millisecond}, nil
	307	+	},
	308	+	}}
	309	+
	310	+	start := time.Now()
	311	+	resp, err := client.Get("http://pipe/")
	312	+	require.NoError(t, err)
	313	+	defer resp.Body.Close()
	314	+	body, err := io.ReadAll(resp.Body)
	315	+	require.NoError(t, err, "a client that keeps reading must not be cut")
	316	+	assert.Len(t, body, size)
	317	+	assert.Greater(t, time.Since(start), testIdle, "the transfer must outlast the idle timeout")
	318	+	}
	319	+
	320	+	func TestTransferIdleTimeout_NonReadingClientUnblocksWrite(t *testing.T) {
	321	+	writeErr := make(chan error, 1)
	322	+	ln := servePipe(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
	323	+	_, err := w.Write([]byte(strings.Repeat("d", 1<<20)))
	324	+	writeErr <- err
	325	+	}))
	326	+	conn := ln.dial()
	327	+	defer conn.Close()
	328	+	_, err := conn.Write([]byte("GET / HTTP/1.1\r\nHost: x\r\n\r\n"))
	329	+	require.NoError(t, err)
	330	+
	331	+	select {
	332	+	case err := <-writeErr:
	333	+	require.Error(t, err, "write to a client that never reads must fail")
	334	+	case <-time.After(10 * testIdle):
	335	+	t.Fatal("blocked write was not unblocked by the inactivity timeout")
	336	+	}
	337	+	}
	338	+
	339	+	func TestTransferIdleTimeout_SilentHandlerContextCanceledAndConnectionClosed(t *testing.T) {
	340	+	ctxDone := make(chan struct{})
	341	+	srv := newIdleServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	342	+	_, _ = w.Write([]byte("start"))
	343	+	w.(http.Flusher).Flush()
	344	+	<-r.Context().Done() // a ReverseProxy would abort its upstream request here
	345	+	close(ctxDone)
	346	+	}))
	347	+
	348	+	client := &http.Client{Timeout: 10 * time.Second}
	349	+	begin := time.Now()
	350	+	resp, err := client.Get(srv.URL)
	351	+	require.NoError(t, err)
	352	+	defer resp.Body.Close()
	353	+	body, err := io.ReadAll(resp.Body)
	354	+	require.Error(t, err, "the client connection must be closed, not completed cleanly")
	355	+	assert.Equal(t, "start", string(body))
	356	+	assert.Less(t, time.Since(begin), 5*time.Second)
	357	+
	358	+	select {
	359	+	case <-ctxDone:
	360	+	case <-time.After(5 * time.Second):
	361	+	t.Fatal("request context was not canceled")
	362	+	}
	363	+	}
	364	+
	365	+	func TestTransferIdleTimeout_ProgressKeepsContextAlive(t *testing.T) {
	366	+	var canceled atomic.Bool
	367	+	srv := newIdleServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	368	+	for range 12 {
	369	+	_, _ = w.Write([]byte("x"))
	370	+	w.(http.Flusher).Flush()
	371	+	select {
	372	+	case <-r.Context().Done():
	373	+	canceled.Store(true)
	374	+	return
	375	+	case <-time.After(50 * time.Millisecond):
	376	+	}
	377	+	}
	378	+	}))
	379	+	resp, err := http.Get(srv.URL)
	380	+	require.NoError(t, err)
	381	+	defer resp.Body.Close()
	382	+	body, err := io.ReadAll(resp.Body)
	383	+	require.NoError(t, err)
	384	+	assert.Len(t, body, 12)
	385	+	assert.False(t, canceled.Load())
	386	+	}
	387	+
	388	+	func TestTransferIdleTimeout_BodyUseAfterHandlerReturnedIsSafe(t *testing.T) {
	389	+	var body io.ReadCloser
	390	+	h := TransferIdleTimeout(time.Second)(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
	391	+	body = r.Body
	392	+	}))
	393	+	req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("payload"))
	394	+	h.ServeHTTP(httptest.NewRecorder(), req)
	395	+
	396	+	// Like http.Transport closing the outbound body after the handler returned.
	397	+	require.NoError(t, body.Close())
	398	+	require.NoError(t, body.Close(), "Close must be idempotent")
	399	+	_, _ = body.Read(make([]byte, 4))
	400	+	}
	401	+
	402	+	// TestTransferIdleTimeout_HTTP2ReverseProxyEarlyResponseDuringUpload covers the
	403	+	// transport closing the outbound request body after the handler returned, when
	404	+	// the upstream answers early (413) while the client is still uploading. Run
	405	+	// with -race.
	406	+	func TestTransferIdleTimeout_HTTP2ReverseProxyEarlyResponseDuringUpload(t *testing.T) {
	407	+	upstream := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
	408	+	http.Error(w, "too large", http.StatusRequestEntityTooLarge)
	409	+	}))
	410	+	upstream.EnableHTTP2 = true
	411	+	upstream.StartTLS()
	412	+	defer upstream.Close()
	413	+
	414	+	target, err := url.Parse(upstream.URL)
	415	+	require.NoError(t, err)
	416	+	proxy := httputil.NewSingleHostReverseProxy(target)
	417	+	proxy.Transport = &http.Transport{
	418	+	TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // test upstream
	419	+	ForceAttemptHTTP2: true,
	420	+	}
	421	+	front := httptest.NewUnstartedServer(TransferIdleTimeout(testIdle)(proxy))
	422	+	front.EnableHTTP2 = true
	423	+	front.StartTLS()
	424	+	defer front.Close()
	425	+
	426	+	var wg sync.WaitGroup
	427	+	for range 8 {
	428	+	wg.Add(1)
	429	+	go func() {
	430	+	defer wg.Done()
	431	+	for range 10 {
	432	+	uploadUntilEarlyResponse(t, front)
	433	+	}
	434	+	}()
	435	+	}
	436	+	wg.Wait()
	437	+	}
	438	+
	439	+	func uploadUntilEarlyResponse(t testing.T, front httptest.Server) {
	440	+	t.Helper()
	441	+	pr, pw := io.Pipe()
	442	+	writerDone := make(chan struct{})
	443	+	go func() {
	444	+	defer close(writerDone)
	445	+	chunk := []byte(strings.Repeat("u", 4096))
	446	+	for range 200 {
	447	+	if _, err := pw.Write(chunk); err != nil {
	448	+	return
	449	+	}
	450	+	time.Sleep(time.Millisecond)
	451	+	}
	452	+	_ = pw.Close()
	453	+	}()
	454	+
	455	+	req, err := http.NewRequest(http.MethodPost, front.URL, pr)
	456	+	if !assert.NoError(t, err) {
	457	+	return
	458	+	}
	459	+	resp, err := front.Client().Do(req)
	460	+	if err == nil {
	461	+	assert.Equal(t, 2, resp.ProtoMajor)
	462	+	assert.Equal(t, http.StatusRequestEntityTooLarge, resp.StatusCode)
	463	+	_, _ = io.Copy(io.Discard, resp.Body)
	464	+	_ = resp.Body.Close()
	465	+	}
	466	+	_ = pr.CloseWithError(io.ErrClosedPipe)
	467	+	<-writerDone
	468	+	}
	469	+
	470	+	// TestTransferIdleTimeout_HTTP2BodyUsedAfterHandlerReturned uses the request
	471	+	// body after net/http's HTTP/2 server finished the handler (it drops its
	472	+	// response writer state then), as http.Transport does when it closes an
	473	+	// outbound body asynchronously. It must neither panic nor race.
	474	+	func TestTransferIdleTimeout_HTTP2BodyUsedAfterHandlerReturned(t *testing.T) {
	475	+	bodyCh := make(chan io.ReadCloser, 1)
	476	+	returned := make(chan struct{})
	477	+	inner := TransferIdleTimeout(testIdle)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	478	+	bodyCh <- r.Body
	479	+	_, _ = w.Write([]byte("early"))
	480	+	}))
	481	+	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	482	+	inner.ServeHTTP(w, r)
	483	+	close(returned)
	484	+	}))
	485	+	srv.EnableHTTP2 = true
	486	+	srv.StartTLS()
	487	+	defer srv.Close()
	488	+
	489	+	pr, pw := io.Pipe()
	490	+	defer pw.Close()
	491	+	req, err := http.NewRequest(http.MethodPost, srv.URL, pr)
	492	+	require.NoError(t, err)
	493	+	go func() { _, _ = pw.Write([]byte("partial")) }()
	494	+	resp, err := srv.Client().Do(req)
	495	+	require.NoError(t, err)
	496	+	_, _ = io.Copy(io.Discard, resp.Body)
	497	+	_ = resp.Body.Close()
	498	+
	499	+	body := <-bodyCh
	500	+	<-returned
	501	+	time.Sleep(100 * time.Millisecond) // let the server loop process handlerDone
	502	+
	503	+	_, _ = body.Read(make([]byte, 8))
	504	+	require.NoError(t, body.Close())
	505	+	require.NoError(t, body.Close())
	506	+	}


@@ -2,11 +2,16 @@ package middleware
2		2
3	import (	3		import (
4	"bufio"	4		"bufio"
		5	+	"context"
		6	+	"crypto/tls"
5	"io"	7		"io"
6	"net"	8		"net"
7	"net/http"	9		"net/http"
8	"net/http/httptest"	10		"net/http/httptest"
		11	+	"net/http/httputil"
		12	+	"net/url"
9	"strings"	13		"strings"
		14	+	"sync"
10	"sync/atomic"	15		"sync/atomic"
11	"testing"	16		"testing"
12	"time"	17		"time"

@@ -227,3 +232,275 @@ func TestTransferIdleTimeout_HTTP2(t *testing.T) {
227	require.NoError(t, err)	232		require.NoError(t, err)
228	assert.Len(t, body, chunks)	233		assert.Len(t, body, chunks)
229	}	234		}
		235	+
		236	+	// pipeListener serves connections created with dial over net.Pipe, which has no
		237	+	// kernel buffering: a write completes only when the peer reads it.
		238	+	type pipeListener struct {
		239	+	conns chan net.Conn
		240	+	done chan struct{}
		241	+	}
		242	+
		243	+	func newPipeListener() *pipeListener {
		244	+	return &pipeListener{conns: make(chan net.Conn), done: make(chan struct{})}
		245	+	}
		246	+
		247	+	func (l *pipeListener) Accept() (net.Conn, error) {
		248	+	select {
		249	+	case c := <-l.conns:
		250	+	return c, nil
		251	+	case <-l.done:
		252	+	return nil, net.ErrClosed
		253	+	}
		254	+	}
		255	+
		256	+	func (l *pipeListener) Close() error {
		257	+	select {
		258	+	case <-l.done:
		259	+	default:
		260	+	close(l.done)
		261	+	}
		262	+	return nil
		263	+	}
		264	+
		265	+	func (l *pipeListener) Addr() net.Addr { return pipeAddr{} }
		266	+
		267	+	func (l *pipeListener) dial() net.Conn {
		268	+	client, server := net.Pipe()
		269	+	l.conns <- server
		270	+	return client
		271	+	}
		272	+
		273	+	type pipeAddr struct{}
		274	+
		275	+	func (pipeAddr) Network() string { return "pipe" }
		276	+	func (pipeAddr) String() string { return "pipe" }
		277	+
		278	+	// slowReadConn reads at most 4 KiB every pause, like a congested client.
		279	+	type slowReadConn struct {
		280	+	net.Conn
		281	+	pause time.Duration
		282	+	}
		283	+
		284	+	func (c *slowReadConn) Read(p []byte) (int, error) {
		285	+	time.Sleep(c.pause)
		286	+	return c.Conn.Read(p[:min(len(p), 4096)])
		287	+	}
		288	+
		289	+	func servePipe(t testing.T, handler http.Handler) pipeListener {
		290	+	t.Helper()
		291	+	ln := newPipeListener()
		292	+	srv := &http.Server{Handler: TransferIdleTimeout(testIdle)(handler), ReadHeaderTimeout: 5 * time.Second}
		293	+	go func() { _ = srv.Serve(ln) }()
		294	+	t.Cleanup(func() { _ = srv.Close() })
		295	+	return ln
		296	+	}
		297	+
		298	+	func TestTransferIdleTimeout_BackpressuredButReadingClientIsNotCut(t *testing.T) {
		299	+	const size = 64 * 1024
		300	+	ln := servePipe(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		301	+	_, _ = w.Write([]byte(strings.Repeat("d", size)))
		302	+	}))
		303	+	client := &http.Client{Transport: &http.Transport{
		304	+	DisableKeepAlives: true,
		305	+	DialContext: func(context.Context, string, string) (net.Conn, error) {
		306	+	return &slowReadConn{Conn: ln.dial(), pause: 25 * time.Millisecond}, nil
		307	+	},
		308	+	}}
		309	+
		310	+	start := time.Now()
		311	+	resp, err := client.Get("http://pipe/")
		312	+	require.NoError(t, err)
		313	+	defer resp.Body.Close()
		314	+	body, err := io.ReadAll(resp.Body)
		315	+	require.NoError(t, err, "a client that keeps reading must not be cut")
		316	+	assert.Len(t, body, size)
		317	+	assert.Greater(t, time.Since(start), testIdle, "the transfer must outlast the idle timeout")
		318	+	}
		319	+
		320	+	func TestTransferIdleTimeout_NonReadingClientUnblocksWrite(t *testing.T) {
		321	+	writeErr := make(chan error, 1)
		322	+	ln := servePipe(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		323	+	_, err := w.Write([]byte(strings.Repeat("d", 1<<20)))
		324	+	writeErr <- err
		325	+	}))
		326	+	conn := ln.dial()
		327	+	defer conn.Close()
		328	+	_, err := conn.Write([]byte("GET / HTTP/1.1\r\nHost: x\r\n\r\n"))
		329	+	require.NoError(t, err)
		330	+
		331	+	select {
		332	+	case err := <-writeErr:
		333	+	require.Error(t, err, "write to a client that never reads must fail")
		334	+	case <-time.After(10 * testIdle):
		335	+	t.Fatal("blocked write was not unblocked by the inactivity timeout")
		336	+	}
		337	+	}
		338	+
		339	+	func TestTransferIdleTimeout_SilentHandlerContextCanceledAndConnectionClosed(t *testing.T) {
		340	+	ctxDone := make(chan struct{})
		341	+	srv := newIdleServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		342	+	_, _ = w.Write([]byte("start"))
		343	+	w.(http.Flusher).Flush()
		344	+	<-r.Context().Done() // a ReverseProxy would abort its upstream request here
		345	+	close(ctxDone)
		346	+	}))
		347	+
		348	+	client := &http.Client{Timeout: 10 * time.Second}
		349	+	begin := time.Now()
		350	+	resp, err := client.Get(srv.URL)
		351	+	require.NoError(t, err)
		352	+	defer resp.Body.Close()
		353	+	body, err := io.ReadAll(resp.Body)
		354	+	require.Error(t, err, "the client connection must be closed, not completed cleanly")
		355	+	assert.Equal(t, "start", string(body))
		356	+	assert.Less(t, time.Since(begin), 5*time.Second)
		357	+
		358	+	select {
		359	+	case <-ctxDone:
		360	+	case <-time.After(5 * time.Second):
		361	+	t.Fatal("request context was not canceled")
		362	+	}
		363	+	}
		364	+
		365	+	func TestTransferIdleTimeout_ProgressKeepsContextAlive(t *testing.T) {
		366	+	var canceled atomic.Bool
		367	+	srv := newIdleServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		368	+	for range 12 {
		369	+	_, _ = w.Write([]byte("x"))
		370	+	w.(http.Flusher).Flush()
		371	+	select {
		372	+	case <-r.Context().Done():
		373	+	canceled.Store(true)
		374	+	return
		375	+	case <-time.After(50 * time.Millisecond):
		376	+	}
		377	+	}
		378	+	}))
		379	+	resp, err := http.Get(srv.URL)
		380	+	require.NoError(t, err)
		381	+	defer resp.Body.Close()
		382	+	body, err := io.ReadAll(resp.Body)
		383	+	require.NoError(t, err)
		384	+	assert.Len(t, body, 12)
		385	+	assert.False(t, canceled.Load())
		386	+	}
		387	+
		388	+	func TestTransferIdleTimeout_BodyUseAfterHandlerReturnedIsSafe(t *testing.T) {
		389	+	var body io.ReadCloser
		390	+	h := TransferIdleTimeout(time.Second)(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
		391	+	body = r.Body
		392	+	}))
		393	+	req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("payload"))
		394	+	h.ServeHTTP(httptest.NewRecorder(), req)
		395	+
		396	+	// Like http.Transport closing the outbound body after the handler returned.
		397	+	require.NoError(t, body.Close())
		398	+	require.NoError(t, body.Close(), "Close must be idempotent")
		399	+	_, _ = body.Read(make([]byte, 4))
		400	+	}
		401	+
		402	+	// TestTransferIdleTimeout_HTTP2ReverseProxyEarlyResponseDuringUpload covers the
		403	+	// transport closing the outbound request body after the handler returned, when
		404	+	// the upstream answers early (413) while the client is still uploading. Run
		405	+	// with -race.
		406	+	func TestTransferIdleTimeout_HTTP2ReverseProxyEarlyResponseDuringUpload(t *testing.T) {
		407	+	upstream := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		408	+	http.Error(w, "too large", http.StatusRequestEntityTooLarge)
		409	+	}))
		410	+	upstream.EnableHTTP2 = true
		411	+	upstream.StartTLS()
		412	+	defer upstream.Close()
		413	+
		414	+	target, err := url.Parse(upstream.URL)
		415	+	require.NoError(t, err)
		416	+	proxy := httputil.NewSingleHostReverseProxy(target)
		417	+	proxy.Transport = &http.Transport{
		418	+	TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // test upstream
		419	+	ForceAttemptHTTP2: true,
		420	+	}
		421	+	front := httptest.NewUnstartedServer(TransferIdleTimeout(testIdle)(proxy))
		422	+	front.EnableHTTP2 = true
		423	+	front.StartTLS()
		424	+	defer front.Close()
		425	+
		426	+	var wg sync.WaitGroup
		427	+	for range 8 {
		428	+	wg.Add(1)
		429	+	go func() {
		430	+	defer wg.Done()
		431	+	for range 10 {
		432	+	uploadUntilEarlyResponse(t, front)
		433	+	}
		434	+	}()
		435	+	}
		436	+	wg.Wait()
		437	+	}
		438	+
		439	+	func uploadUntilEarlyResponse(t testing.T, front httptest.Server) {
		440	+	t.Helper()
		441	+	pr, pw := io.Pipe()
		442	+	writerDone := make(chan struct{})
		443	+	go func() {
		444	+	defer close(writerDone)
		445	+	chunk := []byte(strings.Repeat("u", 4096))
		446	+	for range 200 {
		447	+	if _, err := pw.Write(chunk); err != nil {
		448	+	return
		449	+	}
		450	+	time.Sleep(time.Millisecond)
		451	+	}
		452	+	_ = pw.Close()
		453	+	}()
		454	+
		455	+	req, err := http.NewRequest(http.MethodPost, front.URL, pr)
		456	+	if !assert.NoError(t, err) {
		457	+	return
		458	+	}
		459	+	resp, err := front.Client().Do(req)
		460	+	if err == nil {
		461	+	assert.Equal(t, 2, resp.ProtoMajor)
		462	+	assert.Equal(t, http.StatusRequestEntityTooLarge, resp.StatusCode)
		463	+	_, _ = io.Copy(io.Discard, resp.Body)
		464	+	_ = resp.Body.Close()
		465	+	}
		466	+	_ = pr.CloseWithError(io.ErrClosedPipe)
		467	+	<-writerDone
		468	+	}
		469	+
		470	+	// TestTransferIdleTimeout_HTTP2BodyUsedAfterHandlerReturned uses the request
		471	+	// body after net/http's HTTP/2 server finished the handler (it drops its
		472	+	// response writer state then), as http.Transport does when it closes an
		473	+	// outbound body asynchronously. It must neither panic nor race.
		474	+	func TestTransferIdleTimeout_HTTP2BodyUsedAfterHandlerReturned(t *testing.T) {
		475	+	bodyCh := make(chan io.ReadCloser, 1)
		476	+	returned := make(chan struct{})
		477	+	inner := TransferIdleTimeout(testIdle)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		478	+	bodyCh <- r.Body
		479	+	_, _ = w.Write([]byte("early"))
		480	+	}))
		481	+	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		482	+	inner.ServeHTTP(w, r)
		483	+	close(returned)
		484	+	}))
		485	+	srv.EnableHTTP2 = true
		486	+	srv.StartTLS()
		487	+	defer srv.Close()
		488	+
		489	+	pr, pw := io.Pipe()
		490	+	defer pw.Close()
		491	+	req, err := http.NewRequest(http.MethodPost, srv.URL, pr)
		492	+	require.NoError(t, err)
		493	+	go func() { _, _ = pw.Write([]byte("partial")) }()
		494	+	resp, err := srv.Client().Do(req)
		495	+	require.NoError(t, err)
		496	+	_, _ = io.Copy(io.Discard, resp.Body)
		497	+	_ = resp.Body.Close()
		498	+
		499	+	body := <-bodyCh
		500	+	<-returned
		501	+	time.Sleep(100 * time.Millisecond) // let the server loop process handlerDone
		502	+
		503	+	_, _ = body.Read(make([]byte, 8))
		504	+	require.NoError(t, body.Close())
		505	+	require.NoError(t, body.Close())
		506	+	}

M internal/adapters/in/traffic/http_server_test.go +11 -4


@@ -84,12 +84,19 @@ func TestSmartTCPStreamingResponseOutlivesTransferIdleTimeout(t *testing.T) {
84	84		manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
85	85		silent := r.URL.Path == "/silent"
86	86		for i := 0; i < 12; i++ {
87		-	_, _ = w.Write([]byte("x"))
	87	+	if _, err := w.Write([]byte("x")); err != nil {
	88	+	return
	89	+	}
88	90		w.(http.Flusher).Flush()
89	91		if silent {
90		-	time.Sleep(3 * idle)
91		-	} else {
92		-	time.Sleep(50 * time.Millisecond)
	92	+	// Stay silent until the inactivity timeout cancels the request.
	93	+	<-r.Context().Done()
	94	+	return
	95	+	}
	96	+	select {
	97	+	case <-r.Context().Done():
	98	+	return
	99	+	case <-time.After(50 * time.Millisecond):
93	100		}
94	101		}
95	102		}), nil)

 	manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		silent := r.URL.Path == "/silent"
 		for i := 0; i < 12; i++ {
-			_, _ = w.Write([]byte("x"))
+			if _, err := w.Write([]byte("x")); err != nil {
+				return
+			}
 			w.(http.Flusher).Flush()
 			if silent {
-				time.Sleep(3 * idle)
+				// Stay silent until the inactivity timeout cancels the request.
-			} else {
+				<-r.Context().Done()
-				time.Sleep(50 * time.Millisecond)
+				return
+			}
+			select {
+			case <-r.Context().Done():
+				return
+			case <-time.After(50 * time.Millisecond):
 			}
 		}
 	}), nil)