From 3cd763135f2c20a5b9a4e87ffa503bff669371c5 Mon Sep 17 00:00:00 2001
From: Tor Colvin <tor.colvin@couchbase.com>
Date: Sun, 5 Jul 2026 22:05:57 -0400
Subject: [PATCH 1/7] CBG-5555 better fix for activeOnly+limit

---
 db/channel_cache_single.go |  14 +-
 db/channel_cache_test.go   | 285 +++++++++++++++++++++++++++++++++++++
 2 files changed, 296 insertions(+), 3 deletions(-)

diff --git a/db/channel_cache_single.go b/db/channel_cache_single.go
index 336d08ea63..40b4d80c36 100644
--- a/db/channel_cache_single.go
+++ b/db/channel_cache_single.go
@@ -440,14 +440,22 @@ func (c *singleChannelCacheImpl) GetChanges(ctx context.Context, options Changes
 
 	result := resultFromQuery
 	room := options.Limit - len(result)
-	if (options.Limit == 0 || room > 0) && len(resultFromCache) > 0 {
+	// For activeOnly+limit, only append cache when the query reached the cache boundary.
+	// If the query stopped early (gap exists before cacheValidFrom), the changesFeed pagination
+	// loop must fetch that range before cache results are valid.
+	// The overlap entry at cacheValidFrom is deduplicated below when cache is appended.
+	queryReachedCache := len(resultFromQuery) == 0 || resultFromQuery[len(resultFromQuery)-1].Sequence >= endSeq
+	if (options.Limit == 0 || room > 0 || (options.ActiveOnly && queryReachedCache)) && len(resultFromCache) > 0 {
 		// Concatenate the view & cache results:
 		if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
 			resultFromCache = resultFromCache[1:]
 		}
 		n := len(resultFromCache)
-		if options.Limit > 0 && room > 0 && room < n {
-			n = room
+		// Limit evaluation only valid when not activeOnly, since view and cache results don't apply activeOnly filtering
+		if !options.ActiveOnly {
+			if options.Limit > 0 && room > 0 && room < n {
+				n = room
+			}
 		}
 		result = append(result, resultFromCache[0:n]...)
 	}
diff --git a/db/channel_cache_test.go b/db/channel_cache_test.go
index b1ecd02ab8..8a475167b3 100644
--- a/db/channel_cache_test.go
+++ b/db/channel_cache_test.go
@@ -598,3 +598,288 @@ func TestChannelCacheBackgroundTaskWithIllegalTimeInterval(t *testing.T) {
 	assert.Equal(t, "CleanAgedItems", backgroundTaskError.TaskName)
 	assert.Equal(t, options.ChannelCacheAge, backgroundTaskError.Interval)
 }
+
+// - The channel cache is validFrom sequence n, with one active mutation resident in the cache
+// - There are channel removals (only) in the bucket with m < sequence < n
+// - Client issues a GetChanges request with since=m
+func TestChannelCacheActiveOnlyAndLimit(t *testing.T) {
+	ctx, db, collection := setupDBWithChannelCacheSize(t, 2)
+
+	const (
+		activeChannel   = "active"
+		inactiveChannel = "inactive"
+		doc1            = "doc1"
+		doc2            = "doc2"
+		doc3            = "doc3"
+	)
+
+	// doc1 rev1: channel active
+	// doc1 rev2: channel inactive
+	// doc2 rev1: channel active
+	// doc2 rev2: channel inactive
+	// doc3 rev1: channel active
+	revID, _, err := collection.Put(ctx, doc1, Body{"channels": activeChannel})
+	require.NoError(t, err)
+	_, _, err = collection.Put(ctx, doc1, Body{"channels": inactiveChannel, "_rev": revID})
+	require.NoError(t, err)
+	revID, _, err = collection.Put(ctx, doc2, Body{"channels": activeChannel})
+	require.NoError(t, err)
+	_, _, err = collection.Put(ctx, doc2, Body{"channels": inactiveChannel, "_rev": revID})
+	require.NoError(t, err)
+
+	_, _, err = collection.Put(ctx, doc3, Body{"channels": activeChannel})
+	require.NoError(t, err)
+
+	db.WaitForPendingChanges(t)
+
+	// prime channel cache, doc2 and doc3 should be in cache
+	changesOptions := ChangesOptions{
+		Since:      SequenceID{Seq: 0},
+		ActiveOnly: false,
+		ChangesCtx: base.TestCtx(t),
+	}
+	require.Len(t, getChanges(t, collection, base.SetOf(activeChannel), changesOptions), 3)
+
+	// whether limit or no limit, should only be 1 active entry
+	for _, limit := range []int{0, 1} {
+		t.Run("limit="+fmt.Sprint(limit), func(t *testing.T) {
+			changesOptions = ChangesOptions{
+				Since:      SequenceID{Seq: 0},
+				ActiveOnly: true,
+				ChangesCtx: base.TestCtx(t),
+				Limit:      limit,
+			}
+			require.Len(t, getChanges(t, collection, base.SetOf(activeChannel), changesOptions), 1)
+		})
+	}
+}
+
+func TestChannelCacheActiveOnlyScenarios(t *testing.T) {
+	const activeChannel = "active"
+
+	t.Run("query returns an active rev, cache is all removals", func(t *testing.T) {
+		ctx, db, collection := setupDBWithChannelCacheSize(t, 2)
+
+		// doc1: active (seq 1) - will be in backing store, not cache
+		_, _, _ = collection.Put(ctx, "doc1", Body{"channels": activeChannel})
+
+		// doc2: active (seq 2) -> inactive (seq 3) - seq 3 in cache
+		revID2, _, _ := collection.Put(ctx, "doc2", Body{"channels": activeChannel})
+		_, _, _ = collection.Put(ctx, "doc2", Body{"channels": "other", "_rev": revID2})
+
+		// doc3: active (seq 4) -> inactive (seq 5) - seq 5 in cache
+		revID3, _, _ := collection.Put(ctx, "doc3", Body{"channels": activeChannel})
+		_, _, _ = collection.Put(ctx, "doc3", Body{"channels": "other", "_rev": revID3})
+
+		db.WaitForPendingChanges(t)
+
+		// With limit 1 (before)
+		changesOptions := ChangesOptions{Since: SequenceID{Seq: 0}, ActiveOnly: true, Limit: 1, ChangesCtx: base.TestCtx(t)}
+		changes := getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 1)
+		assert.Equal(t, "doc1", changes[0].ID)
+
+		// No limit
+		changesOptions.Limit = 0
+		changes = getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 1)
+		assert.Equal(t, "doc1", changes[0].ID)
+
+		// With limit 1 (after)
+		changesOptions.Limit = 1
+		changes = getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 1)
+		assert.Equal(t, "doc1", changes[0].ID)
+	})
+
+	t.Run("query returns an active rev, cache also has an active rev", func(t *testing.T) {
+		ctx, db, collection := setupDBWithChannelCacheSize(t, 2)
+
+		// doc1: active (seq 1) - backing store
+		_, _, _ = collection.Put(ctx, "doc1", Body{"channels": activeChannel})
+
+		// doc2: active (seq 2) -> inactive (seq 3) - cache
+		revID2, _, _ := collection.Put(ctx, "doc2", Body{"channels": activeChannel})
+		_, _, _ = collection.Put(ctx, "doc2", Body{"channels": "other", "_rev": revID2})
+
+		// doc3: active (seq 4) - cache
+		_, _, _ = collection.Put(ctx, "doc3", Body{"channels": activeChannel})
+
+		db.WaitForPendingChanges(t)
+
+		// With limit 1 (before)
+		changesOptions := ChangesOptions{Since: SequenceID{Seq: 0}, ActiveOnly: true, Limit: 1, ChangesCtx: base.TestCtx(t)}
+		changes := getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 1)
+		assert.Equal(t, "doc1", changes[0].ID)
+
+		// No limit: should get doc1 and doc3
+		changesOptions.Limit = 0
+		changes = getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 2)
+		assert.Equal(t, "doc1", changes[0].ID)
+		assert.Equal(t, "doc3", changes[1].ID)
+
+		// With limit 1 (after)
+		changesOptions.Limit = 1
+		changes = getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 1)
+		assert.Equal(t, "doc1", changes[0].ID)
+	})
+
+	t.Run("query has no active revs, cache has no active revs", func(t *testing.T) {
+		ctx, db, collection := setupDBWithChannelCacheSize(t, 2)
+
+		// doc1: active (seq 1) -> inactive (seq 2)
+		revID1, _, _ := collection.Put(ctx, "doc1", Body{"channels": activeChannel})
+		_, _, _ = collection.Put(ctx, "doc1", Body{"channels": "other", "_rev": revID1})
+
+		// doc2: active (seq 3) -> inactive (seq 4)
+		revID2, _, _ := collection.Put(ctx, "doc2", Body{"channels": activeChannel})
+		_, _, _ = collection.Put(ctx, "doc2", Body{"channels": "other", "_rev": revID2})
+
+		db.WaitForPendingChanges(t)
+
+		// With limit 1 (before)
+		changesOptions := ChangesOptions{Since: SequenceID{Seq: 0}, ActiveOnly: true, Limit: 1, ChangesCtx: base.TestCtx(t)}
+		changes := getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 0)
+
+		// No limit: should get nothing
+		changesOptions.Limit = 0
+		changes = getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 0)
+
+		// With limit 1 (after)
+		changesOptions.Limit = 1
+		changes = getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 0)
+	})
+	t.Run("cache populated, query requires pagination", func(t *testing.T) {
+		cacheOptions := DefaultCacheOptions()
+		cacheOptions.ChannelCacheMaxLength = 5
+		cacheOptions.ChannelQueryLimit = 5
+		ctx, db, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+		// seed activeChannel in the cache prior to writing docs
+		changesOptions := ChangesOptions{Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t)}
+		_ = getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		// Write 20 docs to the channel.  5 should be cached, 15 require query
+		for i := range 20 {
+			docID := fmt.Sprintf("doc%d", i+1)
+			_, _, _ = collection.Put(ctx, docID, Body{"channels": activeChannel})
+		}
+		db.WaitForPendingChanges(t)
+		changesOptions = ChangesOptions{Since: SequenceID{Seq: 0}, ActiveOnly: true, Limit: 0, ChangesCtx: base.TestCtx(t)}
+		changes := getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 20)
+		for i, change := range changes {
+			assert.Equal(t, fmt.Sprintf("doc%d", i+1), change.ID, "change at index %d", i)
+		}
+	})
+	t.Run("cache populated, query requires pagination with limit", func(t *testing.T) {
+		// With ChannelQueryLimit=5 and Limit=10, each GetChanges call returns 5 entries from the
+		// query range.  Without the queryHitActiveLimit guard, the first call would append the
+		// cache (docs 16-20) immediately and skip docs 6-15.
+		cacheOptions := DefaultCacheOptions()
+		cacheOptions.ChannelCacheMaxLength = 5
+		cacheOptions.ChannelQueryLimit = 5
+		ctx, db, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+		changesOptions := ChangesOptions{Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t)}
+		_ = getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		for i := 1; i <= 20; i++ {
+			_, _, _ = collection.Put(ctx, fmt.Sprintf("doc%d", i), Body{"channels": activeChannel})
+		}
+		db.WaitForPendingChanges(t)
+		// Limit=10 spans two query batches (5+5) before reaching cache. The pagination loop must
+		// not prematurely append cache after the first batch hits the active limit.
+		changesOptions = ChangesOptions{Since: SequenceID{Seq: 0}, ActiveOnly: true, Limit: 10, ChangesCtx: base.TestCtx(t)}
+		changes := getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+		require.Len(t, changes, 10)
+		for i, change := range changes {
+			assert.Equal(t, fmt.Sprintf("doc%d", i+1), change.ID, "change at index %d", i)
+		}
+	})
+}
+
+func setupDBWithChannelCacheSize(t *testing.T, maxLength int) (context.Context, *Database, *DatabaseCollectionWithUser) {
+	cacheOptions := DefaultCacheOptions()
+	cacheOptions.ChannelCacheMaxLength = maxLength
+	return setupDBWithChannelCacheSettings(t, cacheOptions)
+}
+
+func setupDBWithChannelCacheSettings(t *testing.T, cacheOptions CacheOptions) (context.Context, *Database, *DatabaseCollectionWithUser) {
+	db, ctx := setupTestDBWithCacheOptions(t, cacheOptions)
+	t.Cleanup(func() { db.Close(ctx) })
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+	_, err := collection.UpdateSyncFun(ctx, channels.DocChannelsSyncFunction)
+	require.NoError(t, err)
+	return ctx, db, collection
+}
+
+// FuzzChannelCacheActiveOnly verifies that GetChanges with ActiveOnly=true returns every active entry
+// exactly once, in sequence order, regardless of how results are split between backing-store query
+// batches and the in-memory cache.
+func FuzzChannelCacheActiveOnly(f *testing.F) {
+	f.Add(uint8(20), uint8(5), uint8(5), uint8(10)) // query + cache, limit spans two batches
+	f.Add(uint8(5), uint8(5), uint8(5), uint8(0))   // all docs fit in cache, no limit
+	f.Add(uint8(1), uint8(5), uint8(5), uint8(1))   // single doc, limit=1
+	f.Add(uint8(0), uint8(5), uint8(5), uint8(5))   // no docs
+	f.Add(uint8(10), uint8(2), uint8(2), uint8(5))  // tiny cache + single-entry query batches
+	f.Add(uint8(15), uint8(5), uint8(5), uint8(5))  // limit < cache boundary
+
+	f.Fuzz(func(t *testing.T, numDocs, cacheMaxLength, queryLimit, requestLimit uint8) {
+		if cacheMaxLength == 0 {
+			cacheMaxLength = 1
+		}
+		if queryLimit == 0 {
+			queryLimit = 1
+		}
+		const maxDocs = 30
+		n := int(numDocs)
+		if n > maxDocs {
+			n = maxDocs
+		}
+
+		cacheOptions := DefaultCacheOptions()
+		cacheOptions.ChannelCacheMaxLength = int(cacheMaxLength)
+		cacheOptions.ChannelQueryLimit = int(queryLimit)
+		ctx, db, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+
+		// Seed the channel in the cache before writing docs so subsequent writes land
+		// in the cache from sequence 1.
+		_ = getChanges(t, collection, base.SetOf("active"), ChangesOptions{
+			Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t),
+		})
+		for i := 1; i <= n; i++ {
+			_, _, _ = collection.Put(ctx, fmt.Sprintf("doc%d", i), Body{"channels": "active"})
+		}
+		db.WaitForPendingChanges(t)
+
+		changes := getChanges(t, collection, base.SetOf("active"), ChangesOptions{
+			Since:      SequenceID{Seq: 0},
+			ActiveOnly: true,
+			Limit:      int(requestLimit),
+			ChangesCtx: base.TestCtx(t),
+		})
+
+		// No duplicates.
+		seen := make(map[string]bool, len(changes))
+		for _, c := range changes {
+			require.False(t, seen[c.ID], "duplicate entry %s", c.ID)
+			seen[c.ID] = true
+		}
+
+		// Correct count: all docs, or exactly requestLimit if that's the smaller constraint.
+		limit := int(requestLimit)
+		expectedCount := n
+		if limit > 0 && limit < n {
+			expectedCount = limit
+		}
+		require.Len(t, changes, expectedCount)
+
+		// Entries must be the first expectedCount docs in insertion order.
+		for i, c := range changes {
+			require.Equal(t, fmt.Sprintf("doc%d", i+1), c.ID, "wrong doc at index %d", i)
+		}
+	})
+}

From d98a36ada6e39fe4bc5bca5bdb1040ebb0ad4954 Mon Sep 17 00:00:00 2001
From: Tor Colvin <tor.colvin@couchbase.com>
Date: Mon, 6 Jul 2026 12:55:27 -0400
Subject: [PATCH 2/7] more tests

---
 db/changes.go              |   4 +-
 db/changes_test.go         |  55 ++++++
 db/changes_view.go         |  25 ++-
 db/channel_cache.go        |   5 +-
 db/channel_cache_single.go |  13 +-
 db/channel_cache_test.go   | 384 +++++++++++++++++++++++++++++++++++--
 db/database_test.go        |   6 +-
 db/query_test.go           |  16 +-
 db/util_testing.go         |   3 +-
 9 files changed, 464 insertions(+), 47 deletions(-)

