diff --git a/common/metrics/metric_defs.go b/common/metrics/metric_defs.go index f1b47c5eed..ab7d3b1b41 100644 --- a/common/metrics/metric_defs.go +++ b/common/metrics/metric_defs.go @@ -1437,22 +1437,23 @@ var ( HistoryWorkflowExecutionCacheLatency = NewTimerDef("history_workflow_execution_cache_latency") HistoryWorkflowExecutionCacheLockHoldDuration = NewTimerDef("history_workflow_execution_cache_lock_hold_duration") - VisibilityArchiverArchiveNonRetryableErrorCount = NewCounterDef("visibility_archiver_archive_non_retryable_error") - VisibilityArchiverArchiveTransientErrorCount = NewCounterDef("visibility_archiver_archive_transient_error") - VisibilityArchiveSuccessCount = NewCounterDef("visibility_archiver_archive_success") - HistoryScavengerSuccessCount = NewCounterDef("scavenger_success") - HistoryScavengerErrorCount = NewCounterDef("scavenger_errors") - HistoryScavengerSkipCount = NewCounterDef("scavenger_skips") - ScheduleInvariantsScannerOverdueNextActionTimeCount = NewCounterDef("schedule_invariants_scanner_overdue_next_action_time") - ScheduleInvariantsScannerOverdueNextActionTimeCapHitCount = NewCounterDef("schedule_invariants_scanner_overdue_next_action_time_cap_hit") - ScheduleInvariantsScannerStuckOpenCount = NewCounterDef("schedule_invariants_scanner_stuck_open") - ScheduleInvariantsScannerUnknownStateCount = NewCounterDef("schedule_invariants_scanner_unknown_state") - ScheduleInvariantsScannerErrorCount = NewCounterDef("schedule_invariants_scanner_errors") - ExecutionsOutstandingCount = NewGaugeDef("executions_outstanding") - ScavengerValidationRequestsCount = NewCounterDef("scavenger_validation_requests") - ScavengerValidationFailuresCount = NewCounterDef("scavenger_validation_failures") - ScavengerValidationSkipsCount = NewCounterDef("scavenger_validation_skips") - AddSearchAttributesFailuresCount = NewCounterDef("add_search_attributes_failures") + VisibilityArchiverArchiveNonRetryableErrorCount = NewCounterDef("visibility_archiver_archive_non_retryable_error") + VisibilityArchiverArchiveTransientErrorCount = NewCounterDef("visibility_archiver_archive_transient_error") + VisibilityArchiveSuccessCount = NewCounterDef("visibility_archiver_archive_success") + HistoryScavengerSuccessCount = NewCounterDef("scavenger_success") + HistoryScavengerErrorCount = NewCounterDef("scavenger_errors") + HistoryScavengerSkipCount = NewCounterDef("scavenger_skips") + ScheduleInvariantsScannerOverdueNextActionTimeCount = NewCounterDef("schedule_invariants_scanner_overdue_next_action_time") + ScheduleInvariantsScannerOverdueNextActionTimeCapHitCount = NewCounterDef("schedule_invariants_scanner_overdue_next_action_time_cap_hit") + ScheduleInvariantsScannerOverdueNextActionTimeStaleCandidateCount = NewCounterDef("schedule_invariants_scanner_overdue_next_action_time_stale_candidate") + ScheduleInvariantsScannerStuckOpenCount = NewCounterDef("schedule_invariants_scanner_stuck_open") + ScheduleInvariantsScannerUnknownStateCount = NewCounterDef("schedule_invariants_scanner_unknown_state") + ScheduleInvariantsScannerErrorCount = NewCounterDef("schedule_invariants_scanner_errors") + ExecutionsOutstandingCount = NewGaugeDef("executions_outstanding") + ScavengerValidationRequestsCount = NewCounterDef("scavenger_validation_requests") + ScavengerValidationFailuresCount = NewCounterDef("scavenger_validation_failures") + ScavengerValidationSkipsCount = NewCounterDef("scavenger_validation_skips") + AddSearchAttributesFailuresCount = NewCounterDef("add_search_attributes_failures") // Delete Namespace metrics. ReclaimResourcesNamespaceDeleteSuccessCount = NewCounterDef( diff --git a/service/worker/scanner/scheduleinvariants/invariants.go b/service/worker/scanner/scheduleinvariants/invariants.go index efbc8d9d2c..7dae97dd5f 100644 --- a/service/worker/scanner/scheduleinvariants/invariants.go +++ b/service/worker/scanner/scheduleinvariants/invariants.go @@ -22,6 +22,7 @@ import ( "go.temporal.io/server/common/persistence/visibility/manager" "go.temporal.io/server/common/quotas" "go.temporal.io/server/common/sdk" + "google.golang.org/protobuf/types/known/timestamppb" ) const ( @@ -251,6 +252,10 @@ func (a *Activities) ListAllNamespaces() []string { if ns.State() == enumspb.NAMESPACE_STATE_DELETED { continue } + //nolint:forbidigo // scanner works per namespace, not per workflow. + if !ns.ActiveInCluster(a.currentClusterName) { + continue + } names = append(names, ns.Name().String()) } return names @@ -371,9 +376,39 @@ func (a *Activities) scheduleIsExpectedNotToFire(ctx context.Context, nsName, sc len(desc.GetInfo().GetRunningWorkflows()) > 0 { return true } + + // Confirm the invariant against what Describe just returned: the candidate came from + // a ScheduleNextActionTime index entry, which goes stale on replication or indexing + // lag. + if !a.nextActionTimeIsOverdue(desc.GetInfo().GetFutureActionTimes()) { + a.logger.Info("overdue candidate was not overdue on re-check; visibility entry was stale", + tag.WorkflowNamespace(nsName), tag.ScheduleID(scheduleID)) + metrics.ScheduleInvariantsScannerOverdueNextActionTimeStaleCandidateCount.With( + a.metricsHandler.WithTags(metrics.NamespaceTag(nsName))).Record(1) + return true + } return false } +// nextActionTimeIsOverdue applies the candidate query's predicate to futureActionTimes. +func (a *Activities) nextActionTimeIsOverdue(futureActionTimes []*timestamppb.Timestamp) bool { + var earliest time.Time + var found bool + for _, t := range futureActionTimes { + if t == nil { + continue + } + // Ordered soonest-first, but don't rely on it. + if ts := t.AsTime(); !found || ts.Before(earliest) { + earliest, found = ts, true + } + } + if !found { + return false + } + return earliest.Before(a.timeSource.Now().UTC().Add(-a.opts().OverdueNextActionTimeTolerance)) +} + func (a *Activities) emitCount(metricName, namespaceTagValue string, count int64) { if count <= 0 { return diff --git a/service/worker/scanner/scheduleinvariants/invariants_test.go b/service/worker/scanner/scheduleinvariants/invariants_test.go index 630174121e..f1d60ecc8a 100644 --- a/service/worker/scanner/scheduleinvariants/invariants_test.go +++ b/service/worker/scanner/scheduleinvariants/invariants_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "testing" + "time" "github.com/stretchr/testify/require" commonpb "go.temporal.io/api/common/v1" @@ -18,6 +19,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/persistence/visibility/manager" "go.temporal.io/server/common/quotas" @@ -25,10 +27,14 @@ import ( "go.temporal.io/server/common/testing/mockapi/workflowservicemock/v1" "go.temporal.io/server/common/testing/mocksdk" "go.uber.org/mock/gomock" + "google.golang.org/protobuf/types/known/timestamppb" ) const testClusterName = "test-cluster" +// testNow anchors the injected clock; the overdue re-check reads timeSource.Now(). +var testNow = time.Date(2026, 8, 21, 12, 0, 0, 0, time.UTC) + type testDeps struct { ctrl *gomock.Controller visibilityManager *manager.MockVisibilityManager @@ -37,6 +43,7 @@ type testDeps struct { sdkClient *mocksdk.MockClient frontendClient *workflowservicemock.MockWorkflowServiceClient timeSource *clock.EventTimeSource + metricsHandler metrics.Handler } func newTestDeps(t *testing.T) *testDeps { @@ -50,7 +57,9 @@ func newTestDeps(t *testing.T) *testDeps { sdkClient: mocksdk.NewMockClient(ctrl), frontendClient: workflowservicemock.NewMockWorkflowServiceClient(ctrl), timeSource: clock.NewEventTimeSource(), + metricsHandler: metrics.NoopMetricsHandler, } + d.timeSource.Update(testNow) // The DescribeSchedule path always goes via system client → frontend stub. d.sdkClientFactory.EXPECT().GetSystemClient().Return(d.sdkClient).AnyTimes() d.sdkClient.EXPECT().WorkflowService().Return(d.frontendClient).AnyTimes() @@ -66,7 +75,7 @@ func (d *testDeps) newActivitiesWithParams(params dynamicconfig.ScheduleInvarian rl := quotas.NewDefaultOutgoingRateLimiter(quotas.RateFn(dynamicconfig.GetFloatPropertyFn(10000.0))) return &Activities{ logger: log.NewNoopLogger(), - metricsHandler: metrics.NoopMetricsHandler, + metricsHandler: d.metricsHandler, visibilityManager: d.visibilityManager, namespaceRegistry: d.namespaceRegistry, sdkClientFactory: d.sdkClientFactory, @@ -120,7 +129,9 @@ func TestListAllNamespaces_FiltersInactiveAndDeleted(t *testing.T) { }) names := d.newActivities().ListAllNamespaces() - require.ElementsMatch(t, []string{"ns-1", "ns-2", "ns-3"}, names) + require.ElementsMatch(t, []string{"ns-1", "ns-3"}, names, + "ns-2 is active in another cluster: evaluating its invariants here would read a "+ + "standby replica's stale visibility records") } func TestForEachNamespace_InvokesCallbackWithCount(t *testing.T) { @@ -220,7 +231,25 @@ func TestSchedulesInNamespace_YieldsErrorAndStops(t *testing.T) { require.Error(t, iterErr) } -func describeResp(paused bool, overlap enumspb.ScheduleOverlapPolicy, runningCount int) *workflowservice.DescribeScheduleResponse { +var overdueTolerance = dynamicconfig.DefaultScheduleInvariantsScannerParams.OverdueNextActionTimeTolerance + +// overdueActionTime confirms the invariant; pendingActionTime clears it on re-check. +func overdueActionTime() time.Time { + return testNow.Add(-overdueTolerance).Add(-time.Hour) +} + +func pendingActionTime() time.Time { + return testNow.Add(time.Hour) +} + +// describeResp builds a DescribeSchedule response. Passing no futureActionTimes models +// a schedule with no upcoming action. +func describeResp( + paused bool, + overlap enumspb.ScheduleOverlapPolicy, + runningCount int, + futureActionTimes ...time.Time, +) *workflowservice.DescribeScheduleResponse { resp := &workflowservice.DescribeScheduleResponse{ Schedule: &schedulepb.Schedule{ State: &schedulepb.ScheduleState{Paused: paused}, @@ -231,6 +260,9 @@ func describeResp(paused bool, overlap enumspb.ScheduleOverlapPolicy, runningCou for range runningCount { resp.Info.RunningWorkflows = append(resp.Info.RunningWorkflows, &commonpb.WorkflowExecution{WorkflowId: "running"}) } + for _, t := range futureActionTimes { + resp.Info.FutureActionTimes = append(resp.Info.FutureActionTimes, timestamppb.New(t)) + } return resp } @@ -258,17 +290,17 @@ func TestScheduleIsExpectedNotToFire(t *testing.T) { }, { name: "buffer_one_no_running_workflow", - resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, 0), + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, 0, overdueActionTime()), want: false, }, { - name: "skip_policy_with_running_workflow", - resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 1), + name: "skip_policy_with_running_workflow_still_overdue", + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 1, overdueActionTime()), want: false, }, { name: "cancel_other_policy", - resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, 1), + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, 1, overdueActionTime()), want: false, }, { @@ -276,6 +308,69 @@ func TestScheduleIsExpectedNotToFire(t *testing.T) { err: errors.New("describe failed"), want: false, }, + { + // Stale index entry: a standby's frozen record, or a SKIP schedule whose + // action overran while the Generator kept ticking. + name: "next_action_time_still_pending", + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 1, pendingActionTime()), + want: true, + }, + { + // Nothing pending can be late. + name: "no_future_action_times", + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0), + want: true, + }, + { + name: "next_action_time_exactly_at_threshold", + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0, testNow.Add(-overdueTolerance)), + want: true, + }, + { + name: "next_action_time_just_past_threshold", + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0, + testNow.Add(-overdueTolerance).Add(-time.Nanosecond)), + want: false, + }, + { + // The stalled-generator shape: visibility indexes FutureActionTimes[0], + // which is overdue, while the rest of the cached horizon is still future. + // Requiring every entry to be overdue would delay detection by the full + // cache depth. + name: "only_earliest_entry_overdue", + resp: func() *workflowservice.DescribeScheduleResponse { + times := []time.Time{overdueActionTime()} + for i := range 9 { + times = append(times, testNow.Add(time.Duration(i+1)*time.Hour)) + } + return describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0, times...) + }(), + want: false, + }, + { + // Ordering isn't guaranteed: the earliest entry decides, wherever it sits. + name: "unordered_earliest_is_overdue", + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0, + pendingActionTime(), overdueActionTime()), + want: false, + }, + { + // Stale index entry: every cached time is still in the future. + name: "all_entries_pending", + resp: describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0, + pendingActionTime(), pendingActionTime().Add(time.Hour)), + want: true, + }, + { + // A nil entry must not read as the zero time, which would look overdue. + name: "nil_entry_among_pending_times", + resp: func() *workflowservice.DescribeScheduleResponse { + r := describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0, pendingActionTime()) + r.Info.FutureActionTimes = append([]*timestamppb.Timestamp{nil}, r.Info.FutureActionTimes...) + return r + }(), + want: true, + }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -314,10 +409,82 @@ func TestRunOverdueScan_FiltersExpectedNotToFireSchedulesAndCountsRest(t *testin }).Return(describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, 1), nil) d.frontendClient.EXPECT().DescribeSchedule(gomock.Any(), &workflowservice.DescribeScheduleRequest{ Namespace: "ns-1", ScheduleId: "sched-actually-overdue", - }).Return(describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0), nil) + }).Return(describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 0, overdueActionTime()), nil) + + rec := metricstest.NewCaptureHandler() + d.metricsHandler = rec + capture := rec.StartCapture() + defer rec.StopCapture(capture) err := d.newActivities().runOverdueScan(context.Background(), "q") require.NoError(t, err) + + snapshot := capture.Snapshot() + anomalies := snapshot[metrics.ScheduleInvariantsScannerOverdueNextActionTimeCount.Name()] + require.Len(t, anomalies, 1) + require.Equal(t, int64(1), anomalies[0].Value, "only sched-actually-overdue should count") + require.Empty(t, snapshot[metrics.ScheduleInvariantsScannerOverdueNextActionTimeStaleCandidateCount.Name()], + "paused and buffer-waiting are exemptions, not stale candidates") +} + +// Asserts by absence: with no expectations registered, any call for ns-passive fails. +func TestRunOverdueScan_SkipsNamespaceActiveInAnotherCluster(t *testing.T) { + d := newTestDeps(t) + + d.namespaceRegistry.EXPECT().GetAllNamespaces().Return([]*namespace.Namespace{ + globalNS("id-passive", "ns-passive", "other-cluster"), + }) + + err := d.newActivities().runOverdueScan(context.Background(), "q") + require.NoError(t, err) +} + +// Same gate for the count-only scanners, which have no confirmation step at all. +func TestRunScan_SkipsNamespaceActiveInAnotherCluster(t *testing.T) { + d := newTestDeps(t) + + d.namespaceRegistry.EXPECT().GetAllNamespaces().Return([]*namespace.Namespace{ + globalNS("id-passive", "ns-passive", "other-cluster"), + localNS("id-local", "ns-local", testClusterName), + }) + // Only the local namespace is queried. + d.namespaceRegistry.EXPECT().GetNamespaceID(namespace.Name("ns-local")).Return(namespace.ID("id-local"), nil) + d.visibilityManager.EXPECT().CountChasmExecutions(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, req *visibilityservice.CountChasmExecutionsRequest) (*visibilityservice.CountChasmExecutionsResponse, error) { + require.Equal(t, "ns-local", req.Namespace) + return &visibilityservice.CountChasmExecutionsResponse{Count: 3}, nil + }) + + err := d.newActivities().runScan(context.Background(), "stuck_open", "q", "some_metric") + require.NoError(t, err) +} + +// A stale candidate is not an anomaly, but must still be counted so the suppression +// is observable. +func TestRunOverdueScan_StaleCandidateIsCountedSeparatelyNotAsAnomaly(t *testing.T) { + d := newTestDeps(t) + + d.namespaceRegistry.EXPECT().GetAllNamespaces().Return([]*namespace.Namespace{localNS("id-1", "ns-1", testClusterName)}) + d.namespaceRegistry.EXPECT().GetNamespaceID(namespace.Name("ns-1")).Return(namespace.ID("id-1"), nil) + d.visibilityManager.EXPECT().ListChasmExecutions(gomock.Any(), gomock.Any()).Return(&visibilityservice.ListChasmExecutionsResponse{ + Executions: []*chasmspb.VisibilityExecutionInfo{chasmExec("sched-stale")}, + }, nil) + d.frontendClient.EXPECT().DescribeSchedule(gomock.Any(), gomock.Any()). + Return(describeResp(false, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, 1, pendingActionTime()), nil) + + rec := metricstest.NewCaptureHandler() + d.metricsHandler = rec + capture := rec.StartCapture() + defer rec.StopCapture(capture) + + require.NoError(t, d.newActivities().runOverdueScan(context.Background(), "q")) + + snapshot := capture.Snapshot() + require.Empty(t, snapshot[metrics.ScheduleInvariantsScannerOverdueNextActionTimeCount.Name()], + "a stale visibility entry is not an anomaly") + stale := snapshot[metrics.ScheduleInvariantsScannerOverdueNextActionTimeStaleCandidateCount.Name()] + require.Len(t, stale, 1) + require.Equal(t, "ns-1", stale[0].Tags["namespace"]) } func TestRunOverdueScan_ContinuesPastPerNamespaceErrors(t *testing.T) {