Set up simulation test for history queue v2's pending task alert (cadence-workflow#7180)

Shaddoll · ins-tril · commit ef889e1f3620 · 2025-08-28T04:42:37.000+07:00
diff --git a/common/dynamicconfig/dynamicproperties/constants.go b/common/dynamicconfig/dynamicproperties/constants.go
@@ -2614,6 +2614,7 @@ const (
 	// Default value: 5s (5*time.Second)
 	// Allowed filters: N/A
 	QueueProcessorPollBackoffInterval
+	VirtualSliceForceAppendInterval
 	// TimerProcessorUpdateAckInterval is update interval for timer processor
 	// KeyName: history.timerProcessorUpdateAckInterval
 	// Value type: Duration
@@ -5129,6 +5130,11 @@ var DurationKeys = map[DurationKey]DynamicDuration{
 		Description:  "QueueProcessorPollBackoffInterval is the backoff duration when queue processor is throttled",
 		DefaultValue: time.Second * 5,
 	},
+	VirtualSliceForceAppendInterval: {
+		KeyName:      "history.virtualSliceForceAppendInterval",
+		Description:  "VirtualSliceForceAppendInterval is the duration forcing a new virtual slice to be appended to the root virtual queue instead of being merged. It has 2 benefits: First, virtual slices won't grow infinitely, task loading for that slice can complete and its scope can be shrinked. Second, when we need to unload a virtual slice to free memory, we won't unload too many tasks.",
+		DefaultValue: time.Minute * 5,
+	},
 	TimerProcessorUpdateAckInterval: {
 		KeyName:      "history.timerProcessorUpdateAckInterval",
 		Description:  "TimerProcessorUpdateAckInterval is update interval for timer processor",
diff --git a/docker/github_actions/docker-compose-local-history-simulation.yml b/docker/github_actions/docker-compose-local-history-simulation.yml
@@ -49,7 +49,7 @@ services:
       - -e
       - -c
       - >
-        go test -timeout 180s
+        go test -timeout 300s
         -run ^TestHistorySimulation.*$
         -count 1
         -v
diff --git a/service/history/config/config.go b/service/history/config/config.go
@@ -119,6 +119,7 @@ type Config struct {
 	EnableTransferQueueV2PendingTaskCountAlert dynamicproperties.BoolPropertyFnWithShardIDFilter
 	QueueCriticalPendingTaskCount              dynamicproperties.IntPropertyFn
 	QueueMaxVirtualQueueCount                  dynamicproperties.IntPropertyFn
+	VirtualSliceForceAppendInterval            dynamicproperties.DurationPropertyFn
 
 	// QueueProcessor settings
 	QueueProcessorEnableSplit                          dynamicproperties.BoolPropertyFn
@@ -417,6 +418,7 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, maxMessageSize int, i
 		EnableTransferQueueV2PendingTaskCountAlert: dc.GetBoolPropertyFilteredByShardID(dynamicproperties.EnableTransferQueueV2PendingTaskCountAlert),
 		QueueCriticalPendingTaskCount:              dc.GetIntProperty(dynamicproperties.QueueCriticalPendingTaskCount),
 		QueueMaxVirtualQueueCount:                  dc.GetIntProperty(dynamicproperties.QueueMaxVirtualQueueCount),
+		VirtualSliceForceAppendInterval:            dc.GetDurationProperty(dynamicproperties.VirtualSliceForceAppendInterval),
 
 		QueueProcessorEnableSplit:                          dc.GetBoolProperty(dynamicproperties.QueueProcessorEnableSplit),
 		QueueProcessorSplitMaxLevel:                        dc.GetIntProperty(dynamicproperties.QueueProcessorSplitMaxLevel),
diff --git a/service/history/config/config_test.go b/service/history/config/config_test.go
@@ -275,6 +275,7 @@ func TestNewConfig(t *testing.T) {
 		"EnableTransferQueueV2PendingTaskCountAlert":           {dynamicproperties.EnableTransferQueueV2PendingTaskCountAlert, true},
 		"QueueCriticalPendingTaskCount":                        {dynamicproperties.QueueCriticalPendingTaskCount, 100},
 		"QueueMaxVirtualQueueCount":                            {dynamicproperties.QueueMaxVirtualQueueCount, 101},
+		"VirtualSliceForceAppendInterval":                      {dynamicproperties.VirtualSliceForceAppendInterval, time.Second},
 	}
 	client := dynamicconfig.NewInMemoryClient()
 	for fieldName, expected := range fields {
diff --git a/service/history/queuev2/queue_base.go b/service/history/queuev2/queue_base.go
@@ -60,6 +60,7 @@ type (
 		MaxPendingTasksCount                 dynamicproperties.IntPropertyFn
 		PollBackoffInterval                  dynamicproperties.DurationPropertyFn
 		PollBackoffIntervalJitterCoefficient dynamicproperties.FloatPropertyFn
+		VirtualSliceForceAppendInterval      dynamicproperties.DurationPropertyFn
 		// monitor & mitigator options
 		CriticalPendingTaskCount    dynamicproperties.IntPropertyFn
 		EnablePendingTaskCountAlert func() bool
@@ -162,22 +163,25 @@ func newQueueBase(
 		timeSource,
 		quotas.NewDynamicRateLimiter(options.MaxPollRPS.AsFloat64()),
 		monitor,
-		&VirtualQueueOptions{
-			PageSize:                             options.PageSize,
-			MaxPendingTasksCount:                 options.MaxPendingTasksCount,
-			PollBackoffInterval:                  options.PollBackoffInterval,
-			PollBackoffIntervalJitterCoefficient: options.PollBackoffIntervalJitterCoefficient,
-		},
-		&VirtualQueueOptions{
-			PageSize: options.PageSize,
-			// non-root queues should not trigger task unloading
-			// otherwise those virtual queues will keep loading, hit pending task count limit, unload, throttle, load, etc...
-			// use a limit lower than the critical pending task count instead
-			MaxPendingTasksCount: func(opts ...dynamicproperties.FilterOption) int {
-				return int(float64(options.CriticalPendingTaskCount(opts...)) * nonRootQueueMaxPendingTaskCoefficient)
+		&VirtualQueueManagerOptions{
+			RootQueueOptions: &VirtualQueueOptions{
+				PageSize:                             options.PageSize,
+				MaxPendingTasksCount:                 options.MaxPendingTasksCount,
+				PollBackoffInterval:                  options.PollBackoffInterval,
+				PollBackoffIntervalJitterCoefficient: options.PollBackoffIntervalJitterCoefficient,
+			},
+			NonRootQueueOptions: &VirtualQueueOptions{
+				PageSize: options.PageSize,
+				// non-root queues should not trigger task unloading
+				// otherwise those virtual queues will keep loading, hit pending task count limit, unload, throttle, load, etc...
+				// use a limit lower than the critical pending task count instead
+				MaxPendingTasksCount: func(opts ...dynamicproperties.FilterOption) int {
+					return int(float64(options.CriticalPendingTaskCount(opts...)) * nonRootQueueMaxPendingTaskCoefficient)
+				},
+				PollBackoffInterval:                  options.PollBackoffInterval,
+				PollBackoffIntervalJitterCoefficient: options.PollBackoffIntervalJitterCoefficient,
 			},
-			PollBackoffInterval:                  options.PollBackoffInterval,
-			PollBackoffIntervalJitterCoefficient: options.PollBackoffIntervalJitterCoefficient,
+			VirtualSliceForceAppendInterval: options.VirtualSliceForceAppendInterval,
 		},
 		queueState.VirtualQueueStates,
 	)
diff --git a/service/history/queuev2/queue_immediate_test.go b/service/history/queuev2/queue_immediate_test.go
@@ -78,6 +78,7 @@ func TestImmediateQueue_LifeCycle(t *testing.T) {
 		MaxPollRPS:                           dynamicproperties.GetIntPropertyFn(100),
 		MaxPendingTasksCount:                 dynamicproperties.GetIntPropertyFn(100),
 		CriticalPendingTaskCount:             dynamicproperties.GetIntPropertyFn(90),
+		VirtualSliceForceAppendInterval:      dynamicproperties.GetDurationPropertyFn(time.Second * 10),
 		EnablePendingTaskCountAlert:          func() bool { return true },
 		MaxVirtualQueueCount:                 dynamicproperties.GetIntPropertyFn(2),
 	}
diff --git a/service/history/queuev2/queue_scheduled_test.go b/service/history/queuev2/queue_scheduled_test.go
@@ -78,6 +78,7 @@ func TestScheduledQueue_LifeCycle(t *testing.T) {
 		MaxPollRPS:                           dynamicproperties.GetIntPropertyFn(100),
 		MaxPendingTasksCount:                 dynamicproperties.GetIntPropertyFn(100),
 		PollBackoffIntervalJitterCoefficient: dynamicproperties.GetFloatPropertyFn(0.0),
+		VirtualSliceForceAppendInterval:      dynamicproperties.GetDurationPropertyFn(time.Second * 10),
 		CriticalPendingTaskCount:             dynamicproperties.GetIntPropertyFn(90),
 		EnablePendingTaskCountAlert:          func() bool { return true },
 		MaxVirtualQueueCount:                 dynamicproperties.GetIntPropertyFn(2),
diff --git a/service/history/queuev2/timer_queue_factory.go b/service/history/queuev2/timer_queue_factory.go
@@ -151,6 +151,7 @@ func (f *timerQueueFactory) createQueuev2(
 			MaxPendingTasksCount:                 config.QueueMaxPendingTaskCount,
 			PollBackoffInterval:                  config.QueueProcessorPollBackoffInterval,
 			PollBackoffIntervalJitterCoefficient: config.QueueProcessorPollBackoffIntervalJitterCoefficient,
+			VirtualSliceForceAppendInterval:      config.VirtualSliceForceAppendInterval,
 			MaxStartJitterInterval:               dynamicproperties.GetDurationPropertyFn(0),
 			RedispatchInterval:                   config.ActiveTaskRedispatchInterval,
 			CriticalPendingTaskCount:             config.QueueCriticalPendingTaskCount,
diff --git a/service/history/queuev2/transfer_queue_factory.go b/service/history/queuev2/transfer_queue_factory.go
@@ -158,6 +158,7 @@ func (f *transferQueueFactory) createQueuev2(
 			MaxPendingTasksCount:                 config.QueueMaxPendingTaskCount,
 			PollBackoffInterval:                  config.QueueProcessorPollBackoffInterval,
 			PollBackoffIntervalJitterCoefficient: config.QueueProcessorPollBackoffIntervalJitterCoefficient,
+			VirtualSliceForceAppendInterval:      config.VirtualSliceForceAppendInterval,
 			EnableValidator:                      config.TransferProcessorEnableValidator,
 			ValidationInterval:                   config.TransferProcessorValidationInterval,
 			MaxStartJitterInterval:               dynamicproperties.GetDurationPropertyFn(0),
diff --git a/service/history/queuev2/virtual_queue.go b/service/history/queuev2/virtual_queue.go
@@ -348,7 +348,7 @@ func (q *virtualQueueImpl) loadAndSubmitTasks() {
 	if q.pauseController.IsPaused() {
 		// emit a metric indicating that the virtual queue is paused
 		q.metricsScope.UpdateGauge(metrics.VirtualQueuePausedGauge, 1.0)
-		q.logger.Debug("virtual queue is paused")
+		q.logger.Debug("virtual queue is paused", tag.PendingTaskCount(pendingTaskCount), tag.MaxTaskCount(maxTaskCount))
 		return
 	}
 
diff --git a/service/history/queuev2/virtual_queue_manager.go b/service/history/queuev2/virtual_queue_manager.go
@@ -30,6 +30,7 @@ import (
 
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/clock"
+	"github.com/uber/cadence/common/dynamicconfig/dynamicproperties"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/metrics"
@@ -39,20 +40,14 @@ import (
 
 const (
 	rootQueueID = 0
-	// Force creating new slice every forceNewSliceDuration
-	// so that the last slice in the default reader won't grow
-	// infinitely.
-	// The benefit of forcing new slice is:
-	// 1. As long as the last slice won't grow infinitly, task loading
-	// for that slice will complete and it's scope (both range and
-	// predicate) is able to shrink
-	// 2. Current task loading implementation can only unload the entire
-	// slice. If there's only one slice, we may unload all tasks for a
-	// given namespace.
-	forceNewSliceDuration = 5 * time.Minute
 )
 
 type (
+	VirtualQueueManagerOptions struct {
+		RootQueueOptions                *VirtualQueueOptions
+		NonRootQueueOptions             *VirtualQueueOptions
+		VirtualSliceForceAppendInterval dynamicproperties.DurationPropertyFn
+	}
 	VirtualQueueManager interface {
 		common.Daemon
 		VirtualQueues() map[int64]VirtualQueue
@@ -73,8 +68,7 @@ type (
 		timeSource          clock.TimeSource
 		taskLoadRateLimiter quotas.Limiter
 		monitor             Monitor
-		rootQueueOptions    *VirtualQueueOptions
-		nonRootQueueOptions *VirtualQueueOptions
+		options             *VirtualQueueManagerOptions
 
 		sync.RWMutex
 		status               int32
@@ -95,8 +89,7 @@ func NewVirtualQueueManager(
 	timeSource clock.TimeSource,
 	taskLoadRateLimiter quotas.Limiter,
 	monitor Monitor,
-	rootQueueOptions *VirtualQueueOptions,
-	nonRootQueueOptions *VirtualQueueOptions,
+	options *VirtualQueueManagerOptions,
 	virtualQueueStates map[int64][]VirtualSliceState,
 ) VirtualQueueManager {
 	virtualQueues := make(map[int64]VirtualQueue)
@@ -105,13 +98,13 @@ func NewVirtualQueueManager(
 		for i, state := range states {
 			virtualSlices[i] = NewVirtualSlice(state, taskInitializer, queueReader, NewPendingTaskTracker())
 		}
-		var options *VirtualQueueOptions
+		var opts *VirtualQueueOptions
 		if queueID == rootQueueID {
-			options = rootQueueOptions
+			opts = options.RootQueueOptions
 		} else {
-			options = nonRootQueueOptions
+			opts = options.NonRootQueueOptions
 		}
-		virtualQueues[queueID] = NewVirtualQueue(processor, redispatcher, logger.WithTags(tag.VirtualQueueID(queueID)), metricsScope, timeSource, taskLoadRateLimiter, monitor, virtualSlices, options)
+		virtualQueues[queueID] = NewVirtualQueue(processor, redispatcher, logger.WithTags(tag.VirtualQueueID(queueID)), metricsScope, timeSource, taskLoadRateLimiter, monitor, virtualSlices, opts)
 	}
 	return &virtualQueueManagerImpl{
 		processor:           processor,
@@ -123,18 +116,17 @@ func NewVirtualQueueManager(
 		timeSource:          timeSource,
 		taskLoadRateLimiter: taskLoadRateLimiter,
 		monitor:             monitor,
-		rootQueueOptions:    rootQueueOptions,
-		nonRootQueueOptions: nonRootQueueOptions,
+		options:             options,
 		status:              common.DaemonStatusInitialized,
 		virtualQueues:       virtualQueues,
 		createVirtualQueueFn: func(queueID int64, s ...VirtualSlice) VirtualQueue {
-			var options *VirtualQueueOptions
+			var opts *VirtualQueueOptions
 			if queueID == rootQueueID {
-				options = rootQueueOptions
+				opts = options.RootQueueOptions
 			} else {
-				options = nonRootQueueOptions
+				opts = options.NonRootQueueOptions
 			}
-			return NewVirtualQueue(processor, redispatcher, logger.WithTags(tag.VirtualQueueID(queueID)), metricsScope, timeSource, taskLoadRateLimiter, monitor, s, options)
+			return NewVirtualQueue(processor, redispatcher, logger.WithTags(tag.VirtualQueueID(queueID)), metricsScope, timeSource, taskLoadRateLimiter, monitor, s, opts)
 		},
 	}
 }
@@ -230,7 +222,7 @@ func (m *virtualQueueManagerImpl) appendOrMergeSlice(vq VirtualQueue, s VirtualS
 	now := m.timeSource.Now()
 	if now.After(m.nextForceNewSliceTime) {
 		vq.AppendSlices(s)
-		m.nextForceNewSliceTime = now.Add(forceNewSliceDuration)
+		m.nextForceNewSliceTime = now.Add(m.options.VirtualSliceForceAppendInterval())
 		return
 	}
 	vq.MergeSlices(s)
diff --git a/service/history/queuev2/virtual_queue_manager_test.go b/service/history/queuev2/virtual_queue_manager_test.go
@@ -2,6 +2,7 @@ package queuev2
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/mock/gomock"
@@ -195,15 +196,17 @@ func TestVirtualQueueManager_VirtualQueues(t *testing.T) {
 
 			// Create manager instance
 			manager := &virtualQueueManagerImpl{
-				processor:        mockProcessor,
-				taskInitializer:  mockTaskInitializer,
-				redispatcher:     mockRedispatcher,
-				queueReader:      mockQueueReader,
-				logger:           mockLogger,
-				metricsScope:     mockMetricsScope,
-				rootQueueOptions: &VirtualQueueOptions{},
-				nonRootQueueOptions: &VirtualQueueOptions{
-					PageSize: dynamicproperties.GetIntPropertyFn(100),
+				processor:       mockProcessor,
+				taskInitializer: mockTaskInitializer,
+				redispatcher:    mockRedispatcher,
+				queueReader:     mockQueueReader,
+				logger:          mockLogger,
+				metricsScope:    mockMetricsScope,
+				options: &VirtualQueueManagerOptions{
+					RootQueueOptions: &VirtualQueueOptions{},
+					NonRootQueueOptions: &VirtualQueueOptions{
+						PageSize: dynamicproperties.GetIntPropertyFn(100),
+					},
 				},
 				status:        common.DaemonStatusInitialized,
 				virtualQueues: virtualQueues,
@@ -447,15 +450,17 @@ func TestVirtualQueueManager_UpdateAndGetState(t *testing.T) {
 
 			// Create manager instance
 			manager := &virtualQueueManagerImpl{
-				processor:        mockProcessor,
-				taskInitializer:  mockTaskInitializer,
-				redispatcher:     mockRedispatcher,
-				queueReader:      mockQueueReader,
-				logger:           mockLogger,
-				metricsScope:     mockMetricsScope,
-				rootQueueOptions: &VirtualQueueOptions{},
-				nonRootQueueOptions: &VirtualQueueOptions{
-					MaxPendingTasksCount: dynamicproperties.GetIntPropertyFn(100),
+				processor:       mockProcessor,
+				taskInitializer: mockTaskInitializer,
+				redispatcher:    mockRedispatcher,
+				queueReader:     mockQueueReader,
+				logger:          mockLogger,
+				metricsScope:    mockMetricsScope,
+				options: &VirtualQueueManagerOptions{
+					RootQueueOptions: &VirtualQueueOptions{},
+					NonRootQueueOptions: &VirtualQueueOptions{
+						MaxPendingTasksCount: dynamicproperties.GetIntPropertyFn(100),
+					},
 				},
 				status:        common.DaemonStatusInitialized,
 				virtualQueues: virtualQueues,
@@ -554,6 +559,7 @@ func TestVirtualQueueManager_AddNewVirtualSlice(t *testing.T) {
 			// Set up mock expectations
 			tt.setupMockQueues(mockQueues, mockSlice)
 
+			forceNewSliceDuration := time.Minute
 			// Create manager instance
 			manager := &virtualQueueManagerImpl{
 				processor:       mockProcessor,
@@ -563,11 +569,14 @@ func TestVirtualQueueManager_AddNewVirtualSlice(t *testing.T) {
 				logger:          mockLogger,
 				metricsScope:    mockMetricsScope,
 				timeSource:      mockTimeSource,
-				rootQueueOptions: &VirtualQueueOptions{
-					PageSize: dynamicproperties.GetIntPropertyFn(100),
-				},
-				nonRootQueueOptions: &VirtualQueueOptions{
-					PageSize: dynamicproperties.GetIntPropertyFn(100),
+				options: &VirtualQueueManagerOptions{
+					RootQueueOptions: &VirtualQueueOptions{
+						PageSize: dynamicproperties.GetIntPropertyFn(100),
+					},
+					NonRootQueueOptions: &VirtualQueueOptions{
+						PageSize: dynamicproperties.GetIntPropertyFn(100),
+					},
+					VirtualSliceForceAppendInterval: dynamicproperties.GetDurationPropertyFn(forceNewSliceDuration),
 				},
 				status:        common.DaemonStatusInitialized,
 				virtualQueues: virtualQueues,
diff --git a/simulation/history/dynamicconfig/queuev2_split.yaml b/simulation/history/dynamicconfig/queuev2_split.yaml
diff --git a/simulation/history/history_simulation_test.go b/simulation/history/history_simulation_test.go
diff --git a/simulation/history/testdata/history_simulation_queuev2_split.yaml b/simulation/history/testdata/history_simulation_queuev2_split.yaml

Original file line number	Diff line number	Diff line change
`@@ -275,6 +275,7 @@ func TestNewConfig(t *testing.T) {`
`275`	`275`	`"EnableTransferQueueV2PendingTaskCountAlert": {dynamicproperties.EnableTransferQueueV2PendingTaskCountAlert, true},`
`276`	`276`	`"QueueCriticalPendingTaskCount": {dynamicproperties.QueueCriticalPendingTaskCount, 100},`
`277`	`277`	`"QueueMaxVirtualQueueCount": {dynamicproperties.QueueMaxVirtualQueueCount, 101},`
	`278`	`+ "VirtualSliceForceAppendInterval": {dynamicproperties.VirtualSliceForceAppendInterval, time.Second},`
`278`	`279`	`}`
`279`	`280`	`client := dynamicconfig.NewInMemoryClient()`
`280`	`281`	`for fieldName, expected := range fields {`
Original file line number	Diff line number	Diff line change
`@@ -78,6 +78,7 @@ func TestImmediateQueue_LifeCycle(t *testing.T) {`
`78`	`78`	`MaxPollRPS: dynamicproperties.GetIntPropertyFn(100),`
`79`	`79`	`MaxPendingTasksCount: dynamicproperties.GetIntPropertyFn(100),`
`80`	`80`	`CriticalPendingTaskCount: dynamicproperties.GetIntPropertyFn(90),`
	`81`	`+ VirtualSliceForceAppendInterval: dynamicproperties.GetDurationPropertyFn(time.Second * 10),`
`81`	`82`	`EnablePendingTaskCountAlert: func() bool { return true },`
`82`	`83`	`MaxVirtualQueueCount: dynamicproperties.GetIntPropertyFn(2),`
`83`	`84`	`}`
Original file line number	Diff line number	Diff line change
`@@ -348,7 +348,7 @@ func (q *virtualQueueImpl) loadAndSubmitTasks() {`
`348`	`348`	`if q.pauseController.IsPaused() {`
`349`	`349`	`// emit a metric indicating that the virtual queue is paused`
`350`	`350`	`q.metricsScope.UpdateGauge(metrics.VirtualQueuePausedGauge, 1.0)`
`351`		`- q.logger.Debug("virtual queue is paused")`
	`351`	`+ q.logger.Debug("virtual queue is paused", tag.PendingTaskCount(pendingTaskCount), tag.MaxTaskCount(maxTaskCount))`
`352`	`352`	`return`
`353`	`353`	`}`
`354`	`354`