S3: ping a silent HTTP/2 connection and drop it

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.

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)
 		}
 	})
 }

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
 }
 

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{})