feat: add optional streaming object storage capability

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

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

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
+}

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
 )

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=

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
+}