diff --git a/alioss.go b/alioss.go index e420edf..caea0d9 100644 --- a/alioss.go +++ b/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://") diff --git a/azure.go b/azure.go index 375ec8e..dbbff25 100644 --- a/azure.go +++ b/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 diff --git a/conditional_test.go b/conditional_test.go new file mode 100644 index 0000000..14df9c1 --- /dev/null +++ b/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") +} diff --git a/gcp.go b/gcp.go index c68b32a..0be2807 100644 --- a/gcp.go +++ b/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 diff --git a/local.go b/local.go index 03677fa..bd3f051 100644 --- a/local.go +++ b/local.go @@ -15,6 +15,9 @@ package storage import ( + "context" + "errors" + "fmt" "io" "net/url" "os" @@ -91,6 +94,82 @@ func (u *localUploader) UploadData(data []byte, storagePath, _ string) (string, return storagePath, int64(size), nil } +var ( + _ ConditionalUploader = (*localUploader)(nil) + _ RangeDownloader = (*localUploader)(nil) +) + +// createIfAbsent opens the destination for exclusive creation, which is the +// filesystem's own atomic check-and-create. +func (u *localUploader) createIfAbsent(storagePath string) (*os.File, error) { + if dir, _ := path.Split(storagePath); dir != "" { + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, err + } + } + f, err := os.OpenFile(storagePath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0644) + if errors.Is(err, os.ErrExist) { + return nil, fmt.Errorf("%w: %s", ErrObjectExists, storagePath) + } + return f, err +} + +func (u *localUploader) UploadDataIfAbsent(_ context.Context, data []byte, storagePath, _ string) (string, int64, error) { + storagePath = u.location(storagePath) + f, err := u.createIfAbsent(storagePath) + if err != nil { + return "", 0, err + } + defer f.Close() + size, err := f.Write(data) + if err != nil { + return "", 0, err + } + return storagePath, int64(size), nil +} + +func (u *localUploader) UploadFileIfAbsent(_ context.Context, localPath, storagePath, _ string) (string, int64, error) { + storagePath = u.location(storagePath) + local, err := os.Open(localPath) + if err != nil { + return "", 0, err + } + defer local.Close() + f, err := u.createIfAbsent(storagePath) + if err != nil { + return "", 0, err + } + defer f.Close() + size, err := io.Copy(f, local) + if err != nil { + return "", 0, err + } + return storagePath, size, nil +} + +func (u *localUploader) DownloadRange(_ 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) + } + f, err := os.Open(path.Join(u.StorageDir, storagePath)) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, fmt.Errorf("%w: %s", ErrNotFound, storagePath) + } + return nil, err + } + defer f.Close() + buf := make([]byte, n) + read, err := f.ReadAt(buf, off) + if err != nil && !errors.Is(err, io.EOF) { + return nil, err + } + if read == 0 { + return nil, fmt.Errorf("%w: range %d+%d past the end of %s", ErrNotFound, off, n, storagePath) + } + return buf[:read], nil +} + func (u *localUploader) location(storagePath string) string { return path.Join(u.StorageDir, storagePath) } diff --git a/s3.go b/s3.go index ecba17d..5f0eef7 100644 --- a/s3.go +++ b/s3.go @@ -47,14 +47,86 @@ func wrapS3Error(err error) error { } var sc interface{ HTTPStatusCode() int } if errors.As(err, &sc) { + code := sc.HTTPStatusCode() + // The two outcomes callers decide on carry a sentinel as well as + // the status, so errors.Is works without knowing S3's codes. + switch 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: sc.HTTPStatusCode(), + StatusCode: code, } } return err } +var ( + _ ConditionalUploader = (*s3Storage)(nil) + _ RangeDownloader = (*s3Storage)(nil) +) + +// UploadDataIfAbsent stores data at storagePath unless an object is there, +// with S3's If-None-Match condition, so the check and the write are one +// request. +func (s *s3Storage) UploadDataIfAbsent(ctx context.Context, data []byte, storagePath, contentType string) (string, int64, error) { + loc, err := s.uploadWith(ctx, bytes.NewReader(data), storagePath, contentType, true) + if err != nil { + return "", 0, err + } + return loc, int64(len(data)), nil +} + +// UploadFileIfAbsent stores the file at filepath at storagePath unless an +// object is there. The condition rides the multipart completion, so a large +// upload is as atomic as a small one. +func (s *s3Storage) 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 + } + loc, err := s.uploadWith(ctx, file, storagePath, contentType, true) + if err != nil { + return "", 0, err + } + return loc, stat.Size(), nil +} + +// DownloadRange reads n bytes from offset off with a Range request. S3 +// answers a start past the end with 416, which is ErrNotFound here. +func (s *s3Storage) 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) + } + client := s.getClient(nil) + out, err := client.GetObject(ctx, &s3.GetObjectInput{ + Bucket: aws.String(s.conf.Bucket), + Key: aws.String(storagePath), + Range: aws.String(fmt.Sprintf("bytes=%d-%d", off, off+n-1)), + }) + if err != nil { + var sc interface{ HTTPStatusCode() int } + if errors.As(err, &sc) && sc.HTTPStatusCode() == http.StatusRequestedRangeNotSatisfiable { + return nil, &ErrorWithStatusCode{Err: fmt.Errorf("%w: range %d+%d past the end: %w", ErrNotFound, off, n, err), StatusCode: http.StatusRequestedRangeNotSatisfiable} + } + return nil, wrapS3Error(err) + } + defer out.Body.Close() + data, err := io.ReadAll(io.LimitReader(out.Body, n)) + if err != nil { + return nil, err + } + return data, nil +} + const defaultBucketLocation = "us-east-1" type s3Storage struct { @@ -223,6 +295,12 @@ func (s *s3Storage) UploadFile(filepath, storagePath, contentType string) (strin } func (s *s3Storage) upload(reader io.Reader, storagePath, contentType string) (string, error) { + return s.uploadWith(context.Background(), reader, storagePath, contentType, false) +} + +// uploadWith is the one upload path: ifAbsent adds If-None-Match: * so the +// object is written only when none is there. +func (s *s3Storage) uploadWith(ctx context.Context, reader io.Reader, storagePath, contentType string, ifAbsent bool) (string, error) { l := NewS3Logger() client := s.getClient(l) @@ -233,6 +311,9 @@ func (s *s3Storage) upload(reader io.Reader, storagePath, contentType string) (s Key: aws.String(storagePath), Metadata: s.conf.Metadata, } + if ifAbsent { + input.IfNoneMatch = aws.String("*") + } if s.conf.Tagging != "" { input.Tagging = &s.conf.Tagging } @@ -270,8 +351,10 @@ func (s *s3Storage) upload(reader io.Reader, storagePath, contentType string) (s } }) - if _, err := uploader.Upload(context.Background(), input); err != nil { - l.WriteLogs() + if _, err := uploader.Upload(ctx, input); err != nil { + if !errors.Is(wrapS3Error(err), ErrObjectExists) { + l.WriteLogs() + } return "", wrapS3Error(err) } diff --git a/storage.go b/storage.go index 0317964..7446090 100644 --- a/storage.go +++ b/storage.go @@ -14,9 +14,45 @@ package storage -import "time" +import ( + "context" + "errors" + "time" +) +// Errors a backend maps its own to, so callers can decide with errors.Is +// whether an object was missing or a conditional upload lost. They come +// wrapped in the backend's error, which keeps the status code and message. +var ( + ErrNotFound = errors.New("storage: object not found") + ErrObjectExists = errors.New("storage: object already exists") +) + +// ConditionalUploader is the part of Storage whose uploads are conditional +// on the object not existing. Two writers racing for one path see exactly +// one succeed; the other gets ErrObjectExists and changes nothing. Every +// backend implements it: S3 with If-None-Match, GCS with a generation +// precondition, Azure with an ETag condition, OSS with forbid-overwrite, +// local storage with O_EXCL. +type ConditionalUploader interface { + UploadDataIfAbsent(ctx context.Context, data []byte, storagePath, contentType string) (location string, size int64, err error) + UploadFileIfAbsent(ctx context.Context, filepath, storagePath, contentType string) (location string, size int64, err error) +} + +// RangeDownloader is the part of Storage that reads part of an object: n +// bytes from offset off, fewer at the end of the object. An offset past the +// end, or a missing object, is ErrNotFound. +type RangeDownloader interface { + DownloadRange(ctx context.Context, storagePath string, off, n int64) ([]byte, error) +} + +// Storage is one bucket of objects. The conditional and ranged operations +// take a context because a storage-backed service bounds them per request; +// the others predate that and keep their shape. type Storage interface { + ConditionalUploader + RangeDownloader + UploadData(data []byte, storagePath, contentType string) (location string, size int64, err error) UploadFile(filepath, storagePath, contentType string) (location string, size int64, err error)