Pass s3.part_size to the multipart uploader (closes #232)
check / check (push) Waiting to run
check / check (push) Waiting to run
s3.part_size was loaded and defaulted but never reached the S3 client, whose uploader used a fixed 10MiB part. The client now takes the part size from the config, for storage_url and for the s3.* fields, and config load rejects a value below 5MiB or above 5GiB, an explicit 0 included. A blob too large for S3's limit of 10,000 parts at that size is uploaded in larger parts, since the uploader cannot learn the size of the reader it is given. The docs gave the default as 5MB, which the config file reads as 5,000,000 bytes, below the minimum; they now say 5MiB. Judgement call: the 5GiB maximum is enforced along with the 5MiB minimum the issue names. Model: opus-5-5
This commit was merged in pull request #262.
This commit is contained in:
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user