CBG-5223: include both MetadataStore datastores in caching feed

Subscribe the caching DCP feed to both the primary (_system._mobile)
and fallback (_default._default) datastores when the metadata store is
a *base.MetadataStore, so _sync:* mutations are observed regardless of
which collection currently holds the doc during metadata migration.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bbrks

db/change_listener.go

@@ -86,20 +86,35 @@ func (listener *changeListener) OnDocChanged(event sgbucket.FeedEvent, docType D
 	listener.OnChangeCallback(event, docType)
 }
 
+// cachingFeedCollections returns the set of (scope, collection) pairs that the caching DCP
+// feed must subscribe to. When metadataStore is a *base.MetadataStore (active during the
+// metadata migration window), both its primary (_system._mobile) and fallback
+// (_default._default) datastores are included so that _sync:* mutations are observed
+// regardless of which collection currently holds the doc.
+func cachingFeedCollections(metadataStore base.DataStore, scopes map[string]Scope) base.CollectionNameSet {
+	collectionNames := base.NewCollectionNameSet()
+	if ms, ok := metadataStore.(*base.MetadataStore); ok {
+		collectionNames.Add(ms.Primary())
+		collectionNames.Add(ms.Fallback())
+	} else {
+		collectionNames.Add(metadataStore)
+	}
+	for scopeName, collections := range scopes {
+		for collectionName := range collections.Collections {
+			collectionNames.Add(sgbucket.DataStoreNameImpl{Scope: scopeName, Collection: collectionName})
+		}
+	}
+	return collectionNames
+}
+
 // Starts a changeListener on a given Bucket.
 func (listener *changeListener) Start(ctx context.Context, bucket base.Bucket, dbStats *expvar.Map, scopes map[string]Scope, metadataStore base.DataStore) error {
 
 	listener.terminator = make(chan bool)
 	listener.bucket = bucket
 	listener.bucketName = bucket.GetName()
 
-	collectionNames := base.NewCollectionNameSet()
-	collectionNames.Add(metadataStore)
-	for scopeName, collections := range scopes {
-		for collectionName := range collections.Collections {
-			collectionNames.Add(sgbucket.DataStoreNameImpl{Scope: scopeName, Collection: collectionName})
-		}
-	}
+	collectionNames := cachingFeedCollections(metadataStore, scopes)
 	listener.StartNotifierBroadcaster(ctx) // start broadcast changes goroutine
 
 	opts := base.DCPClientOptions{

db/change_listener_dual_metadata_test.go

@@ -0,0 +1,88 @@
+// 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 db
+
+import (
+	"testing"
+
+	"github.com/couchbase/sync_gateway/base"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestCachingFeedCollections_DualMetadataStore verifies that when the metadata store is a
+// *base.MetadataStore, the caching feed subscribes to both the primary (_system._mobile)
+// and fallback (_default._default) datastores so that _sync:* mutations in either
+// collection are observed during the metadata migration window.
+func TestCachingFeedCollections_DualMetadataStore(t *testing.T) {
+	ctx := base.TestCtx(t)
+	bucket := base.GetTestBucket(t)
+	defer bucket.Close(ctx)
+
+	primary := bucket.GetMobileSystemDataStore() // skips if backing store does not support system collections
+	fallback := bucket.DefaultDataStore()
+	ms := base.NewMetadataStore(primary, fallback)
+
+	got := cachingFeedCollections(ms, nil)
+
+	require.Contains(t, got, base.SystemScope, "expected %s scope to be present", base.SystemScope)
+	assert.Contains(t, got[base.SystemScope], base.SystemCollectionMobile)
+
+	require.Contains(t, got, base.DefaultScope, "expected %s scope to be present", base.DefaultScope)
+	assert.Contains(t, got[base.DefaultScope], base.DefaultCollection)
+}
+
+// TestCachingFeedCollections_SingleMetadataStore verifies that when the metadata store is a
+// plain DataStore (not the dual wrapper), the caching feed subscribes only to that store's
+// collection — preserving the pre-CBG-5223 behaviour.
+func TestCachingFeedCollections_SingleMetadataStore(t *testing.T) {
+	ctx := base.TestCtx(t)
+	bucket := base.GetTestBucket(t)
+	defer bucket.Close(ctx)
+
+	metadataStore := bucket.DefaultDataStore()
+
+	got := cachingFeedCollections(metadataStore, nil)
+
+	require.Contains(t, got, base.DefaultScope)
+	assert.Contains(t, got[base.DefaultScope], base.DefaultCollection)
+	// Should not have added any other scope (notably _system).
+	assert.Len(t, got, 1, "expected exactly one scope in single-metadata-store case, got %v", got)
+}
+
+// TestCachingFeedCollections_UserScopesIncluded verifies that user-data scopes/collections
+// are appended to the set regardless of whether the metadata store is a dual wrapper.
+func TestCachingFeedCollections_UserScopesIncluded(t *testing.T) {
+	ctx := base.TestCtx(t)
+	bucket := base.GetTestBucket(t)
+	defer bucket.Close(ctx)
+
+	primary := bucket.GetMobileSystemDataStore()
+	fallback := bucket.DefaultDataStore()
+	ms := base.NewMetadataStore(primary, fallback)
+
+	scopes := map[string]Scope{
+		"myScope": {
+			Collections: map[string]*DatabaseCollection{
+				"collA": nil,
+				"collB": nil,
+			},
+		},
+	}
+
+	got := cachingFeedCollections(ms, scopes)
+
+	require.Contains(t, got, base.SystemScope)
+	assert.Contains(t, got[base.SystemScope], base.SystemCollectionMobile)
+	require.Contains(t, got, base.DefaultScope)
+	assert.Contains(t, got[base.DefaultScope], base.DefaultCollection)
+	require.Contains(t, got, "myScope")
+	assert.Contains(t, got["myScope"], "collA")
+	assert.Contains(t, got["myScope"], "collB")
+}