CBG-5467 when resuming via action=start preserve stats (#8370)

Author: torcolvin

db/background_mgr.go

@@ -52,6 +52,17 @@ var errBackgroundManagerProcessAlreadyRunning = base.HTTPErrorf(http.StatusServi
 // errBackgroundManagerProcessAlreadyStopped is returned when an action to Stop a process occurs but it is already stopped.
 var errBackgroundManagerProcessAlreadyStopped = base.HTTPErrorf(http.StatusServiceUnavailable, "Process already stopped")
 
+// backgroundManagerInitMode is the return type of BackgroundManagerProcessI.Init, indicating whether
+// the process is starting fresh or resuming a prior run.
+type backgroundManagerInitMode int
+
+const (
+	// backgroundManagerInitReset means Init started a new run, discarding any previous state.
+	backgroundManagerInitReset backgroundManagerInitMode = iota
+	// backgroundManagerInitResume means Init found a previous run and will continue from where it left off.
+	backgroundManagerInitResume
+)
+
 // backgroundManagerUpdateClusterStatusMode controls whether updateMultiNodeClusterAwareStatus enforces
 // consistency between the local and cluster states.
 type backgroundManagerUpdateClusterStatusMode int
@@ -130,7 +141,8 @@ type BackgroundManagerStatus struct {
 // Examples of this: ReSync, Compaction, Attachment Migration
 type BackgroundManagerProcessI[O any] interface {
 	// Init is called before Run for setup purposes. If Init errors, Run will not happen.
-	Init(ctx context.Context, options O, clusterStatus []byte) error
+	// Returns backgroundManagerInitResume if resuming an existing (stopped) run, backgroundManagerInitReset if starting a new one.
+	Init(ctx context.Context, options O, clusterStatus []byte) (backgroundManagerInitMode, error)
 	// Run implements all of the work of the process.
 	Run(ctx context.Context, options O, persistClusterStatusCallback updateStatusCallbackFunc, terminator *base.SafeTerminator) error
 	// GetProcessStatus accepts the current BackgroundManagerStatus and the previousStatus that was serialized. previousStatus is
@@ -252,7 +264,7 @@ func (b *BackgroundManager[O]) start(ctx context.Context, options O, processClus
 		b.setStartTime(previousStatus.StartTime)
 	}
 
-	err = b.Process.Init(ctx, options, processClusterStatus)
+	initMode, err := b.Process.Init(ctx, options, processClusterStatus)
 	if err != nil {
 		return err
 	}
@@ -317,7 +329,7 @@ func (b *BackgroundManager[O]) start(ctx context.Context, options O, processClus
 		var err error
 		if b.mode() == backgroundManagerModeMultiNode {
 			mode := backgroundManagerStatusStart
-			if isResume {
+			if isResume || initMode == backgroundManagerInitResume {
 				mode = backgroundManagerStatusResume
 			}
 			err = b.updateMultiNodeClusterAwareStatus(ctx, mode)

db/background_mgr_async_index_init.go

@@ -25,12 +25,12 @@ type AsyncIndexInitManager struct {
 }
 
 // Init is called synchronously to set up a run for the background manager process. See Run() for the async part.
-func (a *AsyncIndexInitManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) error {
+func (a *AsyncIndexInitManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) (backgroundManagerInitMode, error) {
 	a.lock.Lock()
 	defer a.lock.Unlock()
 	a._statusMap = options["statusMap"].(*IndexStatusByCollection)
 	a._doneChan = options["doneChan"].(chan error)
-	return nil
+	return backgroundManagerInitReset, nil
 }
 
 // Run is called inside a goroutine to perform the job of the job. This function should block until the job is complete.

db/background_mgr_attachment_compaction.go

@@ -53,7 +53,7 @@ func NewAttachmentCompactionManager(metadataStore base.DataStore, metaKeys *base
 	}
 }
 
-func (a *AttachmentCompactionManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) error {
+func (a *AttachmentCompactionManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) (backgroundManagerInitMode, error) {
 	database := options["database"].(*Database)
 	database.DbStats.Database().CompactionAttachmentStartTime.Set(uint64(time.Now().UTC().Unix()))
 
@@ -88,7 +88,7 @@ func (a *AttachmentCompactionManager) Init(ctx context.Context, options map[stri
 		// process from scratch with a new compaction ID. Otherwise, we should resume with the compact ID, phase and
 		// stats specified in the doc.
 		if statusDoc.State == BackgroundProcessStateCompleted || err != nil || (reset && ok) {
-			return newRunInit()
+			return backgroundManagerInitReset, newRunInit()
 		} else {
 			a.CompactID = statusDoc.CompactID
 			a.Phase = statusDoc.Phase
@@ -100,11 +100,11 @@ func (a *AttachmentCompactionManager) Init(ctx context.Context, options map[stri
 			base.InfofCtx(ctx, base.KeyAll, "Attachment Compaction: Attempting to resume compaction with compact ID: %q phase %q", a.CompactID, a.Phase)
 		}
 
-		return nil
+		return backgroundManagerInitResume, nil
 
 	}
 
-	return newRunInit()
+	return backgroundManagerInitReset, newRunInit()
 }
 
 func (a *AttachmentCompactionManager) Run(ctx context.Context, options map[string]any, persistClusterStatusCallback updateStatusCallbackFunc, terminator *base.SafeTerminator) error {
@@ -222,7 +222,7 @@ func (a *AttachmentCompactionManager) handleAttachmentCompactionRollbackError(ct
 		}
 		if phase == MarkPhase {
 			// initialise new compaction run as we want to start the phase mark again in event of rollback
-			err = a.Init(ctx, options, nil)
+			_, err = a.Init(ctx, options, nil)
 			if err != nil {
 				base.WarnfCtx(ctx, "error on initialization of new run after rollback has been indicated: %s", err)
 				return false, err

db/background_mgr_attachment_migration.go

@@ -52,7 +52,7 @@ func NewAttachmentMigrationManager(database *DatabaseContext) *BackgroundManager
 	}
 }
 
-func (a *AttachmentMigrationManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) error {
+func (a *AttachmentMigrationManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) (backgroundManagerInitMode, error) {
 	newRunInit := func() error {
 		uniqueUUID, err := uuid.NewRandom()
 		if err != nil {
@@ -76,7 +76,7 @@ func (a *AttachmentMigrationManager) Init(ctx context.Context, options map[strin
 		// If the previous run completed, or there was an error during unmarshalling the status we will start the
 		// process from scratch with a new migration ID. Otherwise, we should resume with the migration ID, stats specified in the doc.
 		if statusDoc.State == BackgroundProcessStateCompleted || err != nil || reset {
-			return newRunInit()
+			return backgroundManagerInitReset, newRunInit()
 		}
 		a.MigrationID = statusDoc.MigrationID
 		a.docsProcessed.Store(statusDoc.DocsProcessed)
@@ -86,10 +86,10 @@ func (a *AttachmentMigrationManager) Init(ctx context.Context, options map[strin
 
 		base.InfofCtx(ctx, base.KeyAll, "Attachment Migration: Resuming migration with migration ID: %s, %d already processed", a.MigrationID, a.docsProcessed.Load())
 
-		return nil
+		return backgroundManagerInitResume, nil
 	}
 
-	return newRunInit()
+	return backgroundManagerInitReset, newRunInit()
 }
 
 func (a *AttachmentMigrationManager) Run(ctx context.Context, options map[string]any, persistClusterStatusCallback updateStatusCallbackFunc, terminator *base.SafeTerminator) error {

db/background_mgr_metadata_migration.go

@@ -68,7 +68,7 @@ type MigrationManagerStatusDoc struct {
 	MigrationManagerResponse `json:"status"`
 }
 
-func (m *MetadataMigrationManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) error {
+func (m *MetadataMigrationManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) (backgroundManagerInitMode, error) {
 	newRunInit := func() error {
 		uniqueUUID, err := uuid.NewRandom()
 		if err != nil {
@@ -92,17 +92,17 @@ func (m *MetadataMigrationManager) Init(ctx context.Context, options map[string]
 		// If the previous run completed, there was an error during unmarshalling the status, or
 		// the caller requested a reset, start again with a fresh migration ID and zeroed counters.
 		if status.State == BackgroundProcessStateCompleted || err != nil || reset {
-			return newRunInit()
+			return backgroundManagerInitReset, newRunInit()
 		}
 		m.docsProcessed.Store(status.DocsProcessed)
 		m.docsFailed.Store(status.DocsFailed)
 		m.docsOutOfScope.Store(status.DocsOutOfScope)
 		m.passes.Store(status.Passes)
 		m.MigrationID = status.MigrationID
 		base.InfofCtx(ctx, base.KeyAll, "Metadata Migration: Resuming migration run with migration ID: %s, docs processed: %d, docs failed: %d, passes: %d", m.MigrationID, status.DocsProcessed, status.DocsFailed, status.Passes)
-		return nil
+		return backgroundManagerInitResume, nil
 	}
-	return newRunInit()
+	return backgroundManagerInitReset, newRunInit()
 }
 
 func (m *MetadataMigrationManager) Run(ctx context.Context, options map[string]any, persistClusterStatusCallback updateStatusCallbackFunc, terminator *base.SafeTerminator) error {

db/background_mgr_resync_dcp.go

@@ -137,15 +137,15 @@ func NewResyncManagerDCP(db *DatabaseContext, distributed bool) *BackgroundManag
 }
 
 // Init processes the options to start a resync process and sets them as struct members.
-func (r *ResyncManagerDCP) Init(ctx context.Context, options ResyncOptions, clusterStatus []byte) error {
+func (r *ResyncManagerDCP) Init(ctx context.Context, options ResyncOptions, clusterStatus []byte) (backgroundManagerInitMode, error) {
 	r.setStartOptions(options)
 
 	var collections DatabaseCollections
 	if len(options.Collections) > 0 {
 		var err error
 		collections, err = r.db.collections(options.Collections)
 		if err != nil {
-			return err
+			return backgroundManagerInitReset, err
 		}
 	} else {
 		collections = slices.Collect(maps.Values(r.db.CollectionByID))
@@ -171,7 +171,7 @@ func (r *ResyncManagerDCP) Init(ctx context.Context, options ResyncOptions, clus
 	} else {
 		r.initializeFromPreviousStatus(statusDoc)
 		base.InfofCtx(ctx, base.KeyAll, "Resuming resync with ID: %q", r.ResyncID)
-		return nil
+		return backgroundManagerInitResume, nil
 	}
 
 	if statusDoc.ResyncID != "" {
@@ -183,7 +183,7 @@ func (r *ResyncManagerDCP) Init(ctx context.Context, options ResyncOptions, clus
 
 	newID, err := uuid.NewRandom()
 	if err != nil {
-		return err
+		return backgroundManagerInitReset, err
 	}
 
 	docsTargeted, err := totalResyncDocs(ctx, collections)
@@ -198,13 +198,13 @@ func (r *ResyncManagerDCP) Init(ctx context.Context, options ResyncOptions, clus
 	if r.Distributed {
 		endSeqNosMap, err := base.GetHighSeqNos(ctx, r.db.Bucket)
 		if err != nil {
-			return err
+			return backgroundManagerInitReset, err
 		}
 		r.setEndSeqNosFromMap(endSeqNosMap)
 	}
 
 	base.InfofCtx(ctx, base.KeyAll, "Running new resync process with ID: %q - %s", r.ResyncID, resetMsg)
-	return nil
+	return backgroundManagerInitReset, nil
 }
 
 // totalResyncDocs returns an estimate of the number of documents processed for resync.

db/background_mgr_resync_dcp_test.go

@@ -162,7 +162,7 @@ func TestResyncDCPInit(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			err = db.ResyncManager.Process.Init(ctx, options, clusterData)
+			_, err = db.ResyncManager.Process.Init(ctx, options, clusterData)
 			require.NoError(t, err)
 
 			response := getResyncStats(t, db)
@@ -404,12 +404,9 @@ func TestResyncManagerDCPResumeStoppedProcess(t *testing.T) {
 	require.NoError(t, err)
 
 	stats = waitForResyncState(t, db, BackgroundProcessStateCompleted)
-	if !db.useShardedDCP() {
-		// CBG-5467 these stats are too low for distributed resync
-		assert.GreaterOrEqual(t, stats.DocsProcessed, int64(docsToCreate))
-		assert.Equal(t, int64(docsToCreate), stats.DocsChanged)
-		// DocsTargeted is preserved from the original run start, even after resume
-	}
+	assert.GreaterOrEqual(t, stats.DocsProcessed, int64(docsToCreate))
+	assert.Equal(t, int64(docsToCreate), stats.DocsChanged)
+	// DocsTargeted is preserved from the original run start, even after resume
 	assert.Equal(t, initialDocsTargeted, stats.DocsTargeted)
 	assert.Equal(t, int64(0), db.DbStats.Database().ResyncDocsTargeted.Value()) // resync finished so reset back to 0
 
@@ -1128,7 +1125,8 @@ func TestResyncDCPInitStoresOptionsInMeta(t *testing.T) {
 		Collections: base.NewCollectionNames(),
 	}
 
-	require.NoError(t, db.ResyncManager.Process.Init(ctx, options, nil))
+	_, initErr := db.ResyncManager.Process.Init(ctx, options, nil)
+	require.NoError(t, initErr)
 
 	_, metaBytes, err := db.ResyncManager.Process.GetProcessStatus(BackgroundManagerStatus{State: BackgroundProcessStateStopped}, nil)
 	require.NoError(t, err)

db/background_mgr_test.go

@@ -31,11 +31,11 @@ type MockProcess struct {
 	lock                 sync.Mutex
 }
 
-func (m *MockProcess) Init(ctx context.Context, options map[string]any, clusterStatus []byte) error {
+func (m *MockProcess) Init(ctx context.Context, options map[string]any, clusterStatus []byte) (backgroundManagerInitMode, error) {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 	m.InitCalled = true
-	return nil
+	return backgroundManagerInitReset, nil
 }
 
 func (m *MockProcess) Run(ctx context.Context, options map[string]any, persistClusterStatusCallback updateStatusCallbackFunc, terminator *base.SafeTerminator) error {
@@ -472,7 +472,7 @@ type ResumableMockProcess struct {
 	optionsLock     sync.RWMutex
 }
 
-func (r *ResumableMockProcess) Init(ctx context.Context, options map[string]any, clusterStatus []byte) error {
+func (r *ResumableMockProcess) Init(ctx context.Context, options map[string]any, clusterStatus []byte) (backgroundManagerInitMode, error) {
 	r.optionsLock.Lock()
 	r.receivedOptions = options
 	r.optionsLock.Unlock()

db/background_mgr_tombstone_compaction.go

@@ -34,11 +34,11 @@ func NewTombstoneCompactionManager() *BackgroundManager[map[string]any] {
 	}
 }
 
-func (t *TombstoneCompactionManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) error {
+func (t *TombstoneCompactionManager) Init(ctx context.Context, options map[string]any, clusterStatus []byte) (backgroundManagerInitMode, error) {
 	database := options["database"].(*Database)
 	database.DbStats.Database().CompactionTombstoneStartTime.Set(uint64(time.Now().UTC().Unix()))
 
-	return nil
+	return backgroundManagerInitReset, nil
 }
 
 func (t *TombstoneCompactionManager) Run(ctx context.Context, options map[string]any, persistClusterStatusCallback updateStatusCallbackFunc, terminator *base.SafeTerminator) error {

db/database_test.go

@@ -4521,8 +4521,8 @@ type testBackgroundProcess[O any] struct {
 	isStoppable bool
 }
 
-func (i *testBackgroundProcess[O]) Init(_ context.Context, _ O, _ []byte) error {
-	return nil
+func (i *testBackgroundProcess[O]) Init(_ context.Context, _ O, _ []byte) (backgroundManagerInitMode, error) {
+	return backgroundManagerInitReset, nil
 }
 
 func (i *testBackgroundProcess[O]) Run(_ context.Context, _ O, _ updateStatusCallbackFunc, terminator *base.SafeTerminator) error {