Repository: blevesearch/bleve
Pull request #2185 (2025-04-24 05:37:41)

[BP] MB-66163: always protect the latest snapshot within bolt
Backporting #2183 for couchbase 7.6.x release cycle builds.


Patch:
diff --git a/index/scorch/event.go b/index/scorch/event.go
index e0bc3b60..b17c7f41 100644
--- a/index/scorch/event.go
+++ b/index/scorch/event.go
@@ -71,3 +71,7 @@ var EventKindPreMergeCheck = EventKind(9)
 // EventKindIndexStart is fired when Index() is invoked which
 // creates a new Document object from an interface using the index mapping.
 var EventKindIndexStart = EventKind(10)
+
+// EventKindPurgerCheck is fired before the purge code is invoked and decides
+// whether to execute or not. For unit test purposes
+var EventKindPurgerCheck = EventKind(11)
diff --git a/index/scorch/merge.go b/index/scorch/merge.go
index b74504ca..6db81cd4 100644
--- a/index/scorch/merge.go
+++ b/index/scorch/merge.go
@@ -80,6 +80,10 @@ OUTER:
 					// Retry instead of blocking/waiting here since a long wait
 					// can result in more segments introduced i.e. s.root will
 					// be updated.
+
+					// decrement the ref count since its no longer needed in this
+					// iteration
+					_ = ourSnapshot.DecRef()
 					continue OUTER
 				}
 
@@ -468,6 +472,11 @@ func (s *Scorch) mergeSegmentBases(snapshot *IndexSnapshot,
 	filename := zapFileName(newSegmentID)
 	path := s.path + string(os.PathSeparator) + filename
 
+	// to prevent accidental cleanup of this newly created file, mark it
+	// as ineligible for removal. this will be flipped back when the bolt
+	// is updated - which is valid, since the snapshot updated in bolt is
+	// cleaned up only if its zero ref'd (MB-66163 for more details)
+	s.markIneligibleForRemoval(filename)
 	newDocNums, _, err :=
 		s.segPlugin.Merge(sbs, sbsDrops, path, s.closeCh, s)
 
diff --git a/index/scorch/persister.go b/index/scorch/persister.go
index d59f733d..05e7254c 100644
--- a/index/scorch/persister.go
+++ b/index/scorch/persister.go
@@ -219,7 +219,9 @@ OUTER:
 		case s.introducerNotifier <- w:
 		}
 
-		s.removeOldData() // might as well cleanup while waiting
+		if ok := s.fireEvent(EventKindPurgerCheck, 0); ok {
+			s.removeOldData() // might as well cleanup while waiting
+		}
 
 		atomic.AddUint64(&s.stats.TotPersistLoopWait, 1)
 
