make a simpler test

Author: torcolvin

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")
+}

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,27 +145,22 @@ 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 {
 			if len(entries) > 0 {
 				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
 }

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

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