Repository: cockroachdb/pebble
Pull request #3765 (2024-07-17 14:41:28)

crl-release-23.2: objstorage: fix race in vfsSync
#### crl-release-23.2: objstorage: improve ParallelSync test to expose race in vfsSync

We improve this test by using a strict MemFS, simulating a crash at
a random point, and checking that all objects that were synced are
accessible.

#### crl-release-23.2: objstorage: fix race in vfsSync

We have a race in `vfsSync`: if two goroutines Sync at the same time,
one might see the flag=false and return immediately, before the other
goroutine actually runs/completes the Sync.

The fix is to store a change counter and wait for any in-progress
syncs as necessary.

Informs https://github.com/cockroachdb/cockroach/issues/124845

cockroach-teamcity: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/cockroachdb/pebble/3765)
<!-- Reviewable:end -->

RaduBerinde: TFTR!

Patch:
diff --git a/objstorage/objstorageprovider/provider.go b/objstorage/objstorageprovider/provider.go
index ea9d08bb..8d5b39c8 100644
--- a/objstorage/objstorageprovider/provider.go
+++ b/objstorage/objstorageprovider/provider.go
@@ -45,9 +45,14 @@ type provider struct {
 			storageObjects map[remote.Locator]remote.Storage
 		}
 
-		// localObjectsChanged is set if non-remote objects were created or deleted
-		// but Sync was not yet called.
-		localObjectsChanged bool
+		// TODO(radu): move these fields to a localLockedState struct.
+		// localObjectsChanged is incremented whenever non-remote objects are created.
+		// The purpose of this counter is to avoid syncing the local filesystem when
+		// only remote objects are changed.
+		localObjectsChangeCounter uint64
+		// localObjectsChangeCounterSynced is the value of localObjectsChangeCounter
+		// value at the time the last completed sync was launched.
+		localObjectsChangeCounterSynced uint64
 
 		// knownObjects maintains information about objects that are known to the provider.
 		// It is initialized with the list of files in the manifest when we open a DB.
@@ -550,7 +555,7 @@ func (p *provider) addMetadata(meta objstorage.ObjectMetadata) {
 			CleanupMethod:  meta.Remote.CleanupMethod,
 		})
 	} else {
-		p.mu.localObjectsChanged = true
+		p.mu.localObjectsChangeCounter++
 	}
 }
 
@@ -566,7 +571,7 @@ func (p *provider) removeMetadata(fileNum base.DiskFileNum) {
 	if meta.IsRemote() {
 		p.mu.remote.catalogBatch.DeleteObject(fileNum)
 	} else {
-		p.mu.localObjectsChanged = true
+		p.mu.localObjectsChangeCounter++
 	}
 }
 
