Support for Conditionnal Write

This commit introduces support for conditional writes for
major cloud providers: aws, gcp, azure.

It does so my adding a new write options: IfNotExist. When
set to true, conditions are set for each cloud storage driver
to enable the desired behavior.

Author: ocampeau

blob/azureblob/azureblob.go

@@ -913,6 +913,14 @@ func (b *bucket) NewTypedWriter(ctx context.Context, key, contentType string, op
 			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,20 @@ 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.
+	// For 'fileblob' and 'memblob', an error of type 'PreconditionFailed'
+	// will be returned by the 'Close' method. This is due to how those driver
+	// are implemented, which buffers the content during writes and perform
+	// the actual write at close time.
+	// For cloud providers (aws, azure, google), an error of type 'PreconditionFailed'
+	// will be returned by the 'Write' method, since a call to 'Write' results
+	// in an API call to the cloud provider.
+	// Default is set to 'false'.
+	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,70 @@ 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
+//	if _, err := w2.Write([]byte(contents)); err != nil {
+//		t.Fatal(err)
+//	}
+//
+//	// Closing the file (ie: thus writing the content to the file)
+//	// should return a 'PreconditionFailed' error since the file
+//	// already exists
+//	if err = w2.Close(); err == nil {
+//		t.Fatal(err)
+//	}
+//}
+
 func benchmarkRead(b *testing.B, bkt *blob.Bucket) {
 	b.Helper()
 

blob/fileblob/fileblob.go

@@ -739,9 +739,10 @@ func (b *bucket) NewTypedWriter(ctx context.Context, key, contentType string, op
 
 	if b.opts.Metadata == MetadataDontWrite {
 		w := &writer{
-			ctx:  ctx,
-			File: f,
-			path: path,
+			ctx:        ctx,
+			File:       f,
+			path:       path,
+			ifNotExist: opts.IfNotExist,
 		}
 		return w, nil
 	}
@@ -765,6 +766,7 @@ func (b *bucket) NewTypedWriter(ctx context.Context, key, contentType string, op
 		attrs:      attrs,
 		contentMD5: opts.ContentMD5,
 		md5hash:    md5.New(),
+		ifNotExist: opts.IfNotExist,
 	}
 	return w, nil
 }
@@ -778,7 +780,8 @@ type writerWithSidecar struct {
 	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
 }
 
 func (w *writerWithSidecar) Write(p []byte) (n int, err error) {
@@ -794,33 +797,55 @@ func (w *writerWithSidecar) Write(p []byte) (n int, err error) {
 	return n, nil
 }
 
-func (w *writerWithSidecar) Close() error {
-	err := w.f.Close()
-	if err != nil {
-		return err
-	}
+func (w *writerWithSidecar) Close() (err error) {
 	// Always delete the temp file. On success, it will have been renamed so
 	// the Remove will fail.
 	defer func() {
 		_ = os.Remove(w.f.Name())
 	}()
 
 	// Check if the write was cancelled.
-	if err := w.ctx.Err(); err != nil {
+	if err = w.ctx.Err(); err != nil {
 		return err
 	}
 
 	md5sum := w.md5hash.Sum(nil)
 	w.attrs.MD5 = md5sum
 
 	// Write the attributes file.
-	if err := setAttrs(w.path, w.attrs); err != nil {
+	if err = setAttrs(w.path, w.attrs); err != nil {
 		return err
 	}
-	// Rename the temp file to path.
-	if err := os.Rename(w.f.Name(), w.path); err != nil {
-		_ = os.Remove(w.path + attrsExt)
-		return err
+
+	defer func() {
+		if err != nil {
+			_ = os.Remove(w.path + attrsExt)
+		}
+	}()
+
+	if w.ifNotExist {
+		fileOptions := os.O_RDWR | os.O_CREATE | os.O_EXCL
+		fd, err := os.OpenFile(w.path, fileOptions, 0666)
+		if err != nil {
+			return gcerr.New(gcerrors.FailedPrecondition, err, 1, "File already exist")
+		}
+		defer func() { _ = fd.Close() }()
+
+		// Set offset at the beginning of the file
+		_, err = w.f.Seek(0, 0)
+		if err != nil {
+			return err
+		}
+
+		// Copy content from the temp file to the new file
+		_, err = w.f.WriteTo(fd)
+		if err != nil {
+			return err
+		}
+	} else {
+		if err := os.Rename(w.f.Name(), w.path); err != nil {
+			return err
+		}
 	}
 	return nil
 }
@@ -831,8 +856,9 @@ func (w *writerWithSidecar) Close() error {
 // 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
 }
 
 func (w *writer) Upload(r io.Reader) error {
@@ -841,10 +867,6 @@ func (w *writer) Upload(r io.Reader) error {
 }
 
 func (w *writer) Close() error {
-	err := w.File.Close()
-	if err != nil {
-		return err
-	}
 	// Always delete the temp file. On success, it will have been renamed so
 	// the Remove will fail.
 	tempname := w.File.Name()
@@ -855,9 +877,29 @@ func (w *writer) Close() error {
 		return err
 	}
 
-	// Rename the temp file to path.
-	if err := os.Rename(tempname, w.path); err != nil {
-		return err
+	if w.ifNotExist {
+		fileOptions := os.O_RDWR | os.O_CREATE | os.O_EXCL
+		fd, err := os.OpenFile(w.path, fileOptions, 0666)
+		if err != nil {
+			return gcerr.New(gcerrors.FailedPrecondition, err, 1, "File already exist")
+		}
+		defer func() { _ = fd.Close() }()
+
+		// Set offset at the beginning of the file
+		_, err = w.File.Seek(0, 0)
+		if err != nil {
+			return err
+		}
+
+		// Copy content from the temp file to the new file
+		_, err = w.File.WriteTo(fd)
+		if err != nil {
+			return err
+		}
+	} else {
+		if err := os.Rename(w.File.Name(), w.path); err != nil {
+			return err
+		}
 	}
 	return nil
 }

blob/gcsblob/gcsblob.go

@@ -626,6 +626,9 @@ func (b *bucket) NewTypedWriter(ctx context.Context, key, contentType string, op
 	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

@@ -291,6 +291,7 @@ func (b *bucket) NewTypedWriter(ctx context.Context, key, contentType string, op
 	for k, v := range opts.Metadata {
 		md[k] = v
 	}
+
 	return &writer{
 		ctx:         ctx,
 		b:           b,
@@ -299,6 +300,7 @@ func (b *bucket) NewTypedWriter(ctx context.Context, key, contentType string, op
 		metadata:    md,
 		opts:        opts,
 		md5hash:     md5.New(),
+		ifNotExist:  opts.IfNotExist,
 	}, nil
 }
 
@@ -312,7 +314,8 @@ type writer struct {
 	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,9 @@ func (w *writer) Close() error {
 	w.b.mu.Lock()
 	defer w.b.mu.Unlock()
 	if prev := w.b.blobs[w.key]; prev != nil {
+		if w.ifNotExist {
+			return errors.New("'IfNotExists' is set and blob already exists for key")
+		}
 		entry.Attributes.CreateTime = prev.Attributes.CreateTime
 	}
 	w.b.blobs[w.key] = entry

blob/s3blob/s3blob.go

@@ -1062,6 +1062,10 @@ func (b *bucket) NewTypedWriter(ctx context.Context, key, contentType string, op
 			Key:         aws.String(key),
 			Metadata:    md,
 		}
+		if opts.IfNotExist {
+			// See https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-writes.html
+			reqV2.IfNoneMatch = aws.String("*")
+		}
 		if opts.CacheControl != "" {
 			reqV2.CacheControl = aws.String(opts.CacheControl)
 		}