Fix: Schedule scanner false-positives for inactive NS (#11703)

## What

This fixes the Schedule invariant scanner's false-positives coming from
replication. Due to carelessness it was firing on the passive side
because I forgot to filter this out, and for a while during
post-replication disconnection, the task processing will cease. Also
adds a small check for Described schedules to filter out visibility
drift.

## How
- Adds a guard for only checking active NS
- Adds a describe check for the next fire time, so that
visibility-delayed schedules are excluded

## Risks:

- That I make a mistake and break the scanner

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

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(

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

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) {