feat(queuev2): add read level sync loop to cachedScheduledQueue

Adds a readLevelSyncLoop goroutine to cachedScheduledQueue that ticks
every TimerProcessorCacheReadLevelSyncInterval (default 1s) and calls
reader.UpdateReadLevel(virtualQueueManager.GetMinReadLevel()).

This keeps the cache lower bound within ~1s of actual processing
progress, compared to the previous ~30s lag from updateQueueStateFn
which is gated on DB writes.

New config: TimerProcessorCacheReadLevelSyncInterval (global, default 1s).

Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>

Author: arzonus

common/dynamicconfig/dynamicproperties/constants.go

@@ -3064,6 +3064,14 @@ const (
 	// Default value: 1s (1*time.Second)
 	// Allowed filters: N/A
 	TimerProcessorCacheTimeEvictionInterval
+	// TimerProcessorCacheReadLevelSyncInterval is how often the cached scheduled
+	// queue syncs the read level from the virtual queue manager to the cache reader,
+	// evicting tasks the processor has already passed.
+	// KeyName: history.timerProcessorCacheReadLevelSyncInterval
+	// Value type: Duration
+	// Default value: 1s (1*time.Second)
+	// Allowed filters: N/A
+	TimerProcessorCacheReadLevelSyncInterval
 	// TransferProcessorFailoverMaxStartJitterInterval is the max jitter interval for starting transfer
 	// failover queue processing. The actual jitter interval used will be a random duration between
 	// 0 and the max interval so that timer failover queue across different shards won't start at
@@ -5754,6 +5762,11 @@ var DurationKeys = map[DurationKey]DynamicDuration{
 		Description:  "TimerProcessorCacheTimeEvictionInterval is how often the time-based eviction loop fires",
 		DefaultValue: time.Second,
 	},
+	TimerProcessorCacheReadLevelSyncInterval: {
+		KeyName:      "history.timerProcessorCacheReadLevelSyncInterval",
+		Description:  "TimerProcessorCacheReadLevelSyncInterval is how often the cached scheduled queue syncs the read level to the cache reader",
+		DefaultValue: time.Second,
+	},
 	TransferProcessorFailoverMaxStartJitterInterval: {
 		KeyName:      "history.transferProcessorFailoverMaxStartJitterInterval",
 		Description:  "TransferProcessorFailoverMaxStartJitterInterval is the max jitter interval for starting transfer failover queue processing. The actual jitter interval used will be a random duration between 0 and the max interval so that timer failover queue across different shards won't start at the same time",

service/history/config/config.go

@@ -166,6 +166,7 @@ type Config struct {
 	TimerProcessorCacheEvictionSafeWindow             dynamicproperties.DurationPropertyFn
 	TimerProcessorCacheMinPrefetchInterval            dynamicproperties.DurationPropertyFn
 	TimerProcessorCacheTimeEvictionInterval           dynamicproperties.DurationPropertyFn
+	TimerProcessorCacheReadLevelSyncInterval          dynamicproperties.DurationPropertyFn
 
 	// TransferQueueProcessor settings
 	TransferTaskBatchSize                                dynamicproperties.IntPropertyFn
@@ -481,6 +482,7 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, maxMessageSize int, i
 		TimerProcessorCacheEvictionSafeWindow:                dc.GetDurationProperty(dynamicproperties.TimerProcessorCacheEvictionSafeWindow),
 		TimerProcessorCacheMinPrefetchInterval:               dc.GetDurationProperty(dynamicproperties.TimerProcessorCacheMinPrefetchInterval),
 		TimerProcessorCacheTimeEvictionInterval:              dc.GetDurationProperty(dynamicproperties.TimerProcessorCacheTimeEvictionInterval),
+		TimerProcessorCacheReadLevelSyncInterval:             dc.GetDurationProperty(dynamicproperties.TimerProcessorCacheReadLevelSyncInterval),
 		TransferTaskBatchSize:                                dc.GetIntProperty(dynamicproperties.TransferTaskBatchSize),
 		TransferTaskDeleteBatchSize:                          dc.GetIntProperty(dynamicproperties.TransferTaskDeleteBatchSize),
 		TransferProcessorFailoverMaxStartJitterInterval:      dc.GetDurationProperty(dynamicproperties.TransferProcessorFailoverMaxStartJitterInterval),

service/history/config/config_test.go

@@ -148,6 +148,7 @@ func TestNewConfig(t *testing.T) {
 		"TimerProcessorCacheEvictionSafeWindow":                {dynamicproperties.TimerProcessorCacheEvictionSafeWindow, time.Second},
 		"TimerProcessorCacheMinPrefetchInterval":               {dynamicproperties.TimerProcessorCacheMinPrefetchInterval, time.Second},
 		"TimerProcessorCacheTimeEvictionInterval":              {dynamicproperties.TimerProcessorCacheTimeEvictionInterval, time.Second},
+		"TimerProcessorCacheReadLevelSyncInterval":             {dynamicproperties.TimerProcessorCacheReadLevelSyncInterval, time.Second},
 		"TransferTaskBatchSize":                                {dynamicproperties.TransferTaskBatchSize, 47},
 		"TransferTaskDeleteBatchSize":                          {dynamicproperties.TransferTaskDeleteBatchSize, 48},
 		"TransferProcessorCompleteTransferFailureRetryCount":   {dynamicproperties.TransferProcessorCompleteTransferFailureRetryCount, 49},

service/history/queuev2/queue_scheduled_cached.go

@@ -24,13 +24,20 @@ package queuev2
 
 import (
 	"context"
+	"sync"
 
+	"github.com/uber/cadence/common/clock"
+	"github.com/uber/cadence/common/dynamicconfig/dynamicproperties"
 	hcommon "github.com/uber/cadence/service/history/common"
 )
 
 type cachedScheduledQueue struct {
 	*scheduledQueue
-	reader CachedQueueReader
+	reader              CachedQueueReader
+	readLevelSyncCancel context.CancelFunc
+	readLevelSyncWg     sync.WaitGroup
+	readLevelInterval   dynamicproperties.DurationPropertyFn
+	clock               clock.TimeSource
 }
 
 func newCachedScheduledQueue(inner *scheduledQueue, reader CachedQueueReader) Queue {
@@ -44,7 +51,13 @@ func newCachedScheduledQueue(inner *scheduledQueue, reader CachedQueueReader) Qu
 		reader.UpdateReadLevel(inner.base.virtualQueueManager.GetMinReadLevel())
 	}
 
-	return &cachedScheduledQueue{scheduledQueue: inner, reader: reader}
+	config := inner.base.shard.GetConfig()
+	return &cachedScheduledQueue{
+		scheduledQueue:    inner,
+		reader:            reader,
+		readLevelInterval: config.TimerProcessorCacheReadLevelSyncInterval,
+		clock:             inner.base.timeSource,
+	}
 }
 
 func (q *cachedScheduledQueue) NotifyNewTask(clusterName string, info *hcommon.NotifyTaskInfo) {
@@ -55,9 +68,38 @@ func (q *cachedScheduledQueue) NotifyNewTask(clusterName string, info *hcommon.N
 func (q *cachedScheduledQueue) Start() {
 	q.reader.Start()
 	q.scheduledQueue.Start()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	q.readLevelSyncCancel = cancel
+	q.readLevelSyncWg.Add(1)
+	go q.readLevelSyncLoop(ctx)
 }
 
 func (q *cachedScheduledQueue) Stop() {
+	q.readLevelSyncCancel()
+	q.readLevelSyncWg.Wait()
+
 	q.scheduledQueue.Stop()
 	q.reader.Stop()
 }
+
+// readLevelSyncLoop periodically syncs the virtual queue manager's current read
+// level to the cache reader, evicting tasks the processor has already passed.
+// This runs more frequently than updateQueueStateFn (which is gated on DB writes)
+// so the cache lower bound stays close to actual processing progress.
+func (q *cachedScheduledQueue) readLevelSyncLoop(ctx context.Context) {
+	defer q.readLevelSyncWg.Done()
+
+	timer := q.clock.NewTimer(q.readLevelInterval())
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.Chan():
+			q.reader.UpdateReadLevel(q.base.virtualQueueManager.GetMinReadLevel())
+			timer.Reset(q.readLevelInterval())
+		}
+	}
+}