Conditional uploads and ranged downloads on every backend

alioss.go

@@ -16,8 +16,11 @@ package storage
 
 import (
 	"bytes"
+	"context"
 	"errors"
+	"fmt"
 	"io"
+	"net/http"
 	"net/url"
 	"os"
 	"strings"
@@ -32,6 +35,14 @@ func wrapAliOSSError(err error) error {
 	}
 	var svcErr oss.ServiceError
 	if errors.As(err, &svcErr) {
+		switch {
+		case svcErr.StatusCode == http.StatusNotFound:
+			err = fmt.Errorf("%w: %w", ErrNotFound, err)
+		case svcErr.Code == "FileAlreadyExists":
+			err = fmt.Errorf("%w: %w", ErrObjectExists, err)
+		case svcErr.StatusCode == http.StatusRequestedRangeNotSatisfiable:
+			err = fmt.Errorf("%w: %w", ErrNotFound, err)
+		}
 		return &ErrorWithStatusCode{
 			Err:        err,
 			StatusCode: svcErr.StatusCode,
@@ -91,6 +102,42 @@ func (s *aliOSSStorage) UploadFile(filepath, storagePath, contentType string) (s
 	return s.location(storagePath), info.Size(), nil
 }
 
+// UploadDataIfAbsent writes with forbid-overwrite: OSS refuses the write
+// with 409 FileAlreadyExists when an object is there.
+func (s *aliOSSStorage) UploadDataIfAbsent(ctx context.Context, data []byte, storagePath, contentType string) (string, int64, error) {
+	if err := s.bucket.PutObject(storagePath, bytes.NewReader(data), oss.ContentType(contentType), oss.ForbidOverWrite(true), oss.WithContext(ctx)); err != nil {
+		return "", 0, wrapAliOSSError(err)
+	}
+	return s.location(storagePath), int64(len(data)), nil
+}
+
+func (s *aliOSSStorage) UploadFileIfAbsent(ctx context.Context, filepath, storagePath, contentType string) (string, int64, error) {
+	stat, err := os.Stat(filepath)
+	if err != nil {
+		return "", 0, err
+	}
+	if err := s.bucket.PutObjectFromFile(storagePath, filepath, oss.ContentType(contentType), oss.ForbidOverWrite(true), oss.WithContext(ctx)); err != nil {
+		return "", 0, wrapAliOSSError(err)
+	}
+	return s.location(storagePath), stat.Size(), nil
+}
+
+func (s *aliOSSStorage) DownloadRange(ctx context.Context, storagePath string, off, n int64) ([]byte, error) {
+	if off < 0 || n <= 0 {
+		return nil, fmt.Errorf("storage: invalid range %d+%d", off, n)
+	}
+	reader, err := s.bucket.GetObject(storagePath, oss.Range(off, off+n-1), oss.WithContext(ctx))
+	if err != nil {
+		return nil, wrapAliOSSError(err)
+	}
+	defer reader.Close()
+	data, err := io.ReadAll(reader)
+	if err != nil {
+		return nil, wrapAliOSSError(err)
+	}
+	return data, nil
+}
+
 func (s *aliOSSStorage) location(storagePath string) string {
 	endpoint := strings.TrimPrefix(s.conf.Endpoint, "https://")
 	endpoint = strings.TrimPrefix(endpoint, "http://")

azure.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
+	"net/http"
 	"net/url"
 	"os"
 	"path"
@@ -27,6 +29,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/sas"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service"
 )
@@ -37,6 +40,12 @@ func wrapAzureError(err error) error {
 	}
 	var respErr *azcore.ResponseError
 	if errors.As(err, &respErr) {
+		switch {
+		case respErr.StatusCode == http.StatusNotFound:
+			err = fmt.Errorf("%w: %w", ErrNotFound, err)
+		case bloberror.HasCode(err, bloberror.BlobAlreadyExists):
+			err = fmt.Errorf("%w: %w", ErrObjectExists, err)
+		}
 		return &ErrorWithStatusCode{
 			Err:        err,
 			StatusCode: respErr.StatusCode,
@@ -45,6 +54,69 @@ func wrapAzureError(err error) error {
 	return err
 }
 
+// ifNoneMatchAny is the access condition that refuses the write when a blob
+// is there: Azure answers 409 BlobAlreadyExists.
+func ifNoneMatchAny() *blob.AccessConditions {
+	return &blob.AccessConditions{ModifiedAccessConditions: &blob.ModifiedAccessConditions{IfNoneMatch: to.Ptr(azcore.ETagAny)}}
+}
+
+func (s *azureBLOBStorage) UploadDataIfAbsent(ctx context.Context, data []byte, storagePath, contentType string) (string, int64, error) {
+	_, err := s.client.UploadBuffer(ctx, s.conf.ContainerName, storagePath, data, &azblob.UploadBufferOptions{
+		HTTPHeaders:      &blob.HTTPHeaders{BlobContentType: &contentType},
+		BlockSize:        azureBlockSize,
+		Concurrency:      azureParallelism,
+		AccessConditions: ifNoneMatchAny(),
+	})
+	if err != nil {
+		return "", 0, wrapAzureError(err)
+	}
+	return s.location(storagePath), int64(len(data)), nil
+}
+
+func (s *azureBLOBStorage) UploadFileIfAbsent(ctx context.Context, filepath, storagePath, contentType string) (string, int64, error) {
+	file, err := os.Open(filepath)
+	if err != nil {
+		return "", 0, err
+	}
+	defer file.Close()
+	stat, err := file.Stat()
+	if err != nil {
+		return "", 0, err
+	}
+	_, err = s.client.UploadFile(ctx, s.conf.ContainerName, storagePath, file, &azblob.UploadFileOptions{
+		HTTPHeaders:      &blob.HTTPHeaders{BlobContentType: &contentType},
+		BlockSize:        azureBlockSize,
+		Concurrency:      azureParallelism,
+		AccessConditions: ifNoneMatchAny(),
+	})
+	if err != nil {
+		return "", 0, wrapAzureError(err)
+	}
+	return s.location(storagePath), stat.Size(), nil
+}
+
+func (s *azureBLOBStorage) DownloadRange(ctx context.Context, storagePath string, off, n int64) ([]byte, error) {
+	if off < 0 || n <= 0 {
+		return nil, fmt.Errorf("storage: invalid range %d+%d", off, n)
+	}
+	resp, err := s.client.DownloadStream(ctx, s.conf.ContainerName, storagePath, &azblob.DownloadStreamOptions{
+		Range: blob.HTTPRange{Offset: off, Count: n},
+	})
+	if err != nil {
+		var respErr *azcore.ResponseError
+		if errors.As(err, &respErr) && respErr.StatusCode == http.StatusRequestedRangeNotSatisfiable {
+			return nil, &ErrorWithStatusCode{Err: fmt.Errorf("%w: range %d+%d past the end: %w", ErrNotFound, off, n, err), StatusCode: respErr.StatusCode}
+		}
+		return nil, wrapAzureError(err)
+	}
+	defer resp.Body.Close()
+	data, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, wrapAzureError(err)
+	}
+	return data, nil
+}
+
 const azureBlockSize = 4 * 1024 * 1024
 const azureParallelism = 16
 

conditional_test.go

@@ -0,0 +1,87 @@
+package storage
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// The conditional and ranged operations behave the same on every backend
+// that has them; run the suite against each one available.
+func testConditionalAndRanged(t *testing.T, s Storage, prefix string) {
+	t.Helper()
+	ctx := context.Background()
+	cu, rd := s, s
+	key := prefix + "/conditional"
+	t.Cleanup(func() { _ = s.DeleteObjects([]string{key, prefix + "/file", prefix + "/ranged"}) })
+
+	if _, n, err := cu.UploadDataIfAbsent(ctx, []byte("first"), key, "text/plain"); err != nil || n != 5 {
+		t.Fatalf("first conditional upload: %v %d", err, n)
+	}
+	if _, _, err := cu.UploadDataIfAbsent(ctx, []byte("second"), key, "text/plain"); !errors.Is(err, ErrObjectExists) {
+		t.Fatalf("second conditional upload: %v, want ErrObjectExists", err)
+	}
+	if got, err := s.DownloadData(key); err != nil || string(got) != "first" {
+		t.Fatalf("the refused upload changed the object: %q %v", got, err)
+	}
+
+	dir := t.TempDir()
+	f := filepath.Join(dir, "f")
+	if err := os.WriteFile(f, []byte(strings.Repeat("x", 100)), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if _, n, err := cu.UploadFileIfAbsent(ctx, f, prefix+"/file", "application/octet-stream"); err != nil || n != 100 {
+		t.Fatalf("conditional file upload: %v %d", err, n)
+	}
+	if _, _, err := cu.UploadFileIfAbsent(ctx, f, prefix+"/file", "application/octet-stream"); !errors.Is(err, ErrObjectExists) {
+		t.Fatalf("conditional file upload over an object: %v", err)
+	}
+
+	if _, _, err := s.UploadData([]byte("0123456789"), prefix+"/ranged", "text/plain"); err != nil {
+		t.Fatal(err)
+	}
+	for _, tc := range []struct {
+		off, n int64
+		want   string
+	}{
+		{2, 3, "234"}, {7, 100, "789"}, {0, 10, "0123456789"},
+	} {
+		got, err := rd.DownloadRange(ctx, prefix+"/ranged", tc.off, tc.n)
+		if err != nil || string(got) != tc.want {
+			t.Fatalf("range %d+%d: %q %v", tc.off, tc.n, got, err)
+		}
+	}
+	if _, err := rd.DownloadRange(ctx, prefix+"/ranged", 10, 1); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("range past the end: %v, want ErrNotFound", err)
+	}
+	if _, err := rd.DownloadRange(ctx, prefix+"/absent", 0, 1); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("range of a missing object: %v, want ErrNotFound", err)
+	}
+}
+
+func TestLocalConditionalAndRanged(t *testing.T) {
+	s, err := NewLocal(&LocalConfig{StorageDir: t.TempDir()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	testConditionalAndRanged(t, s, "p")
+}
+
+// TestS3ConditionalAndRanged runs against any S3-compatible endpoint, for
+// example MinIO: S3_ENDPOINT=http://127.0.0.1:9000 S3_BUCKET=b
+// S3_ACCESS_KEY=... S3_SECRET=... go test -run S3.
+func TestS3ConditionalAndRanged(t *testing.T) {
+	endpoint, bucket := os.Getenv("S3_ENDPOINT"), os.Getenv("S3_BUCKET")
+	key, secret := os.Getenv("S3_ACCESS_KEY"), os.Getenv("S3_SECRET")
+	if bucket == "" || key == "" || secret == "" {
+		t.Skip("Missing env vars")
+	}
+	s, err := NewS3(&S3Config{AccessKey: key, Secret: secret, Endpoint: endpoint, Bucket: bucket, Region: "us-east-1", ForcePathStyle: endpoint != "", MaxRetries: 3})
+	if err != nil {
+		t.Fatal(err)
+	}
+	testConditionalAndRanged(t, s, "conditional-test")
+}

gcp.go

@@ -42,12 +42,18 @@ func wrapGCPError(err error) error {
 	}
 	if errors.Is(err, storage.ErrBucketNotExist) || errors.Is(err, storage.ErrObjectNotExist) {
 		return &ErrorWithStatusCode{
-			Err:        err,
+			Err:        fmt.Errorf("%w: %w", ErrNotFound, err),
 			StatusCode: http.StatusNotFound,
 		}
 	}
 	var apiErr *googleapi.Error
 	if errors.As(err, &apiErr) {
+		switch apiErr.Code {
+		case http.StatusNotFound:
+			err = fmt.Errorf("%w: %w", ErrNotFound, err)
+		case http.StatusPreconditionFailed:
+			err = fmt.Errorf("%w: %w", ErrObjectExists, err)
+		}
 		return &ErrorWithStatusCode{
 			Err:        err,
 			StatusCode: apiErr.Code,
@@ -56,6 +62,73 @@ func wrapGCPError(err error) error {
 	return err
 }
 
+// gcsObjectNotFound is the not-found the client library reports as a
+// sentinel rather than a status.
+func gcsObjectNotFound(err error) error {
+	return &ErrorWithStatusCode{Err: fmt.Errorf("%w: %w", ErrNotFound, err), StatusCode: http.StatusNotFound}
+}
+
+func (s *gcpStorage) UploadDataIfAbsent(ctx context.Context, data []byte, storagePath, contentType string) (string, int64, error) {
+	return s.uploadIfAbsent(ctx, bytes.NewReader(data), storagePath, contentType)
+}
+
+func (s *gcpStorage) UploadFileIfAbsent(ctx context.Context, filepath, storagePath, contentType string) (string, int64, error) {
+	file, err := os.Open(filepath)
+	if err != nil {
+		return "", 0, err
+	}
+	defer file.Close()
+	return s.uploadIfAbsent(ctx, file, storagePath, contentType)
+}
+
+// uploadIfAbsent writes under a DoesNotExist precondition: GCS refuses the
+// write with 412 when an object is there.
+func (s *gcpStorage) uploadIfAbsent(ctx context.Context, reader io.Reader, storagePath, contentType string) (string, int64, error) {
+	wc := s.client.Bucket(s.conf.Bucket).Object(storagePath).
+		If(storage.Conditions{DoesNotExist: true}).
+		Retryer(
+			storage.WithBackoff(gax.Backoff{Initial: time.Millisecond * 100, Max: time.Second * 5, Multiplier: 2}),
+			storage.WithMaxAttempts(5),
+			storage.WithPolicy(storage.RetryAlways),
+		).NewWriter(ctx)
+	wc.ContentType = contentType
+	wc.ChunkRetryDeadline = 0
+	n, err := io.Copy(wc, reader)
+	if err != nil {
+		return "", 0, wrapGCPError(err)
+	}
+	if err = wc.Close(); err != nil {
+		return "", 0, wrapGCPError(err)
+	}
+	return s.location(storagePath), n, nil
+}
+
+func (s *gcpStorage) DownloadRange(ctx context.Context, storagePath string, off, n int64) ([]byte, error) {
+	if off < 0 || n <= 0 {
+		return nil, fmt.Errorf("storage: invalid range %d+%d", off, n)
+	}
+	r, err := s.client.Bucket(s.conf.Bucket).Object(storagePath).Retryer(
+		storage.WithBackoff(gax.Backoff{Initial: time.Millisecond * 100, Max: time.Second * 5, Multiplier: 2}),
+		storage.WithPolicy(storage.RetryAlways),
+	).NewRangeReader(ctx, off, n)
+	if err != nil {
+		if errors.Is(err, storage.ErrObjectNotExist) {
+			return nil, gcsObjectNotFound(err)
+		}
+		var apiErr *googleapi.Error
+		if errors.As(err, &apiErr) && apiErr.Code == http.StatusRequestedRangeNotSatisfiable {
+			return nil, &ErrorWithStatusCode{Err: fmt.Errorf("%w: range %d+%d past the end: %w", ErrNotFound, off, n, err), StatusCode: apiErr.Code}
+		}
+		return nil, wrapGCPError(err)
+	}
+	defer r.Close()
+	data, err := io.ReadAll(r)
+	if err != nil {
+		return nil, wrapGCPError(err)
+	}
+	return data, nil
+}
+
 type gcpStorage struct {
 	conf   *GCPConfig
 	client *storage.Client