From 757eeff9a6ded42b251faed485b75dbc02ec9772 Mon Sep 17 00:00:00 2001 From: Tor Colvin Date: Mon, 5 Oct 2026 16:02:30 -0400 Subject: [PATCH] CBG-5951: make changeCache.initialSequence atomic The caching DCP feed starts before changeCache.Start sets initialSequence, so the feed is running while Start writes the field. DocChanged and processPrincipalDoc read initialSequence without the cache lock. A feed event that arrived while the database was starting raced with the write in Start. TestPushReplicationAPIUpdateDatabase hits this by reloading a database while documents are being written. Store initialSequence as an atomic.Uint64. An event that arrives before Start can still read zero and pass the startup filter, but processEntry waits for the cache lock and then drops it as a duplicate of a sequence below nextSequence. Add TestDocChangedBeforeStart, which delivers feed events between Init and Start. Co-Authored-By: Claude Opus 5.5 (1M context) --- db/change_cache.go | 14 +++---- db/changecachetest/change_cache_test.go | 49 +++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 7 deletions(-) diff --git a/db/change_cache.go b/db/change_cache.go index 4eca7021cd..dfecfc1622 100644 --- a/db/change_cache.go +++ b/db/change_cache.go @@ -67,7 +67,7 @@ type changeCache struct { logCtx context.Context // fix in sg-bucket to ProcessEvent logsDisabled bool // If true, ignore incoming tap changes nextSequence uint64 // Next consecutive sequence number to add. State variable for sequence buffering tracking. Should use getNextSequence() rather than accessing directly. - initialSequence uint64 // DB's current sequence at startup time. + initialSequence atomic.Uint64 // DB's current sequence at startup time. Atomic because DocChanged reads it without the lock, and the feed starts before Start sets it. receivedSeqs map[uint64]struct{} // Set of all sequences received pendingLogs LogPriorityQueue // Out-of-sequence entries waiting to be cached notifyChangeFunc func(context.Context, channels.Set) // Client callback that notifies of channel changes @@ -254,11 +254,11 @@ func (c *changeCache) Clear(ctx context.Context) error { // Reset initialSequence so that any new channel caches have their validFrom set to the current last sequence // the point at which the change cache was initialized / re-initialized. // No need to touch c.nextSequence here, because we don't want to touch the sequence buffering state. - var err error - c.initialSequence, err = c.db.LastSequence(ctx) + initialSequence, err := c.db.LastSequence(ctx) if err != nil { return err } + c.initialSequence.Store(initialSequence) c.pendingLogs = nil heap.Init(&c.pendingLogs) @@ -433,7 +433,7 @@ func (c *changeCache) DocChanged(event sgbucket.FeedEvent, docType DocumentType) return } - if syncData.Sequence <= c.initialSequence { + if syncData.Sequence <= c.initialSequence.Load() { return // DCP is sending us an old value from before I started up; ignore it } @@ -754,7 +754,7 @@ func (c *changeCache) processPrincipalDoc(ctx context.Context, docID string, doc } sequence := princ.Sequence - if sequence <= c.initialSequence { + if sequence <= c.initialSequence.Load() { return // Tap is sending us an old value from before I started up; ignore it } @@ -839,7 +839,7 @@ func (c *changeCache) processEntry(ctx context.Context, change *LogEntry) []chan // Too many pending; add the oldest one: changedChannels = append(changedChannels, c._addPendingLogs(ctx)...) } - } else if sequence > c.initialSequence { + } else if sequence > c.initialSequence.Load() { // Out-of-order sequence received! // Remove from skipped sequence queue if !change.Skipped { @@ -1012,7 +1012,7 @@ func (c *changeCache) getOldestSkippedSequence(ctx context.Context) uint64 { // Set the initial sequence. Presumes that change cache is already locked. func (c *changeCache) _setInitialSequence(initialSequence uint64) { - c.initialSequence = initialSequence + c.initialSequence.Store(initialSequence) c.nextSequence = initialSequence + 1 } diff --git a/db/changecachetest/change_cache_test.go b/db/changecachetest/change_cache_test.go index 05dae81065..0c3f098112 100644 --- a/db/changecachetest/change_cache_test.go +++ b/db/changecachetest/change_cache_test.go @@ -3059,6 +3059,55 @@ func TestDocChangedStartupSequenceFilter(t *testing.T) { assert.Equal(t, "aboveInitial", cached[0].DocID) } +// TestDocChangedBeforeStart delivers feed events between Init and Start, as the DCP feed does on database startup. +// A replay at initialSequence must still be kept out of the cache once Start runs. +func TestDocChangedBeforeStart(t *testing.T) { + const initialSequence = 100 + database, ctx := db.SetupTestDB(t) + defer database.Close(ctx) + collection := db.GetSingleDatabaseCollection(t, database.DatabaseContext) + collectionID := collection.GetCollectionID() + + changeCache := db.NewChangeCacheForTest(t) + require.NoError(t, changeCache.Init(ctx, database.DatabaseContext, database.DatabaseContext.ChannelCacheForTest(t), + nil, nil, database.DatabaseContext.MetadataKeys)) + defer changeCache.Stop(ctx) + + syncDataFor := func(seq uint64) db.SyncData { + return db.SyncData{ + RevAndVersion: channels.RevAndVersion{RevTreeID: "1-abc"}, + Sequence: seq, + Channels: channels.ChannelMap{"ABC": nil}, + History: db.RevTree{"1-abc": &db.RevInfo{ID: "1-abc", Channels: base.SetOf("ABC")}}, + } + } + + // The channel has to exist before the feed event or nothing lands in it. + abcChannel := channels.NewID("ABC", collectionID) + _, err := database.DatabaseContext.ChannelCacheForTest(t).GetCachedChanges(ctx, abcChannel) + require.NoError(t, err) + + events := []sgbucket.FeedEvent{ + feedEventForTest(t, "atInitial", collectionID, syncDataFor(initialSequence)), + feedEventForTest(t, "aboveInitial", collectionID, syncDataFor(initialSequence+1)), + } + var wg sync.WaitGroup + for _, event := range events { + wg.Go(func() { + changeCache.DocChanged(event, db.DocTypeDocument) + }) + } + + require.NoError(t, changeCache.Start(initialSequence)) + wg.Wait() + + assert.Equal(t, uint64(initialSequence+1), changeCache.LastSequence()) + cached, err := database.DatabaseContext.ChannelCacheForTest(t).GetCachedChanges(ctx, abcChannel) + require.NoError(t, err) + require.Len(t, cached, 1) + assert.Equal(t, "aboveInitial", cached[0].DocID) +} + // TestDocChangedRecentSequences covers how DocChanged backfills sequences that DCP deduplicated // away. A document's recent_sequences carry the sequences it consumed but never emitted; the cache // has to synthesise entries for them or sequence buffering stalls. A sequence at which the document