fix(16.2): WR-03 never let concurrent media uploads overwrite each other

uniqueMediaKey checked Exists and the write followed without a precondition, so two uploads of the same name (clipboard images are all image.png) could both pick the free key and the second replaced the first. StoreMedia now reserves the chosen key in-process while it writes, and writes with IfNotExist; a key another process stored after the check moves the upload on to the next free name.
2026-10-10 16:35:40 +02:00
parent 5af746f31f
commit 2c961b8408
2 changed files with 179 additions and 13 deletions
--- a/classes/media_store.go
+++ b/classes/media_store.go
@@ -9,6 +9,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"

 	"git.golem15.com/golem15/summercms/modules/lagoon/attach"
 	"gocloud.dev/blob"
@@ -25,6 +26,27 @@ var MediaFolderPattern = regexp.MustCompile(`^[A-Za-z0-9_\-/]*$`)
 // mediaMaxSuffix is how many numbered names uniqueMediaKey tries.
 const mediaMaxSuffix = 1000

+// mediaWriteAttempts is how many free names StoreMedia tries to write when
+// another writer takes the chosen one first.
+const mediaWriteAttempts = 10
+
+// mediaReservations are the media keys a StoreMedia call in this process
+// has chosen and is writing, per bucket, so two concurrent uploads of the
+// same name never pick the same free key.
+var mediaReservations = struct {
+	sync.Mutex
+	keys map[mediaReservation]bool
+}{keys: map[mediaReservation]bool{}}
+
+type mediaReservation struct {
+	bucket *blob.Bucket
+	key    string
+}
+
+// mediaBeforeWrite, when set, runs after a key is chosen and before it is
+// written. Tests use it to stand in for a writer in another process.
+var mediaBeforeWrite func(ctx context.Context, bucket *blob.Bucket, key string)
+
 // MediaError is an upload StoreMedia refuses. Field is the request field the
 // message belongs to ("file" or "folder") and Message is the validation
 // message of the public upload route.
