Repository: benbjohnson/litestream

Issue:
Upgrade AWS S3 SDK from v1 to v2
## Description

Litestream is currently using v1 of the AWS SDK for Go. This should be upgraded to v2 for better performance, updated features, and ongoing support.

## Current State

- Using AWS SDK for Go v1
- v1 is in maintenance mode with limited updates
- v2 has been stable and recommended for new projects

## Benefits of Upgrading

1. **Performance Improvements**
   - Better connection management
   - Reduced memory allocations
   - Improved retry logic

2. **Modern Go Patterns**
   - Context-based cancellation
   - Better error handling
   - Type-safe configuration

3. **Active Development**
   - Regular updates and bug fixes
   - New AWS features support
   - Security updates

## Migration Scope

- Update S3 client initialization
- Update all S3 API calls to v2 syntax
- Update configuration handling
- Update error handling patterns
- Test with various S3-compatible services (AWS S3, MinIO, etc.)

## Breaking Changes to Consider

- Configuration syntax changes
- API method signature changes
- Error type changes

## Testing Requirements

- Test with AWS S3
- Test with S3-compatible services (MinIO, Backblaze B2, etc.)
- Verify backward compatibility with existing configurations
- Performance benchmarks before/after

## References

- [AWS SDK for Go v2 Migration Guide](https://aws.github.io/aws-sdk-go-v2/docs/migrating/)
- [AWS SDK for Go v2 Documentation](https://aws.github.io/aws-sdk-go-v2/docs/)

## Additional Context

This upgrade was requested by Ben Johnson as part of modernizing the Litestream codebase.

Hints from issue comments:

Solution patch:
diff --git a/etc/s3_mock.py b/etc/s3_mock.py
index 8b61dec..1c9b42b 100755
--- a/etc/s3_mock.py
+++ b/etc/s3_mock.py
@@ -25,7 +25,7 @@ server.start()
 s3 = boto3.client(
     "s3",
     aws_access_key_id=env["LITESTREAM_S3_ACCESS_KEY_ID"],
-    aws_secret_access_key=["LITESTREAM_S3_SECRET_ACCESS_KEY"],
+    aws_secret_access_key=env["LITESTREAM_S3_SECRET_ACCESS_KEY"],
     endpoint_url=env["LITESTREAM_S3_ENDPOINT"]
 ).create_bucket(Bucket=env["LITESTREAM_S3_BUCKET"])
 
diff --git a/go.mod b/go.mod
index 2df4e61..b750cfb 100644
--- a/go.mod
+++ b/go.mod
@@ -9,7 +9,12 @@ require (
 	filippo.io/age v1.1.1
 	github.com/Azure/azure-storage-blob-go v0.15.0
 	github.com/MadAppGang/httplog v1.3.0
-	github.com/aws/aws-sdk-go v1.49.5
+	github.com/aws/aws-sdk-go-v2 v1.37.1
+	github.com/aws/aws-sdk-go-v2/config v1.30.2
+	github.com/aws/aws-sdk-go-v2/credentials v1.18.2
+	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.2
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.85.1
+	github.com/aws/smithy-go v1.22.5
 	github.com/mark3labs/mcp-go v0.32.0
 	github.com/mattn/go-shellwords v1.0.12
 	github.com/mattn/go-sqlite3 v1.14.19
@@ -30,6 +35,19 @@ require (
 	cloud.google.com/go/iam v1.1.5 // indirect
 	github.com/Azure/azure-pipeline-go v0.2.3 // indirect
 	github.com/TylerBrock/colorjson v0.0.0-20200706003622-8a50f05110d2 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.0 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.8.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.26.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.31.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.35.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/fatih/color v1.13.0 // indirect
@@ -42,7 +60,6 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
 	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
-	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/kr/fs v0.1.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-ieproxy v0.0.11 // indirect
diff --git a/go.sum b/go.sum
index 9d8d3f6..86f436e 100644
--- a/go.sum
+++ b/go.sum
@@ -31,8 +31,44 @@ github.com/MadAppGang/httplog v1.3.0 h1:1XU54TO8kiqTeO+7oZLKAM3RP/cJ7SadzslRcKsp
 github.com/MadAppGang/httplog v1.3.0/go.mod h1:gpYEdkjh/Cda6YxtDy4AB7KY+fR7mb3SqBZw74A5hJ4=
 github.com/TylerBrock/colorjson v0.0.0-20200706003622-8a50f05110d2 h1:ZBbLwSJqkHBuFDA6DUhhse0IGJ7T5bemHyNILUjvOq4=
 github.com/TylerBrock/colorjson v0.0.0-20200706003622-8a50f05110d2/go.mod h1:VSw57q4QFiWDbRnjdX8Cb3Ow0SFncRw+bA/ofY6Q83w=
-github.com/aws/aws-sdk-go v1.49.5 h1:y2yfBlwjPDi3/sBVKeznYEdDy6wIhjA2L5NCBMLUIYA=
-github.com/aws/aws-sdk-go v1.49.5/go.mod h1:LF8svs817+Nz+DmiMQKTO3ubZ/6IaTpq3TjupRn3Eqk=
+github.com/aws/aws-sdk-go-v2 v1.37.1 h1:SMUxeNz3Z6nqGsXv0JuJXc8w5YMtrQMuIBmDx//bBDY=
+github.com/aws/aws-sdk-go-v2 v1.37.1/go.mod h1:9Q0OoGQoboYIAJyslFyF1f5K1Ryddop8gqMhWx/n4Wg=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.0 h1:6GMWV6CNpA/6fbFHnoAjrv4+LGfyTqZz2LtCHnspgDg=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.0/go.mod h1:/mXlTIVG9jbxkqDnr5UQNQxW1HRYxeGklkM9vAFeabg=
+github.com/aws/aws-sdk-go-v2/config v1.30.2 h1:YE1BmSc4fFYqFgN1mN8uzrtc7R9x+7oSWeX8ckoltAw=
+github.com/aws/aws-sdk-go-v2/config v1.30.2/go.mod h1:UNrLGZ6jfAVjgVJpkIxjLufRJqTXCVYOpkeVf83kwBo=
+github.com/aws/aws-sdk-go-v2/credentials v1.18.2 h1:mfm0GKY/PHLhs7KO0sUaOtFnIQ15Qqxt+wXbO/5fIfs=
+github.com/aws/aws-sdk-go-v2/credentials v1.18.2/go.mod h1:v0SdJX6ayPeZFQxgXUKw5RhLpAoZUuynxWDfh8+Eknc=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.1 h1:owmNBboeA0kHKDcdF8KiSXmrIuXZustfMGGytv6OMkM=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.1/go.mod h1:Bg1miN59SGxrZqlP8vJZSmXW+1N8Y1MjQDq1OfuNod8=
+github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.2 h1:YFX4DvH1CPQXgQR8935b46Om+L7+6jus4aTdKqyDR84=
+github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.2/go.mod h1:DgMPy7GqxcV0RSyaITnI3rw8HC3lIHB87U3KPQKDxHg=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1 h1:ksZXBYv80EFTcgc8OJO48aQ8XDWXIQL7gGasPeCoTzI=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1/go.mod h1:HSksQyyJETVZS7uM54cir0IgxttTD+8aEoJMPGepHBI=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1 h1:+dn/xF/05utS7tUhjIcndbuaPjfll2LhbH1cCDGLYUQ=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1/go.mod h1:hyAGz30LHdm5KBZDI58MXx5lDVZ5CUfvfTZvMu4HCZo=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.1 h1:4HbnOGE9491a9zYJ9VpPh1ApgEq6ZlD4Kuv1PJenFpc=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.1/go.mod h1:Z6QnHC6TmpJWUxAy8FI4JzA7rTwl6EIANkyK9OR5z5w=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.0 h1:6+lZi2JeGKtCraAj1rpoZfKqnQ9SptseRZioejfUOLM=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.0/go.mod h1:eb3gfbVIxIoGgJsi9pGne19dhCBpK6opTYpQqAmdy44=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.8.1 h1:ps3nrmBWdWwakZBydGX1CxeYFK80HsQ79JLMwm7Y4/c=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.8.1/go.mod h1:bAdfrfxENre68Hh2swNaGEVuFYE74o0SaSCAlaG9E74=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.1 h1:ky79ysLMxhwk5rxJtS+ILd3Mc8kC5fhsLBrP27r6h4I=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.1/go.mod h1:+2MmkvFvPYM1vsozBWduoLJUi5maxFk5B7KJFECujhY=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.1 h1:MdVYlN5pcQu1t1OYx4Ajo3fKl1IEhzgdPQbYFCRjYS8=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.1/go.mod h1:iikmNLrvHm2p4a3/4BPeix2S9P+nW8yM1IZW73x8bFA=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.85.1 h1:Hsqo8+dFxSdDvv9B2PgIx1AJAnDpqgS0znVI+R+MoGY=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.85.1/go.mod h1:8Q0TAPXD68Z8YqlcIGHs/UNIDHsxErV9H4dl4vJEpgw=
+github.com/aws/aws-sdk-go-v2/service/sso v1.26.1 h1:uWaz3DoNK9MNhm7i6UGxqufwu3BEuJZm72WlpGwyVtY=
+github.com/aws/aws-sdk-go-v2/service/sso v1.26.1/go.mod h1:ILpVNjL0BO+Z3Mm0SbEeUoYS9e0eJWV1BxNppp0fcb8=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.31.1 h1:XdG6/o1/ZDmn3wJU5SRAejHaWgKS4zHv0jBamuKuS2k=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.31.1/go.mod h1:oiotGTKadCOCl3vg/tYh4k45JlDF81Ka8rdumNhEnIQ=
+github.com/aws/aws-sdk-go-v2/service/sts v1.35.1 h1:iF4Xxkc0H9c/K2dS0zZw3SCkj0Z7n6AMnUiiyoJND+I=
+github.com/aws/aws-sdk-go-v2/service/sts v1.35.1/go.mod h1:0bxIatfN0aLq4mjoLDeBpOjOke68OsFlXPDFJ7V0MYw=
+github.com/aws/smithy-go v1.22.5 h1:P9ATCXPMb2mPjYBgueqJNCA5S9UfktsW0tTxi+a7eqw=
+github.com/aws/smithy-go v1.22.5/go.mod h1:t1ufH5HMublsJYulve2RKmHDC15xu1f26kHCp/HgceI=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -105,10 +141,6 @@ github.com/googleapis/gax-go/v2 v2.12.0 h1:A+gCJKdRfqXkr+BIRGtZLibNXf0m1f9E4HG56
 github.com/googleapis/gax-go/v2 v2.12.0/go.mod h1:y+aIqrI5eb1YGMVJfuV3185Ts/D7qKpsEkdD5+I6QGU=
 github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f h1:7LYC+Yfkj3CTRcShK0KOL/w6iTiKyqqBA9a41Wnggw8=
 github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f/go.mod h1:pFlLw2CfqZiIBOx6BuCeRLCrfxBJipTY0nIOF/VbGcI=
-github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
-github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
-github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8=
-github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8=
 github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
@@ -303,7 +335,6 @@ google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqw
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
-gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/s3/replica_client.go b/s3/replica_client.go
index 2900e62..e9cb10c 100644
--- a/s3/replica_client.go
+++ b/s3/replica_client.go
@@ -3,27 +3,31 @@ package s3
 import (
 	"context"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net"
 	"net/http"
+	"net/url"
 	"os"
 	"path"
 	"regexp"
+	"strings"
 	"sync"
 	"time"
 
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/awserr"
-	"github.com/aws/aws-sdk-go/aws/credentials"
-	"github.com/aws/aws-sdk-go/aws/session"
-	"github.com/aws/aws-sdk-go/service/s3"
-	"github.com/aws/aws-sdk-go/service/s3/s3manager"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/aws/smithy-go"
+	"github.com/aws/smithy-go/middleware"
+	smithyhttp "github.com/aws/smithy-go/transport/http"
 	"github.com/superfly/ltx"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/benbjohnson/litestream"
-	"github.com/benbjohnson/litestream/internal"
 )
 
 // ReplicaClientType is the client type for this package.
@@ -40,8 +44,8 @@ var _ litestream.ReplicaClient = (*ReplicaClient)(nil)
 // ReplicaClient is a client for writing LTX files to S3.
 type ReplicaClient struct {
 	mu       sync.Mutex
-	s3       *s3.S3 // s3 service
-	uploader *s3manager.Uploader
+	s3       *s3.Client // s3 service
+	uploader *manager.Uploader
 
 	// AWS authentication keys.
 	AccessKeyID     string
@@ -54,6 +58,10 @@ type ReplicaClient struct {
 	Endpoint       string
 	ForcePathStyle bool
 	SkipVerify     bool
+
+	// Upload configuration
+	PartSize    int64 // Part size for multipart uploads (default: 5MB)
+	Concurrency int   // Number of concurrent parts to upload (default: 5)
 }
 
 // NewReplicaClient returns a new instance of ReplicaClient.
@@ -75,6 +83,11 @@ func (c *ReplicaClient) Init(ctx context.Context) (err error) {
 		return nil
 	}
 
+	// Validate required configuration
+	if c.Bucket == "" {
+		return fmt.Errorf("s3: bucket name is required")
+	}
+
 	// Look up region if not specified and no endpoint is used.
 	// Endpoints are typically used for non-S3 object stores and do not
 	// necessarily require a region.
@@ -89,395 +102,530 @@ func (c *ReplicaClient) Init(ctx context.Context) (err error) {
 		}
 	}
 
-	// Create new AWS session.
-	config := c.config()
-	if region != "" {
-		config.Region = aws.String(region)
+	// Create HTTP client with 24 hour timeout for long-running operations
+	httpClient := &http.Client{
+		Timeout: 24 * time.Hour,
 	}
 
-	sess, err := session.NewSession(config)
-	if err != nil {
-		return fmt.Errorf("cannot create aws session: %w", err)
+	// Configure transport for insecure connections if needed
+	if c.SkipVerify {
+		httpClient.Transport = &http.Transport{
+			Proxy: http.ProxyFromEnvironment,
+			DialContext: (&net.Dialer{
+				Timeout:   30 * time.Second,
+				KeepAlive: 30 * time.Second,
+			}).DialContext,
+			ForceAttemptHTTP2:     true,
+			MaxIdleConns:          100,
+			IdleConnTimeout:       90 * time.Second,
+			TLSHandshakeTimeout:   10 * time.Second,
+			ExpectContinueTimeout: 1 * time.Second,
+			TLSClientConfig: &tls.Config{
+				InsecureSkipVerify: true,
+			},
+		}
 	}
-	c.s3 = s3.New(sess)
-	c.uploader = s3manager.NewUploader(sess)
-	return nil
-}
 
-// config returns the AWS configuration. Uses the default credential chain
-// unless a key/secret are explicitly set.
-func (c *ReplicaClient) config() *aws.Config {
-	config := &aws.Config{}
+	// Build configuration options
+	configOpts := []func(*config.LoadOptions) error{
+		config.WithRegion(region),
+		// Use adaptive retry mode for better resilience with 24 hour timeout
+		// This matches Azure's approach for long-running operations
+		config.WithRetryMode(aws.RetryModeAdaptive),
+		config.WithRetryMaxAttempts(10), // Increase retry attempts for resilience
+	}
 
-	if c.AccessKeyID != "" || c.SecretAccessKey != "" {
-		config.Credentials = credentials.NewStaticCredentials(c.AccessKeyID, c.SecretAccessKey, "")
+	// Add HTTP client with proper timeout
+	configOpts = append(configOpts, config.WithHTTPClient(httpClient))
+
+	// Add static credentials if provided, otherwise use default credential chain
+	// Default credential chain includes:
+	// - Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
+	// - Shared credentials file (~/.aws/credentials)
+	// - EC2 Instance Profile credentials
+	// - ECS Task Role credentials
+	// - Web Identity Token credentials (for EKS)
+	if c.AccessKeyID != "" && c.SecretAccessKey != "" {
+		configOpts = append(configOpts, config.WithCredentialsProvider(
+			credentials.NewStaticCredentialsProvider(c.AccessKeyID, c.SecretAccessKey, ""),
+		))
 	}
-	if c.Endpoint != "" {
-		config.Endpoint = aws.String(c.Endpoint)
+
+	// Load AWS configuration
+	cfg, err := config.LoadDefaultConfig(ctx, configOpts...)
+	if err != nil {
+		return fmt.Errorf("cannot load aws config: %w", err)
 	}
-	if c.ForcePathStyle {
-		config.S3ForcePathStyle = aws.Bool(c.ForcePathStyle)
+
+	// Create S3 client options
+	s3Opts := []func(*s3.Options){
+		func(o *s3.Options) {
+			o.UsePathStyle = c.ForcePathStyle
+			// Add User-Agent for telemetry (similar to Azure's ApplicationID)
+			o.APIOptions = append(o.APIOptions, func(stack *middleware.Stack) error {
+				return stack.Build.Add(
+					middleware.BuildMiddlewareFunc(
+						"LitestreamUserAgent",
+						func(ctx context.Context, in middleware.BuildInput, next middleware.BuildHandler) (
+							out middleware.BuildOutput, metadata middleware.Metadata, err error,
+						) {
+							if req, ok := in.Request.(*smithyhttp.Request); ok {
+								req.Header.Add("User-Agent", "litestream")
+							}
+							return next.HandleBuild(ctx, in)
+						},
+					),
+					middleware.After,
+				)
+			})
+		},
 	}
-	if c.SkipVerify {
-		config.HTTPClient = &http.Client{Transport: &http.Transport{
-			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-		}}
+
+	// Add custom endpoint if specified
+	c.configureEndpoint(&s3Opts)
+
+	// Create S3 client
+	c.s3 = s3.NewFromConfig(cfg, s3Opts...)
+
+	// Configure uploader with custom options if specified
+	uploaderOpts := []func(*manager.Uploader){}
+	if c.PartSize > 0 {
+		uploaderOpts = append(uploaderOpts, func(u *manager.Uploader) {
+			u.PartSize = c.PartSize
+		})
 	}
+	if c.Concurrency > 0 {
+		uploaderOpts = append(uploaderOpts, func(u *manager.Uploader) {
+			u.Concurrency = c.Concurrency
+		})
+	}
+	c.uploader = manager.NewUploader(c.s3, uploaderOpts...)
 
-	return config
+	return nil
 }
 
-func (c *ReplicaClient) findBucketRegion(ctx context.Context, bucket string) (string, error) {
-	// Connect to US standard region to fetch info.
-	config := c.config()
-	config.Region = aws.String(DefaultRegion)
-	sess, err := session.NewSession(config)
-	if err != nil {
-		return "", err
+// configureEndpoint adds custom endpoint configuration to S3 client options if needed.
+func (c *ReplicaClient) configureEndpoint(opts *[]func(*s3.Options)) {
+	if c.Endpoint != "" {
+		*opts = append(*opts, func(o *s3.Options) {
+			o.UsePathStyle = c.ForcePathStyle
+			o.BaseEndpoint = aws.String(c.Endpoint)
+			// For MinIO and other S3-compatible services
+			if strings.HasPrefix(c.Endpoint, "http://") {
+				o.EndpointOptions.DisableHTTPS = true
+			}
+		})
 	}
+}
 
-	// Fetch bucket location, if possible. Must be bucket owner.
-	// This call can return a nil location which means it's in us-east-1.
-	if out, err := s3.New(sess).GetBucketLocation(&s3.GetBucketLocationInput{
-		Bucket: aws.String(bucket),
-	}); err != nil {
-		return "", err
-	} else if out.LocationConstraint != nil {
-		return *out.LocationConstraint, nil
+// findBucketRegion looks up the AWS region for a bucket. Returns blank if non-S3.
+func (c *ReplicaClient) findBucketRegion(ctx context.Context, bucket string) (string, error) {
+	// Build a config with credentials but no region
+	configOpts := []func(*config.LoadOptions) error{}
+
+	// Add static credentials if provided
+	if c.AccessKeyID != "" && c.SecretAccessKey != "" {
+		configOpts = append(configOpts, config.WithCredentialsProvider(
+			credentials.NewStaticCredentialsProvider(c.AccessKeyID, c.SecretAccessKey, ""),
+		))
 	}
-	return DefaultRegion, nil
-}
 
-// DeleteAll deletes all LTX files.
-func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
-	if err := c.Init(ctx); err != nil {
-		return err
+	// Load AWS configuration
+	cfg, err := config.LoadDefaultConfig(ctx, configOpts...)
+	if err != nil {
+		return "", fmt.Errorf("cannot load aws config for region lookup: %w", err)
 	}
 
-	// Collect all files.
-	var objIDs []*s3.ObjectIdentifier
-	if err := c.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
-		Bucket: aws.String(c.Bucket),
-		Prefix: aws.String(c.Path),
-	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
-		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
+	// Use default region for initial region lookup
+	cfg.Region = DefaultRegion
 
-		for _, obj := range page.Contents {
-			objIDs = append(objIDs, &s3.ObjectIdentifier{Key: obj.Key})
-		}
-		return true
-	}); err != nil {
-		return err
-	}
+	// Create S3 client options
+	s3Opts := []func(*s3.Options){}
 
-	// Delete all files in batches.
-	for len(objIDs) > 0 {
-		n := MaxKeys
-		if len(objIDs) < n {
-			n = len(objIDs)
-		}
+	// Configure custom endpoint for region lookup
+	c.configureEndpoint(&s3Opts)
 
-		out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
-			Bucket: aws.String(c.Bucket),
-			Delete: &s3.Delete{Objects: objIDs[:n], Quiet: aws.Bool(true)},
-		})
-		if err != nil {
-			return err
-		}
-		if err := deleteOutputError(out); err != nil {
-			return err
-		}
-		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
+	client := s3.NewFromConfig(cfg, s3Opts...)
 
-		objIDs = objIDs[n:]
+	// Get bucket location
+	out, err := client.GetBucketLocation(ctx, &s3.GetBucketLocationInput{
+		Bucket: aws.String(bucket),
+	})
+	if err != nil {
+		return "", err
 	}
 
-	// log.Printf("%s(%s): retainer: deleting: %s", r.db.Path(), r.Name())
-
-	return nil
+	// Convert location constraint to region
+	if out.LocationConstraint == "" {
+		return DefaultRegion, nil
+	}
+	return string(out.LocationConstraint), nil
 }
 
-// LTXFiles returns an iterator over all available LTX files.
+// LTXFiles returns an iterator over all LTX files on the replica for the given level.
 func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
 	}
-	return newLTXFileIterator(ctx, c, level, seek), nil
+	return newFileIterator(ctx, c, level, seek), nil
 }
 
-// WriteLTXFile writes an LTX file from rd into a remote location.
-func (c *ReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, rd io.Reader) (info *ltx.FileInfo, err error) {
+// OpenLTXFile returns a reader that contains an LTX file at a given TXID.
+func (c *ReplicaClient) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID) (io.ReadCloser, error) {
 	if err := c.Init(ctx); err != nil {
-		return info, err
+		return nil, err
 	}
 
-	key := litestream.LTXFilePath(c.Path, level, minTXID, maxTXID)
-	startTime := time.Now()
-
-	rc := internal.NewReadCounter(rd)
-	if _, err := c.uploader.UploadWithContext(ctx, &s3manager.UploadInput{
+	// Build the key from the file info
+	filename := ltx.FormatFilename(minTXID, maxTXID)
+	key := c.Path + "/" + fmt.Sprintf("%04x/%s", level, filename)
+	out, err := c.s3.GetObject(ctx, &s3.GetObjectInput{
 		Bucket: aws.String(c.Bucket),
 		Key:    aws.String(key),
-		Body:   rc,
-	}); err != nil {
-		return info, err
+	})
+	if err != nil {
+		if isNotExists(err) {
+			return nil, os.ErrNotExist
+		}
+		return nil, fmt.Errorf("get object %s: %w", key, err)
 	}
-
-	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "PUT").Inc()
-	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "PUT").Add(float64(rc.N()))
-
-	return &ltx.FileInfo{
-		Level:     level,
-		MinTXID:   minTXID,
-		MaxTXID:   maxTXID,
-		Size:      rc.N(),
-		CreatedAt: startTime.UTC(),
-	}, nil
+	return out.Body, nil
 }
 
-// OpenLTXFile returns a reader for an LTX file
-// Returns os.ErrNotExist if no matching index/offset is found.
-func (c *ReplicaClient) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID) (io.ReadCloser, error) {
+// WriteLTXFile writes an LTX file to the replica.
+func (c *ReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
 	}
 
-	key := litestream.LTXFilePath(c.Path, level, minTXID, maxTXID)
-	out, err := c.s3.GetObjectWithContext(ctx, &s3.GetObjectInput{
+	filename := ltx.FormatFilename(minTXID, maxTXID)
+	key := c.Path + "/" + fmt.Sprintf("%04x/%s", level, filename)
+	out, err := c.uploader.Upload(ctx, &s3.PutObjectInput{
 		Bucket: aws.String(c.Bucket),
 		Key:    aws.String(key),
+		Body:   r,
 	})
-	if isNotExists(err) {
-		return nil, os.ErrNotExist
-	} else if err != nil {
-		return nil, err
+	if err != nil {
+		return nil, fmt.Errorf("upload to %s: %w", key, err)
 	}
-	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "GET").Inc()
-	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "GET").Add(float64(aws.Int64Value(out.ContentLength)))
 
-	return out.Body, nil
+	// Build file info from the uploaded file
+	info := &ltx.FileInfo{
+		Level:     level,
+		MinTXID:   minTXID,
+		MaxTXID:   maxTXID,
+		CreatedAt: time.Now(),
+	}
+
+	// ETag indicates successful upload
+	if out.ETag == nil {
+		return nil, fmt.Errorf("upload failed: no ETag returned")
+	}
+
+	return info, nil
 }
 
-// DeleteLTXFiles deletes a set of LTX files.
+// DeleteLTXFiles deletes one or more LTX files.
 func (c *ReplicaClient) DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error {
 	if err := c.Init(ctx); err != nil {
 		return err
 	}
 
-	objIDs := make([]*s3.ObjectIdentifier, MaxKeys)
-	for len(a) > 0 {
-		n := MaxKeys
-		if len(a) < n {
-			n = len(a)
-		}
+	if len(a) == 0 {
+		return nil
+	}
 
-		// Generate a batch of object IDs for deleting the LTX files.
-		for i, info := range a[:n] {
-			key := litestream.LTXFilePath(c.Path, info.Level, info.MinTXID, info.MaxTXID)
-			objIDs[i] = &s3.ObjectIdentifier{Key: &key}
-		}
+	// Convert file infos to object identifiers
+	objIDs := make([]types.ObjectIdentifier, 0, len(a))
+	for _, info := range a {
+		filename := ltx.FormatFilename(info.MinTXID, info.MaxTXID)
+		key := c.Path + "/" + fmt.Sprintf("%04x/%s", info.Level, filename)
+		objIDs = append(objIDs, types.ObjectIdentifier{Key: aws.String(key)})
+	}
 
-		// Delete S3 objects in bulk.
-		out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+	// Delete in batches
+	for len(objIDs) > 0 {
+		n := min(len(objIDs), MaxKeys)
+
+		out, err := c.s3.DeleteObjects(ctx, &s3.DeleteObjectsInput{
 			Bucket: aws.String(c.Bucket),
-			Delete: &s3.Delete{Objects: objIDs[:n], Quiet: aws.Bool(true)},
+			Delete: &types.Delete{Objects: objIDs[:n], Quiet: aws.Bool(true)},
 		})
 		if err != nil {
+			return fmt.Errorf("delete batch of %d objects: %w", n, err)
+		} else if err := deleteOutputError(out); err != nil {
 			return err
 		}
-		if err := deleteOutputError(out); err != nil {
+
+		objIDs = objIDs[n:]
+	}
+
+	return nil
+}
+
+// DeleteAll deletes all files.
+func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
+	if err := c.Init(ctx); err != nil {
+		return err
+	}
+
+	var objIDs []types.ObjectIdentifier
+
+	// Create paginator for listing objects
+	paginator := s3.NewListObjectsV2Paginator(c.s3, &s3.ListObjectsV2Input{
+		Bucket: aws.String(c.Bucket),
+		Prefix: aws.String(c.Path + "/"),
+	})
+
+	// Iterate through all pages
+	for paginator.HasMorePages() {
+		page, err := paginator.NextPage(ctx)
+		if err != nil {
+			return fmt.Errorf("list objects page: %w", err)
+		}
+
+		// Collect object identifiers
+		for _, obj := range page.Contents {
+			objIDs = append(objIDs, types.ObjectIdentifier{Key: obj.Key})
+		}
+	}
+
+	// Delete all collected objects in batches
+	for len(objIDs) > 0 {
+		n := min(len(objIDs), MaxKeys)
+
+		out, err := c.s3.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: aws.String(c.Bucket),
+			Delete: &types.Delete{Objects: objIDs[:n], Quiet: aws.Bool(true)},
+		})
+		if err != nil {
+			return fmt.Errorf("delete all batch of %d objects: %w", n, err)
+		} else if err := deleteOutputError(out); err != nil {
 			return err
 		}
-		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
 
-		a = a[n:]
+		objIDs = objIDs[n:]
 	}
 
 	return nil
 }
 
-type ltxFileIterator struct {
+// fileIterator represents an iterator over LTX files in S3.
+type fileIterator struct {
+	ctx    context.Context
+	cancel context.CancelFunc
 	client *ReplicaClient
 	level  int
 	seek   ltx.TXID
 
-	ch     chan *ltx.FileInfo
-	g      errgroup.Group
-	ctx    context.Context
-	cancel func()
+	paginator *s3.ListObjectsV2Paginator
+	page      *s3.ListObjectsV2Output
+	pageIndex int
 
-	info *ltx.FileInfo
-	err  error
+	closed bool
+	err    error
+	info   *ltx.FileInfo
 }
 
-func newLTXFileIterator(ctx context.Context, client *ReplicaClient, level int, seek ltx.TXID) *ltxFileIterator {
-	itr := &ltxFileIterator{
+func newFileIterator(ctx context.Context, client *ReplicaClient, level int, seek ltx.TXID) *fileIterator {
+	ctx, cancel := context.WithCancel(ctx)
+
+	itr := &fileIterator{
+		ctx:    ctx,
+		cancel: cancel,
 		client: client,
 		level:  level,
 		seek:   seek,
-		ch:     make(chan *ltx.FileInfo),
 	}
 
-	itr.ctx, itr.cancel = context.WithCancel(ctx)
-	itr.g.Go(itr.fetch)
+	// Create paginator for listing objects with level prefix
+	prefix := client.Path + "/" + fmt.Sprintf("%04x/", level)
+	itr.paginator = s3.NewListObjectsV2Paginator(client.s3, &s3.ListObjectsV2Input{
+		Bucket: aws.String(client.Bucket),
+		Prefix: aws.String(prefix),
+	})
 
 	return itr
 }
 
-// fetch runs in a separate goroutine to fetch pages of objects and stream them to a channel.
-func (itr *ltxFileIterator) fetch() error {
-	defer close(itr.ch)
+// Close stops iteration.
+func (itr *fileIterator) Close() (err error) {
+	itr.closed = true
+	itr.cancel()
+	return nil
+}
 
-	dir := litestream.LTXLevelDir(itr.client.Path, itr.level)
-	prefix := dir + "/"
-	if itr.seek != 0 {
-		prefix += itr.seek.String()
+// Next returns the next file. Returns false when no more files are available.
+func (itr *fileIterator) Next() bool {
+	if itr.closed || itr.err != nil {
+		return false
 	}
 
-	return itr.client.s3.ListObjectsPagesWithContext(itr.ctx, &s3.ListObjectsInput{
-		Bucket:    aws.String(itr.client.Bucket),
-		Prefix:    aws.String(prefix),
-		Delimiter: aws.String("/"),
-	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
-		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
+	// Process objects until we find a valid LTX file
+	for {
+		// Load next page if needed
+		if itr.page == nil || itr.pageIndex >= len(itr.page.Contents) {
+			if !itr.paginator.HasMorePages() {
+				return false
+			}
 
-		for _, obj := range page.Contents {
-			key := path.Base(aws.StringValue(obj.Key))
+			var err error
+			itr.page, err = itr.paginator.NextPage(itr.ctx)
+			if err != nil {
+				itr.err = err
+				return false
+			}
+			itr.pageIndex = 0
+		}
+
+		// Process current object
+		if itr.pageIndex < len(itr.page.Contents) {
+			obj := itr.page.Contents[itr.pageIndex]
+			itr.pageIndex++
+
+			// Extract file info from key
+			key := path.Base(aws.ToString(obj.Key))
 			minTXID, maxTXID, err := ltx.ParseFilename(key)
 			if err != nil {
-				continue
+				continue // Skip non-LTX files
 			}
 
+			// Build file info
 			info := &ltx.FileInfo{
-				Level:     itr.level,
-				MinTXID:   minTXID,
-				MaxTXID:   maxTXID,
-				Size:      aws.Int64Value(obj.Size),
-				CreatedAt: obj.LastModified.UTC(),
+				Level:   itr.level,
+				MinTXID: minTXID,
+				MaxTXID: maxTXID,
 			}
 
-			select {
-			case <-itr.ctx.Done():
-				return false
-			case itr.ch <- info:
+			// Skip if below seek TXID
+			if info.MinTXID < itr.seek {
+				continue
 			}
-		}
-		return true
-	})
-}
 
-func (itr *ltxFileIterator) Close() (err error) {
-	err = itr.err
+			// Skip if wrong level
+			if info.Level != itr.level {
+				continue
+			}
 
-	// Cancel context and wait for error group to finish.
-	itr.cancel()
-	if e := itr.g.Wait(); e != nil && err == nil {
-		err = e
+			// Set file info
+			info.Size = aws.ToInt64(obj.Size)
+			info.CreatedAt = aws.ToTime(obj.LastModified)
+			itr.info = info
+			return true
+		}
 	}
+}
 
-	return err
+// Item returns the metadata for the current file.
+func (itr *fileIterator) Item() *ltx.FileInfo {
+	return itr.info
 }
 
-func (itr *ltxFileIterator) Next() bool {
-	// Exit if an error has already occurred.
-	if itr.err != nil {
-		return false
+// Err returns any error that occurred during iteration.
+func (itr *fileIterator) Err() error {
+	return itr.err
+}
+
+// ParseURL parses an S3 URL into its host and path parts.
+// If endpoint is set, it can override the host.
+func ParseURL(s, endpoint string) (bucket, region, key string, err error) {
+	u, err := url.Parse(s)
+	if err != nil {
+		return "", "", "", err
 	}
 
-	// Return false if context was canceled or if there are no more segments.
-	// Otherwise fetch the next segment and store it on the iterator.
-	select {
-	case <-itr.ctx.Done():
-		return false
-	case info, ok := <-itr.ch:
-		if !ok {
-			return false
+	if u.Scheme != "s3" {
+		return "", "", "", fmt.Errorf("invalid s3 url scheme")
+	}
+
+	// Special handling for filebase.com
+	if u.Host == "filebase.com" {
+		parts := strings.SplitN(strings.TrimPrefix(u.Path, "/"), "/", 2)
+		if len(parts) == 0 {
+			return "", "", "", fmt.Errorf("s3 bucket required")
+		}
+		bucket = parts[0]
+		if len(parts) > 1 {
+			key = parts[1]
 		}
-		itr.info = info
-		return true
+		return bucket, "", key, nil
 	}
-}
 
-func (itr *ltxFileIterator) Err() error { return itr.err }
+	// For other hosts, check if it's a special endpoint
+	bucket, region, _, _ = ParseHost(u.Host)
+	if bucket == "" {
+		bucket = u.Host
+	}
 
-func (itr *ltxFileIterator) Item() *ltx.FileInfo {
-	return itr.info
+	key = strings.TrimPrefix(u.Path, "/")
+	return bucket, region, key, nil
 }
 
-// ParseHost extracts data from a hostname depending on the service provider.
-func ParseHost(s string) (bucket, region, endpoint string, forcePathStyle bool) {
-	// Extract port if one is specified.
-	host, port, err := net.SplitHostPort(s)
-	if err != nil {
-		host = s
-	}
-
-	// Default to path-based URLs, except for with AWS S3 itself.
-	forcePathStyle = true
-
-	// Extract fields from provider-specific host formats.
-	scheme := "https"
-	if a := localhostRegex.FindStringSubmatch(host); a != nil {
-		bucket, region = a[1], "us-east-1"
-		scheme, endpoint = "http", "localhost"
-	} else if a := backblazeRegex.FindStringSubmatch(host); a != nil {
-		bucket, region = a[1], a[2]
-		endpoint = fmt.Sprintf("s3.%s.backblazeb2.com", region)
-	} else if a := filebaseRegex.FindStringSubmatch(host); a != nil {
-		bucket, endpoint = a[1], "s3.filebase.com"
-	} else if a := digitalOceanRegex.FindStringSubmatch(host); a != nil {
-		bucket, region = a[1], a[2]
-		endpoint = fmt.Sprintf("%s.digitaloceanspaces.com", region)
-	} else if a := scalewayRegex.FindStringSubmatch(host); a != nil {
-		bucket, region = a[1], a[2]
-		endpoint = fmt.Sprintf("s3.%s.scw.cloud", region)
-	} else if a := linodeRegex.FindStringSubmatch(host); a != nil {
-		bucket, region = a[1], a[2]
-		endpoint = fmt.Sprintf("%s.linodeobjects.com", region)
-	} else {
-		bucket = host
-		forcePathStyle = false
-	}
-
-	// Add port back to endpoint, if available.
-	if endpoint != "" && port != "" {
-		endpoint = net.JoinHostPort(endpoint, port)
+// ParseHost parses the host/endpoint for an S3-like storage system.
+// Endpoints: https://docs.aws.amazon.com/general/latest/gr/s3.html
+func ParseHost(host string) (bucket, region, endpoint string, forcePathStyle bool) {
+	// Check for MinIO-style hosts (bucket.host:port)
+	if strings.Contains(host, ":") && !strings.Contains(host, ".com") {
+		parts := strings.SplitN(host, ".", 2)
+		if len(parts) == 2 {
+			// Extract bucket from bucket.host:port format
+			bucket = parts[0]
+			endpoint = "http://" + parts[1]
+			return bucket, DefaultRegion, endpoint, true
+		}
+		// No bucket in host, just host:port
+		return "", "", "http://" + host, true
 	}
 
-	// Prepend scheme to endpoint.
-	if endpoint != "" {
-		endpoint = scheme + "://" + endpoint
+	// Check common object storage providers
+	if a := digitaloceanRegex.FindStringSubmatch(host); len(a) > 1 {
+		region = a[2]
+		return "", region, fmt.Sprintf("https://%s.digitaloceanspaces.com", region), false
+	} else if a := backblazeRegex.FindStringSubmatch(host); len(a) > 1 {
+		region = a[2]
+		bucket = a[1]
+		endpoint = fmt.Sprintf("https://s3.%s.backblazeb2.com", region)
+		return bucket, region, endpoint, true
+	} else if a := filebaseRegex.FindStringSubmatch(host); len(a) > 1 {
+		bucket = a[1]
+		endpoint = "s3.filebase.com"
+		return bucket, "", endpoint, false
+	} else if a := scalewayRegex.FindStringSubmatch(host); len(a) > 1 {
+		region = a[2]
+		bucket = a[1]
+		endpoint = fmt.Sprintf("s3.%s.scw.cloud", region)
+		return bucket, region, endpoint, false
 	}
 
-	return bucket, region, endpoint, forcePathStyle
+	// For standard S3, the host is the bucket name
+	return host, "", "", false
 }
 
 var (
-	localhostRegex    = regexp.MustCompile(`^(?:(.+)\.)?localhost$`)
+	digitaloceanRegex = regexp.MustCompile(`^(?:(.+)\.)?([^.]+)\.digitaloceanspaces.com$`)
 	backblazeRegex    = regexp.MustCompile(`^(?:(.+)\.)?s3.([^.]+)\.backblazeb2.com$`)
 	filebaseRegex     = regexp.MustCompile(`^(?:(.+)\.)?s3.filebase.com$`)
-	digitalOceanRegex = regexp.MustCompile(`^(?:(.+)\.)?([^.]+)\.digitaloceanspaces.com$`)
 	scalewayRegex     = regexp.MustCompile(`^(?:(.+)\.)?s3.([^.]+)\.scw\.cloud$`)
-	linodeRegex       = regexp.MustCompile(`^(?:(.+)\.)?([^.]+)\.linodeobjects.com$`)
 )
 
 func isNotExists(err error) bool {
-	switch err := err.(type) {
-	case awserr.Error:
-		return err.Code() == `NoSuchKey`
-	default:
-		return false
+	var apiErr smithy.APIError
+	if errors.As(err, &apiErr) {
+		return apiErr.ErrorCode() == "NoSuchKey"
 	}
+	return false
 }
 
 func deleteOutputError(out *s3.DeleteObjectsOutput) error {
-	switch len(out.Errors) {
-	case 0:
+	if len(out.Errors) == 0 {
 		return nil
-	case 1:
-		return fmt.Errorf("deleting object %s: %s - %s", aws.StringValue(out.Errors[0].Key), aws.StringValue(out.Errors[0].Code), aws.StringValue(out.Errors[0].Message))
-	default:
-		return fmt.Errorf("%d errors occurred deleting objects, %s: %s - (%s (and %d others)",
-			len(out.Errors), aws.StringValue(out.Errors[0].Key), aws.StringValue(out.Errors[0].Code), aws.StringValue(out.Errors[0].Message), len(out.Errors)-1)
 	}
+
+	// Build generic error
+	var b strings.Builder
+	b.WriteString("failed to delete files:")
+	for _, err := range out.Errors {
+		fmt.Fprintf(&b, "\n%s: %s", aws.ToString(err.Key), aws.ToString(err.Message))
+	}
+	return errors.New(b.String())
 }



Test patch:
diff --git a/replica_client_test.go b/replica_client_test.go
index 7bd09d0..453ef13 100644
--- a/replica_client_test.go
+++ b/replica_client_test.go
@@ -1,6 +1,7 @@
 package litestream_test
 
 import (
+	"bytes"
 	"context"
 	"flag"
 	"fmt"
@@ -326,3 +327,116 @@ func stripLTXFileInfo(info *ltx.FileInfo) *ltx.FileInfo {
 	other.CreatedAt = time.Time{}
 	return &other
 }
+
+// TestReplicaClient_S3_UploaderConfig tests S3 uploader configuration for large files
+func TestReplicaClient_S3_UploaderConfig(t *testing.T) {
+	// Only run for S3 integration tests
+	if !strings.Contains(*integration, "s3") {
+		t.Skip("Skipping S3-specific uploader config test")
+	}
+
+	RunWithReplicaClient(t, "LargeFileWithCustomConfig", func(t *testing.T, c litestream.ReplicaClient) {
+		t.Parallel()
+
+		// Type assert to S3 client to set custom config
+		s3Client, ok := c.(*s3.ReplicaClient)
+		if !ok {
+			t.Skip("Not an S3 client")
+		}
+
+		// Set custom upload configuration
+		s3Client.PartSize = 5 * 1024 * 1024 // 5MB parts
+		s3Client.Concurrency = 3            // 3 concurrent parts
+
+		// Determine file size based on whether we're testing against moto or real S3
+		// Moto has issue #8762 where composite checksums for multipart uploads
+		// don't have the -X suffix, causing checksum validation to fail.
+		// Reference: https://github.com/getmoto/moto/issues/8762
+		size := 10 * 1024 * 1024 // 10MB - triggers multipart upload
+
+		// If we're using moto (localhost endpoint), use smaller file to avoid multipart
+		if s3Client.Endpoint != "" && strings.Contains(s3Client.Endpoint, "127.0.0.1") {
+			size = 4 * 1024 * 1024 // 4MB - avoids multipart upload with moto
+			t.Log("Using 4MB file size to work around moto multipart checksum issue")
+		} else {
+			t.Log("Using 10MB file size to test multipart upload")
+		}
+		data := make([]byte, size)
+		for i := range data {
+			data[i] = byte(i % 256)
+		}
+
+		// Upload the file using bytes.Reader to avoid string conversion issues
+		if _, err := c.WriteLTXFile(context.Background(), 0, ltx.TXID(1), ltx.TXID(100), bytes.NewReader(data)); err != nil {
+			t.Fatalf("failed to write large file: %v", err)
+		}
+
+		// Read it back and verify size
+		r, err := c.OpenLTXFile(context.Background(), 0, ltx.TXID(1), ltx.TXID(100))
+		if err != nil {
+			t.Fatalf("failed to open large file: %v", err)
+		}
+		defer r.Close()
+
+		buf, err := io.ReadAll(r)
+		if err != nil {
+			t.Fatalf("failed to read large file: %v", err)
+		}
+
+		if len(buf) != size {
+			t.Errorf("size mismatch: got %d, want %d", len(buf), size)
+		}
+
+		// Verify the data matches what we uploaded
+		if !bytes.Equal(buf, data) {
+			t.Errorf("data mismatch: uploaded and downloaded data do not match")
+		}
+	})
+}
+
+// TestReplicaClient_S3_ErrorContext tests that S3 errors include helpful context
+func TestReplicaClient_S3_ErrorContext(t *testing.T) {
+	// Only run for S3 integration tests
+	if !strings.Contains(*integration, "s3") {
+		t.Skip("Skipping S3-specific error context test")
+	}
+
+	RunWithReplicaClient(t, "ErrorContext", func(t *testing.T, c litestream.ReplicaClient) {
+		t.Parallel()
+
+		// Test OpenLTXFile with non-existent file
+		_, err := c.OpenLTXFile(context.Background(), 0, ltx.TXID(999), ltx.TXID(999))
+		if err == nil {
+			t.Fatal("expected error for non-existent file")
+		}
+
+		// Should return os.ErrNotExist for S3 NoSuchKey
+		if !os.IsNotExist(err) {
+			t.Errorf("expected os.ErrNotExist, got %v", err)
+		}
+	})
+}
+
+// TestReplicaClient_S3_BucketValidation tests bucket validation in S3 client
+func TestReplicaClient_S3_BucketValidation(t *testing.T) {
+	// Only run for S3 integration tests
+	if !strings.Contains(*integration, "s3") {
+		t.Skip("Skipping S3-specific bucket validation test")
+	}
+
+	// Create a new S3 client with empty bucket
+	c := s3.NewReplicaClient()
+	c.AccessKeyID = *s3AccessKeyID
+	c.SecretAccessKey = *s3SecretAccessKey
+	c.Region = *s3Region
+	c.Bucket = "" // Empty bucket name
+
+	// Should fail with bucket validation error
+	err := c.Init(context.Background())
+	if err == nil {
+		t.Fatal("expected error for empty bucket name")
+	}
+	if !strings.Contains(err.Error(), "bucket name is required") {
+		t.Errorf("expected bucket validation error, got: %v", err)
+	}
+}
diff --git a/s3/replica_client_test.go b/s3/replica_client_test.go
new file mode 100644
index 0000000..e3d1885
--- /dev/null
+++ b/s3/replica_client_test.go
@@ -0,0 +1,261 @@
+package s3
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/smithy-go"
+)
+
+// mockAPIError implements smithy.APIError for testing
+type mockAPIError struct {
+	code    string
+	message string
+}
+
+func (e *mockAPIError) Error() string {
+	return e.message
+}
+
+func (e *mockAPIError) ErrorCode() string {
+	return e.code
+}
+
+func (e *mockAPIError) ErrorMessage() string {
+	return e.message
+}
+
+func (e *mockAPIError) ErrorFault() smithy.ErrorFault {
+	return smithy.FaultUnknown
+}
+
+func TestIsNotExists(t *testing.T) {
+	// Test with NoSuchKey error
+	noSuchKeyErr := &mockAPIError{
+		code:    "NoSuchKey",
+		message: "The specified key does not exist",
+	}
+	if !isNotExists(noSuchKeyErr) {
+		t.Error("isNotExists should return true for NoSuchKey error")
+	}
+
+	// Test with different error code
+	differentErr := &mockAPIError{
+		code:    "AccessDenied",
+		message: "Access denied",
+	}
+	if isNotExists(differentErr) {
+		t.Error("isNotExists should return false for non-NoSuchKey error")
+	}
+
+	// Test with non-API error
+	regularErr := errors.New("regular error")
+	if isNotExists(regularErr) {
+		t.Error("isNotExists should return false for non-API error")
+	}
+
+	// Test with nil error
+	if isNotExists(nil) {
+		t.Error("isNotExists should return false for nil error")
+	}
+
+	// Test with wrapped API error
+	wrappedErr := &mockAPIError{
+		code:    "NoSuchKey",
+		message: "wrapped key error",
+	}
+	if !isNotExists(wrappedErr) {
+		t.Error("isNotExists should return true for wrapped NoSuchKey error")
+	}
+}
+
+// TestReplicaClient_Init_BucketValidation tests that Init validates bucket name
+func TestReplicaClient_Init_BucketValidation(t *testing.T) {
+	t.Run("EmptyBucket", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "" // Empty bucket name
+		c.Region = "us-east-1"
+
+		err := c.Init(context.Background())
+		if err == nil {
+			t.Fatal("expected error for empty bucket name")
+		}
+		if !strings.Contains(err.Error(), "bucket name is required") {
+			t.Errorf("unexpected error: %v", err)
+		}
+	})
+
+	t.Run("ValidBucket", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "test-bucket"
+		c.Region = "us-east-1"
+		// Note: This will fail when trying to connect, but should pass bucket validation
+		err := c.Init(context.Background())
+		// We expect a different error (not bucket validation)
+		if err != nil && strings.Contains(err.Error(), "bucket name is required") {
+			t.Errorf("should not fail bucket validation with valid bucket: %v", err)
+		}
+	})
+}
+
+// TestReplicaClient_UploaderConfiguration tests that uploader configuration is applied
+func TestReplicaClient_UploaderConfiguration(t *testing.T) {
+	t.Run("CustomPartSize", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "test-bucket"
+		c.Region = "us-east-1"
+		c.PartSize = 10 * 1024 * 1024 // 10MB
+		c.Concurrency = 10
+
+		// Verify the configuration is set
+		if c.PartSize != 10*1024*1024 {
+			t.Errorf("expected PartSize to be 10MB, got %d", c.PartSize)
+		}
+		if c.Concurrency != 10 {
+			t.Errorf("expected Concurrency to be 10, got %d", c.Concurrency)
+		}
+	})
+
+	t.Run("DefaultConfiguration", func(t *testing.T) {
+		c := NewReplicaClient()
+		// Verify defaults are zero (will use SDK defaults)
+		if c.PartSize != 0 {
+			t.Errorf("expected default PartSize to be 0, got %d", c.PartSize)
+		}
+		if c.Concurrency != 0 {
+			t.Errorf("expected default Concurrency to be 0, got %d", c.Concurrency)
+		}
+	})
+}
+
+// TestReplicaClient_ConfigureEndpoint tests the endpoint configuration helper
+func TestReplicaClient_ConfigureEndpoint(t *testing.T) {
+	tests := []struct {
+		name           string
+		endpoint       string
+		forcePathStyle bool
+		expectHTTPS    bool
+	}{
+		{
+			name:           "HTTPEndpoint",
+			endpoint:       "http://localhost:9000",
+			forcePathStyle: true,
+			expectHTTPS:    false,
+		},
+		{
+			name:           "HTTPSEndpoint",
+			endpoint:       "https://s3.amazonaws.com",
+			forcePathStyle: false,
+			expectHTTPS:    true,
+		},
+		{
+			name:           "EmptyEndpoint",
+			endpoint:       "",
+			forcePathStyle: false,
+			expectHTTPS:    true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := NewReplicaClient()
+			c.Endpoint = tt.endpoint
+			c.ForcePathStyle = tt.forcePathStyle
+
+			// Test that configureEndpoint can be called without error
+			var opts []func(*s3.Options)
+			c.configureEndpoint(&opts)
+
+			// Verify opts were added when endpoint is set
+			if tt.endpoint != "" && len(opts) == 0 {
+				t.Error("expected endpoint options to be added")
+			}
+			if tt.endpoint == "" && len(opts) != 0 {
+				t.Error("expected no endpoint options for empty endpoint")
+			}
+		})
+	}
+}
+
+// TestReplicaClient_HTTPClientConfiguration tests HTTP client setup
+func TestReplicaClient_HTTPClientConfiguration(t *testing.T) {
+	t.Run("WithSkipVerify", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "test-bucket"
+		c.Region = "us-east-1"
+		c.SkipVerify = true
+
+		// We can't directly test the HTTP client configuration without
+		// actually initializing, but we can verify the flag is set
+		if !c.SkipVerify {
+			t.Error("expected SkipVerify to be true")
+		}
+	})
+
+	t.Run("WithoutSkipVerify", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "test-bucket"
+		c.Region = "us-east-1"
+		c.SkipVerify = false
+
+		if c.SkipVerify {
+			t.Error("expected SkipVerify to be false")
+		}
+	})
+}
+
+// TestReplicaClient_CredentialConfiguration tests credential setup
+func TestReplicaClient_CredentialConfiguration(t *testing.T) {
+	t.Run("WithStaticCredentials", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "test-bucket"
+		c.Region = "us-east-1"
+		c.AccessKeyID = "AKIAIOSFODNN7EXAMPLE"
+		c.SecretAccessKey = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+
+		// Verify credentials are set
+		if c.AccessKeyID == "" || c.SecretAccessKey == "" {
+			t.Error("expected credentials to be set")
+		}
+	})
+
+	t.Run("WithDefaultCredentialChain", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "test-bucket"
+		c.Region = "us-east-1"
+		// Leave AccessKeyID and SecretAccessKey empty
+
+		// Verify credentials are not set (will use default chain)
+		if c.AccessKeyID != "" || c.SecretAccessKey != "" {
+			t.Error("expected credentials to be empty for default chain")
+		}
+	})
+}
+
+// TestReplicaClient_DefaultRegionUsage tests that DefaultRegion constant is used consistently
+func TestReplicaClient_DefaultRegionUsage(t *testing.T) {
+	// Test that DefaultRegion is properly defined
+	if DefaultRegion != "us-east-1" {
+		t.Errorf("expected DefaultRegion to be 'us-east-1', got %s", DefaultRegion)
+	}
+
+	// Test ParseHost uses DefaultRegion
+	t.Run("ParseHost_MinIO", func(t *testing.T) {
+		bucket, region, endpoint, forcePathStyle := ParseHost("mybucket.localhost:9000")
+		if region != DefaultRegion {
+			t.Errorf("expected region to be %s, got %s", DefaultRegion, region)
+		}
+		if bucket != "mybucket" {
+			t.Errorf("expected bucket to be 'mybucket', got %s", bucket)
+		}
+		if !strings.Contains(endpoint, "localhost:9000") {
+			t.Errorf("expected endpoint to contain 'localhost:9000', got %s", endpoint)
+		}
+		if !forcePathStyle {
+			t.Error("expected forcePathStyle to be true for MinIO")
+		}
+	})
+}