From 27db174c5a7f1369a619d22f1666a192fc7aa92b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Mon, 19 Oct 2026 18:53:56 -0700 Subject: [PATCH] S3: ping a silent HTTP/2 connection and drop it Requests share an HTTP/2 connection, and the transports set no health check on it: one that went silent without closing held every request on it until the 5 minute idle timeout, which each new request restarted, so a host's whole bucket traffic could stall. The transports now ping a connection silent for 10 s and close it when the ping goes unanswered for 5 s (or a write makes no progress for 10 s); the next request, or the SDK's retry, dials a new one. --- README.md | 2 +- s3.go | 2 +- storage.go | 16 +++++++++ transport_test.go | 85 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 103 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 4d85384..6566f00 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ Object storage on S3 and S3-compatible stores, GCS, Azure Blob, OCI Object Stora - **S3 semantics** where stores differ. `DeleteObjects` of a missing path succeeds, as S3's delete does. It tries every path, and when some are not deleted fails with a `*DeleteError` whose `Failed` names each one with its error (`errors.As`); every other path is deleted, and a retry is safe. - **Keys** are the store's keys as given. The local directory, as a filesystem, drops a leading `/`, cleans `//` and `.`, refuses keys outside it, and cannot hold both `a` and `a/b`. -- **Timeouts**: a connection that moves no bytes either way for 5 minutes is cut; a slow but live transfer never is. The context bounds the call. +- **Timeouts**: a connection that moves no bytes either way for 5 minutes is cut; a slow but live transfer never is. An HTTP/2 connection silent for 10 s is pinged and closed when the ping goes unanswered for 5 s, so requests sharing it do not wait out the 5 minutes. The context bounds the call. - **Retries**: every store retries throttling, 5xx and network errors with backoff, including AliOSS, whose SDK does not. A stream given to `UploadObject` is sent once on AliOSS. - **Proxy**: `ProxyConfig` (S3, GCS) carries its credentials for http and https targets, honors `NO_PROXY`, and never proxies the cloud metadata services. - **Local writes** go in place, as a file is written, so mounted volumes and file watchers see one write; `IfAbsent` writes a temporary file (`.upload-*`, never listed) and links it into place. diff --git a/s3.go b/s3.go index 6c723e4..42b9ff5 100644 --- a/s3.go +++ b/s3.go @@ -218,7 +218,7 @@ func newS3Client(awsConf *aws.Config, conf *S3Config) *s3.Client { // Options run after the SDK sets up the HTTP client, which a defaults // mode does by replacing the dialer the idle timeout wraps. if b, ok := o.HTTPClient.(*awshttp.BuildableClient); ok { - o.HTTPClient = b.WithTransportOptions(setIdleTimeout) + o.HTTPClient = b.WithTransportOptions(setIdleTimeout, setHealthCheck) } }) } diff --git a/storage.go b/storage.go index be0b0be..c1e4ce1 100644 --- a/storage.go +++ b/storage.go @@ -205,6 +205,21 @@ func setIdleTimeout(t *http.Transport) { } } +// HTTP/2 shares a connection among requests, so one that goes silent +// without closing (a peer or middlebox gone quiet) holds every request on +// it, and new requests keep joining it, until the idle timeout, which each +// new request restarts. A connection silent for healthCheck is pinged, and +// closed when the ping goes unanswered for pingTimeout or a write makes no +// progress for healthCheck, so the next request, or the retry, dials anew. +const ( + healthCheck = 10 * time.Second + pingTimeout = 5 * time.Second +) + +func setHealthCheck(t *http.Transport) { + t.HTTP2 = &http.HTTP2Config{SendPingTimeout: healthCheck, PingTimeout: pingTimeout, WriteByteTimeout: healthCheck} +} + // idleConn fails once no bytes move either way for timeout. Each read or write // moves both deadlines: the transport's read for the response is already // waiting while a long body is still being written. A deadline set from above, @@ -294,6 +309,7 @@ func transportWithTimeouts(rt http.RoundTripper) *http.Transport { t.MaxIdleConnsPerHost = max(t.MaxIdleConnsPerHost, maxIdleConnsPerHost) t.TLSHandshakeTimeout = tlsHandshakeTimeout setIdleTimeout(t) + setHealthCheck(t) return t } diff --git a/transport_test.go b/transport_test.go index a6135bd..bd2af57 100644 --- a/transport_test.go +++ b/transport_test.go @@ -21,9 +21,12 @@ import ( "net/http" "net/http/httptest" "os" + "sync" + "sync/atomic" "testing" "time" + awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http" "github.com/stretchr/testify/require" ) @@ -97,6 +100,88 @@ func TestIdleConnKeepsDeadlines(t *testing.T) { require.Less(t, time.Since(start), 5*time.Second) } +// A connection that goes silent without closing held every request HTTP/2 +// shares it with until the idle timeout, which each new request restarted. +// The S3 client pings a silent connection and closes it, and the next +// request dials a new one. +func TestSilentHTTP2ConnectionIsDropped(t *testing.T) { + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("ok")) + })) + srv.EnableHTTP2 = true + l := &freezable{Listener: srv.Listener, frozen: make(chan struct{})} + srv.Listener = l + srv.StartTLS() + t.Cleanup(func() { close(l.frozen); srv.Close() }) + + s, err := NewS3(&S3Config{AccessKey: "key", Secret: "secret", Region: "us-east-1", Endpoint: srv.URL, ForcePathStyle: true, Bucket: "bucket"}) + require.NoError(t, err) + tr := s.(*s3Storage).client.Options().HTTPClient.(*awshttp.BuildableClient).GetTransport() + tr.TLSClientConfig = srv.Client().Transport.(*http.Transport).TLSClientConfig + client := &http.Client{Transport: tr} + get := func() error { + resp, err := client.Get(srv.URL) + if err != nil { + return err + } + resp.Body.Close() + require.Equal(t, 2, resp.ProtoMajor) + return nil + } + + require.NoError(t, get()) + l.freeze() + start := time.Now() + for range 3 { + _ = get() + } + require.Less(t, time.Since(start), 2*healthCheck+pingTimeout, "requests held past the health check") + // Once the silent connection is closed, a request dials a new one. + require.NoError(t, get()) +} + +// freezable is a listener whose accepted connections stop reading once +// frozen, as a peer gone silent without closing. +type freezable struct { + net.Listener + frozen chan struct{} + mu sync.Mutex + conns []*freezableConn +} + +type freezableConn struct { + net.Conn + frozen atomic.Pointer[chan struct{}] +} + +func (c *freezableConn) Read(p []byte) (int, error) { + if f := c.frozen.Load(); f != nil { + <-*f + } + return c.Conn.Read(p) +} + +func (l *freezable) Accept() (net.Conn, error) { + c, err := l.Listener.Accept() + if err != nil { + return nil, err + } + fc := &freezableConn{Conn: c} + l.mu.Lock() + l.conns = append(l.conns, fc) + l.mu.Unlock() + return fc, nil +} + +// freeze stops every connection accepted so far. +func (l *freezable) freeze() { + l.mu.Lock() + defer l.mu.Unlock() + for _, c := range l.conns { + c.frozen.Store(&l.frozen) + } +} + // stalled is a server that accepts requests and never answers them. func stalled(t *testing.T) string { done := make(chan struct{})