diff --git a/internal/s3fs/resilient.go b/internal/s3fs/resilient.go new file mode 100644 index 0000000..6227870 --- /dev/null +++ b/internal/s3fs/resilient.go @@ -0,0 +1,109 @@ +package s3fs + +import ( + "context" + "net/http" + "time" + + awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http" + "github.com/aws/aws-sdk-go-v2/service/s3" +) + +// Stale keep-alive connections are the failure this guards against. The S3 +// client storage-go builds uses the AWS SDK default transport: IdleConnTimeout +// 90s and — unlike storage-go's own bundle client — no ResponseHeaderTimeout. +// Tigris (or an intermediary/NAT) silently drops idle connections well before +// 90s, so a pooled connection can be dead while the client still believes it is +// usable. A request written to such a connection is never answered; with no +// response-header deadline and the call's context.TODO() carrying no timeout, +// the read blocks forever. +// +// The lazy reader (see file.go) made this acute: where the old eager reader did +// one GetObject per pack file in a burst on fresh connections, the lazy reader +// issues many small GetObjects strung out across go-git's processing — long +// enough for pooled connections to go stale before reuse. A single clone of a +// real repository reliably wedges. +// +// hardenedTimeouts bound those two windows: prune idle connections before +// Tigris does (so they are never reused stale), and fail a stalled reused +// connection fast (so the SDK retryer retries the idempotent request on a fresh +// connection instead of hanging). ResponseHeaderTimeout bounds only the wait +// for response headers, not body streaming, so large pack reads are unaffected. +const ( + hardenedIdleConnTimeout = 30 * time.Second + hardenedResponseHeaderTimeout = 30 * time.Second +) + +// newHardenedHTTPClient returns a single AWS HTTP client whose transport prunes +// idle connections early and times out the wait for response headers. It must +// be shared across all calls so its connection pool is reused; a per-call client +// would defeat pooling. +func newHardenedHTTPClient() *awshttp.BuildableClient { + return awshttp.NewBuildableClient().WithTransportOptions(func(t *http.Transport) { + t.IdleConnTimeout = hardenedIdleConnTimeout + t.ResponseHeaderTimeout = hardenedResponseHeaderTimeout + }) +} + +// resilientClient wraps an s3Client, injecting a shared hardened HTTP client +// into every request via a per-call option. The embedded s3Client supplies any +// method not overridden below; all of them are, so the option reaches every S3 +// round-trip the filesystem makes. +type resilientClient struct { + s3Client + opt func(*s3.Options) +} + +// Harden wraps a Tigris/S3 client so every request it makes carries an HTTP +// client that fails fast on stale keep-alive connections rather than hanging +// forever. Pass the result to NewS3FS and NewListingCache. See the package +// constants above for the rationale. +func Harden(c s3Client) s3Client { + hc := newHardenedHTTPClient() + return resilientClient{ + s3Client: c, + opt: func(o *s3.Options) { o.HTTPClient = hc }, + } +} + +// withOpt prepends the hardening option so an explicit per-call option still +// takes precedence (later options win in the SDK's option application). +func (c resilientClient) withOpt(opts []func(*s3.Options)) []func(*s3.Options) { + return append([]func(*s3.Options){c.opt}, opts...) +} + +func (c resilientClient) HeadObject(ctx context.Context, in *s3.HeadObjectInput, opts ...func(*s3.Options)) (*s3.HeadObjectOutput, error) { + return c.s3Client.HeadObject(ctx, in, c.withOpt(opts)...) +} + +func (c resilientClient) GetObject(ctx context.Context, in *s3.GetObjectInput, opts ...func(*s3.Options)) (*s3.GetObjectOutput, error) { + return c.s3Client.GetObject(ctx, in, c.withOpt(opts)...) +} + +func (c resilientClient) PutObject(ctx context.Context, in *s3.PutObjectInput, opts ...func(*s3.Options)) (*s3.PutObjectOutput, error) { + return c.s3Client.PutObject(ctx, in, c.withOpt(opts)...) +} + +func (c resilientClient) ListObjectsV2(ctx context.Context, in *s3.ListObjectsV2Input, opts ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) { + return c.s3Client.ListObjectsV2(ctx, in, c.withOpt(opts)...) +} + +func (c resilientClient) DeleteObject(ctx context.Context, in *s3.DeleteObjectInput, opts ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) { + return c.s3Client.DeleteObject(ctx, in, c.withOpt(opts)...) +} + +func (c resilientClient) RenameObject(ctx context.Context, in *s3.CopyObjectInput, opts ...func(*s3.Options)) (*s3.CopyObjectOutput, error) { + return c.s3Client.RenameObject(ctx, in, c.withOpt(opts)...) +} + +func (c resilientClient) CreateMultipartUpload(ctx context.Context, in *s3.CreateMultipartUploadInput, opts ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) { + return c.s3Client.CreateMultipartUpload(ctx, in, c.withOpt(opts)...) +} + +func (c resilientClient) UploadPart(ctx context.Context, in *s3.UploadPartInput, opts ...func(*s3.Options)) (*s3.UploadPartOutput, error) { + return c.s3Client.UploadPart(ctx, in, c.withOpt(opts)...) +} + +func (c resilientClient) CompleteMultipartUpload(ctx context.Context, in *s3.CompleteMultipartUploadInput, opts ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) { + return c.s3Client.CompleteMultipartUpload(ctx, in, c.withOpt(opts)...) +} diff --git a/internal/s3fs/resilient_test.go b/internal/s3fs/resilient_test.go new file mode 100644 index 0000000..240b2c3 --- /dev/null +++ b/internal/s3fs/resilient_test.go @@ -0,0 +1,146 @@ +package s3fs + +import ( + "context" + "testing" + + awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http" + "github.com/aws/aws-sdk-go-v2/service/s3" +) + +// optRecorder is an s3Client that records the per-call option functions it +// receives for each method, so a test can confirm Harden injects its hardened +// HTTP client on every S3 round-trip. The embedded nil s3Client is never used: +// the wrapper calls only the methods overridden below. +type optRecorder struct { + s3Client + last []func(*s3.Options) +} + +func (r *optRecorder) HeadObject(_ context.Context, _ *s3.HeadObjectInput, opts ...func(*s3.Options)) (*s3.HeadObjectOutput, error) { + r.last = opts + return &s3.HeadObjectOutput{}, nil +} + +func (r *optRecorder) GetObject(_ context.Context, _ *s3.GetObjectInput, opts ...func(*s3.Options)) (*s3.GetObjectOutput, error) { + r.last = opts + return &s3.GetObjectOutput{}, nil +} + +func (r *optRecorder) PutObject(_ context.Context, _ *s3.PutObjectInput, opts ...func(*s3.Options)) (*s3.PutObjectOutput, error) { + r.last = opts + return &s3.PutObjectOutput{}, nil +} + +func (r *optRecorder) ListObjectsV2(_ context.Context, _ *s3.ListObjectsV2Input, opts ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) { + r.last = opts + return &s3.ListObjectsV2Output{}, nil +} + +func (r *optRecorder) DeleteObject(_ context.Context, _ *s3.DeleteObjectInput, opts ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) { + r.last = opts + return &s3.DeleteObjectOutput{}, nil +} + +func (r *optRecorder) RenameObject(_ context.Context, _ *s3.CopyObjectInput, opts ...func(*s3.Options)) (*s3.CopyObjectOutput, error) { + r.last = opts + return &s3.CopyObjectOutput{}, nil +} + +func (r *optRecorder) CreateMultipartUpload(_ context.Context, _ *s3.CreateMultipartUploadInput, opts ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) { + r.last = opts + return &s3.CreateMultipartUploadOutput{}, nil +} + +func (r *optRecorder) UploadPart(_ context.Context, _ *s3.UploadPartInput, opts ...func(*s3.Options)) (*s3.UploadPartOutput, error) { + r.last = opts + return &s3.UploadPartOutput{}, nil +} + +func (r *optRecorder) CompleteMultipartUpload(_ context.Context, _ *s3.CompleteMultipartUploadInput, opts ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) { + r.last = opts + return &s3.CompleteMultipartUploadOutput{}, nil +} + +// TestHardenInjectsHardenedHTTPClient verifies that every method on a Harden-ed +// client carries the hardened HTTP client (bounded ResponseHeaderTimeout and a +// reduced IdleConnTimeout) into its per-call options. Without this, S3 requests +// run on the AWS default transport, whose stale keep-alive connections hang +// forever — the clone-hang root cause this fix addresses. +func TestHardenInjectsHardenedHTTPClient(t *testing.T) { + ctx := context.Background() + + tests := []struct { + name string + invoke func(c s3Client) error + }{ + {"HeadObject", func(c s3Client) error { _, err := c.HeadObject(ctx, &s3.HeadObjectInput{}); return err }}, + {"GetObject", func(c s3Client) error { _, err := c.GetObject(ctx, &s3.GetObjectInput{}); return err }}, + {"PutObject", func(c s3Client) error { _, err := c.PutObject(ctx, &s3.PutObjectInput{}); return err }}, + {"ListObjectsV2", func(c s3Client) error { _, err := c.ListObjectsV2(ctx, &s3.ListObjectsV2Input{}); return err }}, + {"DeleteObject", func(c s3Client) error { _, err := c.DeleteObject(ctx, &s3.DeleteObjectInput{}); return err }}, + {"RenameObject", func(c s3Client) error { _, err := c.RenameObject(ctx, &s3.CopyObjectInput{}); return err }}, + {"CreateMultipartUpload", func(c s3Client) error { + _, err := c.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{}) + return err + }}, + {"UploadPart", func(c s3Client) error { _, err := c.UploadPart(ctx, &s3.UploadPartInput{}); return err }}, + {"CompleteMultipartUpload", func(c s3Client) error { + _, err := c.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{}) + return err + }}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + rec := &optRecorder{} + if err := tt.invoke(Harden(rec)); err != nil { + t.Fatalf("%s: %v", tt.name, err) + } + if len(rec.last) == 0 { + t.Fatalf("%s: no per-call options injected; stale connections would hang", tt.name) + } + + // Apply the recorded options the way the SDK would and inspect the + // resulting HTTP client's transport. + var o s3.Options + for _, fn := range rec.last { + fn(&o) + } + if o.HTTPClient == nil { + t.Fatalf("%s: HTTPClient not set on options", tt.name) + } + bc, ok := o.HTTPClient.(*awshttp.BuildableClient) + if !ok { + t.Fatalf("%s: HTTPClient is %T, want *awshttp.BuildableClient", tt.name, o.HTTPClient) + } + tr := bc.GetTransport() + if tr.ResponseHeaderTimeout != hardenedResponseHeaderTimeout { + t.Errorf("%s: ResponseHeaderTimeout = %v, want %v", tt.name, tr.ResponseHeaderTimeout, hardenedResponseHeaderTimeout) + } + if tr.IdleConnTimeout != hardenedIdleConnTimeout { + t.Errorf("%s: IdleConnTimeout = %v, want %v", tt.name, tr.IdleConnTimeout, hardenedIdleConnTimeout) + } + }) + } +} + +// TestHardenExplicitOptionWins confirms an explicit per-call option still +// overrides the injected default (later options win), so callers retain control. +func TestHardenExplicitOptionWins(t *testing.T) { + rec := &optRecorder{} + override := awshttp.NewBuildableClient() + _, err := Harden(rec).GetObject(context.Background(), &s3.GetObjectInput{}, + func(o *s3.Options) { o.HTTPClient = override }) + if err != nil { + t.Fatal(err) + } + + var o s3.Options + for _, fn := range rec.last { + fn(&o) + } + if got, _ := o.HTTPClient.(*awshttp.BuildableClient); got != override { + t.Errorf("explicit HTTPClient did not win: got %p, want %p", got, override) + } +}