Add observability for multicursor queue-state resolution loss

common/metrics/metric_defs.go

@@ -724,6 +724,9 @@ var (
 	HistoryCount                     = NewDimensionlessHistogramDef("history_count")
 	TasksCompletedPerShardInfoUpdate = NewDimensionlessHistogramDef("tasks_per_shardinfo_update")
 	TimeBetweenShardInfoUpdates      = NewTimerDef("time_between_shardinfo_update")
+	ShardInfoSize                    = NewBytesHistogramDef("shard_info_size")
+	QueueStateSize                   = NewBytesHistogramDef("queue_state_size")
+	QueueStateSizeTotal              = NewCounterDef("queue_state_size_total")
 	SearchAttributesSize             = NewBytesHistogramDef("search_attributes_size")
 	MemoSize                         = NewBytesHistogramDef("memo_size")
 	TooManyPendingChildWorkflows     = NewCounterDef(
@@ -953,6 +956,8 @@ var (
 	QueueScheduleLatency         = NewTimerDef("queue_latency_schedule") // latency for scheduling 100 tasks in one task channel
 	QueueReaderCountHistogram    = NewDimensionlessHistogramDef("queue_reader_count")
 	QueueSliceCountHistogram     = NewDimensionlessHistogramDef("queue_slice_count")
+	QueueSliceCountTotal         = NewCounterDef("queue_slice_count_total")
+	QueueSlicePendingKeys        = NewDimensionlessHistogramDef("queue_slice_pending_keys")
 	QueueActionCounter           = NewCounterDef("queue_actions")
 	QueuePredicateResolutionLoss = NewCounterDef(
 		"queue_predicate_resolution_loss",

service/history/queues/queue_base.go

@@ -329,7 +329,12 @@ func (p *queueBase) checkpoint() {
 		}
 	}
 	metrics.QueueReaderCountHistogram.With(p.metricsHandler).Record(int64(len(readerScopes)))
-	metrics.QueueSliceCountHistogram.With(p.metricsHandler).Record(int64(p.monitor.GetTotalSliceCount()))
+	sliceCount := int64(p.monitor.GetTotalSliceCount())
+	categoryTag := metrics.TaskCategoryTag(p.category.Name())
+	// The counter is a true accumulator; the histogram's _sum is not, since tally's Prometheus
+	// reporter replays each sample as its bucket's upper bound, not the recorded value.
+	metrics.QueueSliceCountHistogram.With(p.metricsHandler).Record(sliceCount, categoryTag)
+	metrics.QueueSliceCountTotal.With(p.metricsHandler).Record(sliceCount, categoryTag)
 	metrics.PendingTasksCounter.With(p.metricsHandler).Record(int64(p.monitor.GetTotalPendingTaskCount()))
 
 	// NOTE: Must range-complete task first.

service/history/queues/queue_base_test.go

@@ -17,6 +17,7 @@ import (
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/metrics/metricstest"
 	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/predicates"
@@ -455,6 +456,102 @@ func (s *queueBaseSuite) TestCheckPoint_NoPendingTasks() {
 	s.True(exclusiveReaderHighWatermark.CompareTo(base.exclusiveDeletionHighWatermark) == 0)
 }
 
+func (s *queueBaseSuite) TestCheckPoint_RecordsSliceCountWithTaskCategoryTag() {
+	numSlices := 3
+	scopes := NewRandomScopes(numSlices)
+	queueState := &queueState{
+		readerScopes: map[int64][]Scope{
+			DefaultReaderId: scopes,
+		},
+		exclusiveReaderHighWatermark: tasks.MaximumKey,
+	}
+	persistenceState := ToPersistenceQueueState(queueState)
+
+	mockShard := shard.NewTestContext(
+		s.controller,
+		&persistencespb.ShardInfo{
+			ShardId: 0,
+			RangeId: 10,
+			QueueStates: map[int32]*persistencespb.QueueState{
+				int32(tasks.CategoryIDTimer): persistenceState,
+			},
+		},
+		s.config,
+	)
+	mockShard.Resource.ClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
+	mockShard.Resource.ClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes()
+
+	captureHandler := metricstest.NewCaptureHandler()
+	capture := captureHandler.StartCapture()
+	defer captureHandler.StopCapture(capture)
+	s.metricsHandler = captureHandler
+
+	base := s.newQueueBase(mockShard, tasks.CategoryTimer, nil)
+	base.checkpointTimer = time.NewTimer(s.options.CheckpointInterval())
+
+	// set to a smaller value so that delete will be triggered, matching TestCheckPoint_SlicePredicateAction
+	base.exclusiveDeletionHighWatermark = tasks.MinimumKey
+
+	mockShard.Resource.ExecutionMgr.EXPECT().RangeCompleteHistoryTasks(gomock.Any(), gomock.Any()).Return(nil).Times(1)
+	mockShard.Resource.ShardMgr.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1)
+
+	base.checkpoint()
+
+	snapshot := capture.Snapshot()
+	recordings := snapshot[metrics.QueueSliceCountHistogram.Name()]
+	s.Require().Len(recordings, 1)
+	s.Equal(int64(numSlices), recordings[0].Value)
+	s.Equal(tasks.CategoryTimer.Name(), recordings[0].Tags["task_category"])
+}
+
+func (s *queueBaseSuite) TestCheckPoint_RecordsSliceCountTotal() {
+	numSlices := 3
+	scopes := NewRandomScopes(numSlices)
+	queueState := &queueState{
+		readerScopes: map[int64][]Scope{
+			DefaultReaderId: scopes,
+		},
+		exclusiveReaderHighWatermark: tasks.MaximumKey,
+	}
+	persistenceState := ToPersistenceQueueState(queueState)
+
+	mockShard := shard.NewTestContext(
+		s.controller,
+		&persistencespb.ShardInfo{
+			ShardId: 0,
+			RangeId: 10,
+			QueueStates: map[int32]*persistencespb.QueueState{
+				int32(tasks.CategoryIDTimer): persistenceState,
+			},
+		},
+		s.config,
+	)
+	mockShard.Resource.ClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
+	mockShard.Resource.ClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes()
+
+	captureHandler := metricstest.NewCaptureHandler()
+	capture := captureHandler.StartCapture()
+	defer captureHandler.StopCapture(capture)
+	s.metricsHandler = captureHandler
+
+	base := s.newQueueBase(mockShard, tasks.CategoryTimer, nil)
+	base.checkpointTimer = time.NewTimer(s.options.CheckpointInterval())
+
+	// set to a smaller value so that delete will be triggered, matching TestCheckPoint_SlicePredicateAction
+	base.exclusiveDeletionHighWatermark = tasks.MinimumKey
+
+	mockShard.Resource.ExecutionMgr.EXPECT().RangeCompleteHistoryTasks(gomock.Any(), gomock.Any()).Return(nil).Times(1)
+	mockShard.Resource.ShardMgr.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1)
+
+	base.checkpoint()
+
+	snapshot := capture.Snapshot()
+	recordings := snapshot[metrics.QueueSliceCountTotal.Name()]
+	s.Require().Len(recordings, 1)
+	s.Equal(int64(numSlices), recordings[0].Value)
+	s.Equal(tasks.CategoryTimer.Name(), recordings[0].Tags["task_category"])
+}
+
 func (s *queueBaseSuite) TestCheckPoint_SlicePredicateAction() {
 	exclusiveReaderHighWatermark := tasks.MaximumKey
 	scopes := NewRandomScopes(3)

service/history/queues/slice.go

@@ -348,6 +348,7 @@ func (s *SliceImpl) shrinkPredicate() {
 
 	// TODO: this should be generic enough to shrink any predicate type, probably doesn't belong here.
 	pendingPerKey := s.pendingPerKey
+	metrics.QueueSlicePendingKeys.With(s.metricsHandler).Record(int64(len(pendingPerKey)))
 	if len(pendingPerKey) > s.maxPendingKeysFn() {
 		// only shrink predicate if there're few keys left
 		metrics.QueuePredicateResolutionLoss.With(s.metricsHandler).Record(

service/history/queues/slice_test.go

@@ -18,6 +18,7 @@ import (
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/metrics/metricstest"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/predicates"
 	ctasks "go.temporal.io/server/common/tasks"
@@ -449,6 +450,80 @@ func (s *sliceSuite) TestShrinkScope_ShrinkPredicate() {
 	}
 }
 
+func (s *sliceSuite) TestShrinkScope_RecordsPendingKeysHistogram() {
+	testCases := []struct {
+		name                 string
+		numPendingNamespaces int
+		leaveIteratorsOpen   bool
+		expectDeclined       bool
+	}{
+		{
+			name:                 "narrowing succeeds",
+			numPendingNamespaces: 3,
+		},
+		{
+			name:                 "narrowing declines",
+			numPendingNamespaces: 12,
+			expectDeclined:       true,
+		},
+		{
+			name:                 "slice still reading its range: no sample recorded",
+			numPendingNamespaces: 3,
+			leaveIteratorsOpen:   true,
+		},
+	}
+
+	for _, tc := range testCases {
+		s.Run(tc.name, func() {
+			r := NewRandomRange()
+			predicate := predicates.Universal[tasks.Task]()
+
+			handler := metricstest.NewCaptureHandler()
+			capture := handler.StartCapture()
+			defer handler.StopCapture(capture)
+
+			slice := NewSlice(nil, s.executableFactory, s.monitor, NewScope(r, predicate), GrouperNamespaceID{}, noPredicateSizeLimit, defaultMaxPendingKeys, handler)
+			if !tc.leaveIteratorsOpen {
+				slice.iterators = []Iterator{} // manually set iterators to be empty to trigger predicate update
+			}
+
+			// One pending executable per namespace, each with its own distinct namespace ID,
+			// so the pending-key count is exact rather than a coincidence of random assignment
+			// across a shared, smaller pool of namespace IDs.
+			executables := s.randomExecutablesInRange(r, tc.numPendingNamespaces)
+			for _, executable := range executables {
+				mockExecutable := executable.(*MockExecutable)
+				mockExecutable.EXPECT().GetTask().Return(mockExecutable).AnyTimes()
+				mockExecutable.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
+				mockExecutable.EXPECT().State().Return(ctasks.TaskStatePending).MaxTimes(1)
+				slice.add(executable)
+			}
+
+			slice.ShrinkScope()
+			s.validateSliceState(slice)
+
+			snapshot := capture.Snapshot()
+
+			if tc.leaveIteratorsOpen {
+				s.Empty(snapshot[metrics.QueueSlicePendingKeys.Name()])
+				s.Empty(snapshot[metrics.QueuePredicateResolutionLoss.Name()])
+				return
+			}
+
+			pendingKeysRecordings := snapshot[metrics.QueueSlicePendingKeys.Name()]
+			s.Require().Len(pendingKeysRecordings, 1)
+			s.Equal(int64(tc.numPendingNamespaces), pendingKeysRecordings[0].Value)
+
+			lossRecordings := snapshot[metrics.QueuePredicateResolutionLoss.Name()]
+			if tc.expectDeclined {
+				s.Len(lossRecordings, 1)
+			} else {
+				s.Empty(lossRecordings)
+			}
+		})
+	}
+}
+
 func (s *sliceSuite) TestSelectTasks_NoError() {
 	r := NewRandomRange()
 	namespaceIDs := []string{uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString()}

service/history/shard/context_impl.go

@@ -1257,6 +1257,21 @@ func (s *ContextImpl) updateShardInfo(
 	s.tasksCompletedSinceLastUpdate = 0
 
 	updatedShardInfo := trimShardInfo(s.config, s.clusterMetadata.GetAllClusterInfo(), s.copyShardInfo(s.shardInfo))
+
+	metrics.ShardInfoSize.With(s.metricsHandler).Record(int64(updatedShardInfo.Size()))
+	for categoryID, queueState := range updatedShardInfo.QueueStates {
+		category, ok := s.taskCategoryRegistry.GetCategoryByID(int(categoryID))
+		if !ok {
+			continue
+		}
+		sizeBytes := int64(queueState.Size())
+		categoryTag := metrics.TaskCategoryTag(category.Name())
+		// The counter is a true accumulator; the histogram's _sum is not, since tally's Prometheus
+		// reporter replays each sample as its bucket's upper bound, not the recorded value.
+		metrics.QueueStateSize.With(s.metricsHandler).Record(sizeBytes, categoryTag)
+		metrics.QueueStateSizeTotal.With(s.metricsHandler).Record(sizeBytes, categoryTag)
+	}
+
 	request := &persistence.UpdateShardRequest{
 		ShardInfo:       updatedShardInfo,
 		PreviousRangeID: s.shardInfo.GetRangeId(),

service/history/shard/context_test.go

@@ -994,6 +994,102 @@ func (s *contextSuite) TestUpdateShardInfo_FirstUpdate() {
 	s.Equal(0, s.mockShard.tasksCompletedSinceLastUpdate)
 }
 
+func (s *contextSuite) TestUpdateShardInfo_RecordsSizeMetrics() {
+	s.mockShard.state = contextStateAcquired
+	s.setImmediateAckLevels(map[int32]int64{
+		int32(tasks.CategoryIDTransfer): 100,
+		int32(tasks.CategoryIDTimer):    200,
+	})
+
+	expectedTransferSize := int64(s.mockShard.shardInfo.QueueStates[int32(tasks.CategoryIDTransfer)].Size())
+	expectedTimerSize := int64(s.mockShard.shardInfo.QueueStates[int32(tasks.CategoryIDTimer)].Size())
+
+	s.mockShardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1)
+
+	captureHandler := metricstest.NewCaptureHandler()
+	s.mockShard.SetMetricsHandler(captureHandler)
+	capture := captureHandler.StartCapture()
+	defer captureHandler.StopCapture(capture)
+
+	err := s.mockShard.updateShardInfo(0, func() {})
+	s.NoError(err)
+
+	snapshot := capture.Snapshot()
+
+	shardInfoSizeRecordings := snapshot[metrics.ShardInfoSize.Name()]
+	s.Require().Len(shardInfoSizeRecordings, 1)
+	s.GreaterOrEqual(shardInfoSizeRecordings[0].Value.(int64), expectedTransferSize)
+	s.GreaterOrEqual(shardInfoSizeRecordings[0].Value.(int64), expectedTimerSize)
+
+	queueStateSizeRecordings := snapshot[metrics.QueueStateSize.Name()]
+	s.Require().Len(queueStateSizeRecordings, 2)
+
+	sizeByCategory := make(map[string]int64, len(queueStateSizeRecordings))
+	for _, recording := range queueStateSizeRecordings {
+		sizeByCategory[recording.Tags["task_category"]] = recording.Value.(int64)
+	}
+	s.Equal(map[string]int64{
+		tasks.CategoryTransfer.Name(): expectedTransferSize,
+		tasks.CategoryTimer.Name():    expectedTimerSize,
+	}, sizeByCategory)
+}
+
+func (s *contextSuite) TestUpdateShardInfo_RecordsQueueStateSizeTotal() {
+	s.mockShard.state = contextStateAcquired
+	s.setImmediateAckLevels(map[int32]int64{
+		int32(tasks.CategoryIDTransfer): 100,
+		int32(tasks.CategoryIDTimer):    200,
+	})
+
+	expectedTransferSize := int64(s.mockShard.shardInfo.QueueStates[int32(tasks.CategoryIDTransfer)].Size())
+	expectedTimerSize := int64(s.mockShard.shardInfo.QueueStates[int32(tasks.CategoryIDTimer)].Size())
+
+	s.mockShardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1)
+
+	captureHandler := metricstest.NewCaptureHandler()
+	s.mockShard.SetMetricsHandler(captureHandler)
+	capture := captureHandler.StartCapture()
+	defer captureHandler.StopCapture(capture)
+
+	err := s.mockShard.updateShardInfo(0, func() {})
+	s.NoError(err)
+
+	snapshot := capture.Snapshot()
+	recordings := snapshot[metrics.QueueStateSizeTotal.Name()]
+	s.Require().Len(recordings, 2)
+
+	sizeByCategory := make(map[string]int64, len(recordings))
+	for _, recording := range recordings {
+		sizeByCategory[recording.Tags["task_category"]] = recording.Value.(int64)
+	}
+	s.Equal(map[string]int64{
+		tasks.CategoryTransfer.Name(): expectedTransferSize,
+		tasks.CategoryTimer.Name():    expectedTimerSize,
+	}, sizeByCategory)
+}
+
+func (s *contextSuite) TestUpdateShardInfo_DoesNotRecordSizeMetrics_WhenThrottled() {
+	s.mockShard.state = contextStateAcquired
+
+	// First call always persists, establishing lastUpdated.
+	s.mockShardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil).Times(1)
+	s.NoError(s.mockShard.updateShardInfo(0, func() {}))
+
+	captureHandler := metricstest.NewCaptureHandler()
+	s.mockShard.SetMetricsHandler(captureHandler)
+	capture := captureHandler.StartCapture()
+	defer captureHandler.StopCapture(capture)
+
+	// No time has passed and too few tasks completed: shouldn't persist, and shouldn't record size.
+	s.mockShardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Times(0)
+	s.NoError(s.mockShard.updateShardInfo(0, func() {}))
+
+	snapshot := capture.Snapshot()
+	s.Empty(snapshot[metrics.ShardInfoSize.Name()])
+	s.Empty(snapshot[metrics.QueueStateSize.Name()])
+	s.Empty(snapshot[metrics.QueueStateSizeTotal.Name()])
+}
+
 // setImmediateAckLevels replaces the shard's queue states so each given immediate category has its
 // ack level at the given task id, i.e. a backlog of everything above it.
 func (s *contextSuite) setImmediateAckLevels(ackLevelByCategoryID map[int32]int64) {