diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 4ddc8e5..2925728 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -77,7 +77,7 @@ jobs: go mod verify go build ./... go vet ./... - go test -race -count=1 -timeout 120s ./... + go test -race -count=1 -timeout 300s ./... publish: name: Create tag and release diff --git a/README.md b/README.md index b52a503..9f82a34 100644 --- a/README.md +++ b/README.md @@ -391,7 +391,7 @@ empty (`omitempty`). | `errors` | int64 | Errors during the measurement window. | | `duration` | duration (ns) | Wall-clock measurement duration (post-warmup). | | `requests_per_sec` | float64 | `requests / duration.Seconds()`. | -| `throughput_bps` | float64 | Response-body bytes per second. Accurate for H1; the H2 client currently counts only the last DATA frame of each response, so multi-frame H2 bodies are undercounted. | +| `throughput_bps` | float64 | Response-body bytes per second, over successful responses: the Content-Length or the chunk sizes for H1, the data of every DATA frame for H2 (padding excluded, as chunk framing is for H1). | | `latency` | object | `{avg, min, max, p50, p75, p90, p99, p99_9, p99_99}`, each a duration in ns. | | `loadgen_version` | string | loadgen build that produced the run. *(optional)* | | `mode` | string | `"saturation"` or `"rated"`. *(optional)* | diff --git a/backoff.go b/backoff.go index f5c4601..da34578 100644 --- a/backoff.go +++ b/backoff.go @@ -46,13 +46,25 @@ type connectBackoff struct { // cancelled or abort is closed, so Close() and run shutdown are never stuck // behind a sleep. A nil abort channel is valid and never fires. func (b *connectBackoff) sleep(ctx context.Context, abort <-chan struct{}) bool { + return sleepFor(ctx, abort, b.step()) +} + +// step returns the current backoff interval, jittered uniformly down to half, +// and doubles the interval for the following call, capped at +// reconnectBackoffMax, without sleeping: a caller that owns the backoff under +// a lock takes the interval there and sleeps it after releasing the lock. +func (b *connectBackoff) step() time.Duration { d := b.next if d <= 0 { d = reconnectBackoffMin } b.next = min(d*2, reconnectBackoffMax) + return d/2 + rand.N(d/2+1) // uniform in [d/2, d] +} - d = d/2 + rand.N(d/2+1) // uniform in [d/2, d] +// sleepFor blocks for d. It returns early — reporting false — when ctx is +// cancelled or abort is closed. A nil abort channel is valid and never fires. +func sleepFor(ctx context.Context, abort <-chan struct{}, d time.Duration) bool { t := time.NewTimer(d) defer t.Stop() select { diff --git a/h2client.go b/h2client.go index 653a4ae..0023094 100644 --- a/h2client.go +++ b/h2client.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "crypto/tls" + "errors" "fmt" "net" "strconv" @@ -99,7 +100,7 @@ type h2Client struct { // would otherwise strand the slot dead and the worker would spin // closed-conn errors with no recovery (fastapi-h2 logged ~1.1B errors / // 0 requests from exactly this). reconnectSlot calls it under the slot - // lock, paced by the slot's backoff. + // lock, paced by the slot's backoff while the server looks down. redial func() (*h2Conn, error) // dialedViaUpgrade reports whether this client's connections were @@ -156,12 +157,16 @@ type h2Conn struct { writeCh chan h2WriteReq // Stream management — lock-free fixed-size slot array. - // Index = (streamID >> 1) % len(streamSlots). The semaphore limits - // concurrent streams, and stream IDs are assigned sequentially by - // writeLoop, so slots cycle predictably with no collisions. - // Sized at 2x effectiveStreams for headroom against wrap-around. + // Index = (streamID >> 1) % len(streamSlots), sized at 2x + // effectiveStreams. The semaphore limits concurrent streams, but a slow + // or abandoned stream can still hold its slot when the IDs come round + // to it again, so writeLoop skips an ID whose slot is held: no two + // pending streams share a slot. nextStreamID atomic.Uint32 streamSlots []h2StreamSlot + // firstStreamID is the connection's first request stream: 1, or 3 on an + // h2c-upgraded connection, whose stream 1 is the upgrade request itself. + firstStreamID uint32 // Channel pool — pools *chan h2Response (heap-allocated pointers). // The pointer in the slot remains valid even after the goroutine exits. @@ -184,26 +189,70 @@ type h2Conn struct { // currently-writing stream's window lives in curStreamWindow keyed by // curStreamID; readLoop replenishes both on WINDOW_UPDATE. Without this a // 64KiB body (post-64k-h2 = 65536 B) exceeds the 65535 window by one byte - // and the request hangs to the 5-min deadline. + // and the request hangs until the run ends. curStreamReset is the ID of + // that stream once the server has reset it (RST_STREAM): the body writer + // then stops before its next DATA frame, whatever window is left (RFC + // 9113 §6.4), and sends no RST_STREAM of its own (§5.4.2). connSendWindow atomic.Int64 curStreamID atomic.Uint32 curStreamWindow atomic.Int64 + curStreamReset atomic.Uint32 serverInitWindow uint32 // Concurrency limit streamSem chan struct{} - // Shutdown signal — closed by closeConn to unblock readLoop/writeLoop/workers - done chan struct{} - - addr string + // Shutdown signal — closed by closeConn (once, under closeOnce) to unblock + // readLoop/writeLoop/workers, whether the client closes the connection or + // it dies under the client (failConn, from readLoop or writeLoop). + done chan struct{} + closeOnce sync.Once + + // loopsDone is closed once readLoop and writeLoop have both returned + // (loopsLeft counts them down), which they do soon after done closes: + // closeConn closes the socket too. Each has sent by then every answer it + // will ever send, so await waits for it before it decides that a request + // on a closed connection got no answer. + loopsDone chan struct{} + loopsLeft atomic.Int32 + + // queueMu makes taking a request off writeCh and answering it one step + // for answerQueued's callers, the writer's last drain and every await on + // a closed connection: once an await's own answerQueued returns, any + // request someone else took off the queue has been answered, its own + // among them, so it can decide that it got no answer. + queueMu sync.Mutex + + addr string + // closed marks a connection that takes no new request: DoRequest + // redials the slot instead. Set before done is closed, and on its own + // when the connection's stream IDs run out (its last streams are still + // in flight then): closing, not closed, says the client closed it. closed atomic.Bool + // served is set by readLoop once the connection carries a response + // frame for one of the run's requests (not the h2c upgrade's own stream + // 1). A connection that ended without one shows the server down or + // dropping every connection: reconnectSlot fails the request that finds + // it, after the slot's backoff. + served atomic.Bool } // h2StreamSlot holds an atomic pointer to a response channel. // writeLoop stores; readLoop loads and clears. +// +// The other fields are the response state of the stream that last sent a +// frame for this slot. Only readLoop touches them, so they need no +// synchronisation, and they are keyed by stream ID: the first frame of a new +// stream resets them (see h2Conn.stream), so nothing leaks from a stream +// that used the slot before, whether it finished, was reset or was +// abandoned. type h2StreamSlot struct { ch atomic.Pointer[chan h2Response] + + streamID uint32 + status int // :status of the response; 0 when absent or not parseable + final bool // the final (non-1xx) response HEADERS arrived; a later HEADERS frame carries trailers + bytes int // body bytes: the data of every DATA frame so far, padding excluded } // h2Response is the result dispatched from the read loop to a waiting worker. @@ -225,7 +274,8 @@ var h2StaticStatus = [7]int{200, 204, 206, 304, 400, 404, 500} // extractStatus extracts the HTTP status code from an HPACK-encoded header block // without allocating. Handles indexed representations from the HPACK static table // (covers >99% of benchmark responses) and falls back to literal parsing. -// Returns 0 for unparseable blocks (treated as success in benchmarks). +// Returns 0 for a block without a :status it can read, which DoRequest counts +// as an error (errH2NoStatus), never as a success. func extractStatus(headerBlock []byte) int { if len(headerBlock) == 0 { return 0 @@ -263,8 +313,11 @@ func extractStatus(headerBlock []byte) int { return h2StaticStatus[b-0x88] } - // Fallback: literal header field with name index 8 (:status). - // Safe because we disabled the HPACK dynamic table (SettingHeaderTableSize=0). + // Fallback: literal header field whose name is a static :status entry. + // Static indices 8-14 all name :status, and encoders pick any of them: + // x/net's (net/http's) names it by the LAST match, 14, so a literal 503 + // arrives as 0x0e 0x03 "503". Safe because we disabled the HPACK dynamic + // table (SettingHeaderTableSize=0). var nameIdx int pos++ @@ -279,26 +332,70 @@ func extractStatus(headerBlock []byte) int { return 0 } - if nameIdx != 8 || pos >= len(headerBlock) { + if nameIdx < 8 || nameIdx > 14 { + // Name index 0: the name is a string literal (RFC 7541 §6.2), legal + // if unusual for a name the static table holds. Accept ":status", + // plain or Huffman-coded, and go on as for a static :status name. + if nameIdx != 0 || pos >= len(headerBlock) { + return 0 + } + nameByte := headerBlock[pos] + pos++ + nameLen := int(nameByte & 0x7F) + if pos+nameLen > len(headerBlock) { + return 0 + } + name := headerBlock[pos : pos+nameLen] + pos += nameLen + if nameByte&0x80 != 0 { + if string(name) != h2HuffmanStatusName { + return 0 + } + } else if string(name) != ":status" { + return 0 + } + } + if pos >= len(headerBlock) { return 0 } valueByte := headerBlock[pos] pos++ - // Skip Huffman-encoded values (unlikely for 3-digit status codes) - if valueByte&0x80 != 0 { + valueLen := int(valueByte & 0x7F) + if pos+valueLen > len(headerBlock) { return 0 } + value := headerBlock[pos : pos+valueLen] - valueLen := int(valueByte & 0x7F) - if valueLen != 3 || pos+3 > len(headerBlock) { - return 0 + // A Huffman-coded value (RFC 7541 §5.2). Encoders pick it whenever it is + // shorter than the literal, which for a status is every code with at + // least two digits from {0, 1, 2}: 401, 402, 410, 412, 502, 103, 201, + // 301 and more (403 and 503 are not). net/http's server sends them so. + if valueByte&0x80 != 0 { + return h2HuffmanStatus[string(value)] // 0 when it is not a 3-digit status } - return parseStatusCode(headerBlock[pos : pos+3]) + if valueLen != 3 { + return 0 + } + return parseStatusCode(value) } +// h2HuffmanStatusName is the HPACK Huffman encoding of the name ":status". +var h2HuffmanStatusName = string(hpack.AppendHuffmanString(nil, ":status")) + +// h2HuffmanStatus maps the HPACK Huffman encoding of every status code +// 100-599 to the code, so extractStatus decodes a Huffman-coded :status with +// one map lookup (string(b) in a map index does not allocate). +var h2HuffmanStatus = func() map[string]int { + m := make(map[string]int, 500) + for status := 100; status < 600; status++ { + m[string(hpack.AppendHuffmanString(nil, strconv.Itoa(status)))] = status + } + return m +}() + // newH2Client creates a new zero-alloc HTTP/2 client. func newH2Client(host, port, path string, cfg Config) (*h2Client, error) { return newH2ClientWithDialer(host, port, path, cfg, false) @@ -382,34 +479,72 @@ func newH2ClientWithDialer(host, port, path string, cfg Config, upgrade bool) (* }, nil } -// reconnectSlot re-establishes a dead slot's connection, single-flighted under -// the slot lock and paced by the slot's backoff. Returns the live conn, or nil -// if the redial failed (caller surfaces a connect error) or ctx ended. A -// concurrent caller that finds the slot already healed returns immediately. -func (c *h2Client) reconnectSlot(ctx context.Context, slot *h2ConnSlot) *h2Conn { +// reconnectSlot replaces a slot's dead connection, single-flighted under the +// slot lock: a caller that finds the slot already healed takes the new +// connection. When the server looks down, it fails the request instead, as +// h1client fails a request whose redial fails: the dial failed (a connect +// error), or the connection being replaced ended before it answered a single +// request (errH2NeverServed: the server completes handshakes and drops them, +// and nothing is dialed for this request). That request then sleeps the +// slot's backoff after it releases the lock, so the sleep paces the server, +// lands in no recorded latency (a failed request records none) and holds up +// none of the slot's other workers; the next request dials at once. The +// backoff doubles while the server keeps looking down and starts over once a +// connection has served. After the run's context ends it dials nothing. +func (c *h2Client) reconnectSlot(ctx context.Context, slot *h2ConnSlot) (*h2Conn, error) { slot.mu.Lock() - defer slot.mu.Unlock() - - if hc := slot.cur.Load(); hc != nil && !hc.closed.Load() { - return hc // another worker already re-dialed this slot + old := slot.cur.Load() + if old != nil && !old.closed.Load() { + slot.mu.Unlock() + return old, nil // another worker already re-dialed this slot } - if old := slot.cur.Load(); old != nil { - old.closeConn() // release fds/goroutines of the dead conn + if ctx.Err() != nil { + slot.mu.Unlock() + return nil, ctx.Err() // Close, which runs after the cancel, would never close a connection dialed now } - // Pace retries so a server that stays down can't be hot-redialled. - if !slot.backoff.sleep(ctx, nil) { - return nil // ctx cancelled mid-backoff + var err error + if old != nil { + old.closeConn() // release fds/goroutines of the dead conn + // served is final once readLoop has returned: when the writer ended + // the connection, readLoop may still be parsing frames it had read, + // the connection's first response among them. Both loops return soon + // after the close. + select { + case <-old.loopsDone: + case <-ctx.Done(): + } + // Checked after the wait, not in its ctx case only: the run may have + // ended with the loops already returned, and select then takes + // either case. + if ctx.Err() != nil { + slot.mu.Unlock() + return nil, ctx.Err() + } + if old.served.Load() { + slot.backoff.reset() // the server was up: redial at once, and the pace starts over + } else { + err = errH2NeverServed + } } - hc, err := c.redial() - if err != nil { + if err == nil { + hc, dialErr := c.redial() + if dialErr == nil { + slot.cur.Store(hc) + slot.mu.Unlock() + return hc, nil + } recordConnectError() - slot.cur.Store(nil) - return nil + err = dialErr } - slot.backoff.reset() - slot.cur.Store(hc) - return hc + + // The server looks down. The next request dials at once; this one fails + // after the backoff, which it sleeps outside the lock. + slot.cur.Store(nil) + pause := slot.backoff.step() + slot.mu.Unlock() + sleepFor(ctx, nil, pause) + return nil, err } // h2HopByHopHeaders lists HTTP/1.1 connection-specific headers that are @@ -611,12 +746,14 @@ func completeH2Handshake(conn net.Conn, br *bufio.Reader, addr string, maxStream } } + // The handshake is done: from here the connection has no deadline, as an + // HTTP/1.1 connection has none. It lives until the server or the network + // ends it or the client closes it, and Close is what interrupts a read or + // a write blocked on a peer that stopped reading. An absolute deadline + // here ended every connection of a run at the same instant, however + // healthy (#88). _ = conn.SetReadDeadline(time.Time{}) - // Set a deadline covering the entire benchmark lifetime. - // Prevents infinite blocks on TCP write if H2 flow control deadlocks. - _ = conn.SetDeadline(time.Now().Add(5 * time.Minute)) - effectiveStreams := min(serverMaxStreams, uint32(maxStreams)) if effectiveStreams < 1 { effectiveStreams = 100 @@ -627,16 +764,18 @@ func completeH2Handshake(conn net.Conn, br *bufio.Reader, addr string, maxStream numSlots := 2 * effectiveStreams hc := &h2Conn{ - conn: conn, - framer: framer, - bufWriter: bw, - writeCh: make(chan h2WriteReq, 4096), - streamSlots: make([]h2StreamSlot, numSlots), + conn: conn, + framer: framer, + bufWriter: bw, + writeCh: make(chan h2WriteReq, 4096), + streamSlots: make([]h2StreamSlot, numSlots), + firstStreamID: initialStreamID, // DATA-send split = the SERVER's max frame size (not our 64KiB receive cap). maxFrameSize: serverMaxFrame, serverInitWindow: serverInitWin, streamSem: make(chan struct{}, effectiveStreams), done: make(chan struct{}), + loopsDone: make(chan struct{}), addr: addr, chanPool: sync.Pool{ New: func() any { @@ -654,6 +793,7 @@ func completeH2Handshake(conn net.Conn, br *bufio.Reader, addr string, maxStream hc.streamSem <- struct{}{} } + hc.loopsLeft.Store(2) go hc.writeLoop() go hc.readLoop() @@ -666,6 +806,7 @@ func completeH2Handshake(conn net.Conn, br *bufio.Reader, addr string, maxStream // It flushes pending connection-level WINDOW_UPDATE (accumulated by readLoop // via atomic counter) both when processing requests AND periodically when idle. func (hc *h2Conn) writeLoop() { + defer hc.loopExited() ticker := time.NewTicker(1 * time.Millisecond) defer ticker.Stop() @@ -684,35 +825,79 @@ func (hc *h2Conn) writeLoop() { count++ if count%64 == 0 { hc.flushWindowUpdate() - _ = hc.bufWriter.Flush() // send WINDOW_UPDATE to the network NOW + hc.writeFailed(hc.bufWriter.Flush()) // send WINDOW_UPDATE to the network NOW } default: break drain } } hc.flushWindowUpdate() - _ = hc.bufWriter.Flush() + hc.writeFailed(hc.bufWriter.Flush()) case <-ticker.C: hc.flushWindowUpdate() - _ = hc.bufWriter.Flush() + hc.writeFailed(hc.bufWriter.Flush()) case <-hc.done: - // Drain pending requests — send errors to respCh so workers unblock. - for { - select { - case req := <-hc.writeCh: - if req.respCh != nil { - *req.respCh <- h2Response{err: fmt.Errorf("connection closing")} - } - default: - return - } + hc.answerQueued() + return + } + } +} + +// answerQueued answers every request still queued for a closed connection's +// writer. None of them was written, so none reached the server: each goes +// to the redialed connection (errH2NotSent), as a request still waiting for +// a stream does, instead of counting as an error. Callers take turns +// (queueMu), so each request is off the queue only once it is answered. +func (hc *h2Conn) answerQueued() { + hc.queueMu.Lock() + defer hc.queueMu.Unlock() + for { + select { + case req := <-hc.writeCh: + if req.respCh != nil { + *req.respCh <- h2Response{err: errH2NotSent} } + default: + return } } } +// loopExited is deferred by readLoop and writeLoop: the second to return +// closes loopsDone. +func (hc *h2Conn) loopExited() { + if hc.loopsLeft.Add(-1) == 0 { + close(hc.loopsDone) + } +} + +// writeFailed ends the connection after a write to it failed (nil: no-op). +// The peer is gone, and readLoop does not always see it: the server may keep +// its side open, or its reset may not have reached the read yet. Without +// this the bufio.Writer keeps the error and fails every later request on the +// dead connection at once, a spin of errors that never redials (#89). +func (hc *h2Conn) writeFailed(err error) { + if err != nil && !hc.closing() { + hc.failConn(fmt.Errorf("connection error: %w", err)) + } +} + +// closing reports whether closeConn has run: the client closed the +// connection (Close, reconnectSlot), or failConn is ending it. closed alone +// does not say so: a connection whose stream IDs ran out is closed to new +// requests while its last streams are still in flight, and if it then dies, +// readLoop and writeFailed must fail it like any other. +func (hc *h2Conn) closing() bool { + select { + case <-hc.done: + return true + default: + return false + } +} + func (hc *h2Conn) flushWindowUpdate() { if pending := hc.pendingConnWindow.Swap(0); pending > 0 { _ = hc.framer.WriteWindowUpdate(0, pending) @@ -722,27 +907,75 @@ func (hc *h2Conn) flushWindowUpdate() { func (hc *h2Conn) processWriteReq(req h2WriteReq) { switch req.kind { case h2WriteHeaders: + // A connection that takes no new request (it died, the client closed + // it, or its stream IDs ran out) never writes this one: it goes to + // the redialed connection, as a request still waiting for a stream + // does. Written, it would fail on bufio's sticky error, or wait for a + // response nobody reads, and count as an error. + if hc.closed.Load() { + if req.respCh != nil { + *req.respCh <- h2Response{err: errH2NotSent} + } + return + } + + // Take the next stream ID whose slot is free. A slot still holds the + // channel of a stream that 2*MaxStreams-1 later streams have outlived + // (a slow response, or a request its worker abandoned); reusing it + // would hand that stream's response to this request and leave its + // worker waiting for the rest of the run. IDs may skip: a new + // stream's ID need only exceed every earlier one (RFC 9113 §5.1.1). + // At most MaxStreams slots are held by streams with a token, so a + // free slot comes within MaxStreams+1 IDs; the bound keeps a slot + // array held throughout (abandoned streams) from looping. + numSlots := uint32(len(hc.streamSlots)) streamID := hc.nextStreamID.Add(2) - 2 + for range numSlots { + if hc.streamSlots[(streamID>>1)%numSlots].ch.Load() == nil { + break + } + streamID = hc.nextStreamID.Add(2) - 2 + } if streamID > 0x7FFFFFFF { + // The connection's stream IDs are used up, which a connection + // that lives as long as the server keeps it (#88) reaches after + // 2^30 requests. RFC 9113 §5.1.1: open a new connection. Marked + // closed, it takes no new request, and DoRequest redials it and + // closes it; this request never reached the server, so it goes + // to the new connection (errH2NotSent), as does every request + // still queued here. + hc.closed.Store(true) if req.respCh != nil { - *req.respCh <- h2Response{err: fmt.Errorf("h2client: stream ID exhausted")} + *req.respCh <- h2Response{err: errH2NotSent} } return } - slotIdx := (streamID >> 1) % uint32(len(hc.streamSlots)) + slotIdx := (streamID >> 1) % numSlots hc.streamSlots[slotIdx].ch.Store(req.respCh) + if req.hasBody { + // Key the send window to this stream before its HEADERS can reach + // the server, which may grant window for it or reset it as soon + // as they do. + hc.curStreamWindow.Store(int64(hc.serverInitWindow)) + hc.curStreamID.Store(streamID) + } err := hc.framer.WriteHeaders(streamID, req.block, !req.hasBody) if err == nil && req.hasBody { - err = hc.writeBodyFlowControlled(streamID, req.data) - } - if err != nil { - hc.streamSlots[slotIdx].ch.Store(nil) - if req.respCh != nil { - *req.respCh <- h2Response{err: err} + err = hc.writeBodyFlowControlled(streamID, &hc.streamSlots[slotIdx], req.respCh, req.data) + if err == errH2BodyAbandoned { + return // the server ended the stream, and its worker has the answer; the connection is fine } } + // Whoever takes respCh out of the slot answers it, exactly once: + // readLoop may have taken it first (a response, a reset, or + // failStreams on a connection that died while this was written). + // Answering it twice would block here on the full channel. + if err != nil && req.respCh != nil && hc.streamSlots[slotIdx].ch.CompareAndSwap(req.respCh, nil) { + *req.respCh <- h2Response{err: err} + } + hc.writeFailed(err) case h2WriteSettingsAck: _ = hc.framer.WriteSettingsAck() @@ -760,54 +993,84 @@ func (hc *h2Conn) processWriteReq(req h2WriteReq) { // is full of worker requests — readLoop blocks, can't read responses, server's // flow control windows exhaust, everything stalls. func (hc *h2Conn) readLoop() { + defer hc.loopExited() numSlots := uint32(len(hc.streamSlots)) for { frame, err := hc.framer.ReadFrame() if err != nil { - if hc.closed.Load() { - return - } - // Notify all pending streams - for i := range hc.streamSlots { - chPtr := hc.streamSlots[i].ch.Swap(nil) - if chPtr != nil { - *chPtr <- h2Response{err: fmt.Errorf("connection error: %w", err)} - } + if hc.closing() { + return // the client closed it (Close, or reconnectSlot replacing it), or failConn is ending it } + // The server or the network ended the connection without a + // GOAWAY: a close, a reset, a timeout. Fail it, so its in-flight + // streams are errors and DoRequest redials. + hc.failConn(fmt.Errorf("connection error: %w", err)) + return + } + + // A HEADERS or DATA frame whose padding does not fit it is a + // connection error (RFC 9113 §6.1, §6.2): nothing in it can be read, + // so the connection ends as a dead one does. + if frame.badPadding() { + hc.failConn(fmt.Errorf("h2client: PROTOCOL_ERROR: frame type %d on stream %d: its Pad Length does not fit its %d-byte payload", + frame.Type, frame.StreamID, frame.Length)) + return + } + // So is a HEADERS frame too short for the priority fields its + // PRIORITY flag announces: a frame size error on a frame that + // carries a field block (RFC 9113 §4.2). + if frame.badPriority() { + hc.failConn(fmt.Errorf("h2client: FRAME_SIZE_ERROR: HEADERS on stream %d: its %d-byte payload has no room for its priority fields", + frame.StreamID, frame.Length)) return } switch frame.Type { case frameHeaders: - status := extractStatus(frame.HeaderBlockFragment()) + // The status is the final response's :status, whatever ends the + // stream: this HEADERS frame, a DATA frame, or trailers. An + // interim 1xx response is replaced by the final one; a HEADERS + // frame after the final one carries trailers, not a status. + st := hc.stream(frame.StreamID) + if !st.final { + st.status = extractStatus(frame.HeaderBlockFragment()) + st.final = st.status < 100 || st.status >= 200 + } if frame.StreamEnded() { - idx := (frame.StreamID >> 1) % numSlots - chPtr := hc.streamSlots[idx].ch.Swap(nil) + chPtr := st.ch.Swap(nil) if chPtr != nil { - *chPtr <- h2Response{status: status, bytesRead: 0} + *chPtr <- h2Response{status: st.status, bytesRead: st.bytes} } } case frameData: - bytesRead := len(frame.Data()) - // Accumulate connection-level WINDOW_UPDATE via atomic counter. // writeLoop flushes this between processing worker requests. - if bytesRead > 0 { - hc.pendingConnWindow.Add(uint32(bytesRead)) + // Flow control counts the whole payload, padding included. + if frame.Length > 0 { + hc.pendingConnWindow.Add(frame.Length) } + // The body is every DATA frame of the stream, not the one that + // ends it: a body split over 16 KiB frames, or followed by an + // empty END_STREAM frame (net/http, after a flush), counts in full. + // Padding is framing, as chunk sizes are in HTTP/1.1: not counted. + st := hc.stream(frame.StreamID) + st.bytes += len(frame.Data()) + if frame.StreamEnded() { - idx := (frame.StreamID >> 1) % numSlots - chPtr := hc.streamSlots[idx].ch.Swap(nil) + chPtr := st.ch.Swap(nil) if chPtr != nil { - *chPtr <- h2Response{status: 200, bytesRead: bytesRead} + *chPtr <- h2Response{status: st.status, bytesRead: st.bytes} } } case frameRSTStream: + if frame.StreamID == hc.curStreamID.Load() { + hc.curStreamReset.Store(frame.StreamID) // before the answer: the body writer reads them in the other order + } idx := (frame.StreamID >> 1) % numSlots chPtr := hc.streamSlots[idx].ch.Swap(nil) if chPtr != nil { @@ -833,13 +1096,7 @@ func (hc *h2Conn) readLoop() { } case frameGoAway: - hc.closed.Store(true) - for i := range hc.streamSlots { - chPtr := hc.streamSlots[i].ch.Swap(nil) - if chPtr != nil { - *chPtr <- h2Response{err: fmt.Errorf("goaway: code=%d", frame.GoAwayErrCode())} - } - } + hc.failConn(fmt.Errorf("goaway: code=%d", frame.GoAwayErrCode())) return case frameWindowUpdate: @@ -857,33 +1114,104 @@ func (hc *h2Conn) readLoop() { } } +// stream returns the slot of streamID with its response state reset if the +// slot last held another stream, and marks the connection served when the +// stream is one of the run's requests. The h2c upgrade's own stream 1 is not: +// the server answers it on every upgraded connection (celeris before it even +// reads the client preface), so it shows nothing about whether the server +// serves the run. readLoop only. +func (hc *h2Conn) stream(streamID uint32) *h2StreamSlot { + st := &hc.streamSlots[(streamID>>1)%uint32(len(hc.streamSlots))] + if st.streamID != streamID { + st.streamID = streamID + st.status = 0 + st.final = false + st.bytes = 0 + if streamID >= hc.firstStreamID && !hc.served.Load() { + hc.served.Store(true) + } + } + return st +} + +// errH2NeverServed is DoRequest's error for a request that found its +// connection ended before the connection answered any request: the server +// completes handshakes and drops the connections, or is shutting down. The +// server looks down, so the request fails (see reconnectSlot). +var errH2NeverServed = errors.New("h2client: the connection ended before it answered a request") + +// errH2NoStatus is DoRequest's error for a response without a final :status +// the client can read: no :status at all, none that extractStatus decodes, or +// only an interim 1xx. It is not a success, as h1client fails a response +// without a status line. +var errH2NoStatus = errors.New("h2client: response without a final :status") + +// errH2NotSent reports a request that never reached the server: the +// connection died before the request was handed to it (roundTrip), or took +// no new request (it died, was closed, or ran out of stream IDs) before its +// writer wrote it (processWriteReq, answerQueued). It is not a failed +// request: DoRequest takes the request to the redialed connection. Never +// returned to DoRequest's caller. +var errH2NotSent = errors.New("h2client: connection gone before the request was sent") + // DoRequest sends an HTTP/2 request and waits for the response. // Fire-and-forget to writeLoop: no resultCh round-trip. Workers wait only on respCh, // which receives from either writeLoop (on error) or readLoop (on response). func (c *h2Client) DoRequest(ctx context.Context, workerID int) (int, error) { idx := workerID % len(c.conns) slot := c.conns[idx] - hc := slot.cur.Load() - - // A server that closed/GOAWAYed this connection mid-cell leaves the slot - // dead; re-dial (paced) so the next request gets a fresh conn instead of - // spinning closed-conn errors. backoff lives in reconnectSlot. - if hc == nil || hc.closed.Load() { - hc = c.reconnectSlot(ctx, slot) - if hc == nil { - if ctx.Err() != nil { - return 0, ctx.Err() + for { + hc := slot.cur.Load() + + // A server that closed, reset or GOAWAYed this connection mid-cell + // leaves the slot dead (failConn); re-dial so the request gets a + // fresh conn instead of spinning closed-conn errors. While the + // server looks down, reconnectSlot fails the request instead, paced + // by the slot's backoff. + if hc == nil || hc.closed.Load() { + var err error + if hc, err = c.reconnectSlot(ctx, slot); err != nil { + switch { + case ctx.Err() != nil: + return 0, ctx.Err() + case err == errH2NeverServed: + return 0, fmt.Errorf("h2client: conn[%d]: %w", idx, err) + } + return 0, fmt.Errorf("h2client: conn[%d] reconnect failed: %w", idx, err) } - return 0, fmt.Errorf("h2client: conn[%d] reconnect failed", idx) } + + n, err := c.roundTrip(ctx, hc, idx) + if err != errH2NotSent { // the sentinel itself, never wrapped + return n, err + } + // The connection ended before this request was written: while it + // waited for a stream, or before the connection's writer reached it + // (the connection died, or ran out of stream IDs). Each pass through + // here needs a connection to have ended, and reconnectSlot redials + // only a connection that answered a request: one that never did + // fails this request, so a server that keeps dropping connections + // cannot make this loop spin uncounted. ctx ends it too. } +} - // Acquire stream semaphore +// roundTrip sends one request on hc and waits for its response. It returns +// errH2NotSent if hc ended before the request was written: before it was +// handed to the writer, or before the writer reached it. +func (c *h2Client) roundTrip(ctx context.Context, hc *h2Conn, idx int) (int, error) { + // Acquire a stream. When the connection dies, the workers queued here + // wake through done, or through the tokens its failed streams return. select { case <-hc.streamSem: + case <-hc.done: + return 0, errH2NotSent case <-ctx.Done(): return 0, ctx.Err() } + if hc.closed.Load() { + hc.streamSem <- struct{}{} + return 0, errH2NotSent + } // Get heap-allocated response channel pointer from pool chPtr := hc.chanPool.Get().(*chan h2Response) @@ -907,25 +1235,25 @@ func (c *h2Client) DoRequest(ctx context.Context, workerID int) (int, error) { case <-hc.done: hc.chanPool.Put(chPtr) hc.streamSem <- struct{}{} - return 0, fmt.Errorf("h2client: conn[%d] connection closing", idx) + return 0, errH2NotSent case <-ctx.Done(): hc.chanPool.Put(chPtr) hc.streamSem <- struct{}{} return 0, ctx.Err() } - // Single wait — receives normal responses (from readLoop) AND write errors (from writeLoop) + return hc.await(ctx, chPtr, idx) +} + +// await waits for the answer to a request handed to hc: a response (from +// readLoop), an error (from writeLoop, or failStreams), or errH2NotSent (from +// writeLoop, for a request it never wrote). Exactly one of them answers each +// request, and done below only when none did, so a request is one outcome, +// never zero or two. +func (hc *h2Conn) await(ctx context.Context, chPtr *chan h2Response, idx int) (int, error) { select { case resp := <-*chPtr: - hc.chanPool.Put(chPtr) - hc.streamSem <- struct{}{} - if resp.err != nil { - return 0, resp.err - } - if resp.status >= 400 { - return resp.bytesRead, statusError(resp.status) - } - return resp.bytesRead, nil + return hc.finish(chPtr, resp) case <-ctx.Done(): // Worker exits, abandons chPtr. The heap-allocated channel (~120 bytes) // will be GC'd when the slot is overwritten by the next stream ID at @@ -934,26 +1262,79 @@ func (c *h2Client) DoRequest(ctx context.Context, workerID int) (int, error) { hc.streamSem <- struct{}{} return 0, ctx.Err() case <-hc.done: + // The connection is closing, and this request's answer may still + // be on its way: readLoop may be delivering a response it took + // before the close, and writeLoop answers every request it never + // wrote with errH2NotSent. Both loops return soon after done closes, + // having sent every answer they will send, so wait for them (or the + // answer), then take the answer if there is one. Deciding at once + // turned both into errors: a response the server sent, and a request + // that never reached it. + select { + case resp := <-*chPtr: + return hc.finish(chPtr, resp) + case <-hc.loopsDone: + case <-ctx.Done(): + hc.streamSem <- struct{}{} + return 0, ctx.Err() + } + // writeLoop has returned, so whatever is still queued for it was + // never written: roundTrip's hand-off can win its race with done + // after the writer's final drain. Answer those requests too, this + // one among them (each is taken off the queue once, so answered + // once). Another await may have taken this one: answerQueued waits + // for it to answer, so the receive below cannot miss that answer. + hc.answerQueued() + select { + case resp := <-*chPtr: + return hc.finish(chPtr, resp) + default: + } hc.streamSem <- struct{}{} return 0, fmt.Errorf("h2client: conn[%d] connection closing", idx) } } +// finish returns the stream's channel and token and turns its answer into +// DoRequest's result: a status >= 400 is an error, as h1client counts it, and +// so is a response with no final :status: none that could be read, or only an +// interim 1xx, which RFC 9113 §8.1 makes a malformed response. +func (hc *h2Conn) finish(chPtr *chan h2Response, resp h2Response) (int, error) { + hc.chanPool.Put(chPtr) + hc.streamSem <- struct{}{} + if resp.err != nil { + return 0, resp.err + } + if resp.status < 200 { + return resp.bytesRead, errH2NoStatus + } + if resp.status >= 400 { + return resp.bytesRead, statusError(resp.status) + } + return resp.bytesRead, nil +} + +// errH2BodyAbandoned is writeBodyFlowControlled's report of a body it stopped +// because the server ended the stream first. Not an error: the stream's worker +// already has its answer, and the connection is fine. +var errH2BodyAbandoned = errors.New("h2client: request body abandoned: the server ended the stream") + // writeBodyFlowControlled sends a request body as DATA frames, respecting BOTH // the server's SETTINGS_MAX_FRAME_SIZE and its connection + per-stream send // windows (RFC 7540 §6.9). The writer goroutine is sequential (one body at a // time), so the active stream's window lives in hc.curStreamWindow keyed by -// hc.curStreamID; readLoop replenishes connSendWindow/curStreamWindow on -// WINDOW_UPDATE. When a window is exhausted we flush the buffered frames (so the -// server can consume + replenish) and poll — no extra goroutine/channel. The -// conn's 5-min deadline (completeH2Handshake) is the deadlock backstop. +// hc.curStreamID (processWriteReq keys them); readLoop replenishes +// connSendWindow/curStreamWindow on WINDOW_UPDATE. When a window is exhausted +// we flush the buffered frames (so the server can consume + replenish) and +// poll — no extra goroutine/channel. The wait ends when that flush fails, when +// the connection is closed (done), which Close does at the end of a run and +// failConn when the connection dies, or when the server has ended the stream +// (st no longer holds respCh): the backstops against a peer that never grants +// window. A stream the server has reset gets no further DATA frame at all. // // Without this, a 64KiB body (post-64k-h2 = 65536 B) either trips FRAME_SIZE_ERROR // (64KiB DATA frame vs a 16384-default server) or overruns the 65535 window. -func (hc *h2Conn) writeBodyFlowControlled(streamID uint32, data []byte) error { - hc.curStreamWindow.Store(int64(hc.serverInitWindow)) - hc.curStreamID.Store(streamID) - +func (hc *h2Conn) writeBodyFlowControlled(streamID uint32, st *h2StreamSlot, respCh *chan h2Response, data []byte) error { maxFrame := int(hc.maxFrameSize) if maxFrame < 16384 { maxFrame = 16384 @@ -967,17 +1348,44 @@ func (hc *h2Conn) writeBodyFlowControlled(streamID uint32, data []byte) error { avail = sw } if avail <= 0 { - // Window exhausted: flush so the peer receives what we've sent and - // can send WINDOW_UPDATE, then wait for readLoop to replenish. - _ = hc.bufWriter.Flush() select { case <-hc.done: return fmt.Errorf("h2client: conn closed mid-body (flow-control wait)") default: } + // The server has answered or reset the stream (readLoop took its + // channel out of the slot): RFC 9113 §8.1 lets it answer before + // the body is complete. It grants no more window for a stream it + // has ended, so stop the body here instead of waiting for the run + // to end with every later request of the connection queued + // behind this one. A response that ended without a reset leaves + // our half of the stream open: close it with RST_STREAM(CANCEL). + if respCh != nil && st.ch.Load() != respCh { + if hc.curStreamReset.Load() != streamID { + _ = hc.framer.WriteRSTStream(streamID, h2ErrCodeCancel) + } + return errH2BodyAbandoned + } + // Window exhausted: flush so the peer receives what we've sent and + // can send WINDOW_UPDATE, then wait for readLoop to replenish. + // The flush carries the receive window readLoop has credited + // meanwhile, as writeLoop's own flushes do: a peer that waits for + // that credit before it grants ours would otherwise deadlock with + // this wait. A failed flush ends the wait and, through the + // caller's writeFailed, the connection (#89). + hc.flushWindowUpdate() + if err := hc.bufWriter.Flush(); err != nil { + return err + } time.Sleep(50 * time.Microsecond) continue } + // The server has reset the stream: send no more frames on it (RFC + // 9113 §6.4), however much window is left. readLoop records the + // reset before it answers the stream's worker. + if hc.curStreamReset.Load() == streamID { + return errH2BodyAbandoned + } if avail > len(data) { avail = len(data) } @@ -1004,10 +1412,36 @@ func (c *h2Client) Close() { } } +// closeConn closes the connection: it takes no new request, done unblocks +// workers, readLoop and writeLoop, and the socket is released. Idempotent, +// and safe from any goroutine. func (hc *h2Conn) closeConn() { - if !hc.closed.CompareAndSwap(false, true) { - return // already closed + hc.closed.Store(true) + hc.closeOnce.Do(func() { + close(hc.done) // unblock workers, readLoop, and writeLoop + _ = hc.conn.Close() + }) +} + +// failConn ends a connection that died under the client: the server closed +// or reset it, sent GOAWAY, or a write to it failed. It marks the connection +// closed first, so DoRequest redials instead of queueing onto a connection +// nobody reads, fails every stream in flight on it once, and closes it, +// which stops writeLoop, releases the socket and wakes the workers queued +// for a stream. Called by readLoop and writeLoop; safe from both at once. +func (hc *h2Conn) failConn(err error) { + hc.closed.Store(true) + hc.failStreams(err) + hc.closeConn() +} + +// failStreams answers every stream still registered on the connection with +// err. Each response channel is taken out of its slot before it is answered, +// so no stream is answered twice. +func (hc *h2Conn) failStreams(err error) { + for i := range hc.streamSlots { + if chPtr := hc.streamSlots[i].ch.Swap(nil); chPtr != nil { + *chPtr <- h2Response{err: err} + } } - close(hc.done) // unblock workers, readLoop, and writeLoop - _ = hc.conn.Close() } diff --git a/h2client_accounting_test.go b/h2client_accounting_test.go new file mode 100644 index 0000000..99c6257 --- /dev/null +++ b/h2client_accounting_test.go @@ -0,0 +1,2468 @@ +package loadgen + +// Regression tests for what the HTTP/2 client records per response and how +// long its connections live: +// +// - #85: a 4xx/5xx response WITH a body was recorded as status 200 (a +// success); the status must come from the response HEADERS frame. +// - #86: throughput counted only the last DATA frame of each response. +// - #88: every connection carried an absolute 5-minute deadline from dial. +// - #89: a connection that died with a read error (no GOAWAY) was never +// redialed, and its workers hung or spun errors; a failed write left a +// sticky error that failed every later request on the dead connection. +// +// Most tests drive a scripted HTTP/2 server (rawH2Server) that writes every +// response frame by frame, so each wire shape is pinned exactly; the +// status and byte parity tests use net/http's own server, answering the same +// route over HTTP/1.1 and HTTP/2, as the reference. + +import ( + "bufio" + "bytes" + "context" + "encoding/binary" + "errors" + "io" + "net" + "net/http" + "net/textproto" + "strconv" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "golang.org/x/net/http2" + "golang.org/x/net/http2/hpack" +) + +// --------------------------------------------------------------------------- +// Scripted HTTP/2 server + +// rawH2Action is what the scripted server does with a connection after a +// handler has answered (or not answered) a request stream. +type rawH2Action int + +const ( + rawKeep rawH2Action = iota // keep serving the connection + rawClose // close the TCP connection, no GOAWAY + rawGoAway // send GOAWAY(last stream = this one), keep the connection open +) + +// rawH2Handler answers one request stream by writing frames on c. +type rawH2Handler func(c *rawH2Conn, streamID uint32, path string) rawH2Action + +// rawH2Server speaks just enough HTTP/2, over prior knowledge or the RFC 7540 +// §3.2 h2c upgrade, for a handler to script each response frame by frame. +type rawH2Server struct { + ln net.Listener + handler rawH2Handler + killAfter time.Duration // > 0: the server closes each connection (no GOAWAY) this long after accepting it + closeAfterData int // > 0: the server closes a connection (no GOAWAY) once it has read this many DATA bytes on it + settings []http2.Setting // the server's SETTINGS (none: every default) + connWindow uint32 // > 0: grant the client this much more connection send window in the handshake + answerUpgrade bool // answer the h2c upgrade request on stream 1, as celeris does (see rawH2Handshake) + goAwayAtOnce bool // send GOAWAY as soon as the client's SETTINGS are acked, before any request + + accepted atomic.Int64 // connections accepted + killed atomic.Int64 // connections the server ended (rawClose or killAfter) + clientClosed atomic.Int64 // connections the client ended first + dataSent atomic.Int64 // DATA frame payload bytes written, padding included + windowCredit atomic.Int64 // connection WINDOW_UPDATE increments received, minus each handshake's own + answeredLate atomic.Int64 // requests answered on a connection other than the first one + clientResets atomic.Int64 // RST_STREAM frames received from the client + clientCancel atomic.Int64 // of which with error code CANCEL + dataAfterReset atomic.Int64 // DATA payload bytes received on streams the server had reset (rawH2Conn.reset) + + mu sync.Mutex + conns []net.Conn + wg sync.WaitGroup +} + +// rawH2Conn is one server-side connection, as a handler sees it. +type rawH2Conn struct { + srv *rawH2Server + nc net.Conn + fr *http2.Framer + enc *hpack.Encoder + hbuf bytes.Buffer + index int64 // accept order, from 1 + requests int // request streams seen on this connection, the current one included + held int // streams a handler started answering and left open + dataRecv int // DATA payload bytes read on this connection + resetIDs map[uint32]bool +} + +// rawH2Opts are the rawH2Server options of the same names. +type rawH2Opts struct { + killAfter time.Duration + closeAfterData int + settings []http2.Setting + connWindow uint32 + answerUpgrade bool + goAwayAtOnce bool +} + +func startRawH2(t *testing.T, h rawH2Handler) *rawH2Server { + return startRawH2With(t, rawH2Opts{}, h) +} + +func startRawH2With(t *testing.T, opts rawH2Opts, h rawH2Handler) *rawH2Server { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + s := &rawH2Server{ln: ln, handler: h, killAfter: opts.killAfter, closeAfterData: opts.closeAfterData, settings: opts.settings, + connWindow: opts.connWindow, answerUpgrade: opts.answerUpgrade, goAwayAtOnce: opts.goAwayAtOnce} + s.wg.Add(1) + go func() { + defer s.wg.Done() + for { + nc, err := ln.Accept() + if err != nil { + return + } + s.mu.Lock() + s.conns = append(s.conns, nc) + s.mu.Unlock() + s.wg.Add(1) + go s.serve(nc) + } + }() + t.Cleanup(func() { + _ = ln.Close() + s.mu.Lock() + for _, nc := range s.conns { + _ = nc.Close() + } + s.mu.Unlock() + s.wg.Wait() + }) + return s +} + +func (s *rawH2Server) url(path string) string { return "http://" + s.ln.Addr().String() + path } + +func (s *rawH2Server) hostPort() (string, string) { + h, p, _ := net.SplitHostPort(s.ln.Addr().String()) + return h, p +} + +func (s *rawH2Server) serve(nc net.Conn) { + defer s.wg.Done() + index := s.accepted.Add(1) + + // ended is set by whichever side ends the connection first, so a read + // error after the server closed it is not counted as the client's doing. + var ended atomic.Bool + serverEnd := func() { + if ended.CompareAndSwap(false, true) { + s.killed.Add(1) + _ = nc.Close() + } + } + if s.killAfter > 0 { + timer := time.AfterFunc(s.killAfter, serverEnd) + defer timer.Stop() + } + + br := bufio.NewReader(nc) + fr := http2.NewFramer(nc, br) + if !rawH2Handshake(nc, br, fr, s.answerUpgrade, s.settings...) { + _ = nc.Close() + return + } + if s.connWindow > 0 { + _ = fr.WriteWindowUpdate(0, s.connWindow) + } + c := &rawH2Conn{srv: s, nc: nc, fr: fr, index: index, resetIDs: map[uint32]bool{}} + c.enc = hpack.NewEncoder(&c.hbuf) + c.enc.SetMaxDynamicTableSizeLimit(0) // the client advertises SETTINGS_HEADER_TABLE_SIZE=0 + dec := hpack.NewDecoder(4096, nil) + handshakeCredit := true + + for { + f, err := fr.ReadFrame() + if err != nil { + if ended.CompareAndSwap(false, true) { + s.clientClosed.Add(1) + } + _ = nc.Close() + return + } + switch f := f.(type) { + case *http2.SettingsFrame: + if !f.IsAck() { + _ = fr.WriteSettingsAck() + if s.goAwayAtOnce { + _ = fr.WriteGoAway(0, http2.ErrCodeNo, nil) // the handshake is done, and no request will be served + } + } + case *http2.RSTStreamFrame: + s.clientResets.Add(1) + if f.ErrCode == http2.ErrCodeCancel { + s.clientCancel.Add(1) + } + case *http2.PingFrame: + if !f.IsAck() { + _ = fr.WritePing(true, f.Data) + } + case *http2.WindowUpdateFrame: + if f.StreamID == 0 { + if handshakeCredit { + handshakeCredit = false // the client's handshake grant, not a credit for DATA we sent + continue + } + s.windowCredit.Add(int64(f.Increment)) + } + case *http2.DataFrame: + c.dataRecv += len(f.Data()) + if c.resetIDs[f.StreamID] { + s.dataAfterReset.Add(int64(len(f.Data()))) + } + if s.closeAfterData > 0 && c.dataRecv >= s.closeAfterData { + serverEnd() + return + } + case *http2.HeadersFrame: + c.requests++ + path := "" + if fields, err := dec.DecodeFull(f.HeaderBlockFragment()); err == nil { + for _, hf := range fields { + if hf.Name == ":path" { + path = hf.Value + } + } + } + switch s.handler(c, f.StreamID, path) { + case rawClose: + serverEnd() + return + case rawGoAway: + _ = fr.WriteGoAway(f.StreamID, http2.ErrCodeNo, nil) + } + } + } +} + +// rawH2Handshake reads the client connection preface, performing the h2c +// upgrade first when the connection starts with an HTTP/1.1 request, and +// sends the server's SETTINGS. With answerUpgrade, the upgrade request is +// answered on stream 1, as RFC 7540 §3.2 requires, in the order celeris +// writes it (internal/conn/h2.go NewH2StateFromUpgrade): SETTINGS, a SETTINGS +// ACK for the client's HTTP2-Settings, then the response, before the client +// preface is read. +func rawH2Handshake(nc net.Conn, br *bufio.Reader, fr *http2.Framer, answerUpgrade bool, settings ...http2.Setting) bool { + head, err := br.Peek(len(http2.ClientPreface)) + if err != nil { + return false + } + if string(head) != http2.ClientPreface { + tp := textproto.NewReader(br) + if _, err := tp.ReadLine(); err != nil { + return false + } + if _, err := tp.ReadMIMEHeader(); err != nil { + return false + } + if _, err := io.WriteString(nc, "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: h2c\r\n\r\n"); err != nil { + return false + } + if err := fr.WriteSettings(settings...); err != nil { + return false + } + if answerUpgrade { + _ = fr.WriteSettingsAck() + _ = fr.WriteHeaders(http2.HeadersFrameParam{StreamID: 1, BlockFragment: []byte{0x88}, EndHeaders: true}) // :status 200 + _ = fr.WriteData(1, true, []byte(rawOKBody)) + } + if head, err = br.Peek(len(http2.ClientPreface)); err != nil || string(head) != http2.ClientPreface { + return false + } + _, err = br.Discard(len(http2.ClientPreface)) + return err == nil + } + if _, err := br.Discard(len(http2.ClientPreface)); err != nil { + return false + } + return fr.WriteSettings(settings...) == nil +} + +// headers writes one HEADERS frame carrying fields, given as name, value pairs. +func (c *rawH2Conn) headers(streamID uint32, endStream bool, fields ...string) { + c.headersFrame(http2.HeadersFrameParam{StreamID: streamID, EndStream: endStream, EndHeaders: true}, fields...) +} + +func (c *rawH2Conn) headersFrame(p http2.HeadersFrameParam, fields ...string) { + c.hbuf.Reset() + for i := 0; i+1 < len(fields); i += 2 { + _ = c.enc.WriteField(hpack.HeaderField{Name: fields[i], Value: fields[i+1]}) + } + p.BlockFragment = c.hbuf.Bytes() + _ = c.fr.WriteHeaders(p) +} + +// data writes one DATA frame with n body bytes and, when pad > 0, pad bytes +// of padding. +func (c *rawH2Conn) data(streamID uint32, endStream bool, n, pad int) { + var padding []byte + length := n + if pad > 0 { + padding = make([]byte, pad) + length += 1 + pad // the Pad Length field plus the padding + } + _ = c.fr.WriteDataPadded(streamID, endStream, make([]byte, n), padding) + c.srv.dataSent.Add(int64(length)) +} + +// tornData writes the 9-byte header of a DATA frame announcing declared bytes +// and only the first sent of them, so the connection ends mid-frame. +func (c *rawH2Conn) tornData(streamID uint32, declared, sent int) { + hdr := []byte{byte(declared >> 16), byte(declared >> 8), byte(declared), 0x0 /* DATA */, 0, 0, 0, 0, 0} + binary.BigEndian.PutUint32(hdr[5:], streamID) + _, _ = c.nc.Write(append(hdr, make([]byte, sent)...)) +} + +// reset ends a stream with RST_STREAM(code). DATA the client sends on it +// afterwards is counted in dataAfterReset. +func (c *rawH2Conn) reset(streamID uint32, code http2.ErrCode) { + c.resetIDs[streamID] = true + _ = c.fr.WriteRSTStream(streamID, code) +} + +// ok answers a stream with 200 and the 2-byte body "OK", in one DATA frame. +func (c *rawH2Conn) ok(streamID uint32) { + c.headers(streamID, false, ":status", "200") + c.data(streamID, true, len(rawOKBody), 0) + if c.index > 1 { + c.srv.answeredLate.Add(1) + } +} + +const rawOKBody = "OK" + +func respondOK(c *rawH2Conn, streamID uint32, _ string) rawH2Action { + c.ok(streamID) + return rawKeep +} + +// --------------------------------------------------------------------------- +// Helpers + +// runBench runs a Benchmarker and fails the test if Run does not return. +func runBench(t *testing.T, cfg Config) *Result { + t.Helper() + if cfg.Method == "" { + cfg.Method = "GET" + } + b, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + type outcome struct { + res *Result + err error + } + done := make(chan outcome, 1) + go func() { + res, err := b.Run(context.Background()) + done <- outcome{res, err} + }() + select { + case o := <-done: + if o.err != nil { + t.Fatalf("Run: %v", o.err) + } + return o.res + case <-time.After(cfg.Warmup + cfg.Duration + 20*time.Second): + t.Fatal("Benchmarker.Run did not return: a worker is hung") + return nil + } +} + +// h2Once sends one request on a fresh prior-knowledge client for path. +func h2Once(t *testing.T, host, port, path string) (int, error) { + t.Helper() + cl, err := newH2Client(host, port, path, testH2Cfg("GET", nil, nil, 1, 10)) + if err != nil { + t.Fatalf("newH2Client(%s): %v", path, err) + } + defer cl.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return cl.DoRequest(ctx, 0) +} + +// h1Once sends one request on a fresh HTTP/1.1 client for path. +func h1Once(t *testing.T, host, port, path string) (int, error) { + t.Helper() + cl, err := newH1Client(host, port, path, Config{ + Method: "GET", Workers: 1, Connections: 1, PoolSize: 1, + DialTimeout: 5 * time.Second, ReadBufferSize: 64 << 10, WriteBufferSize: 64 << 10, MaxResponseSize: -1, + }) + if err != nil { + t.Fatalf("newH1Client(%s): %v", path, err) + } + defer cl.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return cl.DoRequest(ctx, 0) +} + +// h2Status is the status an HTTP/2 DoRequest error reports, 0 for a success. +func h2Status(err error) int { + var se *HTTP2StatusError + if errors.As(err, &se) { + return se.Status + } + if err != nil { + return -1 + } + return 0 +} + +// startReferenceServer serves mux over HTTP/1.1 and prior-knowledge HTTP/2 +// cleartext on one listener, using net/http's own HTTP/2 server. +func startReferenceServer(t *testing.T, mux *http.ServeMux) (host, port string) { + t.Helper() + var protos http.Protocols + protos.SetHTTP1(true) + protos.SetUnencryptedHTTP2(true) + srv := &http.Server{Handler: mux, Protocols: &protos, ReadHeaderTimeout: 5 * time.Second} + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + go func() { _ = srv.Serve(ln) }() + t.Cleanup(func() { _ = srv.Close() }) + host, port, _ = net.SplitHostPort(ln.Addr().String()) + return host, port +} + +// warpThreshold and warpFactor define the deadline warp: a deadline armed +// more than warpThreshold ahead is a connection-lifetime bound, not a +// handshake one (those are 10 s), and fires warpFactor times sooner, so the +// 5-minute bound of #88 fires after 200 ms. +const ( + warpThreshold = 30 * time.Second + warpFactor = 1500 +) + +// warpCounts is what the deadline warp saw: every deadline armed on a warped +// connection (seen) and the long ones it warped (armed). +type warpCounts struct { + seen atomic.Int64 // deadlines armed through the warp, of any length + armed atomic.Int64 // deadlines more than warpThreshold ahead, warped +} + +// warpConn applies the deadline warp and counts the deadlines it saw. +type warpConn struct { + net.Conn + counts *warpCounts +} + +func (c warpConn) warp(t time.Time) time.Time { + if t.IsZero() { + return t + } + c.counts.seen.Add(1) + d := time.Until(t) + if d <= warpThreshold { + return t + } + c.counts.armed.Add(1) + return time.Now().Add(d / warpFactor) +} + +func (c warpConn) SetDeadline(t time.Time) error { return c.Conn.SetDeadline(c.warp(t)) } +func (c warpConn) SetReadDeadline(t time.Time) error { return c.Conn.SetReadDeadline(c.warp(t)) } +func (c warpConn) SetWriteDeadline(t time.Time) error { return c.Conn.SetWriteDeadline(c.warp(t)) } + +// warpLongDeadlines routes every loadgen dial through warpConn for the rest +// of the test and returns what the warp saw. +func warpLongDeadlines(t *testing.T) *warpCounts { + t.Helper() + counts := new(warpCounts) + dial := dialTimeoutFunc + t.Cleanup(func() { dialTimeoutFunc = dial }) + dialTimeoutFunc = func(network, addr string, timeout time.Duration) (net.Conn, error) { + c, err := dial(network, addr, timeout) + if err != nil { + return nil, err + } + return warpConn{Conn: c, counts: counts}, nil + } + return counts +} + +// requireWarped fails the test unless every connection the server accepted +// armed a deadline through the warp: the handshake's 10 s read deadline, at +// least. armed == 0 proves no lifetime deadline only for a connection the +// warp wraps; one dialed around the dialTimeoutFunc hook would carry a real +// 5-minute deadline that no sub-second test can see fire. +func requireWarped(t *testing.T, w *warpCounts, accepted int64) { + t.Helper() + if seen := w.seen.Load(); seen < accepted { + t.Fatalf("the warp saw %d deadline(s) for %d connection(s): the connections were not dialed through dialTimeoutFunc, so armed=%d proves nothing", + seen, accepted, w.armed.Load()) + } +} + +// --------------------------------------------------------------------------- +// #85: the status comes from the response HEADERS, whether or not a body follows + +// TestH2StatusMatchesH1 answers each route over HTTP/1.1 and HTTP/2 from the +// same net/http handler and requires the HTTP/2 client to classify every +// response as the HTTP/1.1 client does: status >= 400 is an error. +func TestH2StatusMatchesH1(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/ok", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("OK")) }) + mux.HandleFunc("/unauthorized", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusUnauthorized) }) + mux.HandleFunc("/bad-gateway", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte("upstream failed\n")) + }) + mux.HandleFunc("/unavailable", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte("try later\n")) + }) + host, port := startReferenceServer(t, mux) + + for _, tc := range []struct { + path string + status int // 0: a success + shape string + }{ + {"/ok", 0, "200, body"}, + {"/missing", 404, "http.NotFound: 404, then a 19-byte body"}, + {"/unauthorized", 401, "401, no body; net/http Huffman-codes the status"}, + {"/bad-gateway", 502, "502, then a body; Huffman-coded status"}, + {"/unavailable", 503, "503, then a body; literal status"}, + } { + _, h1err := h1Once(t, host, port, tc.path) + _, h2err := h2Once(t, host, port, tc.path) + t.Logf("%-13s h1 err=%v | h2 status=%d err=%v", tc.path, h1err, h2Status(h2err), h2err) + if (h1err != nil) != (tc.status != 0) { + t.Fatalf("%s: the HTTP/1.1 control disagrees with the route: err=%v", tc.path, h1err) + } + if got := h2Status(h2err); got != tc.status { + t.Errorf("%s (%s): HTTP/2 recorded status %d (err=%v), HTTP/1.1 recorded %d (err=%v): the HTTP/2 client must take the status from the response HEADERS", + tc.path, tc.shape, got, h2err, tc.status, h1err) + } + } +} + +// TestH2StatusRecordedFromHeaders pins the frame layouts that carry a status +// on a HEADERS frame that does not end the stream. +func TestH2StatusRecordedFromHeaders(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, path string) rawH2Action { + switch path { + case "/404-body": + c.headers(sid, false, ":status", "404") + c.data(sid, true, 19, 0) + case "/500-body-empty-end": + c.headers(sid, false, ":status", "500") + c.data(sid, false, 30, 0) + c.data(sid, true, 0, 0) + case "/503-body-trailers": + c.headers(sid, false, ":status", "503") + c.data(sid, false, 30, 0) + c.headers(sid, true, "x-checksum", "abc") + case "/502-no-body": + c.headers(sid, true, ":status", "502") // Huffman-coded: 16 bits beat the 3 literal bytes + case "/401-body": + c.headers(sid, false, ":status", "401") // Huffman-coded + c.data(sid, true, 12, 0) + case "/103-then-404": + c.headers(sid, false, ":status", "103") // interim response, then the final one + c.headers(sid, false, ":status", "404") + c.data(sid, true, 19, 0) + case "/404-padded-priority": + c.headersFrame(http2.HeadersFrameParam{StreamID: sid, EndHeaders: true, PadLength: 7, + Priority: http2.PriorityParam{Weight: 200}}, ":status", "404") + c.data(sid, true, 19, 0) + case "/103-then-200": + c.headers(sid, false, ":status", "103") + c.headers(sid, false, ":status", "200") + c.data(sid, true, 19, 0) + default: + c.ok(sid) + } + return rawKeep + }) + host, port := srv.hostPort() + + for _, tc := range []struct { + path string + status int // 0: a success + }{ + {"/404-body", 404}, + {"/500-body-empty-end", 500}, + {"/503-body-trailers", 503}, + {"/502-no-body", 502}, + {"/401-body", 401}, + {"/103-then-404", 404}, + {"/404-padded-priority", 404}, + {"/103-then-200", 0}, + {"/ok", 0}, + } { + _, err := h2Once(t, host, port, tc.path) + t.Logf("%-21s status=%d err=%v", tc.path, h2Status(err), err) + if got := h2Status(err); got != tc.status { + t.Errorf("%s: recorded status %d (err=%v), want %d: the status must come from the final response HEADERS, whatever frames follow it", + tc.path, got, err, tc.status) + } + } +} + +// TestExtractStatusEveryEncoding decodes every status 100-599 in every way an +// HPACK encoder may send it when the dynamic table is off: a literal naming +// :status by any of the static indices 8-14 (x/net's encoder, and so +// net/http's server, uses 14), without indexing, never indexed or with +// incremental indexing, and with a plain or a Huffman-coded value (RFC 7541 +// §5.2; an encoder picks Huffman whenever it is shorter, as it is for 401, +// 502, 103 and every code with two digits from {0, 1, 2}). +func TestExtractStatusEveryEncoding(t *testing.T) { + var buf bytes.Buffer + enc := hpack.NewEncoder(&buf) + enc.SetMaxDynamicTableSizeLimit(0) // as the client advertises + missed, total := 0, 0 + check := func(desc string, block []byte, status int) { + total++ + if got := extractStatus(block); got != status { + missed++ + if missed <= 8 { + t.Errorf("extractStatus(%s, % x) = %d, want %d", desc, block, got, status) + } + } + } + for status := 100; status < 600; status++ { + value := strconv.Itoa(status) + buf.Reset() + _ = enc.WriteField(hpack.HeaderField{Name: ":status", Value: value}) + check("x/net hpack encoder", bytes.Clone(buf.Bytes()), status) + + huff := hpack.AppendHuffmanString(nil, value) + for idx := byte(8); idx <= 14; idx++ { + for _, first := range []byte{idx, 0x10 | idx, 0x40 | idx} { // without indexing, never indexed, incremental + check("literal "+value, append([]byte{first, byte(len(value))}, value...), status) + check("Huffman "+value, append([]byte{first, 0x80 | byte(len(huff))}, huff...), status) + } + } + } + if missed > 0 { + t.Errorf("%d of %d encodings of a status were not decoded", missed, total) + } +} + +// TestH2ErrorStatusWithBodyCountsAsError is #85 as the issue measured it: the +// Benchmarker on an all-404-with-body route must report errors and no +// requests, over prior knowledge, the h2c upgrade and a -mix of both. +func TestH2ErrorStatusWithBodyCountsAsError(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + c.headers(sid, false, ":status", "404") + c.data(sid, true, 19, 0) + return rawKeep + }) + opts := HTTP2Options{Connections: 1, MaxStreams: 4} + for _, tc := range []struct { + name string + cfg Config + }{ + {"prior-knowledge", Config{HTTP2: true}}, + {"h2c-upgrade", Config{H2CUpgrade: true}}, + {"mix", Config{Mix: &MixRatio{H2: 1, Upgrade: 1}}}, + } { + t.Run(tc.name, func(t *testing.T) { + cfg := tc.cfg + cfg.URL = srv.url("/missing") + cfg.Duration = 150 * time.Millisecond + cfg.Workers = 4 + cfg.HTTP2Options = opts + res := runBench(t, cfg) + t.Logf("requests=%d errors=%d mix=%+v", res.Requests, res.Errors, res.Mix) + if res.Requests != 0 || res.Errors == 0 { + t.Errorf("requests=%d errors=%d, want 0 requests and some errors: every response was a 404 with a body", res.Requests, res.Errors) + } + if m := res.Mix; m != nil { + if m.H2Conns == 0 || m.UpgradeConns == 0 { + t.Fatalf("mix did not exercise both HTTP/2 paths: %+v", *m) + } + if m.H2Requests != 0 || m.UpgradeRequests != 0 || m.H2Errors == 0 || m.UpgradeErrors == 0 { + t.Errorf("mix stats %+v: want every H2 and upgrade response counted as an error", *m) + } + } + }) + } +} + +// --------------------------------------------------------------------------- +// #86: every DATA frame's payload counts, padding excluded + +// TestH2BodyBytesMatchH1 compares the body bytes each client reports for the +// same net/http route. net/http's HTTP/2 server ends a body it has already +// flushed with an empty DATA frame; HTTP/1.1 counts the Content-Length (or +// the chunk sizes), so HTTP/2 must count every DATA frame's data. +func TestH2BodyBytesMatchH1(t *testing.T) { + body64k := bytes.Repeat([]byte("x"), 64<<10) + body1k := bytes.Repeat([]byte("y"), 1<<10) + mux := http.NewServeMux() + mux.HandleFunc("/ok", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("OK")) }) + mux.HandleFunc("/64k", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write(body64k) }) + mux.HandleFunc("/1k-flushed", func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write(body1k) + w.(http.Flusher).Flush() + }) + host, port := startReferenceServer(t, mux) + + for _, tc := range []struct { + path string + want int + }{ + {"/ok", 2}, + {"/64k", 64 << 10}, + {"/1k-flushed", 1 << 10}, + } { + n1, err1 := h1Once(t, host, port, tc.path) + n2, err2 := h2Once(t, host, port, tc.path) + t.Logf("%-11s h1 bytes=%d | h2 bytes=%d", tc.path, n1, n2) + if err1 != nil || err2 != nil { + t.Fatalf("%s: h1 err=%v h2 err=%v", tc.path, err1, err2) + } + if n1 != tc.want { + t.Fatalf("%s: the HTTP/1.1 control read %d body bytes, want %d", tc.path, n1, tc.want) + } + if n2 != n1 { + t.Errorf("%s: HTTP/2 counted %d body bytes, HTTP/1.1 counted %d: every DATA frame of the response must count", tc.path, n2, n1) + } + } +} + +// TestH2BodyBytesCountEveryDATAFrame pins the DATA frame layouts, including +// padding (never counted as body, as HTTP/1.1 never counts chunk framing) +// and trailers. +func TestH2BodyBytesCountEveryDATAFrame(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, path string) rawH2Action { + switch path { + case "/four-frames": // the usual 16 KiB framing, END_STREAM on the last non-empty frame + c.headers(sid, false, ":status", "200") + for i := range 4 { + c.data(sid, i == 3, 16<<10, 0) + } + case "/empty-end": // what net/http sends for a flushed body + c.headers(sid, false, ":status", "200") + c.data(sid, false, 64<<10, 0) + c.data(sid, true, 0, 0) + case "/trailers": + c.headers(sid, false, ":status", "200") + c.data(sid, false, 1000, 0) + c.data(sid, false, 1000, 0) + c.headers(sid, true, "x-checksum", "abc") + case "/padded": + c.headers(sid, false, ":status", "200") + for i := range 3 { + c.data(sid, i == 2, 1000, 16) + } + default: + c.ok(sid) + } + return rawKeep + }) + host, port := srv.hostPort() + + for _, tc := range []struct { + path string + want int + }{ + {"/four-frames", 64 << 10}, + {"/empty-end", 64 << 10}, + {"/trailers", 2000}, + {"/padded", 3000}, + {"/ok", 2}, + } { + n, err := h2Once(t, host, port, tc.path) + t.Logf("%-12s bytes=%d want=%d err=%v", tc.path, n, tc.want, err) + if err != nil { + t.Fatalf("%s: %v", tc.path, err) + } + if n != tc.want { + t.Errorf("%s: counted %d body bytes, want %d", tc.path, n, tc.want) + } + } +} + +// TestH2FlowControlCreditsPadding guards the other side of #86: the body +// count leaves padding out, but the connection flow-control window must +// still be credited with every DATA payload byte received, padding included +// (RFC 9113 §6.9.1), or padded responses slowly close the window. +func TestH2FlowControlCreditsPadding(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + c.headers(sid, false, ":status", "200") + for i := range 3 { + c.data(sid, i == 2, 1000, 16) + } + return rawKeep + }) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 10)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + for range 5 { + if _, err := cl.DoRequest(ctx, 0); err != nil { + t.Fatal(err) + } + } + // The client flushes its WINDOW_UPDATEs within a millisecond or so. + deadline := time.Now().Add(2 * time.Second) + for srv.windowCredit.Load() != srv.dataSent.Load() && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + credit, sent := srv.windowCredit.Load(), srv.dataSent.Load() + t.Logf("DATA payload sent (padding included)=%d, connection window credited=%d", sent, credit) + if credit != sent { + t.Errorf("the client credited %d bytes of connection window for %d bytes of DATA payload: flow control must count padding too", credit, sent) + } +} + +// TestH2ThroughputCountsEveryDATAFrame is #86 as the issue measured it: +// throughput_bps over a 4 x 16 KiB response must be the whole body per +// request, not a quarter of it. +func TestH2ThroughputCountsEveryDATAFrame(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + c.headers(sid, false, ":status", "200") + for i := range 4 { + c.data(sid, i == 3, 16<<10, 0) + } + return rawKeep + }) + res := runBench(t, Config{URL: srv.url("/"), Duration: 150 * time.Millisecond, Workers: 1, + HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: 4}}) + t.Logf("requests=%d errors=%d rps=%.0f throughput_bps=%.0f bytes/response=%.1f", + res.Requests, res.Errors, res.RequestsPerSec, res.ThroughputBPS, res.ThroughputBPS/res.RequestsPerSec) + if res.Requests == 0 || res.Errors != 0 { + t.Fatalf("requests=%d errors=%d", res.Requests, res.Errors) + } + if perReq := res.ThroughputBPS / res.RequestsPerSec; perReq < 65535.5 || perReq > 65536.5 { + t.Errorf("throughput_bps / requests_per_sec = %.1f bytes per response, want 65536", perReq) + } +} + +// --------------------------------------------------------------------------- +// #88: no lifetime deadline + +// TestH2ConnectionHasNoLifetimeDeadline runs one H2 connection past the +// connection-lifetime bound #88 put on it (5 minutes, warped to 200 ms). Like +// HTTP/1.1, which sets no deadline, the connection must keep serving: no +// errors, no second connection. +func TestH2ConnectionHasNoLifetimeDeadline(t *testing.T) { + warp := warpLongDeadlines(t) + srv := startRawH2(t, respondOK) + res := runBench(t, Config{URL: srv.url("/"), Duration: 600 * time.Millisecond, Workers: 1, MaxRPS: 400, + HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: 4}}) + t.Logf("deadlines seen=%d long deadlines armed=%d requests=%d errors=%d connect_errors=%d connections=%d", + warp.seen.Load(), warp.armed.Load(), res.Requests, res.Errors, res.ConnectErrors, srv.accepted.Load()) + requireWarped(t, warp, srv.accepted.Load()) + if n := warp.armed.Load(); n != 0 { + t.Errorf("%d deadline(s) more than %v ahead were armed on the connection: an H2 connection must not carry a lifetime deadline", n, warpThreshold) + } + if res.Errors != 0 || res.ConnectErrors != 0 || srv.accepted.Load() != 1 { + t.Errorf("errors=%d connect_errors=%d connections=%d, want 0, 0 and 1: the connection died at a lifetime deadline (5 min, warped here to %v)", + res.Errors, res.ConnectErrors, srv.accepted.Load(), 5*time.Minute/warpFactor) + } + if res.Requests == 0 { + t.Error("no request completed") + } +} + +// --------------------------------------------------------------------------- +// #89: a connection that dies without a GOAWAY is redialed + +// TestH2RedialsAfterServerClosesWithoutGOAWAY: the server answers one request +// per connection and closes it, with no GOAWAY. The client must redial and +// keep going, count at most the one in-flight stream of each dead +// connection as an error, and not charge the workers queued for a stream. +func TestH2RedialsAfterServerClosesWithoutGOAWAY(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + c.ok(sid) + return rawClose + }) + // 8 workers queue for 1 stream: the ones waiting when the connection + // dies never reached it, so they are moved to the redialed connection. + res := runBench(t, Config{URL: srv.url("/"), Duration: 300 * time.Millisecond, Workers: 2, + HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: 1}}) + accepted := srv.accepted.Load() + t.Logf("connections=%d requests=%d errors=%d connect_errors=%d", accepted, res.Requests, res.Errors, res.ConnectErrors) + if accepted < 3 { + t.Fatalf("the server accepted %d connection(s) in %v and requests=%d errors=%d connect_errors=%d: the client did not redial after the server closed without GOAWAY", + accepted, 300*time.Millisecond, res.Requests, res.Errors, res.ConnectErrors) + } + if res.Requests < accepted-1 { + t.Errorf("requests=%d over %d connections: each connection answers one request", res.Requests, accepted) + } + if res.Errors > accepted { + t.Errorf("errors=%d over %d connections with 1 stream each: more than one error per dead connection means requests that never reached it were charged, or a worker spun", + res.Errors, accepted) + } +} + +// TestH2ReadErrorMidStreamFailsEachInFlightStreamOnce: on each connection the +// server answers 3 requests, then starts 4 responses (HEADERS and part of the +// body) and ends the connection in the middle of the 4th response's DATA +// frame, so the client's read fails with an unexpected EOF. Each of those 4 +// streams is exactly one error, the client redials, and the torn bodies are +// credited to nothing. (The server closes rather than resets: on Darwin 27 a +// reset was seen to reach the client's read ~334 ms late, a kernel delay +// that would make this test's count depend on the platform.) +func TestH2ReadErrorMidStreamFailsEachInFlightStreamOnce(t *testing.T) { + const answered, inFlight = 3, 4 + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + if c.requests <= answered { + c.ok(sid) + return rawKeep + } + c.headers(sid, false, ":status", "200") + if c.held++; c.held == inFlight { + c.tornData(sid, 100, 40) // the connection ends 40 bytes into a 100-byte frame + return rawClose + } + c.data(sid, false, 100, 0) // the body starts and never ends + return rawKeep + }) + res := runBench(t, Config{URL: srv.url("/"), Duration: 400 * time.Millisecond, Workers: 1, + HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: inFlight}}) + accepted, killed := srv.accepted.Load(), srv.killed.Load() + t.Logf("connections=%d torn=%d requests=%d errors=%d connect_errors=%d bytes/response=%.3f", + accepted, killed, res.Requests, res.Errors, res.ConnectErrors, res.ThroughputBPS/res.RequestsPerSec) + if accepted < 3 { + t.Fatalf("the server accepted %d connection(s) in %v and requests=%d errors=%d: the client did not redial after a read error mid-stream", + accepted, 400*time.Millisecond, res.Requests, res.Errors) + } + // The last torn connection can race the end of the run, which is not an error. + if lo, hi := inFlight*(killed-1), inFlight*killed; res.Errors < lo || res.Errors > hi { + t.Errorf("errors=%d after %d torn connections with %d streams in flight each: want %d..%d, one error per in-flight stream", + res.Errors, killed, inFlight, lo, hi) + } + if perReq := res.ThroughputBPS / res.RequestsPerSec; perReq < float64(len(rawOKBody))-0.01 || perReq > float64(len(rawOKBody))+0.01 { + t.Errorf("throughput_bps / requests_per_sec = %.3f bytes per response, want %d: the partial bodies of failed streams leaked into the count", + perReq, len(rawOKBody)) + } +} + +// TestH2ResponseWinsOverConnectionClose: a worker that reaches its wait +// after readLoop has both answered its stream and failed the connection finds +// the response and the closed connection ready at once, and, since round 1, +// both loops already returned (loopsDone). The response is the request's +// outcome; select alone would pick at random. (Found in review of the #89 +// fix, which makes a close right after a response common.) +func TestH2ResponseWinsOverConnectionClose(t *testing.T) { + hc := &h2Conn{done: make(chan struct{}), loopsDone: make(chan struct{}), streamSem: make(chan struct{}, 1)} + close(hc.done) + close(hc.loopsDone) + for i := range 200 { + ch := make(chan h2Response, 1) + ch <- h2Response{status: 200, bytesRead: 2} + n, err := hc.await(context.Background(), &ch, 0) + <-hc.streamSem // the token await returned + if err != nil || n != 2 { + t.Fatalf("attempt %d: a delivered 200 response on a closed connection returned (%d, %v), want (2, nil)", i, n, err) + } + } +} + +// failWritesConn fails every Write after the first okWrites on the +// connection with ECONNRESET, writing nothing, while reads go on: the +// connection's death shows on the write side only, as when the server is gone +// but its reset has not reached the client's read. +type failWritesConn struct { + net.Conn + okWrites int + writes atomic.Int64 +} + +func (c *failWritesConn) Write(p []byte) (int, error) { + if c.writes.Add(1) > int64(c.okWrites) { + return 0, &net.OpError{Op: "write", Net: "tcp", Err: syscall.ECONNRESET} + } + return c.Conn.Write(p) +} + +// TestH2WriteErrorFailsTheConnection: each connection's writes fail from the +// 6th on (3 go to the handshake) while the server stays silent and keeps the +// connection open, so no read ever fails. The failed write alone must end the +// connection and be redialed. On main a failed flush left bufio's sticky +// error behind, and every later request failed at once on the same dead +// connection: a spin of errors that never redialed. +func TestH2WriteErrorFailsTheConnection(t *testing.T) { + dial := dialTimeoutFunc + t.Cleanup(func() { dialTimeoutFunc = dial }) + dialTimeoutFunc = func(network, addr string, timeout time.Duration) (net.Conn, error) { + c, err := dial(network, addr, timeout) + if err != nil { + return nil, err + } + return &failWritesConn{Conn: c, okWrites: 5}, nil + } + srv := startRawH2(t, respondOK) + const streams = 4 + res := runBench(t, Config{URL: srv.url("/"), Duration: 300 * time.Millisecond, Workers: 1, + HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: streams}}) + accepted := srv.accepted.Load() + t.Logf("connections=%d requests=%d errors=%d connect_errors=%d", accepted, res.Requests, res.Errors, res.ConnectErrors) + if accepted < 3 { + t.Fatalf("the server accepted %d connection(s) in %v and requests=%d errors=%d: a connection whose writes fail was not ended and redialed", + accepted, 300*time.Millisecond, res.Requests, res.Errors) + } + if res.Errors > streams*accepted { + t.Errorf("errors=%d over %d connections with %d streams each: requests kept failing on a dead connection", res.Errors, accepted, streams) + } +} + +// TestH2InFlightStreamAnsweredOnceWhenConnectionDiesMidBody: a POST whose +// body is larger than the send window waits in writeBodyFlowControlled for a +// WINDOW_UPDATE the server never sends; the server reads the 65,535 bytes +// the window allowed and closes the connection. The dead connection must be +// torn down (on main its writer waited for window forever), and the stream +// must be answered exactly once, although both readLoop (the connection +// died) and writeLoop (its body write gave up) fail it. +func TestH2InFlightStreamAnsweredOnceWhenConnectionDiesMidBody(t *testing.T) { + srv := startRawH2With(t, rawH2Opts{closeAfterData: 65535}, func(*rawH2Conn, uint32, string) rawH2Action { + return rawKeep // take the request and its body, never answer, never grant window + }) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/upload", testH2Cfg("POST", nil, make([]byte, 100<<10), 1, 10)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + hc := cl.conns[0].cur.Load() + + // The request, as DoRequest hands it to the connection, with a response + // channel the test owns and never drains until the connection is gone. + ch := make(chan h2Response, 1) + hc.writeCh <- h2WriteReq{kind: h2WriteHeaders, block: cl.headerBlock, data: cl.dataPayload, hasBody: true, respCh: &ch} + + select { + case <-hc.done: + case <-time.After(3 * time.Second): + t.Fatal("the server read the 65,535 body bytes the window allowed and closed the connection, and the client never tore it down: its writer still waits for window on a dead connection") + } + var first h2Response + select { + case first = <-ch: + case <-time.After(2 * time.Second): + t.Fatal("the in-flight stream was never answered") + } + if first.err == nil { + t.Fatalf("the in-flight stream on a dead connection was answered with success: %+v", first) + } + select { + case second := <-ch: + t.Errorf("the in-flight stream was answered twice: %v, then %v", first.err, second.err) + case <-time.After(200 * time.Millisecond): + } + t.Logf("answered once: %v", first.err) +} + +// TestH2GOAWAYConnectionIsReleased: after a GOAWAY the client redials (it +// always has) and must also close the connection it abandons, instead of +// leaking the socket and its writer goroutine. +func TestH2GOAWAYConnectionIsReleased(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + c.ok(sid) + return rawGoAway // and leave the connection open for the client to close + }) + res := runBench(t, Config{URL: srv.url("/"), Duration: 300 * time.Millisecond, Workers: 1, + HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: 1}}) + if srv.accepted.Load() < 2 { + t.Fatalf("the server accepted %d connection(s), requests=%d errors=%d: the client did not redial after GOAWAY", srv.accepted.Load(), res.Requests, res.Errors) + } + deadline := time.Now().Add(2 * time.Second) + for srv.clientClosed.Load() < srv.accepted.Load() && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + t.Logf("connections=%d closed by the client=%d requests=%d errors=%d", srv.accepted.Load(), srv.clientClosed.Load(), res.Requests, res.Errors) + if closed, accepted := srv.clientClosed.Load(), srv.accepted.Load(); closed < accepted { + t.Errorf("the client closed %d of the %d connections it opened, after Run returned: the ones the server GOAWAYed are leaked", closed, accepted) + } +} + +// --------------------------------------------------------------------------- +// #88 and #89 together + +// TestH2ServerEndedConnectionIsRedialedWithoutLifetimeDeadline: with the +// lifetime deadline gone (#88), a connection ends only when the server or the +// network ends it, and that must still be redialed (#89). The server closes +// every connection, without GOAWAY, 300 ms after accepting it; the lifetime +// bound would have fired at 200 ms. +func TestH2ServerEndedConnectionIsRedialedWithoutLifetimeDeadline(t *testing.T) { + warp := warpLongDeadlines(t) + srv := startRawH2With(t, rawH2Opts{killAfter: 300 * time.Millisecond}, respondOK) + const streams = 4 + res := runBench(t, Config{URL: srv.url("/"), Duration: 800 * time.Millisecond, Workers: 1, MaxRPS: 400, + HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: streams}}) + killed := srv.killed.Load() + t.Logf("deadlines seen=%d long deadlines armed=%d connections=%d server closes=%d answered after a redial=%d closed by the client=%d requests=%d errors=%d connect_errors=%d", + warp.seen.Load(), warp.armed.Load(), srv.accepted.Load(), killed, srv.answeredLate.Load(), srv.clientClosed.Load(), res.Requests, res.Errors, res.ConnectErrors) + requireWarped(t, warp, srv.accepted.Load()) + if n := warp.armed.Load(); n != 0 { + t.Errorf("%d deadline(s) more than %v ahead were armed: the connection still has a lifetime deadline", n, warpThreshold) + } + if killed == 0 || srv.accepted.Load() < 2 || srv.answeredLate.Load() == 0 { + t.Fatalf("server closes=%d connections=%d requests answered after a redial=%d (requests=%d errors=%d): the client did not redial the connection the server ended", + killed, srv.accepted.Load(), srv.answeredLate.Load(), res.Requests, res.Errors) + } + if res.Errors > streams*killed { + t.Errorf("errors=%d after %d server closes with at most %d streams in flight each", res.Errors, killed, streams) + } + if srv.clientClosed.Load() > 1 { // the one Close at the end of the run + t.Errorf("the client ended %d connections itself before the server did", srv.clientClosed.Load()) + } +} + +// --------------------------------------------------------------------------- +// Review of the #85-#89 fixes (PR #91, round 1) + +// TestH2StreamIDExhaustionRedials: with no lifetime deadline (#88) a +// connection lives as long as the server keeps it, so a long run can use up +// its 2^30 client stream IDs. RFC 9113 §5.1.1: a client that cannot open a +// new stream opens a new connection. The request that finds the IDs used up +// never reached the server, so it goes to a redialed connection and does not +// fail, and the used-up connection is closed. On 2930ded every request from +// then on failed at once on a connection nothing retired. +func TestH2StreamIDExhaustionRedials(t *testing.T) { + srv := startRawH2(t, respondOK) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 1)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + // Three stream IDs left on the first connection: 2^31-5, 2^31-3, 2^31-1. + cl.conns[0].cur.Load().nextStreamID.Store(0x7FFFFFFB) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + for i := range 8 { + if _, err := cl.DoRequest(ctx, 0); err != nil { + t.Fatalf("request %d: %v (connections=%d): a request that finds the stream IDs used up must go to a new connection, not fail", + i+1, err, srv.accepted.Load()) + } + } + deadline := time.Now().Add(2 * time.Second) + for srv.clientClosed.Load() < 1 && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + t.Logf("8 requests: connections=%d closed by the client=%d", srv.accepted.Load(), srv.clientClosed.Load()) + if n := srv.accepted.Load(); n != 2 { + t.Errorf("connections=%d, want 2: one redial when the first connection's stream IDs ran out", n) + } + if srv.clientClosed.Load() < 1 { + t.Error("the client never closed the connection whose stream IDs ran out") + } +} + +// uploadWindow is the SETTINGS_INITIAL_WINDOW_SIZE of the servers below: a +// request body of 2,000 bytes sends 1,000 and then waits for window that the +// server never grants. +var uploadWindow = []http2.Setting{{ID: http2.SettingInitialWindowSize, Val: 1000}} + +// sendUpload hands one POST with a 2,000-byte body to the client's first +// connection, as DoRequest does, with a response channel the test owns. +func sendUpload(t *testing.T, srv *rawH2Server) (*h2Conn, chan h2Response) { + t.Helper() + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/upload", testH2Cfg("POST", nil, make([]byte, 2000), 1, 10)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(cl.Close) + hc := cl.conns[0].cur.Load() + ch := make(chan h2Response, 1) + hc.writeCh <- h2WriteReq{kind: h2WriteHeaders, block: cl.headerBlock, data: cl.dataPayload, hasBody: true, respCh: &ch} + return hc, ch +} + +// TestH2WriteErrorWhileBodyWaitsForWindow: a POST body waits for send window, +// and the flush it makes while waiting is the connection's first write to +// fail. The server stays silent and keeps the connection open, so no read +// fails. The failed flush must end the connection and answer the stream, +// once, as a failed write anywhere else does (#89). On 2930ded the wait +// discarded the flush error and polled for window until the run ended, with +// every worker of the connection stuck behind it. +func TestH2WriteErrorWhileBodyWaitsForWindow(t *testing.T) { + dial := dialTimeoutFunc + t.Cleanup(func() { dialTimeoutFunc = dial }) + dialTimeoutFunc = func(network, addr string, timeout time.Duration) (net.Conn, error) { + c, err := dial(network, addr, timeout) + if err != nil { + return nil, err + } + return &failWritesConn{Conn: c, okWrites: 3}, nil // the handshake's 3 writes succeed, every later one fails + } + srv := startRawH2With(t, rawH2Opts{settings: uploadWindow}, func(*rawH2Conn, uint32, string) rawH2Action { + return rawKeep // never answer, never grant window + }) + hc, ch := sendUpload(t, srv) + + select { + case <-hc.done: + case <-time.After(3 * time.Second): + t.Fatal("the flush made while the body waited for window failed, and the client never ended the connection: its writer still polls for window on a connection it cannot write to") + } + var first h2Response + select { + case first = <-ch: + case <-time.After(2 * time.Second): + t.Fatal("the stream whose body could not be written was never answered") + } + if first.err == nil { + t.Fatalf("the stream whose body could not be written was answered with success: %+v", first) + } + select { + case second := <-ch: + t.Errorf("the stream was answered twice: %v, then %v", first.err, second.err) + case <-time.After(200 * time.Millisecond): + } + t.Logf("answered once: %v", first.err) +} + +// TestH2WindowCreditSentWhileBodyWaitsForWindow: while a POST body waits for +// send window, the client must still return receive window to the server. +// The server answers the POST at once with 10,000 bytes of a body it does not +// end, and never grants window for the upload. The client must credit those +// bytes to the connection window although its writer is parked in the +// flow-control wait: a server that waits for that credit before it reads or +// grants more deadlocks with the client otherwise. On 2930ded the credit was +// sent only once the wait ended, which it never did. +func TestH2WindowCreditSentWhileBodyWaitsForWindow(t *testing.T) { + const respBody = 10000 + srv := startRawH2With(t, rawH2Opts{settings: uploadWindow}, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + c.headers(sid, false, ":status", "200") + c.data(sid, false, respBody, 0) + return rawKeep // never end the response, never grant window + }) + sendUpload(t, srv) + + deadline := time.Now().Add(2 * time.Second) + for srv.windowCredit.Load() < respBody && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + t.Logf("response bytes received by the client=%d, connection window credited=%d", srv.dataSent.Load(), srv.windowCredit.Load()) + if got := srv.windowCredit.Load(); got < respBody { + t.Errorf("the client credited %d of the %d response bytes it received while its request body waited for window: the writer must send WINDOW_UPDATE while it waits", got, respBody) + } +} + +// TestH2RedialBacksOffOnlyWhenTheServerLooksDown: how a slot redials, matched +// to h1client (h1client.go DoRequest). h1client redials a connection the +// server ended at once, and when the redial fails it counts a connect error +// and sleeps its backoff inside the request it returns as an error, so the +// sleep reaches no recorded latency; the next request dials at once. +// +// HTTP/2 does the same, and treats a connection that ended before it carried +// a response to any request as the server looking down too: the request that +// finds it fails and sleeps, and nothing is dialed for it, so a server that +// completes each handshake and then drops the connection is paced and shows +// as errors. The pace doubles while the server keeps looking down, and is +// reset once a connection has served. Every subtest drives DoRequest; the +// backoff is set far beyond each context where a sleep must not happen. +func TestH2RedialBacksOffOnlyWhenTheServerLooksDown(t *testing.T) { + srv := startRawH2(t, respondOK) + host, port := srv.hostPort() + newClient := func(t *testing.T) *h2Client { + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 1)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(cl.Close) + return cl + } + const longBackoff = 20 * time.Second // sleep() jitters it to 10-20 s + // endConn ends the slot's connection as a server that closes it does. + endConn := func(slot *h2ConnSlot, why string) { + slot.cur.Load().failConn(errors.New("test: " + why)) + } + // emptySlot leaves the slot as a failed redial leaves it. + emptySlot := func(slot *h2ConnSlot) { + slot.cur.Load().closeConn() + slot.cur.Store(nil) + } + ctxFor := func(t *testing.T, d time.Duration) context.Context { + ctx, cancel := context.WithTimeout(context.Background(), d) + t.Cleanup(cancel) + return ctx + } + + t.Run("answered", func(t *testing.T) { + cl := newClient(t) + dials := countDials(cl, 0) + ctx := ctxFor(t, 3*time.Second) + if _, err := cl.DoRequest(ctx, 0); err != nil { + t.Fatal(err) + } + slot := cl.conns[0] + endConn(slot, "the server ended the connection") + slot.backoff.next = longBackoff + start := time.Now() + _, err := cl.DoRequest(ctx, 0) + t.Logf("request after a connection that answered: %v, err=%v, dials=%d", time.Since(start), err, dials.Load()) + if err != nil { + t.Fatalf("the request that redials a connection which had answered failed after %v (%v): h1client redials such a connection at once, without its backoff", + time.Since(start), err) + } + }) + t.Run("never answered", func(t *testing.T) { + cl := newClient(t) + dials := countDials(cl, 0) + slot := cl.conns[0] + endConn(slot, "the server dropped the connection before it answered anything") + slot.backoff.next = 200 * time.Millisecond // sleep() jitters it to 100-200 ms + start := time.Now() + _, err := cl.DoRequest(ctxFor(t, 3*time.Second), 0) + elapsed := time.Since(start) + t.Logf("request that found a connection which never answered: %v, err=%v, dials=%d", elapsed, err, dials.Load()) + if err == nil { + t.Fatalf("the request that found a connection which ended before it answered anything succeeded, after %v and %d dial(s): the server looks down, so it must fail, sleep the backoff inside that failure and dial nothing, or the sleep lands in a recorded latency and the outage shows no error", + elapsed, dials.Load()) + } + if n := dials.Load(); n != 0 { + t.Errorf("%d dial(s) for the request that found a connection which never answered: it fails, and the next request dials", n) + } + if elapsed < 100*time.Millisecond { + t.Errorf("the request returned after %v: it did not sleep the backoff (100-200 ms), so a server that drops every connection is redialed in a loop", elapsed) + } + }) + t.Run("after a failed redial", func(t *testing.T) { + cl := newClient(t) + slot := cl.conns[0] + emptySlot(slot) + slot.backoff.next = longBackoff + start := time.Now() + _, err := cl.DoRequest(ctxFor(t, 3*time.Second), 0) + t.Logf("request after a failed redial: %v, err=%v", time.Since(start), err) + if err != nil { + t.Fatalf("the request after a failed redial failed after %v (%v): it slept the backoff before it dialed, so the sleep lands in its latency when the dial succeeds; h1client sleeps only inside the request whose redial failed", + time.Since(start), err) + } + }) + t.Run("a failed redial", func(t *testing.T) { + cl := newClient(t) + ctx := ctxFor(t, 3*time.Second) + if _, err := cl.DoRequest(ctx, 0); err != nil { + t.Fatal(err) + } + dials := countDials(cl, 1) + slot := cl.conns[0] + endConn(slot, "the server ended the connection") + start := time.Now() + _, err := cl.DoRequest(ctx, 0) + elapsed := time.Since(start) + t.Logf("request whose redial failed: %v, err=%v, dials=%d", elapsed, err, dials.Load()) + if err == nil || dials.Load() != 1 { + t.Fatalf("err=%v dials=%d, want an error after 1 failed dial", err, dials.Load()) + } + // The connection it replaces had served, so the pace starts over: + // the first step sleeps 5-10 ms. + if elapsed < reconnectBackoffMin/2 { + t.Errorf("the request whose redial failed returned after %v without sleeping the backoff (at least %v): as in h1client it paces the server that looks down inside the request it fails", + elapsed, reconnectBackoffMin/2) + } + }) + t.Run("the sleep holds no lock", func(t *testing.T) { + cl := newClient(t) + slot := cl.conns[0] + emptySlot(slot) + slot.backoff.next = longBackoff + failed := make(chan struct{}) + var n atomic.Int64 + redial := cl.redial + cl.redial = func() (*h2Conn, error) { + if n.Add(1) == 1 { + close(failed) + return nil, errors.New("test: connection refused") + } + return redial() + } + ctxA, cancelA := context.WithTimeout(context.Background(), 10*time.Second) + defer cancelA() + aDone := make(chan error, 1) + go func() { + _, err := cl.DoRequest(ctxA, 0) + aDone <- err + }() + select { + case <-failed: + case <-time.After(3 * time.Second): + t.Fatal("the request on an empty slot did not dial within 3 s: it slept the backoff before it dialed") + } + // A's redial failed and A sleeps 10-20 s. B, on the same slot, must + // not wait for that sleep: it dials at once. + ctxB, cancelB := context.WithTimeout(context.Background(), 3*time.Second) + defer cancelB() + bDone := make(chan error, 1) + start := time.Now() + go func() { + _, err := cl.DoRequest(ctxB, 1) + bDone <- err + }() + select { + case err := <-bDone: + t.Logf("second request while the first sleeps its backoff: %v, err=%v", time.Since(start), err) + if err != nil { + t.Fatalf("the second request failed (%v): it waited for the backoff sleep of the request whose redial failed", err) + } + case <-time.After(5 * time.Second): + t.Fatal("the second request did not return within 5 s: the request whose redial failed sleeps its backoff holding the slot lock, and every worker of the slot waits for it") + } + aReturned := false + select { + case err := <-aDone: + aReturned = true + t.Errorf("the request whose redial failed returned (%v) before its backoff ended", err) + default: + } + cancelA() + if !aReturned { + <-aDone + } + }) + t.Run("grows until a connection serves", func(t *testing.T) { + // Every connection is dialed by DoRequest (reconnectSlot), as in a + // run: while down, the server completes each handshake and drops the + // connection at its first request, unanswered; up, it answers. + var up atomic.Bool + flaky := startRawH2(t, func(c *rawH2Conn, sid uint32, path string) rawH2Action { + if up.Load() { + return respondOK(c, sid, path) + } + return rawClose + }) + fh, fp := flaky.hostPort() + cl, err := newH2Client(fh, fp, "/", testH2Cfg("GET", nil, nil, 1, 1)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(cl.Close) + dials := countDials(cl, 0) + slot := cl.conns[0] + ctx := ctxFor(t, 5*time.Second) + for round := 1; round <= 3; round++ { + // The request on the slot's connection (from round 2 on, one + // DoRequest dialed after the round before): the server drops it. + if _, err := cl.DoRequest(ctx, 0); err == nil { + t.Fatalf("round %d: the request the server dropped unanswered succeeded", round) + } + // The next request finds a connection that never answered. + if _, err := cl.DoRequest(ctx, 0); !errors.Is(err, errH2NeverServed) { + t.Fatalf("round %d: the request that found a connection which never answered got %v, want errH2NeverServed", round, err) + } + want := min(reconnectBackoffMin< time.Second || slot.backoff.next != reconnectBackoffMin<<1 { + t.Errorf("the first backoff after a connection served took %v and left backoff.next=%v: the pace was not reset (want 5-10 ms and %v)", + d, slot.backoff.next, reconnectBackoffMin<<1) + } + }) + t.Run("h2c: only the upgrade's own stream answered", func(t *testing.T) { + // The server answers the upgrade request on stream 1 (RFC 7540 §3.2; + // celeris writes it before it reads the preface) and refuses every + // request of the run. The upgrade's response is a handshake artefact, + // not an answer to a request: the connection has served nothing. + up := startRawH2With(t, rawH2Opts{answerUpgrade: true}, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + c.reset(sid, http2.ErrCodeRefusedStream) + return rawKeep + }) + uh, uport := up.hostPort() + cl, err := newH2CUpgradeClient(uh, uport, "/", testH2Cfg("GET", nil, nil, 1, 1)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(cl.Close) + dials := countDials(cl, 0) + slot := cl.conns[0] + // The refusal arrives after the stream-1 response, which readLoop has + // therefore read. + var re *HTTP2ResetError + if _, err := cl.DoRequest(ctxFor(t, 3*time.Second), 0); !errors.As(err, &re) || re.Code != uint32(http2.ErrCodeRefusedStream) { + t.Fatalf("got %v, want the REFUSED_STREAM reset", err) + } + if slot.cur.Load().served.Load() { + t.Error("the h2c upgrade's own stream-1 response marked the connection served, though no request of the run was answered on it") + } + endConn(slot, "the server dropped the connection") + slot.backoff.next = longBackoff + _, err = cl.DoRequest(ctxFor(t, 300*time.Millisecond), 0) + t.Logf("request after an upgraded connection that answered only stream 1: err=%v, dials=%d", err, dials.Load()) + if n := dials.Load(); n != 0 { + t.Errorf("the connection that answered only the upgrade's stream 1 was redialed at once (%d dial(s)): a server that answers the upgrade and then drops every connection is redialed in a loop", n) + } + }) + // Guard: with no sleep in the way, the redial must still not outlive + // the run. Benchmarker.Run cancels the context before it calls Close, so + // a connection dialed after that would be one Close never sees. + t.Run("run over", func(t *testing.T) { + cl := newClient(t) + ctx, cancel := context.WithCancel(context.Background()) + if _, err := cl.DoRequest(ctx, 0); err != nil { + t.Fatal(err) + } + dials := countDials(cl, 0) + endConn(cl.conns[0], "the server ended the connection") + cancel() + if _, err := cl.DoRequest(ctx, 0); err == nil || dials.Load() != 0 { + t.Fatalf("err=%v dials=%d after the run's context ended: a connection dialed now is one Close, which runs after the cancel, never closes", err, dials.Load()) + } + }) +} + +// countDials routes the client's redials through a counter and fails the +// first failFirst of them, as a refused connection would. +func countDials(cl *h2Client, failFirst int64) *atomic.Int64 { + var n atomic.Int64 + redial := cl.redial + cl.redial = func() (*h2Conn, error) { + if n.Add(1) <= failFirst { + return nil, errors.New("test: connection refused") + } + return redial() + } + return &n +} + +// TestH2ResponseWithoutAReadableStatusIsNotASuccess: a response counts as a +// success only when its :status says so, as an HTTP/1.1 response does only +// with a status line (h1client: "short status line" is an error). A :status +// sent as a literal with a literal name (RFC 7541 §6.2, name index 0) is a +// status like any other; a response with no :status, or a HEADERS or DATA +// frame whose pad length does not fit its payload (RFC 9113 §6.1, §6.2: a +// connection error), is an error. On 2930ded each of these was a success. +func TestH2ResponseWithoutAReadableStatusIsNotASuccess(t *testing.T) { + literalName := func(first byte, name []byte, huffman bool, value string) []byte { + n := byte(len(name)) + if huffman { + n |= 0x80 + } + b := append([]byte{first, n}, name...) + return append(append(b, byte(len(value))), value...) + } + plain, huff := []byte(":status"), hpack.AppendHuffmanString(nil, ":status") + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, path string) rawH2Action { + endAll := http2.FlagHeadersEndHeaders | http2.FlagHeadersEndStream + switch path { + case "/literal-name-404": // without indexing, literal name + _ = c.fr.WriteRawFrame(http2.FrameHeaders, endAll, sid, literalName(0x00, plain, false, "404")) + case "/literal-name-huffman-503": // never indexed, Huffman-coded literal name + _ = c.fr.WriteRawFrame(http2.FrameHeaders, endAll, sid, literalName(0x10, huff, true, "503")) + case "/literal-name-indexed-502": // incremental indexing, literal name + _ = c.fr.WriteRawFrame(http2.FrameHeaders, endAll, sid, literalName(0x40, plain, false, "502")) + case "/no-status": // a field block without :status + c.headers(sid, true, "content-type", "text/plain") + case "/headers-bad-padding": // Pad Length 200 in a 2-byte payload, then :status 200 + _ = c.fr.WriteRawFrame(http2.FrameHeaders, endAll|http2.FlagHeadersPadded, sid, []byte{200, 0x88}) + case "/data-bad-padding": // :status 200, then DATA whose Pad Length 50 exceeds its 11-byte payload + c.headers(sid, false, ":status", "200") + _ = c.fr.WriteRawFrame(http2.FrameData, http2.FlagDataEndStream|http2.FlagDataPadded, sid, append([]byte{50}, make([]byte, 10)...)) + default: + c.ok(sid) + } + return rawKeep + }) + host, port := srv.hostPort() + + for _, tc := range []struct { + path string + status int // 0: a success; -1: an error other than a status + }{ + {"/literal-name-404", 404}, + {"/literal-name-huffman-503", 503}, + {"/literal-name-indexed-502", 502}, + {"/no-status", -1}, + {"/headers-bad-padding", -1}, + {"/data-bad-padding", -1}, + {"/ok", 0}, + } { + n, err := h2Once(t, host, port, tc.path) + t.Logf("%-26s status=%d bytes=%d err=%v", tc.path, h2Status(err), n, err) + if got := h2Status(err); got != tc.status { + t.Errorf("%s: recorded status %d (err=%v), want %d (0: success, -1: an error): a response is a success only when its :status says so", + tc.path, got, err, tc.status) + } + } +} + +// TestH2StreamOutlivedByLaterStreamsGetsItsResponse: a connection keeps each +// stream's response channel in slot (streamID/2) mod 2*MaxStreams, so a +// stream still waiting when 2*MaxStreams-1 later streams have been opened +// shares its slot with the next one. With 2 streams (4 slots), the server +// holds the first stream while a second worker sends 4 more; the 5th stream +// on the connection would take the held stream's slot. The server then +// answers both. Each worker must get its own response. On main (a89f02d) +// and on 2930ded the 5th stream overwrote the held stream's channel: its +// response went to the other worker, and the held stream's worker waited +// for the rest of the run. (Found in round 1, when an h2c benchmark with no +// deadline hung for 11 minutes on a loaded machine.) +func TestH2StreamOutlivedByLaterStreamsGetsItsResponse(t *testing.T) { + firstSeen := make(chan struct{}) + var held uint32 + srv := startRawH2With(t, rawH2Opts{settings: []http2.Setting{{ID: http2.SettingMaxConcurrentStreams, Val: 2}}}, + func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + switch c.requests { + case 1: // hold the first stream + held = sid + close(firstSeen) + case 5: // the stream after 3 more: it would reuse the held stream's slot + c.ok(sid) + c.ok(held) + default: + c.ok(sid) + } + return rawKeep + }) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 2)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + type result struct { + n int + err error + } + first := make(chan result, 1) + go func() { + n, err := cl.DoRequest(ctx, 0) + first <- result{n, err} + }() + select { + case <-firstSeen: + case <-time.After(3 * time.Second): + t.Fatal("the first request never reached the server") + } + for i := range 4 { + if n, err := cl.DoRequest(ctx, 0); err != nil || n != len(rawOKBody) { + t.Fatalf("second worker, request %d: bytes=%d err=%v", i+1, n, err) + } + } + r := <-first + t.Logf("held stream: bytes=%d err=%v (server streams: held=%d)", r.n, r.err, held) + if r.err != nil || r.n != len(rawOKBody) { + t.Errorf("the server answered the held stream, but its worker got bytes=%d err=%v: a later stream took its slot and its response", r.n, r.err) + } +} + +// TestH2RequestsTheConnectionNeverWroteAreRetried: 4 workers share one +// connection with 4 streams and POST 100 KiB bodies; the server starts each +// response, reads the 65,535 bytes the connection window allows and closes +// the connection. The first request's body was being written, so it is in +// flight and is one error. The other three were still queued for the writer: +// they never reached the server, so, like a request still waiting for a +// stream, they go to the redialed connection and are not errors. Before the +// fix their workers woke on the closed connection and gave up before the +// writer answered, and the writer failed the ones it reached: 4 errors per +// connection. (The server starts each response so that its connections have +// served: since round 2 a connection that ends before it answered anything +// costs the request that finds it, which is TestH2RedialBacksOff...'s case.) +func TestH2RequestsTheConnectionNeverWroteAreRetried(t *testing.T) { + srv := startRawH2With(t, rawH2Opts{closeAfterData: 65535}, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + c.headers(sid, false, ":status", "200") // start the response, never end it + return rawKeep + }) + const streams = 4 + res := runBench(t, Config{URL: srv.url("/upload"), Method: "POST", Body: make([]byte, 100<<10), Duration: 400 * time.Millisecond, + Workers: streams, HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: streams}}) + killed := srv.killed.Load() + t.Logf("connections=%d closed by the server=%d requests=%d errors=%d errors per closed connection=%.2f", + srv.accepted.Load(), killed, res.Requests, res.Errors, float64(res.Errors)/float64(max(killed, 1))) + if killed < 3 { + t.Fatalf("the server closed %d connection(s): the client did not keep redialing", killed) + } + if res.Errors > killed { + t.Errorf("errors=%d over %d connections the server closed with one body in flight each: requests the connection never wrote were charged as errors", + res.Errors, killed) + } +} + +// TestH2ClosedConnectionWritesNoNewRequest: once a connection takes no new +// request, its writer writes none of those it still holds: the one it takes +// next, and the ones still queued when it stops. Each is answered +// errH2NotSent, so DoRequest takes it to the redialed connection. +func TestH2ClosedConnectionWritesNoNewRequest(t *testing.T) { + var wire bytes.Buffer + bw := bufio.NewWriter(&wire) + hc := &h2Conn{bufWriter: bw, framer: newH2Framer(bw, nil), writeCh: make(chan h2WriteReq, 4), streamSlots: make([]h2StreamSlot, 4)} + hc.nextStreamID.Store(1) + hc.closed.Store(true) + + chans := make([]chan h2Response, 4) + for i := range chans { + chans[i] = make(chan h2Response, 1) + } + hc.processWriteReq(h2WriteReq{kind: h2WriteHeaders, block: []byte{0x82}, respCh: &chans[0]}) // the one it takes next + for i := 1; i < len(chans); i++ { + hc.writeCh <- h2WriteReq{kind: h2WriteHeaders, block: []byte{0x82}, respCh: &chans[i]} // still queued + } + hc.answerQueued() + if err := bw.Flush(); err != nil { + t.Fatal(err) + } + for i, ch := range chans { + select { + case r := <-ch: + if r.err != errH2NotSent { + t.Errorf("request %d on a closed connection was answered %+v, want errH2NotSent", i, r) + } + default: + t.Errorf("request %d on a closed connection was never answered", i) + } + } + if wire.Len() != 0 { + t.Errorf("the writer wrote %d bytes for requests on a closed connection, want 0", wire.Len()) + } +} + +// TestH2AwaitTakesAnAnswerThatArrivesAfterTheClose: a worker that sees its +// connection close may still be owed an answer: a response readLoop took +// before the close and is delivering, or writeLoop's errH2NotSent for a +// request it never wrote. await waits until both loops have returned +// (loopsDone), and only a request no loop answered is "connection closing". +// Deciding at once made the response an error (N1) and the unsent request +// an error instead of a retry (N2). +func TestH2AwaitTakesAnAnswerThatArrivesAfterTheClose(t *testing.T) { + for _, tc := range []struct { + name string + answer *h2Response // nil: no loop answers + }{ + {"response in delivery", &h2Response{status: 200, bytesRead: 2}}, + {"request never written", &h2Response{err: errH2NotSent}}, + {"no answer", nil}, + } { + t.Run(tc.name, func(t *testing.T) { + hc := &h2Conn{done: make(chan struct{}), loopsDone: make(chan struct{}), streamSem: make(chan struct{}, 1)} + close(hc.done) + ch := make(chan h2Response, 1) + go func() { + time.Sleep(20 * time.Millisecond) // the loops are still at work when the worker sees done + if tc.answer != nil { + ch <- *tc.answer + } + close(hc.loopsDone) + }() + n, err := hc.await(context.Background(), &ch, 0) + <-hc.streamSem // the token await returned + t.Logf("await returned (%d, %v)", n, err) + switch { + case tc.answer == nil: + if err == nil || errors.Is(err, errH2NotSent) { + t.Errorf("no loop answered, await returned (%d, %v), want a connection-closing error", n, err) + } + case tc.answer.err != nil: + if err != errH2NotSent { + t.Errorf("writeLoop answered errH2NotSent after the close, await returned (%d, %v)", n, err) + } + default: + if err != nil || n != 2 { + t.Errorf("readLoop delivered a 200 after the close, await returned (%d, %v), want (2, nil)", n, err) + } + } + }) + } +} + +// TestH2ConnectionLoopsEndAfterClose: await relies on readLoop and writeLoop +// returning soon after a connection closes, and on loopsDone saying so. Close +// a live connection and require loopsDone within 2 s. +func TestH2ConnectionLoopsEndAfterClose(t *testing.T) { + srv := startRawH2(t, respondOK) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 4)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if _, err := cl.DoRequest(ctx, 0); err != nil { + t.Fatal(err) + } + hc := cl.conns[0].cur.Load() + hc.closeConn() + select { + case <-hc.loopsDone: + case <-time.After(2 * time.Second): + t.Fatal("2 s after the connection closed, its readLoop and writeLoop had not both returned: a worker waiting in await for them would wait until its context ends") + } +} + +// --------------------------------------------------------------------------- +// Review of the fixes, round 2 + +// TestH2CUpgradeServerThatAnswersOnlyTheUpgradeIsPaced: an h2c server that +// answers the upgrade request on stream 1, as RFC 7540 §3.2 requires, and +// then closes each connection when the first request of the run arrives, +// without answering it. The stream-1 response must not count as the +// connection having served: every connection ends before it answered a +// request, so the server looks down, and the redials are paced. On e37777e +// the stream-1 response marked each connection served, so each was redialed +// at once: a dial storm, as on a dead port before the backoff existed. +func TestH2CUpgradeServerThatAnswersOnlyTheUpgradeIsPaced(t *testing.T) { + srv := startRawH2With(t, rawH2Opts{answerUpgrade: true}, func(*rawH2Conn, uint32, string) rawH2Action { + return rawClose + }) + res := runBench(t, Config{URL: srv.url("/"), Duration: 300 * time.Millisecond, Workers: 1, + H2CUpgrade: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: 4}}) + accepted := srv.accepted.Load() + t.Logf("connections=%d requests=%d errors=%d connect_errors=%d", accepted, res.Requests, res.Errors, res.ConnectErrors) + if accepted > 60 { + t.Errorf("%d connections in %v to a server that answered no request: they were redialed without backoff", accepted, 300*time.Millisecond) + } + if res.Requests != 0 || res.Errors == 0 { + t.Errorf("requests=%d errors=%d, want 0 and at least 1: the server answered nothing", res.Requests, res.Errors) + } +} + +// TestH2ServerThatDropsEveryConnectionShowsErrors: the server completes each +// handshake, then sends GOAWAY before any request. No request reaches it, so +// none is in flight when a connection ends; each still costs the request that +// finds the connection gone, as h1client charges each request to a server +// that drops its connections, and the pace grows, so the run shows the outage +// as errors and a handful of connections. On e37777e the pace restarted at +// every dial that succeeded (44 and 46 connections in two runs), and whether +// a request was an error or retried silently depended on whether it was +// written before the GOAWAY was read (43 and 24 errors). The server never +// answers a request, so a client that ignored the GOAWAY would wait on its +// first connection for the whole run: 1 connection and no error, which the +// floor below fails. Fixed code, 42 runs: 10-11 connections and 10-17 errors +// in 400 ms. +func TestH2ServerThatDropsEveryConnectionShowsErrors(t *testing.T) { + srv := startRawH2With(t, rawH2Opts{goAwayAtOnce: true}, func(*rawH2Conn, uint32, string) rawH2Action { + return rawKeep + }) + res := runBench(t, Config{URL: srv.url("/"), Duration: 400 * time.Millisecond, Workers: 1, + HTTP2: true, HTTP2Options: HTTP2Options{Connections: 1, MaxStreams: 4}}) + accepted := srv.accepted.Load() + t.Logf("connections=%d requests=%d errors=%d connect_errors=%d", accepted, res.Requests, res.Errors, res.ConnectErrors) + if accepted < 3 || res.Errors == 0 { + t.Errorf("connections=%d errors=%d, want at least 3 and 1: the server drops every connection at once, so the client must keep redialing it and count the outage, not wait on one connection (#89)", + accepted, res.Errors) + } + if res.Errors < accepted-2 { + t.Errorf("errors=%d over %d connections that each ended before answering anything: the outage is not counted", res.Errors, accepted) + } + if accepted > 25 { + t.Errorf("%d connections in %v to a server that drops every one: the backoff does not grow while the server looks down", accepted, 400*time.Millisecond) + } +} + +// TestH2BodyWaitEndsWhenTheServerEndsTheStream: a POST body larger than the +// server's stream window waits for a WINDOW_UPDATE; the server ends the +// stream first and never grants more. RFC 9113 §8.1 lets a server answer +// before the body is complete and then RST_STREAM(NO_ERROR) (x/net's server +// does so when its handler returns without reading the body); a server may +// also refuse the stream. It sends no window for a stream it has ended, so +// the wait must end there, without failing the connection: the later requests +// of the connection are queued behind the writer. A response that ends +// without a reset leaves the client's half of the stream open, which the +// client closes with RST_STREAM(CANCEL); after a reset it sends none (§5.4.2). +// On e37777e the writer polled for window until the run ended, and every +// later request of the connection hung, uncounted. +func TestH2BodyWaitEndsWhenTheServerEndsTheStream(t *testing.T) { + const requests = 3 + for _, tc := range []struct { + name string + answer func(c *rawH2Conn, sid uint32) + success bool // the request is a success (else the reset) + minCancels int64 // RST_STREAM(CANCEL) the client must send, at least + maxResets int64 // RST_STREAM the client may send, at most + }{ + {"response then RST_STREAM(NO_ERROR)", func(c *rawH2Conn, sid uint32) { c.ok(sid); c.reset(sid, http2.ErrCodeNo) }, true, 0, requests}, + {"response only", func(c *rawH2Conn, sid uint32) { c.ok(sid) }, true, requests, requests}, + {"RST_STREAM(REFUSED_STREAM)", func(c *rawH2Conn, sid uint32) { c.reset(sid, http2.ErrCodeRefusedStream) }, false, 0, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + srv := startRawH2With(t, rawH2Opts{settings: uploadWindow}, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + tc.answer(c, sid) + return rawKeep // never grant window + }) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/upload", testH2Cfg("POST", nil, make([]byte, 2000), 1, 4)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + for i := range requests { + _, err := cl.DoRequest(ctx, 0) + var re *HTTP2ResetError + if tc.success && err != nil || !tc.success && !errors.As(err, &re) { + t.Fatalf("request %d: %v (connections=%d): the server ended each stream at once, but the connection's writer still waits for window for a body the server will never take", + i+1, err, srv.accepted.Load()) + } + } + deadline := time.Now().Add(2 * time.Second) + for srv.clientCancel.Load() < tc.minCancels && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + time.Sleep(50 * time.Millisecond) // a reset the client should not send would be flushed by now + t.Logf("connections=%d client RST_STREAM=%d (CANCEL %d)", srv.accepted.Load(), srv.clientResets.Load(), srv.clientCancel.Load()) + if n := srv.accepted.Load(); n != 1 { + t.Errorf("connections=%d, want 1: a stream the server ended is not a connection error", n) + } + if got := srv.clientCancel.Load(); got < tc.minCancels { + t.Errorf("the client sent %d RST_STREAM(CANCEL), want %d: a body it stops sending leaves its half of the stream open on the server", got, tc.minCancels) + } + if got := srv.clientResets.Load(); got > tc.maxResets { + t.Errorf("the client sent %d RST_STREAM, want at most %d: never one for a stream the server reset", got, tc.maxResets) + } + }) + } +} + +// TestH2AwaitAnswersARequestQueuedAfterTheWriterReturned: a worker's hand-off +// to writeCh and the connection's done can be ready together, and select +// picks either, so a request can be queued after writeLoop's final drain +// (answerQueued). Nobody wrote it, so it never reached the server: await, +// once both loops have returned, answers it errH2NotSent (DoRequest takes it +// to the redialed connection) instead of "connection closing", an error. On +// e37777e it was an error. +func TestH2AwaitAnswersARequestQueuedAfterTheWriterReturned(t *testing.T) { + hc := &h2Conn{done: make(chan struct{}), loopsDone: make(chan struct{}), streamSem: make(chan struct{}, 1), writeCh: make(chan h2WriteReq, 4)} + close(hc.done) + close(hc.loopsDone) + ch := make(chan h2Response, 1) + hc.writeCh <- h2WriteReq{kind: h2WriteHeaders, block: []byte{0x82}, respCh: &ch} // queued after the writer returned + n, err := hc.await(context.Background(), &ch, 0) + <-hc.streamSem // the token await returned + t.Logf("await returned (%d, %v)", n, err) + if err != errH2NotSent { + t.Errorf("a request still queued for a writer that has returned was answered (%d, %v), want errH2NotSent: it was never written", n, err) + } + if len(hc.writeCh) != 0 { + t.Errorf("%d request(s) left in writeCh after await", len(hc.writeCh)) + } +} + +// TestH2ResponseWithOnlyAnInterimStatusIsNotASuccess: a stream that ends +// while its only :status is an interim 1xx has no final response, which RFC +// 9113 §8.1 makes malformed: an error (errH2NoStatus), not a success. On +// e37777e both shapes were a success with a 1xx status. +func TestH2ResponseWithOnlyAnInterimStatusIsNotASuccess(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, path string) rawH2Action { + switch path { + case "/interim-ends-stream": // HEADERS 103 with END_STREAM + c.headers(sid, true, ":status", "103") + case "/interim-then-data": // HEADERS 100, then the stream ends on DATA + c.headers(sid, false, ":status", "100") + c.data(sid, true, len(rawOKBody), 0) + case "/interim-then-final": // HEADERS 103, then the final 200: a success + c.headers(sid, false, ":status", "103") + c.ok(sid) + default: + c.ok(sid) + } + return rawKeep + }) + host, port := srv.hostPort() + for _, tc := range []struct { + path string + status int // 0: a success; -1: an error other than a status + }{ + {"/interim-ends-stream", -1}, + {"/interim-then-data", -1}, + {"/interim-then-final", 0}, + {"/ok", 0}, + } { + n, err := h2Once(t, host, port, tc.path) + t.Logf("%-22s status=%d bytes=%d err=%v", tc.path, h2Status(err), n, err) + if got := h2Status(err); got != tc.status { + t.Errorf("%s: recorded status %d (err=%v), want %d (0: success, -1: an error): a response with no final status is not a success", + tc.path, got, err, tc.status) + } + } +} + +// TestH2HeadersTooShortForTheirPriorityFieldsEndTheConnection: a HEADERS +// frame flagged PRIORITY whose payload has no room for the 5 bytes of +// priority fields is a frame size error on a frame that carries a field +// block, so a connection error (RFC 9113 §4.2, §6.2), as a Pad Length that +// does not fit is since round 1. On e37777e it was read as an empty block: +// the stream failed with errH2NoStatus and the connection went on. The +// boundary is exact: 4 bytes are too short, while 5 bytes are the priority +// fields and an empty field block, which is only a response without a +// :status (a stream error), and 6 carry :status 200. +func TestH2HeadersTooShortForTheirPriorityFieldsEndTheConnection(t *testing.T) { + priority := []byte{0, 0, 0, 0, 15} // stream dependency 0, weight 16 + for _, tc := range []struct { + name string + payload []byte + connError bool // a connection error; else the connection goes on + success bool + }{ + {"1 byte: :status 200 where the priority fields should be", []byte{0x88}, true, false}, + {"4 bytes", priority[:4], true, false}, + {"5 bytes: the priority fields, an empty field block", priority, false, false}, + {"6 bytes: the priority fields, :status 200", append(append([]byte{}, priority...), 0x88), false, true}, + } { + t.Run(tc.name, func(t *testing.T) { + srv := startRawH2(t, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + _ = c.fr.WriteRawFrame(http2.FrameHeaders, http2.FlagHeadersEndHeaders|http2.FlagHeadersEndStream|http2.FlagHeadersPriority, sid, tc.payload) + return rawKeep + }) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 1)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + hc := cl.conns[0].cur.Load() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _, err = cl.DoRequest(ctx, 0) + t.Logf("%d-byte payload: err=%v connection closed=%t", len(tc.payload), err, hc.closed.Load()) + if tc.success != (err == nil) { + t.Errorf("err=%v, want a success=%t", err, tc.success) + } + if hc.closed.Load() != tc.connError { + t.Errorf("connection closed=%t after a HEADERS frame flagged PRIORITY with a %d-byte payload (err=%v), want %t: only fewer than 5 bytes are a connection error", + hc.closed.Load(), len(tc.payload), err, tc.connError) + } + }) + } +} + +// --------------------------------------------------------------------------- +// Review of the fixes, round 3 (at 147edb4) + +// TestH2BodyStopsAtTheServersReset: once the server resets a stream, the +// client sends no more frames on it (RFC 9113 §6.4, §5.1 "closed"), however +// much send window the body still has. The server grants 128 MiB of window, +// resets a 32 MiB upload at its HEADERS and reads nothing for 300 ms, so the +// client's writer can be ahead of the reset by no more than the socket +// buffers. The second request's HEADERS reach the server after every DATA +// frame the client wrote for the first stream: the server counts those at +// that moment. On 147edb4 the writer checked for the reset only once the +// window ran out, so it sent the whole body into a stream that no longer +// existed (a Go server answers each such frame with RST_STREAM(STREAM_CLOSED)). +func TestH2BodyStopsAtTheServersReset(t *testing.T) { + const body = 32 << 20 + var afterFirst atomic.Int64 // DATA bytes of the first stream the server read before the second stream's HEADERS + srv := startRawH2With(t, rawH2Opts{ + settings: []http2.Setting{{ID: http2.SettingInitialWindowSize, Val: 128 << 20}}, + connWindow: 128 << 20, + }, func(c *rawH2Conn, sid uint32, _ string) rawH2Action { + if c.requests == 2 { + afterFirst.Store(c.srv.dataAfterReset.Load()) + } + c.reset(sid, http2.ErrCodeRefusedStream) + if c.requests == 1 { + time.Sleep(300 * time.Millisecond) // let the client read the reset while its writer is still in the body + } + return rawKeep + }) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/upload", testH2Cfg("POST", nil, make([]byte, body), 1, 4)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + for i := range 2 { + var re *HTTP2ResetError + if _, err := cl.DoRequest(ctx, 0); !errors.As(err, &re) { + t.Fatalf("request %d: got %v, want the server's RST_STREAM(REFUSED_STREAM)", i+1, err) + } + } + sent := afterFirst.Load() + t.Logf("DATA bytes the client sent on the first stream after the server reset it: %d of the %d-byte body; client RST_STREAM=%d connections=%d", + sent, body, srv.clientResets.Load(), srv.accepted.Load()) + if sent >= body/2 { + t.Errorf("the client sent %d of the body's %d bytes on a stream the server had reset: after RST_STREAM it must send no more frames on it (RFC 9113 §6.4)", sent, body) + } + if n := srv.clientResets.Load(); n != 0 { + t.Errorf("the client sent %d RST_STREAM: never one for a stream the server reset (RFC 9113 §5.4.2)", n) + } + if n := srv.accepted.Load(); n != 1 { + t.Errorf("connections=%d, want 1: a stream the server reset is not a connection error", n) + } +} + +// TestH2ConnectionOutOfStreamIDsIsFailedWhenItDies: a connection whose stream +// IDs have run out takes no new request (closed), but its last streams are +// still in flight. If the server then closes it, or a write to it fails, the +// connection has died: its streams must fail and done must close, as on any +// other connection. No request comes to redial it here (the requests are +// handed to the writer directly), so nothing else closes it. On 147edb4 +// readLoop and writeFailed took closed for the client's own close and +// returned without failing it: the stream in flight waited for the rest of +// the run. +func TestH2ConnectionOutOfStreamIDsIsFailedWhenItDies(t *testing.T) { + for _, tc := range []struct { + name string + failWrites bool // a write fails; else the server closes the connection + }{ + {"the server closes it", false}, + {"a write to it fails", true}, + } { + t.Run(tc.name, func(t *testing.T) { + if tc.failWrites { + dial := dialTimeoutFunc + t.Cleanup(func() { dialTimeoutFunc = dial }) + dialTimeoutFunc = func(network, addr string, timeout time.Duration) (net.Conn, error) { + c, err := dial(network, addr, timeout) + if err != nil { + return nil, err + } + return &failWritesConn{Conn: c, okWrites: 4}, nil // the handshake's 3 writes and the last stream's HEADERS + } + } + held := make(chan *rawH2Conn, 1) + srv := startRawH2(t, func(c *rawH2Conn, _ uint32, _ string) rawH2Action { + held <- c // never answer + return rawKeep + }) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 4)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(cl.Close) + hc := cl.conns[0].cur.Load() + hc.nextStreamID.Store(0x7FFFFFFF) // the last stream ID a connection may use + last := make(chan h2Response, 1) + hc.writeCh <- h2WriteReq{kind: h2WriteHeaders, block: cl.headerBlock, respCh: &last} + var sc *rawH2Conn + select { + case sc = <-held: + case <-time.After(3 * time.Second): + t.Fatal("the stream with the last ID never reached the server") + } + next := make(chan h2Response, 1) + hc.writeCh <- h2WriteReq{kind: h2WriteHeaders, block: cl.headerBlock, respCh: &next} + select { + case r := <-next: + if r.err != errH2NotSent { + t.Fatalf("the request after the last stream ID was answered %+v, want errH2NotSent", r) + } + case <-time.After(3 * time.Second): + t.Fatal("the request after the last stream ID was never answered") + } + if !hc.closed.Load() { + t.Fatal("a connection out of stream IDs still takes new requests") + } + if tc.failWrites { + _ = sc.fr.WritePing(false, [8]byte{1}) // the client's PING ACK is its next write, which fails + } else { + _ = sc.nc.Close() + } + select { + case r := <-last: + t.Logf("the stream in flight was answered: %v", r.err) + if r.err == nil { + t.Errorf("the stream in flight on a connection that died was a success: %+v", r) + } + case <-time.After(3 * time.Second): + t.Fatal("the stream in flight was never answered after its connection died: a connection out of stream IDs was taken for one the client closed, and nothing failed it") + } + select { + case <-hc.done: + case <-time.After(time.Second): + t.Error("done never closed on a connection that died") + } + }) + } +} + +// heldReadConn holds each read, once armed, after the bytes are read from +// the socket and before the reader gets them, until release is closed: the +// reader then parses a response that had arrived before the connection +// ended. +type heldReadConn struct { + net.Conn + armed atomic.Bool + holding chan struct{} // closed when an armed read holds bytes + once sync.Once + release chan struct{} +} + +func (c *heldReadConn) Read(p []byte) (int, error) { + n, err := c.Conn.Read(p) + if n > 0 && c.armed.Load() { + c.once.Do(func() { close(c.holding) }) + <-c.release + } + return n, err +} + +// TestH2RedialTakesTheResponseTheConnectionHadReceived: the writer ends a +// connection (a failed write: failConn) just after the server's first +// response reached the client, and before readLoop has parsed it. That +// connection has served, so the request that finds it gone redials at once. +// reconnectSlot must read served only once readLoop has returned, having +// parsed what it had read. On 147edb4 it read served at once: the connection +// looked as if it had never served, so the request failed (errH2NeverServed) +// and slept the backoff, and nothing was dialed. +func TestH2RedialTakesTheResponseTheConnectionHadReceived(t *testing.T) { + dial := dialTimeoutFunc + t.Cleanup(func() { dialTimeoutFunc = dial }) + gate := &heldReadConn{holding: make(chan struct{}), release: make(chan struct{})} + var dialed atomic.Int64 + dialTimeoutFunc = func(network, addr string, timeout time.Duration) (net.Conn, error) { + c, err := dial(network, addr, timeout) + if err != nil || dialed.Add(1) > 1 { + return c, err + } + gate.Conn = c + return gate, nil + } + srv := startRawH2(t, respondOK) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 4)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + hc := cl.conns[0].cur.Load() + gate.armed.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + first := make(chan error, 1) + go func() { + _, err := cl.DoRequest(ctx, 0) + first <- err + }() + select { + case <-gate.holding: + case <-time.After(3 * time.Second): + close(gate.release) + t.Fatal("the server's first response never reached the client") + } + hc.failConn(errors.New("test: a write to the connection failed")) // as writeFailed does + t.Logf("the request whose response was still unparsed: %v", <-first) + go func() { + time.Sleep(100 * time.Millisecond) + close(gate.release) + }() + dials := countDials(cl, 0) + start := time.Now() + _, err = cl.DoRequest(ctx, 0) + t.Logf("the request that found the connection gone: %v, err=%v, dials=%d, served=%t", time.Since(start), err, dials.Load(), hc.served.Load()) + if err != nil || dials.Load() != 1 { + t.Errorf("err=%v dials=%d, want a success after 1 dial: the connection had received the answer to a request, so it served, and the request that finds it gone redials at once", err, dials.Load()) + } +} + +// endsAfterFirstCheck is a context that ends as its first Err returns: the +// run's context ending just after a check of it. That first Err closes Done +// and returns nil; every later Err returns context.Canceled. +type endsAfterFirstCheck struct { + context.Context + done chan struct{} + checked atomic.Bool +} + +func newEndsAfterFirstCheck() *endsAfterFirstCheck { + return &endsAfterFirstCheck{Context: context.Background(), done: make(chan struct{})} +} + +func (c *endsAfterFirstCheck) Done() <-chan struct{} { return c.done } + +func (c *endsAfterFirstCheck) Err() error { + if c.checked.CompareAndSwap(false, true) { + close(c.done) + return nil + } + return context.Canceled +} + +// TestH2RedialDialsNothingOnceTheRunHasEnded: once the run's context has +// ended, reconnectSlot dials nothing, as its doc says. Close runs after the +// cancel and would not close a connection dialed then, and a dial that fails +// then is a connect error in the Result of a run that had already ended +// (bench.go takes snapshotConnectErrors after the workers return). The +// context can end while reconnectSlot waits for the old connection's loops +// (readLoop still parsing what it had read), or after reconnectSlot's first +// check and before that wait, when the loops may already have returned: the +// wait's two cases are then both ready and select takes either. On 0af01b3 +// the wait fell through on the cancel and dialed: 1 dial, and 1 connect error +// when the server was down. +func TestH2RedialDialsNothingOnceTheRunHasEnded(t *testing.T) { + for _, down := range []bool{false, true} { + name := "the run ends during the wait for the loops, server up" + if down { + name = "the run ends during the wait for the loops, server down" + } + t.Run(name, func(t *testing.T) { + dial := dialTimeoutFunc + t.Cleanup(func() { dialTimeoutFunc = dial }) + gate := &heldReadConn{holding: make(chan struct{}), release: make(chan struct{})} + var dialed atomic.Int64 + dialTimeoutFunc = func(network, addr string, timeout time.Duration) (net.Conn, error) { + c, err := dial(network, addr, timeout) + if err != nil || dialed.Add(1) > 1 { + return c, err + } + gate.Conn = c + return gate, nil + } + srv := startRawH2(t, respondOK) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 4)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + var released sync.Once + release := func() { released.Do(func() { close(gate.release) }) } + defer release() + live, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + hc := cl.conns[0].cur.Load() + if _, err := cl.DoRequest(live, 0); err != nil || !hc.served.Load() { + t.Fatalf("first request: err=%v served=%t, want a success", err, hc.served.Load()) + } + // readLoop holds the second response's bytes until release, so + // the connection's loops outlast its end. + gate.armed.Store(true) + go func() { _, _ = cl.DoRequest(live, 1) }() + select { + case <-gate.holding: + case <-time.After(3 * time.Second): + t.Fatal("the server's second response never reached the client") + } + hc.failConn(errors.New("test: a write to the connection failed")) // as writeFailed does + if down { + _ = srv.ln.Close() + } + dials := countDials(cl, 0) + before := connectErrorsCounter.Swap(0) + defer connectErrorsCounter.Add(before) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + time.AfterFunc(50*time.Millisecond, cancel) // the run ends + _, err = cl.DoRequest(ctx, 0) + connectErrors := connectErrorsCounter.Swap(0) + t.Logf("the request that found the connection gone as the run ended: err=%v dials=%d connect_errors=%d", err, dials.Load(), connectErrors) + if !errors.Is(err, context.Canceled) || dials.Load() != 0 || connectErrors != 0 { + t.Errorf("err=%v dials=%d connect_errors=%d, want context.Canceled, 0 and 0: the run ended while reconnectSlot waited for the old connection's loops, and a slot dials nothing after the run's end", + err, dials.Load(), connectErrors) + } + }) + } + + t.Run("the run ends just before the wait, the loops already returned", func(t *testing.T) { + srv := startRawH2(t, respondOK) + host, port := srv.hostPort() + cl, err := newH2Client(host, port, "/", testH2Cfg("GET", nil, nil, 1, 4)) + if err != nil { + t.Fatal(err) + } + defer cl.Close() + live, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + slot := cl.conns[0] + hc := slot.cur.Load() + if _, err := cl.DoRequest(live, 0); err != nil || !hc.served.Load() { + t.Fatalf("first request: err=%v served=%t, want a success", err, hc.served.Load()) + } + hc.failConn(errors.New("test: a write to the connection failed")) + select { + case <-hc.loopsDone: + case <-time.After(3 * time.Second): + t.Fatal("the connection's loops never returned after it ended") + } + dials := countDials(cl, 0) + // Both of the wait's cases are ready and select takes either at + // random, so a check made in only one of them dials half the time. + const attempts = 64 + for i := range attempts { + if _, err := cl.reconnectSlot(newEndsAfterFirstCheck(), slot); !errors.Is(err, context.Canceled) || dials.Load() != 0 { + t.Fatalf("attempt %d of %d: err=%v dials=%d, want context.Canceled and no dial: the run ended after reconnectSlot's first check of it, and a slot dials nothing after the run's end", + i+1, attempts, err, dials.Load()) + } + } + }) +} + +// TestH2AwaitAnswersEachQueuedRequestOnce: requests still queued for a +// connection whose writer has returned are answered by whichever await runs +// answerQueued first; several awaits on one dead connection drain the queue +// together. An await that finds the queue empty and its own channel empty +// must not decide that nobody answered it while another await has taken its +// request off the queue and not yet answered it: that request never reached +// the server, so it is errH2NotSent (retried), not "connection closing" (an +// error). 8 awaits per dead connection, 20000 connections. On +// 147edb4 a probe of this shape counted 913 of 160,000 such requests +// answered as errors under -race, and 99-104 without. +func TestH2AwaitAnswersEachQueuedRequestOnce(t *testing.T) { + const trials, waiters = 20000, 8 + var miscounted, total int64 + for range trials { + hc := &h2Conn{done: make(chan struct{}), loopsDone: make(chan struct{}), streamSem: make(chan struct{}, waiters), + writeCh: make(chan h2WriteReq, waiters)} + close(hc.done) + close(hc.loopsDone) + chans := make([]chan h2Response, waiters) + for i := range chans { + chans[i] = make(chan h2Response, 1) + hc.writeCh <- h2WriteReq{kind: h2WriteHeaders, block: []byte{0x82}, respCh: &chans[i]} + } + var wg sync.WaitGroup + var bad atomic.Int64 + start := make(chan struct{}) + for i := range chans { + wg.Add(1) + go func() { + defer wg.Done() + <-start + if _, err := hc.await(context.Background(), &chans[i], 0); err != errH2NotSent { + bad.Add(1) + } + }() + } + close(start) + wg.Wait() + miscounted += bad.Load() + total += waiters + } + t.Logf("requests never written, answered as errors: %d of %d", miscounted, total) + if miscounted != 0 { + t.Errorf("%d of %d requests that were never written were answered \"connection closing\": an await gave up while another await held its request, taken off the queue and not yet answered", miscounted, total) + } +} diff --git a/h2framer.go b/h2framer.go index f006358..5f5a58c 100644 --- a/h2framer.go +++ b/h2framer.go @@ -22,6 +22,13 @@ const ( flagEndStream = 0x1 flagACK = 0x1 flagEndHeaders = 0x4 + flagPadded = 0x8 // DATA, HEADERS: Pad Length byte first, that many padding bytes last + flagPriority = 0x20 // HEADERS: 5 bytes of stream dependency and weight before the block +) + +// H2 error codes (RFC 9113 §7) the client sends. +const ( + h2ErrCodeCancel = 0x8 ) // H2 settings IDs. @@ -51,11 +58,52 @@ func (f *h2Frame) StreamEnded() bool { return f.Flags&flagEndStream != 0 } // IsAck reports whether the ACK flag is set (for SETTINGS and PING). func (f *h2Frame) IsAck() bool { return f.Flags&flagACK != 0 } -// HeaderBlockFragment returns the header block fragment from a HEADERS frame. -func (f *h2Frame) HeaderBlockFragment() []byte { return f.payload } +// HeaderBlockFragment returns the header block fragment from a HEADERS frame, +// without the padding and priority fields a server may add (RFC 9113 §6.2). +func (f *h2Frame) HeaderBlockFragment() []byte { + p := f.unpadded() + if f.Flags&flagPriority != 0 { + if len(p) < 5 { + return nil + } + p = p[5:] + } + return p +} + +// unpadded returns the payload of a DATA or HEADERS frame without its Pad +// Length field and padding. A pad length that does not fit the payload is a +// PROTOCOL_ERROR (RFC 9113 §6.1); nothing in such a payload is usable. +func (f *h2Frame) unpadded() []byte { + p := f.payload + if f.Flags&flagPadded == 0 { + return p + } + if len(p) == 0 || int(p[0]) >= len(p) { + return nil + } + return p[1 : len(p)-int(p[0])] +} + +// badPadding reports a DATA or HEADERS frame whose Pad Length does not fit +// its payload: a connection error of type PROTOCOL_ERROR (RFC 9113 §6.1, +// §6.2). Nothing in such a frame can be read (unpadded returns nil). +func (f *h2Frame) badPadding() bool { + return (f.Type == frameData || f.Type == frameHeaders) && f.Flags&flagPadded != 0 && + (len(f.payload) == 0 || int(f.payload[0]) >= len(f.payload)) +} + +// badPriority reports a HEADERS frame flagged PRIORITY whose payload, without +// its padding, is shorter than the 5 bytes of priority fields: a frame size +// error on a frame that carries a field block, so a connection error (RFC 9113 +// §4.2, §6.2). Check badPadding first. +func (f *h2Frame) badPriority() bool { + return f.Type == frameHeaders && f.Flags&flagPriority != 0 && len(f.unpadded()) < 5 +} -// Data returns the data payload from a DATA frame. -func (f *h2Frame) Data() []byte { return f.payload } +// Data returns the data of a DATA frame, without its padding. Flow control +// counts the whole payload, padding included (RFC 9113 §6.9.1): use Length. +func (f *h2Frame) Data() []byte { return f.unpadded() } // ErrCode returns the error code from a RST_STREAM frame (payload[0:4]). func (f *h2Frame) ErrCode() uint32 { @@ -259,6 +307,19 @@ func (fr *h2Framer) WriteWindowUpdate(streamID, incr uint32) error { return err } +// WriteRSTStream writes a RST_STREAM frame. Single 13-byte write via wbuf. +func (fr *h2Framer) WriteRSTStream(streamID, errCode uint32) error { + fr.wbuf[0] = 0 // length = 4 + fr.wbuf[1] = 0 + fr.wbuf[2] = 4 + fr.wbuf[3] = frameRSTStream + fr.wbuf[4] = 0 + binary.BigEndian.PutUint32(fr.wbuf[5:9], streamID) + binary.BigEndian.PutUint32(fr.wbuf[9:13], errCode) + _, err := fr.bw.Write(fr.wbuf[:13]) + return err +} + // WriteGoAway writes a GOAWAY frame. func (fr *h2Framer) WriteGoAway(lastStreamID, errCode uint32, debugData []byte) error { if len(debugData) == 0 {