diff --git a/db/changes.go b/db/changes.go
index aa95cf4d25..b73290ea53 100644
--- a/db/changes.go
+++ b/db/changes.go
@@ -534,7 +534,9 @@ func (db *DatabaseCollectionWithUser) changesFeed(ctx context.Context, singleCha
 					base.DebugfCtx(ctx, base.KeyChanges, "Terminating channel feed %s", base.UD(to))
 					return
 				case feed <- &change:
-					sentChanges++
+					if !options.ActiveOnly || (!change.Deleted && len(change.Removed) == 0) {
+						sentChanges++
+					}
 				}
 			}
 
diff --git a/db/changes_test.go b/db/changes_test.go
index af90dfb5f5..84a176b501 100644
--- a/db/changes_test.go
+++ b/db/changes_test.go
@@ -517,3 +517,58 @@ func TestCurrentVersionPopulationOnChannelCache(t *testing.T) {
 	assert.Equal(t, doc.HLV.SourceID, entries[0].SourceID)
 	assert.Equal(t, doc.HLV.Version, entries[0].Version)
 }
+
+// TestActiveOnlyWithLimit verifies that when querying with ActiveOnly: true and a Limit,
+// the pagination inside changesFeed does not terminate prematurely due to counting inactive/deleted
+// entries as "sent", ensuring the client receives the requested number of active changes when available.
+func TestActiveOnlyWithLimit(t *testing.T) {
+	cacheOptions := DefaultCacheOptions()
+	cacheOptions.ChannelQueryLimit = 3
+	db, ctx := setupTestDBWithCacheOptions(t, cacheOptions)
+	defer db.Close(ctx)
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+
+	base.SetUpTestLogging(t, base.LevelInfo, base.KeyChanges, base.KeyCache)
+
+	// 1. Create 6 documents that we will subsequently delete
+	revs := make(map[string]string)
+	for i := 1; i <= 6; i++ {
+		key := fmt.Sprintf("doc_del_%d", i)
+		body := Body{"foo": "bar"}
+		revId, _, err := collection.Put(ctx, key, body)
+		require.NoError(t, err)
+		revs[key] = revId
+	}
+
+	// 2. Delete those 6 documents so we have 6 deleted sequences at the start of the feed
+	for i := 1; i <= 6; i++ {
+		key := fmt.Sprintf("doc_del_%d", i)
+		_, _, err := collection.DeleteDoc(ctx, key, DocVersion{RevTreeID: revs[key]})
+		require.NoError(t, err)
+	}
+
+	// 3. Create 4 active documents
+	for i := 1; i <= 4; i++ {
+		key := fmt.Sprintf("doc_act_%d", i)
+		body := Body{"foo": "bar"}
+		_, _, err := collection.Put(ctx, key, body)
+		require.NoError(t, err)
+	}
+
+	db.WaitForPendingChanges(t)
+
+	// Get changes with active_only=true and Limit=3
+	changesOptions := ChangesOptions{
+		Since:      SequenceID{Seq: 0},
+		ActiveOnly: true,
+		Limit:      3,
+		ChangesCtx: base.TestCtx(t),
+	}
+
+	changes := getChanges(t, collection, base.SetOf("*"), changesOptions)
+	// We should receive exactly 3 active changes ("doc_act_1", "doc_act_2", "doc_act_3")
+	require.Len(t, changes, 3)
+	assert.Equal(t, "doc_act_1", changes[0].ID)
+	assert.Equal(t, "doc_act_2", changes[1].ID)
+	assert.Equal(t, "doc_act_3", changes[2].ID)
+}
diff --git a/db/changes_view.go b/db/changes_view.go
index bd683c1ed3..87fff30b66 100644
--- a/db/changes_view.go
+++ b/db/changes_view.go
@@ -90,14 +90,19 @@ func (dbc *DatabaseContext) getQueryHandlerForCollection(collectionID uint32) (C
 }
 
 // Queries the 'channels' view to get a range of sequences of a single channel as LogEntries.
-func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (LogEntries, error) {
+// The second return value, reachedEnd, is true if the query scanned all the way through to endSeq
+// without stopping early due to the activeOnly+limit active-entry count being satisfied first. When
+// reachedEnd is false, there may be additional matching entries between the last returned entry and
+// endSeq that were never scanned - it is unsafe to assume the range up to endSeq is fully covered.
+func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (entries LogEntries, reachedEnd bool, err error) {
 	if c.dataStore == nil {
-		return nil, errors.New("No data store available for channel query")
+		return nil, false, errors.New("No data store available for channel query")
 	}
 	start := time.Now()
 	usingViews := c.useViews()
-	entries := make(LogEntries, 0)
+	entries = make(LogEntries, 0)
 	activeEntryCount := 0
+	reachedEnd = true
 
 	base.InfofCtx(ctx, base.KeyCache, "  Querying 'channels' for %q (start=#%d, end=#%d, limit=%d)", base.UD(channelName), startSeq, endSeq, limit)
 
@@ -110,7 +115,7 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 		// Query the view or index
 		queryResults, err := c.QueryChannels(ctx, channelName, startSeq, endSeq, limit, activeOnly)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		queryRowCount := 0
 
@@ -145,7 +150,7 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 		// Close query results
 		closeErr := queryResults.Close(ctx)
 		if closeErr != nil {
-			return nil, closeErr
+			return nil, false, closeErr
 		}
 
 		if queryRowCount == 0 {
@@ -153,14 +158,16 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 				break
 			}
 			base.InfofCtx(ctx, base.KeyCache, "    Got no rows from query for channel:%q", base.UD(channelName))
-			return nil, nil
+			return nil, true, nil
 		}
 
 		// If active-only, loop until either retrieve (limit) active entries, or reach endSeq.  Non-active entries are still
 		// included in the result set for potential cache prepend
 		if activeOnly {
-			// If we've reached limit, we're done
-			if activeEntryCount >= limit || limit == 0 {
+			// If we've reached limit before exhausting the range up to endSeq, we're done, but the
+			// range beyond the last returned entry hasn't been fully scanned.
+			if limit != 0 && activeEntryCount >= limit {
+				reachedEnd = endSeq > 0 && highSeq >= endSeq
 				break
 			}
 			// If we've reached endSeq, we're done
@@ -186,5 +193,5 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 			elapsed, len(entries), base.UD(channelName), startSeq, endSeq, limit)
 	}
 	c.dbStats().Cache().ViewQueries.Add(1)
-	return entries, nil
+	return entries, reachedEnd, nil
 }
diff --git a/db/channel_cache.go b/db/channel_cache.go
index 8697cfeea6..0a269ca69f 100644
--- a/db/channel_cache.go
+++ b/db/channel_cache.go
@@ -80,7 +80,10 @@ var _ ChannelCache = &channelCacheImpl{}
 
 // ChannelQueryHandler interface is implemented by databaseContext and databaseCollection.
 type ChannelQueryHandler interface {
-	getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (LogEntries, error)
+	// getChangesInChannelFromQuery returns the LogEntries for a channel within [startSeq, endSeq], and
+	// reachedEnd, which is true if the query scanned all the way through to endSeq without stopping
+	// early due to satisfying an activeOnly+limit active-entry count first.
+	getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (entries LogEntries, reachedEnd bool, err error)
 }
 
 // Function that returns a ChannelQueryHandlerFunc for the specified collectionID
