diff --git a/cmd/autocar/common.go b/cmd/autocar/common.go index f7f8786..6d06e38 100644 --- a/cmd/autocar/common.go +++ b/cmd/autocar/common.go @@ -34,6 +34,7 @@ type tunnelFlags struct { dialTimeout time.Duration primaryTimeout time.Duration openTimeout time.Duration + h2WriteTimeout time.Duration fallbackTTL time.Duration h3Fingerprint string pacing string @@ -56,6 +57,7 @@ func addTunnelFlags(fs *flag.FlagSet, flags *tunnelFlags) { fs.DurationVar(&flags.dialTimeout, "dial-timeout", 5*time.Second, "transport network dial timeout") fs.DurationVar(&flags.primaryTimeout, "quic-attempt-timeout", 5*time.Second, "entire UDP primary phase budget before auto-mode TCP fallback") fs.DurationVar(&flags.openTimeout, "open-timeout", 15*time.Second, "overall remote stream open timeout") + fs.DurationVar(&flags.h2WriteTimeout, "h2-write-timeout", 30*time.Second, "shared H2 connection write timeout (h2/web-auto only; 0 uses 30s; not an idle or per-stream timeout)") fs.DurationVar(&flags.fallbackTTL, "fallback-cooldown", 30*time.Second, "base time to prefer the TCP fallback after a UDP path failure (each retry is jittered +/-20%)") fs.StringVar(&flags.h3Fingerprint, "h3-fingerprint", string(tunnel.H3FingerprintChrome202608), "web H3 wire profile: chrome-2026-08 or native") fs.StringVar(&flags.pacing, "pacing", "adaptive", "QUIC application pacing: adaptive, reno, or fixed-rate") @@ -80,6 +82,9 @@ func buildTunnelDialer(flags tunnelFlags) (closeDialer, error) { if flags.dialTimeout <= 0 || flags.openTimeout <= 0 { return nil, errors.New("--dial-timeout and --open-timeout must be positive") } + if flags.h2WriteTimeout < 0 { + return nil, errors.New("--h2-write-timeout must not be negative; zero uses the 30s default") + } if (mode == "auto" || mode == "web-auto") && (flags.primaryTimeout <= 0 || flags.primaryTimeout >= flags.openTimeout) { return nil, fmt.Errorf("%s mode requires 0 < --quic-attempt-timeout < --open-timeout so the TCP fallback retains time", mode) } @@ -225,6 +230,7 @@ func buildTunnelDialer(flags tunnelFlags) (closeDialer, error) { TLSConfig: tlsConfig, HandshakeTimeout: flags.openTimeout, DialTimeout: flags.dialTimeout, + WriteByteTimeout: flags.h2WriteTimeout, }) if err != nil { return nil, err @@ -239,6 +245,7 @@ func buildTunnelDialer(flags tunnelFlags) (closeDialer, error) { HandshakeTimeout: flags.openTimeout, H3DialTimeout: flags.dialTimeout, H2DialTimeout: flags.dialTimeout, + H2WriteByteTimeout: flags.h2WriteTimeout, PrimaryAttemptTimeout: flags.primaryTimeout, FallbackCooldown: flags.fallbackTTL, }) diff --git a/cmd/autocar/h2_write_timeout_test.go b/cmd/autocar/h2_write_timeout_test.go new file mode 100644 index 0000000..d51e056 --- /dev/null +++ b/cmd/autocar/h2_write_timeout_test.go @@ -0,0 +1,96 @@ +package main + +import ( + "context" + "flag" + "io" + "strings" + "testing" + "time" +) + +func TestH2WriteTimeoutFlagAndConfig(t *testing.T) { + for _, test := range []struct { + name string + config string + args []string + want time.Duration + }{ + {name: "default", want: 30 * time.Second}, + {name: "CLI", args: []string{"--h2-write-timeout=90s"}, want: 90 * time.Second}, + {name: "zero selects default", args: []string{"--h2-write-timeout=0"}, want: 0}, + {name: "JSON", config: `{"h2-write-timeout":"2m"}`, want: 2 * time.Minute}, + {name: "CLI overrides JSON", config: `{"h2-write-timeout":"2m"}`, args: []string{"--h2-write-timeout=45s"}, want: 45 * time.Second}, + } { + t.Run(test.name, func(t *testing.T) { + fs := flag.NewFlagSet("test", flag.ContinueOnError) + var flags tunnelFlags + addTunnelFlags(fs, &flags) + args := test.args + if test.config != "" { + args = append([]string{"--config", writeTestCommandConfig(t, test.config)}, args...) + } + if err := parseFlagsWithConfig(fs, args); err != nil { + t.Fatal(err) + } + if flags.h2WriteTimeout != test.want { + t.Fatalf("H2 write timeout = %v, want %v", flags.h2WriteTimeout, test.want) + } + if flags.openTimeout != 15*time.Second || flags.dialTimeout != 5*time.Second { + t.Fatal("H2 write timeout changed unrelated setup timeouts") + } + }) + } +} + +func TestH2WriteTimeoutConfigRejectsMalformedValues(t *testing.T) { + for _, value := range []string{`12`, `null`, `"secret-invalid-duration"`} { + fs := flag.NewFlagSet("test", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags tunnelFlags + addTunnelFlags(fs, &flags) + path := writeTestCommandConfig(t, `{"h2-write-timeout":`+value+`}`) + err := parseFlagsWithConfig(fs, []string{"--config", path, "--h2-write-timeout=30s"}) + if err == nil || strings.Contains(err.Error(), "secret-invalid-duration") { + t.Fatalf("malformed duration must fail without leaking its value: %v", err) + } + } +} + +func TestH2WriteTimeoutPreflight(t *testing.T) { + clearPreflightEnvironment(t) + files := newPreflightFiles(t) + dnsCalls := denyPreflightDNS(t) + for _, mode := range []string{"h2", "web-auto"} { + for _, value := range []string{"0", "125ms", "2m", "-1s"} { + t.Run(mode+"/"+value, func(t *testing.T) { + args := append(files.clientArgs(), "--transport", mode, "--h2-write-timeout="+value) + err := runClient(context.Background(), args) + if value == "-1s" { + if err == nil || !strings.Contains(err.Error(), "--h2-write-timeout") { + t.Fatalf("negative timeout error = %v", err) + } + } else if err != nil { + t.Fatal(err) + } + }) + } + } + if got := dnsCalls.Load(); got != 0 { + t.Fatalf("offline H2 checks attempted %d DNS connections", got) + } +} + +func TestH2WriteTimeoutHelp(t *testing.T) { + fs := flag.NewFlagSet("test", flag.ContinueOnError) + var output strings.Builder + fs.SetOutput(&output) + var flags tunnelFlags + addTunnelFlags(fs, &flags) + fs.PrintDefaults() + for _, want := range []string{"h2-write-timeout", "h2/web-auto only", "0 uses 30s", "not an idle or per-stream timeout"} { + if !strings.Contains(output.String(), want) { + t.Fatalf("help omitted %q", want) + } + } +} diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index ee14445..e80c644 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -281,6 +281,25 @@ it on the server as `--destination-write-timeout=30s` or JSON 32 KiB writes can legitimately take longer. Negative durations are rejected by normal startup and `--check`. +Source builds provide a separate client `--h2-write-timeout` for explicit `h2` +and the H2 fallback of `web-auto`. Its default is `30s`; `0` selects that +default, and negative values are rejected. JSON uses +`"h2-write-timeout": "30s"`. It is independent of `--open-timeout`, local proxy +idle timeouts and server destination-write timeouts, and does not change H3 or +native transport settings. This option is not in the v1.0.1 release. + +This bounds pending HTTP/2 physical-connection writes, including control +frames, so a non-reading peer cannot indefinitely hold the shared writer lock +and prevent stream cleanup. No write is pending on a purely idle connection, +and successful writes clear their deadlines. It is a TLS write-call budget, +not a precise kernel-level byte-idleness timer: an exceptionally slow write +can time out despite partial network progress. Increase it if needed for such +links. The `30s` default is an operational choice, not a measured universal +optimum. A TLS write timeout can terminate **all streams on that connection**; +new requests establish a fresh authenticated session. Normal per-stream +cancellation remains isolated. Stream close may wait for the pending write's +budget and cleanup; this is bounded recovery, not an immediate-close promise. + Before leaving a client running, verify a real authenticated relay path with the same connection flags: diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index a18d2f6..ca32105 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -64,6 +64,15 @@ H2's handshake budget covers both TLS negotiation and the initial HTTP/2 preface/SETTINGS write. Caller cancellation or client shutdown also interrupts that initialization, before the connection enters the reusable session pool. +Source builds additionally bound H2 physical writes with a separate +`--h2-write-timeout` (default `30s`, zero selects the default). It applies to +explicit `h2` and `web-auto`'s H2 fallback, including shared control-frame +writes; a stalled peer can no longer hold that writer indefinitely. This is +not an idle or per-stream deadline. A physical TLS write timeout can end all +streams on the affected connection, while ordinary stream cancellation must +preserve healthy siblings. See [timeout semantics and configuration](DEPLOYMENT.md) +for partial-progress and cleanup boundaries. This setting is not in v1.0.1. + There is no `autocar/2` ALPN or AutoCAR binary stream header on these paths. The web ALPNs are `h2`, `h3`, and `http/1.1`. Native and web transports remain separate modes and are not wire-compatible. diff --git a/internal/tunnel/web_client.go b/internal/tunnel/web_client.go index f318503..44a9798 100644 --- a/internal/tunnel/web_client.go +++ b/internal/tunnel/web_client.go @@ -34,6 +34,12 @@ type WebClientConfig struct { HandshakeTimeout time.Duration H3DialTimeout time.Duration H2DialTimeout time.Duration + // H2WriteByteTimeout limits physical HTTP/2 writes, not stream lifetime or + // idle time. Zero uses thirty seconds; negative values are invalid. As with + // WebH2ClientConfig.WriteByteTimeout, partial network progress does not + // guarantee survival. A timeout can terminate every stream sharing the H2 + // fallback connection; it does not affect HTTP/3. + H2WriteByteTimeout time.Duration // PrimaryAttemptTimeout bounds an HTTP/3 CONNECT attempt, including a // request on an already warm connection. Zero defaults to five seconds. @@ -98,7 +104,7 @@ func NewWebClient(config WebClientConfig) (*WebClient, error) { if config.ServerAddress == "" { return nil, errors.New("tunnel: web-cover server address is required") } - if config.HandshakeTimeout < 0 || config.H3DialTimeout < 0 || config.H2DialTimeout < 0 || + if config.HandshakeTimeout < 0 || config.H3DialTimeout < 0 || config.H2DialTimeout < 0 || config.H2WriteByteTimeout < 0 || config.PrimaryAttemptTimeout < 0 || config.FallbackCooldown < 0 { return nil, errors.New("tunnel: web-cover client timeouts cannot be negative") } @@ -133,6 +139,7 @@ func NewWebClient(config WebClientConfig) (*WebClient, error) { TLSConfig: config.TLSConfig, HandshakeTimeout: config.HandshakeTimeout, DialTimeout: config.H2DialTimeout, + WriteByteTimeout: config.H2WriteByteTimeout, }) if err != nil { _ = h3.Close() diff --git a/internal/tunnel/web_h2_client.go b/internal/tunnel/web_h2_client.go index 1b99469..6af17a1 100644 --- a/internal/tunnel/web_h2_client.go +++ b/internal/tunnel/web_h2_client.go @@ -19,6 +19,8 @@ import ( "golang.org/x/net/http2" ) +const defaultWebH2WriteByteTimeout = 30 * time.Second + // WebH2ClientConfig configures the HTTP/2 side of the web-cover transport. type WebH2ClientConfig struct { ServerAddress string @@ -29,6 +31,13 @@ type WebH2ClientConfig struct { FingerprintProfile FingerprintProfile HandshakeTimeout time.Duration DialTimeout time.Duration + // WriteByteTimeout limits stalled writes on the shared HTTP/2 connection. + // Zero uses a conservative thirty-second default; negative values are + // invalid. This is a TLS write-call budget, not a precise TCP byte-idle + // timer: it can expire despite partial network progress. A timeout can + // terminate all streams on that physical TLS connection; it is not a + // stream deadline or an idle-connection timeout. + WriteByteTimeout time.Duration } // WebH2Client implements transport.Dialer with one standard HTTP/2 CONNECT @@ -107,7 +116,7 @@ func newWebH2ClientWithSigner(config WebH2ClientConfig, auth *webAuthSigner, cla if err := validateWebAuthClaims(claims); err != nil { return nil, err } - if config.HandshakeTimeout < 0 || config.DialTimeout < 0 { + if config.HandshakeTimeout < 0 || config.DialTimeout < 0 || config.WriteByteTimeout < 0 { return nil, errors.New("tunnel: web-cover HTTP/2 timeouts cannot be negative") } tlsConfig, err := webClientTLSConfig(config.TLSConfig, config.ServerAddress, webH2ALPN) @@ -135,6 +144,10 @@ func newWebH2ClientWithSigner(config WebH2ClientConfig, auth *webAuthSigner, cla if dialTimeout == 0 { dialTimeout = defaultDialTimeout } + writeByteTimeout := config.WriteByteTimeout + if writeByteTimeout == 0 { + writeByteTimeout = defaultWebH2WriteByteTimeout + } ctx, cancel := context.WithCancel(context.Background()) return &WebH2Client{ address: config.ServerAddress, @@ -148,6 +161,7 @@ func newWebH2ClientWithSigner(config WebH2ClientConfig, auth *webAuthSigner, cla DisableCompression: true, StrictMaxConcurrentStreams: true, MaxHeaderListSize: defaultWebClientMaxResponseHeaderBytes, + WriteByteTimeout: writeByteTimeout, }, utlsSessionCache: utlsSessionCache, ctx: ctx, diff --git a/internal/tunnel/web_h2_write_timeout_integration_test.go b/internal/tunnel/web_h2_write_timeout_integration_test.go new file mode 100644 index 0000000..1f87bfd --- /dev/null +++ b/internal/tunnel/web_h2_write_timeout_integration_test.go @@ -0,0 +1,346 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +const webH2IntegrationWriteTimeout = 300 * time.Millisecond + +func TestWebH2WriteByteTimeoutPreservesHealthyTLS(t *testing.T) { + for _, profile := range []FingerprintProfile{FingerprintNative, FingerprintChrome133} { + t.Run(string(profile), func(t *testing.T) { + client := newWebH2WriteTimeoutIntegrationClient(t, profile) + target := startWebTCPEcho(t) + sibling := dialWebH2WriteTimeoutStream(t, client, target) + assertWebH2WriteTimeoutEcho(t, sibling) + session := verifiedWebH2WriteTimeoutSession(t, client) + // This option is a blocked-write timeout, not an idle timeout. No + // TLS writes are pending during this deliberately longer interval. + time.Sleep(2*webH2IntegrationWriteTimeout + 50*time.Millisecond) + reused := dialWebH2WriteTimeoutStream(t, client, target) + assertWebH2WriteTimeoutEcho(t, reused) + assertWebH2DeadlineSessionUnchanged(t, client, session) + + address, payload, started, result := startWebH2WriteTimeoutReply(t) + conn := dialWebH2WriteTimeoutStream(t, client, address) + if _, err := io.WriteString(conn, "complete upload"); err != nil { + t.Fatal(err) + } + if err := conn.(interface{ CloseWrite() error }).CloseWrite(); err != nil { + t.Fatal(err) + } + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("half-closed upload did not start its reply") + } + assertWebH2WriteTimeoutEcho(t, sibling) + // The twelve spaced response writes take longer than the physical + // write timeout; upload FIN must not truncate this download. + assertWebEOFBody(t, conn, payload) + if err := <-result; err != nil { + t.Fatal(err) + } + assertWebH2WriteTimeoutEcho(t, sibling) + assertWebH2DeadlineSessionUnchanged(t, client, session) + }) + } +} + +func TestWebH2WriteByteTimeoutReachesTLSRawConnection(t *testing.T) { + for _, profile := range []FingerprintProfile{FingerprintNative, FingerprintChrome133} { + t.Run(string(profile), func(t *testing.T) { + client := newWebH2WriteTimeoutIntegrationClient(t, profile) + target := startWebTCPEcho(t) + conn := dialWebH2WriteTimeoutStream(t, client, target) + sibling := dialWebH2WriteTimeoutStream(t, client, target) + assertWebH2WriteTimeoutEcho(t, conn) + assertWebH2WriteTimeoutEcho(t, sibling) + session := verifiedWebH2WriteTimeoutSession(t, client) + wire := session.raw.(*webH2WriteTimeoutWire) + wire.bridge.paused.Store(true) + + writeDone, siblingDone := make(chan error, 1), make(chan error, 1) + // Exceed x/net's 512 KiB request scratch buffer: accepting a + // smaller caller Write is not proof its bytes reached the wire. + go func() { _, err := conn.Write(bytes.Repeat([]byte("upload"), 128<<10)); writeDone <- err }() + go func() { var data [1]byte; _, err := sibling.Read(data[:]); siblingDone <- err }() + // Even a failed assertion must release and join our I/O workers. + t.Cleanup(func() { + _ = client.Close() + for _, done := range []<-chan error{writeDone, siblingDone} { + select { + case <-done: + case <-time.After(2 * time.Second): + t.Error("TLS stall worker did not stop") + } + } + }) + select { + case <-wire.bridge.entered: + case <-time.After(2 * time.Second): + t.Fatal("encrypted client writes did not reach the paused byte bridge") + } + select { + case observed := <-wire.timedOut: + if !errors.Is(observed.err, os.ErrDeadlineExceeded) || observed.deadline.IsZero() { + t.Fatalf("underlying write error/deadline = %v/%v", observed.err, observed.deadline) + } + if observed.budget <= 0 || observed.budget > webH2IntegrationWriteTimeout+100*time.Millisecond { + t.Fatalf("TLS forwarded write deadline budget %v, want at most %v", observed.budget, webH2IntegrationWriteTimeout) + } + case <-time.After(2 * time.Second): + t.Fatal("real blocked raw Write did not return its deadline error") + } + // This is a physical TLS failure, so all streams on that session + // fail. Unlike per-stream deadlines, it cannot preserve siblings. + for _, done := range []chan error{writeDone, siblingDone} { + select { + case err := <-done: + done <- err // Leave a completion receipt for cleanup as well. + if err == nil { + t.Fatal("stalled physical connection left stream I/O successful") + } + case <-time.After(2 * time.Second): + t.Fatal("physical write timeout did not release sibling/upload") + } + } + if session.h2.CanTakeNewRequest() { + t.Fatal("timed-out physical session remained reusable") + } + fresh := dialWebH2WriteTimeoutStream(t, client, target) + assertWebH2WriteTimeoutEcho(t, fresh) + if verifiedWebH2WriteTimeoutSession(t, client) == session { + t.Fatal("recovery reused the failed TLS connection") + } + }) + } +} + +func newWebH2WriteTimeoutIntegrationClient(t *testing.T, profile FingerprintProfile) *WebH2Client { + t.Helper() + serverTLS, clientTLS := testTLSConfigs(t) + server := startWebH2TestServer(t, WebH2ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + Dialer: &net.Dialer{}, Cover: http.NotFoundHandler(), + }) + client, err := NewWebH2Client(WebH2ClientConfig{ + ServerAddress: server.Addr().String(), Token: webTestToken, TLSConfig: clientTLS, + FingerprintProfile: profile, WriteByteTimeout: webH2IntegrationWriteTimeout, + }) + if err != nil { + t.Fatal(err) + } + var mu sync.Mutex + var bridges []*webH2WriteTimeoutBridge + client.dialer = transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + tcp, err := (&net.Dialer{}).DialContext(ctx, network, address) + if err != nil { + return nil, err + } + raw, peer := net.Pipe() + bridge := &webH2WriteTimeoutBridge{raw: raw, peer: peer, tcp: tcp, entered: make(chan struct{}), closed: make(chan struct{}), done: make(chan struct{})} + wire := &webH2WriteTimeoutWire{Conn: raw, bridge: bridge, timedOut: make(chan webH2WriteTimeoutObservation, 1)} + mu.Lock() + bridges = append(bridges, bridge) + mu.Unlock() + var copies sync.WaitGroup + copies.Add(2) + go func() { defer copies.Done(); defer bridge.close(); _, _ = io.Copy(tcp, bridge) }() + go func() { defer copies.Done(); defer bridge.close(); _, _ = io.Copy(peer, tcp) }() + go func() { copies.Wait(); close(bridge.done) }() + return wire, nil + }) + t.Cleanup(func() { + _ = client.Close() + mu.Lock() + all := append([]*webH2WriteTimeoutBridge(nil), bridges...) + mu.Unlock() + for _, bridge := range all { + bridge.close() + select { + case <-bridge.done: + case <-time.After(2 * time.Second): + t.Error("TLS byte bridge did not stop") + } + } + }) + return client +} + +func dialWebH2WriteTimeoutStream(t *testing.T, client *WebH2Client, target string) net.Conn { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + conn, err := client.DialContext(ctx, "tcp", target) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +func verifiedWebH2WriteTimeoutSession(t *testing.T, client *WebH2Client) *webH2ClientSession { + t.Helper() + client.mu.Lock() + session := client.current + client.mu.Unlock() + if session == nil { + t.Fatal("missing authenticated TLS session") + } + state := session.conn.ConnectionState() + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != webH2ALPN || len(state.VerifiedChains) == 0 { + t.Fatalf("expected verified TLS 1.3 and h2, got version=%x ALPN=%q chains=%d", state.Version, state.NegotiatedProtocol, len(state.VerifiedChains)) + } + return session +} + +func assertWebH2WriteTimeoutEcho(t *testing.T, conn net.Conn) { + t.Helper() + assertWebSessionSiblingEcho(t, conn) + if err := conn.SetDeadline(time.Time{}); err != nil { + t.Fatal(err) + } +} + +func startWebH2WriteTimeoutReply(t *testing.T) (string, []byte, <-chan struct{}, <-chan error) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + started, done, result := make(chan struct{}), make(chan struct{}), make(chan error, 1) + payload := bytes.Repeat([]byte("response after upload FIN\n"), 12) + go func() { + defer close(done) + result <- func() error { + conn, err := listener.Accept() + if err != nil { + return err + } + defer conn.Close() + stop := context.AfterFunc(ctx, func() { _ = conn.Close() }) + defer stop() + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + request, err := io.ReadAll(conn) + if err != nil || string(request) != "complete upload" { + return fmt.Errorf("half-closed request=%q error=%v", request, err) + } + ticker := time.NewTicker(50 * time.Millisecond) + defer ticker.Stop() + chunk := len(payload) / 12 + for i := range 12 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + if _, err := conn.Write(payload[i*chunk : (i+1)*chunk]); err != nil { + return err + } + if i == 0 { + close(started) + } + } + return nil + }() + }() + t.Cleanup(func() { + cancel() + _ = listener.Close() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Error("delayed half-close target did not stop") + } + }) + return listener.Addr().String(), payload, started, result +} + +// The bridge forwards actual TLS bytes to the production loopback H2 server. +// Pausing its read gives net.Pipe a deterministic real raw-write stall without +// fabricating TLS state, I/O errors, or timeout notifications. This is not a +// kernel TCP congestion or real-network performance experiment. +type webH2WriteTimeoutBridge struct { + raw, peer, tcp net.Conn + paused atomic.Bool + entered chan struct{} + closed chan struct{} + done chan struct{} + enterOnce sync.Once + closeOnce sync.Once +} + +func (b *webH2WriteTimeoutBridge) Read(data []byte) (int, error) { + if b.paused.Load() { + b.enterOnce.Do(func() { close(b.entered) }) + <-b.closed + return 0, net.ErrClosed + } + return b.peer.Read(data) +} + +func (b *webH2WriteTimeoutBridge) close() { + b.closeOnce.Do(func() { + close(b.closed) + _ = b.raw.Close() + _ = b.peer.Close() + _ = b.tcp.Close() + }) +} + +type webH2WriteTimeoutObservation struct { + err error + deadline time.Time + budget time.Duration +} + +type webH2WriteTimeoutWire struct { + net.Conn + bridge *webH2WriteTimeoutBridge + mu sync.Mutex + deadline time.Time + budget time.Duration + timedOut chan webH2WriteTimeoutObservation +} + +func (w *webH2WriteTimeoutWire) SetWriteDeadline(deadline time.Time) error { + w.mu.Lock() + w.deadline, w.budget = deadline, time.Until(deadline) + w.mu.Unlock() + return w.Conn.SetWriteDeadline(deadline) +} + +func (w *webH2WriteTimeoutWire) Write(data []byte) (int, error) { + n, err := w.Conn.Write(data) + if errors.Is(err, os.ErrDeadlineExceeded) { + w.mu.Lock() + observed := webH2WriteTimeoutObservation{err: err, deadline: w.deadline, budget: w.budget} + w.mu.Unlock() + select { + case w.timedOut <- observed: + default: + } + } + return n, err +} + +func (w *webH2WriteTimeoutWire) Close() error { + w.bridge.close() + return nil +} diff --git a/internal/tunnel/web_h2_write_timeout_test.go b/internal/tunnel/web_h2_write_timeout_test.go new file mode 100644 index 0000000..97c05bf --- /dev/null +++ b/internal/tunnel/web_h2_write_timeout_test.go @@ -0,0 +1,246 @@ +package tunnel + +import ( + "bytes" + "context" + "errors" + "io" + "net" + "net/http" + "os" + "testing" + "time" + + "golang.org/x/net/http2" + "golang.org/x/net/http2/hpack" +) + +func TestWebH2WriteByteTimeoutConfiguration(t *testing.T) { + _, clientTLS := testTLSConfigs(t) + for _, tc := range []struct { + name string + value, want time.Duration + }{ + {"default", 0, 30 * time.Second}, + {"override", 2 * time.Second, 2 * time.Second}, + {"negative", -time.Second, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + client, err := NewWebH2Client(WebH2ClientConfig{ + ServerAddress: "127.0.0.1:443", Token: webTestToken, TLSConfig: clientTLS, + WriteByteTimeout: tc.value, + }) + if tc.value < 0 { + if err == nil { + _ = client.Close() + t.Fatal("negative timeout accepted") + } + return + } + if err != nil { + t.Fatal(err) + } + defer client.Close() + if got := client.transport.WriteByteTimeout; got != tc.want { + t.Fatalf("transport timeout = %v, want %v", got, tc.want) + } + }) + } +} + +func TestWebClientH2WriteByteTimeoutConfiguration(t *testing.T) { + _, clientTLS := testTLSConfigs(t) + for _, tc := range []struct { + name string + value, want time.Duration + }{ + {"default", 0, 30 * time.Second}, + {"override", 2 * time.Second, 2 * time.Second}, + {"negative", -time.Second, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + client, err := NewWebClient(WebClientConfig{ + ServerAddress: "127.0.0.1:443", Token: webTestToken, TLSConfig: clientTLS, + H2WriteByteTimeout: tc.value, + }) + if tc.value < 0 { + if err == nil { + _ = client.Close() + t.Fatal("negative timeout accepted") + } + return + } + if err != nil { + t.Fatal(err) + } + defer client.Close() + h2 := client.fallback.dialer.(*WebH2Client) + if got := h2.transport.WriteByteTimeout; got != tc.want { + t.Fatalf("fallback timeout = %v, want %v", got, tc.want) + } + }) + } +} + +func TestWebH2WriteByteTimeoutReleasesUnreadResponse(t *testing.T) { + for _, mode := range []string{"current close", "current deadline", "retired close"} { + t.Run(mode, func(t *testing.T) { + // Use the production constructor's transport, not an independently + // configured http2.Transport that could hide missing option wiring. + _, clientTLS := testTLSConfigs(t) + client, err := NewWebH2Client(WebH2ClientConfig{ + ServerAddress: "127.0.0.1:443", Token: webTestToken, TLSConfig: clientTLS, + WriteByteTimeout: 250 * time.Millisecond, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = client.Close() }) + stalled, conn := newWebH2UnreadResponse(t, client, mode != "retired close") + if mode == "current deadline" { + released := make(chan struct{}) + release := conn.onClose + conn.onClose = func() { release(); close(released) } + if err := conn.SetDeadline(time.Now()); err != nil { + t.Fatal(err) + } + select { + case <-released: + case <-time.After(time.Second): + t.Fatal("deadline did not start closing the stream") + } + } + closed := make(chan error, 1) + go func() { closed <- conn.Close() }() + select { + case err := <-closed: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + _ = client.Close() + select { + case <-closed: + case <-time.After(time.Second): + t.Fatal("stream Close stayed blocked after physical client Close") + } + t.Fatal("stream Close waited indefinitely behind a stalled HTTP/2 write") + } + if mode == "current deadline" { + _, err := conn.Read(make([]byte, 8)) + if !errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("expired Read = %v, want deadline exceeded", err) + } + } + // Body.Close can return as soon as the write lock is released, before + // x/net's request cleanup has closed the failed physical session. This + // test only checks cleanup: real next-Dial recovery is covered by the + // TLS integration tests, without manually retiring this current session. + wire := stalled.raw.(*webH2StalledWriter) + deadline := time.Now().Add(time.Second) + for (stalled.h2.CanTakeNewRequest() || wire.closes.Load() == 0) && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if stalled.h2.CanTakeNewRequest() || wire.closes.Load() == 0 { + t.Fatal("stalled physical session was not closed") + } + client.mu.Lock() + _, retained := client.sessions[stalled] + active := stalled.active + client.mu.Unlock() + if active != 0 || (mode == "retired close" && retained) { + t.Fatalf("closed stream active=%d retired session retained=%v", active, retained) + } + }) + } +} + +// The peer acknowledges startup, sends a successful response and unread DATA, +// then stops reading after another SETTINGS. Its ACK holds x/net's write mutex, +// which response Body.Close also needs to return unread flow-control credit. +func newWebH2UnreadResponse(t *testing.T, client *WebH2Client, current bool) (*webH2ClientSession, *webH2Conn) { + t.Helper() + raw, peer := net.Pipe() + wire := &webH2StalledWriter{Conn: raw, started: make(chan struct{})} + t.Cleanup(func() { _ = raw.Close(); _ = peer.Close() }) + _ = peer.SetDeadline(time.Now().Add(3 * time.Second)) + peerDone := make(chan error, 1) + go func() { + peerDone <- func() error { + prefix := make([]byte, len(http2.ClientPreface)) + if _, err := io.ReadFull(peer, prefix); err != nil { + return err + } + fr := http2.NewFramer(peer, peer) + for range 2 { + if _, err := fr.ReadFrame(); err != nil { + return err + } + } + if err := fr.WriteSettings(); err != nil { + return err + } + var streamID uint32 + acknowledged := false + for streamID == 0 || !acknowledged { + frame, err := fr.ReadFrame() + if err != nil { + return err + } + if headers, ok := frame.(*http2.HeadersFrame); ok { + streamID = headers.StreamID + } + if settings, ok := frame.(*http2.SettingsFrame); ok && settings.IsAck() { + acknowledged = true + } + } + var headers bytes.Buffer + if err := hpack.NewEncoder(&headers).WriteField(hpack.HeaderField{Name: ":status", Value: "200"}); err != nil { + return err + } + if err := fr.WriteHeaders(http2.HeadersFrameParam{StreamID: streamID, BlockFragment: headers.Bytes(), EndHeaders: true}); err != nil { + return err + } + if err := fr.WriteData(streamID, false, []byte("ab")); err != nil { + return err + } + wire.signal.Store(true) + return fr.WriteSettings() + }() + }() + h2, err := client.transport.NewClientConn(wire) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = raw.Close(); _ = h2.Close() }) + streamCtx, streamCancel := context.WithTimeout(client.ctx, 3*time.Second) + t.Cleanup(streamCancel) + requestReader, requestWriter := io.Pipe() + t.Cleanup(func() { _ = requestReader.Close(); _ = requestWriter.Close() }) + request, err := http.NewRequestWithContext(streamCtx, http.MethodConnect, "https://target.invalid:443", requestReader) + if err != nil { + t.Fatal(err) + } + response, err := h2.RoundTrip(request) + if err != nil { + t.Fatal(err) + } + select { + case <-wire.started: + case <-time.After(time.Second): + t.Fatal("peer did not stall the HTTP/2 writer") + } + if err := <-peerDone; err != nil { + t.Fatal(err) + } + session := &webH2ClientSession{raw: wire, h2: h2, active: 1, authState: webH2ClientAuthReady} + client.mu.Lock() + client.sessions[session] = struct{}{} + if current { + client.current = session + } + client.mu.Unlock() + conn := newWebH2Conn(response.Body, requestWriter, streamCancel, raw.LocalAddr(), raw.RemoteAddr()) + conn.onClose = func() { client.releaseSessionStream(session) } + return session, conn +}