MB-60659 implement cbgt rollback hook

context: https://review.couchbase.org/c/cbgt/+/205499

Aim is to make rollback phases atomic and serialize them as a whole.
This PR register the cbft specific rollback steps (rolling back
the scorch) to cbgt as a hook.

Change-Id: Id7559b4708b911414e3666bc513393f566f1e5b0
Reviewed-on: https://review.couchbase.org/c/cbft/+/205500
Tested-by: Mohd Shaad Khan <[email protected]>
Well-Formed: Restriction Checker
Reviewed-by: Aditi Ahuja <[email protected]>
Well-Formed: Build Bot <[email protected]>

Author: moshaad7

pindex_bleve.go

@@ -306,25 +306,33 @@ func (sr *SearchRequest) ConvertToBleveSearchRequest() (*bleve.SearchRequest, er
 	return r, r.Validate()
 }
 
+type RollbackInfo struct {
+	partition   string
+	vBucketUUID uint64
+	rollbackSeq uint64
+}
+
 type BleveDest struct {
 	path       string
 	sourceName string
 
 	bleveDocConfig BleveDocumentConfig
 
 	// Invoked when mgr should rollback this BleveDest
-	rollback func()
+	rollback     func()
+	rollbackInfo *RollbackInfo
 
 	stats     *cbgt.PIndexStoreStats
 	copyStats *CopyPartitionStats
 
-	m           sync.RWMutex // Protects the fields that follow.
-	bindex      bleve.Index
-	partitions  map[string]*BleveDestPartition
-	rev         uint64 // Incremented whenever bindex changes.
-	batchReqChs []chan *batchRequest
-	stopCh      chan struct{}
-	removeCh    chan struct{}
+	m                    sync.RWMutex // Protects the fields that follow.
+	bindex               bleve.Index
+	partitions           map[string]*BleveDestPartition
+	rev                  uint64 // Incremented whenever bindex changes.
+	batchReqChs          []chan *batchRequest
+	stopCh               chan struct{}
+	removeCh             chan struct{}
+	isRollbackInProgress bool
 }
 
 // Used to track state for a single partition.

pindex_bleve_rollback.go

@@ -15,9 +15,63 @@ import (
 
 	"github.com/blevesearch/bleve/v2/index/scorch"
 
+	"github.com/couchbase/cbgt"
 	log "github.com/couchbase/clog"
 )
 
