[4.1.2 Backport] CBG-5614: lateLogs compaction (#8634)

Co-authored-by: Gregory Newman-Smith <gregory.newmansmith@couchbase.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: 4 people

base/stats.go

@@ -455,6 +455,11 @@ type CacheStats struct {
 	ChannelCacheChannelsEvictedInactive *SgwIntStat `json:"chan_cache_channels_evicted_inactive"`
 	// The total number of active channel cache channels evicted, based on ‘not recently used’ criteria.
 	ChannelCacheChannelsEvictedNRU *SgwIntStat `json:"chan_cache_channels_evicted_nru"`
+	// The total number of entries currently held across all channels' late-arriving-sequence queues (lateLogs).
+	NumEntriesInLateFeed *SgwIntStat `json:"num_entries_in_late_feed"`
+	// The total number of times a continuous _changes feed was forced to roll back to its low sequence because
+	// its lastSequence was pruned from a channel's lateLogs (length/age cap firing on a lagging feed).
+	LateFeedForcedRollbacks *SgwIntStat `json:"late_feed_forced_rollbacks"`
 	// The total number of channel cache compaction runs.
 	ChannelCacheCompactCount *SgwIntStat `json:"chan_cache_compact_count"`
 	// The total amount of time taken by channel cache compaction across all compaction runs.
@@ -1481,6 +1486,14 @@ func (d *DbStats) initCacheStats() error {
 	if err != nil {
 		return err
 	}
+	resUtil.NumEntriesInLateFeed, err = NewIntStat(SubsystemCacheKey, "num_entries_in_late_feed", StatUnitNoUnits, NumEntriesInLateFeedDesc, StatAddedVersion4dot1dot0, StatDeprecatedVersionNotDeprecated, StatStabilityInternal, labelKeys, labelVals, prometheus.GaugeValue, 0)
+	if err != nil {
+		return err
+	}
+	resUtil.LateFeedForcedRollbacks, err = NewIntStat(SubsystemCacheKey, "late_feed_forced_rollbacks", StatUnitNoUnits, LateFeedForcedRollbacksDesc, StatAddedVersion4dot1dot0, StatDeprecatedVersionNotDeprecated, StatStabilityInternal, labelKeys, labelVals, prometheus.CounterValue, 0)
+	if err != nil {
+		return err
+	}
 	resUtil.ChannelCacheCompactCount, err = NewIntStat(SubsystemCacheKey, "chan_cache_compact_count", StatUnitNoUnits, ChanCacheCompactCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
@@ -1597,6 +1610,8 @@ func (d *DbStats) unregisterCacheStats() {
 	prometheus.Unregister(d.CacheStats.ChannelCacheChannelsAdded)
 	prometheus.Unregister(d.CacheStats.ChannelCacheChannelsEvictedInactive)
 	prometheus.Unregister(d.CacheStats.ChannelCacheChannelsEvictedNRU)
+	prometheus.Unregister(d.CacheStats.NumEntriesInLateFeed)
+	prometheus.Unregister(d.CacheStats.LateFeedForcedRollbacks)
 	prometheus.Unregister(d.CacheStats.ChannelCacheCompactCount)
 	prometheus.Unregister(d.CacheStats.ChannelCacheCompactTime)
 	prometheus.Unregister(d.CacheStats.ChannelCacheHits)

base/stats_descriptions.go

@@ -98,6 +98,10 @@ const (
 
 	ChanCacheChannelsEvictedNRUDesc = "The total number of active channel cache channels evicted, based on 'not recently used' criteria."
 
+	NumEntriesInLateFeedDesc = "The total number of late-arriving-sequence entries currently held across all channels."
+
+	LateFeedForcedRollbacksDesc = "The total number of times a continuous _changes feed was forced to roll back to its low sequence because its lastSequence had been pruned from a channel's lateLogs by the length or age cap."
+
 	ChanCacheCompactCountDesc = "The total number of channel cache compaction runs."
 
 	ChanCacheCompactTimeDesc = "The total amount of time taken by channel cache compaction across all compaction runs."

db/change_cache.go

@@ -124,13 +124,20 @@ type CacheOptions struct {
 	CachePendingSeqMaxWait time.Duration // Max wait for pending sequence before skipping
 	CachePendingSeqMaxNum  int           // Max number of pending sequences before skipping
 	CacheSkippedSeqMaxWait time.Duration // Max wait for skipped sequence before abandoning
+	// Broadcast ticker intervals for waking continuous _changes feeds. Zero means use the package default
+	// (DefaultBroadcastChangesTime / SkippedSequenceBroadcastChangesTime). Internal tuning knob - primarily
+	// so tests can avoid the 500ms skipped-sequence slow mode dominating runtime. Not exposed via REST config.
+	BroadcastChangesInterval         time.Duration // Normal-mode feed broadcast interval
+	SkippedSequenceBroadcastInterval time.Duration // Slow-mode feed broadcast interval (skipped sequences present)
 }
 
 func DefaultCacheOptions() CacheOptions {
 	return CacheOptions{
-		CachePendingSeqMaxWait: DefaultCachePendingSeqMaxWait,
-		CachePendingSeqMaxNum:  DefaultCachePendingSeqMaxNum,
-		CacheSkippedSeqMaxWait: DefaultSkippedSeqMaxWait,
+		CachePendingSeqMaxWait:           DefaultCachePendingSeqMaxWait,
+		CachePendingSeqMaxNum:            DefaultCachePendingSeqMaxNum,
+		CacheSkippedSeqMaxWait:           DefaultSkippedSeqMaxWait,
+		BroadcastChangesInterval:         DefaultBroadcastChangesTime,
+		SkippedSequenceBroadcastInterval: SkippedSequenceBroadcastChangesTime,
 		ChannelCacheOptions: ChannelCacheOptions{
 			ChannelCacheAge:             DefaultChannelCacheAge,
 			ChannelCacheMinLength:       DefaultChannelCacheMinLength,
@@ -139,6 +146,7 @@ func DefaultCacheOptions() CacheOptions {
 			CompactHighWatermarkPercent: DefaultCompactHighWatermarkPercent,
 			CompactLowWatermarkPercent:  DefaultCompactLowWatermarkPercent,
 			ChannelQueryLimit:           DefaultQueryPaginationLimit,
+			LateLogAge:                  DefaultLateLogAge,
 		},
 	}
 }

db/change_listener.go

@@ -257,9 +257,9 @@ func (listener *changeListener) Notify(ctx context.Context, keys channels.Set) {
 }
 
 func (listener *changeListener) StartNotifierBroadcaster(ctx context.Context) {
-	ticker := time.NewTicker(DefaultBroadcastChangesTime)
-	// boolean to indicate whether ticker is using the default value, this is needed so we don't call reset on ticker
-	// for a value it already has
+	ticker := time.NewTicker(listener.broadcastInterval(false))
+	// Tracks whether we're currently using the slow-mode (skipped sequences present) interval, so we only
+	// Reset the ticker when the mode changes.
 	broadcastSlowMode := false
 	go func(terminator chan bool, doneChan chan struct{}) {
 		defer func() {
@@ -285,7 +285,7 @@ func (listener *changeListener) StartNotifierBroadcaster(ctx context.Context) {
 				newBroadcastSlowMode := listener.dbCtx.BroadcastSlowMode.Load()
 				if broadcastSlowMode != newBroadcastSlowMode {
 					// broadcast changes interval has changed, reset ticker
-					duration := tickerValForBroadcastSpeed(newBroadcastSlowMode)
+					duration := listener.broadcastInterval(newBroadcastSlowMode)
 					base.DebugfCtx(ctx, base.KeyChanges, "Updating broadcast changes interval for %q to %v", base.MD(listener.bucketName), duration)
 					broadcastSlowMode = newBroadcastSlowMode
 					ticker.Reset(duration)
@@ -295,14 +295,20 @@ func (listener *changeListener) StartNotifierBroadcaster(ctx context.Context) {
 	}(listener.terminator, listener.broadcastChangesDoneChan)
 }
 
-// tickerValForBroadcastSpeed will return the duration for the ticker to be reset to based on input boolean to indicate
-// if skipped sequences are present or not
-func tickerValForBroadcastSpeed(skippedSequencePresent bool) time.Duration {
-	// if the skipped sequence broadcast is enabled, return the slow ticker value
+// broadcastInterval returns the duration for the broadcast ticker based on whether skipped sequences are
+// present, honoring the optional per-DB CacheOptions override and falling back to the package defaults when
+// unset. The override is an internal tuning (primarily to keep tests fast); production uses the defaults.
+func (listener *changeListener) broadcastInterval(skippedSequencePresent bool) time.Duration {
+	opts := listener.dbCtx.Options.CacheOptions
 	if skippedSequencePresent {
+		if opts != nil && opts.SkippedSequenceBroadcastInterval > 0 {
+			return opts.SkippedSequenceBroadcastInterval
+		}
 		return SkippedSequenceBroadcastChangesTime
 	}
-	// otherwise return the default ticker value
+	if opts != nil && opts.BroadcastChangesInterval > 0 {
+		return opts.BroadcastChangesInterval
+	}
 	return DefaultBroadcastChangesTime
 }
 

db/channel_cache.go

@@ -31,6 +31,7 @@ var (
 	DefaultChannelCacheMaxNumber       = 50000            // Default of 50k channel caches
 	DefaultCompactHighWatermarkPercent = 80               // Default compaction high watermark (percent of MaxNumber)
 	DefaultCompactLowWatermarkPercent  = 60               // Default compaction low watermark (percent of MaxNumber)
+	DefaultLateLogAge                  = 5 * time.Minute  // Force-prune late-arriving entries older than this, even if a feed is still parked on them
 )
 
 type ChannelCache interface {
@@ -131,6 +132,18 @@ func newChannelCache(ctx context.Context, dbName string, options ChannelCacheOpt
 		return nil, err
 	}
 	channelCache.backgroundTasks = append(channelCache.backgroundTasks, bgt)
+
+	// Late-log pruning runs on its own task at LateLogAge, independent of the (potentially much larger)
+	// ChannelCacheAge, so LateLogAge is honored regardless of how ChannelCacheAge is set.
+	lateLogAge := options.LateLogAge
+	if lateLogAge <= 0 {
+		lateLogAge = DefaultLateLogAge
+	}
+	lateLogBGT, err := NewBackgroundTask(ctx, "CleanAgedLateLogs", channelCache.cleanAgedLateLogs, lateLogAge, channelCache.terminator)
+	if err != nil {
+		return nil, err
+	}
+	channelCache.backgroundTasks = append(channelCache.backgroundTasks, lateLogBGT)
 	base.DebugfCtx(ctx, base.KeyCache, "Initialized channel cache with maxChannels:%d, HWM: %d, LWM: %d",
 		channelCache.maxChannels, channelCache.compactHighWatermark, channelCache.compactLowWatermark)
 	return channelCache, nil
@@ -139,6 +152,9 @@ func newChannelCache(ctx context.Context, dbName string, options ChannelCacheOpt
 func (c *channelCacheImpl) Clear() {
 	c.seqLock.Lock()
 	c.channelCaches.Init()
+	// All channel caches (and their lateLogs, including sentinels) have just been dropped, so reset the
+	// late-feed gauge rather than leaking it across change-cache reinitialization.
+	c.cacheStats.NumEntriesInLateFeed.Set(0)
 	c.seqLock.Unlock()
 }
 
@@ -315,6 +331,24 @@ func (c *channelCacheImpl) cleanAgedItems(ctx context.Context) error {
 	return nil
 }
 
+// cleanAgedLateLogs prunes each channel's late-arriving sequence queue by LateLogAge.  It runs on its own
+// background task (separate from cleanAgedItems) so late_log_expiry_seconds is honored on its own cadence,
+// independent of the channel cache's expiry_seconds.  Error returned to fulfill BackgroundTaskFunc signature.
+func (c *channelCacheImpl) cleanAgedLateLogs(ctx context.Context) error {
+
+	callback := func(v any) bool {
+		channelCache := AsSingleChannelCache(ctx, v)
+		if channelCache == nil {
+			return false
+		}
+		channelCache.pruneLateLogAge(ctx)
+		return true
+	}
+	c.channelCaches.Range(callback)
+
+	return nil
+}
+
 func (c *channelCacheImpl) getChannelCache(ctx context.Context, channel channels.ID) (SingleChannelCache, error) {
 
 	cacheValue, found := c.channelCaches.Get(channel)
@@ -396,6 +430,11 @@ func (c *channelCacheImpl) addChannelCache(ctx context.Context, channel channels
 	cacheValue, created, cacheSize := c.channelCaches.GetOrInsert(channel, singleChannelCache)
 	c.validFromLock.Unlock()
 
+	// If another goroutine won the insert race our freshly-built cache is discarded, but it never contributed to
+	// NumEntriesInLateFeed: initializeLateLogs doesn't count the retained placeholder, and since the discarded
+	// instance was never in channelCaches no late arrival could have been added to it. So there's nothing to
+	// release here.
+
 	singleChannelCache = AsSingleChannelCache(ctx, cacheValue)
 
 	if cacheSize > c.compactHighWatermark {
@@ -547,6 +586,18 @@ func (c *channelCacheImpl) compactChannelCache(ctx context.Context) {
 
 		cacheSize = c.channelCaches.RemoveElements(evictionElements)
 
+		// Evicted channel caches take their lateLogs entries (including the retained placeholder) with them, and
+		// those entries never go through the per-entry purge paths, so release each cache's contribution to the
+		// late-feed gauge here to stop NumEntriesInLateFeed from leaking upward as channels are evicted.
+		// releaseLateLogsForEviction reads-and-decrements atomically under the cache's lateLogLock (and runs
+		// after RemoveElements, so any add that landed just before removal is captured by the count read),
+		// rather than subtracting a pre-removal snapshot that could race concurrent adds/prunes.
+		for _, elem := range evictionElements {
+			if scc, ok := elem.Value.(*singleChannelCacheImpl); ok {
+				scc.releaseLateLogsForEviction()
+			}
+		}
+
 		// Update eviction stats
 		c.updateEvictionStats(inactiveEvictCount, len(evictionElements), compactIterationStart)