@@ -286,7 +288,9 @@ func (s *Scorch) pausePersisterForMergerCatchUp(lastPersistedEpoch uint64,
 	// 1. Too many older snapshots awaiting the clean up.
 	// 2. The merger could be lagging behind on merging the disk files.
 	if numFilesOnDisk > uint64(po.PersisterNapUnderNumFiles) {
-		s.removeOldData()
+		if ok := s.fireEvent(EventKindPurgerCheck, 0); ok {
+			s.removeOldData()
+		}
 		numFilesOnDisk, _, _ = s.diskFileStats(nil)
 	}
 
@@ -994,9 +998,12 @@ func getTimeSeriesSnapshots(maxDataPoints int, interval time.Duration,
 // by a time duration of RollbackSamplingInterval.
 func getProtectedSnapshots(rollbackSamplingInterval time.Duration,
 	numSnapshotsToKeep int,
-	persistedSnapshots []*snapshotMetaData) map[uint64]time.Time {
-
-	lastPoint, protectedEpochs := getTimeSeriesSnapshots(numSnapshotsToKeep,
+	persistedSnapshots []*snapshotMetaData,
+) map[uint64]time.Time {
+	// keep numSnapshotsToKeep - 1 worth of time series snapshots, because we always
+	// must preserve the very latest snapshot in bolt as well to avoid accidental
+	// deletes of bolt entries and cleanups by the purger code.
+	lastPoint, protectedEpochs := getTimeSeriesSnapshots(numSnapshotsToKeep-1,
 		rollbackSamplingInterval, persistedSnapshots)
 	if len(protectedEpochs) < numSnapshotsToKeep {
 		numSnapshotsNeeded := numSnapshotsToKeep - len(protectedEpochs)
@@ -1182,7 +1189,10 @@ type snapshotMetaData struct {
 func (s *Scorch) rootBoltSnapshotMetaData() ([]*snapshotMetaData, error) {
 	var rv []*snapshotMetaData
 	currTime := time.Now()
-	expirationDuration := time.Duration(s.numSnapshotsToKeep) * s.rollbackSamplingInterval
+	// including the very latest snapshot there should be n snapshots, so the
+	// very last one would be tc - (n-1) * d
+	// for eg for n = 3 the checkpoints preserved should be tc, tc - d, tc - 2d
+	expirationDuration := time.Duration(s.numSnapshotsToKeep-1) * s.rollbackSamplingInterval
 
 	err := s.rootBolt.View(func(tx *bolt.Tx) error {
 		snapshots := tx.Bucket(boltSnapshotsBucket)
@@ -1191,6 +1201,7 @@ func (s *Scorch) rootBoltSnapshotMetaData() ([]*snapshotMetaData, error) {
 		}
 		sc := snapshots.Cursor()
 		var found bool
+		// traversal order - latest -> oldest epoch
 		for sk, _ := sc.Last(); sk != nil; sk, _ = sc.Prev() {
 			_, snapshotEpoch, err := decodeUvarintAscending(sk)
 			if err != nil {
@@ -1240,7 +1251,6 @@ func (s *Scorch) rootBoltSnapshotMetaData() ([]*snapshotMetaData, error) {
 					err = nil
 				}
 			}
-
 		}
 		return nil
 	})


Test patch:
diff --git a/index/scorch/rollback_test.go b/index/scorch/rollback_test.go
index 007ac2aa..c52b30c2 100644
--- a/index/scorch/rollback_test.go
+++ b/index/scorch/rollback_test.go
@@ -15,6 +15,10 @@
 package scorch
 
 import (
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
@@ -283,15 +287,14 @@ func TestGetProtectedSnapshots(t *testing.T) {
 				{epoch: 99, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 12))},
 				{epoch: 88, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 6))},
 				{epoch: 50, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval))},
-				{epoch: 35, timeStamp: currentTimeStamp.Add(-(6 * RollbackSamplingInterval / 5))},
-				{epoch: 10, timeStamp: currentTimeStamp.Add(-(2 * RollbackSamplingInterval))},
 			},
 			numSnapshotsToKeep: 2,
 			expCount:           2,
-			expEpochs:          []uint64{50, 10},
+			expEpochs:          []uint64{100, 50},
 		},
 		{
-			title: "epochs that have timestamps approximated to the expected value",
+			title: "epochs that have timestamps approximated to the expected value, " +
+				"always retain the latest one",
 			metaData: []*snapshotMetaData{
 				{epoch: 100, timeStamp: currentTimeStamp},
 				{epoch: 99, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 12))},
@@ -302,7 +305,7 @@ func TestGetProtectedSnapshots(t *testing.T) {
 			},
 			numSnapshotsToKeep: 3,
 			expCount:           3,
-			expEpochs:          []uint64{50, 35, 10},
+			expEpochs:          []uint64{100, 35, 10},
 		},
 		{
 			title: "protecting epochs when we don't have enough snapshots with RollbackSamplingInterval" +
@@ -351,3 +354,199 @@ func TestGetProtectedSnapshots(t *testing.T) {
 		}
 	}
 }
