diff --git a/rest/cluster_compat.go b/rest/cluster_compat.go index 88bdecaf09..51e63bab57 100644 --- a/rest/cluster_compat.go +++ b/rest/cluster_compat.go @@ -452,9 +452,10 @@ func (m *clusterCompatManager) Refresh(ctx context.Context) { // refreshNodeRegistrations iterates the tracked buckets, registers this node, and returns // the per-node version map and the aggregate freeze record across those registries. Returns -// an error if every tracked bucket failed so callers can leave the previously-cached state -// in place — stale is preferable to flipping the cluster compat version to nil on a transient -// bucket outage. +// an error only if every tracked bucket failed for a reason other than vanishing — stale cache +// is preferable to flipping the cluster compat version to nil on a transient bucket outage. If +// every tracked bucket vanished instead, that's reported as success with empty results, so the +// caller updates the cache to reflect that no buckets remain tracked. func (m *clusterCompatManager) refreshNodeRegistrations(ctx context.Context) (map[string]base.ClusterCompatVersion, *base.RegistryFreeze, map[string]*base.RegistryPreCCVAwareNode, *base.ClusterCompatVersion, error) { buckets := m.trackedBucketList() if len(buckets) == 0 { @@ -475,6 +476,7 @@ func (m *clusterCompatManager) refreshNodeRegistrations(ctx context.Context) (ma var aggregateFreeze *base.RegistryFreeze var aggregateHWM *base.ClusterCompatVersion succeeded := 0 + vanished := 0 for _, bucket := range buckets { _, ratchet := eligibleBuckets[bucket] preCCVAwarePeers := m.sc.observePreCCVAwarePeersForBucket(ctx, bucket) @@ -489,6 +491,12 @@ func (m *clusterCompatManager) refreshNodeRegistrations(ctx context.Context) (ma RatchetHWM: ratchet, }) if err != nil { + if errors.Is(err, errBucketDoesNotExist) { + base.InfofCtx(ctx, base.KeyConfig, "Bucket %s no longer exists in cluster; removing from cluster compatibility manager: %v", base.MD(bucket), err) + m.releaseBucket(bucket) + vanished++ + continue + } base.WarnfCtx(ctx, "Failed to register node version in bucket %s: %v", base.MD(bucket), err) continue } @@ -525,9 +533,11 @@ func (m *clusterCompatManager) refreshNodeRegistrations(ctx context.Context) (ma preCCVAwareMap[uuid] = &cp } } - if succeeded == 0 { + if succeeded == 0 && vanished < len(buckets) { return nil, nil, nil, nil, fmt.Errorf("no tracked bucket registries could be updated (%d tracked)", len(buckets)) } + // If every bucket vanished, the aggregates below are still empty/nil — correctly + // representing "no buckets tracked" for the cache. return nodeMap, aggregateFreeze, preCCVAwareMap, aggregateHWM, nil } diff --git a/rest/cluster_compat_test.go b/rest/cluster_compat_test.go index ed8ed4e680..54e3cc6352 100644 --- a/rest/cluster_compat_test.go +++ b/rest/cluster_compat_test.go @@ -9,6 +9,8 @@ package rest import ( + "context" + "errors" "fmt" "maps" "net/http" @@ -19,6 +21,8 @@ import ( "github.com/couchbase/sync_gateway/db" "github.com/couchbase/sync_gateway/testing/assert" "github.com/couchbase/sync_gateway/testing/require" + "github.com/couchbase/sync_gateway/testing/sgtest" + "github.com/couchbaselabs/rosmar" "golang.org/x/sync/errgroup" ) @@ -1878,3 +1882,145 @@ func TestMergeRegistryIntoCache_PreCCVAwarePeerLowerWins(t *testing.T) { assert.Equal(t, later, m.cachedPreCCVAwareNodes[peerUUID].LastObservedAt, "LastObservedAt must advance even when the higher reading is discarded") }) } + +func TestClusterCompatRefreshAndStopUntrackVanishedBucket(t *testing.T) { + ctx := base.TestCtx(t) + fakeBucket := "non_existent_fake_bucket_123" + + if sgtest.UnitTestUrlIsWalrus() { + defer func() { + // Under Rosmar, any read against fakeBucket auto-creates it. + // Make sure to delete it regardless of what happened in the test. + bucket, err := rosmar.OpenBucketIn(sgtest.UnitTestUrl(), fakeBucket, rosmar.CreateOrOpen) + require.NoError(t, err) + require.NoError(t, bucket.CloseAndDelete(ctx)) + }() + } + + rt := NewRestTesterPersistentConfig(t) + defer rt.Close() + + sc := rt.ServerContext() + clusterCompat := sc.ClusterCompat + require.NotNil(t, clusterCompat) + defer clusterCompat.Stop(ctx) + + bucketName := rt.Bucket().GetName() + + // Verify bucket is currently tracked and exists + require.Contains(t, clusterCompat.trackedBucketList(), bucketName) + exists, err := sc.BootstrapContext.bucketExists(ctx, bucketName) + require.NoError(t, err) + require.True(t, exists, "active bucket should exist") + + // Verify non-existent bucket doesn't exist + exists, err = sc.BootstrapContext.bucketExists(ctx, fakeBucket) + require.NoError(t, err) + require.False(t, exists, "non-existent bucket should not exist") + + // Track the fake bucket and confirm refreshNodeRegistrations doesn't hang or panic on a tracked + // bucket that no longer exists. Under Rosmar the read auto-creates fakeBucket (see above), so + // it's never classified as vanished and stays tracked. Under Couchbase Server the read fails + // for real, so it's classified as errBucketDoesNotExist and gets untracked. + clusterCompat.mu.Lock() + clusterCompat.trackedBuckets[fakeBucket] = struct{}{} + clusterCompat.mu.Unlock() + require.Contains(t, clusterCompat.trackedBucketList(), fakeBucket, "fake bucket should be tracked") + + _, _, _, _, err = clusterCompat.refreshNodeRegistrations(ctx) + require.NoError(t, err) + require.Contains(t, clusterCompat.trackedBucketList(), bucketName, "real bucket should remain tracked") + + if sgtest.UnitTestUrlIsWalrus() { + require.Contains(t, clusterCompat.trackedBucketList(), fakeBucket, "Rosmar auto-creates buckets on read, so the fake bucket is never classified as vanished") + } else { + require.NotContains(t, clusterCompat.trackedBucketList(), fakeBucket, "vanished bucket should be untracked") + } +} + +// fakeVanishingBootstrapConnection simulates a bucket that no longer exists in the cluster: +// GetMetadataDocument always fails, and any bucket absent from knownBuckets is then classified +// as errBucketDoesNotExist by getGatewayRegistry. Only the methods refreshNodeRegistrations +// actually calls are implemented; anything else panics via the nil embedded interface. +type fakeVanishingBootstrapConnection struct { + base.BootstrapConnection + knownBuckets []string +} + +func (f *fakeVanishingBootstrapConnection) GetConfigBuckets(_ context.Context) ([]string, error) { + return f.knownBuckets, nil +} + +func (f *fakeVanishingBootstrapConnection) GetMetadataDocument(_ context.Context, _, _ string, _ any) (uint64, error) { + return 0, errors.New("simulated: unable to read metadata document") +} + +// newTestClusterCompatManager builds a clusterCompatManager backed by a fake bootstrap +// connection, skipping the full RestTester/ServerContext setup — refreshNodeRegistrations only +// touches fields that are safe at their zero value here (NodeUID, Config, _databases(Lock)). +func newTestClusterCompatManager(knownBuckets []string, trackedBuckets ...string) *clusterCompatManager { + sc := &ServerContext{ + Config: &StartupConfig{}, + NodeUID: "test-node", + BootstrapContext: &bootstrapContext{ + Connection: &fakeVanishingBootstrapConnection{knownBuckets: knownBuckets}, + }, + } + tracked := make(map[string]struct{}, len(trackedBuckets)) + for _, b := range trackedBuckets { + tracked[b] = struct{}{} + } + return &clusterCompatManager{sc: sc, trackedBuckets: tracked} +} + +// TestClusterCompatRefreshNodeRegistrationsAllBucketsVanished verifies that when every tracked +// bucket vanishes in the same pass, refreshNodeRegistrations reports success with empty +// results rather than an error — otherwise Refresh would never clear the cache, since it stops +// calling refreshNodeRegistrations once trackedBuckets is empty. +func TestClusterCompatRefreshNodeRegistrationsAllBucketsVanished(t *testing.T) { + ctx := base.TestCtx(t) + m := newTestClusterCompatManager(nil, "vanished-bucket-1", "vanished-bucket-2") + + nodes, freeze, preCCVAwareNodes, hwm, err := m.refreshNodeRegistrations(ctx) + require.NoError(t, err, "all-vanished tracked buckets must not be reported as a failure") + assert.Empty(t, nodes) + assert.Nil(t, freeze) + assert.Empty(t, preCCVAwareNodes) + assert.Nil(t, hwm) + assert.Empty(t, m.trackedBucketList(), "vanished buckets must be untracked") +} + +// TestClusterCompatRefreshAllBucketsVanishedClearsCache verifies that Refresh actually clears a +// previously-cached, now-stale cluster compat version once every tracked bucket has vanished. +func TestClusterCompatRefreshAllBucketsVanishedClearsCache(t *testing.T) { + ctx := base.TestCtx(t) + m := newTestClusterCompatManager(nil, "vanished-bucket") + + staleVersion := base.NewClusterCompatVersion(3, 3) + m.cachedVersion = &staleVersion + m.cachedNodes = map[string]base.ClusterCompatVersion{"some-other-node": staleVersion} + + m.Refresh(ctx) + + assert.Nil(t, m.getCachedVersion(), "cache must be cleared once no buckets remain tracked") + assert.Empty(t, m.NodeVersions()) + assert.Empty(t, m.trackedBucketList()) +} + +// TestClusterCompatRefreshNodeRegistrationsPartialVanishKeepsError verifies the fix is scoped +// to the all-vanished case: if a vanished bucket is mixed with a bucket that fails for another +// reason, the result is still reported as an error so Refresh keeps the old cached state +// instead of caching an incomplete view. +func TestClusterCompatRefreshNodeRegistrationsPartialVanishKeepsError(t *testing.T) { + ctx := base.TestCtx(t) + // flaky-bucket exists (per GetConfigBuckets) but every read against it fails, so it's a + // real, non-vanished failure. vanished-bucket doesn't exist, so it's classified as vanished. + m := newTestClusterCompatManager([]string{"flaky-bucket"}, "vanished-bucket", "flaky-bucket") + + _, _, _, _, err := m.refreshNodeRegistrations(ctx) + require.Error(t, err, "a mix of vanished and non-vanished failures must still be reported as an error") + + tracked := m.trackedBucketList() + assert.NotContains(t, tracked, "vanished-bucket", "vanished bucket must still be untracked") + assert.Contains(t, tracked, "flaky-bucket", "non-vanished failures must not untrack the bucket") +} diff --git a/rest/config_manager.go b/rest/config_manager.go index c0f2daf5b2..8b866faaf0 100644 --- a/rest/config_manager.go +++ b/rest/config_manager.go @@ -12,6 +12,7 @@ import ( "context" "errors" "fmt" + "slices" "time" "github.com/couchbase/sync_gateway/base" @@ -647,6 +648,24 @@ func (b *bootstrapContext) rollbackRegistry(ctx context.Context, bucketName, gro return err } +// errBucketDoesNotExist wraps a getGatewayRegistry failure that was caused by the target +// bucket no longer existing in the cluster (as opposed to a transient I/O error), so callers can +// stop tracking/retrying the bucket via errors.Is instead of separately re-checking existence. +var errBucketDoesNotExist = errors.New("bucket no longer exists in cluster") + +// bucketExists returns whether the specified bucket exists in the cluster. Will return an error +// if communicating with Couchbase Server does not work. +func (b *bootstrapContext) bucketExists(ctx context.Context, bucketName string) (bool, error) { + if b.Connection == nil { + return false, errors.New("nil bootstrap connection") + } + buckets, err := b.Connection.GetConfigBuckets(ctx) + if err != nil { + return false, err + } + return slices.Contains(buckets, bucketName), nil +} + // getGatewayRegistry returns the database registry document for the bucket func (b *bootstrapContext) getGatewayRegistry(ctx context.Context, bucketName string) (result *GatewayRegistry, err error) { @@ -656,6 +675,10 @@ func (b *bootstrapContext) getGatewayRegistry(ctx context.Context, bucketName st if base.IsDocNotFoundError(getErr) { return NewGatewayRegistry(b.sgVersion), nil } + // A failure to list cluster buckets must not be misread as bucketName having vanished. + if exists, checkErr := b.bucketExists(ctx, bucketName); checkErr == nil && !exists { + return nil, fmt.Errorf("%w: %w", errBucketDoesNotExist, getErr) + } return nil, getErr } if registry.SGVersion.String() == "" { @@ -838,6 +861,10 @@ func (b *bootstrapContext) DeregisterNodeVersion(ctx context.Context, bucketName for attempt := 1; attempt <= nodeVersionUpdateMaxRetryAttempts; attempt++ { registry, err := b.getGatewayRegistry(ctx, bucketName) if err != nil { + if errors.Is(err, errBucketDoesNotExist) { + base.InfofCtx(ctx, base.KeyConfig, "DeregisterNodeVersion: Bucket %s no longer exists in cluster. Skipping deregistration.", base.MD(bucketName)) + return + } base.WarnfCtx(ctx, "Failed to get registry for node deregistration from bucket %s: %v", base.MD(bucketName), err) return }