From 8ab32de8e1774047d49c5cf43a2401ae9d8d357b Mon Sep 17 00:00:00 2001 From: Rafael Garcia Date: Sat, 8 Aug 2026 12:32:12 -0400 Subject: [PATCH 1/3] http2: stop over-crediting stream WINDOW_UPDATE by buffered body bytes transportResponseBody.Read computed the stream receive-window refresh as unsent = streamFlow - available + bufPipe.Len() bufPipe.Len() is body data received but not yet consumed by the application. The amount that is safe to return to the peer is streamFlow - available - buffered (what golang.org/x/net/http2 computes), so the buffered term must be subtracted. Adding it over-credits the stream window by 2x buffered on every refresh. With a slow reader (proxy relaying to a backpressured client, rate- limited download), the receive buffer stays large, the advertised stream window desyncs from the real connection accounting, and a large download dies partway with: stream error: stream ID N; FLOW_CONTROL_ERROR Add a regression test that downloads 48 MiB over an in-process server through a Transport configured with a Chrome-like 6 MiB stream window while consuming the body at ~12 MiB/s. It fails on master with FLOW_CONTROL_ERROR after ~46 MiB and passes with this fix. Refs https://github.com/bogdanfinn/tls-client/issues/257 --- http2/transport.go | 2 +- http2/transport_flow_test.go | 100 +++++++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) create mode 100644 http2/transport_flow_test.go diff --git a/http2/transport.go b/http2/transport.go index 594516af..cafcb92e 100644 --- a/http2/transport.go +++ b/http2/transport.go @@ -2456,7 +2456,7 @@ func (b transportResponseBody) Read(p []byte) (n int, err error) { // stream. // Use dynamic streamFlow logic - unsent := int(cc.streamFlow) - int(cs.inflow.available()) + cs.bufPipe.Len() + unsent := int(cc.streamFlow) - int(cs.inflow.available()) - cs.bufPipe.Len() // ------------------------------------------------------------------ // FIX: Adaptive Logic diff --git a/http2/transport_flow_test.go b/http2/transport_flow_test.go new file mode 100644 index 00000000..0917f9c5 --- /dev/null +++ b/http2/transport_flow_test.go @@ -0,0 +1,100 @@ +// Copyright 2026 The fhttp Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package http2 + +import ( + "fmt" + "io" + "net" + "testing" + "time" + + http "github.com/bogdanfinn/fhttp" +) + +// throttledSink caps the rate at which a response body is consumed, simulating +// a slow reader: a proxy relaying to a backpressured client, a rate-limited +// download, etc. +type throttledSink struct{ bytesPerSecond int } + +func (s throttledSink) Write(p []byte) (int, error) { + time.Sleep(time.Duration(len(p)) * time.Second / time.Duration(s.bytesPerSecond)) + return len(p), nil +} + +// TestTransportSlowReaderLargeResponse verifies that a response body much +// larger than the stream and connection flow-control windows is delivered in +// full when the application consumes it slowly. +// +// Regression test for a WINDOW_UPDATE accounting bug: Read credited the peer +// for buffered-but-unread body bytes (adding cs.bufPipe.Len() where it should +// subtract it), so the advertised stream window desynced from the real one and +// the transfer died partway with "stream error: ...; FLOW_CONTROL_ERROR". +func TestTransportSlowReaderLargeResponse(t *testing.T) { + const ( + bodySize = 48 << 20 // 48 MiB, ~3x the 15.6 MiB connection window + bytesPerSecond = 12 << 20 + ) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + s := &Server{} + s.ServeConn(conn, &ServeConnOpts{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", fmt.Sprint(bodySize)) + w.WriteHeader(200) + chunk := make([]byte, 1<<20) + for remain := bodySize; remain > 0; remain -= len(chunk) { + if _, err := w.Write(chunk); err != nil { + return + } + } + })}) + }() + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + tr := &Transport{ + Settings: map[SettingID]uint32{ + SettingInitialWindowSize: 6291456, + }, + SettingsOrder: []SettingID{SettingInitialWindowSize}, + ConnectionFlow: 15663105, + PseudoHeaderOrder: []string{":method", ":authority", ":scheme", ":path"}, + } + cc, err := tr.NewClientConn(conn) + if err != nil { + t.Fatal(err) + } + defer cc.Close() + + req, err := http.NewRequest("GET", "https://example.test/", nil) + if err != nil { + t.Fatal(err) + } + resp, err := cc.RoundTrip(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + n, err := io.Copy(throttledSink{bytesPerSecond: bytesPerSecond}, resp.Body) + if err != nil { + t.Fatalf("slow read of %d-byte response failed after %d bytes: %v", bodySize, n, err) + } + if n != bodySize { + t.Fatalf("read %d bytes, want %d", n, bodySize) + } +} From 75a3b989a47286a68544afb797084580c7692cf6 Mon Sep 17 00:00:00 2001 From: Rafael Garcia Date: Sat, 8 Aug 2026 12:37:58 -0400 Subject: [PATCH 2/3] module: rename to github.com/kernel/fhttp So Kernel services can pin this fork via a go.mod replace while the upstream WINDOW_UPDATE fix (bogdanfinn/fhttp#24) is pending review. Import-path-only change; no code changes beyond the preceding fix. --- alpn_test.go | 4 ++-- cgi/child.go | 2 +- cgi/child_test.go | 2 +- cgi/host.go | 2 +- cgi/host_test.go | 4 ++-- cgi/integration_test.go | 6 +++--- client_test.go | 6 +++--- clientserver_test.go | 6 +++--- cookiejar/dummy_publicsuffix_test.go | 2 +- cookiejar/example_test.go | 6 +++--- cookiejar/jar.go | 2 +- cookiejar/jar_test.go | 2 +- example_client_test.go | 4 ++-- example_filesystem_test.go | 2 +- example_handle_test.go | 2 +- example_test.go | 2 +- fcgi/child.go | 4 ++-- fcgi/fcgi_test.go | 2 +- fs_test.go | 4 ++-- go.mod | 2 +- h2_bundle.go | 6 +++--- header.go | 2 +- header_test.go | 2 +- http.go | 2 +- http2/client_conn_pool.go | 2 +- http2/fhttp_test.go | 8 ++++---- http2/frame.go | 2 +- http2/frame_test.go | 2 +- http2/go111.go | 2 +- http2/h2c/h2c.go | 6 +++--- http2/h2c/h2c_test.go | 4 ++-- http2/h2i/h2i.go | 4 ++-- http2/header_order_test.go | 4 ++-- http2/headermap.go | 2 +- http2/http2.go | 2 +- http2/http2_test.go | 4 ++-- http2/not_go111.go | 2 +- http2/push_consume.go | 2 +- http2/push_consume_test.go | 4 ++-- http2/server.go | 4 ++-- http2/server_push_test.go | 2 +- http2/server_test.go | 4 ++-- http2/transport.go | 6 +++--- http2/transport_flow_test.go | 2 +- http2/transport_test.go | 8 ++++---- http2/write.go | 4 ++-- http_test.go | 2 +- httptest/example_test.go | 4 ++-- httptest/httptest.go | 2 +- httptest/httptest_test.go | 2 +- httptest/recorder.go | 2 +- httptest/recorder_test.go | 2 +- httptest/server.go | 4 ++-- httptest/server_test.go | 2 +- httptrace/example_test.go | 2 +- httptrace/trace.go | 2 +- httputil/dump.go | 2 +- httputil/dump_test.go | 2 +- httputil/example_test.go | 6 +++--- httputil/httputil.go | 2 +- httputil/persist.go | 2 +- httputil/reverseproxy.go | 2 +- httputil/reverseproxy_test.go | 4 ++-- internal/testenv/testenv.go | 2 +- main_test.go | 2 +- pprof/pprof.go | 4 ++-- pprof/pprof_test.go | 6 +++--- request.go | 2 +- request_test.go | 4 ++-- response_test.go | 2 +- serve_test.go | 10 +++++----- sniff_test.go | 2 +- transfer.go | 4 ++-- transport.go | 2 +- transport_internal_test.go | 2 +- transport_test.go | 12 ++++++------ triv.go | 2 +- 77 files changed, 129 insertions(+), 129 deletions(-) diff --git a/alpn_test.go b/alpn_test.go index bab6a657..2cfad7f2 100644 --- a/alpn_test.go +++ b/alpn_test.go @@ -15,8 +15,8 @@ import ( tls "github.com/bogdanfinn/utls" - . "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" + . "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" ) func TestNextProtoUpgrade(t *testing.T) { diff --git a/cgi/child.go b/cgi/child.go index d71dd0cb..39f9cb69 100644 --- a/cgi/child.go +++ b/cgi/child.go @@ -20,7 +20,7 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // Request returns the HTTP request as represented in the current diff --git a/cgi/child_test.go b/cgi/child_test.go index 42321d19..816be5a2 100644 --- a/cgi/child_test.go +++ b/cgi/child_test.go @@ -13,7 +13,7 @@ import ( "strings" "testing" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) func TestRequest(t *testing.T) { diff --git a/cgi/host.go b/cgi/host.go index 33d515cf..3afe08df 100644 --- a/cgi/host.go +++ b/cgi/host.go @@ -29,7 +29,7 @@ import ( "strconv" "strings" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" "golang.org/x/net/http/httpguts" ) diff --git a/cgi/host_test.go b/cgi/host_test.go index a7879e6b..902a2c22 100644 --- a/cgi/host_test.go +++ b/cgi/host_test.go @@ -22,8 +22,8 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" ) func newRequest(httpreq string) *http.Request { diff --git a/cgi/integration_test.go b/cgi/integration_test.go index e1c3b02c..d6689e0b 100644 --- a/cgi/integration_test.go +++ b/cgi/integration_test.go @@ -19,9 +19,9 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" - "github.com/bogdanfinn/fhttp/internal/testenv" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" + "github.com/kernel/fhttp/internal/testenv" ) // This test is a CGI host (testing host.go) that runs its own binary diff --git a/client_test.go b/client_test.go index c274acf4..65d0d7b3 100644 --- a/client_test.go +++ b/client_test.go @@ -26,9 +26,9 @@ import ( tls "github.com/bogdanfinn/utls" - . "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/cookiejar" - "github.com/bogdanfinn/fhttp/httptest" + . "github.com/kernel/fhttp" + "github.com/kernel/fhttp/cookiejar" + "github.com/kernel/fhttp/httptest" ) var robotsTxtHandler = HandlerFunc(func(w ResponseWriter, r *Request) { diff --git a/clientserver_test.go b/clientserver_test.go index 1ecc9f1e..3ffc2999 100644 --- a/clientserver_test.go +++ b/clientserver_test.go @@ -29,9 +29,9 @@ import ( tls "github.com/bogdanfinn/utls" - . "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" - "github.com/bogdanfinn/fhttp/httputil" + . "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" + "github.com/kernel/fhttp/httputil" ) type clientServerTest struct { diff --git a/cookiejar/dummy_publicsuffix_test.go b/cookiejar/dummy_publicsuffix_test.go index 9c405e97..a54a4c3c 100644 --- a/cookiejar/dummy_publicsuffix_test.go +++ b/cookiejar/dummy_publicsuffix_test.go @@ -4,7 +4,7 @@ package cookiejar_test -import "github.com/bogdanfinn/fhttp/cookiejar" +import "github.com/kernel/fhttp/cookiejar" type dummypsl struct { List cookiejar.PublicSuffixList diff --git a/cookiejar/example_test.go b/cookiejar/example_test.go index 74da2b55..c43751f4 100644 --- a/cookiejar/example_test.go +++ b/cookiejar/example_test.go @@ -9,9 +9,9 @@ import ( "log" "net/url" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/cookiejar" - "github.com/bogdanfinn/fhttp/httptest" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/cookiejar" + "github.com/kernel/fhttp/httptest" ) func ExampleNew() { diff --git a/cookiejar/jar.go b/cookiejar/jar.go index 9b7dd220..5cf4e95e 100644 --- a/cookiejar/jar.go +++ b/cookiejar/jar.go @@ -15,7 +15,7 @@ import ( "sync" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // PublicSuffixList provides the public suffix of a domain. For example: diff --git a/cookiejar/jar_test.go b/cookiejar/jar_test.go index f5d3f159..7e899f36 100644 --- a/cookiejar/jar_test.go +++ b/cookiejar/jar_test.go @@ -12,7 +12,7 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // tNow is the synthetic current time used as now during testing. diff --git a/example_client_test.go b/example_client_test.go index 04bd35a8..42bce166 100644 --- a/example_client_test.go +++ b/example_client_test.go @@ -13,8 +13,8 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/http2" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2" ) // Basic http test with Header Order + enable push diff --git a/example_filesystem_test.go b/example_filesystem_test.go index 06bb6b46..cd50cb98 100644 --- a/example_filesystem_test.go +++ b/example_filesystem_test.go @@ -9,7 +9,7 @@ import ( "log" "strings" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // containsDotFile reports whether name contains a path element starting with a period. diff --git a/example_handle_test.go b/example_handle_test.go index 6e5d1d2d..b6a469a8 100644 --- a/example_handle_test.go +++ b/example_handle_test.go @@ -9,7 +9,7 @@ import ( "log" "sync" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) type countHandler struct { diff --git a/example_test.go b/example_test.go index df6d239e..ca7c43bc 100644 --- a/example_test.go +++ b/example_test.go @@ -12,7 +12,7 @@ import ( "os" "os/signal" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) func ExampleHijacker() { diff --git a/fcgi/child.go b/fcgi/child.go index fd403ef9..05129ee6 100644 --- a/fcgi/child.go +++ b/fcgi/child.go @@ -17,8 +17,8 @@ import ( "sync" "time" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/cgi" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/cgi" ) // request holds the state for an in-progress request. As soon as it's complete, diff --git a/fcgi/fcgi_test.go b/fcgi/fcgi_test.go index 0f2e3f0e..b6f51fce 100644 --- a/fcgi/fcgi_test.go +++ b/fcgi/fcgi_test.go @@ -11,7 +11,7 @@ import ( "strings" "testing" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) var sizeTests = []struct { diff --git a/fs_test.go b/fs_test.go index 3787389c..9eb7c7ab 100644 --- a/fs_test.go +++ b/fs_test.go @@ -27,8 +27,8 @@ import ( "testing" "time" - . "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" + . "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" ) const ( diff --git a/go.mod b/go.mod index 2468b750..025b467a 100644 --- a/go.mod +++ b/go.mod @@ -1,4 +1,4 @@ -module github.com/bogdanfinn/fhttp +module github.com/kernel/fhttp go 1.24.1 diff --git a/h2_bundle.go b/h2_bundle.go index 48796ba9..e319794e 100644 --- a/h2_bundle.go +++ b/h2_bundle.go @@ -2,7 +2,7 @@ // +build !nethttpomithttp2 // Code generated by golang.org/x/tools/cmd/bundle. DO NOT EDIT. -// $ bundle -o=h2_bundle.go -prefix=http2 -tags=!nethttpomithttp2 github.com/bogdanfinn/fhttp/http2 +// $ bundle -o=h2_bundle.go -prefix=http2 -tags=!nethttpomithttp2 github.com/kernel/fhttp/http2 // Package http2 implements the HTTP/2 protocol. // @@ -50,8 +50,8 @@ import ( "sync/atomic" "time" - "github.com/bogdanfinn/fhttp/http2/hpack" - "github.com/bogdanfinn/fhttp/httptrace" + "github.com/kernel/fhttp/http2/hpack" + "github.com/kernel/fhttp/httptrace" "golang.org/x/net/http/httpguts" "golang.org/x/net/idna" ) diff --git a/header.go b/header.go index b6481131..0a105ff4 100644 --- a/header.go +++ b/header.go @@ -12,7 +12,7 @@ import ( "sync" "time" - "github.com/bogdanfinn/fhttp/httptrace" + "github.com/kernel/fhttp/httptrace" ) // A Header represents the Key-value pairs in an HTTP header. diff --git a/header_test.go b/header_test.go index 3d9361bd..216ae5d6 100644 --- a/header_test.go +++ b/header_test.go @@ -11,7 +11,7 @@ import ( "testing" "time" - "github.com/bogdanfinn/fhttp/internal/race" + "github.com/kernel/fhttp/internal/race" ) var headerWriteTests = []struct { diff --git a/http.go b/http.go index eea931ce..bc68e846 100644 --- a/http.go +++ b/http.go @@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -//go:generate bundle -o=h2_bundle.go -prefix=http2 -tags=!nethttpomithttp2 github.com/bogdanfinn/fhttp/http2 +//go:generate bundle -o=h2_bundle.go -prefix=http2 -tags=!nethttpomithttp2 github.com/kernel/fhttp/http2 package http diff --git a/http2/client_conn_pool.go b/http2/client_conn_pool.go index 8f683002..e754d344 100644 --- a/http2/client_conn_pool.go +++ b/http2/client_conn_pool.go @@ -11,7 +11,7 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // ClientConnPool manages a pool of HTTP/2 client connections. diff --git a/http2/fhttp_test.go b/http2/fhttp_test.go index 746cd67e..a164c759 100644 --- a/http2/fhttp_test.go +++ b/http2/fhttp_test.go @@ -13,13 +13,13 @@ import ( "strings" "testing" - "github.com/bogdanfinn/fhttp/cookiejar" - "github.com/bogdanfinn/fhttp/httptest" + "github.com/kernel/fhttp/cookiejar" + "github.com/kernel/fhttp/httptest" tls "github.com/bogdanfinn/utls" "golang.org/x/net/publicsuffix" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/http2" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2" ) // Tests if connection settings are written correctly diff --git a/http2/frame.go b/http2/frame.go index 9a3869a7..1eee568a 100644 --- a/http2/frame.go +++ b/http2/frame.go @@ -14,7 +14,7 @@ import ( "strings" "sync" - "github.com/bogdanfinn/fhttp/http2/hpack" + "github.com/kernel/fhttp/http2/hpack" "golang.org/x/net/http/httpguts" ) diff --git a/http2/frame_test.go b/http2/frame_test.go index cb4c03f1..f59c2e56 100644 --- a/http2/frame_test.go +++ b/http2/frame_test.go @@ -13,7 +13,7 @@ import ( "testing" "unsafe" - "github.com/bogdanfinn/fhttp/http2/hpack" + "github.com/kernel/fhttp/http2/hpack" ) func testFramer() (*Framer, *bytes.Buffer) { diff --git a/http2/go111.go b/http2/go111.go index 12874467..b4b84e09 100644 --- a/http2/go111.go +++ b/http2/go111.go @@ -10,7 +10,7 @@ package http2 import ( "net/textproto" - "github.com/bogdanfinn/fhttp/httptrace" + "github.com/kernel/fhttp/httptrace" ) func traceHasWroteHeaderField(trace *httptrace.ClientTrace) bool { diff --git a/http2/h2c/h2c.go b/http2/h2c/h2c.go index c0d44dd8..62f259e2 100644 --- a/http2/h2c/h2c.go +++ b/http2/h2c/h2c.go @@ -22,9 +22,9 @@ import ( "os" "strings" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/http2" - "github.com/bogdanfinn/fhttp/http2/hpack" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2" + "github.com/kernel/fhttp/http2/hpack" "golang.org/x/net/http/httpguts" ) diff --git a/http2/h2c/h2c_test.go b/http2/h2c/h2c_test.go index fddebbe3..9cb55bc7 100644 --- a/http2/h2c/h2c_test.go +++ b/http2/h2c/h2c_test.go @@ -11,8 +11,8 @@ import ( "log" "testing" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/http2" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2" ) func TestSettingsAckSwallowWriter(t *testing.T) { diff --git a/http2/h2i/h2i.go b/http2/h2i/h2i.go index 28bc1db4..3f759d99 100644 --- a/http2/h2i/h2i.go +++ b/http2/h2i/h2i.go @@ -37,8 +37,8 @@ import ( tls "github.com/bogdanfinn/utls" - "github.com/bogdanfinn/fhttp/http2" - "github.com/bogdanfinn/fhttp/http2/hpack" + "github.com/kernel/fhttp/http2" + "github.com/kernel/fhttp/http2/hpack" "golang.org/x/term" ) diff --git a/http2/header_order_test.go b/http2/header_order_test.go index 465ab9ce..050d6759 100644 --- a/http2/header_order_test.go +++ b/http2/header_order_test.go @@ -6,8 +6,8 @@ import ( "strings" "testing" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptrace" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptrace" ) func TestHeaderOrder(t *testing.T) { diff --git a/http2/headermap.go b/http2/headermap.go index c90a8432..c394c22a 100644 --- a/http2/headermap.go +++ b/http2/headermap.go @@ -8,7 +8,7 @@ import ( "strings" "sync" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) var ( diff --git a/http2/http2.go b/http2/http2.go index b77d8917..05b5654b 100644 --- a/http2/http2.go +++ b/http2/http2.go @@ -27,7 +27,7 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" "golang.org/x/net/http/httpguts" ) diff --git a/http2/http2_test.go b/http2/http2_test.go index 336ad517..e3176cd1 100644 --- a/http2/http2_test.go +++ b/http2/http2_test.go @@ -15,9 +15,9 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" - "github.com/bogdanfinn/fhttp/http2/hpack" + "github.com/kernel/fhttp/http2/hpack" ) var knownFailing = flag.Bool("known_failing", false, "Run known-failing tests.") diff --git a/http2/not_go111.go b/http2/not_go111.go index 7b7f148f..bdcb8572 100644 --- a/http2/not_go111.go +++ b/http2/not_go111.go @@ -10,7 +10,7 @@ package http2 import ( "net/textproto" - "github.com/bogdanfinn/fhttp/httptrace" + "github.com/kernel/fhttp/httptrace" ) func traceHasWroteHeaderField(trace *httptrace.ClientTrace) bool { return false } diff --git a/http2/push_consume.go b/http2/push_consume.go index 37006099..52ab9e62 100644 --- a/http2/push_consume.go +++ b/http2/push_consume.go @@ -8,7 +8,7 @@ import ( "errors" "net/url" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) var ( diff --git a/http2/push_consume_test.go b/http2/push_consume_test.go index 24cd8780..6f44ba5d 100644 --- a/http2/push_consume_test.go +++ b/http2/push_consume_test.go @@ -9,8 +9,8 @@ import ( "reflect" "testing" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/http2/hpack" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2/hpack" ) func TestPushPromiseHeadersToHTTPRequest(t *testing.T) { diff --git a/http2/server.go b/http2/server.go index 9173f1c4..b1f4a423 100644 --- a/http2/server.go +++ b/http2/server.go @@ -47,8 +47,8 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/http2/hpack" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2/hpack" "golang.org/x/net/http/httpguts" ) diff --git a/http2/server_push_test.go b/http2/server_push_test.go index ab963579..ebaa3d97 100644 --- a/http2/server_push_test.go +++ b/http2/server_push_test.go @@ -15,7 +15,7 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) func TestServer_Push_Success(t *testing.T) { diff --git a/http2/server_test.go b/http2/server_test.go index fa50680f..f64c3478 100644 --- a/http2/server_test.go +++ b/http2/server_test.go @@ -29,8 +29,8 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" "golang.org/x/net/http2/hpack" ) diff --git a/http2/transport.go b/http2/transport.go index cafcb92e..d4d568a1 100644 --- a/http2/transport.go +++ b/http2/transport.go @@ -29,10 +29,10 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptrace" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptrace" - "github.com/bogdanfinn/fhttp/http2/hpack" + "github.com/kernel/fhttp/http2/hpack" "golang.org/x/net/http/httpguts" "golang.org/x/net/idna" ) diff --git a/http2/transport_flow_test.go b/http2/transport_flow_test.go index 0917f9c5..453eae26 100644 --- a/http2/transport_flow_test.go +++ b/http2/transport_flow_test.go @@ -11,7 +11,7 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // throttledSink caps the rate at which a response body is consumed, simulating diff --git a/http2/transport_test.go b/http2/transport_test.go index 8490d9d0..0d563f03 100644 --- a/http2/transport_test.go +++ b/http2/transport_test.go @@ -33,10 +33,10 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/http2/hpack" - "github.com/bogdanfinn/fhttp/httptest" - "github.com/bogdanfinn/fhttp/httptrace" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2/hpack" + "github.com/kernel/fhttp/httptest" + "github.com/kernel/fhttp/httptrace" ) var ( diff --git a/http2/write.go b/http2/write.go index 68df079a..03e416bf 100644 --- a/http2/write.go +++ b/http2/write.go @@ -10,8 +10,8 @@ import ( "log" "net/url" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/http2/hpack" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2/hpack" "golang.org/x/net/http/httpguts" ) diff --git a/http_test.go b/http_test.go index 7559b068..597ce9cf 100644 --- a/http_test.go +++ b/http_test.go @@ -13,7 +13,7 @@ import ( "reflect" "testing" - "github.com/bogdanfinn/fhttp/internal/testenv" + "github.com/kernel/fhttp/internal/testenv" ) func TestForeachHeaderElement(t *testing.T) { diff --git a/httptest/example_test.go b/httptest/example_test.go index f1e0183c..8162887f 100644 --- a/httptest/example_test.go +++ b/httptest/example_test.go @@ -9,8 +9,8 @@ import ( "io" "log" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" ) func ExampleResponseRecorder() { diff --git a/httptest/httptest.go b/httptest/httptest.go index 4fb5dea2..fb58a41f 100644 --- a/httptest/httptest.go +++ b/httptest/httptest.go @@ -13,7 +13,7 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // NewRequest returns a new incoming server Request, suitable diff --git a/httptest/httptest_test.go b/httptest/httptest_test.go index 05749398..2ebcd09c 100644 --- a/httptest/httptest_test.go +++ b/httptest/httptest_test.go @@ -13,7 +13,7 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) func TestNewRequest(t *testing.T) { diff --git a/httptest/recorder.go b/httptest/recorder.go index ba6a1051..f792ecee 100644 --- a/httptest/recorder.go +++ b/httptest/recorder.go @@ -12,7 +12,7 @@ import ( "strconv" "strings" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" "golang.org/x/net/http/httpguts" ) diff --git a/httptest/recorder_test.go b/httptest/recorder_test.go index 5a59f933..b59f58f9 100644 --- a/httptest/recorder_test.go +++ b/httptest/recorder_test.go @@ -9,7 +9,7 @@ import ( "io" "testing" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) func TestRecorder(t *testing.T) { diff --git a/httptest/server.go b/httptest/server.go index 464aaf35..ef226534 100644 --- a/httptest/server.go +++ b/httptest/server.go @@ -19,8 +19,8 @@ import ( tls "github.com/bogdanfinn/utls" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/internal" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/internal" ) // A Server is an HTTP server listening on a system-chosen port on the diff --git a/httptest/server_test.go b/httptest/server_test.go index 74e87dc2..41bdd620 100644 --- a/httptest/server_test.go +++ b/httptest/server_test.go @@ -10,7 +10,7 @@ import ( "net" "testing" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) type newServerFunc func(http.Handler) *Server diff --git a/httptrace/example_test.go b/httptrace/example_test.go index 099da09c..42be66b5 100644 --- a/httptrace/example_test.go +++ b/httptrace/example_test.go @@ -9,7 +9,7 @@ import ( "log" "net/http" - "github.com/bogdanfinn/fhttp/httptrace" + "github.com/kernel/fhttp/httptrace" ) func Example() { diff --git a/httptrace/trace.go b/httptrace/trace.go index 11a81a34..fe20492b 100644 --- a/httptrace/trace.go +++ b/httptrace/trace.go @@ -15,7 +15,7 @@ import ( tls "github.com/bogdanfinn/utls" - "github.com/bogdanfinn/fhttp/internal/nettrace" + "github.com/kernel/fhttp/internal/nettrace" ) // unique type to prevent assignment. diff --git a/httputil/dump.go b/httputil/dump.go index 7b04ff34..1bbca8b9 100644 --- a/httputil/dump.go +++ b/httputil/dump.go @@ -15,7 +15,7 @@ import ( "strings" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // drainBody reads all of b to memory and then returns two equivalent diff --git a/httputil/dump_test.go b/httputil/dump_test.go index 4db5e452..b00c2adc 100644 --- a/httputil/dump_test.go +++ b/httputil/dump_test.go @@ -18,7 +18,7 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) type eofReader struct{} diff --git a/httputil/example_test.go b/httputil/example_test.go index 63ee8299..922e2135 100644 --- a/httputil/example_test.go +++ b/httputil/example_test.go @@ -11,9 +11,9 @@ import ( "net/url" "strings" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" - "github.com/bogdanfinn/fhttp/httputil" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" + "github.com/kernel/fhttp/httputil" ) func ExampleDumpRequest() { diff --git a/httputil/httputil.go b/httputil/httputil.go index e3f5b4a7..f7ba110d 100644 --- a/httputil/httputil.go +++ b/httputil/httputil.go @@ -9,7 +9,7 @@ package httputil import ( "io" - "github.com/bogdanfinn/fhttp/internal" + "github.com/kernel/fhttp/internal" ) // NewChunkedReader returns a new chunkedReader that translates the data read from r diff --git a/httputil/persist.go b/httputil/persist.go index 54989282..59dd575e 100644 --- a/httputil/persist.go +++ b/httputil/persist.go @@ -12,7 +12,7 @@ import ( "net/textproto" "sync" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) var ( diff --git a/httputil/reverseproxy.go b/httputil/reverseproxy.go index 4b11e05a..16e15527 100644 --- a/httputil/reverseproxy.go +++ b/httputil/reverseproxy.go @@ -18,7 +18,7 @@ import ( "sync" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" "golang.org/x/net/http/httpguts" ) diff --git a/httputil/reverseproxy_test.go b/httputil/reverseproxy_test.go index 195a5d50..6666355c 100644 --- a/httputil/reverseproxy_test.go +++ b/httputil/reverseproxy_test.go @@ -24,8 +24,8 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" ) const fakeHopHeader = "X-Fake-Hop-Header-For-Test" diff --git a/internal/testenv/testenv.go b/internal/testenv/testenv.go index cefe1589..8ab5b1ca 100644 --- a/internal/testenv/testenv.go +++ b/internal/testenv/testenv.go @@ -22,7 +22,7 @@ import ( "sync" "testing" - "github.com/bogdanfinn/fhttp/internal/cfg" + "github.com/kernel/fhttp/internal/cfg" ) // Builder reports the name of the builder running this test diff --git a/main_test.go b/main_test.go index 308ff4a1..1f153dcd 100644 --- a/main_test.go +++ b/main_test.go @@ -15,7 +15,7 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) var quietLog = log.New(io.Discard, "", 0) diff --git a/pprof/pprof.go b/pprof/pprof.go index c4630755..9fa56542 100644 --- a/pprof/pprof.go +++ b/pprof/pprof.go @@ -74,8 +74,8 @@ import ( "strings" "time" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/internal/profile" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/internal/profile" ) func init() { diff --git a/pprof/pprof_test.go b/pprof/pprof_test.go index d728cae2..4a29ccab 100644 --- a/pprof/pprof_test.go +++ b/pprof/pprof_test.go @@ -16,9 +16,9 @@ import ( "testing" "time" - http "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" - "github.com/bogdanfinn/fhttp/internal/profile" + http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" + "github.com/kernel/fhttp/internal/profile" ) // TestDescriptions checks that the profile names under runtime/pprof package diff --git a/request.go b/request.go index 2163a8c7..a2c1fdc3 100644 --- a/request.go +++ b/request.go @@ -26,7 +26,7 @@ import ( tls "github.com/bogdanfinn/utls" - "github.com/bogdanfinn/fhttp/httptrace" + "github.com/kernel/fhttp/httptrace" "golang.org/x/net/idna" ) diff --git a/request_test.go b/request_test.go index 6f6a0634..8bd8acf0 100644 --- a/request_test.go +++ b/request_test.go @@ -21,8 +21,8 @@ import ( "strings" "testing" - . "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" + . "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" ) func TestQuery(t *testing.T) { diff --git a/response_test.go b/response_test.go index da2c23c1..147a04b2 100644 --- a/response_test.go +++ b/response_test.go @@ -18,7 +18,7 @@ import ( "strings" "testing" - "github.com/bogdanfinn/fhttp/internal" + "github.com/kernel/fhttp/internal" ) type respTest struct { diff --git a/serve_test.go b/serve_test.go index e6c34cf9..cf52a242 100644 --- a/serve_test.go +++ b/serve_test.go @@ -37,11 +37,11 @@ import ( tls "github.com/bogdanfinn/utls" - . "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" - "github.com/bogdanfinn/fhttp/httputil" - "github.com/bogdanfinn/fhttp/internal" - "github.com/bogdanfinn/fhttp/internal/testenv" + . "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" + "github.com/kernel/fhttp/httputil" + "github.com/kernel/fhttp/internal" + "github.com/kernel/fhttp/internal/testenv" ) type dummyAddr string diff --git a/sniff_test.go b/sniff_test.go index 9e4afaf0..57a300f3 100644 --- a/sniff_test.go +++ b/sniff_test.go @@ -14,7 +14,7 @@ import ( "strings" "testing" - . "github.com/bogdanfinn/fhttp" + . "github.com/kernel/fhttp" ) var sniffTests = []struct { diff --git a/transfer.go b/transfer.go index 76d62ed5..cd77b697 100644 --- a/transfer.go +++ b/transfer.go @@ -18,8 +18,8 @@ import ( "sync" "time" - "github.com/bogdanfinn/fhttp/httptrace" - "github.com/bogdanfinn/fhttp/internal" + "github.com/kernel/fhttp/httptrace" + "github.com/kernel/fhttp/internal" "golang.org/x/net/http/httpguts" ) diff --git a/transport.go b/transport.go index ba675a29..ae270a26 100644 --- a/transport.go +++ b/transport.go @@ -36,7 +36,7 @@ import ( tls "github.com/bogdanfinn/utls" - "github.com/bogdanfinn/fhttp/httptrace" + "github.com/kernel/fhttp/httptrace" "golang.org/x/net/http/httpguts" "golang.org/x/net/http/httpproxy" diff --git a/transport_internal_test.go b/transport_internal_test.go index 5cccc345..0c9f1252 100644 --- a/transport_internal_test.go +++ b/transport_internal_test.go @@ -18,7 +18,7 @@ import ( tls "github.com/bogdanfinn/utls" - "github.com/bogdanfinn/fhttp/internal" + "github.com/kernel/fhttp/internal" ) // Issue 15446: incorrect wrapping of errors when server closes an idle connection. diff --git a/transport_test.go b/transport_test.go index c8a569b7..cead0a1d 100644 --- a/transport_test.go +++ b/transport_test.go @@ -39,12 +39,12 @@ import ( tls "github.com/bogdanfinn/utls" - . "github.com/bogdanfinn/fhttp" - "github.com/bogdanfinn/fhttp/httptest" - "github.com/bogdanfinn/fhttp/httptrace" - "github.com/bogdanfinn/fhttp/httputil" - "github.com/bogdanfinn/fhttp/internal" - "github.com/bogdanfinn/fhttp/internal/nettrace" + . "github.com/kernel/fhttp" + "github.com/kernel/fhttp/httptest" + "github.com/kernel/fhttp/httptrace" + "github.com/kernel/fhttp/httputil" + "github.com/kernel/fhttp/internal" + "github.com/kernel/fhttp/internal/nettrace" "golang.org/x/net/http/httpguts" ) diff --git a/triv.go b/triv.go index 04a80519..d6dc8ccb 100644 --- a/triv.go +++ b/triv.go @@ -19,7 +19,7 @@ import ( "strconv" "sync" - http "github.com/bogdanfinn/fhttp" + http "github.com/kernel/fhttp" ) // hello world, the web server From 99252f9f5e06e5b107236d1b2e3d2a778347390f Mon Sep 17 00:00:00 2001 From: hiroTamada <88675973+hiroTamada@users.noreply.github.com> Date: Sat, 29 Aug 2026 20:45:51 +0000 Subject: [PATCH 3/3] http2: port upstream inbound flow control fixes to stop window leaks Port two golang.org/x/net fixes the fork predates: - 7805fdc3 "http2: rewrite inbound flow control tracking" (client side): replace the threshold-based WINDOW_UPDATE refresh with the inflow accounting type. The old Read path refreshed the connection window only once available dropped below half of cc.connFlow, and Close refunded only unread bytes, so bytes read from a body that was then closed were never returned to the peer. Aborted downloads accumulate the stranded deficit until the connection window reaches zero, at which point every later stream on the pooled connection starves with zero bytes: no read can happen, so no refresh can ever fire again. The refresh ceiling was also cc.connFlow while the advertised window is cc.connFlow+65535, permanently stranding the protocol-default slack. inflow credits every consumed byte exactly once and batches wire updates (>=4KiB, or enough to at least double the peer's remaining window). - 9f24bb44 "http2: properly discard data received after request/response body is closed": break the body pipe before refunding unread credit (not after), fail pipe writes after a break instead of silently discarding, and refund connection-level credit for DATA that loses the race with Close. Also the server-side handler-closed-body refund. TestTransportBodyCloseRaceRefundsConnFlow drives 200 aborted-mid-flood downloads against a frame-level server that strictly obeys the client's advertised windows, then asserts a full-body canary request still completes. It wedges on master (canary starves with zero bytes) and passes with this change. --- http2/flow.go | 68 ++++++++++++++ http2/pipe.go | 6 +- http2/pipe_test.go | 10 +- http2/server.go | 7 +- http2/transport.go | 147 +++++++++++++---------------- http2/transport_flow_test.go | 177 +++++++++++++++++++++++++++++++++++ 6 files changed, 323 insertions(+), 92 deletions(-) diff --git a/http2/flow.go b/http2/flow.go index b51f0e0c..eaf7b723 100644 --- a/http2/flow.go +++ b/http2/flow.go @@ -50,3 +50,71 @@ func (f *flow) add(n int32) bool { } return false } + +// inflowMinRefresh is the minimum number of bytes we'll send for a +// flow control window update. +const inflowMinRefresh = 4 << 10 + +// inflow accounts for an inbound flow control window. +// It tracks both the latest window sent to the peer (used for enforcement) +// and the accumulated unsent window. +type inflow struct { + avail int32 + unsent int32 +} + +// init sets the initial window. +func (f *inflow) init(n int32) { + f.avail = n +} + +// add adds n bytes to the window, with a maximum window size of max, +// indicating that the peer can now send us more data. +// For example, the user read from a {Request,Response} body and consumed +// some of the buffered data, so the peer can now send more. +// It returns the number of bytes to send in a WINDOW_UPDATE frame to the peer. +// Window updates are accumulated and sent when the unsent capacity +// is at least inflowMinRefresh or will at least double the peer's available window. +func (f *inflow) add(n int) (connAdd int32) { + if n < 0 { + panic("negative update") + } + unsent := int64(f.unsent) + int64(n) + // "A sender MUST NOT allow a flow-control window to exceed 2^31-1 octets." + // RFC 7540 Section 6.9.1. + const maxWindow = 1<<31 - 1 + if unsent+int64(f.avail) > maxWindow { + panic("flow control update exceeds maximum window size") + } + f.unsent = int32(unsent) + if f.unsent < inflowMinRefresh && f.unsent < f.avail { + // If there aren't at least inflowMinRefresh bytes of window to send, + // and this update won't at least double the window, buffer the update for later. + return 0 + } + f.avail += f.unsent + f.unsent = 0 + return int32(unsent) +} + +// take attempts to take n bytes from the peer's flow control window. +// It reports whether the window has available capacity. +func (f *inflow) take(n uint32) bool { + if n > uint32(f.avail) { + return false + } + f.avail -= int32(n) + return true +} + +// takeInflows attempts to take n bytes from two inflows, +// typically connection-level and stream-level flows. +// It reports whether both windows have available capacity. +func takeInflows(f1, f2 *inflow, n uint32) bool { + if n > uint32(f1.avail) || n > uint32(f2.avail) { + return false + } + f1.avail -= int32(n) + f2.avail -= int32(n) + return true +} diff --git a/http2/pipe.go b/http2/pipe.go index 5d7a844b..7f63ff9b 100644 --- a/http2/pipe.go +++ b/http2/pipe.go @@ -80,13 +80,9 @@ func (p *pipe) Write(d []byte) (n int, err error) { p.c.L = &p.mu } defer p.c.Signal() - if p.err != nil { + if p.err != nil || p.breakErr != nil { return 0, errClosedPipeWrite } - if p.breakErr != nil { - p.unread += len(d) - return len(d), nil // discard when there is no reader - } // pipe.setBuffer is never invoked, leaving the buffer uninitialized. // We shouldn't try to write to an uninitialized pipe, // but returning an error is better than panicking. diff --git a/http2/pipe_test.go b/http2/pipe_test.go index 83d2dfd2..2574d2ae 100644 --- a/http2/pipe_test.go +++ b/http2/pipe_test.go @@ -125,15 +125,15 @@ func TestPipeBreakWithError(t *testing.T) { if p.Len() != 3 { t.Errorf("pipe should have 3 unread bytes") } - // Write should succeed silently. - if n, err := p.Write([]byte("abc")); err != nil || n != 3 { - t.Errorf("Write(abc) after break\ngot %v, %v\nwant 0, nil", n, err) + // Write should fail. + if n, err := p.Write([]byte("abc")); err != errClosedPipeWrite || n != 0 { + t.Errorf("Write(abc) after break\ngot %v, %v\nwant 0, errClosedPipeWrite", n, err) } if p.b != nil { t.Errorf("buffer should be nil after Write") } - if p.Len() != 6 { - t.Errorf("pipe should have 6 unread bytes") + if p.Len() != 3 { + t.Errorf("pipe should have 3 unread bytes") } // Read should fail. if n, err := p.Read(make([]byte, 1)); err == nil || n != 0 { diff --git a/http2/server.go b/http2/server.go index b1f4a423..7a13dfd4 100644 --- a/http2/server.go +++ b/http2/server.go @@ -1702,15 +1702,18 @@ func (sc *serverConn) processData(f *DataFrame) error { st.inflow.take(int32(f.Length)) if len(data) > 0 { + st.bodyBytes += int64(len(data)) wrote, err := st.body.Write(data) if err != nil { + // The handler has closed the request body. + // Return the connection-level flow control for the discarded data, + // but not the stream-level flow control. sc.sendWindowUpdate(nil, int(f.Length)-wrote) - return streamError(id, ErrCodeStreamClosed) + return nil } if wrote != len(data) { panic("internal error: bad Writer") } - st.bodyBytes += int64(len(data)) } // Return any padded flow control now, since we won't diff --git a/http2/transport.go b/http2/transport.go index d4d568a1..e47f3868 100644 --- a/http2/transport.go +++ b/http2/transport.go @@ -47,10 +47,6 @@ const ( // control tokens we announce to the peer, and how many bytes // we buffer per stream. transportDefaultStreamFlow = 4 << 20 - - // transportDefaultStreamMinRefresh is the minimum number of bytes we'll send - // a stream-level WINDOW_UPDATE for at a time. - transportDefaultStreamMinRefresh = 4 << 10 ) // Transport is an HTTP/2 Transport. @@ -327,7 +323,7 @@ type ClientConn struct { idleTimeout time.Duration // or 0 for never idleTimer *time.Timer - inflow flow // peer's conn-level flow control + inflow inflow // peer's conn-level flow control initialWindowSize uint32 lastActive time.Time @@ -377,8 +373,8 @@ type clientStream struct { flow flow // guarded by cc.mu gotEndStream bool // got frame with END_STREAM flag set ID uint32 - inflow flow // guarded by cc.mu - num1xx uint8 // number of 1xx responses seen + inflow inflow // guarded by cc.mu + num1xx uint8 // number of 1xx responses seen on100 func() // optional code to run if get a 100 continue response @@ -876,7 +872,7 @@ func (t *Transport) newClientConn(c net.Conn, addr string, singleUse bool) (*Cli } // Use the dynamic connection flow value we calculated earlier - cc.inflow.add(int32(cc.connFlow) + int32(initialWindowSize)) + cc.inflow.init(int32(cc.connFlow) + initialWindowSize) cc.bw.Flush() if cc.werr != nil { @@ -1997,8 +1993,7 @@ func (cc *ClientConn) newStreamWithID(streamID uint32, incNext bool) *clientStre } cs.flow.add(int32(cc.initialWindowSize)) cs.flow.setConnFlow(&cc.flow) - cs.inflow.add(int32(cc.streamFlow)) - cs.inflow.setConnFlow(&cc.inflow) + cs.inflow.init(int32(cc.streamFlow)) cc.streams[cs.ID] = cs if incNext { @@ -2441,46 +2436,17 @@ func (b transportResponseBody) Read(p []byte) (n int, err error) { cc.mu.Lock() defer cc.mu.Unlock() - var connAdd, streamAdd int32 - - // Check the conn-level first, before the stream-level. - // Use dynamic connFlow logic - if v := cc.inflow.available(); v < int32(cc.connFlow/2) { - connAdd = int32(cc.connFlow) - v - cc.inflow.add(connAdd) - } - - if err == nil { - // Consider any buffered body data (read from the conn but not - // consumed by the client) when computing flow control for this - // stream. - - // Use dynamic streamFlow logic - unsent := int(cc.streamFlow) - int(cs.inflow.available()) - cs.bufPipe.Len() - - // ------------------------------------------------------------------ - // FIX: Adaptive Logic - // ------------------------------------------------------------------ - const aggressiveThreshold = 16384 // 16KB - - // Check if the configured initial window is small (e.g. Firefox's 128KB or 65KB). - // If so, we need to be aggressive with updates. - isSmallWindow := cc.initialWindowSize < 1048576 // < 1MB - - if isSmallWindow { - if unsent > aggressiveThreshold { - streamAdd = int32(unsent) - cs.inflow.add(streamAdd) - } - } else { - // Fallback to standard behavior for large windows (Chrome/Default). - // FIX: Replaced transportDefaultStreamFlow constant with cc.streamFlow. - // This ensures correct behavior if a user sets a custom Large window (e.g. 6MB). - if unsent > transportDefaultStreamMinRefresh && unsent > int(cc.streamFlow)/2 { - streamAdd = int32(unsent) - cs.inflow.add(streamAdd) - } - } + // Credit every byte the application consumes back to the flow control + // windows. The inflow accounting batches wire updates (at least + // inflowMinRefresh bytes, or enough to at least double the peer's + // remaining window), and unlike a threshold refresh it conserves + // credit exactly: bytes read from a body that is closed shortly after + // are still refunded by the next stream's reads instead of stranding + // until the connection window strangles to zero. + connAdd := cc.inflow.add(n) + var streamAdd int32 + if err == nil { // No need to refresh if the stream is over or failed. + streamAdd = cs.inflow.add(n) } if connAdd != 0 || streamAdd != 0 { @@ -2505,26 +2471,35 @@ func (b transportResponseBody) Close() error { cc := cs.cc serverSentStreamEnd := cs.bufPipe.Err() == io.EOF + + // Break the pipe before returning flow control credit for unread data. + // Pipe writes fail from here on, so no data can land in the pipe (and + // silently lose its connection-level credit) between the refund below and + // the break. See golang.org/x/net commit 9f24bb44. + cs.bufPipe.BreakWithError(errClosedResponseBody) + unread := cs.bufPipe.Len() if unread > 0 || !serverSentStreamEnd { cc.mu.Lock() + var connAdd int32 + if unread > 0 { + // Return connection-level flow control. + connAdd = cc.inflow.add(unread) + } cc.wmu.Lock() if !serverSentStreamEnd { cc.fr.WriteRSTStream(cs.ID, ErrCodeCancel) cs.didReset = true } - // Return connection-level flow control. - if unread > 0 { - cc.inflow.add(int32(unread)) - cc.fr.WriteWindowUpdate(0, uint32(unread)) + if connAdd > 0 { + cc.fr.WriteWindowUpdate(0, uint32(connAdd)) } cc.bw.Flush() cc.wmu.Unlock() cc.mu.Unlock() } - cs.bufPipe.BreakWithError(errClosedResponseBody) cc.forgetStreamID(cs.ID) return nil @@ -2554,13 +2529,18 @@ func (rl *clientConnReadLoop) processData(f *DataFrame) error { // But at least return their flow control: if f.Length > 0 { cc.mu.Lock() - cc.inflow.add(int32(f.Length)) + ok := cc.inflow.take(f.Length) + connAdd := cc.inflow.add(int(f.Length)) cc.mu.Unlock() - - cc.wmu.Lock() - cc.fr.WriteWindowUpdate(0, uint32(f.Length)) - cc.bw.Flush() - cc.wmu.Unlock() + if !ok { + return ConnectionError(ErrCodeFlowControl) + } + if connAdd > 0 { + cc.wmu.Lock() + cc.fr.WriteWindowUpdate(0, uint32(connAdd)) + cc.bw.Flush() + cc.wmu.Unlock() + } } return nil @@ -2589,9 +2569,7 @@ func (rl *clientConnReadLoop) processData(f *DataFrame) error { } // Check connection-level flow control. cc.mu.Lock() - if cs.inflow.available() >= int32(f.Length) { - cs.inflow.take(int32(f.Length)) - } else { + if !takeInflows(&cc.inflow, &cs.inflow, f.Length) { cc.mu.Unlock() return ConnectionError(ErrCodeFlowControl) @@ -2602,31 +2580,40 @@ func (rl *clientConnReadLoop) processData(f *DataFrame) error { if pad := int(f.Length) - len(data); pad > 0 { refund += pad } + + didReset := cs.didReset + if len(data) > 0 && !didReset { + if _, err := cs.bufPipe.Write(data); err != nil { + // The response body has been closed; writes to its pipe + // fail, so the data will never be read. Fall through to + // refund its connection-level flow control below rather + // than silently losing the credit. + didReset = true + } + } // Return len(data) now if the stream is already closed, // since data will never be read. - didReset := cs.didReset if didReset { refund += len(data) } - if refund > 0 { - cc.inflow.add(int32(refund)) - cc.wmu.Lock() - cc.fr.WriteWindowUpdate(0, uint32(refund)) - if !didReset { - cs.inflow.add(int32(refund)) - cc.fr.WriteWindowUpdate(cs.ID, uint32(refund)) - } - cc.bw.Flush() - cc.wmu.Unlock() + + sendConn := cc.inflow.add(refund) + var sendStream int32 + if !didReset { + sendStream = cs.inflow.add(refund) } cc.mu.Unlock() - if len(data) > 0 && !didReset { - if _, err := cs.bufPipe.Write(data); err != nil { - rl.endStreamError(cs, err) - - return err + if sendConn > 0 || sendStream > 0 { + cc.wmu.Lock() + if sendConn > 0 { + cc.fr.WriteWindowUpdate(0, uint32(sendConn)) + } + if sendStream > 0 { + cc.fr.WriteWindowUpdate(cs.ID, uint32(sendStream)) } + cc.bw.Flush() + cc.wmu.Unlock() } } diff --git a/http2/transport_flow_test.go b/http2/transport_flow_test.go index 453eae26..6c29ffe2 100644 --- a/http2/transport_flow_test.go +++ b/http2/transport_flow_test.go @@ -5,13 +5,16 @@ package http2 import ( + "bytes" "fmt" "io" "net" + "strconv" "testing" "time" http "github.com/kernel/fhttp" + "github.com/kernel/fhttp/http2/hpack" ) // throttledSink caps the rate at which a response body is consumed, simulating @@ -98,3 +101,177 @@ func TestTransportSlowReaderLargeResponse(t *testing.T) { t.Fatalf("read %d bytes, want %d", n, bodySize) } } + +// TestTransportBodyCloseRaceRefundsConnFlow verifies that closing a response +// body while DATA frames for it are still arriving does not leak +// connection-level flow control credit. +// +// Regression test for a lost-refund race: processData checked cs.didReset +// under cc.mu but wrote to the body pipe after releasing it, while Close +// refunded buffered bytes and only then broke the pipe. Data written in +// between was silently discarded by the broken pipe, and its +// connection-level credit was never returned. Read-path window refreshes +// cannot heal the deficit because they only run when a read returns data, +// so once the window hit zero every later stream on the connection starved +// with zero bytes until the origin reset it. See golang.org/x/net commit +// 9f24bb44 (golang/go#57578). +// +// The peer is a frame-level server that strictly obeys the flow control +// windows the client advertises, so a starved transfer can only mean the +// client failed to refund credit. The race is probabilistic; the abort loop +// amplifies it. The test never fails on correct code and reliably starves +// the canary request on code with the leak. +func TestTransportBodyCloseRaceRefundsConnFlow(t *testing.T) { + const ( + bodySize = 256 << 10 + iterations = 200 + frameSize = 8 << 10 + ) + + ct := newClientTester(t) + ct.tr.Settings = map[SettingID]uint32{ + SettingInitialWindowSize: 65535, + } + ct.tr.SettingsOrder = []SettingID{SettingInitialWindowSize} + ct.tr.ConnectionFlow = 65535 + + // If the connection wedges (the failure mode under test), unblock all + // reads so the test fails fast instead of hitting the suite timeout. + watchdog := time.AfterFunc(60*time.Second, func() { + ct.sc.Close() + ct.cc.Close() + }) + defer watchdog.Stop() + + ct.client = func() error { + for i := 0; i <= iterations; i++ { + req, err := http.NewRequest("GET", "https://dummy.tld/", nil) + if err != nil { + return err + } + resp, err := ct.tr.RoundTrip(req) + if err != nil { + return fmt.Errorf("iteration %d: RoundTrip: %w", i, err) + } + if i < iterations { + // Abort mid-transfer so Close races DATA delivery. + if _, err := io.CopyN(io.Discard, resp.Body, 4<<10); err != nil { + resp.Body.Close() + return fmt.Errorf("iteration %d: read: %w", i, err) + } + resp.Body.Close() + continue + } + // Canary: with the connection window intact this completes + // instantly; with leaked credit it starves with zero bytes, + // like every later stream on a wedged connection. + n, err := io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if err != nil { + return fmt.Errorf("canary read failed after %d of %d bytes (connection-level flow control credit leaked?): %w", n, bodySize, err) + } + if n != bodySize { + return fmt.Errorf("canary read %d bytes, want %d", n, bodySize) + } + } + return nil + } + + ct.server = func() error { + ct.greet() + ct.fr.ReadMetaHeaders = hpack.NewDecoder(initialHeaderTableSize, nil) + + var hbuf bytes.Buffer + henc := hpack.NewEncoder(&hbuf) + + // The client's connection-level receive window: the 65535 default + // plus whatever WINDOW_UPDATEs it sends, starting with the + // ConnectionFlow announcement. Stream windows start at the client's + // SETTINGS_INITIAL_WINDOW_SIZE. + connSend := 65535 + streamSend := make(map[uint32]int) + remaining := make(map[uint32]int) + var lastStream uint32 + + push := func() error { + for { + wrote := false + for id, rem := range remaining { + n := frameSize + if n > rem { + n = rem + } + if n > streamSend[id] { + n = streamSend[id] + } + if n > connSend { + n = connSend + } + if n <= 0 { + continue + } + if err := ct.fr.WriteData(id, false, make([]byte, n)); err != nil { + return err + } + remaining[id] -= n + streamSend[id] -= n + connSend -= n + if remaining[id] == 0 { + if err := ct.fr.WriteData(id, true, nil); err != nil { + return err + } + delete(remaining, id) + delete(streamSend, id) + } + wrote = true + } + if !wrote { + return nil + } + } + } + + for { + f, err := ct.fr.ReadFrame() + if err != nil { + return fmt.Errorf("server ReadFrame: %w", err) + } + switch f := f.(type) { + case *MetaHeadersFrame: + id := f.StreamID + lastStream = id + hbuf.Reset() + henc.WriteField(hpack.HeaderField{Name: ":status", Value: "200"}) + henc.WriteField(hpack.HeaderField{Name: "content-length", Value: strconv.Itoa(bodySize)}) + if err := ct.fr.WriteHeaders(HeadersFrameParam{ + StreamID: id, + BlockFragment: hbuf.Bytes(), + EndHeaders: true, + }); err != nil { + return err + } + streamSend[id] = 65535 + remaining[id] = bodySize + case *WindowUpdateFrame: + if f.StreamID == 0 { + connSend += int(f.Increment) + } else if _, ok := streamSend[f.StreamID]; ok { + streamSend[f.StreamID] += int(f.Increment) + } + case *RSTStreamFrame: + delete(remaining, f.StreamID) + delete(streamSend, f.StreamID) + } + if err := push(); err != nil { + return err + } + // The canary stream is the (iterations+1)-th; once it has been + // served in full, the script is complete. + if lastStream == 2*iterations+1 && len(remaining) == 0 { + return nil + } + } + } + + ct.run() +}