use preconditions

Signed-off-by: Jason Hall <imjasonh@gmail.com>

writer.go

@@ -46,6 +46,7 @@ type Writer struct {
 	created bool // whether the GCS object exists yet
 	closed  bool
 	err     error
+	gen     int64 // expected generation of the GCS object
 
 	stopCh chan struct{}
 	doneCh chan struct{}
@@ -137,7 +138,8 @@ func (w *Writer) flushLocked() {
 
 	if !w.created {
 		// First write: create the object directly.
-		gw := objh.NewWriter(w.ctx)
+		// DoesNotExist ensures we don't overwrite an object created by someone else.
+		gw := objh.If(storage.Conditions{DoesNotExist: true}).NewWriter(w.ctx)
 		if _, err := gw.Write(data); err != nil {
 			w.err = fmt.Errorf("writing initial object: %w", err)
 			return
@@ -146,6 +148,7 @@ func (w *Writer) flushLocked() {
 			w.err = fmt.Errorf("closing initial object writer: %w", err)
 			return
 		}
+		w.gen = gw.Attrs().Generation
 		w.created = true
 		w.buf.Reset()
 		return
@@ -155,7 +158,7 @@ func (w *Writer) flushLocked() {
 	tmpname := fmt.Sprintf("%s-temp-%d", w.object, now().UnixNano())
 	tmpobj := bh.Object(tmpname)
 
-	gw := tmpobj.NewWriter(w.ctx)
+	gw := tmpobj.If(storage.Conditions{DoesNotExist: true}).NewWriter(w.ctx)
 	if _, err := gw.Write(data); err != nil {
 		w.err = fmt.Errorf("writing temp object: %w", err)
 		return
@@ -166,7 +169,7 @@ func (w *Writer) flushLocked() {
 	}
 
 	// Check component count before composing.
-	attrs, err := objh.Attrs(w.ctx)
+	attrs, err := objh.If(storage.Conditions{GenerationMatch: w.gen}).Attrs(w.ctx)
 	if err != nil {
 		w.err = fmt.Errorf("getting object attrs: %w", err)
 		return
@@ -180,10 +183,13 @@ func (w *Writer) flushLocked() {
 	}
 
 	// Compose existing + temp into existing.
-	if _, err := objh.ComposerFrom(objh, tmpobj).Run(w.ctx); err != nil {
+	composer := objh.If(storage.Conditions{GenerationMatch: w.gen}).ComposerFrom(objh, tmpobj)
+	composeAttrs, err := composer.Run(w.ctx)
+	if err != nil {
 		w.err = fmt.Errorf("composing objects: %w", err)
 		return
 	}
+	w.gen = composeAttrs.Generation
 
 	// Delete temp object.
 	if err := tmpobj.Delete(w.ctx); err != nil {
@@ -200,14 +206,18 @@ func (w *Writer) resetComponentCount(bh *storage.BucketHandle, objh *storage.Obj
 	tmpobj := bh.Object(tmpname)
 
 	// Copy to temp (this creates a non-composite object).
-	if _, err := tmpobj.CopierFrom(objh).Run(w.ctx); err != nil {
+	// Source precondition ensures we're copying the generation we expect.
+	if _, err := tmpobj.CopierFrom(objh.If(storage.Conditions{GenerationMatch: w.gen})).Run(w.ctx); err != nil {
 		return fmt.Errorf("copying to reset temp: %w", err)
 	}
 
 	// Copy back.
-	if _, err := objh.CopierFrom(tmpobj).Run(w.ctx); err != nil {
+	// Destination precondition ensures nobody modified the main object while we copied.
+	attrs, err := objh.If(storage.Conditions{GenerationMatch: w.gen}).CopierFrom(tmpobj).Run(w.ctx)
+	if err != nil {
 		return fmt.Errorf("copying back from reset temp: %w", err)
 	}
+	w.gen = attrs.Generation
 
 	// Delete the reset temp.
 	if err := tmpobj.Delete(w.ctx); err != nil {
@@ -244,7 +254,7 @@ func (w *Writer) Close() error {
 	// Set done metadata. Only if the object was created.
 	if w.created {
 		objh := w.client.Bucket(w.bucket).Object(w.object)
-		if _, err := objh.Update(w.ctx, storage.ObjectAttrsToUpdate{
+		if _, err := objh.If(storage.Conditions{GenerationMatch: w.gen}).Update(w.ctx, storage.ObjectAttrsToUpdate{
 			Metadata: map[string]string{"done": "true"},
 		}); err != nil {
 			return fmt.Errorf("setting done metadata: %w", err)

writer_test.go

@@ -208,6 +208,33 @@ func TestWriterCloseNoData(t *testing.T) {
 	}
 }
 
+func TestWriterExternalWriteDetected(t *testing.T) {
+	client := setup(t)
+	ctx := context.Background()
+
+	// External process creates the object first.
+	extw := client.Bucket("test").Object("obj").NewWriter(ctx)
+	extw.Write([]byte("external"))
+	if err := extw.Close(); err != nil {
+		t.Fatalf("external write: %v", err)
+	}
+
+	// Our writer tries to create the same object — DoesNotExist should fail.
+	w := NewWriter(ctx, WriterOptions{
+		Client:        client,
+		Bucket:        "test",
+		Object:        "obj",
+		FlushInterval: time.Hour,
+	})
+
+	w.Write([]byte("hello"))
+	err := w.Flush()
+	if err == nil {
+		t.Fatal("expected error from flush when object already exists, got nil")
+	}
+	t.Logf("got expected error: %v", err)
+}
+
 func TestWriterComponentCountReset(t *testing.T) {
 	client := setup(t)
 	ctx := context.Background()