+func init() {
+	cbgt.RollbackHook = func(phase cbgt.RollbackPhase, pindex *cbgt.PIndex) (err error) {
+		df, ok := pindex.Dest.(*cbgt.DestForwarder)
+		if !ok || df == nil {
+			return fmt.Errorf("rollbackhook: invalid dest for pindex:%s",
+				pindex.Name)
+		}
+		bd, ok := df.DestProvider.(*BleveDest)
+		if !ok || bd == nil {
+			return fmt.Errorf("rollbackhook: invalid destProvider for pindex:%s",
+				pindex.Name)
+		}
+
+		bd.m.Lock()
+		defer bd.m.Unlock()
+
+		switch phase {
+		case cbgt.RollbackInit:
+
+			pindexName := bd.bindex.Name()
+			wasClosed, wasPartial, err := bd.partialRollbackLOCKED()
+
+			log.Printf("pindex_bleve_rollback: path: %s, wasClosed: %t, "+
+				"wasPartial: %t, err: %v", bd.path, wasClosed, wasPartial, err)
+
+			if !wasClosed {
+				bd.closeLOCKED(false)
+			}
+
+			if !wasPartial {
+				atomic.AddUint64(&TotRollbackFull, 1)
+				if ServerlessMode {
+					// this is a full rollback, so the paritition is going to be
+					// rebuilt a fresh. The reason we are refunding over here is
+					// because this is not a end-user problem, but rather a
+					// couchbase cluster problem. So, once the partition is built
+					// afresh, we would essentially any loss of cost by charging
+					// for 0 - original high seq no. and after that we will
+					// actually start costing the user.
+					RollbackRefund(pindexName, bd.sourceName, 0)
+				}
+				os.RemoveAll(bd.path) // Full rollback to zero.
+			} else {
+				atomic.AddUint64(&TotRollbackPartial, 1)
+			}
+		case cbgt.RollbackCompleted:
+			bd.isRollbackInProgress = false
+		}
+
+		return nil
+	}
+}
+
 func (t *BleveDest) Rollback(partition string, vBucketUUID uint64, rollbackSeq uint64) error {
 	t.AddError("dest rollback", partition, nil, rollbackSeq, nil, nil)
 
@@ -28,47 +82,27 @@ func (t *BleveDest) Rollback(partition string, vBucketUUID uint64, rollbackSeq u
 
 	// The BleveDest may be closed due to another partition(BleveDestPartition) of
 	// the same pindex being rolled back earlier.
-	if t.bindex != nil {
-		pindexName := t.bindex.Name()
-		wasClosed, wasPartial, err := t.partialRollbackLOCKED(partition,
-			vBucketUUID, rollbackSeq)
-
-		log.Printf("pindex_bleve_rollback: path: %s,"+
-			" wasClosed: %t, wasPartial: %t, err: %v",
-			t.path, wasClosed, wasPartial, err)
-
-		if !wasClosed {
-			t.closeLOCKED(false)
-		}
+	if t.bindex == nil || t.isRollbackInProgress {
+		return nil
+	}
 
-		if !wasPartial {
-			atomic.AddUint64(&TotRollbackFull, 1)
-			if ServerlessMode {
-				// this is a full rollback, so the paritition is going to be
-				// rebuilt a fresh. The reason we are refunding over here is
-				// because this is not a end-user problem, but rather a
-				// couchbase cluster problem. So, once the partition is built
-				// afresh, we would essentially any loss of cost by charging
-				// for 0 - original high seq no. and after that we will
-				// actually start costing the user.
-				RollbackRefund(pindexName, t.sourceName, 0)
-			}
-			os.RemoveAll(t.path) // Full rollback to zero.
-		} else {
-			atomic.AddUint64(&TotRollbackPartial, 1)
-		}
+	t.isRollbackInProgress = true
 
-		// Whether partial or full rollback, restart the BleveDest so that
-		// feeds are restarted.
-		t.rollback()
+	t.rollbackInfo = &RollbackInfo{
+		partition:   partition,
+		vBucketUUID: vBucketUUID,
+		rollbackSeq: rollbackSeq,
 	}
 
+	// Whether partial or full rollback, restart the BleveDest so that
+	// feeds are restarted.
+	t.rollback()
+
 	return nil
 }
 
 // Attempt partial rollback.
-func (t *BleveDest) partialRollbackLOCKED(partition string,
-	vBucketUUID uint64, rollbackSeq uint64) (bool, bool, error) {
+func (t *BleveDest) partialRollbackLOCKED() (bool, bool, error) {
 	if t.bindex == nil {
 		return false, false, nil
 	}
@@ -79,8 +113,7 @@ func (t *BleveDest) partialRollbackLOCKED(partition string,
 	}
 
 	if sh, ok := index.(*scorch.Scorch); ok {
-		return t.partialScorchRollbackLOCKED(sh,
-			partition, vBucketUUID, rollbackSeq)
+		return t.partialScorchRollbackLOCKED(sh)
 	}
 
 	return false, false, fmt.Errorf("unknown index type")

pindex_bleve_scorch_rollback.go

@@ -25,8 +25,17 @@ import (
 // Determine which point is of relevance to get to at or before the
 // wanted rollbackSeq and vBucketUUID. If found, rollback to that
 // particular point.
-func (t *BleveDest) partialScorchRollbackLOCKED(sh *scorch.Scorch,
-	partition string, vBucketUUID, rollbackSeq uint64) (bool, bool, error) {
+func (t *BleveDest) partialScorchRollbackLOCKED(sh *scorch.Scorch) (
+	bool, bool, error) {
+	if t == nil || t.rollbackInfo == nil {
+		return false, false, fmt.Errorf("pindex_bleve_scorch_rollback: invalid" +
+			" bleveDest or rollbackInfo")
+	}
+
+	partition := t.rollbackInfo.partition
+	vBucketUUID := t.rollbackInfo.vBucketUUID
+	rollbackSeq := t.rollbackInfo.rollbackSeq
+
 	seqMaxKey := []byte(partition)
 
 	// get vBucketMap/Opaque key