diff --git a/rest/admin_api.go b/rest/admin_api.go index cc13935a60..20cc2a315c 100644 --- a/rest/admin_api.go +++ b/rest/admin_api.go @@ -1081,7 +1081,10 @@ func (h *handler) handlePutDbConfig() (err error) { h.setEtag(updatedDbConfig.Version) h.server._databasesLock.Lock() defer h.server._databasesLock.Unlock() - h.server._dbConfigs[dbName].cfgCas = cas + // Another request or the config poller can remove or reload the database after this update loaded it. + if loadedConfig, ok := h.server._dbConfigs[dbName]; ok && loadedConfig.Version == updatedDbConfig.Version { + loadedConfig.cfgCas = cas + } base.Audit(h.ctx(), base.AuditIDUpdateDatabaseConfig, auditFields) return base.HTTPErrorf(http.StatusCreated, "updated") diff --git a/rest/cluster_compat_test.go b/rest/cluster_compat_test.go index 984cd4437d..7ea25fe1d9 100644 --- a/rest/cluster_compat_test.go +++ b/rest/cluster_compat_test.go @@ -14,6 +14,7 @@ import ( "fmt" "maps" "net/http" + "sync/atomic" "testing" "time" @@ -643,6 +644,48 @@ func TestClusterCompatAppliedDBVersionUpdatedByUpdateConfigAndReloadDatabase(t * assert.NotEqual(t, versionAfterCreate, versionAfterOffline, "version should change after taking db offline") } +// TestUpdateDbConfigWithConcurrentPeerHeartbeats verifies that POST /{db}/_config succeeds when a peer +// heartbeat writes the registry during every database reload, and that the reload is not repeated. +func TestUpdateDbConfigWithConcurrentPeerHeartbeats(t *testing.T) { + // Heartbeat from inside the reload, between UpdateConfig's registry read and write, so every + // attempt that reloads the database gets a CAS mismatch. + var reloads atomic.Int32 + var rt *RestTester + rt = NewRestTester(t, &RestTesterConfig{ + PersistentConfig: true, + ConnectToBucketFn: func(ctx context.Context, spec base.BucketSpec, failFast bool) (base.Bucket, error) { + reloads.Add(1) + _, err := rt.ServerContext().BootstrapContext.RegisterNodeVersion(ctx, RegisterNodeVersionOpts{ + BucketName: spec.BucketName, + NodeUID: "peer", + Version: base.NodeClusterCompatVersion, + HeartbeatExpiry: time.Hour, + }) + assert.NoError(t, err) + return db.ConnectToBucket(ctx, spec, failFast) + }, + }) + defer rt.Close() + RequireStatus(t, rt.CreateDatabase("db", rt.NewDbConfig()), http.StatusCreated) + reloads.Store(0) + sc := rt.ServerContext() + + dbConfig := rt.NewDbConfig() + dbConfig.RevsLimit = new(uint32(123)) + RequireStatus(t, rt.UpsertDbConfig("db", dbConfig), http.StatusCreated) + assert.Equal(t, int32(1), reloads.Load()) + + var bucketConfig DatabaseConfig + bucketCas, err := sc.BootstrapContext.GetConfig(rt.Context(), rt.Bucket().GetName(), sc.Config.Bootstrap.ConfigGroupID, "db", &bucketConfig) + require.NoError(t, err) + require.Equal(t, uint32(123), *bucketConfig.RevsLimit) + loadedConfig := sc.GetDatabaseConfig("db") + require.NotNil(t, loadedConfig) + assert.Equal(t, bucketConfig.Version, loadedConfig.Version) + assert.Equal(t, bucketCas, loadedConfig.cfgCas) + assert.Equal(t, uint32(123), *loadedConfig.RevsLimit) +} + // TestIsConfigFullyApplied is a truth-table test covering the edge cases of // GatewayRegistry.IsConfigFullyApplied. func TestIsConfigFullyApplied(t *testing.T) { diff --git a/rest/config_concurrency_test.go b/rest/config_concurrency_test.go new file mode 100644 index 0000000000..0bc645e668 --- /dev/null +++ b/rest/config_concurrency_test.go @@ -0,0 +1,206 @@ +// Copyright 2026-Present Couchbase, Inc. +// +// Use of this software is governed by the Business Source License included +// in the file licenses/BSL-Couchbase.txt. As of the Change Date specified +// in that file, in accordance with the Business Source License, use of this +// software will be governed by the Apache License, Version 2.0, included in +// the file licenses/APL2.txt. + +package rest + +import ( + "context" + "fmt" + "math/rand/v2" + "net/http" + "slices" + "sync" + "testing" + "time" + + "github.com/couchbase/sync_gateway/base" + "github.com/couchbase/sync_gateway/db" + "github.com/couchbase/sync_gateway/testing/assert" + "github.com/couchbase/sync_gateway/testing/require" +) + +// dbConfigOperation is one admin API operation that TestConcurrentDbConfigOperationsAcrossNodes sends to a random node. +type dbConfigOperation struct { + name string + weight int + allowed []int // statuses expected while other nodes create, update and delete the same database + sendFunc func(rt *RestTester, rng *rand.Rand) *TestResponse +} + +func dbConfigOperations() []dbConfigOperation { + return []dbConfigOperation{ + { + name: "create", + weight: 2, + allowed: []int{http.StatusCreated, http.StatusPreconditionFailed}, + sendFunc: func(rt *RestTester, _ *rand.Rand) *TestResponse { + return rt.SendAdminRequest(http.MethodPut, "/db/", string(base.MustJSONMarshal(rt.TB(), rt.NewDbConfig()))) + }, + }, + { + name: "upsert config", + weight: 4, + allowed: []int{http.StatusCreated, http.StatusNotFound}, + sendFunc: func(rt *RestTester, rng *rand.Rand) *TestResponse { + dbConfig := rt.NewDbConfig() + dbConfig.RevsLimit = new(uint32(100 + rng.IntN(100))) + return rt.SendAdminRequest(http.MethodPost, "/db/_config", string(base.MustJSONMarshal(rt.TB(), dbConfig))) + }, + }, + { + name: "replace config", + weight: 2, + allowed: []int{http.StatusCreated, http.StatusNotFound}, + sendFunc: func(rt *RestTester, rng *rand.Rand) *TestResponse { + dbConfig := rt.NewDbConfig() + dbConfig.RevsLimit = new(uint32(100 + rng.IntN(100))) + return rt.SendAdminRequest(http.MethodPut, "/db/_config", string(base.MustJSONMarshal(rt.TB(), dbConfig))) + }, + }, + { + name: "update sync function", + weight: 2, + allowed: []int{http.StatusOK, http.StatusNotFound}, + sendFunc: func(rt *RestTester, rng *rand.Rand) *TestResponse { + return rt.SendAdminRequest(http.MethodPut, "/db/_config/sync", fmt.Sprintf(`function(doc) { channel("ch%d"); }`, rng.IntN(10))) + }, + }, + { + name: "offline", + weight: 1, + allowed: []int{http.StatusOK, http.StatusNotFound}, + sendFunc: func(rt *RestTester, _ *rand.Rand) *TestResponse { + return rt.SendAdminRequest(http.MethodPost, "/db/_offline", "") + }, + }, + { + name: "online", + weight: 1, + allowed: []int{http.StatusOK, http.StatusNotFound}, + sendFunc: func(rt *RestTester, _ *rand.Rand) *TestResponse { + return rt.SendAdminRequest(http.MethodPost, "/db/_online", "") + }, + }, + { + name: "delete", + weight: 1, + allowed: []int{http.StatusOK, http.StatusNotFound}, + sendFunc: func(rt *RestTester, _ *rand.Rand) *TestResponse { + return rt.SendAdminRequest(http.MethodDelete, "/db/", "") + }, + }, + } +} + +// isExpectedStatus reports whether resp is one of the operation's allowed statuses. +func (op dbConfigOperation) isExpectedStatus(resp *TestResponse) bool { + return slices.Contains(op.allowed, resp.Code) +} + +// pickDbConfigOperation returns a random operation, chosen in proportion to its weight. +func pickDbConfigOperation(rng *rand.Rand, operations []dbConfigOperation) dbConfigOperation { + totalWeight := 0 + for _, op := range operations { + totalWeight += op.weight + } + n := rng.IntN(totalWeight) + for _, op := range operations { + if n < op.weight { + return op + } + n -= op.weight + } + return operations[len(operations)-1] +} + +// TestConcurrentDbConfigOperationsAcrossNodes sends random database config operations to random nodes of a +// cluster that polls configs, and therefore heartbeats into the registry, more often than a database reloads. +// It then checks that every node and the registry converge on one config. The seed is logged, but goroutine +// scheduling still varies between runs, so a failure is not guaranteed to reproduce. +func TestConcurrentDbConfigOperationsAcrossNodes(t *testing.T) { + t.Skip("CBG-5935 config writes can panic when a concurrent config write reloads the database") + if testing.Short() { + t.Skip("skipping randomized concurrency test in short mode") + } + const ( + numNodes = 3 + numWorkers = 4 + opsPerWorker = 15 + pollInterval = 50 * time.Millisecond + reloadDelay = 150 * time.Millisecond // longer than pollInterval, as a Couchbase Server backed reload is in a real cluster + convergeTimeout = 30 * time.Second + ) + seed := rand.Uint64() + t.Logf("seed: %d", seed) + + ctx := base.TestCtx(t) + rtc := NewRestTesterCluster(t, &RestTesterClusterConfig{ + NumNodes: numNodes, + MutateStartupConfig: func(config *StartupConfig) { + config.Bootstrap.ConfigUpdateFrequency = base.NewConfigDuration(pollInterval) + }, + ConnectToBucketFn: func(ctx context.Context, spec base.BucketSpec, failFast bool) (base.Bucket, error) { + time.Sleep(reloadDelay) + return db.ConnectToBucket(ctx, spec, failFast) + }, + }) + defer rtc.Close(ctx) + + operations := dbConfigOperations() + var wg sync.WaitGroup + for worker := range numWorkers { + wg.Go(func() { + rng := rand.New(rand.NewPCG(seed, uint64(worker))) + for i := range opsPerWorker { + op := pickDbConfigOperation(rng, operations) + node := rng.IntN(numNodes) + resp := op.sendFunc(rtc.Node(node), rng) + assert.Truef(t, op.isExpectedStatus(resp), "worker %d op %d: %s on node %d returned %d, expected one of %v: %s", + worker, i, op.name, node, resp.Code, op.allowed, resp.Body.String()) + } + }) + } + wg.Wait() + + // Leave the database online with a known config, then wait for every node and the registry to agree on it. + rtA := rtc.Node(0) + require.EventuallyWithT(t, func(c *assert.CollectT) { + resp := rtA.CreateDatabase("db", rtA.NewDbConfig()) + assert.Contains(c, []int{http.StatusCreated, http.StatusPreconditionFailed}, resp.Code, resp.Body.String()) + resp = rtA.UpsertDbConfig("db", rtA.NewDbConfig()) + assert.Equal(c, http.StatusCreated, resp.Code, resp.Body.String()) + resp = rtA.SendAdminRequest(http.MethodPost, "/db/_online", "") + assert.Equal(c, http.StatusOK, resp.Code, resp.Body.String()) + }, convergeTimeout, 100*time.Millisecond) + + bootstrap := rtA.ServerContext().BootstrapContext + bucketName := rtA.Bucket().GetName() + groupID := rtA.ServerContext().Config.Bootstrap.ConfigGroupID + require.EventuallyWithT(t, func(c *assert.CollectT) { + var bucketConfig DatabaseConfig + _, err := bootstrap.GetConfig(ctx, bucketName, groupID, "db", &bucketConfig) + if !assert.NoError(c, err) { + return + } + registry, err := bootstrap.getGatewayRegistry(ctx, bucketName) + if !assert.NoError(c, err) { + return + } + registryDb, ok := registry.getRegistryDatabase(groupID, "db") + if assert.True(c, ok, "db missing from registry") { + assert.Equal(c, bucketConfig.Version, registryDb.Version, "registry version") + assert.Nil(c, registryDb.PreviousVersion, "registry has an unfinished update") + } + rtc.ForEachNode(func(rt *RestTester) { + nodeConfig := rt.ServerContext().GetDatabaseConfig("db") + if assert.NotNil(c, nodeConfig, "db not loaded on node") { + assert.Equal(c, bucketConfig.Version, nodeConfig.Version, "node config version") + } + }) + }, convergeTimeout, 100*time.Millisecond) +} diff --git a/rest/config_manager.go b/rest/config_manager.go index 6f2da7c0b8..51d342e562 100644 --- a/rest/config_manager.go +++ b/rest/config_manager.go @@ -160,6 +160,7 @@ func (b *bootstrapContext) InsertConfig(ctx context.Context, bucketName, groupID // 4. Update the registry to remove the previous version func (b *bootstrapContext) UpdateConfig(ctx context.Context, bucketName, groupID, dbName string, updateCallback func(bucketConfig *DatabaseConfig) (updatedConfig *DatabaseConfig, err error)) (newCAS uint64, err error) { var updatedConfig *DatabaseConfig + var callbackCfgCas uint64 // cfgCas of the bucket config that produced updatedConfig var previousVersion string var createdAtTime *time.Time @@ -183,10 +184,15 @@ func (b *bootstrapContext) UpdateConfig(ctx context.Context, bucketName, groupID // Step 2. Update registry to update registry entry, and move previous registry entry to PreviousVersion previousVersion = existingConfig.Version - var callbackErr error - updatedConfig, callbackErr = updateCallback(existingConfig) - if callbackErr != nil { - return false, callbackErr, nil + // A retry after a concurrent registry write (e.g. a heartbeat) reuses the callback's result, because the callback + // can have side effects such as reloading the database. An unchanged cfgCas means the bucket config is unchanged. + if updatedConfig == nil || existingConfig.cfgCas != callbackCfgCas { + var callbackErr error + updatedConfig, callbackErr = updateCallback(existingConfig) + if callbackErr != nil { + return false, callbackErr, nil + } + callbackCfgCas = existingConfig.cfgCas } // Update database in registry @@ -459,6 +465,7 @@ func (b *bootstrapContext) GetDatabaseConfigs(ctx context.Context, bucketName, g // getConfigVersionWithRetry attempts to retrieve the specified config file. // On file not found, will perform backoff retry up to specified timeout. On timeout, returns the nil and rollback error. // On mismatched version when registry version is newer, will perform backoff retry up to timeout. On timeout, returns the config and rollback error. +// While retrying, returns ErrConfigRegistryReloadRequired if the registry no longer has the requested version. // On mismatched version when config version is newer, returns the config and ErrConfigVersionMismatch. func (b *bootstrapContext) getConfigVersionWithRetry(ctx context.Context, bucketName, groupID, dbName, version string) (*DatabaseConfig, error) { @@ -472,7 +479,7 @@ func (b *bootstrapContext) getConfigVersionWithRetry(ctx context.Context, bucket metadataKey := PersistentConfigKey(ctx, groupID, dbName) cas, err := b.Connection.GetMetadataDocument(ctx, bucketName, metadataKey, config) if base.IsDocNotFoundError(err) { - return true, base.ErrConfigRegistryRollback, nil + return b.retryUnlessRegistryVersionChanged(ctx, bucketName, groupID, dbName, version, nil) } if err != nil { return false, err, nil @@ -501,7 +508,7 @@ func (b *bootstrapContext) getConfigVersionWithRetry(ctx context.Context, bucket } base.InfofCtx(ctx, base.KeyConfig, "getConfigVersionWithRetry for key %s found version mismatch, retrying. Requested: %s, Found: %s", metadataKey, version, config.Version) - return true, base.ErrConfigRegistryRollback, config + return b.retryUnlessRegistryVersionChanged(ctx, bucketName, groupID, dbName, version, config) } // Kick off the retry loop @@ -528,6 +535,21 @@ func (b *bootstrapContext) getConfigVersionWithRetry(ctx context.Context, bucket return nil, err } +// retryUnlessRegistryVersionChanged keeps waiting for an in-flight write of version while the registry still has it, +// and otherwise (for example after a concurrent delete) asks the caller to reload the registry. +func (b *bootstrapContext) retryUnlessRegistryVersionChanged(ctx context.Context, bucketName, groupID, dbName, version string, config *DatabaseConfig) (shouldRetry bool, err error, value any) { + registry, err := b.getGatewayRegistry(ctx, bucketName) + if err != nil { + return false, err, nil + } + registryDb, ok := registry.getRegistryDatabase(groupID, dbName) + if !ok || registryDb.IsDeleted() || registryDb.Version != version { + base.InfofCtx(ctx, base.KeyConfig, "Registry no longer has version %s for db %s while waiting for its config - reload required", version, base.MD(dbName)) + return false, base.ErrConfigRegistryReloadRequired, nil + } + return true, base.ErrConfigRegistryRollback, config +} + // getDatabaseConfig retrieves the database config, and enforces version match. On config not found or mismatched // version, will retry with backoff (to wait for in-flight updates to complete). After retry timeout, // triggers registry rollback and returns rollback error @@ -536,7 +558,11 @@ func (b *bootstrapContext) getDatabaseConfig(ctx context.Context, bucketName, gr ctx = b.addDatabaseLogContext(ctx, &DbConfig{Name: dbName}) config, err := b.getConfigVersionWithRetry(ctx, bucketName, groupID, dbName, version) if err != nil { - if err == base.ErrConfigRegistryRollback { + // A config newer than the registry means a concurrent update wrote both after the registry was read. + if errors.Is(err, base.ErrConfigVersionMismatch) { + return nil, base.ErrConfigRegistryReloadRequired + } + if errors.Is(err, base.ErrConfigRegistryRollback) { base.InfofCtx(ctx, base.KeyConfig, "Registry rollback required for bucket: %s, groupID: %s, dbName:%s", bucketName, groupID, dbName) rollbackErr := b.rollbackRegistry(ctx, bucketName, groupID, dbName, config, registry) // On successful registry rollback, caller needs reload registry diff --git a/rest/config_manager_test.go b/rest/config_manager_test.go index 30e2539ce4..8ff4dbc96a 100644 --- a/rest/config_manager_test.go +++ b/rest/config_manager_test.go @@ -10,7 +10,9 @@ package rest import ( "fmt" + "net/http" "testing" + "time" "github.com/couchbase/sync_gateway/base" "github.com/couchbase/sync_gateway/testing/assert" @@ -230,3 +232,49 @@ func TestLongMetadataID(t *testing.T) { assert.NotEqual(t, rehashMetadataID, longMetadataID) } + +// TestGetDatabaseConfigNewerThanRegistry verifies that reading a config written by a concurrent update after the +// registry was read asks the caller to reload the registry, rather than failing. +func TestGetDatabaseConfigNewerThanRegistry(t *testing.T) { + rt := NewRestTesterPersistentConfig(t) + defer rt.Close() + + ctx := rt.Context() + sc := rt.ServerContext() + bucketName := rt.Bucket().GetName() + groupID := sc.Config.Bootstrap.ConfigGroupID + + staleRegistry, staleConfig, err := sc.BootstrapContext.getRegistryAndDatabase(ctx, bucketName, groupID, "db") + require.NoError(t, err) + require.NotNil(t, staleConfig) + + dbConfig := rt.NewDbConfig() + dbConfig.RevsLimit = new(uint32(123)) + RequireStatus(t, rt.UpsertDbConfig("db", dbConfig), http.StatusCreated) + + _, err = sc.BootstrapContext.getDatabaseConfig(ctx, bucketName, groupID, "db", staleConfig.Version, staleRegistry) + require.ErrorIs(t, err, base.ErrConfigRegistryReloadRequired) +} + +// TestGetDatabaseConfigDeletedAfterRegistryRead verifies that reading a config deleted by a concurrent delete after +// the registry was read asks the caller to reload the registry, rather than waiting for the config to reappear. +func TestGetDatabaseConfigDeletedAfterRegistryRead(t *testing.T) { + rt := NewRestTesterPersistentConfig(t) + defer rt.Close() + + ctx := rt.Context() + sc := rt.ServerContext() + bucketName := rt.Bucket().GetName() + groupID := sc.Config.Bootstrap.ConfigGroupID + // Waiting for the deleted config makes the test hang until the go test timeout, instead of passing after a rollback. + sc.BootstrapContext.configRetryTimeout = 24 * time.Hour + + staleRegistry, staleConfig, err := sc.BootstrapContext.getRegistryAndDatabase(ctx, bucketName, groupID, "db") + require.NoError(t, err) + require.NotNil(t, staleConfig) + + RequireStatus(t, rt.SendAdminRequest(http.MethodDelete, "/db/", ""), http.StatusOK) + + _, err = sc.BootstrapContext.getDatabaseConfig(ctx, bucketName, groupID, "db", staleConfig.Version, staleRegistry) + require.ErrorIs(t, err, base.ErrConfigRegistryReloadRequired) +} diff --git a/rest/utilities_testing.go b/rest/utilities_testing.go index 4bbd03f252..b5bdfefa49 100644 --- a/rest/utilities_testing.go +++ b/rest/utilities_testing.go @@ -62,6 +62,7 @@ type RestTesterConfig struct { EnableUserQueries bool // Enable the feature-flag for user N1QL/etc queries CustomTestBucket *base.TestBucket // If set, use this bucket instead of requesting a new one. LeakyBucketConfig *base.LeakyBucketConfig // Set to create and use a leaky bucket on the RT and DB. A test bucket cannot be passed in if using this option. + ConnectToBucketFn db.OpenBucketFn // Replaces the function used to connect to buckets when loading databases. Cannot be used with LeakyBucketConfig. adminInterface string // adminInterface overrides the default admin interface. SgReplicateEnabled bool // SgReplicateManager disabled by default for RestTester ISGRSupportedBLIPSubprotocols []string // Forces the BLIP subprotocols used by this node's ISGR active replicators - see Bucket(). Not supported with PersistentConfig. @@ -230,6 +231,9 @@ func (rt *RestTester) Bucket() base.Bucket { rt.TB().Fatalf("A passed in TestBucket cannot be used on the RestTester when defining a LeakyBucketConfig") } rt.TestBucket = testBucket + if rt.ConnectToBucketFn != nil { + require.Nil(rt.TB(), rt.LeakyBucketConfig, "ConnectToBucketFn cannot be used on the RestTester with a LeakyBucketConfig") + } if rt.PersistentConfig != false { require.Zero(rt.TB(), rt.InitSyncSeq, "RestTesterConfig.InitSyncSeq is not supported with RestTesterConfig.PersistentConfig = true") @@ -309,8 +313,9 @@ func (rt *RestTester) Bucket() base.Bucket { rt.RestTesterServerContext = NewServerContext(base.TestCtx(rt.TB()), &sc, rt.RestTesterConfig.PersistentConfig) - _, isLeaky := base.AsLeakyBucket(rt.TestBucket) - if rt.LeakyBucketConfig != nil || isLeaky { + if rt.ConnectToBucketFn != nil { + rt.RestTesterServerContext.connectToBucketFn = rt.ConnectToBucketFn + } else if _, isLeaky := base.AsLeakyBucket(rt.TestBucket); rt.LeakyBucketConfig != nil || isLeaky { rt.RestTesterServerContext.connectToBucketFn = func(ctx context.Context, spec base.BucketSpec, failfast bool) (base.Bucket, error) { if spec.BucketName == testBucket.GetName() { return testBucket.NoCloseClone(), nil @@ -751,6 +756,10 @@ func (rt *RestTester) SendUserRequestWithHeaders(method, resource string, body s // templateResource is a non-fatal version of rt.mustTemplateResource func (rt *RestTester) templateResource(resource string) (string, error) { + // Skip the database lookup, which can race with a concurrent database delete, when there is nothing to template. + if !strings.Contains(resource, "{{") { + return resource, nil + } tmpl, err := template.New("urltemplate"). Option("missingkey=error"). Parse(resource) diff --git a/rest/utilities_testing_rest_tester_cluster.go b/rest/utilities_testing_rest_tester_cluster.go index fb35aeb946..3082dc03af 100644 --- a/rest/utilities_testing_rest_tester_cluster.go +++ b/rest/utilities_testing_rest_tester_cluster.go @@ -15,6 +15,7 @@ import ( "testing" "github.com/couchbase/sync_gateway/base" + "github.com/couchbase/sync_gateway/db" "github.com/couchbase/sync_gateway/testing/require" "github.com/google/uuid" ) @@ -83,6 +84,7 @@ func (rtc *RestTesterCluster) AddNode() *RestTester { PersistentConfig: true, CustomTestBucket: rtc.testBucket.NoCloseClone(), MutateStartupConfig: rtc.config.MutateStartupConfig, + ConnectToBucketFn: rtc.config.ConnectToBucketFn, LeakyBootstrapConnectionConfig: rtc.config.LeakyBootstrapConnectionConfigs[len(nodes)], } rt := NewRestTester(rtc.t, rtConfig) @@ -110,6 +112,7 @@ func (rtc *RestTesterCluster) Close(ctx context.Context) { type RestTesterClusterConfig struct { NumNodes uint8 // Number of RestTester objects to create MutateStartupConfig func(*StartupConfig) // Passes this option to the RestTesterConfig for each RestTester + ConnectToBucketFn db.OpenBucketFn // Passes this option to the RestTesterConfig for each RestTester // LeakyBootstrapConnectionConfigs maps a node index to the LeakyBootstrapConnectionConfig for that node, including nodes started by AddNode LeakyBootstrapConnectionConfigs map[int]*base.LeakyBootstrapConnectionConfig } @@ -142,6 +145,7 @@ func NewRestTesterCluster(t *testing.T, config *RestTesterClusterConfig) *RestTe PersistentConfig: true, CustomTestBucket: tb.NoCloseClone(), MutateStartupConfig: config.MutateStartupConfig, + ConnectToBucketFn: config.ConnectToBucketFn, LeakyBootstrapConnectionConfig: config.LeakyBootstrapConnectionConfigs[int(i)], } rt := NewRestTester(t, rtConfig)