diff --git a/db/channel_cache_single.go b/db/channel_cache_single.go
index 40b4d80c36..10d524f84c 100644
--- a/db/channel_cache_single.go
+++ b/db/channel_cache_single.go
@@ -419,7 +419,7 @@ func (c *singleChannelCacheImpl) GetChanges(ctx context.Context, options Changes
 	// overlap, which helps confirm that we've got everything.
 	c.cacheStats.ChannelCacheMisses.Add(1)
 	endSeq := cacheValidFrom
-	resultFromQuery, err := c.queryHandler.getChangesInChannelFromQuery(ctx, c.channelID.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
+	resultFromQuery, queryReachedEnd, err := c.queryHandler.getChangesInChannelFromQuery(ctx, c.channelID.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
 	if err != nil {
 		return nil, err
 	}
@@ -440,12 +440,14 @@ func (c *singleChannelCacheImpl) GetChanges(ctx context.Context, options Changes
 
 	result := resultFromQuery
 	room := options.Limit - len(result)
-	// For activeOnly+limit, only append cache when the query reached the cache boundary.
+	// For activeOnly+limit, only append cache when the query reached the cache boundary
+	// (queryReachedEnd, reported by the query itself - not inferred from sequence numbers, since
+	// this channel's entries are sparse across the global sequence space and the last entry's
+	// sequence is often below endSeq even when the query fully scanned through to endSeq).
 	// If the query stopped early (gap exists before cacheValidFrom), the changesFeed pagination
 	// loop must fetch that range before cache results are valid.
 	// The overlap entry at cacheValidFrom is deduplicated below when cache is appended.
-	queryReachedCache := len(resultFromQuery) == 0 || resultFromQuery[len(resultFromQuery)-1].Sequence >= endSeq
-	if (options.Limit == 0 || room > 0 || (options.ActiveOnly && queryReachedCache)) && len(resultFromCache) > 0 {
+	if (options.Limit == 0 || room > 0 || (options.ActiveOnly && queryReachedEnd)) && len(resultFromCache) > 0 {
 		// Concatenate the view & cache results:
 		if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
 			resultFromCache = resultFromCache[1:]
@@ -849,7 +851,8 @@ type bypassChannelCache struct {
 func (b *bypassChannelCache) GetChanges(ctx context.Context, options ChangesOptions) ([]*LogEntry, error) {
 	startSeq := options.Since.SafeSequence() + 1
 	endSeq := uint64(math.MaxUint64)
-	return b.queryHandler.getChangesInChannelFromQuery(ctx, b.channel.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
+	entries, _, err := b.queryHandler.getChangesInChannelFromQuery(ctx, b.channel.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
+	return entries, err
 }
 
 // No cached changes for bypassChannelCache
diff --git a/db/channel_cache_test.go b/db/channel_cache_test.go
index 8a475167b3..61e61e400b 100644
--- a/db/channel_cache_test.go
+++ b/db/channel_cache_test.go
@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"log"
 	"math/rand"
+	"sort"
 	"sync"
 	"testing"
 	"time"
@@ -544,8 +545,9 @@ func (qh *testQueryHandler) asFactory(collectionID uint32) (ChannelQueryHandler,
 	return qh, nil
 }
 
-func (qh *testQueryHandler) getChangesInChannelFromQuery(ctx context.Context, channel string, startSeq, endSeq uint64, limit int, activeOnly bool) (LogEntries, error) {
+func (qh *testQueryHandler) getChangesInChannelFromQuery(ctx context.Context, channel string, startSeq, endSeq uint64, limit int, activeOnly bool) (LogEntries, bool, error) {
 	queryEntries := make(LogEntries, 0)
+	reachedEnd := true
 	qh.lock.RLock()
 	for _, entry := range qh.entries {
 		_, ok := entry.Channels[channel]
@@ -555,6 +557,7 @@ func (qh *testQueryHandler) getChangesInChannelFromQuery(ctx context.Context, ch
 			}
 			queryEntries = append(queryEntries, entry)
 			if limit > 0 && len(queryEntries) >= limit {
+				reachedEnd = endSeq > 0 && entry.Sequence >= endSeq
 				break
 			}
 		}
@@ -564,7 +567,7 @@ func (qh *testQueryHandler) getChangesInChannelFromQuery(ctx context.Context, ch
 	qh.lock.Lock()
 	qh.queryCount++
 	qh.lock.Unlock()
-	return queryEntries, nil
+	return queryEntries, reachedEnd, nil
 }
 
 func (qh *testQueryHandler) seedEntries(seededEntries LogEntries) {
@@ -820,12 +823,37 @@ func setupDBWithChannelCacheSettings(t *testing.T, cacheOptions CacheOptions) (c
 // exactly once, in sequence order, regardless of how results are split between backing-store query
 // batches and the in-memory cache.
 func FuzzChannelCacheActiveOnly(f *testing.F) {
-	f.Add(uint8(20), uint8(5), uint8(5), uint8(10)) // query + cache, limit spans two batches
-	f.Add(uint8(5), uint8(5), uint8(5), uint8(0))   // all docs fit in cache, no limit
-	f.Add(uint8(1), uint8(5), uint8(5), uint8(1))   // single doc, limit=1
-	f.Add(uint8(0), uint8(5), uint8(5), uint8(5))   // no docs
-	f.Add(uint8(10), uint8(2), uint8(2), uint8(5))  // tiny cache + single-entry query batches
-	f.Add(uint8(15), uint8(5), uint8(5), uint8(5))  // limit < cache boundary
+	// --- SEED CORPUS FOR EXPLICIT BOUNDARY CASES ---
+
+	// 1. request limit (R) aligns exactly with query pagination limit (Q) (R == Q, or R == k * Q)
+	f.Add(uint8(15), uint8(5), uint8(5), uint8(5))  // R == Q (both 5)
+	f.Add(uint8(20), uint8(5), uint8(5), uint8(10)) // R == 2 * Q (request 10, page size 5)
+	f.Add(uint8(10), uint8(2), uint8(2), uint8(4))  // R == 2 * Q with tiny limits (request 4, page size 2, cache size 2)
+
+	// 2. single query, no cache (cache validFrom is high, so all historical entries must be fetched from DB)
+	// (cacheMaxLength=1 means cache is effectively empty for other sequences; query limit is large enough to fetch all in one page)
+	f.Add(uint8(20), uint8(1), uint8(15), uint8(5))
+
+	// 3. Query gap before boundary, cache at boundary (query limit restricts DB page size, cacheMaxLength=1 keeps cache minimal)
+	f.Add(uint8(15), uint8(1), uint8(2), uint8(10)) // request limit 10, database page size 2, no cache
+
+	// 4. single query + partial cache (cacheMaxLength is large enough to hold some but not all docs, single DB query)
+	f.Add(uint8(10), uint8(5), uint8(10), uint8(8))
+
+	// 5. multiple queries + partial cache (cache holds part, DB pagination requested to get the rest over multiple pages)
+	f.Add(uint8(12), uint8(4), uint8(3), uint8(10))
+
+	// 6. single query + full cache (cacheMaxLength is large enough to hold all doc sequences, query limit is large)
+	f.Add(uint8(10), uint8(10), uint8(15), uint8(10))
+
+	// 7. multiple queries + full cache (cacheMaxLength is large, but query limit is small)
+	f.Add(uint8(15), uint8(10), uint8(3), uint8(12))
+
+	// 8. Other general permutations
+	f.Add(uint8(5), uint8(5), uint8(5), uint8(0))  // all docs fit in cache, no request limit
+	f.Add(uint8(1), uint8(5), uint8(5), uint8(1))  // single doc, limit=1
+	f.Add(uint8(0), uint8(5), uint8(5), uint8(5))  // no docs
+	f.Add(uint8(15), uint8(5), uint8(5), uint8(5)) // request limit < cache boundary
 
 	f.Fuzz(func(t *testing.T, numDocs, cacheMaxLength, queryLimit, requestLimit uint8) {
 		if cacheMaxLength == 0 {
@@ -850,9 +878,82 @@ func FuzzChannelCacheActiveOnly(f *testing.F) {
 		_ = getChanges(t, collection, base.SetOf("active"), ChangesOptions{
 			Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t),
 		})
+
+		type docState struct {
+			id     string
+			revID  string
+			seq    uint64
+			active bool
+		}
+
+		docsMap := make(map[string]*docState)
+		var docIDs []string
+
+		// Deterministic random generator based on inputs to ensure reproducibility
+		rng := rand.New(rand.NewSource(int64(numDocs) + int64(cacheMaxLength)*100 + int64(queryLimit)*10000 + int64(requestLimit)*1000000))
+
+		nextDocID := 1
 		for i := 1; i <= n; i++ {
-			_, _, _ = collection.Put(ctx, fmt.Sprintf("doc%d", i), Body{"channels": "active"})
+			// Choose action:
+			// 0: Create new active doc
+			// 1: Create new other (inactive) doc
+			// 2: Update existing to active
+			// 3: Update existing to other (inactive)
+			action := rng.Intn(4)
+			if len(docIDs) == 0 {
+				action = rng.Intn(2) // Only create actions are possible initially
+			}
+
+			var docID string
+			var body Body
+			var active bool
+
+			switch action {
+			case 0:
+				docID = fmt.Sprintf("doc_act_%d", nextDocID)
+				nextDocID++
+				body = Body{"channels": "active"}
+				active = true
+			case 1:
+				docID = fmt.Sprintf("doc_oth_%d", nextDocID)
+				nextDocID++
+				body = Body{"channels": "other"}
+				active = false
+			case 2:
+				// Update existing to active
+				idx := rng.Intn(len(docIDs))
+				docID = docIDs[idx]
+				state := docsMap[docID]
+				body = Body{"channels": "active", "_rev": state.revID}
+				active = true
+			case 3:
+				// Update existing to other
+				idx := rng.Intn(len(docIDs))
+				docID = docIDs[idx]
+				state := docsMap[docID]
+				body = Body{"channels": "other", "_rev": state.revID}
+				active = false
+			}
+
+			newRevID, doc, err := collection.Put(ctx, docID, body)
+			require.NoError(t, err)
+
+			if state, exists := docsMap[docID]; exists {
+				state.revID = newRevID
+				state.seq = doc.Sequence
+				state.active = active
+			} else {
+				docsMap[docID] = &docState{
+					id:     docID,
+					revID:  newRevID,
+					seq:    doc.Sequence,
+					active: active,
+				}
+				docIDs = append(docIDs, docID)
+				sort.Strings(docIDs) // maintain sorted order for deterministic selection
+			}
 		}
+
 		db.WaitForPendingChanges(t)
 
 		changes := getChanges(t, collection, base.SetOf("active"), ChangesOptions{
@@ -862,24 +963,269 @@ func FuzzChannelCacheActiveOnly(f *testing.F) {
 			ChangesCtx: base.TestCtx(t),
 		})
 
-		// No duplicates.
-		seen := make(map[string]bool, len(changes))
-		for _, c := range changes {
-			require.False(t, seen[c.ID], "duplicate entry %s", c.ID)
-			seen[c.ID] = true
+		// Collect the expected active docs at their final sequence numbers
+		var activeDocs []*docState
+		for _, state := range docsMap {
+			if state.active {
+				activeDocs = append(activeDocs, state)
+			}
 		}
+		sort.Slice(activeDocs, func(i, j int) bool {
+			return activeDocs[i].seq < activeDocs[j].seq
+		})
 
-		// Correct count: all docs, or exactly requestLimit if that's the smaller constraint.
+		// Correct count: all active docs, or exactly requestLimit if that is smaller
 		limit := int(requestLimit)
-		expectedCount := n
-		if limit > 0 && limit < n {
+		expectedCount := len(activeDocs)
+		if limit > 0 && limit < len(activeDocs) {
 			expectedCount = limit
 		}
+
+		// No duplicates in the returned changes
+		seen := make(map[string]bool, len(changes))
+		for _, c := range changes {
+			require.False(t, seen[c.ID], "duplicate entry %s", c.ID)
+			seen[c.ID] = true
+		}
 		require.Len(t, changes, expectedCount)
 
-		// Entries must be the first expectedCount docs in insertion order.
+		// Entries must be the first expectedCount active docs in final sequence order
 		for i, c := range changes {
-			require.Equal(t, fmt.Sprintf("doc%d", i+1), c.ID, "wrong doc at index %d", i)
+			require.Equal(t, activeDocs[i].id, c.ID, "wrong doc ID at index %d", i)
+			require.Equal(t, activeDocs[i].seq, c.Seq.Seq, "wrong sequence at index %d", i)
 		}
 	})
 }
+
+// TestChannelCacheActiveOnlyLimitWithCrossChannelGap reproduces a scenario where the cache's
+// validFrom sequence doesn't correspond to any entry in the queried channel, because the
+// intervening sequence(s) belong to a *different* channel. The query's last returned entry can
+// then be well below cacheValidFrom (e.g. N), while the cache's first entry starts above it
+// (e.g. N+2), even though the query fully scanned through to the cache boundary with no gap.
+func TestChannelCacheActiveOnlyLimitWithCrossChannelGap(t *testing.T) {
+	cacheOptions := DefaultCacheOptions()
+	cacheOptions.ChannelCacheMaxLength = 2
+	ctx, db, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+
+	const activeChannel = "active"
+	const otherChannel = "other"
+
+	// Prime the cache *before* any writes so subsequent docs are appended live (and pruned
+	// live via _pruneCacheLength), which is the path where validFrom can land on a sequence
+	// that isn't an entry in this channel at all (see _pruneCacheLength).
+	primingOptions := ChangesOptions{Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t)}
+	_ = getChanges(t, collection, base.SetOf(activeChannel), primingOptions)
+
+	// doc1: active (seq1) -> removed from channel (seq2)
+	revID, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel})
+	require.NoError(t, err)
+	_, _, err = collection.Put(ctx, "doc1", Body{"channels": otherChannel, "_rev": revID})
+	require.NoError(t, err)
+
+	// doc2: active (seq3) -> removed from channel (seq4)
+	revID, _, err = collection.Put(ctx, "doc2", Body{"channels": activeChannel})
+	require.NoError(t, err)
+	_, _, err = collection.Put(ctx, "doc2", Body{"channels": otherChannel, "_rev": revID})
+	require.NoError(t, err)
+
+	// docGap: seq5, entirely in a different channel - creates a sequence gap for activeChannel.
+	_, _, err = collection.Put(ctx, "docGap", Body{"channels": otherChannel})
+	require.NoError(t, err)
+
+	// doc3, doc4: active (seq6, seq7) - stay active. With ChannelCacheMaxLength=2 these live
+	// appends prune doc1/doc2's entries out of the cache, pushing validFrom to 5 (docGap's
+	// sequence), which isn't an entry in activeChannel at all.
+	_, _, err = collection.Put(ctx, "doc3", Body{"channels": activeChannel})
+	require.NoError(t, err)
+	_, _, err = collection.Put(ctx, "doc4", Body{"channels": activeChannel})
+	require.NoError(t, err)
+
+	db.WaitForPendingChanges(t)
+
+	changesOptions := ChangesOptions{
+		Since:      SequenceID{Seq: 0},
+		ActiveOnly: true,
+		Limit:      1,
+		ChangesCtx: base.TestCtx(t),
+	}
+	changes := getChanges(t, collection, base.SetOf(activeChannel), changesOptions)
+	require.Len(t, changes, 1)
+	assert.Equal(t, "doc3", changes[0].ID)
+}
+
+func TestChannelCacheActiveOnlyBoundariesAndGaps(t *testing.T) {
+	const activeChannel = "active"
+	const otherChannel = "other"
+
+	t.Run("Query and cache both exactly at boundary (No Gap)", func(t *testing.T) {
+		cacheOptions := DefaultCacheOptions()
+		cacheOptions.ChannelCacheMaxLength = 2
+		ctx, db, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+
+		// Prime cache
+		_ = getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t),
+		})
+
+		_, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel})
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc2", Body{"channels": activeChannel})
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc3", Body{"channels": activeChannel})
+		require.NoError(t, err)
+
+		db.WaitForPendingChanges(t)
+
+		// Since doc1 is pruned (cache length 2), validFrom is 2 (doc2's sequence is 2).
+		// Querying with Limit=10 should return all 3 docs: doc1 (Seq 1), doc2 (Seq 2), and doc3 (Seq 3).
+		// Since query reached boundary (Seq 2 >= 2), the cache should be appended, deduplicating doc2.
+		changes := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since:      SequenceID{Seq: 0},
+			ActiveOnly: true,
+			Limit:      10,
+			ChangesCtx: base.TestCtx(t),
+		})
+		require.Len(t, changes, 3)
+		assert.Equal(t, "doc1", changes[0].ID)
+		assert.Equal(t, "doc2", changes[1].ID)
+		assert.Equal(t, "doc3", changes[2].ID)
+	})
+
+	t.Run("Query gap before boundary, cache at boundary", func(t *testing.T) {
+		cacheOptions := DefaultCacheOptions()
+		cacheOptions.ChannelCacheMaxLength = 2
+		ctx, db, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+
+		// Prime cache
+		_ = getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t),
+		})
+
+		_, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel}) // Seq 1
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "docGap", Body{"channels": otherChannel}) // Seq 2
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc2", Body{"channels": activeChannel}) // Seq 3
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc3", Body{"channels": activeChannel}) // Seq 4
+		require.NoError(t, err)
+
+		db.WaitForPendingChanges(t)
+
+		// doc1 is pruned, cache contains doc2 and doc3. validFrom is 2 (doc1.Seq + 1 = 2).
+		// Querying with Limit=1 should get doc1 (Seq 1) and stop early because limit=1 is met before Seq 2 (validFrom).
+		// Therefore reachedEnd is false, cache is not appended, only doc1 is returned.
+		changes := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since:      SequenceID{Seq: 0},
+			ActiveOnly: true,
+			Limit:      1,
+			ChangesCtx: base.TestCtx(t),
+		})
+		require.Len(t, changes, 1)
+		assert.Equal(t, "doc1", changes[0].ID)
+	})
+
+	t.Run("Query at boundary, cache gap after boundary", func(t *testing.T) {
+		cacheOptions := DefaultCacheOptions()
+		cacheOptions.ChannelCacheMaxLength = 2
+		ctx, db, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+
+		// Prime cache
+		_ = getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t),
+		})
+
+		_, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel}) // Seq 1
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc2", Body{"channels": activeChannel}) // Seq 2
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "docGap", Body{"channels": otherChannel}) // Seq 3
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc3", Body{"channels": activeChannel}) // Seq 4
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc4", Body{"channels": activeChannel}) // Seq 5
+		require.NoError(t, err)
+
+		db.WaitForPendingChanges(t)
+
+		// doc1 and doc2 are pruned. Cache contains doc3 and doc4. validFrom is 3 (doc2.Seq + 1 = 3).
+		// Querying with Limit=2 gets doc1 (Seq 1) and doc2 (Seq 2) and stops early (highSeq = 2 < 3).
+		// Therefore reachedEnd is false, cache is not appended.
+		changes := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since:      SequenceID{Seq: 0},
+			ActiveOnly: true,
+			Limit:      2,
+			ChangesCtx: base.TestCtx(t),
+		})
+		require.Len(t, changes, 2)
+		assert.Equal(t, "doc1", changes[0].ID)
+		assert.Equal(t, "doc2", changes[1].ID)
+
+		// Querying with Limit=10 gets doc1, doc2, and the query iterator finishes normally at Seq 3 (no other active).
+		// Thus reachedEnd is true, cache is appended, and limit=10 allows all 4 to be returned.
+		changes10 := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since:      SequenceID{Seq: 0},
+			ActiveOnly: true,
+			Limit:      10,
+			ChangesCtx: base.TestCtx(t),
+		})
+		require.Len(t, changes10, 4)
+		assert.Equal(t, "doc1", changes10[0].ID)
+		assert.Equal(t, "doc2", changes10[1].ID)
+		assert.Equal(t, "doc3", changes10[2].ID)
+		assert.Equal(t, "doc4", changes10[3].ID)
+	})
+
+	t.Run("Gaps in both directions", func(t *testing.T) {
+		cacheOptions := DefaultCacheOptions()
+		cacheOptions.ChannelCacheMaxLength = 2
+		ctx, db, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+
+		// Prime cache
+		_ = getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since: SequenceID{Seq: 0}, ChangesCtx: base.TestCtx(t),
+		})
+
+		_, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel}) // Seq 1
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "docGap1", Body{"channels": otherChannel}) // Seq 2
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc2", Body{"channels": activeChannel}) // Seq 3
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "docGap2", Body{"channels": otherChannel}) // Seq 4
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc3", Body{"channels": activeChannel}) // Seq 5
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "doc4", Body{"channels": activeChannel}) // Seq 6
+		require.NoError(t, err)
+
+		db.WaitForPendingChanges(t)
+
+		// doc1 and doc2 are pruned. Cache contains doc3 and doc4. validFrom is 4 (doc2.Seq + 1 = 4).
+		// Querying with Limit=2 gets doc1 (Seq 1) and doc2 (Seq 3) and stops early (highSeq = 3 < 4).
+		// Thus reachedEnd is false, cache is not appended.
+		changes := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since:      SequenceID{Seq: 0},
+			ActiveOnly: true,
+			Limit:      2,
+			ChangesCtx: base.TestCtx(t),
+		})
+		require.Len(t, changes, 2)
+		assert.Equal(t, "doc1", changes[0].ID)
+		assert.Equal(t, "doc2", changes[1].ID)
+
+		// Querying with Limit=10 gets doc1, doc2, and query iterator reaches endSeq normally (Seq 4 reached).
+		// Thus reachedEnd is true, cache is appended, and limit=10 allows all 4 to be returned.
+		changes10 := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
+			Since:      SequenceID{Seq: 0},
+			ActiveOnly: true,
+			Limit:      10,
+			ChangesCtx: base.TestCtx(t),
+		})
+		require.Len(t, changes10, 4)
+		assert.Equal(t, "doc1", changes10[0].ID)
+		assert.Equal(t, "doc2", changes10[1].ID)
+		assert.Equal(t, "doc3", changes10[2].ID)
+		assert.Equal(t, "doc4", changes10[3].ID)
+	})
+}
diff --git a/db/database_test.go b/db/database_test.go
index dd37046176..8350ce3173 100644
--- a/db/database_test.go
+++ b/db/database_test.go
@@ -2975,7 +2975,7 @@ func TestChannelView(t *testing.T) {
 	// Query view (retry loop to wait for indexing)
 	for i := range 10 {
 		var err error
-		entries, err = collection.getChangesInChannelFromQuery(ctx, "*", 0, 100, 0, false)
+		entries, _, err = collection.getChangesInChannelFromQuery(ctx, "*", 0, 100, 0, false)
 
 		assert.NoError(t, err, "Couldn't create document")
 		if len(entries) >= 1 {
@@ -3048,7 +3048,7 @@ func TestChannelQuery(t *testing.T) {
 
 	for _, testCase := range testCases {
 		t.Run(testCase.testName, func(t *testing.T) {
-			entries, err = collection.getChangesInChannelFromQuery(ctx, testCase.channelName, 0, 100, 0, false)
+			entries, _, err = collection.getChangesInChannelFromQuery(ctx, testCase.channelName, 0, 100, 0, false)
 			require.NoError(t, err)
 
 			for i, entry := range entries {
@@ -3132,7 +3132,7 @@ func TestChannelQueryRevocation(t *testing.T) {
 
 	for _, testCase := range testCases {
 		t.Run(testCase.testName, func(t *testing.T) {
-			entries, err = collection.getChangesInChannelFromQuery(ctx, testCase.channelName, 0, 100, 0, false)
+			entries, _, err = collection.getChangesInChannelFromQuery(ctx, testCase.channelName, 0, 100, 0, false)
 			require.NoError(t, err)
 
 			for i, entry := range entries {
diff --git a/db/query_test.go b/db/query_test.go
index f704e4edee..5fbb3052e6 100644
--- a/db/query_test.go
+++ b/db/query_test.go
@@ -486,49 +486,49 @@ func TestQueryChannelsActiveOnlyWithLimit(t *testing.T) {
 
 	// Get changes from channel "ABC" with limit and activeOnly true
 
-	entries, err := collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 25, true)
+	entries, _, err := collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 25, true)
 	require.NoError(t, err, "Couldn't query active docs from channel ABC with limit")
 	require.Len(t, entries, 25)
 	checkFlags(entries)
 
 	// Get changes from channel "*" with limit and activeOnly true
-	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 25, true)
+	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 25, true)
 	require.NoError(t, err, "Couldn't query active docs from channel * with limit")
 	require.Len(t, entries, 25)
 	checkFlags(entries)
 
 	// Get changes from channel "ABC" without limit and activeOnly true
-	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 0, true)
+	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 0, true)
 	require.NoError(t, err, "Couldn't query active docs from channel ABC with limit")
 	require.Len(t, entries, 30)
 	checkFlags(entries)
 
 	// Get changes from channel "*" without limit and activeOnly true
-	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 0, true)
+	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 0, true)
 	require.NoError(t, err, "Couldn't query active docs from channel * with limit")
 	require.Len(t, entries, 30)
 	checkFlags(entries)
 
 	// Get changes from channel "ABC" with limit and activeOnly false
-	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 45, false)
+	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 45, false)
 	require.NoError(t, err, "Couldn't query active docs from channel ABC with limit")
 	require.Len(t, entries, 45)
 	checkFlags(entries)
 
 	// Get changes from channel "*" with limit and activeOnly false
-	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 45, false)
+	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 45, false)
 	require.NoError(t, err, "Couldn't query active docs from channel * with limit")
 	require.Len(t, entries, 45)
 	checkFlags(entries)
 
 	// Get changes from channel "ABC" without limit and activeOnly false
-	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 0, false)
+	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 0, false)
 	require.NoError(t, err, "Couldn't query active docs from channel ABC with limit")
 	require.Len(t, entries, 50)
 	checkFlags(entries)
 
 	// Get changes from channel "*" without limit and activeOnly true
