CBG-5555 better fix for activeOnly+limit

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
@@ -543,12 +543,14 @@ func (db *DatabaseCollectionWithUser) changesFeed(ctx context.Context, singleCha
 				return
 			}
 
-			// If we've reached the request limit, we're done
-			itemsSent += sentChanges
-			if requestLimit > 0 && itemsSent >= requestLimit {
-				return
+			// 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.
+			if !options.ActiveOnly {
+				itemsSent += sentChanges
+				if requestLimit > 0 && itemsSent >= requestLimit {
+					return
+				}
 			}
-
 			paginationOptions.Since.Seq = lastSeq
 		}
 	}()

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) {
@@ -517,3 +519,249 @@ 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)
+}
+
+// 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
+	entries   []*LogEntry // full ordered set of entries in the channel, by sequence
+	calls     int
+}
+
+func (s *stubSingleChannelCache) GetChanges(_ context.Context, options ChangesOptions) ([]*LogEntry, error) {
+	s.calls++
+	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) {
+	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. 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
+	ctx, _, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+	collectionID := collection.GetCollectionID()
+	channelID := channels.NewID("active", collectionID)
+
+	stub := &stubSingleChannelCache{
+		channelID: channelID,
+		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:      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,
+	// 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 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 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
+	ctx, _, collection := setupDBWithChannelCacheSettings(t, cacheOptions)
+	collectionID := collection.GetCollectionID()
+	channelID := channels.NewID("active", 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,
+		entries:   entries,
+	}
+
+	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, 13)
+	assert.Equal(t, "active1", received[9].ID)
+	assert.Equal(t, "active2", received[10].ID)
+	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
+// 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,
+		entries: []*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")
+}