diff --git a/db/changes.go b/db/changes.go index aa95cf4d25..0719341ada 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 @@ -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 } }() diff --git a/db/changes_test.go b/db/changes_test.go index af90dfb5f5..3f500249b8 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) { @@ -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") +} diff --git a/db/channel_cache_test.go b/db/channel_cache_test.go index b1ecd02ab8..c3fdd4d4c9 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" @@ -598,3 +599,631 @@ 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, 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 + 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) { + // --- 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 { + 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), + }) + + 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++ { + // 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{ + Since: SequenceID{Seq: 0}, + ActiveOnly: true, + Limit: int(requestLimit), + ChangesCtx: base.TestCtx(t), + }) + + // 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 active docs, or exactly requestLimit if that is smaller + limit := int(requestLimit) + 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 active docs in final sequence order + for i, c := range changes { + 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). + // 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, + 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). + // 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, + 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 (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, + 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). + // 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, + 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 (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, + 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) + }) +}