centralize validation

Author: torcolvin

db/background_mgr_resync_dcp.go

@@ -30,7 +30,7 @@ import (
 // Resync Implementation of Background Manager Process using DCP stream
 // =====================================================================
 
-const DefaultResyncPartitions = 64
+const DefaultResyncPartitions uint16 = 64
 
 type ResyncManagerDCP struct {
 	db                           *DatabaseContext
@@ -413,12 +413,7 @@ func (r *ResyncManagerDCP) Run(ctx context.Context, options ResyncOptions, persi
 			return fmt.Errorf("Error generating CBGT index name: %v", err)
 		}
 
-		var partitionCount uint16
-		if db.Options.UnsupportedOptions != nil && db.Options.UnsupportedOptions.ResyncPartitions != nil && *db.Options.UnsupportedOptions.ResyncPartitions > 0 {
-			partitionCount = *db.Options.UnsupportedOptions.ResyncPartitions
-		} else {
-			partitionCount = DefaultResyncPartitions
-		}
+		partitionCount := r.db.GetResyncPartitionCount()
 		base.DebugfCtx(ctx, base.KeyAll, "Using %d partitions for resync", partitionCount)
 
 		opts := base.ShardedDCPOptions{

db/background_mgr_resync_dcp_test.go

@@ -243,6 +243,8 @@ func TestResyncManagerDCPStart(t *testing.T) {
 		require.NoError(t, db.ResyncManager.Start(ctx, options))
 		stats := waitForResyncState(t, db, BackgroundProcessStateCompleted)
 
+		assert.Equal(t, DefaultResyncPartitions, db.GetResyncPartitionCount())
+
 		assert.GreaterOrEqual(t, stats.DocsProcessed, int64(docsToCreate)) // may be processing tombstones from previous tests
 		assert.Equal(t, int64(0), stats.DocsChanged)
 		assert.GreaterOrEqual(t, stats.DocsTargeted, uint64(docsToCreate))
@@ -284,6 +286,8 @@ func TestResyncManagerDCPStart(t *testing.T) {
 
 		RequireBackgroundManagerState(t, db.ResyncManager, BackgroundProcessStateCompleted)
 
+		assert.Equal(t, DefaultResyncPartitions, db.GetResyncPartitionCount())
+
 		stats := getResyncStats(t, db)
 		// If there are tombstones from older docs which have been deleted from the bucket, processed docs will
 		// be greater than DocsChanged

db/database.go

@@ -471,6 +471,11 @@ func NewDatabaseContext(ctx context.Context, dbName string, bucket base.Bucket,
 	if err != nil {
 		return nil, err
 	}
+	if dbContext.useShardedDCP() {
+		if partitionCount := dbContext.GetResyncPartitionCount(); partitionCount > dbContext.numVBuckets {
+			return nil, NewDatabaseError(DatabaseInvalidResyncPartitions)
+		}
+	}
 	err = dbContext.updateCCVSettings(ctx)
 	if err != nil {
 		return nil, err
@@ -2726,6 +2731,15 @@ func (db *DatabaseContext) NumVBuckets() uint16 {
 	return db.numVBuckets
 }
 
+// GetResyncPartitionCount returns the number of partitions to use for the DCP resync feed.
+// Returns the configured ResyncPartitions value if set, otherwise DefaultResyncPartitions.
+func (db *DatabaseContext) GetResyncPartitionCount() uint16 {
+	if db.Options.UnsupportedOptions != nil && db.Options.UnsupportedOptions.ResyncPartitions != nil && *db.Options.UnsupportedOptions.ResyncPartitions > 0 {
+		return *db.Options.UnsupportedOptions.ResyncPartitions
+	}
+	return DefaultResyncPartitions
+}
+
 // InitializeOfflineMode starts polling the database state when the database transitions to offline mode.
 // This enables the database to detect state changes (such as resync requests) from other nodes in the cluster
 // while it is offline. The polling mechanism watches the metadata store for updates to the database state

db/database_error.go

@@ -53,3 +53,7 @@ func NewDatabaseError(code databaseErrorCode) *DatabaseError {
 		Code:   code,
 	}
 }
+
+func (e *DatabaseError) Error() string {
+	return e.ErrMsg
+}

docs/api/components/schemas.yaml

@@ -1725,7 +1725,7 @@ Database:
           type: boolean
         resync_partitions:
           description: |-
-            Number of partitions to use for distributed DCP resync. If not set, defaults to the import partitions value.
+            Number of partitions to use for distributed DCP resync. Maximum number is the number of vBuckets for the backing bucket.
           type: integer
           minimum: 1
           maximum: 1024

rest/adminapitest/admin_api_test.go

@@ -751,6 +751,9 @@ func TestResyncUsingDCPStream(t *testing.T) {
 
 			resyncManagerStatus := rt.WaitForResyncDCPStatus(db.BackgroundProcessStateCompleted)
 
+			dbCtx := rt.GetDatabase()
+			assert.Equal(t, db.DefaultResyncPartitions, dbCtx.GetResyncPartitionCount())
+
 			assert.Equal(t, testCase.docsCreated, int(rt.GetDatabase().DbStats.Database().SyncFunctionCount.Value()))
 			if !base.UnitTestUrlIsWalrus() && !base.TestsDisableGSI() {
 				// It is possible for Couchbase Server GSI runs which use DCP purge to two DCP events from a previous

rest/adminapitest/resync_test.go

@@ -467,23 +467,44 @@ func TestResyncRequireResyncDefaultMetadataID(t *testing.T) {
 	rt.WaitForDatabaseState(db2Name, db.RunStateString[db.DBOnline])
 }
 
-func TestResyncPartitionsMaximumValidation(t *testing.T) {
+func TestResyncPartitionsValidation(t *testing.T) {
 	rt := rest.NewRestTester(t, &rest.RestTesterConfig{
 		PersistentConfig: true,
 	})
 	defer rt.Close()
 
-	dbConfig := rt.NewDbConfig()
-	numVBuckets, err := rt.TestBucket.GetMaxVbno(base.TestCtx(t))
+	numVBuckets, err := rt.TestBucket.GetMaxVbno(rt.Context())
 	require.NoError(t, err)
-	invalidPartitions := numVBuckets + 1
-	dbConfig.Unsupported = &db.UnsupportedOptions{
-		ResyncPartitions: &invalidPartitions,
-	}
 
-	resp := rt.CreateDatabase("db1", dbConfig)
-	rest.RequireStatus(t, resp, http.StatusInternalServerError)
-	assert.Contains(t, resp.Body.String(), "resync_partitions must be between 1 and")
+	t.Run("invalid partitions exceeding vbucket count are rejected", func(t *testing.T) {
+		// Partition count validation only applies when using sharded DCP (EE + CBS).
+		if !base.IsEnterpriseEdition() || base.UnitTestUrlIsWalrus() {
+			t.Skip("Partition count validation only applies to EE with CBS")
+		}
+		dbConfig := rt.NewDbConfig()
+		invalidPartitions := numVBuckets + 1
+		dbConfig.Unsupported = &db.UnsupportedOptions{
+			ResyncPartitions: &invalidPartitions,
+		}
+		resp := rt.CreateDatabase("db1", dbConfig)
+		rest.RequireStatus(t, resp, http.StatusInternalServerError)
+		assert.Contains(t, resp.Body.String(), "resync_partitions must be between 1 and")
+	})
+
+	t.Run("valid custom partitions are used by ResyncManagerDCP", func(t *testing.T) {
+		dbConfig := rt.NewDbConfig()
+		customPartitions := uint16(16)
+		dbConfig.Unsupported = &db.UnsupportedOptions{
+			ResyncPartitions: &customPartitions,
+		}
+		resp := rt.CreateDatabase("db2", dbConfig)
+		rest.RequireStatus(t, resp, http.StatusCreated)
+
+		dbCtx, err := rt.ServerContext().GetDatabase(rt.Context(), "db2")
+		require.NoError(t, err)
+
+		assert.Equal(t, customPartitions, dbCtx.GetResyncPartitionCount())
+	})
 }
 
 // TestDistributedResync is a long-running dev-time test used to validate cbgt rebalance behavior during resync.

rest/server_context.go

@@ -1037,7 +1037,11 @@ func (sc *ServerContext) _getOrAddDatabaseFromConfig(ctx context.Context, config
 	dbcontext, err = db.NewDatabaseContext(ctx, dbName, bucket, autoImport, contextOptions)
 	if err != nil {
 		if options.loadFromBucket {
-			sc._handleInvalidDatabaseConfig(ctx, spec.BucketName, config, db.NewDatabaseError(db.DatabaseCreateDatabaseContextError))
+			var dbErr *db.DatabaseError
+			if !errors.As(err, &dbErr) {
+				dbErr = db.NewDatabaseError(db.DatabaseCreateDatabaseContextError)
+			}
+			sc._handleInvalidDatabaseConfig(ctx, spec.BucketName, config, dbErr)
 		}
 		return nil, err
 	}
@@ -1093,13 +1097,6 @@ func (sc *ServerContext) _getOrAddDatabaseFromConfig(ctx context.Context, config
 		}
 	}
 
-	if config.Unsupported != nil && config.Unsupported.ResyncPartitions != nil && *config.Unsupported.ResyncPartitions > dbcontext.NumVBuckets() {
-		if options.loadFromBucket {
-			sc._handleInvalidDatabaseConfig(ctx, spec.BucketName, config, db.NewDatabaseError(db.DatabaseInvalidResyncPartitions))
-		}
-		return nil, fmt.Errorf("resync_partitions must be between 1 and %d, got %d", dbcontext.NumVBuckets(), *config.Unsupported.ResyncPartitions)
-	}
-
 	if config.CORS != nil {
 		dbcontext.CORS = config.DbConfig.CORS
 	} else {