blob: introduce conditional write option for storage

blob/azureblob/azureblob.go

@@ -913,6 +913,14 @@
 			BlobContentType:        &contentType,
 		},
 	}
+	if opts.IfNotExist {
+		etagAny := azcore.ETagAny
+		uploadOpts.AccessConditions = &azblob.AccessConditions{
+			ModifiedAccessConditions: &azblobblob.ModifiedAccessConditions{
+				IfNoneMatch: &etagAny,
+			},
+		}
+	}
 	if opts.BeforeWrite != nil {
 		asFunc := func(i any) bool {
 			p, ok := i.(**azblob.UploadStreamOptions)

blob/blob.go

@@ -1098,6 +1098,7 @@ func (b *Bucket) NewWriter(ctx context.Context, key string, opts *WriterOptions)
 		MaxConcurrency:              opts.MaxConcurrency,
 		BeforeWrite:                 opts.BeforeWrite,
 		DisableContentTypeDetection: opts.DisableContentTypeDetection,
+		IfNotExist:                  opts.IfNotExist,
 	}
 	if len(opts.Metadata) > 0 {
 		// Services are inconsistent, but at least some treat keys
@@ -1422,6 +1423,13 @@ type WriterOptions struct {
 	// asFunc converts its argument to driver-specific types.
 	// See https://gocloud.dev/concepts/as/ for background information.
 	BeforeWrite func(asFunc func(any) bool) error
+
+	// IfNotExist is used for conditional writes. When set to 'true',
+	// if a blob exists for the same key in the bucket, the write
+	// operation won't succeed and the current blob for the key will
+	// be left untouched. An error for which gcerrors.Code will return
+	// gcerrors.PreconditionFailed will be returned by Write or Close.
+	IfNotExist bool
 }
 
 // CopyOptions sets options for Copy.

blob/driver/driver.go

@@ -110,6 +110,11 @@ type WriterOptions struct {
 	// asFunc allows drivers to expose driver-specific types;
 	// see Bucket.As for more details.
 	BeforeWrite func(asFunc func(any) bool) error
+
+	// IfNotExist is used for conditional writes.
+	// When set to true, if a blob exists for the same key in the bucket, the write operation
+	// won't take place.
+	IfNotExist bool
 }
 
 // CopyOptions controls options for Copy.

blob/drivertest/drivertest.go

@@ -260,6 +260,9 @@ func RunConformanceTests(t *testing.T, newHarness HarnessMaker, asTests []AsTest
 	t.Run("TestSignedURL", func(t *testing.T) {
 		testSignedURL(t, newHarness)
 	})
+	//t.Run("TestIfNotExist", func(t *testing.T) {
+	//	testIfNotExist(t, newHarness)
+	//})
 	asTests = append(asTests, verifyAsFailsOnNil{})
 	t.Run("TestAs", func(t *testing.T) {
 		for _, st := range asTests {
@@ -2734,6 +2737,73 @@ func testAs(t *testing.T, newHarness HarnessMaker, st AsTest) {
 	}
 }
 
+func testIfNotExist(t *testing.T, newHarness HarnessMaker) {
+	t.Helper()
+
+	const key = "blob-for-if-not-exist"
+	const contents = "up and down"
+
+	ctx := context.Background()
+	h, err := newHarness(ctx, t)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer h.Close()
+	drv, err := h.MakeDriver(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	b := blob.NewBucket(drv)
+	defer func() { _ = b.Close() }()
+
+	opts := blob.WriterOptions{
+		ContentType: "text",
+		IfNotExist:  true,
+	}
+
+	// Create one file for the key
+	w1, err := b.NewWriter(ctx, key, &opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		_ = b.Delete(ctx, key)
+	}()
+
+	// Write to the file
+	if _, err := w1.Write([]byte(contents)); err != nil {
+		t.Fatal(err)
+	}
+
+	// Closing the file (ie: thus writing the content to the file)
+	// should not return an error
+	if err := w1.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Create a new writer for the same key
+	w2, err := b.NewWriter(ctx, key, &opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Write to the file
+	// We expect an error either from `Write` or `Close`
+	if _, err = w2.Write([]byte(contents)); err == nil {
+		err = w2.Close()
+	} else {
+		_ = w2.Close()
+	}
+
+	if err == nil {
+		t.Error("expected error rewriting key with IfNotExist, got nil")
+	}
+	if code := gcerrors.Code(err); code != gcerrors.FailedPrecondition {
+		t.Errorf("expected FailedPrecondition error, got %v", code)
+	}
+}
+
 func benchmarkRead(b *testing.B, bkt *blob.Bucket) {
 	b.Helper()
 

blob/fileblob/fileblob.go

@@ -78,6 +78,7 @@
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"gocloud.dev/blob"
@@ -739,9 +740,11 @@
 
 	if b.opts.Metadata == MetadataDontWrite {
 		w := &writer{
-			ctx:  ctx,
-			File: f,
-			path: path,
+			ctx:        ctx,
+			File:       f,
+			path:       path,
+			ifNotExist: opts.IfNotExist,
+			mu:         &sync.Mutex{},
 		}
 		return w, nil
 	}
@@ -765,6 +768,8 @@
 		attrs:      attrs,
 		contentMD5: opts.ContentMD5,
 		md5hash:    md5.New(),
+		ifNotExist: opts.IfNotExist,
+		mu:         &sync.Mutex{},
 	}
 	return w, nil
 }
@@ -778,7 +783,9 @@
 	contentMD5 []byte
 	// We compute the MD5 hash so that we can store it with the file attributes,
 	// not for verification.
-	md5hash hash.Hash
+	md5hash    hash.Hash
+	ifNotExist bool
+	mu         *sync.Mutex
 }
 
 func (w *writerWithSidecar) Write(p []byte) (n int, err error) {
@@ -817,6 +824,15 @@
 	if err := setAttrs(w.path, w.attrs); err != nil {
 		return err
 	}
+
+	if w.ifNotExist {
+		w.mu.Lock()
+		defer w.mu.Unlock()
+		_, err = os.Stat(w.path)
+		if err == nil {
+			return gcerr.New(gcerrors.FailedPrecondition, err, 1, "File already exist")
+		}
+	}
 	// Rename the temp file to path.
 	if err := os.Rename(w.f.Name(), w.path); err != nil {
 		_ = os.Remove(w.path + attrsExt)
@@ -831,8 +847,10 @@
 // which is why it is not folded into writerWithSidecar.
 type writer struct {
 	*os.File
-	ctx  context.Context
-	path string
+	ctx        context.Context
+	path       string
+	ifNotExist bool
+	mu         *sync.Mutex
 }
 
 func (w *writer) Upload(r io.Reader) error {
@@ -855,6 +873,15 @@
 		return err
 	}
 
+	if w.ifNotExist {
+		w.mu.Lock()
+		defer w.mu.Unlock()
+		_, err = os.Stat(w.path)
+		if err == nil {
+			return gcerr.New(gcerrors.FailedPrecondition, err, 1, "File already exist")
+		}
+	}
+
 	// Rename the temp file to path.
 	if err := os.Rename(tempname, w.path); err != nil {
 		return err

blob/gcsblob/gcsblob.go

@@ -626,6 +626,9 @@
 	bkt := b.client.Bucket(b.name)
 	obj := bkt.Object(key)
 
+	if opts.IfNotExist {
+		obj = obj.If(storage.Conditions{DoesNotExist: true})
+	}
 	// Add an extra level of indirection so that BeforeWrite can replace obj
 	// if needed. For example, ObjectHandle.If returns a new ObjectHandle.
 	// Also, make the Writer lazily in case this replacement happens.

blob/memblob/memblob.go

@@ -33,6 +33,7 @@
 	"crypto/md5"
 	"errors"
 	"fmt"
+	"gocloud.dev/internal/gcerr"
 	"hash"
 	"io"
 	"net/url"
@@ -299,6 +300,7 @@
 		metadata:    md,
 		opts:        opts,
 		md5hash:     md5.New(),
+		ifNotExist:  opts.IfNotExist,
 	}, nil
 }
 
@@ -312,7 +314,8 @@
 	buf         bytes.Buffer
 	// We compute the MD5 hash so that we can store it with the file attributes,
 	// not for verification.
-	md5hash hash.Hash
+	md5hash    hash.Hash
+	ifNotExist bool
 }
 
 func (w *writer) Write(p []byte) (n int, err error) {
@@ -355,6 +358,10 @@
 	w.b.mu.Lock()
 	defer w.b.mu.Unlock()
 	if prev := w.b.blobs[w.key]; prev != nil {
+		if w.ifNotExist {
+			err := errors.New("file already exist")
+			return gcerr.New(gcerrors.FailedPrecondition, err, 1, "a blob already exist for key")
+		}
 		entry.Attributes.CreateTime = prev.Attributes.CreateTime
 	}
 	w.b.blobs[w.key] = entry

blob/s3blob/s3blob.go

@@ -748,6 +748,11 @@
 		Key:         aws.String(key),
 		Metadata:    md,
 	}
+
+	if opts.IfNotExist {
+		// See https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-writes.html
+		req.IfNoneMatch = aws.String("*")
+	}
 	if opts.CacheControl != "" {
 		req.CacheControl = aws.String(opts.CacheControl)
 	}