diff --git a/objstorage/objstorageprovider/vfs.go b/objstorage/objstorageprovider/vfs.go
index 7f73a237..ae34da63 100644
--- a/objstorage/objstorageprovider/vfs.go
+++ b/objstorage/objstorageprovider/vfs.go
@@ -83,19 +83,23 @@ func (p *provider) vfsInit() error {
 
 func (p *provider) vfsSync() error {
 	p.mu.Lock()
-	shouldSync := p.mu.localObjectsChanged
-	p.mu.localObjectsChanged = false
+	counterVal := p.mu.localObjectsChangeCounter
+	lastSynced := p.mu.localObjectsChangeCounterSynced
 	p.mu.Unlock()
 
-	if !shouldSync {
+	if lastSynced >= counterVal {
 		return nil
 	}
 	if err := p.fsDir.Sync(); err != nil {
-		p.mu.Lock()
-		defer p.mu.Unlock()
-		p.mu.localObjectsChanged = true
 		return err
 	}
+
+	p.mu.Lock()
+	if p.mu.localObjectsChangeCounterSynced < counterVal {
+		p.mu.localObjectsChangeCounterSynced = counterVal
+	}
+	p.mu.Unlock()
+
 	return nil
 }
 


Test patch:
diff --git a/objstorage/objstorageprovider/provider_test.go b/objstorage/objstorageprovider/provider_test.go
index a2e5f47a..0f60eae2 100644
--- a/objstorage/objstorageprovider/provider_test.go
+++ b/objstorage/objstorageprovider/provider_test.go
@@ -10,7 +10,9 @@ import (
 	"math/rand"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -575,7 +577,8 @@ func TestParallelSync(t *testing.T) {
 			name = "shared"
 		}
 		t.Run(name, func(t *testing.T) {
-			st := DefaultSettings(vfs.NewMem(), "")
+			fs := vfs.NewStrictMem()
+			st := DefaultSettings(fs, "")
 			st.Remote.StorageFactory = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
 				"": remote.NewInMem(),
 			})
@@ -586,15 +589,35 @@ func TestParallelSync(t *testing.T) {
 			require.NoError(t, err)
 			require.NoError(t, p.SetCreatorID(1))
 
-			const numGoroutines = 4
+			const numGoroutines = 32
 			const numOps = 100
 			var wg sync.WaitGroup
+			wg.Add(numGoroutines + 1)
+
+			var mustExistMu struct {
+				sync.Mutex
+				m map[base.DiskFileNum]struct{}
+			}
+			mustExistMu.m = make(map[base.DiskFileNum]struct{})
+			setMustExist := func(num base.DiskFileNum, val bool) {
+				mustExistMu.Lock()
+				defer mustExistMu.Unlock()
+				if val {
+					mustExistMu.m[num] = struct{}{}
+				} else {
+					delete(mustExistMu.m, num)
+				}
+			}
+
+			var stop atomic.Bool
 			for n := 0; n < numGoroutines; n++ {
-				wg.Add(1)
 				go func(startNum int, shared bool) {
 					defer wg.Done()
 					rng := rand.New(rand.NewSource(int64(startNum)))
 					for i := 0; i < numOps; i++ {
+						if stop.Load() {
+							return
+						}
 						num := base.FileNum(startNum + i).DiskFileNum()
 						w, _, err := p.Create(context.Background(), base.FileTypeTable, num, objstorage.CreateOptions{
 							PreferSharedStorage: shared,
@@ -606,22 +629,68 @@ func TestParallelSync(t *testing.T) {
 							panic(err)
 						}
 						if rng.Intn(2) == 0 {
+							if stop.Load() {
+								return
+							}
 							if err := p.Sync(); err != nil {
 								panic(err)
 							}
+							setMustExist(num, true)
 						}
-						if err := p.Remove(base.FileTypeTable, num); err != nil {
-							panic(err)
-						}
-						if rng.Intn(2) == 0 {
-							if err := p.Sync(); err != nil {
+
+						if rng.Intn(4) == 0 {
+							setMustExist(num, false)
+							if err := p.Remove(base.FileTypeTable, num); err != nil {
 								panic(err)
 							}
+							if rng.Intn(2) == 0 {
+								if err := p.Sync(); err != nil {
+									panic(err)
+								}
+							}
 						}
 					}
 				}(numOps*n, shared)
 			}
+			mustExist := make(map[base.DiskFileNum]struct{})
+			// "Crash" at a random time.
+			time.AfterFunc(time.Duration(rand.Int63n(int64(10*time.Millisecond))), func() {
+				defer wg.Done()
+				if shared {
+					// TODO(radu): we cannot simulate a crash in shared mode because we
+					// have no way to restore the remote.Storage to the state at the time
+					// of the crash.
+					return
+				}
+				// Grab a consistent snapshot of the current mustExist map.
+				mustExistMu.Lock()
+				for n := range mustExistMu.m {
+					mustExist[n] = struct{}{}
+				}
+				fs.SetIgnoreSyncs(true)
+				mustExistMu.Unlock()
+				stop.Store(true)
+			})
+			// Wait until the timer function above and all the goroutines finish.
 			wg.Wait()
+			// Now close the provider, reset the filesystem, and check that all files
+			// we expect to exist are there.
+			require.NoError(t, p.Close())
+			fs.ResetToSyncedState()
+
+			p, err = Open(st)
+			require.NoError(t, err)
+			// Check that all objects exist and can be opened.
+			for num := range mustExist {
+				if _, err := p.Lookup(base.FileTypeTable, num); err != nil {
+					t.Fatalf("object %s not present after crash", num)
+				}
+				r, err := p.OpenForReading(context.Background(), base.FileTypeTable, num, objstorage.OpenOptions{})
+				if err != nil {
+					t.Fatalf("object %s cannot be opened after crash: %s", num, err)
+				}
+				require.NoError(t, r.Close())
+			}
 		})
 	}
 }