From 2c961b8408f7c2448e22530e222bc019f0c2fbe6 Mon Sep 17 00:00:00 2001 From: Jakub Zych Date: Sat, 10 Oct 2026 16:35:40 +0200 Subject: [PATCH] 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. --- classes/media_store.go | 100 +++++++++++++++++++++++++++++++----- classes/media_store_test.go | 92 +++++++++++++++++++++++++++++++++ 2 files changed, 179 insertions(+), 13 deletions(-) diff --git a/classes/media_store.go b/classes/media_store.go index 1fe37f0..4572d65 100644 --- 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// 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 diff --git a/classes/media_store_test.go b/classes/media_store_test.go index 0c23545..30771bc 100644 --- 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() })