CBG-5951: make changeCache.initialSequence atomic

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
 }
 

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