@@ -55,7 +77,10 @@ func validMediaFolder(folder string) bool {
 // file name's base is kept; its extension must be jpg, jpeg, png, gif or
 // webp, and the body may hold at most MediaMaxBytes. A refused upload is a
 // *MediaError with the public route's message; any other error (no bucket,
-// a read or write failure) is not.
+// a read or write failure) is not. It never replaces an existing object:
+// concurrent calls in this process reserve the key they chose, and the
+// write is conditional (IfNotExist), so a key another process stored after
+// the existence check moves the upload to the next free name.
 func StoreMedia(ctx context.Context, bucket *blob.Bucket, folder, filename string, body io.Reader) (StoredMedia, error) {
 	if !validMediaFolder(folder) {
 		return StoredMedia{}, &MediaError{Field: "folder", Message: "The folder format is invalid."}
@@ -77,23 +102,68 @@ func StoreMedia(ctx context.Context, bucket *blob.Bucket, folder, filename strin
 	if len(data) > MediaMaxBytes {
 		return StoredMedia{}, &MediaError{Field: "file", Message: "The file may not be greater than 10240 kilobytes."}
 	}
-	keyPath, err = uniqueMediaKey(ctx, bucket, keyPath)
-	if err != nil {
-		return StoredMedia{}, err
+	for attempt := 0; attempt < mediaWriteAttempts; attempt++ {
+		chosen, release, err := reserveMediaKey(ctx, bucket, keyPath)
+		if err != nil {
+			return StoredMedia{}, err
+		}
+		key := strings.TrimPrefix(chosen, "/")
+		if mediaBeforeWrite != nil {
+			mediaBeforeWrite(ctx, bucket, key)
+		}
+		err = writeNewMedia(ctx, bucket, key, data)
+		release()
+		if gcerrors.Code(err) == gcerrors.FailedPrecondition {
+			// Another writer (another process) stored this key after the
+			// existence check; take the next free name.
+			continue
+		}
+		if err != nil {
+			return StoredMedia{}, err
+		}
+		return StoredMedia{URL: attach.PublicURL(key), Path: chosen}, nil
 	}
-	key := strings.TrimPrefix(keyPath, "/")
-	w, err := bucket.NewWriter(ctx, key, nil)
+	return StoredMedia{}, fmt.Errorf("journal: no free media name for %s after %d attempts", keyPath, mediaWriteAttempts)
+}
+
+// reserveMediaKey chooses a free key for pathName (uniqueMediaKey, skipping
+// keys this process is writing) and reserves it until release is called.
+func reserveMediaKey(ctx context.Context, bucket *blob.Bucket, pathName string) (string, func(), error) {
+	mediaReservations.Lock()
+	defer mediaReservations.Unlock()
+	reserved := func(p string) bool {
+		return mediaReservations.keys[mediaReservation{bucket: bucket, key: p}]
+	}
+	chosen, err := uniqueMediaKey(ctx, bucket, pathName, reserved)
 	if err != nil {
-		return StoredMedia{}, fmt.Errorf("journal: media writer: %w", err)
+		return "", nil, err
+	}
+	slot := mediaReservation{bucket: bucket, key: chosen}
+	mediaReservations.keys[slot] = true
+	release := func() {
+		mediaReservations.Lock()
+		delete(mediaReservations.keys, slot)
+		mediaReservations.Unlock()
+	}
+	return chosen, release, nil
+}
+
+// writeNewMedia writes data under key only when no object exists there
+// (blob.WriterOptions.IfNotExist): an existing object is never replaced,
+// and the error then has the gcerrors.FailedPrecondition code.
+func writeNewMedia(ctx context.Context, bucket *blob.Bucket, key string, data []byte) error {
+	w, err := bucket.NewWriter(ctx, key, &blob.WriterOptions{IfNotExist: true})
+	if err != nil {
+		return fmt.Errorf("journal: media writer: %w", err)
 	}
 	if _, err := w.Write(data); err != nil {
 		_ = w.Close()
-		return StoredMedia{}, fmt.Errorf("journal: write media: %w", err)
+		return fmt.Errorf("journal: write media: %w", err)
 	}
 	if err := w.Close(); err != nil {
-		return StoredMedia{}, fmt.Errorf("journal: close media: %w", err)
+		return fmt.Errorf("journal: close media: %w", err)
 	}
-	return StoredMedia{URL: attach.PublicURL(key), Path: keyPath}, nil
+	return nil
 }

 // mediaObjectPath is /journal/<folder>/<fileName> with empty, "." and ".."
@@ -117,10 +187,14 @@ func mediaObjectPath(folder, fileName string) string {
 }

 // uniqueMediaKey returns pathName, or pathName with "-1", "-2", ... before
-// its extension when an object already exists there. It never returns the
-// path of an existing object: when every numbered name is taken it fails.
-func uniqueMediaKey(ctx context.Context, bucket *blob.Bucket, pathName string) (string, error) {
+// its extension when an object already exists there or reserved reports the
+// path as taken. It never returns the path of an existing object: when every
+// numbered name is taken it fails.
+func uniqueMediaKey(ctx context.Context, bucket *blob.Bucket, pathName string, reserved func(string) bool) (string, error) {
 	exists := func(p string) (bool, error) {
+		if reserved != nil && reserved(p) {
+			return true, nil
+		}
 		ok, err := bucket.Exists(ctx, strings.TrimPrefix(p, "/"))
 		if err != nil && gcerrors.Code(err) != gcerrors.NotFound {
 			return false, err
--- a/classes/media_store_test.go
+++ b/classes/media_store_test.go
@@ -4,11 +4,14 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"

 	"git.golem15.com/golem15/summercms/modules/lagoon/attach"
+	"gocloud.dev/blob"
 	"gocloud.dev/blob/memblob"
 )

@@ -101,6 +104,95 @@ func TestStoreMedia(t *testing.T) {
 		}
 	})

+	// WR-03: concurrent uploads of the same name each get their own key and
+	// keep their own bytes.
+	t.Run("concurrent uploads of one name never overwrite each other", func(t *testing.T) {
+		race := memblob.OpenBucket(nil)
+		t.Cleanup(func() { _ = race.Close() })
+		const uploads = 16
+		var wg sync.WaitGroup
+		results := make([]StoredMedia, uploads)
+		errs := make([]error, uploads)
+		bodies := make([][]byte, uploads)
+		for i := range uploads {
+			bodies[i] = append(append([]byte{}, testPNG...), []byte(fmt.Sprintf("upload-%02d", i))...)
+		}
+		start := make(chan struct{})
+		for i := range uploads {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				<-start
+				results[i], errs[i] = StoreMedia(ctx, race, "editor", "image.png", bytes.NewReader(bodies[i]))
+			}()
+		}
+		close(start)
+		wg.Wait()
+		seen := map[string]int{}
+		for i := range uploads {
+			if errs[i] != nil {
+				t.Fatalf("upload %d: %v", i, errs[i])
+			}
+			if prev, dup := seen[results[i].Path]; dup {
+				t.Fatalf("uploads %d and %d both stored %s", prev, i, results[i].Path)
+			}
+			seen[results[i].Path] = i
+			data, err := race.ReadAll(ctx, strings.TrimPrefix(results[i].Path, "/"))
+			if err != nil || !bytes.Equal(data, bodies[i]) {
+				t.Fatalf("upload %d at %s holds another upload's bytes (err %v)", i, results[i].Path, err)
+			}
+		}
+		if len(mediaReservations.keys) != 0 {
+			t.Fatalf("reservations left behind: %v", mediaReservations.keys)
+		}
+	})
+
+	t.Run("a key another process stores after the check moves the upload on", func(t *testing.T) {
+		other := memblob.OpenBucket(nil)
+		t.Cleanup(func() { _ = other.Close() })
+		planted := map[string]bool{}
+		mediaBeforeWrite = func(ctx context.Context, b *blob.Bucket, key string) {
+			if len(planted) < 2 {
+				planted[key] = true
+				if err := b.WriteAll(ctx, key, []byte("other process"), nil); err != nil {
+					t.Error(err)
+				}
+			}
+		}
+		t.Cleanup(func() { mediaBeforeWrite = nil })
+		got, err := StoreMedia(ctx, other, "editor", "image.png", bytes.NewReader(testPNG))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got.Path != "/journal/editor/image-2.png" {
+			t.Fatalf("path = %s, want the first name nobody else wrote", got.Path)
+		}
+		for key := range planted {
+			if data, _ := other.ReadAll(ctx, key); string(data) != "other process" {
+				t.Fatalf("%s was overwritten", key)
+			}
+		}
+		if data, _ := other.ReadAll(ctx, "journal/editor/image-2.png"); !bytes.Equal(data, testPNG) {
+			t.Fatal("the upload's bytes are not at its path")
+		}
+	})
+
+	t.Run("a writer that always loses gives up with an opaque error", func(t *testing.T) {
+		busy := memblob.OpenBucket(nil)
+		t.Cleanup(func() { _ = busy.Close() })
+		mediaBeforeWrite = func(ctx context.Context, b *blob.Bucket, key string) {
+			_ = b.WriteAll(ctx, key, []byte("other process"), nil)
+		}
+		t.Cleanup(func() { mediaBeforeWrite = nil })
+		_, err := StoreMedia(ctx, busy, "editor", "image.png", bytes.NewReader(testPNG))
+		if err == nil || errors.As(err, new(*MediaError)) {
+			t.Fatalf("error = %v, want an opaque error", err)
+		}
+		if len(mediaReservations.keys) != 0 {
+			t.Fatalf("reservations left behind: %v", mediaReservations.keys)
+		}
+	})
+
 	t.Run("a name with every suffix taken fails instead of overwriting", func(t *testing.T) {
 		full := memblob.OpenBucket(nil)
 		t.Cleanup(func() { _ = full.Close() })