+
+func indexDummyData(t *testing.T, scorchi *Scorch, i int) {
+	// create a batch, insert 2 new documents
+	batch := index.NewBatch()
+	doc := document.NewDocument(fmt.Sprintf("%d", i))
+	doc.AddField(document.NewTextField("name", []uint64{}, []byte("test1")))
+	batch.Update(doc)
+	doc = document.NewDocument(fmt.Sprintf("%d", i+1))
+	doc.AddField(document.NewTextField("name", []uint64{}, []byte("test2")))
+	batch.Update(doc)
+
+	err := scorchi.Batch(batch)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+type testFSDirector string
+
+func (f testFSDirector) GetWriter(filePath string) (io.WriteCloser,
+	error) {
+	dir, file := filepath.Split(filePath)
+	if dir != "" {
+		err := os.MkdirAll(filepath.Join(string(f), dir), os.ModePerm)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return os.OpenFile(filepath.Join(string(f), dir, file),
+		os.O_RDWR|os.O_CREATE, 0600)
+}
+
+func TestLatestSnapshotProtected(t *testing.T) {
+	cfg := CreateConfig("TestLatestSnapshotProtected")
+	numSnapshotsToKeepOrig := NumSnapshotsToKeep
+	NumSnapshotsToKeep = 3
+	rollbackSamplingIntervalOrig := RollbackSamplingInterval
+	RollbackSamplingInterval = 10 * time.Second
+
+	err := InitTest(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		NumSnapshotsToKeep = numSnapshotsToKeepOrig
+		RollbackSamplingInterval = rollbackSamplingIntervalOrig
+		err := DestroyTest(cfg)
+		if err != nil {
+			t.Log(err)
+		}
+	}()
+
+	// disable merger and purger
+	RegistryEventCallbacks["test"] = func(e Event) bool {
+		if e.Kind == EventKindPreMergeCheck || e.Kind == EventKindPurgerCheck {
+			return false
+		}
+		return true
+	}
+	cfg["eventCallbackName"] = "test"
+	analysisQueue := index.NewAnalysisQueue(1)
+	idx, err := NewScorch(Name, cfg, analysisQueue)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	scorchi, ok := idx.(*Scorch)
+	if !ok {
+		t.Fatalf("Not a scorch index?")
+	}
+
+	err = scorchi.Open()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// replicate the following scenario of persistence of snapshots
+	// tc, tc - d/12, tc - d/6, tc - 3d/4, tc - 5d/6, tc - 6d/5
+	// approximate timestamps where there's a chance that the latest snapshot
+	// might not fit into the time-series
+	indexDummyData(t, scorchi, 1)
+	persistedSnapshots, err := scorchi.rootBoltSnapshotMetaData()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(persistedSnapshots) != 1 {
+		t.Fatalf("expected 1 persisted snapshot, got %d", len(persistedSnapshots))
+	}
+	time.Sleep(4 * RollbackSamplingInterval / 5)
+	indexDummyData(t, scorchi, 3)
+	time.Sleep(9 * RollbackSamplingInterval / 20)
+	indexDummyData(t, scorchi, 5)
+	time.Sleep(7 * RollbackSamplingInterval / 12)
+	indexDummyData(t, scorchi, 7)
+	time.Sleep(1 * RollbackSamplingInterval / 12)
+	indexDummyData(t, scorchi, 9)
+
+	persistedSnapshots, err = scorchi.rootBoltSnapshotMetaData()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	protectedSnapshots := getProtectedSnapshots(RollbackSamplingInterval, NumSnapshotsToKeep, persistedSnapshots)
+	if len(protectedSnapshots) != 3 {
+		t.Fatalf("expected %d protected snapshots, got %d", NumSnapshotsToKeep, len(protectedSnapshots))
+	}
+	if _, ok := protectedSnapshots[persistedSnapshots[0].epoch]; !ok {
+		t.Fatalf("expected %d to be protected, but not found", persistedSnapshots[0].epoch)
+	}
+}
+
+func TestBackupRacingWithPurge(t *testing.T) {
+	cfg := CreateConfig("TestBackupRacingWithPurge")
+	numSnapshotsToKeepOrig := NumSnapshotsToKeep
+	NumSnapshotsToKeep = 3
+	rollbackSamplingIntervalOrig := RollbackSamplingInterval
+	RollbackSamplingInterval = 10 * time.Second
+	err := InitTest(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		NumSnapshotsToKeep = numSnapshotsToKeepOrig
+		RollbackSamplingInterval = rollbackSamplingIntervalOrig
+		err := DestroyTest(cfg)
+		if err != nil {
+			t.Log(err)
+		}
+	}()
+
+	// disable merger and purger
+	RegistryEventCallbacks["test"] = func(e Event) bool {
+		if e.Kind == EventKindPreMergeCheck || e.Kind == EventKindPurgerCheck {
+			return false
+		}
+		return true
+	}
+	cfg["eventCallbackName"] = "test"
+	analysisQueue := index.NewAnalysisQueue(1)
+	idx, err := NewScorch(Name, cfg, analysisQueue)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer idx.Close()
+
+	scorchi, ok := idx.(*Scorch)
+	if !ok {
+		t.Fatalf("Not a scorch index?")
+	}
+
+	err = scorchi.Open()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// replicate the following scenario of persistence of snapshots
+	// tc, tc - d/12, tc - d/6, tc - 3d/4, tc - 5d/6, tc - 6d/5, tc - 2d
+	// approximate timestamps where there's a chance that the latest snapshot
+	// might not fit into the time-series
+	indexDummyData(t, scorchi, 1)
+	time.Sleep(4 * RollbackSamplingInterval / 5)
+	indexDummyData(t, scorchi, 3)
+	time.Sleep(9 * RollbackSamplingInterval / 20)
+	indexDummyData(t, scorchi, 5)
+	time.Sleep(7 * RollbackSamplingInterval / 12)
+	indexDummyData(t, scorchi, 7)
+	time.Sleep(1 * RollbackSamplingInterval / 12)
+	indexDummyData(t, scorchi, 9)
+
+	// now if the purge code is invoked, there's a possiblity of the latest snapshot
+	// being removed from bolt and the corresponding file segment getting cleaned up.
+	scorchi.removeOldData()
+
+	copyReader := scorchi.CopyReader()
+	defer func() { copyReader.CloseCopyReader() }()
+
+	backupidxConfig := CreateConfig("backup-directory")
+	err = InitTest(backupidxConfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		err := DestroyTest(backupidxConfig)
+		if err != nil {
+			t.Log(err)
+		}
+	}()
+
+	// if the latest snapshot was purged, the following will return error
+	err = copyReader.CopyTo(testFSDirector(backupidxConfig["path"].(string)))
+	if err != nil {
+		t.Fatalf("error copying the index: %v", err)
+	}
+}