diff --git a/db/background_mgr_resync_dcp.go b/db/background_mgr_resync_dcp.go index a0aa6615ad..a155e64c85 100644 --- a/db/background_mgr_resync_dcp.go +++ b/db/background_mgr_resync_dcp.go @@ -30,6 +30,8 @@ import ( // Resync Implementation of Background Manager Process using DCP stream // ===================================================================== +const DefaultResyncPartitions uint16 = 64 + type ResyncManagerDCP struct { db *DatabaseContext docsProcessedLocal atomic.Int64 // number of documents processed locally on this node since the last start or resume of resync @@ -411,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 = db.Options.ImportOptions.ImportPartitions - } + partitionCount := r.db.GetResyncPartitionCount() base.DebugfCtx(ctx, base.KeyAll, "Using %d partitions for resync", partitionCount) opts := base.ShardedDCPOptions{ diff --git a/db/background_mgr_resync_dcp_test.go b/db/background_mgr_resync_dcp_test.go index 5e7774ab80..c1c9458ee7 100644 --- a/db/background_mgr_resync_dcp_test.go +++ b/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 diff --git a/db/database.go b/db/database.go index 74616d671a..4f6d941dd5 100644 --- a/db/database.go +++ b/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 diff --git a/db/database_error.go b/db/database_error.go index f89f317c6e..a524503f8b 100644 --- a/db/database_error.go +++ b/db/database_error.go @@ -53,3 +53,7 @@ func NewDatabaseError(code databaseErrorCode) *DatabaseError { Code: code, } } + +func (e *DatabaseError) Error() string { + return e.ErrMsg +} diff --git a/docs/api/components/schemas.yaml b/docs/api/components/schemas.yaml index b9301306d1..51e3863772 100644 --- a/docs/api/components/schemas.yaml +++ b/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 diff --git a/rest/adminapitest/admin_api_test.go b/rest/adminapitest/admin_api_test.go index cdffa1bf57..96a61a9d5c 100644 --- a/rest/adminapitest/admin_api_test.go +++ b/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 diff --git a/rest/adminapitest/resync_test.go b/rest/adminapitest/resync_test.go index b802df001d..e431a31777 100644 --- a/rest/adminapitest/resync_test.go +++ b/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. diff --git a/rest/server_context.go b/rest/server_context.go index c5456ceade..bf7207458f 100644 --- a/rest/server_context.go +++ b/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 {