-	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 0, false)
+	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 0, false)
 	require.NoError(t, err, "Couldn't query active docs from channel * with limit")
 	require.Len(t, entries, 50)
 	checkFlags(entries)
diff --git a/db/util_testing.go b/db/util_testing.go
index 15fc0ffe45..0e7630171a 100644
--- a/db/util_testing.go
+++ b/db/util_testing.go
@@ -443,7 +443,8 @@ func (dbc *DatabaseContext) ChannelViewForTest(tb testing.TB, channelName string
 }
 
 func (dbc *DatabaseContext) CollectionChannelViewForTest(tb testing.TB, collection *DatabaseCollection, channelName string, startSeq, endSeq uint64) (LogEntries, error) {
-	return collection.getChangesInChannelFromQuery(base.TestCtx(tb), channelName, startSeq, endSeq, 0, false)
+	entries, _, err := collection.getChangesInChannelFromQuery(base.TestCtx(tb), channelName, startSeq, endSeq, 0, false)
+	return entries, err
 }
 
 // Test-only version of GetPrincipal that doesn't trigger channel/role recalculation

From cd26067a8e6820ec325fd4b7515d67c9aea57042 Mon Sep 17 00:00:00 2001
From: Tor Colvin <tor.colvin@couchbase.com>
Date: Mon, 6 Jul 2026 14:24:45 -0400
Subject: [PATCH 3/7] add fixes

---
 db/changes_view.go       |  12 ++-
 db/channel_cache_test.go |  53 +++++++++++++
 db/query_test.go         | 157 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 219 insertions(+), 3 deletions(-)

diff --git a/db/changes_view.go b/db/changes_view.go
index 87fff30b66..62d1fd2df7 100644
--- a/db/changes_view.go
+++ b/db/changes_view.go
@@ -164,10 +164,16 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 		// If active-only, loop until either retrieve (limit) active entries, or reach endSeq.  Non-active entries are still
 		// included in the result set for potential cache prepend
 		if activeOnly {
-			// If we've reached limit before exhausting the range up to endSeq, we're done, but the
-			// range beyond the last returned entry hasn't been fully scanned.
+			// If we've reached limit before exhausting the range up to endSeq, we're done.  Whether
+			// the range beyond the last returned entry has been fully scanned can't be determined by
+			// comparing highSeq to endSeq alone - channels are often sparse, so the last real entry
+			// is frequently well below endSeq even when nothing was missed.  Instead, check whether
+			// this query call itself returned fewer rows than requested: if so, the store had no more
+			// matching entries anywhere in [startSeq, endSeq], regardless of where highSeq landed. If
+			// the call returned exactly `limit` rows, the LIMIT clause may have truncated further
+			// matches, so fall back to the highSeq/endSeq comparison.
 			if limit != 0 && activeEntryCount >= limit {
-				reachedEnd = endSeq > 0 && highSeq >= endSeq
+				reachedEnd = queryRowCount < limit || (endSeq > 0 && highSeq >= endSeq)
 				break
 			}
 			// If we've reached endSeq, we're done
diff --git a/db/channel_cache_test.go b/db/channel_cache_test.go
index 61e61e400b..ba1af4fdac 100644
--- a/db/channel_cache_test.go
+++ b/db/channel_cache_test.go
@@ -802,6 +802,59 @@ func TestChannelCacheActiveOnlyScenarios(t *testing.T) {
 			assert.Equal(t, fmt.Sprintf("doc%d", i+1), change.ID, "change at index %d", i)
 		}
 	})
