engines: HTTP/2 WriteTimeout bounds a response that waits for the client's window, as a minimum rate: stage 2 of the per-stream deadlines (v1.6.0, Refs #889)

.github/root-test-shards.txt

@@ -60,3 +60,32 @@ b TestGlobalMiddlewareOfAnAsyncRouteKeepsTheLoop848
 b TestH2MaxConcurrentStreamsHeadersContinuation981
 b TestNoDeprecatedPublicAPI830
 b TestHijackedConnNotCutByServerTimeoutsStd855
+
+# The write deadline's cells (celeris#889, D2): 3 to 28 s each at the root, 70 s together, shared out so that the rest stays inside its timeout.
+a TestH2SparseWindowUpdatesAreNotCut889
+a TestH2PausedAndDribbledStreamsOnAHealthyConnAreBounded889
+a TestH2StalledTailIsResetAtWriteTimeout889
+a TestH2WindowUpdateDribblerIsResetAtTheFirstJudgement889
+a TestH2SilentClientWithASmallRemainderIsLetGo889
+b TestH2ParallelDownloadsSharingASlowLinkAreNotCut889
+b TestH2WriteRateFloor889
+b TestH2ConnWindowStarvedStreamsAreJudgedPerConnection889
+b TestH2FlowingDownloadAndQuietStreamAreNotCut889
+b TestH2SilentClientWithAStalledResponseKeepsTheConnUntilTheStreamIsReset889
+
+# The rest step of the Unit job (test/server, -race, -timeout=540s) was at 539.2 s on the merge of #1169 (main alone: 501 s), 0.8 s from its timeout:
+# the 14 slowest tests of the root packages still in it (8 to 34 s each, 216 s of CI test time) move to the shards, which have 300 s of room.
+a TestShutdownConnCloseContract884
+a TestStreamWriterOverCapH1_875
+a TestDefaultMaxRequestBodySize4MiB889
+a TestConcurrentResponsesOwnTheirBodies
+a TestShutdownRunningAsyncHandlerAnswersCloseWithSlowDownloader884
+a TestShutdownPoolWaitStopsServingH1AndDetached884
+a TestSplitBodyResponsesKeepTheConnection
+b TestIOUringShutdownGateConns884
+b TestSeveralCookieLinesAreOneFieldOnEveryEngine889
+b TestExpectContinueOverLimit889
+b TestEpollStaticFileDownloadAtAModerateRateCompletes885
+b TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885
+b TestShutdownIdleAndInProgressH1ConnsWithoutHTTP2_1165
+b TestShutdownPoolWaitGateClosesIdleH1AndEndsDetached885

config.go

