Allow resync background manager Resume to round trip options (#8331)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Adam Fraser <adam.fraser@couchbase.com>

Author: 3 people

db/background_mgr.go

@@ -89,10 +89,10 @@ func (a *AsyncIndexInitManager) ResetStatus() {
 	return
 }
 
-var _ BackgroundManagerProcessI = &AsyncIndexInitManager{}
+var _ BackgroundManagerProcessI[map[string]any] = &AsyncIndexInitManager{}
 
-func NewAsyncIndexInitManager(metadataStore base.DataStore, metaKeys *base.MetadataKeys) *BackgroundManager {
-	return &BackgroundManager{
+func NewAsyncIndexInitManager(metadataStore base.DataStore, metaKeys *base.MetadataKeys) *BackgroundManager[map[string]any] {
+	return &BackgroundManager[map[string]any]{
 		name:    "index_init",
 		Process: &AsyncIndexInitManager{},
 		clusterAwareOptions: &ClusterAwareBackgroundManagerOptions{

db/background_mgr_async_index_init.go

@@ -38,10 +38,10 @@ type AttachmentCompactionManager struct {
 	runFunctionStartedCallback atomic.Pointer[runFunctionStartedCallbackFunc]
 }
 
-var _ BackgroundManagerProcessI = &AttachmentCompactionManager{}
+var _ BackgroundManagerProcessI[map[string]any] = &AttachmentCompactionManager{}
 
-func NewAttachmentCompactionManager(metadataStore base.DataStore, metaKeys *base.MetadataKeys) *BackgroundManager {
-	return &BackgroundManager{
+func NewAttachmentCompactionManager(metadataStore base.DataStore, metaKeys *base.MetadataKeys) *BackgroundManager[map[string]any] {
+	return &BackgroundManager[map[string]any]{
 		name:    "attachment_compaction",
 		Process: &AttachmentCompactionManager{},
 		clusterAwareOptions: &ClusterAwareBackgroundManagerOptions{

db/background_mgr_attachment_compaction.go

@@ -31,14 +31,14 @@ type AttachmentMigrationManager struct {
 	lock          sync.RWMutex
 }
 
-var _ BackgroundManagerProcessI = &AttachmentMigrationManager{}
+var _ BackgroundManagerProcessI[map[string]any] = &AttachmentMigrationManager{}
 
 const MetaVersionValue = "4.0.0" // Meta version to set in syncInfo document upon completion of attachment migration for collection
 
-func NewAttachmentMigrationManager(database *DatabaseContext) *BackgroundManager {
+func NewAttachmentMigrationManager(database *DatabaseContext) *BackgroundManager[map[string]any] {
 	metadataStore := database.MetadataStore
 	metaKeys := database.MetadataKeys
-	return &BackgroundManager{
+	return &BackgroundManager[map[string]any]{
 		name: "attachment_migration",
 		Process: &AttachmentMigrationManager{
 			databaseCtx: database,

db/background_mgr_attachment_migration.go

@@ -34,15 +34,15 @@ type MetadataMigrationManager struct {
 
 const MetadataMigrationManagerName = "metadata_migration"
 
-var _ BackgroundManagerProcessI = &MetadataMigrationManager{}
+var _ BackgroundManagerProcessI[map[string]any] = &MetadataMigrationManager{}
 
 // errMetadataMigrationTerminated is the cancellation cause propagated to the run context when
 // the BackgroundManager terminator fires, so ops blocked on ctx (seq-counter retry loop, range
 // scan iteration) can distinguish an operator stop from a parent-context cancellation.
 var errMetadataMigrationTerminated = errors.New("metadata migration terminated by stop request")
 
-func NewMetadataMigrationManager(dbContext *DatabaseContext) *BackgroundManager {
-	return &BackgroundManager{
+func NewMetadataMigrationManager(dbContext *DatabaseContext) *BackgroundManager[map[string]any] {
+	return &BackgroundManager[map[string]any]{
 		name:    MetadataMigrationManagerName,
 		Process: &MetadataMigrationManager{dbContext: dbContext},
 		clusterAwareOptions: &ClusterAwareBackgroundManagerOptions{

db/background_mgr_metadata_migration.go

@@ -45,7 +45,7 @@ type ResyncManagerDCP struct {
 	ResyncID                     string
 	VBUUIDs                      []uint64
 	ResyncedCollections          base.CollectionNames
-	startOptions                 map[string]any // options from the most recent Start call, persisted to meta for Resume
+	startOptions                 ResyncOptions // options from the most recent Start call, persisted to meta for Resume
 	resyncCollectionInfo
 	lock              sync.RWMutex
 	Distributed       bool
@@ -94,13 +94,20 @@ type resyncCollectionInfo struct {
 	hasAllCollections bool
 }
 
-var _ BackgroundManagerProcessI = &ResyncManagerDCP{}
+// ResyncOptions are used to initialize a resync process.
+type ResyncOptions struct {
+	Collections         base.CollectionNames `json:"collections,omitempty"`
+	Reset               bool                 `json:"reset,omitempty"`
+	RegenerateSequences bool                 `json:"regenerateSequences,omitempty"`
+}
+
+var _ BackgroundManagerProcessI[ResyncOptions] = &ResyncManagerDCP{}
 
 // NewResyncManagerDCP returns a new instance of ResyncManagerDCP wrapped in a BackgroundManager. If distributed is
 // true, the manager will be set up to run in a distributed manner across multiple nodes, otherwise it will run on a
 // single node.
-func NewResyncManagerDCP(db *DatabaseContext, distributed bool) *BackgroundManager {
-	b := &BackgroundManager{
+func NewResyncManagerDCP(db *DatabaseContext, distributed bool) *BackgroundManager[ResyncOptions] {
+	b := &BackgroundManager[ResyncOptions]{
 		name: "resync",
 		Process: &ResyncManagerDCP{
 			db:                db,
@@ -126,18 +133,14 @@ func NewResyncManagerDCP(db *DatabaseContext, distributed bool) *BackgroundManag
 	return b
 }
 
-// Init processes the options to start a resync process and sets them as struct memebers.
-func (r *ResyncManagerDCP) Init(ctx context.Context, options map[string]any, clusterStatus []byte) error {
-	resyncCollections, ok := options["collections"].(base.CollectionNames)
-	if !ok {
-		return errors.New("collections option is required and must be of type base.CollectionNames")
-	}
+// 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 {
 	r.setStartOptions(options)
 
 	var collections DatabaseCollections
-	if len(resyncCollections) > 0 {
+	if len(options.Collections) > 0 {
 		var err error
-		collections, err = r.db.collections(resyncCollections)
+		collections, err = r.db.collections(options.Collections)
 		if err != nil {
 			return err
 		}
@@ -154,7 +157,7 @@ func (r *ResyncManagerDCP) Init(ctx context.Context, options map[string]any, clu
 	var statusDoc ResyncManagerStatusDocDCP
 	if clusterStatus == nil {
 		resetMsg = "no previous run found"
-	} else if resetOpt, _ := options["reset"].(bool); resetOpt {
+	} else if options.Reset {
 		resetMsg = "reset option requested"
 	} else if err := base.JSONUnmarshal(clusterStatus, &statusDoc); err != nil {
 		resetMsg = "failed to unmarshal cluster status"
@@ -216,7 +219,7 @@ func (r *ResyncManagerDCP) purgeCheckpoints(ctx context.Context, resyncID string
 }
 
 // setStartOptions stores the options used to start the current run so that Resume can reconstruct them.
-func (r *ResyncManagerDCP) setStartOptions(options map[string]any) {
+func (r *ResyncManagerDCP) setStartOptions(options ResyncOptions) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 	r.startOptions = options
@@ -230,12 +233,9 @@ func (r *ResyncManagerDCP) SetVBUUIDs(vbuuids []uint64) {
 }
 
 // Run starts a DCP feed to process documents for resync.
-func (r *ResyncManagerDCP) Run(ctx context.Context, options map[string]any, persistClusterStatusCallback updateStatusCallbackFunc, terminator *base.SafeTerminator) (err error) {
+func (r *ResyncManagerDCP) Run(ctx context.Context, options ResyncOptions, persistClusterStatusCallback updateStatusCallbackFunc, terminator *base.SafeTerminator) (err error) {
 	db := r.db
-	regenerateSequences, ok := options["regenerateSequences"].(bool)
-	if !ok {
-		return errors.New("regenerateSequences option is required and must be of type bool")
-	}
+	regenerateSequences := options.RegenerateSequences
 	ctx = context.WithoutCancel(ctx) // drop cancellation from parent context
 	ctx = base.CorrelationIDLogCtx(ctx, r.ResyncID)
 	ctx, cancelResync := context.WithCancelCause(ctx)
@@ -789,9 +789,9 @@ type resyncManagerCompletedVBuckets struct {
 }
 
 type ResyncManagerMeta struct {
-	VBUUIDs       []uint64       `json:"vbuuids"`
-	CollectionIDs []uint32       `json:"collection_ids,omitempty"`
-	Options       map[string]any `json:"options,omitempty"` // start options, persisted for Resume
+	VBUUIDs       []uint64      `json:"vbuuids"`
+	CollectionIDs []uint32      `json:"collection_ids,omitempty"`
+	Options       ResyncOptions `json:"options,omitempty"` // start options, persisted for Resume
 	resyncManagerCompletedVBuckets
 }