feat: add Bundle API support for streaming multi-object tar download

Add BundleObjects method that POSTs to /{bucket}?bundle with SigV4
signing and returns a streaming io.ReadCloser for tar consumption.

Includes WithBundleCompression/WithBundleOnError header helpers,
unit tests, and godoc example.

Assisted-by: Claude Opus 4.6 via Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Yevgeniy Firsov <firsov@tigrisdata.com>

Author: efirs

bundle.go

@@ -0,0 +1,190 @@
+package storage
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"time"
+
+	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
+)
+
+const (
+	// BundleFormatTar is the tar archive format for bundle responses.
+	BundleFormatTar = "tar"
+
+	// BundleCompressionNone disables compression (default).
+	BundleCompressionNone = "none"
+	// BundleCompressionGzip enables gzip compression.
+	BundleCompressionGzip = "gzip"
+	// BundleCompressionZstd enables zstd compression.
+	BundleCompressionZstd = "zstd"
+
+	// BundleOnErrorSkip silently omits missing objects from the archive (default).
+	BundleOnErrorSkip = "skip"
+	// BundleOnErrorFail returns an error if any object is missing.
+	BundleOnErrorFail = "fail"
+)
+
+// bundleHTTPClient is reused across calls. No overall timeout — the caller's
+// context controls cancellation, which avoids cutting off streaming reads.
+var bundleHTTPClient = &http.Client{
+	Transport: &http.Transport{
+		Proxy:                 http.ProxyFromEnvironment,
+		DialContext:           (&net.Dialer{Timeout: 30 * time.Second}).DialContext,
+		TLSHandshakeTimeout:  10 * time.Second,
+		ResponseHeaderTimeout: 60 * time.Second,
+	},
+}
+
+// BundleObjectsInput is the input for a BundleObjects request.
+type BundleObjectsInput struct {
+	// Bucket is the name of the bucket containing the objects. Required.
+	Bucket string
+
+	// Keys is the list of object keys to include in the bundle. Required.
+	// Maximum 5,000 keys per request.
+	Keys []string
+
+	// Compression sets the compression algorithm for the response.
+	// Valid values: "none" (default), "gzip", "zstd".
+	Compression string
+
+	// OnError controls behavior when objects are missing.
+	// "skip" (default): omit missing objects and append __bundle_errors.json to the tar.
+	// "fail": return an error before streaming if any object is missing.
+	OnError string
+}
+
+// BundleObjectsOutput is the response from a BundleObjects request.
+//
+// The Body contains a streaming tar archive. Callers are responsible for closing Body.
+// Use archive/tar to iterate entries:
+//
+//	tr := tar.NewReader(output.Body)
+//	for {
+//	    hdr, err := tr.Next()
+//	    if err == io.EOF { break }
+//	    // process hdr.Name, tr
+//	}
+//
+// If compression was requested, wrap Body with the appropriate decompressor first:
+//
+//	gz, _ := gzip.NewReader(output.Body)
+//	tr := tar.NewReader(gz)
+type BundleObjectsOutput struct {
+	// Body is the streaming tar archive. Must be closed by the caller.
+	Body io.ReadCloser
+
+	// ContentType is the response Content-Type (e.g. "application/x-tar", "application/gzip").
+	ContentType string
+
+	// StatusCode is the HTTP status code of the response.
+	StatusCode int
+}
+
+type bundleRequestBody struct {
+	Keys []string `json:"keys"`
+}
+
+// BundleObjects fetches multiple objects from a bucket as a streaming tar archive
+// in a single HTTP request.
+//
+// This is a Tigris extension to the S3 API, designed for ML training workloads
+// that need to fetch thousands of objects per batch without per-object HTTP overhead.
+//
+// The caller is responsible for closing the returned Body.
+func (c *Client) BundleObjects(ctx context.Context, in *BundleObjectsInput) (*BundleObjectsOutput, error) {
+	if in.Bucket == "" {
+		return nil, fmt.Errorf("storage: BundleObjects: bucket is required")
+	}
+	if len(in.Keys) == 0 {
+		return nil, fmt.Errorf("storage: BundleObjects: at least one key is required")
+	}
+
+	compression := in.Compression
+	if compression == "" {
+		compression = BundleCompressionNone
+	}
+
+	onError := in.OnError
+	if onError == "" {
+		onError = BundleOnErrorSkip
+	}
+
+	opts := c.Client.Options()
+
+	endpoint := GlobalEndpoint
+	if opts.BaseEndpoint != nil {
+		endpoint = *opts.BaseEndpoint
+	}
+	endpoint = strings.TrimRight(endpoint, "/")
+
+	reqURL := fmt.Sprintf("%s/%s?bundle", endpoint, in.Bucket)
+
+	body, err := json.Marshal(bundleRequestBody{Keys: in.Keys})
+	if err != nil {
+		return nil, fmt.Errorf("storage: BundleObjects: failed to marshal keys: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(body))
+	if err != nil {
+		return nil, fmt.Errorf("storage: BundleObjects: failed to create request: %w", err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Tigris-Bundle-Format", BundleFormatTar)
+	req.Header.Set("X-Tigris-Bundle-Compression", compression)
+	req.Header.Set("X-Tigris-Bundle-On-Error", onError)
+
+	// Sign request with SigV4.
+	if opts.Credentials != nil {
+		creds, err := opts.Credentials.Retrieve(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("storage: BundleObjects: failed to retrieve credentials: %w", err)
+		}
+
+		payloadHash := sha256Hex(body)
+		req.Header.Set("X-Amz-Content-Sha256", payloadHash)
+
+		signer := v4.NewSigner()
+		region := opts.Region
+		if region == "" {
+			region = "auto"
+		}
+
+		err = signer.SignHTTP(ctx, creds, req, payloadHash, "s3", region, time.Now())
+		if err != nil {
+			return nil, fmt.Errorf("storage: BundleObjects: failed to sign request: %w", err)
+		}
+	}
+
+	resp, err := bundleHTTPClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("storage: BundleObjects: request failed: %w", err)
+	}
+
+	if resp.StatusCode >= 400 {
+		defer resp.Body.Close()
+		errBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		return nil, fmt.Errorf("storage: BundleObjects: HTTP %d: %s", resp.StatusCode, string(errBody))
+	}
+
+	return &BundleObjectsOutput{
+		Body:        resp.Body,
+		ContentType: resp.Header.Get("Content-Type"),
+		StatusCode:  resp.StatusCode,
+	}, nil
+}
+
+func sha256Hex(data []byte) string {
+	h := sha256.Sum256(data)
+	return hex.EncodeToString(h[:])
+}

bundle_test.go

@@ -0,0 +1,161 @@
+package storage
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestBundleObjects_Validation(t *testing.T) {
+	cli := &Client{}
+
+	t.Run("missing bucket", func(t *testing.T) {
+		_, err := cli.BundleObjects(context.Background(), &BundleObjectsInput{
+			Keys: []string{"a"},
+		})
+		if err == nil {
+			t.Fatal("expected error for missing bucket")
+		}
+	})
+
+	t.Run("missing keys", func(t *testing.T) {
+		_, err := cli.BundleObjects(context.Background(), &BundleObjectsInput{
+			Bucket: "test-bucket",
+		})
+		if err == nil {
+			t.Fatal("expected error for missing keys")
+		}
+	})
+
+	t.Run("empty keys", func(t *testing.T) {
+		_, err := cli.BundleObjects(context.Background(), &BundleObjectsInput{
+			Bucket: "test-bucket",
+			Keys:   []string{},
+		})
+		if err == nil {
+			t.Fatal("expected error for empty keys")
+		}
+	})
+}
+
+func TestBundleObjects_RequestConstruction(t *testing.T) {
+	var capturedReq *http.Request
+	var capturedBody []byte
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		capturedReq = r
+		capturedBody, _ = io.ReadAll(r.Body)
+		w.Header().Set("Content-Type", "application/x-tar")
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	cli, err := New(context.Background(),
+		WithEndpoint(server.URL),
+		WithAccessKeypair("test-key", "test-secret"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("default options", func(t *testing.T) {
+		output, err := cli.BundleObjects(context.Background(), &BundleObjectsInput{
+			Bucket: "my-bucket",
+			Keys:   []string{"a.jpg", "b.jpg"},
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer output.Body.Close()
+
+		if capturedReq.Method != "POST" {
+			t.Errorf("method = %q, want POST", capturedReq.Method)
+		}
+		if capturedReq.URL.Path != "/my-bucket" {
+			t.Errorf("path = %q, want /my-bucket", capturedReq.URL.Path)
+		}
+		if capturedReq.URL.Query().Get("bundle") != "" {
+			// query param "bundle" should be present with empty value
+		} else if !capturedReq.URL.Query().Has("bundle") {
+			t.Error("missing ?bundle query parameter")
+		}
+
+		if capturedReq.Header.Get("Content-Type") != "application/json" {
+			t.Errorf("content-type = %q, want application/json", capturedReq.Header.Get("Content-Type"))
+		}
+		if capturedReq.Header.Get("X-Tigris-Bundle-Format") != "tar" {
+			t.Errorf("bundle-format = %q, want tar", capturedReq.Header.Get("X-Tigris-Bundle-Format"))
+		}
+		if capturedReq.Header.Get("X-Tigris-Bundle-Compression") != "none" {
+			t.Errorf("compression = %q, want none", capturedReq.Header.Get("X-Tigris-Bundle-Compression"))
+		}
+		if capturedReq.Header.Get("X-Tigris-Bundle-On-Error") != "skip" {
+			t.Errorf("on-error = %q, want skip", capturedReq.Header.Get("X-Tigris-Bundle-On-Error"))
+		}
+
+		// Verify body contains keys.
+		var body bundleRequestBody
+		if err := json.Unmarshal(capturedBody, &body); err != nil {
+			t.Fatalf("failed to unmarshal body: %v", err)
+		}
+		if len(body.Keys) != 2 || body.Keys[0] != "a.jpg" || body.Keys[1] != "b.jpg" {
+			t.Errorf("body keys = %v, want [a.jpg b.jpg]", body.Keys)
+		}
+
+		// Verify SigV4 authorization header is present.
+		if capturedReq.Header.Get("Authorization") == "" {
+			t.Error("missing Authorization header (SigV4)")
+		}
+
+		if output.ContentType != "application/x-tar" {
+			t.Errorf("content-type = %q, want application/x-tar", output.ContentType)
+		}
+	})
+
+	t.Run("custom compression and error mode", func(t *testing.T) {
+		output, err := cli.BundleObjects(context.Background(), &BundleObjectsInput{
+			Bucket:      "my-bucket",
+			Keys:        []string{"x.txt"},
+			Compression: BundleCompressionGzip,
+			OnError:     BundleOnErrorFail,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer output.Body.Close()
+
+		if capturedReq.Header.Get("X-Tigris-Bundle-Compression") != "gzip" {
+			t.Errorf("compression = %q, want gzip", capturedReq.Header.Get("X-Tigris-Bundle-Compression"))
+		}
+		if capturedReq.Header.Get("X-Tigris-Bundle-On-Error") != "fail" {
+			t.Errorf("on-error = %q, want fail", capturedReq.Header.Get("X-Tigris-Bundle-On-Error"))
+		}
+	})
+}
+
+func TestBundleObjects_HTTPError(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte(`<Error><Code>InvalidArgument</Code></Error>`))
+	}))
+	defer server.Close()
+
+	cli, err := New(context.Background(),
+		WithEndpoint(server.URL),
+		WithAccessKeypair("test-key", "test-secret"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = cli.BundleObjects(context.Background(), &BundleObjectsInput{
+		Bucket: "my-bucket",
+		Keys:   []string{"a.jpg"},
+	})
+	if err == nil {
+		t.Fatal("expected error for HTTP 400")
+	}
+}