diff --git a/TODO.md b/TODO.md index e8dd4f6..1a1b960 100644 --- a/TODO.md +++ b/TODO.md @@ -22,6 +22,16 @@ the tag exists and is exercised; what is left is merging `next` to # Completed Steps +- 2026-10-07: Made `s3.part_size` set the multipart upload part size + ([issue #232](https://git.eeqj.de/sneak/vaultik/issues/232)). It was + loaded and defaulted but never passed to the S3 client, whose uploader + used a fixed 10MiB part. It now reaches the uploader for `storage_url` + and for the `s3.*` fields, and a part size S3 refuses, below 5MiB or + above 5GiB, `0` included, fails at config load. A blob too large for + S3's limit of 10,000 parts at the configured size is uploaded in larger + parts. The docs gave the default as `5MB`, which the config file reads + as 5,000,000 bytes, below the minimum; they now say `5MiB`. + - 2026-10-07: Made per-name retention work when the hostname contains `_` ([issue #230](https://git.eeqj.de/sneak/vaultik/issues/230)). A snapshot ID is `hostname_name_timestamp`, and the name was read as diff --git a/config.example.yml b/config.example.yml index 6dc50de..3ad5870 100644 --- a/config.example.yml +++ b/config.example.yml @@ -287,10 +287,11 @@ storage_url: "rclone://myremote/path/to/backups" # #use_ssl: true # # # Part size for multipart uploads -# # Minimum 5MB, affects memory usage during upload -# # Supports: 5MB, 10M, 100MiB, etc. -# # Default: 5MB -# #part_size: 5MB +# # Minimum 5MiB, maximum 5GiB; affects memory usage during upload +# # A blob too large for 10,000 parts of this size gets larger parts +# # Supports: 10MB, 16MiB, 100MiB, etc. (5MB is below the minimum) +# # Default: 5MiB +# #part_size: 5MiB # Path to local SQLite index database # This database tracks file state for incremental backups diff --git a/internal/cli/config.go b/internal/cli/config.go index b00a602..545f81a 100644 --- a/internal/cli/config.go +++ b/internal/cli/config.go @@ -205,7 +205,7 @@ storage_url: "" # access_key_id: YOUR_ACCESS_KEY # secret_access_key: YOUR_SECRET_KEY # # region: us-east-1 # Default: us-east-1 -# # part_size: 5MB # Multipart upload part size. Default: 5MB +# # part_size: 5MiB # Upload part size, 5MiB to 5GiB. Default: 5MiB # # For the s3:// form, disable TLS with ?ssl=false in the URL, not use_ssl. # ─── OPTIONAL ──────────────────────────────────────────────────────────────── diff --git a/internal/config/config.go b/internal/config/config.go index 1448a32..e5192b0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -33,11 +33,14 @@ const secretKeyPrefix = "AGE-SECRET-KEY-" const ( defaultBlobSizeLimit = Size(10 * 1024 * 1024 * 1024) // 10GB defaultChunkSize = Size(10 * 1024 * 1024) // 10MB - defaultS3PartSize = Size(5 * 1024 * 1024) // 5MB + defaultS3PartSize = Size(5 * 1024 * 1024) // 5MiB defaultCompressionLevel = 3 minChunkSize = 1024 * 1024 // 1MB minCompressionLevel = 1 maxCompressionLevel = 19 + // S3 accepts a multipart upload part from 5MiB to 5GiB. + minS3PartSize = 5 * 1024 * 1024 + maxS3PartSize = 5 * 1024 * 1024 * 1024 ) // Sentinel validation errors. @@ -55,6 +58,7 @@ var ( "blob_size_limit must be at least the largest chunk the chunker can " + "emit (chunk_size times the FastCDC size spread)") errBadCompression = errors.New("compression_level must be between 1 and 19") + errBadS3PartSize = errors.New("s3.part_size must be between 5MiB and 5GiB") errBadStorageScheme = errors.New( "storage_url must start with s3://, file://, or rclone://") errStorageNotConfigured = errors.New( @@ -244,6 +248,7 @@ func Load(path string) (*Config, error) { ChunkSize: defaultChunkSize, IndexPath: filepath.Join(xdg.DataHome, appName, "index.sqlite"), CompressionLevel: defaultCompressionLevel, + S3: S3Config{PartSize: defaultS3PartSize}, } // Convert smartconfig data to YAML then unmarshal @@ -294,10 +299,6 @@ func Load(path string) (*Config, error) { cfg.S3.Region = "us-east-1" } - if cfg.S3.PartSize == 0 { - cfg.S3.PartSize = defaultS3PartSize - } - // Check config file permissions (warn if world or group readable) //nolint:gosec // G703: config path is operator-supplied by design info, statErr := os.Stat(path) @@ -332,6 +333,7 @@ func Load(path string) (*Config, error) { // (chunk_size times chunker.ChunkSizeSpread), so a single-chunk blob never // exceeds the configured limit // - Compression level must be between 1 and 19 +// - S3 part size must be between 5MiB and 5GiB, the part sizes S3 accepts // // Returns an error describing the first validation failure encountered. func (c *Config) Validate() error { @@ -376,6 +378,11 @@ func (c *Config) Validate() error { return errBadCompression } + if c.S3.PartSize.Int64() < minS3PartSize || + c.S3.PartSize.Int64() > maxS3PartSize { + return errBadS3PartSize + } + return nil } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 52b46ad..c9bb10d 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -166,6 +166,7 @@ func TestValidateBlobSizeLimit(t *testing.T) { ChunkSize: chunkSize, BlobSizeLimit: blobLimit, CompressionLevel: 3, + S3: S3Config{PartSize: defaultS3PartSize}, } } @@ -221,6 +222,120 @@ func TestValidateBlobSizeLimit(t *testing.T) { } } +// TestValidateS3PartSize checks that s3.part_size is held to the part sizes +// S3 accepts, 5MiB to 5GiB, by changing only the part size of the test +// config. "5MB" in the config file is 5,000,000 bytes, below the minimum. +func TestValidateS3PartSize(t *testing.T) { + t.Parallel() + + base, err := Load(os.Getenv("VAULTIK_CONFIG")) + if err != nil { + t.Fatalf("Failed to load config: %v", err) + } + + tests := []struct { + name string + partSize Size + wantErr bool + }{ + { + name: "5MB is rejected", + partSize: 5_000_000, + wantErr: true, + }, + { + name: "one byte below 5MiB is rejected", + partSize: minS3PartSize - 1, + wantErr: true, + }, + { + name: "5MiB is accepted", + partSize: minS3PartSize, + wantErr: false, + }, + { + name: "5GiB is accepted", + partSize: maxS3PartSize, + wantErr: false, + }, + { + name: "one byte above 5GiB is rejected", + partSize: maxS3PartSize + 1, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + cfg := *base + cfg.S3.PartSize = tt.partSize + + err := cfg.Validate() + if tt.wantErr { + if !errors.Is(err, errBadS3PartSize) { + t.Fatalf("Validate() error = %v, want errBadS3PartSize", err) + } + + return + } + + if err != nil { + t.Fatalf("Validate() unexpected error: %v", err) + } + }) + } +} + +// TestLoadS3PartSize checks that a config file without s3.part_size loads +// with the 5MiB default, and that an explicit 0 fails at load like any other +// part size S3 refuses. +func TestLoadS3PartSize(t *testing.T) { + t.Parallel() + + const withoutPartSize = "snapshots:\n" + + " test:\n" + + " paths: [/tmp/vaultik-test-source]\n" + + "storage_url: file:///tmp/vaultik-test-storage\n" + + writeConfig := func(t *testing.T, text string) string { + t.Helper() + + path := filepath.Join(t.TempDir(), "config.yml") + + err := os.WriteFile(path, []byte(text), 0o600) + if err != nil { + t.Fatalf("write config: %v", err) + } + + return path + } + + t.Run("absent loads as 5MiB", func(t *testing.T) { + t.Parallel() + + cfg, err := Load(writeConfig(t, withoutPartSize)) + if err != nil { + t.Fatalf("Load() unexpected error: %v", err) + } + + if cfg.S3.PartSize != defaultS3PartSize { + t.Errorf("s3.part_size = %d, want %d", + cfg.S3.PartSize, defaultS3PartSize) + } + }) + + t.Run("0 is rejected", func(t *testing.T) { + t.Parallel() + + _, err := Load(writeConfig(t, withoutPartSize+"s3:\n part_size: 0\n")) + if !errors.Is(err, errBadS3PartSize) { + t.Fatalf("Load() error = %v, want errBadS3PartSize", err) + } + }) +} + // TestValidateAgeRecipients checks that recipients are parsed at config load // (a bad entry fails immediately, not mid-backup) and that no invalid entry — // least of all a pasted secret key — is echoed in the error. An empty list @@ -236,6 +351,7 @@ func TestValidateAgeRecipients(t *testing.T) { ChunkSize: Size(10 * 1024 * 1024), BlobSizeLimit: Size(10 * 1024 * 1024 * 1024), CompressionLevel: 3, + S3: S3Config{PartSize: defaultS3PartSize}, } } diff --git a/internal/s3/client.go b/internal/s3/client.go index 43f8725..f5fc91c 100644 --- a/internal/s3/client.go +++ b/internal/s3/client.go @@ -27,10 +27,12 @@ type Client struct { bucket string prefix string endpoint string + partSize int64 } // Config contains S3 client configuration. -// All fields are required except Prefix, which defaults to an empty string. +// All fields are required except Prefix, which defaults to an empty string, +// and PartSize, where zero means the SDK default of 5 MiB. // A non-empty Prefix is joined to every key with one "/", whether or not // it ends with one. // The Endpoint field should include the protocol (http:// or https://). @@ -41,6 +43,9 @@ type Config struct { AccessKeyID string SecretAccessKey string Region string + // PartSize is the size in bytes of each part of a multipart upload. + // An upload too large for S3's limit of 10,000 parts gets larger parts. + PartSize int64 } // nopLogger is a logger that discards all output. @@ -90,6 +95,7 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) { bucket: cfg.Bucket, prefix: prefix, endpoint: cfg.Endpoint, + partSize: cfg.PartSize, }, nil } @@ -123,12 +129,9 @@ func (c *Client) PutObjectWithProgress( ) error { fullKey := c.prefix + key - // uploadPartSize is 10MB for better progress granularity. - const uploadPartSize = 10 * 1024 * 1024 - // Create an uploader with the S3 client uploader := manager.NewUploader(c.s3Client, func(u *manager.Uploader) { - u.PartSize = uploadPartSize + u.PartSize = uploadPartSize(c.partSize, size) }) // Create a progress reader that tracks upload progress @@ -149,6 +152,21 @@ func (c *Client) PutObjectWithProgress( return err } +// uploadPartSize returns the part size for an upload of size bytes: the +// configured part size (the SDK default when zero), raised where needed so +// the upload fits in S3's limit of 10,000 parts. The uploader cannot raise +// it itself, because it cannot seek the progress reader to learn its size. +func uploadPartSize(configured, size int64) int64 { + if configured == 0 { + configured = manager.DefaultUploadPartSize + } + + maxParts := int64(manager.MaxUploadParts) + smallestThatFits := (size + maxParts - 1) / maxParts // rounded up + + return max(configured, smallestThatFits) +} + // GetObject downloads an object from S3 with the specified key. // The key is automatically prefixed with the configured prefix. // Returns a ReadCloser containing the object data. The caller must diff --git a/internal/s3/client_internal_test.go b/internal/s3/client_internal_test.go new file mode 100644 index 0000000..552059d --- /dev/null +++ b/internal/s3/client_internal_test.go @@ -0,0 +1,55 @@ +package s3 + +import "testing" + +// TestUploadPartSize checks that an upload too large for 10,000 parts of the +// configured size gets parts just large enough to fit in 10,000. +func TestUploadPartSize(t *testing.T) { + t.Parallel() + + const mib = 1024 * 1024 + + tests := []struct { + name string + configured int64 + size int64 + want int64 + }{ + { + name: "an upload that fits keeps the configured size", + configured: 5 * mib, + size: 10 * 1024 * mib, + want: 5 * mib, + }, + { + name: "exactly 10,000 parts keeps the configured size", + configured: 6 * mib, + size: 10_000 * 6 * mib, + want: 6 * mib, + }, + { + name: "one byte more than 10,000 parts adds a byte to each", + configured: 6 * mib, + size: 10_000*6*mib + 1, + want: 6*mib + 1, + }, + { + name: "zero means the SDK default of 5MiB", + configured: 0, + size: 1, + want: 5 * mib, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got := uploadPartSize(tt.configured, tt.size) + if got != tt.want { + t.Errorf("uploadPartSize(%d, %d) = %d, want %d", + tt.configured, tt.size, got, tt.want) + } + }) + } +} diff --git a/internal/s3/module.go b/internal/s3/module.go index 33b24ba..71f1b88 100644 --- a/internal/s3/module.go +++ b/internal/s3/module.go @@ -28,6 +28,7 @@ func provideClient(lc fx.Lifecycle, cfg *config.Config) (*Client, error) { AccessKeyID: cfg.S3.AccessKeyID, SecretAccessKey: cfg.S3.SecretAccessKey, Region: cfg.S3.Region, + PartSize: cfg.S3.PartSize.Int64(), }) if err != nil { return nil, err diff --git a/internal/storage/module.go b/internal/storage/module.go index 0e400fd..8d00ea1 100644 --- a/internal/storage/module.go +++ b/internal/storage/module.go @@ -99,6 +99,7 @@ func storerFromParsedS3URL(parsed *URL, cfg *config.Config) (Storer, error) { AccessKeyID: cfg.S3.AccessKeyID, SecretAccessKey: cfg.S3.SecretAccessKey, Region: region, + PartSize: cfg.S3.PartSize.Int64(), }) if err != nil { return nil, fmt.Errorf("creating S3 client: %w", err) @@ -134,6 +135,7 @@ func storerFromLegacyS3Config(cfg *config.Config) (Storer, error) { AccessKeyID: cfg.S3.AccessKeyID, SecretAccessKey: cfg.S3.SecretAccessKey, Region: region, + PartSize: cfg.S3.PartSize.Int64(), }) if err != nil { return nil, fmt.Errorf("creating S3 client: %w", err) diff --git a/internal/storage/s3_test.go b/internal/storage/s3_test.go index 2d581f7..2434eb5 100644 --- a/internal/storage/s3_test.go +++ b/internal/storage/s3_test.go @@ -1,11 +1,14 @@ package storage_test import ( + "bytes" "context" "errors" + "net/http" "net/http/httptest" "slices" "strings" + "sync/atomic" "testing" "github.com/johannesboyne/gofakes3" @@ -19,6 +22,13 @@ import ( // s3TestBucket is the bucket created for each in-process S3 server. const s3TestBucket = "test-bucket" +// Credentials for the tests that build a storer from a config.Config. The +// in-process S3 server accepts any. +const ( + s3TestAccessKeyID = "key" + s3TestSecretAccessKey = "secret" +) + // newS3Storer builds an s3:// backend backed by a fresh in-process // S3 server. It reuses the same in-memory S3 harness (gofakes3 + s3mem // over httptest) that internal/s3 and the not-found regression test use, @@ -128,8 +138,8 @@ func TestS3URLPrefixKeyLayout(t *testing.T) { storer, err := storage.NewStorer(&config.Config{ StorageURL: storageURL + "?endpoint=" + srv.URL, S3: config.S3Config{ - AccessKeyID: "key", - SecretAccessKey: "secret", + AccessKeyID: s3TestAccessKeyID, + SecretAccessKey: s3TestSecretAccessKey, }, }) if err != nil { @@ -171,6 +181,90 @@ func TestS3URLPrefixKeyLayout(t *testing.T) { } } +// TestS3UploadUsesConfiguredPartSize checks that s3.part_size reaches the +// multipart uploader, through storage_url and through the s3.* fields. An +// object three parts long must arrive as three parts; at the SDK's default +// of 5 MiB it would arrive as four. +func TestS3UploadUsesConfiguredPartSize(t *testing.T) { + t.Parallel() + + const ( + partSize = 6 * 1024 * 1024 + wantParts = 3 + ) + + backend := s3mem.New() + + err := backend.CreateBucket(s3TestBucket) + if err != nil { + t.Fatalf("create bucket: %v", err) + } + + // Every part of a multipart upload is one request with a partNumber. + var parts atomic.Int32 + + fake := gofakes3.New(backend).Server() + srv := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Has("partNumber") { + parts.Add(1) + } + + fake.ServeHTTP(w, r) + })) + t.Cleanup(srv.Close) + + cases := []struct { + name string + cfg *config.Config + }{ + { + name: "storage_url", + cfg: &config.Config{ + StorageURL: "s3://" + s3TestBucket + "?endpoint=" + srv.URL, + S3: config.S3Config{ + AccessKeyID: s3TestAccessKeyID, + SecretAccessKey: s3TestSecretAccessKey, + PartSize: partSize, + }, + }, + }, + { + name: "s3.endpoint", + cfg: &config.Config{ + S3: config.S3Config{ + Endpoint: srv.URL, + Bucket: s3TestBucket, + AccessKeyID: s3TestAccessKeyID, + SecretAccessKey: s3TestSecretAccessKey, + PartSize: partSize, + }, + }, + }, + } + + for _, tc := range cases { + parts.Store(0) + + storer, err := storage.NewStorer(tc.cfg) + if err != nil { + t.Fatalf("%s: NewStorer: %v", tc.name, err) + } + + data := bytes.NewReader(make([]byte, wantParts*partSize)) + + err = storer.PutWithProgress( + context.Background(), "blob", data, data.Size(), nil) + if err != nil { + t.Fatalf("%s: PutWithProgress: %v", tc.name, err) + } + + if got := parts.Load(); got != wantParts { + t.Errorf("%s: uploaded in %d parts, want %d", tc.name, got, wantParts) + } + } +} + // listStreamKeys returns the keys ListStream yields under a prefix, and // fails the test on a listing error. func listStreamKeys(t *testing.T, s storage.Storer, prefix string) []string { diff --git a/test/config.yaml b/test/config.yaml index f67577f..cd0e303 100644 --- a/test/config.yaml +++ b/test/config.yaml @@ -19,7 +19,7 @@ s3: secret_access_key: test-secret-key region: us-east-1 use_ssl: true - part_size: 5242880 # 5MB + part_size: 5242880 # 5MiB index_path: /tmp/vaultik-test.sqlite chunk_size: 10MB blob_size_limit: 10GB