@@ -259,8 +259,9 @@ type Config struct {
 	// read. A request without a body has no deadline.
 	//
 	// The connection's own clock (the last read) does not close an HTTP/2
-	// connection that has a stream open whose request is whole and no response
-	// waiting for the client's flow-control window: a stream the server is
+	// connection that has a stream open whose request is whole (a response
+	// waiting for the client's flow-control window is bounded by WriteTimeout,
+	// see there, and no longer takes the exemption away): a stream the server is
 	// writing to (an SSE stream) outlives a client that sends no frame. A stream
 	// whose handler never answers is exempt for as long as it holds the stream,
 	// where an HTTP/1.1 async handler that overruns ReadTimeout is closed. When
@@ -310,7 +311,8 @@ type Config struct {
 	// OnDisconnect runs), whatever the middleware's idle deadline or the last
 	// read of the connection. A peer that reads, however slowly, and a
 	// connection with nothing queued are not closed (there is no minimum
-	// rate); -1 sets no such bound. The close comes between about 3/4
+	// rate; an HTTP/2 stream whose client lets its flow-control window run out
+	// is bounded by the minimum rate described below); -1 sets no such bound. The close comes between about 3/4
 	// and 5/4 of WriteTimeout after the peer last took a byte (its progress
 	// is sampled every quarter of WriteTimeout; a SEND completion restarts the
 	// clock at once), plus the engine's sweep interval (up to about 0.1 s on
@@ -372,10 +374,54 @@ type Config struct {
 	// last flush that took bytes there, as before the rule. The rule covers an
 	// HTTP/2 response parked on the socket too. A response that waits for the
 	// client's HTTP/2 flow-control window is not parked (nothing is queued on
-	// the socket) and WriteTimeout does not bound it on epoll: ReadTimeout and
-	// IdleTimeout do, from the client's last frame, so a client that sends a
-	// frame of any kind (a PING) at least once per ReadTimeout holds it, until
-	// per-stream deadlines are added. A request error read while a
+	// the socket); WriteTimeout bounds its stream in its own right, as a
+	// minimum rate, on the native engines: in each WriteTimeout window a stream
+	// that holds buffered bytes must have had at least 64 KiB of them (or all
+	// it held, if less) sent, released by the client's WINDOW_UPDATEs, or it is
+	// reset with RST_STREAM(INTERNAL_ERROR), as net/http resets a stream at its
+	// write deadline. There is no ceiling, where std's deadline is absolute: a
+	// stream that releases at least that on average, lagging the rate by no
+	// more than four windows' worth (surplus released in one update carries to
+	// the next windows, and the bytes in flight when the stream stalled cover its
+	// first), is not cut however long its response takes. A PING or a
+	// WINDOW_UPDATE of a byte is not progress. A client that updates in big steps
+	// (nghttp2 and curl, at half the window) is fine while the gap between its
+	// updates stays under about five WriteTimeouts: half the window divided by the
+	// rate it reads at. When the connection as a whole releases at least 64 KiB
+	// in a window, a stream that takes less (parallel downloads share the
+	// client's socket) is held while it takes half its fair share of it, and for
+	// three windows in a row when it takes less: a download queued behind its
+	// siblings' bytes in the client's socket is cut where the bytes ahead of it
+	// take longer than three WriteTimeouts to read (N streams of W bytes in flight
+	// on a link of L bytes/s: N x W / L), so a server whose clients are on slow
+	// links should set WriteTimeout to 5s or more. When the connection itself
+	// releases less than 64 KiB while it holds more, the first window is the
+	// verdict. The end of a response is given grace: a stream that holds less than
+	// 64 KiB, and has had half of it sent in the window, or any stream while the
+	// connection as a whole holds less than 64 KiB (the last bytes of the transfer
+	// are on the wire, and the grant for them can wait behind its siblings' bytes in
+	// the client's socket for more than a window), is held for three windows more
+	// and reset in the fourth if nothing moves it. A stream the client never
+	// grants is reset after 1+k windows alone, or 4+k with siblings that keep the
+	// connection moving, k being the number of whole 64 KiB in the bytes in flight
+	// when it stalled (about the peer's initial window), at most four: 1 and 4
+	// WriteTimeouts for the default 64 KiB window, 5 and 8 for a window of 256 KiB or
+	// more. When what it has left to send is under 64 KiB (the response is a
+	// little longer than the window), the bytes in flight are spent a whole floor a
+	// window, and the tail grace above follows: 5 WriteTimeouts for the default
+	// window, 6 for 128 KiB, 8 for 256 KiB or more. A one-byte WINDOW_UPDATE
+	// dribbler is reset in window max(2, 1+k): the 2nd of the default window, the
+	// 3rd of 128 KiB, the 4th of 192 KiB, the 5th of 256 KiB or more. A stream
+	// whose own window has room but whose bytes wait for the connection's is
+	// credited with what the whole connection released. While such a stream
+	// lives, ReadTimeout and IdleTimeout do not close the connection
+	// (WriteTimeout is its bound), so a silent client's unread response, and the
+	// buffer it pins, lives for up to 8 WriteTimeouts (5, with the default window)
+	// where it used to be dropped with the whole connection at ReadTimeout. A
+	// handler blocked waiting for window keeps its own, absolute, bound of
+	// WriteTimeout from the start of the wait, whatever the client reads: parallel
+	// downloads on one connection whose responses exceed the connection's 4 MiB
+	// output budget can lose the later ones to it. A request error read while a
 	// response is queued (a malformed request) puts the connection on the
 	// closing drain below, which counts each byte the client acknowledges and
 	// has no floor: a client that sends one and then takes a few bytes per

internal/conn/h2.go

@@ -625,6 +625,7 @@ func NewH2State(handler stream.Handler, cfg H2Config, write func([]byte), wake *
 	s.idleNs = int64(cfg.IdleTimeout)
 	proc.GetManager().SetHints(&s.hints)
 	proc.GetManager().SetRequestTimeout(cfg.ReadTimeout)
+	proc.GetManager().SetWriteTimeout(cfg.WriteTimeout)
 	proc.SetHeaderBlockTimeout(cfg.ReadHeaderTimeout)
 	if lg := cfg.Logger; lg != nil {
 		proc.OnHandlerPanic = func(id uint32, r any, stack []byte) {
@@ -1125,8 +1126,8 @@ func (s *H2State) Sweep(now int64, write func([]byte)) (res SweepResult) {
 			res.CloseNow = true
 		}
 	}
-	dl, td, hd := h.ReqDL.Load(), h.TombDL.Load(), h.HdrDL.Load()
-	if (dl == 0 || now < dl) && (td == 0 || now < td) && (hd == 0 || now < hd) {
+	dl, td, hd, wd := h.ReqDL.Load(), h.TombDL.Load(), h.HdrDL.Load(), h.WrDL.Load()
+	if (dl == 0 || now < dl) && (td == 0 || now < td) && (hd == 0 || now < hd) && (wd == 0 || now < wd) {
 		return res
 	}
 	if !s.mu.TryLock() {
@@ -1153,6 +1154,24 @@ func (s *H2State) Sweep(now int64, write func([]byte)) (res SweepResult) {
 			}
 		}
 	}
+	if wd != 0 && now >= wd {
+		// The write windows that have ended: the streams that released less than
+		// their floor are reset with RST_STREAM(INTERNAL_ERROR), in batches and
+		// under the same budget as the requests (a stalled client with many
+		// streams is one window's worth).
+		var ids [32]uint32
+		began := time.Now()
+		for {
+			got := m.ExpiredWrites(now, ids[:0])
+			for _, id := range got {
+				p.ExpireWrite(id)
+				res.Expired++
+			}
+			if len(got) < len(ids) || time.Since(began) > expireBudget {
+				break
+			}
+		}
+	}
 	if td != 0 && now >= td {
 		p.DropTombstones(now)
 	}

internal/conn/h2_bench_cfg_889_test.go

@@ -3,9 +3,8 @@ package conn
 import "time"
 
 // benchCfg889 is the H2Config the request-deadline benchmarks run with: the
-// defaults of ReadTimeout and ReadHeaderTimeout, so that a request with a body
-// arms its deadline. (main has no such fields; the comparison builds main with a
-// copy of this file that returns H2Config{}.)
+// defaults of ReadTimeout, ReadHeaderTimeout and WriteTimeout, so that a request
+// with a body arms its deadline and the write deadline is on.
 func benchCfg889() H2Config {
-	return H2Config{ReadTimeout: 60 * time.Second, ReadHeaderTimeout: 10 * time.Second}
+	return H2Config{ReadTimeout: 60 * time.Second, ReadHeaderTimeout: 10 * time.Second, WriteTimeout: 60 * time.Second}
 }

internal/conn/h2_sweep_889_test.go

@@ -9,6 +9,8 @@ import (
 
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
+
+	"github.com/goceleris/celeris/internal/protocol/h2/stream"
 )
 
 // H2State.Sweep (celeris#889): the engines' per-connection look at the request
@@ -537,3 +539,62 @@ func TestH2StateStaysInItsSizeClass889(t *testing.T) {
 		t.Errorf("sizeof(H2State) + the 8-byte malloc header = %d, want at most 704 (its size class)", got)
 	}
 }
+
+// TestSweepResetsAStalledStreamWithInternalError889: a stream that holds bytes for
+// the client's window and releases less than its floor in WriteTimeout is reset
+// by the sweep with RST_STREAM(INTERNAL_ERROR), written to the connection through
+// the write queue; one that released its floor is not.
+func TestSweepResetsAStalledStreamWithInternalError889(t *testing.T) {
+	r := newSweepRig889(t, H2Config{WriteTimeout: time.Second})
+	r.written()
+	m := r.st.processor.GetManager()
+	m.UpdateConnectionWindow(1 << 20)
+	// A client with an initial window of 0: nothing is in flight when a stream stalls,
+	// so the first window is judged on what is released and not on the seed.
+	r.frames(func(fr *http2.Framer) {
+		_ = fr.WriteSettings(http2.Setting{ID: http2.SettingInitialWindowSize, Val: 0})
+	})
+	mk := func(id uint32) *stream.Stream {
+		s := m.CreateStream(id)
+		s.SetState(stream.StateHalfClosedRemote)
+		s.SetWindowSize(0)
+		s.BufferOutbound(make([]byte, 1<<20), true)
+		return s
+	}
+	a, b := mk(1), mk(3)
+	now := time.Now().UnixNano()
+	if res := r.st.Sweep(now+int64(500*time.Millisecond), r.write); res.Expired != 0 {
+		t.Fatalf("expired half a window in: %+v", res)
+	}
+	// Stream 3 releases its floor in every window; stream 1 nothing. The connection
+	// moves, so stream 1 is held for three windows (it may be queued behind its
+	// sibling) and the fourth is the verdict.
+	var res SweepResult
+	for w := 1; w <= 4; w++ {
+		b.SetWindowSize(65536)
+		m.FlushOutbound(b, func(uint32, bool, []byte) {})
+		res = r.st.Sweep(now+int64(w)*int64(1100*time.Millisecond), r.write)
+		if (w < 4) != (res.Expired == 0) {
+			t.Fatalf("window %d: %+v", w, res)
+		}
+	}
+	if res.Expired != 1 || !res.Wrote {
+		t.Fatalf("sweep after the window: %+v, want one stream expired and the conn written", res)
+	}
+	var rsts []wframe889
+	for _, f := range r.written() {
+		if f.typ == http2.FrameRSTStream {
+			rsts = append(rsts, f)
+		}
+	}
+	if len(rsts) != 1 || rsts[0].stream != 1 || rsts[0].code != http2.ErrCodeInternal {
+		t.Fatalf("frames written: %+v, want RST_STREAM(INTERNAL_ERROR) for stream 1 only", rsts)
+	}
+	if _, ok := m.GetStream(1); ok {
+		t.Fatal("stream 1 is still on the connection")
+	}
+	if _, ok := m.GetStream(3); !ok {
+		t.Fatal("stream 3, which released its floor, was taken")
+	}
+	_ = a
+}

internal/protocol/h2/stream/deadline.go

@@ -163,7 +163,7 @@ func (p *Processor) TakeIgnoredActivity() bool {
 // ReadHeaderTimeout): ReadHeaderTimeout after the first HEADERS frame of a block
 // that does not end with it, the connection ends with GOAWAY(ENHANCE_YOUR_CALM).
 // 0 turns it off. Set once, before the first frame.
-func (p *Processor) SetHeaderBlockTimeout(d time.Duration) { p.hdrTimeout = d }
+func (p *Processor) SetHeaderBlockTimeout(d time.Duration) { p.hdrTimeout = clampTimeout(d) }
 
 // armHeaderBlock starts the deadline of the header block that has just begun
 // (a HEADERS frame without END_HEADERS).

internal/protocol/h2/stream/flowcontrol.go

@@ -94,6 +94,7 @@ func (m *Manager) ReserveSendWindow(s *Stream, want int32) int32 {
 			m.notifyStreamWindow(s.ID)
 		}
 	}
+	m.creditConn(connGrant)
 	return connGrant
 }
 

internal/protocol/h2/stream/manager.go

@@ -21,12 +21,14 @@ type Manager struct {
 	maxFrameSize            uint32 // peer's SETTINGS_MAX_FRAME_SIZE (caps what WE send)
 	localMaxFrameSize       uint32 // our advertised SETTINGS_MAX_FRAME_SIZE (caps what PEER sends us)
 	initialWindowSize       uint32
+	initWin                 atomic.Uint32 // initialWindowSize for readers that hold no lock (write_deadline.go)
 	activeStreams           atomic.Uint32
 	requestOpen             atomic.Uint32
 	hints                   *DeadlineHints
 	ownHints                DeadlineHints
 	reqTimeout              time.Duration
-	headerTableSize         uint32 // peer's SETTINGS_HEADER_TABLE_SIZE (bound on our HPACK dynamic table)
+	wrTimeout               time.Duration // the write deadline (SetWriteTimeout)
+	headerTableSize         uint32        // peer's SETTINGS_HEADER_TABLE_SIZE (bound on our HPACK dynamic table)
 	pendingConnWindowUpdate uint32
 	pendingStreamUpdates    map[uint32]uint32
 	windowUpdateMu          sync.Mutex
@@ -45,6 +47,15 @@ type Manager struct {
 	sendWindowMu      sync.Mutex
 	sendWaiters       []*sendWaiter
 
+	// The write deadline's books (write_deadline.go): a record per stream that
+	// holds buffered bytes, in stalls, guarded by stallMu; stallN is how many, for
+	// a load on the paths that have none; connReleased is the bytes of DATA the
+	// connection released while a stream was stalled (ReserveSendWindow).
+	stallMu      sync.Mutex
+	stalls       map[uint32]*stall
+	stallN       atomic.Int32
+	connReleased atomic.Int64
+
 	// retired holds the pool-handler streams whose handler has returned and
 	// which are out of the map, waiting for the event loop to return them to
 	// the stream pool (retire, drainRetired; celeris#951). hasRetired is
@@ -89,6 +100,7 @@ func NewManager() *Manager {
 		headerTableSize:   4096, // RFC 7540 §6.5.2 default
 	}
 	m.hints = &m.ownHints
+	m.initWin.Store(65535)
 	return m
 }
 
@@ -377,6 +389,7 @@ func (m *Manager) takeLocked(id uint32, s *Stream) (release bool) {
 	prev := State(s.state.Swap(int32(StateClosed)))
 	m.updateActiveCount(prev, StateClosed)
 	m.dropPlaceholderLocked(s)
+	m.dropStall(id)
 	f := s.flags.Load()
 	if f&flagReqDeadline != 0 {
 		m.disarmRequestLocked(s)
@@ -428,6 +441,7 @@ func (m *Manager) RemoveStreamFromMap(id uint32) {
 	delete(m.streams, id)
 	m.priorityTree.RemoveStream(id)
 	m.mu.Unlock()
+	m.dropStall(id)
 
 	m.windowUpdateMu.Lock()
 	delete(m.pendingStreamUpdates, id)
@@ -478,6 +492,7 @@ func (m *Manager) retire(s *Stream) {
 		s.OutboundBuffer = nil
 	}
 	id := s.ID
+	m.dropStall(id)
 	// The request is dropped now, not at the loop's next batch: a detached
 	// Context that reads the stream meanwhile sees nothing, as it did when
 	// this goroutine reset the stream (celeris#904), and a connection that
@@ -601,7 +616,7 @@ func (m *Manager) CountActiveStreams() int {
 func (m *Manager) ServerOwes() bool {
 	return m.activeStreams.Load() > 0 &&
 		(m.reqTimeout > 0 || m.requestOpen.Load() == 0) &&
-		m.outboundHeld.Load() == 0 &&
+		(m.wrTimeout > 0 || m.outboundHeld.Load() == 0) &&
 		m.sendWindowWaiters.Load() == 0
 }
 
@@ -684,6 +699,7 @@ func (m *Manager) ApplySetting(id uint16, val uint32) {
 		}
 		m.mu.Lock()
 		m.initialWindowSize = val
+		m.initWin.Store(val)
 		m.mu.Unlock()
 	case 0x5: // MAX_FRAME_SIZE
 		if val < 16384 || val > 16777215 {
@@ -791,6 +807,7 @@ func (m *Manager) Close() {
 	// that returns from here on releases its stream itself (retire).
 	m.closed = true
 	m.placeholders = 0
+	m.dropAllStalls()
 	retired := m.retired
 	m.retired = nil
 	m.hasRetired.Store(false)
@@ -809,7 +826,7 @@ func (m *Manager) Close() {
 // SetRequestTimeout sets the request deadline (the server's ReadTimeout): a
 // stream whose request is not whole that long after its HEADERS is expired by
 // ExpiredRequests. 0 turns it off. Set once, before the first frame.
-func (m *Manager) SetRequestTimeout(d time.Duration) { m.reqTimeout = d }
+func (m *Manager) SetRequestTimeout(d time.Duration) { m.reqTimeout = clampTimeout(d) }
 
 // RequestTimeout is the deadline SetRequestTimeout set.
 func (m *Manager) RequestTimeout() time.Duration { return m.reqTimeout }
@@ -873,6 +890,7 @@ type DeadlineHints struct {
 	ReqDL  atomic.Int64
 	TombDL atomic.Int64
 	HdrDL  atomic.Int64
+	WrDL   atomic.Int64 // the earliest end of a stalled stream's write window
 }
 
 // SetHints points the manager (and the processor that owns it) at h, which the

internal/protocol/h2/stream/outbound.go

@@ -464,6 +464,7 @@ func (m *Manager) FlushOutbound(s *Stream, send OutboundSend) (finished bool) {
 	finished = last && (s.OutboundEndStream || s.outboundAbandoned)
 	send(s.ID, end, buf.Next(n))
 	m.refundOutbound(n) // celeris#893
+	m.noteReleased(s.ID, n)
 	if buf.Len() == 0 {
 		buf.Reset()
 	}
@@ -538,6 +539,7 @@ func (m *Manager) sendOrBufferLocked(s *Stream, data []byte, send OutboundSend)
 		}
 		s.OutboundBuffer.Write(data[n:])
 		m.chargeOutbound(len(data) - n)
+		m.noteBuffered(s, len(data)-n)
 	}
 }
 
@@ -713,6 +715,7 @@ func (m *Manager) DiscardOutbound(s *Stream) {
 		m.refundOutbound(buf.Len())
 		buf.Reset()
 	}
+	m.dropStall(s.ID)
 	s.OutboundEndStream = false
 	s.outboundAbandoned = false
 }