+	t.Run("query exhausts channel before endSeq, reachedEnd must be true", func(t *testing.T) {
+		// doc1: active - contributes a single active row to the channel log.
+		// docA: active, then removed from the channel - the add is superseded by the removal,
+		// so it contributes a single non-active (removal) row, not two rows.
+		// docB: active - the only entry remaining in the channel after docA's removal.
+		// Nothing else exists in the channel all the way out to endSeq, so a query that reaches
+		// docB should report reachedEnd=true even though docB's sequence is far below endSeq -
+		// the gap up to endSeq was fully (if implicitly) scanned, not skipped.
+		ctx, db, collection := setupDBWithChannelCacheSize(t, 2)
+
+		_, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel})
+		require.NoError(t, err)
+		revID, _, err := collection.Put(ctx, "docA", Body{"channels": activeChannel})
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "docA", Body{"channels": "other", "_rev": revID})
+		require.NoError(t, err)
+		_, _, err = collection.Put(ctx, "docB", Body{"channels": activeChannel})
+		require.NoError(t, err)
+		db.WaitForPendingChanges(t)
+
+		// limit=2 forces pagination: the first query call returns doc1's active row and docA's
+		// removal row (2 rows, hitting the row cap) with only 1 active entry seen so far; the
+		// second call returns just docB's active row (1 row, under the row cap), pushing
+		// activeEntryCount to the limit. endSeq is set far beyond any real sequence to exercise
+		// the sparse-channel case.
+		entries, reachedEnd, err := collection.getChangesInChannelFromQuery(ctx, activeChannel, 0, 1000, 2, true)
+		require.NoError(t, err)
+		require.Len(t, entries, 3)
+		assert.True(t, reachedEnd, "query fully exhausted the channel and should report reachedEnd=true")
+	})
+	t.Run("query call lands exactly on limit at the true end of channel, reachedEnd stays conservatively false", func(t *testing.T) {
+		// This pins a deliberate, documented limitation rather than a bug: a single active doc is
+		// the only entry in the channel, and limit=1 exactly matches the row count that call
+		// returns. From the row count alone there's no way to distinguish "the LIMIT clause
+		// truncated a real remaining entry" from "the store coincidentally had exactly `limit`
+		// matching rows" - so reachedEnd must stay conservatively false here, per
+		// getChangesInChannelFromQuery's doc comment ("false" means unknown/maybe, not "definitely
+		// more"). The CBG-5555 fix only resolves the *unambiguous* case, where a query call returns
+		// fewer rows than requested (see the sibling subtest above) - it does not, and cannot
+		// without an extra probe query, resolve this exact-match ambiguity. If this test ever
+		// starts asserting reachedEnd=true here, it means the row-count heuristic changed in a way
+		// that may no longer be safe - verify it can't produce false positives before updating it.
+		ctx, db, collection := setupDBWithChannelCacheSize(t, 2)
+
+		_, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel})
+		require.NoError(t, err)
+		db.WaitForPendingChanges(t)
+
+		entries, reachedEnd, err := collection.getChangesInChannelFromQuery(ctx, activeChannel, 0, 1000, 1, true)
+		require.NoError(t, err)
+		require.Len(t, entries, 1)
+		assert.False(t, reachedEnd, "query call was row-capped exactly at limit; reachedEnd must stay conservatively false")
+	})
 }
 
 func setupDBWithChannelCacheSize(t *testing.T, maxLength int) (context.Context, *Database, *DatabaseCollectionWithUser) {
diff --git a/db/query_test.go b/db/query_test.go
index 5fbb3052e6..9bad946125 100644
--- a/db/query_test.go
+++ b/db/query_test.go
@@ -13,6 +13,8 @@ package db
 import (
 	"context"
 	"fmt"
+	"math/rand"
+	"sort"
 	"strconv"
 	"testing"
 
@@ -534,6 +536,161 @@ func TestQueryChannelsActiveOnlyWithLimit(t *testing.T) {
 	checkFlags(entries)
 }
 
+// FuzzGetChangesInChannelFromQueryActiveOnly exercises getChangesInChannelFromQuery's activeOnly+limit
+// pagination directly (the layer TestQueryChannelsActiveOnlyWithLimit covers, but only against
+// Couchbase Server/N1QL) against the default Rosmar/views backing store. It checks that the returned
+// active entries form a correctly ordered, duplicate-free prefix of the true active set, and that
+// reachedEnd never falsely claims the range was fully scanned (reachedEnd=false is allowed to be
+// conservative - see the doc comment on getChangesInChannelFromQuery - but reachedEnd=true is a hard
+// promise that guards a consumer from silently dropping entries it assumes were already covered).
+// This doesn't pin down the specific reachedEnd fix from CBG-5555 (an unnecessary but safe false
+// negative) - see TestChannelCacheActiveOnlyScenarios's "query exhausts channel before endSeq"
+// subtest in channel_cache_test.go for that regression coverage, engineered to land in the
+// unambiguous case; a black-box fuzz property can't reliably distinguish that from the inherent,
+// accepted ambiguity when a query call happens to return exactly `limit` rows.
+func FuzzGetChangesInChannelFromQueryActiveOnly(f *testing.F) {
+	f.Add(uint8(0), uint8(5))
+	f.Add(uint8(1), uint8(1))
+	f.Add(uint8(5), uint8(0)) // no limit
+	f.Add(uint8(10), uint8(3))
+	f.Add(uint8(20), uint8(7))
+	f.Add(uint8(30), uint8(25))
+	f.Add(uint8(30), uint8(2))
+
+	f.Fuzz(func(t *testing.T, numDocs, limit uint8) {
+		const maxDocs = 30
+		n := min(int(numDocs), maxDocs)
+
+		db, ctx := setupTestDBAllowConflicts(t)
+		defer db.Close(ctx)
+		collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+		collection.ChannelMapper = channels.NewChannelMapper(ctx, channels.DocChannelsSyncFunction, db.Options.JavascriptTimeout)
+
+		const channelName = "ABC"
+
+		type docState struct {
+			id     string
+			revID  string
+			seq    uint64
+			active bool
+		}
+		docsMap := make(map[string]*docState)
+		var docIDs []string
+		nextDocID := 1
+
+		// Deterministic random generator based on inputs to ensure reproducibility
+		rng := rand.New(rand.NewSource(int64(numDocs)*1000 + int64(limit)))
+
+		var endSeq uint64
+		for i := 1; i <= n; i++ {
+			// Choose action:
+			// 0: Create new active doc
+			// 1: Create new other (inactive) doc
+			// 2: Update existing to active
+			// 3: Update existing to other (inactive)
+			action := rng.Intn(4)
+			if len(docIDs) == 0 {
+				action = rng.Intn(2) // Only create actions are possible initially
+			}
+
+			var docID string
+			var body Body
+			var active bool
+			switch action {
+			case 0:
+				docID = fmt.Sprintf("doc_act_%d", nextDocID)
+				nextDocID++
+				body = Body{"channels": channelName}
+				active = true
+			case 1:
+				docID = fmt.Sprintf("doc_oth_%d", nextDocID)
+				nextDocID++
+				body = Body{"channels": "other"}
+				active = false
+			case 2:
+				idx := rng.Intn(len(docIDs))
+				docID = docIDs[idx]
+				body = Body{"channels": channelName, "_rev": docsMap[docID].revID}
+				active = true
+			case 3:
+				idx := rng.Intn(len(docIDs))
+				docID = docIDs[idx]
+				body = Body{"channels": "other", "_rev": docsMap[docID].revID}
+				active = false
+			}
+
+			newRevID, doc, err := collection.Put(ctx, docID, body)
+			require.NoError(t, err)
+			endSeq = doc.Sequence
+
+			if state, exists := docsMap[docID]; exists {
+				state.revID = newRevID
+				state.seq = doc.Sequence
+				state.active = active
+			} else {
+				docsMap[docID] = &docState{id: docID, revID: newRevID, seq: doc.Sequence, active: active}
+				docIDs = append(docIDs, docID)
+			}
+		}
+
+		if n == 0 {
+			return
+		}
+
+		entries, reachedEnd, err := collection.getChangesInChannelFromQuery(ctx, channelName, 0, endSeq, int(limit), true)
+		require.NoError(t, err)
+
+		// Ground truth: active docs at their final sequence, in sequence order.
+		var activeDocs []*docState
+		for _, state := range docsMap {
+			if state.active {
+				activeDocs = append(activeDocs, state)
+			}
+		}
+		sort.Slice(activeDocs, func(i, j int) bool { return activeDocs[i].seq < activeDocs[j].seq })
+
+		// entries may include non-active rows (kept for potential cache prepend) - filter to active
+		// entries for comparison against the true active list, and separately check overall ordering.
+		var highSeq uint64
+		seen := make(map[string]bool, len(entries))
+		var activeReturned []*LogEntry
+		for _, e := range entries {
+			require.False(t, seen[e.DocID], "duplicate entry %s", e.DocID)
+			seen[e.DocID] = true
+			require.GreaterOrEqual(t, e.Sequence, highSeq, "entries must be returned in ascending sequence order")
+			highSeq = e.Sequence
+			if e.IsActive() {
+				activeReturned = append(activeReturned, e)
+			}
+		}
+
+		expectedMin := len(activeDocs)
+		if int(limit) > 0 && int(limit) < expectedMin {
+			expectedMin = int(limit)
+		}
+		require.GreaterOrEqual(t, len(activeReturned), expectedMin,
+			"expected at least %d active entries (limit=%d, total active=%d)", expectedMin, limit, len(activeDocs))
+
+		for i, e := range activeReturned {
+			require.Equal(t, activeDocs[i].id, e.DocID, "wrong doc ID at active index %d", i)
+			require.Equal(t, activeDocs[i].seq, e.Sequence, "wrong sequence at active index %d", i)
+		}
+
+		// reachedEnd's contract (see getChangesInChannelFromQuery's doc comment) is one-directional:
+		// false means "unknown, there might be more" (a query call landing exactly on `limit` rows
+		// can't tell a coincidental match from LIMIT truncation, so a conservative false is expected
+		// and not a bug) - but true is a hard promise that nothing remains. Verify that promise: it's
+		// the direction that matters for correctness, since a false positive here would mean a
+		// consumer trusts the range as fully scanned when it isn't, silently dropping entries.
+		if reachedEnd {
+			followUp, _, err := collection.getChangesInChannelFromQuery(ctx, channelName, highSeq+1, endSeq, 0, false)
+			require.NoError(t, err)
+			require.Len(t, followUp, 0,
+				"reachedEnd=true but remaining range [%d,%d] still has entries", highSeq+1, endSeq)
+		}
+	})
+}
+
 func TestCountAllDocs(t *testing.T) {
 	db, ctx := setupTestDB(t)
 	defer db.Close(ctx)

From 869dca66c96a625ebd327113c4d1db7f1cc7fb22 Mon Sep 17 00:00:00 2001
From: Tor Colvin <tor.colvin@couchbase.com>
Date: Mon, 6 Jul 2026 14:38:27 -0400
Subject: [PATCH 4/7] simplify code

---
 db/changes.go              |  2 +-
 db/changes_view.go         | 17 ++++++-----------
 db/channel_cache_single.go | 34 ++++++++++++++++++----------------
 3 files changed, 25 insertions(+), 28 deletions(-)

diff --git a/db/changes.go b/db/changes.go
index b73290ea53..5070649526 100644
--- a/db/changes.go
+++ b/db/changes.go
@@ -534,7 +534,7 @@ func (db *DatabaseCollectionWithUser) changesFeed(ctx context.Context, singleCha
 					base.DebugfCtx(ctx, base.KeyChanges, "Terminating channel feed %s", base.UD(to))
 					return
 				case feed <- &change:
-					if !options.ActiveOnly || (!change.Deleted && len(change.Removed) == 0) {
+					if !options.ActiveOnly || logEntry.IsActive() {
 						sentChanges++
 					}
 				}
diff --git a/db/changes_view.go b/db/changes_view.go
index 62d1fd2df7..b96fbde4c2 100644
--- a/db/changes_view.go
+++ b/db/changes_view.go
@@ -91,9 +91,7 @@ func (dbc *DatabaseContext) getQueryHandlerForCollection(collectionID uint32) (C
 
 // Queries the 'channels' view to get a range of sequences of a single channel as LogEntries.
 // The second return value, reachedEnd, is true if the query scanned all the way through to endSeq
-// without stopping early due to the activeOnly+limit active-entry count being satisfied first. When
-// reachedEnd is false, there may be additional matching entries between the last returned entry and
-// endSeq that were never scanned - it is unsafe to assume the range up to endSeq is fully covered.
+// rather than stopping early once an activeOnly+limit active-entry count was satisfied.
 func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (entries LogEntries, reachedEnd bool, err error) {
 	if c.dataStore == nil {
 		return nil, false, errors.New("No data store available for channel query")
@@ -164,14 +162,11 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 		// If active-only, loop until either retrieve (limit) active entries, or reach endSeq.  Non-active entries are still
 		// included in the result set for potential cache prepend
 		if activeOnly {
-			// If we've reached limit before exhausting the range up to endSeq, we're done.  Whether
-			// the range beyond the last returned entry has been fully scanned can't be determined by
-			// comparing highSeq to endSeq alone - channels are often sparse, so the last real entry
-			// is frequently well below endSeq even when nothing was missed.  Instead, check whether
-			// this query call itself returned fewer rows than requested: if so, the store had no more
-			// matching entries anywhere in [startSeq, endSeq], regardless of where highSeq landed. If
-			// the call returned exactly `limit` rows, the LIMIT clause may have truncated further
-			// matches, so fall back to the highSeq/endSeq comparison.
+			// If we've reached limit before exhausting the range up to endSeq, we're done. highSeq
+			// alone can't tell us whether the range was fully scanned - channels are often sparse,
+			// so the last entry is frequently well below endSeq even when nothing was missed. But a
+			// query call returning fewer than `limit` rows means the store had no more matching
+			// entries in range, regardless of highSeq; otherwise fall back to the highSeq comparison.
 			if limit != 0 && activeEntryCount >= limit {
 				reachedEnd = queryRowCount < limit || (endSeq > 0 && highSeq >= endSeq)
 				break
diff --git a/db/channel_cache_single.go b/db/channel_cache_single.go
index 10d524f84c..bad710173d 100644
--- a/db/channel_cache_single.go
+++ b/db/channel_cache_single.go
@@ -439,27 +439,29 @@ func (c *singleChannelCacheImpl) GetChanges(ctx context.Context, options Changes
 	}
 
 	result := resultFromQuery
-	room := options.Limit - len(result)
-	// For activeOnly+limit, only append cache when the query reached the cache boundary
-	// (queryReachedEnd, reported by the query itself - not inferred from sequence numbers, since
-	// this channel's entries are sparse across the global sequence space and the last entry's
-	// sequence is often below endSeq even when the query fully scanned through to endSeq).
-	// If the query stopped early (gap exists before cacheValidFrom), the changesFeed pagination
-	// loop must fetch that range before cache results are valid.
-	// The overlap entry at cacheValidFrom is deduplicated below when cache is appended.
-	if (options.Limit == 0 || room > 0 || (options.ActiveOnly && queryReachedEnd)) && len(resultFromCache) > 0 {
-		// Concatenate the view & cache results:
-		if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
-			resultFromCache = resultFromCache[1:]
+	if options.ActiveOnly {
+		// The query's row limit doesn't correspond to a count of active entries, so cache is only
+		// safe to append once the query itself confirms (via queryReachedEnd) that it scanned all
+		// the way to the cache boundary - otherwise there may be unscanned active entries in between.
+		if queryReachedEnd && len(resultFromCache) > 0 {
+			if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
+				resultFromCache = resultFromCache[1:]
+			}
+			result = append(result, resultFromCache...)
 		}
-		n := len(resultFromCache)
-		// Limit evaluation only valid when not activeOnly, since view and cache results don't apply activeOnly filtering
-		if !options.ActiveOnly {
+	} else {
+		room := options.Limit - len(result)
+		if (options.Limit == 0 || room > 0) && len(resultFromCache) > 0 {
+			// Concatenate the view & cache results:
+			if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
+				resultFromCache = resultFromCache[1:]
+			}
+			n := len(resultFromCache)
 			if options.Limit > 0 && room > 0 && room < n {
 				n = room
 			}
+			result = append(result, resultFromCache[0:n]...)
 		}
-		result = append(result, resultFromCache[0:n]...)
 	}
 	base.InfofCtx(ctx, base.KeyCache, "GetChangesInChannel(%q) --> %d rows", base.UD(c.channelID), len(result))
 

From 656a87a579f4540b5053b3b8c5654a67bd5d0377 Mon Sep 17 00:00:00 2001
From: Tor Colvin <tor.colvin@couchbase.com>
Date: Mon, 6 Jul 2026 15:55:06 -0400
Subject: [PATCH 5/7] make a simpler test

---
 db/changes_test.go         | 189 +++++++++++++++++++++++++++++++++++++
 db/changes_view.go         |  26 ++---
 db/channel_cache.go        |   5 +-
 db/channel_cache_single.go |  35 +++----
 db/channel_cache_test.go   |  77 +++------------
 db/database_test.go        |   6 +-
 db/query_test.go           | 173 ++-------------------------------
 db/util_testing.go         |   3 +-
 8 files changed, 233 insertions(+), 281 deletions(-)

diff --git a/db/changes_test.go b/db/changes_test.go
index 84a176b501..c5e2ed89d7 100644
--- a/db/changes_test.go
+++ b/db/changes_test.go
@@ -9,6 +9,7 @@
 package db
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"reflect"
@@ -19,6 +20,7 @@ import (
 	"github.com/couchbase/sync_gateway/channels"
 	"github.com/couchbase/sync_gateway/testing/assert"
 	"github.com/couchbase/sync_gateway/testing/require"
+	"github.com/google/uuid"
 )
 
 func TestFilterToAvailableChannels(t *testing.T) {
@@ -572,3 +574,190 @@ func TestActiveOnlyWithLimit(t *testing.T) {
 	assert.Equal(t, "doc_act_2", changes[1].ID)
 	assert.Equal(t, "doc_act_3", changes[2].ID)
 }
+
+// stubSingleChannelCache is a minimal SingleChannelCache test double that returns a scripted
+// sequence of batches from GetChanges, one per call, in call order. It lets tests drive
+// changesFeed's own pagination bookkeeping directly, without needing real documents, DCP, or a
+// backing query/view implementation. Only GetChanges and ChannelID are exercised by changesFeed;
+// the remaining methods are unused stubs to satisfy the interface.
+type stubSingleChannelCache struct {
+	channelID channels.ID
+	batches   [][]*LogEntry // one slice per call to GetChanges; calls past the end return empty
+	calls     int
+}
+
+func (s *stubSingleChannelCache) GetChanges(_ context.Context, _ ChangesOptions) ([]*LogEntry, error) {
+	if s.calls >= len(s.batches) {
+		return nil, nil
+	}
+	batch := s.batches[s.calls]
+	s.calls++
+	return batch, nil
+}
+
+func (s *stubSingleChannelCache) GetCachedChanges(_ ChangesOptions) (uint64, []*LogEntry) {
+	return 0, nil
+}
+
+func (s *stubSingleChannelCache) ChannelID() channels.ID {
+	return s.channelID
+}
+
+func (s *stubSingleChannelCache) SupportsLateFeed() bool {
+	return false
+}
+
+func (s *stubSingleChannelCache) LateSequenceUUID() uuid.UUID {
+	return uuid.UUID{}
+}
+
+func (s *stubSingleChannelCache) GetLateSequencesSince(_ uint64) ([]*LogEntry, uint64, error) {
+	return nil, 0, nil
+}
+
+func (s *stubSingleChannelCache) RegisterLateSequenceClient() uint64 {
+	return 0
+}
+
+func (s *stubSingleChannelCache) ReleaseLateSequenceClient(_ uint64) bool {
+	return false
+}
+
+// drainChangesFeed reads a changesFeed's output channel to completion, failing the test on any
+// error entry.
+func drainChangesFeed(t *testing.T, feed <-chan *ChangeEntry) []*ChangeEntry {
+	var received []*ChangeEntry
+	for entry := range feed {
+		require.NoError(t, entry.Err)
+		received = append(received, entry)
+	}
+	return received
+}
+
+// TestChangesFeedActiveOnlyContinuesPastInactiveBatch reproduces the CBG-5555 bug directly against
+// changesFeed, bypassing the cache/query layer entirely: a changesFeed that counts every entry it
+// forwards (active or not) against the caller's requested Limit will believe it's done as soon as
+// it has forwarded `Limit` raw entries - even if none of them were active. Here the first batch is
+// two channel-removal entries (exactly Limit=2 raw rows, zero active), and the second batch holds
+// the two active entries the caller actually asked for. A correct changesFeed must call GetChanges
+// a second time to find them.
+func TestChangesFeedActiveOnlyContinuesPastInactiveBatch(t *testing.T) {
+	db, ctx := setupTestDB(t)
+	defer db.Close(ctx)
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+	collectionID := collection.GetCollectionID()
+	channelID := channels.NewID("active", collectionID)
+
+	stub := &stubSingleChannelCache{
+		channelID: channelID,
+		batches: [][]*LogEntry{
+			{
+				{DocID: "removed1", RevID: "1-a", Sequence: 1, Flags: channels.Removed, CollectionID: collectionID},
+				{DocID: "removed2", RevID: "1-a", Sequence: 2, Flags: channels.Removed, CollectionID: collectionID},
+			},
+			{
+				{DocID: "active1", RevID: "1-a", Sequence: 3, CollectionID: collectionID},
+				{DocID: "active2", RevID: "1-a", Sequence: 4, CollectionID: collectionID},
+			},
+		},
+	}
+
+	options := ChangesOptions{
+		Since:      SequenceID{Seq: 0},
+		ActiveOnly: true,
+		Limit:      2,
+		ChangesCtx: base.TestCtx(t),
+	}
+
+	received := drainChangesFeed(t, collection.changesFeed(ctx, stub, options, "test"))
+
+	// changesFeed forwards every entry it sees, active or not - ActiveOnly filtering happens
+	// upstream in SimpleMultiChangesFeed. What matters here is that all 4 entries were retrieved at
+	// all, which requires a second call to GetChanges.
+	require.Len(t, received, 4)
+	assert.Equal(t, "removed1", received[0].ID)
+	assert.Equal(t, "removed2", received[1].ID)
+	assert.Equal(t, "active1", received[2].ID)
+	assert.Equal(t, "active2", received[3].ID)
+	assert.Equal(t, 2, stub.calls, "changesFeed should have called GetChanges a second time to find the requested 2 active entries")
+}
+
+// TestChangesFeedActiveOnlyMultipleInactiveBatches is the same shape as
+// TestChangesFeedActiveOnlyContinuesPastInactiveBatch but spans three all-inactive batches before
+// the active entries appear, verifying pagination genuinely continues round after round rather than
+// tolerating a single extra retry.
+func TestChangesFeedActiveOnlyMultipleInactiveBatches(t *testing.T) {
+	db, ctx := setupTestDB(t)
+	defer db.Close(ctx)
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+	collectionID := collection.GetCollectionID()
+	channelID := channels.NewID("active", collectionID)
+
+	inactiveBatch := func(seqStart uint64) []*LogEntry {
+		return []*LogEntry{
+			{DocID: fmt.Sprintf("removed%d", seqStart), RevID: "1-a", Sequence: seqStart, Flags: channels.Removed, CollectionID: collectionID},
+			{DocID: fmt.Sprintf("removed%d", seqStart+1), RevID: "1-a", Sequence: seqStart + 1, Flags: channels.Removed, CollectionID: collectionID},
+		}
+	}
+
+	stub := &stubSingleChannelCache{
+		channelID: channelID,
+		batches: [][]*LogEntry{
+			inactiveBatch(1),
+			inactiveBatch(3),
+			inactiveBatch(5),
+			{
+				{DocID: "active1", RevID: "1-a", Sequence: 7, CollectionID: collectionID},
+				{DocID: "active2", RevID: "1-a", Sequence: 8, CollectionID: collectionID},
+			},
+		},
+	}
+
+	options := ChangesOptions{
+		Since:      SequenceID{Seq: 0},
+		ActiveOnly: true,
+		Limit:      2,
+		ChangesCtx: base.TestCtx(t),
+	}
+
+	received := drainChangesFeed(t, collection.changesFeed(ctx, stub, options, "test"))
+
+	require.Len(t, received, 8)
+	assert.Equal(t, "active1", received[6].ID)
+	assert.Equal(t, "active2", received[7].ID)
+	assert.Equal(t, 4, stub.calls, "changesFeed should have paged through all three inactive batches to find the requested 2 active entries")
+}
+
+// TestChangesFeedActiveOnlyStopsWhenChannelExhausted verifies changesFeed terminates (rather than
+// looping forever) when the channel runs out of data before satisfying the requested ActiveOnly
+// Limit: the final batch is shorter than the pagination limit requested for that call, which is
+// changesFeed's signal that the channel has no more data.
+func TestChangesFeedActiveOnlyStopsWhenChannelExhausted(t *testing.T) {
+	db, ctx := setupTestDB(t)
+	defer db.Close(ctx)
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+	collectionID := collection.GetCollectionID()
+	channelID := channels.NewID("active", collectionID)
+
+	stub := &stubSingleChannelCache{
+		channelID: channelID,
+		batches: [][]*LogEntry{
+			{
+				{DocID: "removed1", RevID: "1-a", Sequence: 1, Flags: channels.Removed, CollectionID: collectionID},
+			},
+		},
+	}
+
+	options := ChangesOptions{
+		Since:      SequenceID{Seq: 0},
+		ActiveOnly: true,
+		Limit:      5,
+		ChangesCtx: base.TestCtx(t),
+	}
+
+	received := drainChangesFeed(t, collection.changesFeed(ctx, stub, options, "test"))
+
+	require.Len(t, received, 1)
+	assert.Equal(t, "removed1", received[0].ID)
+	assert.Equal(t, 1, stub.calls, "changesFeed should stop after a short batch signals the channel is exhausted")
+}
diff --git a/db/changes_view.go b/db/changes_view.go
index b96fbde4c2..bd683c1ed3 100644
--- a/db/changes_view.go
+++ b/db/changes_view.go
@@ -90,17 +90,14 @@ func (dbc *DatabaseContext) getQueryHandlerForCollection(collectionID uint32) (C
 }
 
 // Queries the 'channels' view to get a range of sequences of a single channel as LogEntries.
-// The second return value, reachedEnd, is true if the query scanned all the way through to endSeq
-// rather than stopping early once an activeOnly+limit active-entry count was satisfied.
-func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (entries LogEntries, reachedEnd bool, err error) {
+func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (LogEntries, error) {
 	if c.dataStore == nil {
-		return nil, false, errors.New("No data store available for channel query")
+		return nil, errors.New("No data store available for channel query")
 	}
 	start := time.Now()
 	usingViews := c.useViews()
-	entries = make(LogEntries, 0)
+	entries := make(LogEntries, 0)
 	activeEntryCount := 0
-	reachedEnd = true
 
 	base.InfofCtx(ctx, base.KeyCache, "  Querying 'channels' for %q (start=#%d, end=#%d, limit=%d)", base.UD(channelName), startSeq, endSeq, limit)
 
@@ -113,7 +110,7 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 		// Query the view or index
 		queryResults, err := c.QueryChannels(ctx, channelName, startSeq, endSeq, limit, activeOnly)
 		if err != nil {
-			return nil, false, err
+			return nil, err
 		}
 		queryRowCount := 0
 
@@ -148,7 +145,7 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 		// Close query results
 		closeErr := queryResults.Close(ctx)
 		if closeErr != nil {
-			return nil, false, closeErr
+			return nil, closeErr
 		}
 
 		if queryRowCount == 0 {
@@ -156,19 +153,14 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 				break
 			}
 			base.InfofCtx(ctx, base.KeyCache, "    Got no rows from query for channel:%q", base.UD(channelName))
-			return nil, true, nil
+			return nil, nil
 		}
 
 		// If active-only, loop until either retrieve (limit) active entries, or reach endSeq.  Non-active entries are still
 		// included in the result set for potential cache prepend
 		if activeOnly {
-			// If we've reached limit before exhausting the range up to endSeq, we're done. highSeq
-			// alone can't tell us whether the range was fully scanned - channels are often sparse,
-			// so the last entry is frequently well below endSeq even when nothing was missed. But a
-			// query call returning fewer than `limit` rows means the store had no more matching
-			// entries in range, regardless of highSeq; otherwise fall back to the highSeq comparison.
-			if limit != 0 && activeEntryCount >= limit {
-				reachedEnd = queryRowCount < limit || (endSeq > 0 && highSeq >= endSeq)
+			// If we've reached limit, we're done
+			if activeEntryCount >= limit || limit == 0 {
 				break
 			}
 			// If we've reached endSeq, we're done
@@ -194,5 +186,5 @@ func (c *DatabaseCollection) getChangesInChannelFromQuery(ctx context.Context, c
 			elapsed, len(entries), base.UD(channelName), startSeq, endSeq, limit)
 	}
 	c.dbStats().Cache().ViewQueries.Add(1)
-	return entries, reachedEnd, nil
+	return entries, nil
 }
diff --git a/db/channel_cache.go b/db/channel_cache.go
index 0a269ca69f..8697cfeea6 100644
--- a/db/channel_cache.go
+++ b/db/channel_cache.go
@@ -80,10 +80,7 @@ var _ ChannelCache = &channelCacheImpl{}
 
 // ChannelQueryHandler interface is implemented by databaseContext and databaseCollection.
 type ChannelQueryHandler interface {
-	// getChangesInChannelFromQuery returns the LogEntries for a channel within [startSeq, endSeq], and
-	// reachedEnd, which is true if the query scanned all the way through to endSeq without stopping
-	// early due to satisfying an activeOnly+limit active-entry count first.
-	getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (entries LogEntries, reachedEnd bool, err error)
+	getChangesInChannelFromQuery(ctx context.Context, channelName string, startSeq, endSeq uint64, limit int, activeOnly bool) (LogEntries, error)
 }
 
 // Function that returns a ChannelQueryHandlerFunc for the specified collectionID
diff --git a/db/channel_cache_single.go b/db/channel_cache_single.go
index bad710173d..336d08ea63 100644
--- a/db/channel_cache_single.go
+++ b/db/channel_cache_single.go
@@ -419,7 +419,7 @@ func (c *singleChannelCacheImpl) GetChanges(ctx context.Context, options Changes
 	// overlap, which helps confirm that we've got everything.
 	c.cacheStats.ChannelCacheMisses.Add(1)
 	endSeq := cacheValidFrom
-	resultFromQuery, queryReachedEnd, err := c.queryHandler.getChangesInChannelFromQuery(ctx, c.channelID.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
+	resultFromQuery, err := c.queryHandler.getChangesInChannelFromQuery(ctx, c.channelID.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
 	if err != nil {
 		return nil, err
 	}
@@ -439,29 +439,17 @@ func (c *singleChannelCacheImpl) GetChanges(ctx context.Context, options Changes
 	}
 
 	result := resultFromQuery
-	if options.ActiveOnly {
-		// The query's row limit doesn't correspond to a count of active entries, so cache is only
-		// safe to append once the query itself confirms (via queryReachedEnd) that it scanned all
-		// the way to the cache boundary - otherwise there may be unscanned active entries in between.
-		if queryReachedEnd && len(resultFromCache) > 0 {
-			if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
-				resultFromCache = resultFromCache[1:]
-			}
-			result = append(result, resultFromCache...)
+	room := options.Limit - len(result)
+	if (options.Limit == 0 || room > 0) && len(resultFromCache) > 0 {
+		// Concatenate the view & cache results:
+		if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
+			resultFromCache = resultFromCache[1:]
 		}
-	} else {
-		room := options.Limit - len(result)
-		if (options.Limit == 0 || room > 0) && len(resultFromCache) > 0 {
-			// Concatenate the view & cache results:
-			if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
-				resultFromCache = resultFromCache[1:]
-			}
-			n := len(resultFromCache)
-			if options.Limit > 0 && room > 0 && room < n {
-				n = room
-			}
-			result = append(result, resultFromCache[0:n]...)
+		n := len(resultFromCache)
+		if options.Limit > 0 && room > 0 && room < n {
+			n = room
 		}
+		result = append(result, resultFromCache[0:n]...)
 	}
 	base.InfofCtx(ctx, base.KeyCache, "GetChangesInChannel(%q) --> %d rows", base.UD(c.channelID), len(result))
 
@@ -853,8 +841,7 @@ type bypassChannelCache struct {
 func (b *bypassChannelCache) GetChanges(ctx context.Context, options ChangesOptions) ([]*LogEntry, error) {
 	startSeq := options.Since.SafeSequence() + 1
 	endSeq := uint64(math.MaxUint64)
-	entries, _, err := b.queryHandler.getChangesInChannelFromQuery(ctx, b.channel.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
-	return entries, err
+	return b.queryHandler.getChangesInChannelFromQuery(ctx, b.channel.Name, startSeq, endSeq, options.Limit, options.ActiveOnly)
 }
 
 // No cached changes for bypassChannelCache
diff --git a/db/channel_cache_test.go b/db/channel_cache_test.go
index ba1af4fdac..c3fdd4d4c9 100644
--- a/db/channel_cache_test.go
+++ b/db/channel_cache_test.go
@@ -545,9 +545,8 @@ func (qh *testQueryHandler) asFactory(collectionID uint32) (ChannelQueryHandler,
 	return qh, nil
 }
 
-func (qh *testQueryHandler) getChangesInChannelFromQuery(ctx context.Context, channel string, startSeq, endSeq uint64, limit int, activeOnly bool) (LogEntries, bool, error) {
+func (qh *testQueryHandler) getChangesInChannelFromQuery(ctx context.Context, channel string, startSeq, endSeq uint64, limit int, activeOnly bool) (LogEntries, error) {
 	queryEntries := make(LogEntries, 0)
-	reachedEnd := true
 	qh.lock.RLock()
 	for _, entry := range qh.entries {
 		_, ok := entry.Channels[channel]
@@ -557,7 +556,6 @@ func (qh *testQueryHandler) getChangesInChannelFromQuery(ctx context.Context, ch
 			}
 			queryEntries = append(queryEntries, entry)
 			if limit > 0 && len(queryEntries) >= limit {
-				reachedEnd = endSeq > 0 && entry.Sequence >= endSeq
 				break
 			}
 		}
@@ -567,7 +565,7 @@ func (qh *testQueryHandler) getChangesInChannelFromQuery(ctx context.Context, ch
 	qh.lock.Lock()
 	qh.queryCount++
 	qh.lock.Unlock()
-	return queryEntries, reachedEnd, nil
+	return queryEntries, nil
 }
 
 func (qh *testQueryHandler) seedEntries(seededEntries LogEntries) {
@@ -781,8 +779,8 @@ func TestChannelCacheActiveOnlyScenarios(t *testing.T) {
 	})
 	t.Run("cache populated, query requires pagination with limit", func(t *testing.T) {
 		// With ChannelQueryLimit=5 and Limit=10, each GetChanges call returns 5 entries from the
-		// query range.  Without the queryHitActiveLimit guard, the first call would append the
-		// cache (docs 16-20) immediately and skip docs 6-15.
+		// query range, which equals the requested limit, so there's no room to also append the
+		// cache (docs 16-20) - if there were, the first call would skip docs 6-15.
 		cacheOptions := DefaultCacheOptions()
 		cacheOptions.ChannelCacheMaxLength = 5
 		cacheOptions.ChannelQueryLimit = 5
@@ -802,59 +800,6 @@ func TestChannelCacheActiveOnlyScenarios(t *testing.T) {
 			assert.Equal(t, fmt.Sprintf("doc%d", i+1), change.ID, "change at index %d", i)
 		}
 	})
-	t.Run("query exhausts channel before endSeq, reachedEnd must be true", func(t *testing.T) {
-		// doc1: active - contributes a single active row to the channel log.
-		// docA: active, then removed from the channel - the add is superseded by the removal,
-		// so it contributes a single non-active (removal) row, not two rows.
-		// docB: active - the only entry remaining in the channel after docA's removal.
-		// Nothing else exists in the channel all the way out to endSeq, so a query that reaches
-		// docB should report reachedEnd=true even though docB's sequence is far below endSeq -
-		// the gap up to endSeq was fully (if implicitly) scanned, not skipped.
-		ctx, db, collection := setupDBWithChannelCacheSize(t, 2)
-
-		_, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel})
-		require.NoError(t, err)
-		revID, _, err := collection.Put(ctx, "docA", Body{"channels": activeChannel})
-		require.NoError(t, err)
-		_, _, err = collection.Put(ctx, "docA", Body{"channels": "other", "_rev": revID})
-		require.NoError(t, err)
-		_, _, err = collection.Put(ctx, "docB", Body{"channels": activeChannel})
-		require.NoError(t, err)
-		db.WaitForPendingChanges(t)
-
-		// limit=2 forces pagination: the first query call returns doc1's active row and docA's
-		// removal row (2 rows, hitting the row cap) with only 1 active entry seen so far; the
-		// second call returns just docB's active row (1 row, under the row cap), pushing
-		// activeEntryCount to the limit. endSeq is set far beyond any real sequence to exercise
-		// the sparse-channel case.
-		entries, reachedEnd, err := collection.getChangesInChannelFromQuery(ctx, activeChannel, 0, 1000, 2, true)
-		require.NoError(t, err)
-		require.Len(t, entries, 3)
-		assert.True(t, reachedEnd, "query fully exhausted the channel and should report reachedEnd=true")
-	})
-	t.Run("query call lands exactly on limit at the true end of channel, reachedEnd stays conservatively false", func(t *testing.T) {
-		// This pins a deliberate, documented limitation rather than a bug: a single active doc is
-		// the only entry in the channel, and limit=1 exactly matches the row count that call
-		// returns. From the row count alone there's no way to distinguish "the LIMIT clause
-		// truncated a real remaining entry" from "the store coincidentally had exactly `limit`
-		// matching rows" - so reachedEnd must stay conservatively false here, per
-		// getChangesInChannelFromQuery's doc comment ("false" means unknown/maybe, not "definitely
-		// more"). The CBG-5555 fix only resolves the *unambiguous* case, where a query call returns
-		// fewer rows than requested (see the sibling subtest above) - it does not, and cannot
-		// without an extra probe query, resolve this exact-match ambiguity. If this test ever
-		// starts asserting reachedEnd=true here, it means the row-count heuristic changed in a way
-		// that may no longer be safe - verify it can't produce false positives before updating it.
-		ctx, db, collection := setupDBWithChannelCacheSize(t, 2)
-
-		_, _, err := collection.Put(ctx, "doc1", Body{"channels": activeChannel})
-		require.NoError(t, err)
-		db.WaitForPendingChanges(t)
-
-		entries, reachedEnd, err := collection.getChangesInChannelFromQuery(ctx, activeChannel, 0, 1000, 1, true)
-		require.NoError(t, err)
-		require.Len(t, entries, 1)
-		assert.False(t, reachedEnd, "query call was row-capped exactly at limit; reachedEnd must stay conservatively false")
-	})
 }
 
 func setupDBWithChannelCacheSize(t *testing.T, maxLength int) (context.Context, *Database, *DatabaseCollectionWithUser) {
@@ -1167,7 +1112,7 @@ func TestChannelCacheActiveOnlyBoundariesAndGaps(t *testing.T) {
 
 		// doc1 is pruned, cache contains doc2 and doc3. validFrom is 2 (doc1.Seq + 1 = 2).
 		// Querying with Limit=1 should get doc1 (Seq 1) and stop early because limit=1 is met before Seq 2 (validFrom).
-		// Therefore reachedEnd is false, cache is not appended, only doc1 is returned.
+		// The query returns exactly `limit` (1) row, so there's no room left to append the cache; only doc1 is returned.
 		changes := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
 			Since:      SequenceID{Seq: 0},
 			ActiveOnly: true,
@@ -1203,7 +1148,7 @@ func TestChannelCacheActiveOnlyBoundariesAndGaps(t *testing.T) {
 
 		// doc1 and doc2 are pruned. Cache contains doc3 and doc4. validFrom is 3 (doc2.Seq + 1 = 3).
 		// Querying with Limit=2 gets doc1 (Seq 1) and doc2 (Seq 2) and stops early (highSeq = 2 < 3).
-		// Therefore reachedEnd is false, cache is not appended.
+		// The query returns exactly `limit` (2) rows, so there's no room left to append the cache.
 		changes := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
 			Since:      SequenceID{Seq: 0},
 			ActiveOnly: true,
@@ -1214,8 +1159,8 @@ func TestChannelCacheActiveOnlyBoundariesAndGaps(t *testing.T) {
 		assert.Equal(t, "doc1", changes[0].ID)
 		assert.Equal(t, "doc2", changes[1].ID)
 
-		// Querying with Limit=10 gets doc1, doc2, and the query iterator finishes normally at Seq 3 (no other active).
-		// Thus reachedEnd is true, cache is appended, and limit=10 allows all 4 to be returned.
+		// Querying with Limit=10 gets doc1, doc2 (2 rows, under the limit), leaving room to append
+		// the cache (doc3, doc4), so all 4 are returned.
 		changes10 := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
 			Since:      SequenceID{Seq: 0},
 			ActiveOnly: true,
@@ -1256,7 +1201,7 @@ func TestChannelCacheActiveOnlyBoundariesAndGaps(t *testing.T) {
 
 		// doc1 and doc2 are pruned. Cache contains doc3 and doc4. validFrom is 4 (doc2.Seq + 1 = 4).
 		// Querying with Limit=2 gets doc1 (Seq 1) and doc2 (Seq 3) and stops early (highSeq = 3 < 4).
-		// Thus reachedEnd is false, cache is not appended.
+		// The query returns exactly `limit` (2) rows, so there's no room left to append the cache.
 		changes := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
 			Since:      SequenceID{Seq: 0},
 			ActiveOnly: true,
@@ -1267,8 +1212,8 @@ func TestChannelCacheActiveOnlyBoundariesAndGaps(t *testing.T) {
 		assert.Equal(t, "doc1", changes[0].ID)
 		assert.Equal(t, "doc2", changes[1].ID)
 
-		// Querying with Limit=10 gets doc1, doc2, and query iterator reaches endSeq normally (Seq 4 reached).
-		// Thus reachedEnd is true, cache is appended, and limit=10 allows all 4 to be returned.
+		// Querying with Limit=10 gets doc1, doc2 (2 rows, under the limit), leaving room to append
+		// the cache (doc3, doc4), so all 4 are returned.
 		changes10 := getChanges(t, collection, base.SetOf(activeChannel), ChangesOptions{
 			Since:      SequenceID{Seq: 0},
 			ActiveOnly: true,
diff --git a/db/database_test.go b/db/database_test.go
index 8350ce3173..dd37046176 100644
--- a/db/database_test.go
+++ b/db/database_test.go
@@ -2975,7 +2975,7 @@ func TestChannelView(t *testing.T) {
 	// Query view (retry loop to wait for indexing)
 	for i := range 10 {
 		var err error
-		entries, _, err = collection.getChangesInChannelFromQuery(ctx, "*", 0, 100, 0, false)
+		entries, err = collection.getChangesInChannelFromQuery(ctx, "*", 0, 100, 0, false)
 
 		assert.NoError(t, err, "Couldn't create document")
 		if len(entries) >= 1 {
@@ -3048,7 +3048,7 @@ func TestChannelQuery(t *testing.T) {
 
 	for _, testCase := range testCases {
 		t.Run(testCase.testName, func(t *testing.T) {
-			entries, _, err = collection.getChangesInChannelFromQuery(ctx, testCase.channelName, 0, 100, 0, false)
+			entries, err = collection.getChangesInChannelFromQuery(ctx, testCase.channelName, 0, 100, 0, false)
 			require.NoError(t, err)
 
 			for i, entry := range entries {
@@ -3132,7 +3132,7 @@ func TestChannelQueryRevocation(t *testing.T) {
 
 	for _, testCase := range testCases {
 		t.Run(testCase.testName, func(t *testing.T) {
-			entries, _, err = collection.getChangesInChannelFromQuery(ctx, testCase.channelName, 0, 100, 0, false)
+			entries, err = collection.getChangesInChannelFromQuery(ctx, testCase.channelName, 0, 100, 0, false)
 			require.NoError(t, err)
 
 			for i, entry := range entries {
diff --git a/db/query_test.go b/db/query_test.go
index 9bad946125..f704e4edee 100644
--- a/db/query_test.go
+++ b/db/query_test.go
@@ -13,8 +13,6 @@ package db
 import (
 	"context"
 	"fmt"
-	"math/rand"
-	"sort"
 	"strconv"
 	"testing"
 
@@ -488,209 +486,54 @@ func TestQueryChannelsActiveOnlyWithLimit(t *testing.T) {
 
 	// Get changes from channel "ABC" with limit and activeOnly true
 
-	entries, _, err := collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 25, true)
+	entries, err := collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 25, true)
 	require.NoError(t, err, "Couldn't query active docs from channel ABC with limit")
 	require.Len(t, entries, 25)
 	checkFlags(entries)
 
 	// Get changes from channel "*" with limit and activeOnly true
-	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 25, true)
+	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 25, true)
 	require.NoError(t, err, "Couldn't query active docs from channel * with limit")
 	require.Len(t, entries, 25)
 	checkFlags(entries)
 
 	// Get changes from channel "ABC" without limit and activeOnly true
-	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 0, true)
+	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 0, true)
 	require.NoError(t, err, "Couldn't query active docs from channel ABC with limit")
 	require.Len(t, entries, 30)
 	checkFlags(entries)
 
 	// Get changes from channel "*" without limit and activeOnly true
-	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 0, true)
+	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 0, true)
 	require.NoError(t, err, "Couldn't query active docs from channel * with limit")
 	require.Len(t, entries, 30)
 	checkFlags(entries)
 
 	// Get changes from channel "ABC" with limit and activeOnly false
-	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 45, false)
+	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 45, false)
 	require.NoError(t, err, "Couldn't query active docs from channel ABC with limit")
 	require.Len(t, entries, 45)
 	checkFlags(entries)
 
 	// Get changes from channel "*" with limit and activeOnly false
-	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 45, false)
+	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 45, false)
 	require.NoError(t, err, "Couldn't query active docs from channel * with limit")
 	require.Len(t, entries, 45)
 	checkFlags(entries)
 
 	// Get changes from channel "ABC" without limit and activeOnly false
-	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 0, false)
+	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "ABC", startSeq, endSeq, 0, false)
 	require.NoError(t, err, "Couldn't query active docs from channel ABC with limit")
 	require.Len(t, entries, 50)
 	checkFlags(entries)
 
 	// Get changes from channel "*" without limit and activeOnly true
-	entries, _, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 0, false)
+	entries, err = collection.getChangesInChannelFromQuery(base.TestCtx(t), "*", startSeq, endSeq, 0, false)
 	require.NoError(t, err, "Couldn't query active docs from channel * with limit")
 	require.Len(t, entries, 50)
 	checkFlags(entries)
 }
 
-// FuzzGetChangesInChannelFromQueryActiveOnly exercises getChangesInChannelFromQuery's activeOnly+limit
-// pagination directly (the layer TestQueryChannelsActiveOnlyWithLimit covers, but only against
-// Couchbase Server/N1QL) against the default Rosmar/views backing store. It checks that the returned
-// active entries form a correctly ordered, duplicate-free prefix of the true active set, and that
-// reachedEnd never falsely claims the range was fully scanned (reachedEnd=false is allowed to be
-// conservative - see the doc comment on getChangesInChannelFromQuery - but reachedEnd=true is a hard
-// promise that guards a consumer from silently dropping entries it assumes were already covered).
-// This doesn't pin down the specific reachedEnd fix from CBG-5555 (an unnecessary but safe false
-// negative) - see TestChannelCacheActiveOnlyScenarios's "query exhausts channel before endSeq"
-// subtest in channel_cache_test.go for that regression coverage, engineered to land in the
-// unambiguous case; a black-box fuzz property can't reliably distinguish that from the inherent,
-// accepted ambiguity when a query call happens to return exactly `limit` rows.
-func FuzzGetChangesInChannelFromQueryActiveOnly(f *testing.F) {
-	f.Add(uint8(0), uint8(5))
-	f.Add(uint8(1), uint8(1))
-	f.Add(uint8(5), uint8(0)) // no limit
-	f.Add(uint8(10), uint8(3))
-	f.Add(uint8(20), uint8(7))
-	f.Add(uint8(30), uint8(25))
-	f.Add(uint8(30), uint8(2))
-
-	f.Fuzz(func(t *testing.T, numDocs, limit uint8) {
-		const maxDocs = 30
-		n := min(int(numDocs), maxDocs)
-
-		db, ctx := setupTestDBAllowConflicts(t)
-		defer db.Close(ctx)
-		collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
-		collection.ChannelMapper = channels.NewChannelMapper(ctx, channels.DocChannelsSyncFunction, db.Options.JavascriptTimeout)
-
-		const channelName = "ABC"
-
-		type docState struct {
-			id     string
-			revID  string
-			seq    uint64
-			active bool
-		}
-		docsMap := make(map[string]*docState)
-		var docIDs []string
-		nextDocID := 1
-
-		// Deterministic random generator based on inputs to ensure reproducibility
-		rng := rand.New(rand.NewSource(int64(numDocs)*1000 + int64(limit)))
-
-		var endSeq uint64
-		for i := 1; i <= n; i++ {
-			// Choose action:
-			// 0: Create new active doc
-			// 1: Create new other (inactive) doc
-			// 2: Update existing to active
-			// 3: Update existing to other (inactive)
-			action := rng.Intn(4)
-			if len(docIDs) == 0 {
-				action = rng.Intn(2) // Only create actions are possible initially
-			}
-
-			var docID string
-			var body Body
-			var active bool
-			switch action {
-			case 0:
-				docID = fmt.Sprintf("doc_act_%d", nextDocID)
-				nextDocID++
-				body = Body{"channels": channelName}
-				active = true
-			case 1:
-				docID = fmt.Sprintf("doc_oth_%d", nextDocID)
-				nextDocID++
-				body = Body{"channels": "other"}
-				active = false
-			case 2:
-				idx := rng.Intn(len(docIDs))
-				docID = docIDs[idx]
-				body = Body{"channels": channelName, "_rev": docsMap[docID].revID}
-				active = true
-			case 3:
-				idx := rng.Intn(len(docIDs))
-				docID = docIDs[idx]
-				body = Body{"channels": "other", "_rev": docsMap[docID].revID}
-				active = false
-			}
-
-			newRevID, doc, err := collection.Put(ctx, docID, body)
-			require.NoError(t, err)
-			endSeq = doc.Sequence
-
-			if state, exists := docsMap[docID]; exists {
-				state.revID = newRevID
-				state.seq = doc.Sequence
-				state.active = active
-			} else {
-				docsMap[docID] = &docState{id: docID, revID: newRevID, seq: doc.Sequence, active: active}
-				docIDs = append(docIDs, docID)
-			}
-		}
-
-		if n == 0 {
-			return
-		}
-
-		entries, reachedEnd, err := collection.getChangesInChannelFromQuery(ctx, channelName, 0, endSeq, int(limit), true)
-		require.NoError(t, err)
-
-		// Ground truth: active docs at their final sequence, in sequence order.
-		var activeDocs []*docState
-		for _, state := range docsMap {
-			if state.active {
-				activeDocs = append(activeDocs, state)
-			}
-		}
-		sort.Slice(activeDocs, func(i, j int) bool { return activeDocs[i].seq < activeDocs[j].seq })
-
-		// entries may include non-active rows (kept for potential cache prepend) - filter to active
-		// entries for comparison against the true active list, and separately check overall ordering.
-		var highSeq uint64
-		seen := make(map[string]bool, len(entries))
-		var activeReturned []*LogEntry
-		for _, e := range entries {
-			require.False(t, seen[e.DocID], "duplicate entry %s", e.DocID)
-			seen[e.DocID] = true
-			require.GreaterOrEqual(t, e.Sequence, highSeq, "entries must be returned in ascending sequence order")
-			highSeq = e.Sequence
-			if e.IsActive() {
-				activeReturned = append(activeReturned, e)
-			}
-		}
-
-		expectedMin := len(activeDocs)
-		if int(limit) > 0 && int(limit) < expectedMin {
-			expectedMin = int(limit)
-		}
-		require.GreaterOrEqual(t, len(activeReturned), expectedMin,
-			"expected at least %d active entries (limit=%d, total active=%d)", expectedMin, limit, len(activeDocs))
-
-		for i, e := range activeReturned {
-			require.Equal(t, activeDocs[i].id, e.DocID, "wrong doc ID at active index %d", i)
-			require.Equal(t, activeDocs[i].seq, e.Sequence, "wrong sequence at active index %d", i)
-		}
-
-		// reachedEnd's contract (see getChangesInChannelFromQuery's doc comment) is one-directional:
-		// false means "unknown, there might be more" (a query call landing exactly on `limit` rows
-		// can't tell a coincidental match from LIMIT truncation, so a conservative false is expected
-		// and not a bug) - but true is a hard promise that nothing remains. Verify that promise: it's
-		// the direction that matters for correctness, since a false positive here would mean a
-		// consumer trusts the range as fully scanned when it isn't, silently dropping entries.
-		if reachedEnd {
-			followUp, _, err := collection.getChangesInChannelFromQuery(ctx, channelName, highSeq+1, endSeq, 0, false)
-			require.NoError(t, err)
-			require.Len(t, followUp, 0,
-				"reachedEnd=true but remaining range [%d,%d] still has entries", highSeq+1, endSeq)
-		}
-	})
-}
-
 func TestCountAllDocs(t *testing.T) {
 	db, ctx := setupTestDB(t)
 	defer db.Close(ctx)
diff --git a/db/util_testing.go b/db/util_testing.go
index 0e7630171a..15fc0ffe45 100644
--- a/db/util_testing.go
+++ b/db/util_testing.go
@@ -443,8 +443,7 @@ func (dbc *DatabaseContext) ChannelViewForTest(tb testing.TB, channelName string
 }
 
 func (dbc *DatabaseContext) CollectionChannelViewForTest(tb testing.TB, collection *DatabaseCollection, channelName string, startSeq, endSeq uint64) (LogEntries, error) {
-	entries, _, err := collection.getChangesInChannelFromQuery(base.TestCtx(tb), channelName, startSeq, endSeq, 0, false)
-	return entries, err
+	return collection.getChangesInChannelFromQuery(base.TestCtx(tb), channelName, startSeq, endSeq, 0, false)
 }
 
 // Test-only version of GetPrincipal that doesn't trigger channel/role recalculation

From 5bee74cd96de5521281668e51274913180ef5072 Mon Sep 17 00:00:00 2001
From: Tor Colvin <tor.colvin@couchbase.com>
Date: Mon, 6 Jul 2026 16:34:18 -0400
Subject: [PATCH 6/7] filter in changesFeed

---
 db/changes.go      | 16 ++++++------
 db/changes_test.go | 65 +++++++++++++++++++++++++---------------------
 2 files changed, 43 insertions(+), 38 deletions(-)

diff --git a/db/changes.go b/db/changes.go
index 5070649526..64c1bb6b52 100644
--- a/db/changes.go
+++ b/db/changes.go
@@ -486,7 +486,7 @@ func (db *DatabaseCollectionWithUser) changesFeed(ctx context.Context, singleCha
 				return
 			}
 			// Calculate limit for this iteration
-			if requestLimit == 0 {
+			if requestLimit == 0 || options.ActiveOnly {
 				paginationOptions.Limit = queryLimit
 			} else {
 				remainingLimit := requestLimit - itemsSent
@@ -534,9 +534,7 @@ func (db *DatabaseCollectionWithUser) changesFeed(ctx context.Context, singleCha
 					base.DebugfCtx(ctx, base.KeyChanges, "Terminating channel feed %s", base.UD(to))
 					return
 				case feed <- &change:
-					if !options.ActiveOnly || logEntry.IsActive() {
-						sentChanges++
-					}
+					sentChanges++
 				}
 			}
 
@@ -545,12 +543,14 @@ func (db *DatabaseCollectionWithUser) changesFeed(ctx context.Context, singleCha
 				return
 			}
 
-			// If we've reached the request limit, we're done
+			// If we've reached the request limit we're done.  If this is an ActiveOnly changes feed there
+			// is additional filtering in the main changes loop, so we can't apply the requestLimit here.
 			itemsSent += sentChanges
-			if requestLimit > 0 && itemsSent >= requestLimit {
-				return
+			if !options.ActiveOnly {
+				if requestLimit > 0 && itemsSent >= requestLimit {
+					return
+				}
 			}
-
 			paginationOptions.Since.Seq = lastSeq
 		}
 	}()
diff --git a/db/changes_test.go b/db/changes_test.go
index c5e2ed89d7..a8535c9904 100644
--- a/db/changes_test.go
+++ b/db/changes_test.go
@@ -635,16 +635,18 @@ func drainChangesFeed(t *testing.T, feed <-chan *ChangeEntry) []*ChangeEntry {
 }
 
 // TestChangesFeedActiveOnlyContinuesPastInactiveBatch reproduces the CBG-5555 bug directly against
-// changesFeed, bypassing the cache/query layer entirely: a changesFeed that counts every entry it
-// forwards (active or not) against the caller's requested Limit will believe it's done as soon as
-// it has forwarded `Limit` raw entries - even if none of them were active. Here the first batch is
-// two channel-removal entries (exactly Limit=2 raw rows, zero active), and the second batch holds
-// the two active entries the caller actually asked for. A correct changesFeed must call GetChanges
-// a second time to find them.
+// changesFeed, bypassing the cache/query layer entirely. For ActiveOnly feeds, changesFeed pages by
+// ChannelQueryLimit rather than the caller's requested Limit (the caller's Limit is enforced further
+// upstream, in SimpleMultiChangesFeed, since raw entries and active entries aren't the same count) -
+// so a changesFeed that mistook a full-sized batch of inactive rows for exhaustion would stop before
+// ever finding an active entry. Here ChannelQueryLimit is set to 3: the first batch is three
+// channel-removal entries (a full page, zero active), and the second batch - shorter than the page
+// size - holds the active entries and signals the channel is exhausted. A correct changesFeed must
+// call GetChanges a second time to find them.
 func TestChangesFeedActiveOnlyContinuesPastInactiveBatch(t *testing.T) {
-	db, ctx := setupTestDB(t)
-	defer db.Close(ctx)
-	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+	cacheOptions := DefaultCacheOptions()
+	cacheOptions.ChannelQueryLimit = 3
+	ctx, _, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
 	collectionID := collection.GetCollectionID()
 	channelID := channels.NewID("active", collectionID)
 
@@ -654,10 +656,11 @@ func TestChangesFeedActiveOnlyContinuesPastInactiveBatch(t *testing.T) {
 			{
 				{DocID: "removed1", RevID: "1-a", Sequence: 1, Flags: channels.Removed, CollectionID: collectionID},
 				{DocID: "removed2", RevID: "1-a", Sequence: 2, Flags: channels.Removed, CollectionID: collectionID},
+				{DocID: "removed3", RevID: "1-a", Sequence: 3, Flags: channels.Removed, CollectionID: collectionID},
 			},
 			{
-				{DocID: "active1", RevID: "1-a", Sequence: 3, CollectionID: collectionID},
-				{DocID: "active2", RevID: "1-a", Sequence: 4, CollectionID: collectionID},
+				{DocID: "active1", RevID: "1-a", Sequence: 4, CollectionID: collectionID},
+				{DocID: "active2", RevID: "1-a", Sequence: 5, CollectionID: collectionID},
 			},
 		},
 	}
@@ -672,24 +675,25 @@ func TestChangesFeedActiveOnlyContinuesPastInactiveBatch(t *testing.T) {
 	received := drainChangesFeed(t, collection.changesFeed(ctx, stub, options, "test"))
 
 	// changesFeed forwards every entry it sees, active or not - ActiveOnly filtering happens
-	// upstream in SimpleMultiChangesFeed. What matters here is that all 4 entries were retrieved at
+	// upstream in SimpleMultiChangesFeed. What matters here is that all 5 entries were retrieved at
 	// all, which requires a second call to GetChanges.
-	require.Len(t, received, 4)
+	require.Len(t, received, 5)
 	assert.Equal(t, "removed1", received[0].ID)
 	assert.Equal(t, "removed2", received[1].ID)
-	assert.Equal(t, "active1", received[2].ID)
-	assert.Equal(t, "active2", received[3].ID)
-	assert.Equal(t, 2, stub.calls, "changesFeed should have called GetChanges a second time to find the requested 2 active entries")
+	assert.Equal(t, "removed3", received[2].ID)
+	assert.Equal(t, "active1", received[3].ID)
+	assert.Equal(t, "active2", received[4].ID)
+	assert.Equal(t, 2, stub.calls, "changesFeed should have called GetChanges a second time to find the active entries")
 }
 
 // TestChangesFeedActiveOnlyMultipleInactiveBatches is the same shape as
-// TestChangesFeedActiveOnlyContinuesPastInactiveBatch but spans three all-inactive batches before
-// the active entries appear, verifying pagination genuinely continues round after round rather than
-// tolerating a single extra retry.
+// TestChangesFeedActiveOnlyContinuesPastInactiveBatch but spans three all-inactive, full-page batches
+// before the active entries appear, verifying pagination genuinely continues round after round rather
+// than tolerating a single extra retry.
 func TestChangesFeedActiveOnlyMultipleInactiveBatches(t *testing.T) {
-	db, ctx := setupTestDB(t)
-	defer db.Close(ctx)
-	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+	cacheOptions := DefaultCacheOptions()
+	cacheOptions.ChannelQueryLimit = 3
+	ctx, _, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
 	collectionID := collection.GetCollectionID()
 	channelID := channels.NewID("active", collectionID)
 
@@ -697,6 +701,7 @@ func TestChangesFeedActiveOnlyMultipleInactiveBatches(t *testing.T) {
 		return []*LogEntry{
 			{DocID: fmt.Sprintf("removed%d", seqStart), RevID: "1-a", Sequence: seqStart, Flags: channels.Removed, CollectionID: collectionID},
 			{DocID: fmt.Sprintf("removed%d", seqStart+1), RevID: "1-a", Sequence: seqStart + 1, Flags: channels.Removed, CollectionID: collectionID},
+			{DocID: fmt.Sprintf("removed%d", seqStart+2), RevID: "1-a", Sequence: seqStart + 2, Flags: channels.Removed, CollectionID: collectionID},
 		}
 	}
 
@@ -704,11 +709,11 @@ func TestChangesFeedActiveOnlyMultipleInactiveBatches(t *testing.T) {
 		channelID: channelID,
 		batches: [][]*LogEntry{
 			inactiveBatch(1),
-			inactiveBatch(3),
-			inactiveBatch(5),
+			inactiveBatch(4),
+			inactiveBatch(7),
 			{
-				{DocID: "active1", RevID: "1-a", Sequence: 7, CollectionID: collectionID},
-				{DocID: "active2", RevID: "1-a", Sequence: 8, CollectionID: collectionID},
+				{DocID: "active1", RevID: "1-a", Sequence: 10, CollectionID: collectionID},
+				{DocID: "active2", RevID: "1-a", Sequence: 11, CollectionID: collectionID},
 			},
 		},
 	}
@@ -722,10 +727,10 @@ func TestChangesFeedActiveOnlyMultipleInactiveBatches(t *testing.T) {
 
 	received := drainChangesFeed(t, collection.changesFeed(ctx, stub, options, "test"))
 
-	require.Len(t, received, 8)
-	assert.Equal(t, "active1", received[6].ID)
-	assert.Equal(t, "active2", received[7].ID)
-	assert.Equal(t, 4, stub.calls, "changesFeed should have paged through all three inactive batches to find the requested 2 active entries")
+	require.Len(t, received, 11)
+	assert.Equal(t, "active1", received[9].ID)
+	assert.Equal(t, "active2", received[10].ID)
+	assert.Equal(t, 4, stub.calls, "changesFeed should have paged through all three inactive batches to find the active entries")
 }
 
 // TestChangesFeedActiveOnlyStopsWhenChannelExhausted verifies changesFeed terminates (rather than

From b8e6b67592b498b851ae2f5844c84102456efa62 Mon Sep 17 00:00:00 2001
From: Tor Colvin <tor.colvin@couchbase.com>
Date: Mon, 6 Jul 2026 17:25:16 -0400
Subject: [PATCH 7/7] improve tests

---
 db/changes.go      |   2 +-
 db/changes_test.go | 113 ++++++++++++++++++++++-----------------------
 2 files changed, 57 insertions(+), 58 deletions(-)

diff --git a/db/changes.go b/db/changes.go
index 64c1bb6b52..0719341ada 100644
--- a/db/changes.go
+++ b/db/changes.go
@@ -545,8 +545,8 @@ func (db *DatabaseCollectionWithUser) changesFeed(ctx context.Context, singleCha
 
 			// If we've reached the request limit we're done.  If this is an ActiveOnly changes feed there
 			// is additional filtering in the main changes loop, so we can't apply the requestLimit here.
-			itemsSent += sentChanges
 			if !options.ActiveOnly {
+				itemsSent += sentChanges
 				if requestLimit > 0 && itemsSent >= requestLimit {
 					return
 				}
diff --git a/db/changes_test.go b/db/changes_test.go
index a8535c9904..3f500249b8 100644
--- a/db/changes_test.go
+++ b/db/changes_test.go
@@ -575,24 +575,32 @@ func TestActiveOnlyWithLimit(t *testing.T) {
 	assert.Equal(t, "doc_act_3", changes[2].ID)
 }
 
-// stubSingleChannelCache is a minimal SingleChannelCache test double that returns a scripted
-// sequence of batches from GetChanges, one per call, in call order. It lets tests drive
-// changesFeed's own pagination bookkeeping directly, without needing real documents, DCP, or a
-// backing query/view implementation. Only GetChanges and ChannelID are exercised by changesFeed;
-// the remaining methods are unused stubs to satisfy the interface.
+// stubSingleChannelCache is a minimal SingleChannelCache test double that serves entries from a
+// fixed, sequence-ordered list, honoring Since and Limit the way a real channel cache would. This
+// lets tests drive changesFeed's own pagination bookkeeping directly, without needing real
+// documents, DCP, or a backing query/view implementation, while still having changesFeed's actual
+// Since/Limit choices (not a pre-scripted call count) determine what gets returned. Only GetChanges
+// and ChannelID are exercised by changesFeed; the remaining methods are unused stubs to satisfy the
+// interface.
 type stubSingleChannelCache struct {
 	channelID channels.ID
-	batches   [][]*LogEntry // one slice per call to GetChanges; calls past the end return empty
+	entries   []*LogEntry // full ordered set of entries in the channel, by sequence
 	calls     int
 }
 
-func (s *stubSingleChannelCache) GetChanges(_ context.Context, _ ChangesOptions) ([]*LogEntry, error) {
-	if s.calls >= len(s.batches) {
-		return nil, nil
-	}
-	batch := s.batches[s.calls]
+func (s *stubSingleChannelCache) GetChanges(_ context.Context, options ChangesOptions) ([]*LogEntry, error) {
 	s.calls++
-	return batch, nil
+	var result []*LogEntry
+	for _, entry := range s.entries {
+		if entry.Sequence <= options.Since.Seq {
+			continue
+		}
+		result = append(result, entry)
+		if options.Limit > 0 && len(result) >= options.Limit {
+			break
+		}
+	}
+	return result, nil
 }
 
 func (s *stubSingleChannelCache) GetCachedChanges(_ ChangesOptions) (uint64, []*LogEntry) {
@@ -635,14 +643,16 @@ func drainChangesFeed(t *testing.T, feed <-chan *ChangeEntry) []*ChangeEntry {
 }
 
 // TestChangesFeedActiveOnlyContinuesPastInactiveBatch reproduces the CBG-5555 bug directly against
-// changesFeed, bypassing the cache/query layer entirely. For ActiveOnly feeds, changesFeed pages by
-// ChannelQueryLimit rather than the caller's requested Limit (the caller's Limit is enforced further
-// upstream, in SimpleMultiChangesFeed, since raw entries and active entries aren't the same count) -
-// so a changesFeed that mistook a full-sized batch of inactive rows for exhaustion would stop before
-// ever finding an active entry. Here ChannelQueryLimit is set to 3: the first batch is three
-// channel-removal entries (a full page, zero active), and the second batch - shorter than the page
-// size - holds the active entries and signals the channel is exhausted. A correct changesFeed must
-// call GetChanges a second time to find them.
+// changesFeed, bypassing the cache/query layer entirely. For ActiveOnly feeds, changesFeed must page
+// by ChannelQueryLimit rather than the caller's requested Limit (the caller's Limit is enforced
+// further upstream, in SimpleMultiChangesFeed, since raw entries and active entries aren't the same
+// count). The buggy version instead kept shrinking its per-call Limit toward the small requested
+// Limit and stopped as soon as it had sent that many *active* entries - so with Limit=1 here, it
+// would give up after finding a single active entry even though the channel has a second one still
+// to come. ChannelQueryLimit is set to 3: the channel has three channel-removal entries (a full page,
+// zero active) followed by two active entries. The stub honors Since/Limit like a real channel
+// cache, so this only passes if changesFeed's actual pagination choices - not a pre-scripted call
+// count - are what produce the full result.
 func TestChangesFeedActiveOnlyContinuesPastInactiveBatch(t *testing.T) {
 	cacheOptions := DefaultCacheOptions()
 	cacheOptions.ChannelQueryLimit = 3
@@ -652,44 +662,41 @@ func TestChangesFeedActiveOnlyContinuesPastInactiveBatch(t *testing.T) {
 
 	stub := &stubSingleChannelCache{
 		channelID: channelID,
-		batches: [][]*LogEntry{
-			{
-				{DocID: "removed1", RevID: "1-a", Sequence: 1, Flags: channels.Removed, CollectionID: collectionID},
-				{DocID: "removed2", RevID: "1-a", Sequence: 2, Flags: channels.Removed, CollectionID: collectionID},
-				{DocID: "removed3", RevID: "1-a", Sequence: 3, Flags: channels.Removed, CollectionID: collectionID},
-			},
-			{
-				{DocID: "active1", RevID: "1-a", Sequence: 4, CollectionID: collectionID},
-				{DocID: "active2", RevID: "1-a", Sequence: 5, CollectionID: collectionID},
-			},
+		entries: []*LogEntry{
+			{DocID: "removed1", RevID: "1-a", Sequence: 1, Flags: channels.Removed, CollectionID: collectionID},
+			{DocID: "removed2", RevID: "1-a", Sequence: 2, Flags: channels.Removed, CollectionID: collectionID},
+			{DocID: "removed3", RevID: "1-a", Sequence: 3, Flags: channels.Removed, CollectionID: collectionID},
+			{DocID: "active1", RevID: "1-a", Sequence: 4, CollectionID: collectionID},
+			{DocID: "active2", RevID: "1-a", Sequence: 5, CollectionID: collectionID},
 		},
 	}
 
 	options := ChangesOptions{
 		Since:      SequenceID{Seq: 0},
 		ActiveOnly: true,
-		Limit:      2,
+		Limit:      1,
 		ChangesCtx: base.TestCtx(t),
 	}
 
 	received := drainChangesFeed(t, collection.changesFeed(ctx, stub, options, "test"))
 
 	// changesFeed forwards every entry it sees, active or not - ActiveOnly filtering happens
-	// upstream in SimpleMultiChangesFeed. What matters here is that all 5 entries were retrieved at
-	// all, which requires a second call to GetChanges.
+	// upstream in SimpleMultiChangesFeed. What matters here is that all 5 entries were retrieved,
+	// including active2, which the buggy version dropped.
 	require.Len(t, received, 5)
 	assert.Equal(t, "removed1", received[0].ID)
 	assert.Equal(t, "removed2", received[1].ID)
 	assert.Equal(t, "removed3", received[2].ID)
 	assert.Equal(t, "active1", received[3].ID)
 	assert.Equal(t, "active2", received[4].ID)
-	assert.Equal(t, 2, stub.calls, "changesFeed should have called GetChanges a second time to find the active entries")
+	assert.Equal(t, 2, stub.calls, "changesFeed should page by ChannelQueryLimit, not the much smaller requested Limit, for ActiveOnly feeds")
 }
 
 // TestChangesFeedActiveOnlyMultipleInactiveBatches is the same shape as
 // TestChangesFeedActiveOnlyContinuesPastInactiveBatch but spans three all-inactive, full-page batches
-// before the active entries appear, verifying pagination genuinely continues round after round rather
-// than tolerating a single extra retry.
+// before four active entries appear - more than the requested Limit of 2. The buggy version stopped
+// as soon as it had sent Limit-many active entries, so it would give up after active2 and never see
+// active3 or active4, even though the channel has more data.
 func TestChangesFeedActiveOnlyMultipleInactiveBatches(t *testing.T) {
 	cacheOptions := DefaultCacheOptions()
 	cacheOptions.ChannelQueryLimit = 3
@@ -697,25 +704,17 @@ func TestChangesFeedActiveOnlyMultipleInactiveBatches(t *testing.T) {
 	collectionID := collection.GetCollectionID()
 	channelID := channels.NewID("active", collectionID)
 
-	inactiveBatch := func(seqStart uint64) []*LogEntry {
-		return []*LogEntry{
-			{DocID: fmt.Sprintf("removed%d", seqStart), RevID: "1-a", Sequence: seqStart, Flags: channels.Removed, CollectionID: collectionID},
-			{DocID: fmt.Sprintf("removed%d", seqStart+1), RevID: "1-a", Sequence: seqStart + 1, Flags: channels.Removed, CollectionID: collectionID},
-			{DocID: fmt.Sprintf("removed%d", seqStart+2), RevID: "1-a", Sequence: seqStart + 2, Flags: channels.Removed, CollectionID: collectionID},
-		}
+	var entries []*LogEntry
+	for seq := uint64(1); seq <= 9; seq++ {
+		entries = append(entries, &LogEntry{DocID: fmt.Sprintf("removed%d", seq), RevID: "1-a", Sequence: seq, Flags: channels.Removed, CollectionID: collectionID})
+	}
+	for i, seq := 1, uint64(10); seq <= 13; i, seq = i+1, seq+1 {
+		entries = append(entries, &LogEntry{DocID: fmt.Sprintf("active%d", i), RevID: "1-a", Sequence: seq, CollectionID: collectionID})
 	}
 
 	stub := &stubSingleChannelCache{
 		channelID: channelID,
-		batches: [][]*LogEntry{
-			inactiveBatch(1),
-			inactiveBatch(4),
-			inactiveBatch(7),
-			{
-				{DocID: "active1", RevID: "1-a", Sequence: 10, CollectionID: collectionID},
-				{DocID: "active2", RevID: "1-a", Sequence: 11, CollectionID: collectionID},
-			},
-		},
+		entries:   entries,
 	}
 
 	options := ChangesOptions{
@@ -727,10 +726,12 @@ func TestChangesFeedActiveOnlyMultipleInactiveBatches(t *testing.T) {
 
 	received := drainChangesFeed(t, collection.changesFeed(ctx, stub, options, "test"))
 
-	require.Len(t, received, 11)
+	require.Len(t, received, 13)
 	assert.Equal(t, "active1", received[9].ID)
 	assert.Equal(t, "active2", received[10].ID)
-	assert.Equal(t, 4, stub.calls, "changesFeed should have paged through all three inactive batches to find the active entries")
+	assert.Equal(t, "active3", received[11].ID)
+	assert.Equal(t, "active4", received[12].ID)
+	assert.Equal(t, 5, stub.calls, "changesFeed should have paged through all three inactive batches and kept going to find all four active entries")
 }
 
 // TestChangesFeedActiveOnlyStopsWhenChannelExhausted verifies changesFeed terminates (rather than
@@ -746,10 +747,8 @@ func TestChangesFeedActiveOnlyStopsWhenChannelExhausted(t *testing.T) {
 
 	stub := &stubSingleChannelCache{
 		channelID: channelID,
-		batches: [][]*LogEntry{
-			{
-				{DocID: "removed1", RevID: "1-a", Sequence: 1, Flags: channels.Removed, CollectionID: collectionID},
-			},
+		entries: []*LogEntry{
+			{DocID: "removed1", RevID: "1-a", Sequence: 1, Flags: channels.Removed, CollectionID: collectionID},
 		},
 	}