CBG-5555 better fix for activeOnly+limit

Author: torcolvin

db/channel_cache_single.go

@@ -440,7 +440,12 @@ func (c *singleChannelCacheImpl) GetChanges(ctx context.Context, options Changes
 
 	result := resultFromQuery
 	room := options.Limit - len(result)
-	if (options.Limit == 0 || room > 0 || options.ActiveOnly) && 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:]

db/channel_cache_test.go

@@ -755,15 +755,131 @@ func TestChannelCacheActiveOnlyScenarios(t *testing.T) {
 		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)
+			_, _, _ = 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)
+		}
+	})
+}