diff --git a/README.md b/README.md index 809626b..2a13578 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,31 @@ Choose the construction style that fits your application: All storage operations also expose `*Context` equivalents for deadlines and cancellation. The default methods use `context.Background()`. +### Streaming objects + +`storage.Objects(disk)` discovers the optional `storagecore.ObjectStorage` +capability. The S3 driver supports paged object walking, streaming reads and +writes, metadata preservation, version-pinned reads, and conditional reads using +an observed ETag. Unsupported drivers return `storage.ErrUnsupported`; there is +no implicit fallback that buffers whole objects in memory. + +Object keys are opaque and relative to the configured disk prefix. Unlike the +existing file API, these methods preserve dot segments and repeated slashes. +Each operation takes an explicit context. Close every returned `ObjectReader`. +Writes declare an exact size and leave ownership of the input reader with the +caller; provide a seekable reader when provider retries are required. +For non-seekable S3 uploads, HTTPS retains the SDK's streaming checksums. Plain +HTTP requires explicit `s3storage.Config.UnsignedPayload` opt-in; only that case +skips optional SDK checksum precomputation. Signed HTTP uploads require a seekable +reader. The library never silently disables configured payload signing or spools +whole objects to disk. Cancellation stops subsequent source reads, including SDK +prehashing; a reader that blocks inside its own `Read` must honor cancellation or +be unblocked by its owner. + +Bucket creation remains explicit through `s3storage.PrepareBucketContext`. +Ordinary storage construction and object reads never provision provider resources. +Existing buffered methods, driver interfaces, and minimum Go versions are unchanged. + ### Common operations ```go diff --git a/docs/bench/go.mod b/docs/bench/go.mod index 56965b9..557b8cc 100644 --- a/docs/bench/go.mod +++ b/docs/bench/go.mod @@ -133,7 +133,7 @@ require ( github.com/go-resty/resty/v2 v2.17.2 // indirect github.com/go-viper/mapstructure/v2 v2.5.0 // indirect github.com/goccy/go-json v0.10.6 // indirect - github.com/goforj/storage/storagecore v0.5.1 // indirect + github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b // indirect github.com/gofrs/flock v0.13.0 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang-jwt/jwt/v4 v4.5.2 // indirect diff --git a/driver/s3storage/bucket.go b/driver/s3storage/bucket.go new file mode 100644 index 0000000..697ee69 --- /dev/null +++ b/driver/s3storage/bucket.go @@ -0,0 +1,49 @@ +package s3storage + +import ( + "context" + "errors" + "fmt" + + "github.com/aws/aws-sdk-go-v2/aws" + "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/goforj/storage/storagecore" +) + +// bucketAPI is separate from ordinary object operations because provisioning is explicitly requested. +type bucketAPI interface { + HeadBucket(context.Context, *s3.HeadBucketInput, ...func(*s3.Options)) (*s3.HeadBucketOutput, error) + CreateBucket(context.Context, *s3.CreateBucketInput, ...func(*s3.Options)) (*s3.CreateBucketOutput, error) +} + +// PrepareBucketContext creates a missing bucket using the same configuration and credentials as the driver. +// Existing buckets are unchanged. Permission failures never trigger a creation attempt. +func PrepareBucketContext(ctx context.Context, cfg Config) error { + ctx = normalizeContext(ctx) + store, err := newFromDiskConfig(ctx, cfg.ResolvedConfig()) + if err != nil { + return err + } + client, ok := store.(*driver).client.(bucketAPI) + if !ok { + return storagecore.ErrUnsupported + } + if _, err = client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(cfg.Bucket)}); err == nil { + return nil + } + var api smithy.APIError + if !errors.As(err, &api) || api.ErrorCode() != "NotFound" && api.ErrorCode() != "NoSuchBucket" { + return fmt.Errorf("storage: inspect bucket: %w", wrapError(err)) + } + input := &s3.CreateBucketInput{Bucket: aws.String(cfg.Bucket)} + if cfg.Region != "us-east-1" { + input.CreateBucketConfiguration = &types.CreateBucketConfiguration{LocationConstraint: types.BucketLocationConstraint(cfg.Region)} + } + _, err = client.CreateBucket(ctx, input) + if err != nil { + return fmt.Errorf("storage: prepare bucket: %w", wrapError(err)) + } + return nil +} diff --git a/driver/s3storage/go.mod b/driver/s3storage/go.mod index 75cbf84..894d9bb 100644 --- a/driver/s3storage/go.mod +++ b/driver/s3storage/go.mod @@ -8,7 +8,7 @@ require ( github.com/aws/aws-sdk-go-v2/credentials v1.20.3 github.com/aws/aws-sdk-go-v2/service/s3 v1.111.0 github.com/aws/smithy-go v1.28.1 - github.com/goforj/storage/storagecore v0.5.1 + github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b github.com/johannesboyne/gofakes3 v1.2.0 github.com/stretchr/testify v1.12.1 ) diff --git a/driver/s3storage/go.sum b/driver/s3storage/go.sum index 2bd593d..3dd9115 100644 --- a/driver/s3storage/go.sum +++ b/driver/s3storage/go.sum @@ -96,8 +96,8 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU= github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8= github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8= -github.com/goforj/storage/storagecore v0.5.1 h1:q9B6F9ne/KqZtLDM5wqWcmoZz+teWzhu7OLGp3aUCqY= -github.com/goforj/storage/storagecore v0.5.1/go.mod h1:n6epEk/dinioVPKa8VkWduaf0b1ggcnycXRmd6nURCY= +github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b h1:DoxWa3MWGyCYqF7BngY1KKMZfUpGWCSr7NNMvqZ72Jo= +github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b/go.mod h1:n6epEk/dinioVPKa8VkWduaf0b1ggcnycXRmd6nURCY= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= diff --git a/driver/s3storage/objects.go b/driver/s3storage/objects.go new file mode 100644 index 0000000..272f6b4 --- /dev/null +++ b/driver/s3storage/objects.go @@ -0,0 +1,211 @@ +package s3storage + +import ( + "context" + "fmt" + "io" + "net/url" + "strings" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/goforj/storage/storagecore" +) + +// objectKey preserves opaque keys while keeping all access within the configured prefix. +func (d *driver) objectKey(key string) (string, error) { + if strings.ContainsRune(key, '\x00') || d.prefix == "" && key == "" { + return "", fmt.Errorf("%w: invalid object key", storagecore.ErrForbidden) + } + if d.prefix != "" { + return d.prefix + "/" + key, nil + } + return key, nil +} + +// objectPrefix scopes flat listings without treating keys as filesystem paths. +func (d *driver) objectPrefix(prefix string) (string, error) { + if strings.ContainsRune(prefix, '\x00') { + return "", fmt.Errorf("%w: invalid object prefix", storagecore.ErrForbidden) + } + if d.prefix != "" { + return d.prefix + "/" + prefix, nil + } + return prefix, nil +} + +// WalkObjectsContext iterates raw objects one provider page at a time and propagates callback cancellation. +func (d *driver) WalkObjectsContext(ctx context.Context, prefix string, visit func(storagecore.ObjectInfo) error) error { + ctx = normalizeContext(ctx) + if err := ctx.Err(); err != nil { + return err + } + if visit == nil { + return fmt.Errorf("storage: object visitor is required") + } + scoped, err := d.objectPrefix(prefix) + if err != nil { + return err + } + var token *string + for { + page, err := d.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(d.bucket), Prefix: aws.String(scoped), ContinuationToken: token, EncodingType: types.EncodingTypeUrl}) + if err != nil { + return wrapError(err) + } + for _, object := range page.Contents { + key := aws.ToString(object.Key) + if page.EncodingType == types.EncodingTypeUrl { + key, err = url.PathUnescape(key) + if err != nil { + return fmt.Errorf("storage: decode object key: %w", err) + } + } + if !strings.HasPrefix(key, scoped) { + return fmt.Errorf("%w: object outside disk prefix", storagecore.ErrForbidden) + } + relative := key + if d.prefix != "" { + relative = strings.TrimPrefix(key, d.prefix+"/") + } + info := storagecore.ObjectInfo{Key: relative, Size: aws.ToInt64(object.Size), ETag: aws.ToString(object.ETag)} + if object.LastModified != nil { + info.ModifiedAt = *object.LastModified + } + if err := ctx.Err(); err != nil { + return err + } + if err := visit(info); err != nil { + return err + } + } + if !aws.ToBool(page.IsTruncated) { + return nil + } + next := aws.ToString(page.NextContinuationToken) + if next == "" || next == aws.ToString(token) { + return fmt.Errorf("storage: object listing did not advance") + } + token = page.NextContinuationToken + } +} + +// OpenObjectContext streams a prefix-scoped object with optional version and entity-tag conditions. +func (d *driver) OpenObjectContext(ctx context.Context, key string, options storagecore.OpenObjectOptions) (*storagecore.ObjectReader, error) { + ctx = normalizeContext(ctx) + if err := ctx.Err(); err != nil { + return nil, err + } + scoped, err := d.objectKey(key) + if err != nil { + return nil, err + } + input := &s3.GetObjectInput{Bucket: aws.String(d.bucket), Key: aws.String(scoped)} + if options.Version != "" { + input.VersionId = aws.String(options.Version) + } + if options.IfMatch != "" { + input.IfMatch = aws.String(options.IfMatch) + } + out, err := d.client.GetObject(ctx, input) + if err != nil { + return nil, wrapError(err) + } + if options.IfMatch != "" && options.IfMatch != "*" && strings.Trim(options.IfMatch, "\"") != strings.Trim(aws.ToString(out.ETag), "\"") { + out.Body.Close() + return nil, fmt.Errorf("storage: object changed during conditional read") + } + info := storagecore.ObjectInfo{Key: key, Size: aws.ToInt64(out.ContentLength), ETag: aws.ToString(out.ETag), Version: aws.ToString(out.VersionId), Metadata: storagecore.ObjectMetadata{ContentType: aws.ToString(out.ContentType), ContentEncoding: aws.ToString(out.ContentEncoding), ContentDisposition: aws.ToString(out.ContentDisposition), CacheControl: aws.ToString(out.CacheControl), Values: out.Metadata}} + if out.LastModified != nil { + info.ModifiedAt = *out.LastModified + } + return &storagecore.ObjectReader{ReadCloser: out.Body, Info: info}, nil +} + +// PutObjectContext passes a bounded stream and its metadata directly to the provider. +func (d *driver) PutObjectContext(ctx context.Context, key string, body io.Reader, options storagecore.PutObjectOptions) (storagecore.ObjectInfo, error) { + ctx = normalizeContext(ctx) + if err := ctx.Err(); err != nil { + return storagecore.ObjectInfo{}, err + } + scoped, err := d.objectKey(key) + if err != nil { + return storagecore.ObjectInfo{}, err + } + if body == nil || options.Size < 0 { + return storagecore.ObjectInfo{}, fmt.Errorf("storage: object body and nonnegative size are required") + } + _, seekable := body.(io.Seeker) + var operationOptions []func(*s3.Options) + if !seekable && d.httpEndpoint && options.Size > 0 { + if !d.unsignedPayload { + return storagecore.ObjectInfo{}, fmt.Errorf("%w: non-seekable HTTP uploads require UnsignedPayload or a seekable body", storagecore.ErrUnsupported) + } + // The caller already opted out of payload signing; optional checksums also require rewindable HTTP streams. + operationOptions = append(operationOptions, func(options *s3.Options) { + options.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired + }) + } + body = contextObjectBody(ctx, body) + metadata := options.Metadata + input := &s3.PutObjectInput{Bucket: aws.String(d.bucket), Key: aws.String(scoped), Body: body, ContentLength: aws.Int64(options.Size), Metadata: metadata.Values} + if metadata.ContentType != "" { + input.ContentType = aws.String(metadata.ContentType) + } + if metadata.ContentEncoding != "" { + input.ContentEncoding = aws.String(metadata.ContentEncoding) + } + if metadata.ContentDisposition != "" { + input.ContentDisposition = aws.String(metadata.ContentDisposition) + } + if metadata.CacheControl != "" { + input.CacheControl = aws.String(metadata.CacheControl) + } + out, err := d.client.PutObject(ctx, input, operationOptions...) + if err != nil { + return storagecore.ObjectInfo{}, wrapError(err) + } + return storagecore.ObjectInfo{Key: key, Size: options.Size, ETag: aws.ToString(out.ETag), Version: aws.ToString(out.VersionId), Metadata: metadata}, nil +} + +// objectContextReader checks cancellation during SDK hashing as well as transport reads. +type objectContextReader struct { + ctx context.Context + reader io.Reader +} + +// Read stops additional source consumption as soon as cancellation is observed. +func (r objectContextReader) Read(p []byte) (int, error) { + if err := r.ctx.Err(); err != nil { + return 0, err + } + n, err := r.reader.Read(p) + if canceled := r.ctx.Err(); canceled != nil { + return n, canceled + } + return n, err +} + +// objectContextReadSeeker preserves SDK rewind support for caller-owned seekable sources. +type objectContextReadSeeker struct { + objectContextReader + seeker io.Seeker +} + +// Seek preserves the caller's stream position semantics while honoring cancellation. +func (r objectContextReadSeeker) Seek(offset int64, whence int) (int64, error) { + if err := r.ctx.Err(); err != nil { + return 0, err + } + return r.seeker.Seek(offset, whence) +} + +// contextObjectBody retains seekability without exposing a whole-stream copy fast path that bypasses cancellation. +func contextObjectBody(ctx context.Context, body io.Reader) io.Reader { + reader := objectContextReader{ctx: ctx, reader: body} + if seeker, ok := body.(io.Seeker); ok { + return objectContextReadSeeker{objectContextReader: reader, seeker: seeker} + } + return reader +} diff --git a/driver/s3storage/objects_contract_test.go b/driver/s3storage/objects_contract_test.go new file mode 100644 index 0000000..93d74b4 --- /dev/null +++ b/driver/s3storage/objects_contract_test.go @@ -0,0 +1,150 @@ +package s3storage + +import ( + "context" + "errors" + "io" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "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/goforj/storage/storagecore" +) + +// bucketFixture records provisioning separately from ordinary object operations. +type bucketFixture struct { + fakeS3 + inspectErr error + createErr error + created []*s3.CreateBucketInput +} + +// HeadBucket supplies provider failure classes without network retry delays. +func (f *bucketFixture) HeadBucket(context.Context, *s3.HeadBucketInput, ...func(*s3.Options)) (*s3.HeadBucketOutput, error) { + return &s3.HeadBucketOutput{}, f.inspectErr +} + +// CreateBucket records the region constraint and preserves provisioning errors. +func (f *bucketFixture) CreateBucket(_ context.Context, input *s3.CreateBucketInput, _ ...func(*s3.Options)) (*s3.CreateBucketOutput, error) { + f.created = append(f.created, input) + return &s3.CreateBucketOutput{}, f.createErr +} + +// TestBucketPreparationFailureBoundaries ensures only missing buckets trigger provisioning. +func TestBucketPreparationFailureBoundaries(t *testing.T) { + original := buildS3Clients + t.Cleanup(func() { buildS3Clients = original }) + cfg := Config{Bucket: "fixture", Region: "eu-west-1", AccessKeyID: "fixture", SecretAccessKey: "fixture"} + for _, test := range []struct { + name string + client s3API + cfg Config + wantError bool + wantCreates int + }{ + {name: "invalid configuration", client: &bucketFixture{}, cfg: Config{}, wantError: true}, + {name: "unsupported provisioning", client: &fakeS3{}, cfg: cfg, wantError: true}, + {name: "transport failure", client: &bucketFixture{inspectErr: errors.New("transport failed")}, cfg: cfg, wantError: true}, + {name: "missing regional bucket", client: &bucketFixture{inspectErr: &smithy.GenericAPIError{Code: "NoSuchBucket"}}, cfg: cfg, wantCreates: 1}, + {name: "create failure", client: &bucketFixture{inspectErr: &smithy.GenericAPIError{Code: "NotFound"}, createErr: errors.New("creation refused")}, cfg: cfg, wantError: true, wantCreates: 1}, + } { + t.Run(test.name, func(t *testing.T) { + buildS3Clients = func(aws.Config, storagecore.ResolvedConfig) (s3API, s3PresignAPI) { return test.client, nil } + err := PrepareBucketContext(context.Background(), test.cfg) + if (err != nil) != test.wantError { + t.Fatal("unexpected provisioning outcome", err) + } + if fixture, ok := test.client.(*bucketFixture); ok { + if len(fixture.created) != test.wantCreates { + t.Fatal("unexpected creation attempt", fixture.created) + } + for _, created := range fixture.created { + if created.CreateBucketConfiguration == nil || string(created.CreateBucketConfiguration.LocationConstraint) != "eu-west-1" { + t.Fatal("regional constraint lost", created) + } + } + } + }) + } +} + +// objectResponseFixture supplies streaming response ownership and metadata independently of list fixtures. +type objectResponseFixture struct { + fakeS3 + response *s3.GetObjectOutput +} + +// GetObject records conditional inputs while returning the configured streaming response. +func (f *objectResponseFixture) GetObject(_ context.Context, input *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) { + f.getInputs = append(f.getInputs, input) + return f.response, nil +} + +// TestObjectPaginationPreservesContinuationAndCancellation checks page boundaries and callback stop behavior. +func TestObjectPaginationPreservesContinuationAndCancellation(t *testing.T) { + modified := time.Date(2026, 10, 18, 0, 0, 0, 0, time.UTC) + fake := &fakeS3{listSeq: []*s3.ListObjectsV2Output{ + {Contents: []types.Object{{Key: aws.String("first"), Size: aws.Int64(1), LastModified: &modified}}, IsTruncated: aws.Bool(true), NextContinuationToken: aws.String("second-page")}, + {Contents: []types.Object{{Key: aws.String("second"), Size: aws.Int64(2)}}}, + }} + disk := &driver{client: fake, bucket: "fixture"} + var found []storagecore.ObjectInfo + if err := disk.WalkObjectsContext(context.Background(), "", func(info storagecore.ObjectInfo) error { found = append(found, info); return nil }); err != nil { + t.Fatal(err) + } + if len(found) != 2 || found[0].Key != "first" || !found[0].ModifiedAt.Equal(modified) || found[1].Key != "second" || len(fake.listInputs) != 2 || aws.ToString(fake.listInputs[1].ContinuationToken) != "second-page" { + t.Fatal("pagination lost data", found, fake.listInputs) + } + fake.listSeq = []*s3.ListObjectsV2Output{{IsTruncated: aws.Bool(true), NextContinuationToken: aws.String("repeated")}, {IsTruncated: aws.Bool(true), NextContinuationToken: aws.String("repeated")}} + if err := disk.WalkObjectsContext(context.Background(), "", func(storagecore.ObjectInfo) error { return nil }); err == nil { + t.Fatal("repeated token accepted") + } + ctx, cancel := context.WithCancel(context.Background()) + fake.listHook = cancel + fake.listOut = &s3.ListObjectsV2Output{Contents: []types.Object{{Key: aws.String("after-cancellation")}}} + called := false + if err := disk.WalkObjectsContext(ctx, "", func(storagecore.ObjectInfo) error { called = true; return nil }); !errors.Is(err, context.Canceled) || called { + t.Fatal("canceled page delivered", called, err) + } +} + +// TestObjectStreamingOwnershipAndMetadata verifies stream lifetime and metadata at the provider boundary. +func TestObjectStreamingOwnershipAndMetadata(t *testing.T) { + modified := time.Date(2026, 10, 18, 0, 0, 0, 0, time.UTC) + body := &trackedReadCloser{reader: strings.NewReader("data")} + fixture := &objectResponseFixture{response: &s3.GetObjectOutput{Body: body, ContentLength: aws.Int64(4), ETag: aws.String("\"observed\""), VersionId: aws.String("version"), LastModified: &modified, ContentType: aws.String("text/plain"), ContentEncoding: aws.String("identity"), ContentDisposition: aws.String("attachment"), CacheControl: aws.String("max-age=60"), Metadata: map[string]string{"owner": "fixture"}}} + disk := &driver{client: fixture, bucket: "fixture"} + reader, err := disk.OpenObjectContext(context.Background(), "path/../opaque", storagecore.OpenObjectOptions{Version: "version", IfMatch: "observed"}) + if err != nil { + t.Fatal(err) + } + if body.closed { + t.Fatal("stream closed before caller finished") + } + data, err := io.ReadAll(reader) + if err != nil || string(data) != "data" || reader.Info.Version != "version" || !reader.Info.ModifiedAt.Equal(modified) { + t.Fatal("stream metadata lost", reader.Info, err) + } + if err := reader.Close(); err != nil || !body.closed { + t.Fatal("caller close lost", err) + } + metadata := reader.Info.Metadata + if metadata.ContentType != "text/plain" || metadata.ContentEncoding != "identity" || metadata.ContentDisposition != "attachment" || metadata.CacheControl != "max-age=60" || metadata.Values["owner"] != "fixture" { + t.Fatal("read metadata lost", metadata) + } + if _, err := disk.PutObjectContext(context.Background(), "path/../opaque", strings.NewReader("data"), storagecore.PutObjectOptions{Size: 4, Metadata: metadata}); err != nil { + t.Fatal(err) + } + put := fixture.putInputs[0] + if aws.ToString(put.Key) != "path/../opaque" || aws.ToString(put.ContentEncoding) != "identity" || aws.ToString(put.ContentDisposition) != "attachment" || aws.ToString(put.CacheControl) != "max-age=60" || put.Metadata["owner"] != "fixture" { + t.Fatal("write metadata or key changed", put) + } + body.closed = false + if _, err := disk.OpenObjectContext(context.Background(), "key", storagecore.OpenObjectOptions{IfMatch: "stale"}); err == nil || !body.closed { + t.Fatal("stale stream retained") + } +} diff --git a/driver/s3storage/objects_protocol_test.go b/driver/s3storage/objects_protocol_test.go new file mode 100644 index 0000000..37b47d9 --- /dev/null +++ b/driver/s3storage/objects_protocol_test.go @@ -0,0 +1,197 @@ +package s3storage + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/goforj/storage/storagecore" +) + +// TestObjectListingDecodesOpaqueKeysExactlyOnce prevents XML-sensitive keys from poisoning a provider page. +func TestObjectListingDecodesOpaqueKeysExactlyOnce(t *testing.T) { + keys := []string{"tenant/control\x01key", "tenant/plus+percent%2F space", "tenant/雪"} + malformed := false + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("encoding-type") != "url" { + t.Error("listing did not request URL encoding") + } + w.Header().Set("Content-Type", "application/xml") + fmt.Fprint(w, "urlfalse") + for _, key := range keys { + encoded := strings.ReplaceAll(url.QueryEscape(key), "+", "%20") + if malformed { + encoded = "%ZZ" + } + fmt.Fprintf(w, "%s1", encoded) + } + fmt.Fprint(w, "") + })) + defer server.Close() + disk, err := New(Config{Bucket: "fixture", Region: "us-east-1", Endpoint: server.URL, AccessKeyID: "fixture", SecretAccessKey: "fixture", UsePathStyle: true, Prefix: "tenant"}) + if err != nil { + t.Fatal(err) + } + objects := disk.(storagecore.ObjectStorage) + var found []string + if err := objects.WalkObjectsContext(context.Background(), "", func(info storagecore.ObjectInfo) error { found = append(found, info.Key); return nil }); err != nil { + t.Fatal(err) + } + if len(found) != len(keys) { + t.Fatal("objects omitted", found) + } + for i, key := range keys { + if found[i] != strings.TrimPrefix(key, "tenant/") { + t.Errorf("opaque key changed: got %q want %q", found[i], key) + } + } + malformed = true + called := false + if err := objects.WalkObjectsContext(context.Background(), "", func(storagecore.ObjectInfo) error { called = true; return nil }); err == nil || called { + t.Fatal("malformed encoding accepted", called, err) + } +} + +// TestNonseekableUploadsRespectHTTPPayloadSigning preserves signing policy without buffering entire objects. +func TestNonseekableUploadsRespectHTTPPayloadSigning(t *testing.T) { + for _, mode := range []string{"explicit", "uppercase", "environment", "service-environment"} { + for _, unsigned := range []bool{false, true} { + t.Run(fmt.Sprint(mode, "/unsigned=", unsigned), func(t *testing.T) { + requests := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + body, err := io.ReadAll(r.Body) + if err != nil || string(body) != "payload" { + t.Errorf("stream changed: %q, %v", body, err) + } + if r.Header.Get("X-Amz-Content-Sha256") != "UNSIGNED-PAYLOAD" { + t.Error("explicit unsigned policy not honored") + } + w.Header().Set("ETag", "\"fixture\"") + })) + defer server.Close() + endpoint := server.URL + switch mode { + case "uppercase": + endpoint = strings.Replace(endpoint, "http:", "HTTP:", 1) + case "environment": + t.Setenv("AWS_ENDPOINT_URL", endpoint) + endpoint = "" + case "service-environment": + t.Setenv("AWS_ENDPOINT_URL_S3", endpoint) + endpoint = "" + } + disk, err := New(Config{Bucket: "fixture", Region: "us-east-1", Endpoint: endpoint, AccessKeyID: "fixture", SecretAccessKey: "fixture", UsePathStyle: true, UnsignedPayload: unsigned}) + if err != nil { + t.Fatal(err) + } + _, err = disk.(storagecore.ObjectStorage).PutObjectContext(context.Background(), "key", io.LimitReader(strings.NewReader("payload"), 7), storagecore.PutObjectOptions{Size: 7}) + if unsigned { + if err != nil || requests != 1 { + t.Fatal("opted-in streaming upload failed", requests, err) + } + } else if !errors.Is(err, storagecore.ErrUnsupported) || requests != 0 { + t.Fatal("signed payload policy weakened", requests, err) + } + }) + } + } + +} + +// TestNonseekableHTTPSUploadsRetainStreamingChecksums exercises the normal SDK streaming protocol over TLS. +func TestNonseekableHTTPSUploadsRetainStreamingChecksums(t *testing.T) { + requests := 0 + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + body, err := io.ReadAll(r.Body) + if err != nil || !strings.Contains(string(body), "payload") { + t.Errorf("streaming payload lost: %q, %v", body, err) + } + if !strings.Contains(r.Header.Get("X-Amz-Trailer"), "x-amz-checksum-") || r.Header.Get("X-Amz-Content-Sha256") != "STREAMING-UNSIGNED-PAYLOAD-TRAILER" { + t.Errorf("HTTPS checksum headers: algorithm=%q trailer=%q hash=%q encoding=%q", r.Header.Get("X-Amz-Sdk-Checksum-Algorithm"), r.Header.Get("X-Amz-Trailer"), r.Header.Get("X-Amz-Content-Sha256"), r.Header.Get("Content-Encoding")) + } + w.Header().Set("ETag", "\"fixture\"") + })) + defer server.Close() + original := buildS3Clients + t.Cleanup(func() { buildS3Clients = original }) + buildS3Clients = func(cfg aws.Config, resolved storagecore.ResolvedConfig) (s3API, s3PresignAPI) { + cfg.HTTPClient = server.Client() + return original(cfg, resolved) + } + disk, err := New(Config{Bucket: "fixture", Region: "us-east-1", Endpoint: server.URL, AccessKeyID: "fixture", SecretAccessKey: "fixture", UsePathStyle: true}) + if err != nil { + t.Fatal(err) + } + _, err = disk.(storagecore.ObjectStorage).PutObjectContext(context.Background(), "key", io.LimitReader(strings.NewReader("payload"), 7), storagecore.PutObjectOptions{Size: 7}) + if err != nil || requests != 1 { + t.Fatal("TLS stream failed", requests, err) + } +} + +// cancelUploadReader records source consumption when cancellation arrives during SDK checksum computation. +type cancelUploadReader struct { + reader *strings.Reader + cancel context.CancelFunc + bytes int + calls int +} + +// Read cancels the request after its first bounded read to expose uninterruptible prehash loops. +func (r *cancelUploadReader) Read(p []byte) (int, error) { + n, err := r.reader.Read(p) + r.bytes += n + r.calls++ + if r.calls == 1 { + r.cancel() + } + return n, err +} + +// Seek retains the source's rewindability so the SDK takes its prehash path. +func (r *cancelUploadReader) Seek(offset int64, whence int) (int64, error) { + return r.reader.Seek(offset, whence) +} + +// TestUploadCancellationStopsPrehashConsumption verifies cancellation after hashing has started, before any network request. +func TestUploadCancellationStopsPrehashConsumption(t *testing.T) { + requests := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests++; w.WriteHeader(500) })) + defer server.Close() + disk, err := New(Config{Bucket: "fixture", Region: "us-east-1", Endpoint: server.URL, AccessKeyID: "fixture", SecretAccessKey: "fixture", UsePathStyle: true}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + body := &cancelUploadReader{reader: strings.NewReader(strings.Repeat("x", 4<<20)), cancel: cancel} + _, err = disk.(storagecore.ObjectStorage).PutObjectContext(ctx, "key", body, storagecore.PutObjectOptions{Size: 4 << 20}) + if !errors.Is(err, context.Canceled) || body.calls != 1 || body.bytes >= 4<<20 || requests != 0 { + t.Fatal("cancellation consumed the remaining source", body.calls, body.bytes, requests, err) + } +} + +// TestCanceledObjectBodyDoesNotTouchSource ensures canceled SDK reads and rewinds leave caller state untouched. +func TestCanceledObjectBodyDoesNotTouchSource(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + source := strings.NewReader("payload") + body := contextObjectBody(ctx, source) + cancel() + if n, err := body.Read(make([]byte, 3)); n != 0 || !errors.Is(err, context.Canceled) { + t.Fatalf("canceled read = %d, %v", n, err) + } + if _, err := body.(io.Seeker).Seek(2, io.SeekStart); !errors.Is(err, context.Canceled) { + t.Fatalf("canceled seek = %v", err) + } + if position, err := source.Seek(0, io.SeekCurrent); err != nil || position != 0 { + t.Fatalf("canceled operation moved source: %d, %v", position, err) + } +} diff --git a/driver/s3storage/objects_test.go b/driver/s3storage/objects_test.go new file mode 100644 index 0000000..87fcc4d --- /dev/null +++ b/driver/s3storage/objects_test.go @@ -0,0 +1,167 @@ +package s3storage + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/goforj/storage/storagecore" + "github.com/johannesboyne/gofakes3" + "github.com/johannesboyne/gofakes3/backend/s3mem" +) + +// TestStreamingObjectsPreserveOpaqueKeysAndMetadata exercises the optional capability against the provider protocol. +func TestStreamingObjectsPreserveOpaqueKeysAndMetadata(t *testing.T) { + server := fakeServer(t, gofakes3.New(s3mem.New())) + defer server.Close() + cfg := Config{Bucket: "streaming", Endpoint: server.URL, Region: "us-east-1", AccessKeyID: "access", SecretAccessKey: "secret", UsePathStyle: true, Prefix: "tenant"} + if err := PrepareBucketContext(context.Background(), cfg); err != nil { + t.Fatal(err) + } + store, err := New(cfg) + if err != nil { + t.Fatal(err) + } + objects := store.(storagecore.ObjectStorage) + ctx := context.Background() + key := "path/../odd//name" + info, err := objects.PutObjectContext(ctx, key, strings.NewReader("payload"), storagecore.PutObjectOptions{Size: 7, Metadata: storagecore.ObjectMetadata{ContentType: "text/plain", Values: map[string]string{"owner": "fixture"}}}) + if err != nil || info.Size != 7 { + t.Fatal(info, err) + } + count := 0 + if err := objects.WalkObjectsContext(ctx, "", func(info storagecore.ObjectInfo) error { + count++ + if info.Key != key || info.Size != 7 { + t.Errorf("key or size changed: %#v", info) + } + return nil + }); err != nil || count != 1 { + t.Fatal(count, err) + } + reader, err := objects.OpenObjectContext(ctx, key, storagecore.OpenObjectOptions{IfMatch: info.ETag}) + if err != nil { + t.Fatal(err) + } + data, err := io.ReadAll(reader) + reader.Close() + if err != nil || string(data) != "payload" || reader.Info.Metadata.ContentType != "text/plain" || reader.Info.Metadata.Values["owner"] != "fixture" { + t.Fatal("stream or metadata changed", err, reader.Info) + } + if _, err := objects.OpenObjectContext(ctx, key, storagecore.OpenObjectOptions{IfMatch: "different"}); err == nil { + t.Fatal("conditional read accepted changed object") + } + stop := errors.New("stop") + if err := objects.WalkObjectsContext(ctx, "", func(storagecore.ObjectInfo) error { return stop }); !errors.Is(err, stop) { + t.Fatal("callback failure lost", err) + } + canceled, cancel := context.WithCancel(ctx) + cancel() + if err := objects.WalkObjectsContext(canceled, "", func(storagecore.ObjectInfo) error { return nil }); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if _, err := objects.OpenObjectContext(canceled, key, storagecore.OpenObjectOptions{}); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if _, err := objects.PutObjectContext(canceled, key, strings.NewReader(""), storagecore.PutObjectOptions{}); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + for _, bad := range []string{"bad\x00key"} { + if _, err := objects.OpenObjectContext(ctx, bad, storagecore.OpenObjectOptions{}); !errors.Is(err, storagecore.ErrForbidden) { + t.Fatal(err) + } + if _, err := objects.PutObjectContext(ctx, bad, strings.NewReader(""), storagecore.PutObjectOptions{}); !errors.Is(err, storagecore.ErrForbidden) { + t.Fatal(err) + } + if err := objects.WalkObjectsContext(ctx, bad, func(storagecore.ObjectInfo) error { return nil }); !errors.Is(err, storagecore.ErrForbidden) { + t.Fatal(err) + } + } + if err := objects.WalkObjectsContext(ctx, "", nil); err == nil { + t.Fatal("nil visitor accepted") + } + if _, err := objects.PutObjectContext(ctx, key, nil, storagecore.PutObjectOptions{}); err == nil { + t.Fatal("nil body accepted") + } + if _, err := objects.PutObjectContext(ctx, key, strings.NewReader(""), storagecore.PutObjectOptions{Size: -1}); err == nil { + t.Fatal("negative size accepted") + } +} + +// TestStreamingObjectFailuresKeepScopeAndStopPagination validates provider and pagination failures without buffering. +func TestStreamingObjectFailuresKeepScopeAndStopPagination(t *testing.T) { + ctx := context.Background() + failure := errors.New("provider failed") + for _, op := range []string{"list", "read", "write"} { + fake := &fakeS3{listErr: failure, getErr: failure, putErr: failure} + disk := &driver{client: fake, prefix: "tenant", bucket: "fixture"} + var err error + switch op { + case "list": + err = disk.WalkObjectsContext(ctx, "", func(storagecore.ObjectInfo) error { return nil }) + case "read": + _, err = disk.OpenObjectContext(ctx, "key", storagecore.OpenObjectOptions{Version: "pinned-version"}) + if len(fake.getInputs) != 1 || *fake.getInputs[0].VersionId != "pinned-version" { + t.Fatal("version pin lost") + } + case "write": + _, err = disk.PutObjectContext(ctx, "key", strings.NewReader("data"), storagecore.PutObjectOptions{Size: 4}) + } + if err == nil { + t.Fatal("provider error swallowed", op) + } + } + for _, page := range []*s3.ListObjectsV2Output{ + {IsTruncated: aws.Bool(true)}, + {Contents: []types.Object{{Key: aws.String("tenant-other/key")}}}, + } { + disk := &driver{client: &fakeS3{listOut: page}, prefix: "tenant", bucket: "fixture"} + if err := disk.WalkObjectsContext(ctx, "", func(storagecore.ObjectInfo) error { return nil }); err == nil { + t.Fatal("invalid listing accepted") + } + } + disk := &driver{client: &fakeS3{}, bucket: "fixture"} + if _, err := disk.OpenObjectContext(ctx, "", storagecore.OpenObjectOptions{}); !errors.Is(err, storagecore.ErrForbidden) { + t.Fatal("empty root key accepted", err) + } +} + +// TestPrepareBucketDistinguishesForbiddenAndMissing verifies provisioning stays opt-in and permission failures are preserved. +func TestPrepareBucketDistinguishesForbiddenAndMissing(t *testing.T) { + for _, status := range []int{200, 403, 404} { + t.Run(http.StatusText(status), func(t *testing.T) { + created := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodHead { + w.WriteHeader(status) + return + } + if r.Method == http.MethodPut { + created++ + w.WriteHeader(200) + return + } + w.WriteHeader(500) + })) + defer server.Close() + err := PrepareBucketContext(context.Background(), Config{Bucket: "fixture", Region: "us-east-1", Endpoint: server.URL, AccessKeyID: "fixture", SecretAccessKey: "fixture", UsePathStyle: true}) + if status == 403 { + if err == nil || created != 0 { + t.Fatal("forbidden bucket provisioned", err, created) + } + } else if err != nil { + t.Fatal(err) + } + if status == 404 && created != 1 { + t.Fatal("missing bucket not prepared", created) + } + }) + } +} diff --git a/driver/s3storage/s3.go b/driver/s3storage/s3.go index dad5498..41317bc 100644 --- a/driver/s3storage/s3.go +++ b/driver/s3storage/s3.go @@ -33,10 +33,12 @@ func init() { } type driver struct { - client s3API - presign s3PresignAPI - bucket string - prefix string + httpEndpoint bool + unsignedPayload bool + client s3API + presign s3PresignAPI + bucket string + prefix string } type s3API interface { @@ -158,9 +160,11 @@ func newFromDiskConfig(ctx context.Context, cfg storagecore.ResolvedConfig) (sto return nil, fmt.Errorf("storage: s3 access key ID and secret access key must be provided together") } if cfg.S3Endpoint != "" { - if _, err := normalizeEndpoint(cfg.S3Endpoint); err != nil { + endpoint, err := normalizeEndpoint(cfg.S3Endpoint) + if err != nil { return nil, err } + cfg.S3Endpoint = endpoint } prefix, err := storagecore.NormalizePath(cfg.Prefix) @@ -173,12 +177,21 @@ func newFromDiskConfig(ctx context.Context, cfg storagecore.ResolvedConfig) (sto return nil, fmt.Errorf("storage: load aws config: %w", err) } client, presign := buildS3Clients(awsCfg, cfg) + endpoint := cfg.S3Endpoint + // SDK options include endpoint overrides inherited from environment and shared AWS configuration. + if endpoint == "" { + if configured, ok := client.(interface{ Options() s3.Options }); ok { + endpoint = aws.ToString(configured.Options().BaseEndpoint) + } + } return &driver{ - client: client, - presign: presign, - bucket: cfg.S3Bucket, - prefix: prefix, + httpEndpoint: strings.HasPrefix(strings.ToLower(endpoint), "http://"), + unsignedPayload: cfg.S3UnsignedPayload, + client: client, + presign: presign, + bucket: cfg.S3Bucket, + prefix: prefix, }, nil } diff --git a/examples/go.mod b/examples/go.mod index b62e4e2..7f04372 100644 --- a/examples/go.mod +++ b/examples/go.mod @@ -140,7 +140,7 @@ require ( github.com/go-resty/resty/v2 v2.17.2 // indirect github.com/go-viper/mapstructure/v2 v2.5.0 // indirect github.com/goccy/go-json v0.10.6 // indirect - github.com/goforj/storage/storagecore v0.5.1 // indirect + github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b // indirect github.com/gofrs/flock v0.13.0 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang-jwt/jwt/v4 v4.5.2 // indirect diff --git a/go.mod b/go.mod index 48e2c65..a6fae84 100644 --- a/go.mod +++ b/go.mod @@ -2,4 +2,4 @@ module github.com/goforj/storage go 1.24.4 -require github.com/goforj/storage/storagecore v0.5.1 +require github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b diff --git a/go.sum b/go.sum index 4b2ec25..c60f599 100644 --- a/go.sum +++ b/go.sum @@ -1,2 +1,2 @@ -github.com/goforj/storage/storagecore v0.5.1 h1:q9B6F9ne/KqZtLDM5wqWcmoZz+teWzhu7OLGp3aUCqY= -github.com/goforj/storage/storagecore v0.5.1/go.mod h1:n6epEk/dinioVPKa8VkWduaf0b1ggcnycXRmd6nURCY= +github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b h1:DoxWa3MWGyCYqF7BngY1KKMZfUpGWCSr7NNMvqZ72Jo= +github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b/go.mod h1:n6epEk/dinioVPKa8VkWduaf0b1ggcnycXRmd6nURCY= diff --git a/integration/go.mod b/integration/go.mod index 743f398..70e1981 100644 --- a/integration/go.mod +++ b/integration/go.mod @@ -155,7 +155,7 @@ require ( github.com/go-resty/resty/v2 v2.17.2 // indirect github.com/go-viper/mapstructure/v2 v2.5.0 // indirect github.com/goccy/go-json v0.10.6 // indirect - github.com/goforj/storage/storagecore v0.5.1 // indirect + github.com/goforj/storage/storagecore v0.5.2-0.20261018064156-39419460440b // indirect github.com/gofrs/flock v0.13.0 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang-jwt/jwt/v4 v4.5.2 // indirect diff --git a/objects.go b/objects.go new file mode 100644 index 0000000..ec62efd --- /dev/null +++ b/objects.go @@ -0,0 +1,21 @@ +package storage + +import "github.com/goforj/storage/storagecore" + +// Objects returns a driver's optional streaming object capability without changing Storage. +// Every operation takes its own context; a handle's WithContext binding does not replace it. +// Unsupported drivers return ErrUnsupported rather than buffering a stream as a fallback. +func Objects(disk Storage) (storagecore.ObjectStorage, error) { + if bound, ok := disk.(*boundStorage); ok { + if bound.bindErr != nil { + return nil, bound.bindErr + } + if objects, ok := bound.inner.(storagecore.ObjectStorage); ok { + return objects, nil + } + } + if objects, ok := disk.(storagecore.ObjectStorage); ok { + return objects, nil + } + return nil, ErrUnsupported +} diff --git a/objects_test.go b/objects_test.go new file mode 100644 index 0000000..2eab6ab --- /dev/null +++ b/objects_test.go @@ -0,0 +1,51 @@ +package storage + +import ( + "context" + "errors" + "io" + "testing" + + "github.com/goforj/storage/storagecore" +) + +// streamFixture exposes the optional capability without adding it to the base Storage interface. +type streamFixture struct{ stubFS } + +// WithContext preserves the fixture capability across public bindings. +func (s *streamFixture) WithContext(context.Context) Storage { return s } + +// WalkObjectsContext keeps the fixture focused on capability discovery. +func (*streamFixture) WalkObjectsContext(context.Context, string, func(storagecore.ObjectInfo) error) error { + return nil +} + +// OpenObjectContext keeps the fixture focused on capability discovery. +func (*streamFixture) OpenObjectContext(context.Context, string, storagecore.OpenObjectOptions) (*storagecore.ObjectReader, error) { + return nil, nil +} + +// PutObjectContext keeps the fixture focused on capability discovery. +func (*streamFixture) PutObjectContext(context.Context, string, io.Reader, storagecore.PutObjectOptions) (storagecore.ObjectInfo, error) { + return storagecore.ObjectInfo{}, nil +} + +// TestObjectsCapabilityPreservesExistingStorage verifies supported, unsupported, and failed bindings. +func TestObjectsCapabilityPreservesExistingStorage(t *testing.T) { + stream := &streamFixture{} + for _, disk := range []Storage{stream, wrapStorage(stream), wrapStorage(stream).WithContext(context.Background())} { + got, err := Objects(disk) + if err != nil || got != stream { + t.Fatal("streaming capability lost", got, err) + } + } + for _, disk := range []Storage{nil, &stubFS{}, wrapStorage(&stubFS{})} { + if _, err := Objects(disk); !errors.Is(err, ErrUnsupported) { + t.Fatal(err) + } + } + failure := errors.New("binding failed") + if _, err := Objects(&boundStorage{inner: stream, bindErr: failure}); !errors.Is(err, failure) { + t.Fatal(err) + } +} diff --git a/storagecore/objects.go b/storagecore/objects.go new file mode 100644 index 0000000..7218130 --- /dev/null +++ b/storagecore/objects.go @@ -0,0 +1,55 @@ +package storagecore + +import ( + "context" + "io" + "time" +) + +// ObjectMetadata contains portable HTTP content attributes retained by object stores. +type ObjectMetadata struct { + ContentType string + ContentEncoding string + ContentDisposition string + CacheControl string + Values map[string]string +} + +// ObjectInfo describes an object using its exact key relative to the configured disk prefix. +// Keys are opaque, including dot segments and repeated slashes, and are never filesystem paths. +type ObjectInfo struct { + Key string + Size int64 + ETag string + Version string + ModifiedAt time.Time + Metadata ObjectMetadata +} + +// OpenObjectOptions optionally pins a read to a version or a previously observed entity tag. +type OpenObjectOptions struct { + Version string + IfMatch string +} + +// ObjectReader owns a streaming response that the caller must close, including after partial reads. +type ObjectReader struct { + io.ReadCloser + Info ObjectInfo +} + +// PutObjectOptions declares the exact stream size and portable metadata without buffering the object. +// The caller owns the reader and should provide a seekable reader when retries are required. +type PutObjectOptions struct { + Size int64 + Metadata ObjectMetadata +} + +// ObjectStorage is an optional streaming capability for opaque object keys. +// Existing Storage path semantics and buffered methods remain unchanged. Walk visits +// objects page by page without synthesizing directories or accumulating the whole listing. +type ObjectStorage interface { + WalkObjectsContext(ctx context.Context, prefix string, visit func(ObjectInfo) error) error + OpenObjectContext(ctx context.Context, key string, options OpenObjectOptions) (*ObjectReader, error) + PutObjectContext(ctx context.Context, key string, body io.Reader, options PutObjectOptions) (ObjectInfo, error) +}