From bdd12146ae87dff7f01fc9167f414c6cb7622f73 Mon Sep 17 00:00:00 2001 From: long-nt-tran Date: Wed, 1 Jul 2026 18:29:38 -0400 Subject: [PATCH 1/5] [testcore] Shared cluster pool for clusters that require WorkerService --- tests/activity_api_batch_reset_test.go | 2 +- tests/activity_api_batch_unpause_test.go | 9 +- .../activity_api_batch_update_options_test.go | 4 +- tests/advanced_visibility_test.go | 2 +- tests/chasm_test.go | 2 +- tests/client_misc_test.go | 6 +- tests/dlq_test.go | 6 +- tests/leakcheck/leak_test.go | 3 +- tests/namespace_test.go | 4 +- tests/poller_scaling_test.go | 5 +- tests/purge_dlq_tasks_api_test.go | 6 +- tests/schedule_migration_test.go | 23 +++-- tests/schedule_test.go | 4 +- tests/task_queue_stats_test.go | 6 +- tests/testcore/onebox.go | 4 + tests/testcore/test_cluster_pool.go | 83 ++++++++++++------- tests/testcore/test_cluster_pool_test.go | 77 ++++++++++++++++- tests/testcore/test_env.go | 25 ++++-- tests/versioning_3_test.go | 2 +- tests/worker_deployment_version_test.go | 6 +- tests/worker_registry_test.go | 4 +- tests/workflow_alias_search_attribute_test.go | 5 +- tests/workflow_reset_test.go | 4 +- 23 files changed, 209 insertions(+), 83 deletions(-) diff --git a/tests/activity_api_batch_reset_test.go b/tests/activity_api_batch_reset_test.go index 332e945bf5a..0c2a58b2e8f 100644 --- a/tests/activity_api_batch_reset_test.go +++ b/tests/activity_api_batch_reset_test.go @@ -32,7 +32,7 @@ func TestActivityAPIBatchResetClientTestSuite(t *testing.T) { func newBatchResetEnv(t *testing.T) *testcore.TestEnv { return testcore.NewEnv( t, - testcore.WithWorkerService("batch operations"), + testcore.WithWorkerService(), // These tests intentionally start multiple batch operations in the same namespace. // The default per-namespace limit is 1, so raise it to the functional test limit. testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, testcore.ClientSuiteLimit), diff --git a/tests/activity_api_batch_unpause_test.go b/tests/activity_api_batch_unpause_test.go index 0b2f8c0b2a9..fa4322a0177 100644 --- a/tests/activity_api_batch_unpause_test.go +++ b/tests/activity_api_batch_unpause_test.go @@ -31,8 +31,7 @@ type ActivityApiBatchUnpauseClientTestSuite struct { } func TestActivityApiBatchUnpauseClientTestSuite(t *testing.T) { - testcore.UseSuiteScopedCluster(t) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. - parallelsuite.RunLegacySequential(t, &ActivityApiBatchUnpauseClientTestSuite{}) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. + parallelsuite.Run(t, &ActivityApiBatchUnpauseClientTestSuite{}) } type internalTestWorkflow struct { @@ -93,7 +92,7 @@ func (s *ActivityApiBatchUnpauseClientTestSuite) createWorkflow(env *testcore.Te } func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Success() { - env := testcore.NewEnv(s.T()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() @@ -200,7 +199,7 @@ func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Succes } func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Failed() { - env := testcore.NewEnv(s.T()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) // neither activity type not "match all" is provided _, err := env.SdkClient().WorkflowService().StartBatchOperation(context.Background(), &workflowservice.StartBatchOperationRequest{ @@ -238,7 +237,7 @@ func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Failed // This is an end-to-end complement to the unit-level checkNamespace tests: it // exercises the full path from StartBatchOperation through the batcher worker. func (s *ActivityApiBatchUnpauseClientTestSuite) TestBatchTerminate_NamespaceIsolation() { - env := testcore.NewEnv(s.T()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() diff --git a/tests/activity_api_batch_update_options_test.go b/tests/activity_api_batch_update_options_test.go index 5c34ca96de6..3edc4d2d0a1 100644 --- a/tests/activity_api_batch_update_options_test.go +++ b/tests/activity_api_batch_update_options_test.go @@ -46,7 +46,7 @@ func (s *ActivityAPIBatchUpdateOptionsSuite) createBatchUpdateOptionsWorkflow(en func (s *ActivityAPIBatchUpdateOptionsSuite) TestActivityBatchUpdateOptionsSuccess() { env := testcore.NewEnv(s.T(), - testcore.WithWorkerService("batch operations"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, testcore.ClientSuiteLimit), ) @@ -194,7 +194,7 @@ func (s *ActivityAPIBatchUpdateOptionsSuite) TestActivityBatchUpdateOptionsSucce func (s *ActivityAPIBatchUpdateOptionsSuite) TestActivityBatchUpdateOptionsFailed() { env := testcore.NewEnv(s.T(), - testcore.WithWorkerService("batch operations"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, testcore.ClientSuiteLimit), ) diff --git a/tests/advanced_visibility_test.go b/tests/advanced_visibility_test.go index 49077aa2aed..9b4f90f56b1 100644 --- a/tests/advanced_visibility_test.go +++ b/tests/advanced_visibility_test.go @@ -2411,7 +2411,7 @@ func (s *AdvancedVisibilitySuite) TestWorkerTaskReachability_Unversioned_InTaskQ } func (s *AdvancedVisibilitySuite) TestBuildIdScavenger_DeletesUnusedBuildId(enableUnifiedQueryConverter bool) { - env := s.newTestEnv(enableUnifiedQueryConverter, testcore.WithWorkerService("build id scavenger workflow")) + env := s.newTestEnv(enableUnifiedQueryConverter, testcore.WithWorkerService()) tq := s.T().Name() buildIdv0 := s.T().Name() + "-v0" buildIdv1 := s.T().Name() + "-v1" diff --git a/tests/chasm_test.go b/tests/chasm_test.go index 970b1d3b040..d3e8af5219b 100644 --- a/tests/chasm_test.go +++ b/tests/chasm_test.go @@ -63,7 +63,7 @@ func newChasmTestEnv(t *testing.T, unified bool) chasmTestEnv { env := testcore.NewEnv( t, testcore.WithDedicatedCluster(), - testcore.WithWorkerService("delete namespace workflow"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.VisibilityEnableUnifiedQueryConverter, unified), testcore.WithDynamicConfig(dynamicconfig.DeleteNamespaceUseChasmDeleteExecution, true), diff --git a/tests/client_misc_test.go b/tests/client_misc_test.go index 44cbd305797..caf46b97230 100644 --- a/tests/client_misc_test.go +++ b/tests/client_misc_test.go @@ -1063,7 +1063,7 @@ func (s *ClientMiscTestSuite) Test_StickyWorkerRestartWorkflowTask() { } func (s *ClientMiscTestSuite) TestBatchSignal() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch operations")) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) type myData struct { Stuff string @@ -1118,7 +1118,7 @@ func (s *ClientMiscTestSuite) TestBatchSignal() { } func (s *ClientMiscTestSuite) TestBatchReset() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch operations")) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) var count atomic.Int32 activityFn := func(ctx context.Context) (int32, error) { @@ -1182,7 +1182,7 @@ func (s *ClientMiscTestSuite) TestBatchReset() { } func (s *ClientMiscTestSuite) TestBatchResetByBuildId() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch operations")) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) tq := testcore.RandomizeStr(s.T().Name()) buildPrefix := uuid.NewString()[:6] + "-" buildIdv1 := buildPrefix + "v1" diff --git a/tests/dlq_test.go b/tests/dlq_test.go index ee2c43178c4..ff78342df1d 100644 --- a/tests/dlq_test.go +++ b/tests/dlq_test.go @@ -252,7 +252,7 @@ func (s *DLQSuite) TestReadArtificialDLQTasks() { // DLQ, this test then purges the DLQ and verifies that the task was deleted. // This test will then call DescribeDLQJob and CancelDLQJob api to verify. func (s *DLQSuite) TestPurgeRealWorkflow() { - env := s.newTestEnv(testcore.WithWorkerService("dlq purge workflow")) + env := s.newTestEnv(testcore.WithWorkerService()) _, dlqMessageID := s.executeDoomedWorkflow(env) @@ -281,7 +281,7 @@ func (s *DLQSuite) TestPurgeRealWorkflow() { // above test is more for testing specific CLI flags when reading from the DLQ. // This test will then call DescribeDLQJob and CancelDLQJob api to verify. func (s *DLQSuite) TestMergeRealWorkflow() { - env := s.newTestEnv(testcore.WithWorkerService("dlq merge workflow")) + env := s.newTestEnv(testcore.WithWorkerService()) // Verify that we can execute a normal workflow. run := s.executeWorkflow(env, "dlq-test-ok-workflow-id") @@ -324,7 +324,7 @@ func (s *DLQSuite) TestMergeRealWorkflow() { } func (s *DLQSuite) TestCancelRunningMerge() { - env := s.newTestEnv(testcore.WithWorkerService("dlq merge workflow")) + env := s.newTestEnv(testcore.WithWorkerService()) env.deleteBlockCh = make(chan any) // Execute several doomed workflows. diff --git a/tests/leakcheck/leak_test.go b/tests/leakcheck/leak_test.go index 169d0cec2f1..7ed2583ddb0 100644 --- a/tests/leakcheck/leak_test.go +++ b/tests/leakcheck/leak_test.go @@ -150,7 +150,8 @@ func buildRunTeardownCluster(t *testing.T, leakCheck *objectleak.ObjectLeakCheck t.Run("cluster", func(t *testing.T) { env := testcore.NewEnv(t, testcore.WithDedicatedCluster(), - testcore.WithWorkerService("leak regression test")) + testcore.WithWorkerService(), + testcore.ForLeakChecker()) env.SdkWorker().RegisterWorkflow(smokeWorkflow) run, err := env.SdkClient().ExecuteWorkflow( diff --git a/tests/namespace_test.go b/tests/namespace_test.go index 3ce173e87cf..d581de7a27c 100644 --- a/tests/namespace_test.go +++ b/tests/namespace_test.go @@ -31,12 +31,12 @@ type ( ) func TestNamespaceSuite(t *testing.T) { - parallelsuite.RunLegacySequential(t, &namespaceTestSuite{}) //nolint:staticcheck // SA1019: namespace deletion tests use dedicated worker-service clusters. + parallelsuite.Run(t, &namespaceTestSuite{}) } func (s *namespaceTestSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv { baseOpts := []testcore.TestOption{ - testcore.WithWorkerService("namespace deletion tests require the system worker service"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.TransferProcessorUpdateAckInterval, 1*time.Second), testcore.WithDynamicConfig(dynamicconfig.VisibilityProcessorUpdateAckInterval, 1*time.Second), } diff --git a/tests/poller_scaling_test.go b/tests/poller_scaling_test.go index bde1c0dc9c1..10c0da48cf7 100644 --- a/tests/poller_scaling_test.go +++ b/tests/poller_scaling_test.go @@ -31,12 +31,13 @@ type PollerScalingIntegSuite struct { } func TestPollerScalingFunctionalSuite(t *testing.T) { - testcore.UseSuiteScopedCluster(t) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. - parallelsuite.RunLegacySequential(t, &PollerScalingIntegSuite{}) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. + parallelsuite.Run(t, &PollerScalingIntegSuite{}) } func (s *PollerScalingIntegSuite) setupEnv(opts ...testcore.TestOption) *testcore.TestEnv { opts = append([]testcore.TestOption{ + testcore.WithWorkerService(), + // Force one partition so we can reliably see the backlog testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1), testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1), diff --git a/tests/purge_dlq_tasks_api_test.go b/tests/purge_dlq_tasks_api_test.go index 776b6e04ed7..33fa2212ea1 100644 --- a/tests/purge_dlq_tasks_api_test.go +++ b/tests/purge_dlq_tasks_api_test.go @@ -40,7 +40,7 @@ func (s *PurgeDLQTasksSuite) TestPurgeDLQTasks() { }{ { name: "HappyPath", - envOptions: []testcore.TestOption{testcore.WithWorkerService("purge DLQ workflow")}, + envOptions: []testcore.TestOption{testcore.WithWorkerService()}, }, { name: "MissingSourceCluster", @@ -58,7 +58,7 @@ func (s *PurgeDLQTasksSuite) TestPurgeDLQTasks() { }, { name: "QueueDoesNotExist", - envOptions: []testcore.TestOption{testcore.WithWorkerService("purge DLQ workflow")}, + envOptions: []testcore.TestOption{testcore.WithWorkerService()}, mutateRequest: func(request *adminservice.PurgeDLQTasksRequest) { request.DlqKey.TargetCluster = "does-not-exist" }, @@ -67,7 +67,7 @@ func (s *PurgeDLQTasksSuite) TestPurgeDLQTasks() { { name: "DeleteTasksUnavailableError", envOptions: []testcore.TestOption{ - testcore.WithWorkerService("purge DLQ workflow"), + testcore.WithWorkerService(), testcore.WithPersistenceFaultInjection(&config.FaultInjection{ Injector: func(target config.FaultInjectionTarget) error { if target.Store == config.QueueV2Name && diff --git a/tests/schedule_migration_test.go b/tests/schedule_migration_test.go index f0efc1b4985..7bac8af086d 100644 --- a/tests/schedule_migration_test.go +++ b/tests/schedule_migration_test.go @@ -43,13 +43,13 @@ type ScheduleMigrationTestSuite struct { } func TestScheduleMigrationTestSuite(t *testing.T) { - testcore.UseSuiteScopedCluster(t) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. - parallelsuite.RunLegacySequential(t, &ScheduleMigrationTestSuite{}) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. + parallelsuite.Run(t, &ScheduleMigrationTestSuite{}) } func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2AlreadyExists() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -291,6 +291,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1BlockedBySentine func (s *ScheduleMigrationTestSuite) TestScheduleMigrationDynamicConfig() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigration, true), ) @@ -400,6 +401,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationDynamicConfig() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV1ToV2() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -536,6 +538,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV1ToV2() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, false), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, false), @@ -742,6 +745,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1Idempotent() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, false), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, false), @@ -812,6 +816,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1Idempotent() { func (s *ScheduleMigrationTestSuite) TestCHASMScheduleDescribeAfterDisablingCreationAndMigration() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigration, true), @@ -913,6 +918,7 @@ func (s *ScheduleMigrationTestSuite) TestCHASMScheduleDescribeAfterDisablingCrea func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1RoutingFallback() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, true), @@ -1048,6 +1054,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1RoutingFallback( func (s *ScheduleMigrationTestSuite) TestScheduleUpdateAfterDelete() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, true), @@ -1158,6 +1165,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleUpdateAfterDelete() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV1ToV2WithClosedV2() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -1314,7 +1322,7 @@ func TestScheduleMigrationV1ToV2NoDuplicateRecentActions(t *testing.T) { // a CHASM sentinel (which would block the migration activity). env := testcore.NewEnv( t, - testcore.WithWorkerService("V1 scheduler"), + testcore.WithWorkerService(), testcore.WithSdkWorker(), ) @@ -1462,7 +1470,7 @@ func TestScheduleMigrationDeferredWithRunningWorkflow(t *testing.T) { // (workflow-backed) schedule rather than a CHASM sentinel. env := testcore.NewEnv( t, - testcore.WithWorkerService("V1 scheduler"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigrationWithRunningWorkflows, false), ) @@ -1598,6 +1606,7 @@ func TestScheduleMigrationDeferredWithRunningWorkflow(t *testing.T) { func (s *ScheduleMigrationTestSuite) TestDeleteScheduleContextMetadata() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerSentinels, true), @@ -1801,6 +1810,7 @@ func (s *ScheduleMigrationTestSuite) TestDeleteScheduleContextMetadata() { func (s *ScheduleMigrationTestSuite) TestPatchScheduleContextMetadata() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerSentinels, true), @@ -1963,7 +1973,7 @@ func (s *ScheduleMigrationTestSuite) TestPatchScheduleContextMetadata() { func TestScheduleMigration_StaleRunningDoesNotSkipPending(t *testing.T) { env := testcore.NewEnv( t, - testcore.WithWorkerService("scheduler operations"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -2111,6 +2121,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationRolloutPercent() { t := s.T() env := testcore.NewEnv( t, + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigration, true), testcore.WithDynamicConfig(dynamicconfig.CHASMSchedulerMigrationRolloutPercent, 50), @@ -2236,7 +2247,7 @@ func TestScheduleMigration_NoRunningWorkflows_GeneratorStarts(t *testing.T) { env := testcore.NewEnv( t, - testcore.WithWorkerService("scheduler operations"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) env.OverrideDynamicConfig(chasmscheduler.CurrentTweakables, tweakables) diff --git a/tests/schedule_test.go b/tests/schedule_test.go index c62c65e0806..89043bea544 100644 --- a/tests/schedule_test.go +++ b/tests/schedule_test.go @@ -69,7 +69,7 @@ func scheduleCommonOpts(t *testing.T) []testcore.TestOption { } if strings.HasPrefix(t.Name(), "TestScheduleV1") { // only v1 needs the worker service - opts = append(opts, testcore.WithWorkerService("V1 scheduler")) + opts = append(opts, testcore.WithWorkerService()) } return opts } @@ -3596,7 +3596,7 @@ func testUpdateScheduleBlobSizeLimit(t *testing.T, newContext contextFactory) { func TestScheduleCreationRolloutPercent(t *testing.T) { opts := append(scheduleCommonOpts(t), // V1 worker is needed because at 50% rollout some schedules land on V1. - testcore.WithWorkerService("V1 scheduler"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true), testcore.WithDynamicConfig(dynamicconfig.CHASMSchedulerCreationRolloutPercent, 50), ) diff --git a/tests/task_queue_stats_test.go b/tests/task_queue_stats_test.go index 22a2fde478e..3140ad66d7f 100644 --- a/tests/task_queue_stats_test.go +++ b/tests/task_queue_stats_test.go @@ -67,6 +67,7 @@ func newTaskQueueStatsContext( extraOpts ...testcore.TestOption, ) *taskQueueStatsContext { opts := []testcore.TestOption{ + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.EnableDeploymentVersions, true), testcore.WithDynamicConfig(dynamicconfig.FrontendEnableWorkerVersioningWorkflowAPIs, true), testcore.WithDynamicConfig(dynamicconfig.MatchingUseNewMatcher, usePriMatcher), @@ -94,8 +95,7 @@ type TaskQueueStatsSuite struct { } func TestTaskQueueStats_Pri_Suite(t *testing.T) { - testcore.UseSuiteScopedCluster(t) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. - parallelsuite.RunLegacySequential(t, &TaskQueueStatsSuite{}, true) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. + parallelsuite.Run(t, &TaskQueueStatsSuite{}, true) } func (s *TaskQueueStatsSuite) newTaskQueueStatsContext( @@ -189,7 +189,7 @@ func (s *TaskQueueStatsSuite) TestAddMultipleTasks_ValidateStats_Cached(usePriMa func (s *TaskQueueStatsSuite) TestVersioningSuite(usePriMatcher bool) { for _, behavior := range testcore.AllMatchingBehaviors() { s.T().Run(behavior.Name()+"Suite", func(t *testing.T) { //nolint:testifylint // nested parallelsuite.Run needs raw *testing.T - parallelsuite.RunLegacySequential(t, &TaskQueueStatsVersionSuite{}, usePriMatcher, behavior) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. + parallelsuite.Run(t, &TaskQueueStatsVersionSuite{}, usePriMatcher, behavior) }) } } diff --git a/tests/testcore/onebox.go b/tests/testcore/onebox.go index 26ed8ef7816..7e94f0f6763 100644 --- a/tests/testcore/onebox.go +++ b/tests/testcore/onebox.go @@ -923,6 +923,10 @@ func sdkClientFactoryProvider( } func (c *TemporalImpl) setNexusCallbackURL() { + addrs := c.hostsByProtocolByService[httpProtocol][primitives.FrontendService].All + if len(addrs) == 0 { + return + } // Set Nexus callback URL with the cluster's HTTP address. This is a sensible default to avoid // users to need to manually set this. //nolint:revive // test callback endpoints are served by the local HTTP API in functional tests diff --git a/tests/testcore/test_cluster_pool.go b/tests/testcore/test_cluster_pool.go index 8211c7fe403..5e74fc94fd8 100644 --- a/tests/testcore/test_cluster_pool.go +++ b/tests/testcore/test_cluster_pool.go @@ -1,6 +1,7 @@ package testcore import ( + "fmt" "os" "runtime" "strconv" @@ -14,28 +15,35 @@ import ( var testClusterRouter *clusterRouter func init() { - sharedSize := max(1, runtime.GOMAXPROCS(0)/2) - if v := os.Getenv("TEMPORAL_TEST_SHARED_CLUSTERS"); v != "" { - n, err := strconv.Atoi(v) - if err != nil || n <= 0 { - panic("TEMPORAL_TEST_SHARED_CLUSTERS must be a positive integer") + applyPoolSizeOverride := func(size *int, envVar string) { + if v := os.Getenv(envVar); v != "" { + n, err := strconv.Atoi(v) + if err != nil { + panic(fmt.Sprintf("Failed to parse %s with %v", envVar, err)) + } else if n <= 0 { + panic(fmt.Sprintf("%s must be a positive integer", envVar)) + } + *size = n } - sharedSize = n } + sharedSize := max(1, runtime.GOMAXPROCS(0)/2) + applyPoolSizeOverride(&sharedSize, "TEMPORAL_TEST_SHARED_CLUSTERS") + + // System workers require significantly more memory, so default to a smaller pool. + sharedWorkerSize := max(1, runtime.GOMAXPROCS(0)/4) + applyPoolSizeOverride(&sharedWorkerSize, "TEMPORAL_TEST_SHARED_WORKER_CLUSTERS") + dedicatedSize := runtime.GOMAXPROCS(0) - if v := os.Getenv("TEMPORAL_TEST_DEDICATED_CLUSTERS"); v != "" { - n, err := strconv.Atoi(v) - if err != nil || n <= 0 { - panic("TEMPORAL_TEST_DEDICATED_CLUSTERS must be a positive integer") - } - dedicatedSize = n - } + applyPoolSizeOverride(&dedicatedSize, "TEMPORAL_TEST_DEDICATED_CLUSTERS") - testClusterRouter = &clusterRouter{ - shared: newClusterPool(sharedSize, false, 0), - dedicated: newClusterPool(dedicatedSize, true, 0), + r := &clusterRouter{ + shared: newClusterPool(sharedSize, false, 0), + sharedWithWorker: newClusterPool(sharedWorkerSize, false, 0), + dedicated: newClusterPool(dedicatedSize, true, 0), } + r.clusterCreatorFn = r.createCluster + testClusterRouter = r } // clusterPool manages a fixed number of test [clusterPoolSlot]s. @@ -158,9 +166,14 @@ func (s *clusterPoolSlot) tearDownLocked(t *testing.T) { // clusterRouter routes tests to shared/dedicated [clusterPool] or [suiteScopedCluster]s. type clusterRouter struct { - shared *clusterPool - dedicated *clusterPool - suiteScoped sync.Map + shared *clusterPool + sharedWithWorker *clusterPool + dedicated *clusterPool + suiteScoped sync.Map + + // clusterCreatorFn is the function used to create new clusters. This allows us to inject a mock + // creation function in unit tests of this module to avoid spinning up clusters. + clusterCreatorFn func(*testing.T, map[dynamicconfig.Key]any, bool, []TestClusterOption) *FunctionalTestBase } // suiteScopedCluster owns one lazily created legacy suite cluster. @@ -196,24 +209,29 @@ func UseSuiteScopedCluster(t *testing.T) { }) } -func (p *clusterRouter) get(t *testing.T, dedicated bool, dynamicConfig map[dynamicconfig.Key]any, clusterOpts []TestClusterOption) (tb *FunctionalTestBase) { +func (p *clusterRouter) get(t *testing.T, dedicated bool, needWorkerService bool, dynamicConfig map[dynamicconfig.Key]any, clusterOpts []TestClusterOption) (tb *FunctionalTestBase) { defer func() { if tb != nil { tb.RegisterTest(t) } }() if dedicated || len(dynamicConfig) > 0 || len(clusterOpts) > 0 { - return p.getDedicated(t, dynamicConfig, clusterOpts) + return p.getDedicated(t, needWorkerService, dynamicConfig, clusterOpts) } if cluster := p.getSuiteScoped(t); cluster != nil { return cluster } - return p.getShared(t) + return p.getShared(t, needWorkerService) } -func (p *clusterRouter) getShared(t *testing.T) *FunctionalTestBase { +func (p *clusterRouter) getShared(t *testing.T, needWorkerService bool) *FunctionalTestBase { + if needWorkerService { + return p.sharedWithWorker.get(t, func() *FunctionalTestBase { + return p.clusterCreatorFn(t, nil, true, []TestClusterOption{withWorkerService(true)}) + }) + } return p.shared.get(t, func() *FunctionalTestBase { - return p.createCluster(t, nil, true, nil) + return p.clusterCreatorFn(t, nil, true, nil) }) } @@ -235,17 +253,20 @@ func (p *clusterRouter) getSuiteScoped(t *testing.T) *FunctionalTestBase { // TODO(stephan, #10580): remove this workaround once the proper cluster-pool fix lands. // Enable the worker service on suite-scoped clusters. The only current user (Versioning3) needs the system // worker for worker-deployment APIs. - suiteCluster.cluster = p.createCluster(t, nil, true, []TestClusterOption{withWorkerService(true)}) + suiteCluster.cluster = p.clusterCreatorFn(t, nil, true, []TestClusterOption{withWorkerService(true)}) }) suiteCluster.cluster.SetT(t) return suiteCluster.cluster } -func (p *clusterRouter) getDedicated(t *testing.T, dynamicConfig map[dynamicconfig.Key]any, clusterOpts []TestClusterOption) *FunctionalTestBase { - if len(dynamicConfig) > 0 || len(clusterOpts) > 0 { - // Custom config or fx options require a fresh cluster (can't reuse). +func (p *clusterRouter) getDedicated(t *testing.T, needWorkerService bool, dynamicConfig map[dynamicconfig.Key]any, clusterOpts []TestClusterOption) *FunctionalTestBase { + if len(dynamicConfig) > 0 || len(clusterOpts) > 0 || needWorkerService { + // Always create a new cluster in the following cases: + // - custom configs are set, since they can't be shared across tests + // - worker service is needed, since goroutines (system workers, matching partition managers) don't clean up between + // tests and would accumulate in a long-lived pooled cluster. p.dedicated.reserveSlot(t) - cluster := p.createCluster(t, dynamicConfig, false, clusterOpts) + cluster := p.clusterCreatorFn(t, dynamicConfig, false, append(clusterOpts, withWorkerService(needWorkerService))) // Register cleanup to tear down the cluster when the test completes. t.Cleanup(func() { @@ -257,9 +278,9 @@ func (p *clusterRouter) getDedicated(t *testing.T, dynamicConfig map[dynamicconf return cluster } - // If no custom config is provided, reuse an existing cluster. + // If no custom config or worker service is needed, reuse an existing cluster. return p.dedicated.get(t, func() *FunctionalTestBase { - return p.createCluster(t, nil, false, nil) + return p.clusterCreatorFn(t, nil, false, nil) }) } diff --git a/tests/testcore/test_cluster_pool_test.go b/tests/testcore/test_cluster_pool_test.go index 2243ed81385..a6f23cea207 100644 --- a/tests/testcore/test_cluster_pool_test.go +++ b/tests/testcore/test_cluster_pool_test.go @@ -8,7 +8,39 @@ import ( "go.temporal.io/server/common/dynamicconfig" ) -func TestGlobalOverridesSurviveTestCleanup(t *testing.T) { +// clusterCreateFnArgs allow us to inspect the arguments passed to the cluster creation function in unit tests of this module. +// This works when paired with a mocked clusterCreationFn that populates a slice of these structs for later inspection. +type clusterCreateFnArgs struct { + shared bool + workerEnabled bool +} + +// newMockClusterPoolRouter creates a clusterRouter with small pools and a mockClusterCreationFn +// that simply records calls. It allows testing routing logic without starting a real server. +// Note that for the function calls, we need to return *[]clusterCreateFnArgs so that the appended +// clusterCreateFnArgs are visible to the caller. +func newMockClusterPoolRouter(t *testing.T) (*clusterRouter, *[]clusterCreateFnArgs) { + t.Helper() + + var calls []clusterCreateFnArgs + mockClusterCreationFn := func(t *testing.T, dc map[dynamicconfig.Key]any, shared bool, opts []TestClusterOption) *FunctionalTestBase { + // Keep the worker service off unless explicitly enabled via WithWorkerService (this mirrors the actual createClusterFn implementation). + baseOpts := []TestClusterOption{withWorkerService(false)} + params := ApplyTestClusterOptions(append(baseOpts, opts...)) + calls = append(calls, clusterCreateFnArgs{shared: shared, workerEnabled: params.EnableWorkerService}) + return &FunctionalTestBase{} + } + + r := &clusterRouter{ + shared: newClusterPool(2, false, 0), + sharedWithWorker: newClusterPool(1, false, 0), // small pool to force reuse of the same cluster for multiple tests + dedicated: newClusterPool(2, true, 0), + clusterCreatorFn: mockClusterCreationFn, + } + return r, &calls +} + +func TestClusterPool_GlobalOverridesSurviveTestCleanup(t *testing.T) { var dcClient *dynamicconfig.MemoryClient t.Run("create", func(t *testing.T) { @@ -104,3 +136,46 @@ func TestClusterPool_PoisonedActiveClusterSwapsWithoutRecycling(t *testing.T) { require.Equal(t, 2, slot.activeLeases) require.Equal(t, 1, slot.leaseCount) } + +func TestClusterPool_SharedWorkerServiceRouting(t *testing.T) { + t.Run("worker service uses and reuses shared-with-worker pool", func(t *testing.T) { + router, calls := newMockClusterPoolRouter(t) + var first, second *FunctionalTestBase + + t.Run("first call creates cluster", func(t *testing.T) { + first = router.get(t, false /* dedicated */, true /* workerService */, nil, nil) + }) + require.Equal(t, 1, router.sharedWithWorker.allSlots[0].leaseCount, "sharedWithWorker should have one lease") + require.Equal(t, 0, router.shared.allSlots[0].leaseCount, "shared pool should be unused") + require.Len(t, *calls, 1) + require.Equal(t, clusterCreateFnArgs{shared: true, workerEnabled: true}, (*calls)[0]) + + t.Run("second call reuses cluster", func(t *testing.T) { + second = router.get(t, false /* dedicated */, true /* workerService */, nil, nil) + }) + require.Same(t, first, second, "second call should return the same cluster, not allocate a new one") + require.Len(t, *calls, 1, "no new cluster should have been created") + require.Equal(t, 2, router.sharedWithWorker.allSlots[0].leaseCount, "slot 0 should have been leased twice") + require.Equal(t, 0, router.shared.allSlots[0].leaseCount+router.shared.allSlots[1].leaseCount, + "plain shared pool should be unused") + }) + + t.Run("worker service with dedicated uses dedicated pool", func(t *testing.T) { + router, calls := newMockClusterPoolRouter(t) + + router.get(t, true /* dedicated */, true /* workerService */, nil, nil) + require.Equal(t, 0, router.sharedWithWorker.allSlots[0].leaseCount, "sharedWithWorker pool should be unused") + require.Len(t, *calls, 1) + require.Equal(t, clusterCreateFnArgs{shared: false, workerEnabled: true}, (*calls)[0]) + }) + + t.Run("no worker service uses plain shared pool", func(t *testing.T) { + router, calls := newMockClusterPoolRouter(t) + + router.get(t, false /* dedicated */, false /* workerService */, nil, nil) + require.Equal(t, 1, router.shared.allSlots[0].leaseCount, "shared pool should have one lease") + require.Equal(t, 0, router.sharedWithWorker.allSlots[0].leaseCount, "sharedWithWorker pool should be unused") + require.Len(t, *calls, 1) + require.Equal(t, clusterCreateFnArgs{shared: true, workerEnabled: false}, (*calls)[0]) + }) +} diff --git a/tests/testcore/test_env.go b/tests/testcore/test_env.go index da8a97a770a..b17e96e5925 100644 --- a/tests/testcore/test_env.go +++ b/tests/testcore/test_env.go @@ -90,6 +90,7 @@ type TestOption func(*testOptions) type testOptions struct { dedicatedCluster bool + needWorkerService bool dedicatedReason string disableTestloggerFailure bool dynamicConfigSettings []dynamicConfigOverride @@ -113,6 +114,17 @@ func WithDedicatedCluster() TestOption { } } +// ForLeakChecker marks the test as intentionally using a dedicated cluster for +// leak measurement purposes, satisfying the [WithDedicatedCluster] guard without +// requiring an actual cluster-global side effect. +// +// NOTE: This option should NEVER be use anywhere outside of the object leak checker. +func ForLeakChecker() TestOption { + return func(o *testOptions) { + o.dedicatedReason = "leak checker" + } +} + // WithDisableTestloggerFailure disables the test logger's behavior of failing // the test when an error log matches a registered expectation (e.g. soft-assert // errors tagged with tag.FailedAssertion). Use for tests that intentionally @@ -152,12 +164,13 @@ func WithFxOptions(serviceName primitives.ServiceName, opts ...fx.Option) TestOp } // WithWorkerService enables the system worker service. The service is off by -// default to avoid the worker overhead. This implies a dedicated cluster. -func WithWorkerService(reason string) TestOption { +// default to avoid the worker overhead. +// If this is not coupled with any option that requires a dedicated cluster, +// we will create a cluster that will be recycled by other tests that also +// need the system worker service. +func WithWorkerService() TestOption { return func(o *testOptions) { - o.dedicatedCluster = true - o.clusterOptions = append(o.clusterOptions, withWorkerService(true)) - o.dedicatedReason = "worker service required: " + reason + o.needWorkerService = true } } @@ -278,7 +291,7 @@ func NewEnv(t *testing.T, opts ...TestOption) *TestEnv { } // Obtain the test cluster from the router. - base := testClusterRouter.get(t, options.dedicatedCluster, startupConfig, options.clusterOptions) + base := testClusterRouter.get(t, options.dedicatedCluster, options.needWorkerService, startupConfig, options.clusterOptions) cluster := base.GetTestCluster() // Create a dedicated namespace for the test to help with test isolation. diff --git a/tests/versioning_3_test.go b/tests/versioning_3_test.go index ec0c966aed9..d28a75db3a2 100644 --- a/tests/versioning_3_test.go +++ b/tests/versioning_3_test.go @@ -6712,7 +6712,7 @@ func (s *Versioning3Suite) TestPinnedCaN_FailedTransientNotificationRefiresDespi // made current again, then reset-by-build-ID resets the workflow before v2 usage // so the reset run resumes on v1. func (s *Versioning3Suite) TestPinnedCaN_ResetByBuildIDAfterRollback() { - env := s.setupEnv(testcore.WithWorkerService("batch operations")) + env := s.setupEnv(testcore.WithWorkerService()) tv := env.Tv() tv1 := tv.WithBuildIDNumber(1) diff --git a/tests/worker_deployment_version_test.go b/tests/worker_deployment_version_test.go index aa3e7874503..4b8da3874f5 100644 --- a/tests/worker_deployment_version_test.go +++ b/tests/worker_deployment_version_test.go @@ -1875,7 +1875,7 @@ func (s *DeploymentVersionSuite) setAndCheckOverrideWithExpectedOutput(env *test func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetPinned_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. - testcore.WithWorkerService("worker-deployment version membership cache test"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) @@ -2854,7 +2854,7 @@ func (s *DeploymentVersionSuite) makeAutoUpgradeOverride() *workflowpb.Versionin func (s *DeploymentVersionSuite) TestStartWorkflowExecution_WithPinnedOverride_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. - testcore.WithWorkerService("worker-deployment version membership cache test"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) @@ -2906,7 +2906,7 @@ func (s *DeploymentVersionSuite) TestStartWorkflowExecution_WithUnpinnedOverride func (s *DeploymentVersionSuite) TestSignalWithStartWorkflowExecution_WithPinnedOverride_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. - testcore.WithWorkerService("worker-deployment version membership cache test"), + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) diff --git a/tests/worker_registry_test.go b/tests/worker_registry_test.go index 5f74681d4e0..968b9841e5d 100644 --- a/tests/worker_registry_test.go +++ b/tests/worker_registry_test.go @@ -22,12 +22,12 @@ type WorkerRegistryTestSuite struct { } func TestWorkerRegistryTestSuite(t *testing.T) { - testcore.UseSuiteScopedCluster(t) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. - parallelsuite.RunLegacySequential(t, &WorkerRegistryTestSuite{}) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. + parallelsuite.Run(t, &WorkerRegistryTestSuite{}) } func (s *WorkerRegistryTestSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv { baseOpts := []testcore.TestOption{ + testcore.WithWorkerService(), testcore.WithDynamicConfig(dynamicconfig.WorkerHeartbeatsEnabled, true), } return testcore.NewEnv(s.T(), append(baseOpts, opts...)...) diff --git a/tests/workflow_alias_search_attribute_test.go b/tests/workflow_alias_search_attribute_test.go index 5f11a2d7253..86b4c781978 100644 --- a/tests/workflow_alias_search_attribute_test.go +++ b/tests/workflow_alias_search_attribute_test.go @@ -27,12 +27,13 @@ type WorkflowAliasSearchAttributeTestSuite struct { } func TestWorkflowAliasSearchAttributeTestSuite(t *testing.T) { - testcore.UseSuiteScopedCluster(t) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. - parallelsuite.RunLegacySequential(t, &WorkflowAliasSearchAttributeTestSuite{}) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. + parallelsuite.Run(t, &WorkflowAliasSearchAttributeTestSuite{}) } func (s *WorkflowAliasSearchAttributeTestSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv { opts = append([]testcore.TestOption{ + testcore.WithWorkerService(), + // Keep deployment versions short because worker-deployment system workflow IDs must fit into 255 characters. testcore.WithTestVars(func(tv *testvars.TestVars) *testvars.TestVars { return tv.WithDeploymentSeries("alias-sa").WithBuildID("v1") diff --git a/tests/workflow_reset_test.go b/tests/workflow_reset_test.go index d189f7826c1..f18637d917f 100644 --- a/tests/workflow_reset_test.go +++ b/tests/workflow_reset_test.go @@ -215,7 +215,7 @@ func (s *WorkflowResetSuite) TestOriginalExecutionRunId() { // Test that the workflow options are updated when the workflow is reset. func (s *WorkflowResetSuite) TestResetWorkflowWithOptionsUpdate() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService("worker-deployment version registration")) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) deploymentName := "testing" buildID := "v.123" @@ -293,7 +293,7 @@ func (s *WorkflowResetSuite) TestResetWorkflowWithOptionsUpdate() { // Test batch reset operation with version update as post reset operation func (s *WorkflowResetSuite) TestBatchResetWithOptionsUpdate() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch operations")) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) deploymentName := "batch-testing" buildID := "v.456" From 5a001a3c9c4945f070552a677740e6c5ee1acadb Mon Sep 17 00:00:00 2001 From: long-nt-tran Date: Tue, 7 Jul 2026 12:10:34 -0400 Subject: [PATCH 2/5] Address PR feedback --- Makefile | 6 ++- docs/development/testing.md | 1 + tests/activity_api_batch_reset_test.go | 2 +- tests/activity_api_batch_unpause_test.go | 6 +-- .../activity_api_batch_update_options_test.go | 4 +- tests/advanced_visibility_test.go | 2 +- tests/chasm_test.go | 2 +- tests/client_misc_test.go | 6 +-- tests/dlq_test.go | 6 +-- tests/leakcheck/leak_test.go | 5 +- tests/namespace_test.go | 2 +- tests/poller_scaling_test.go | 2 +- tests/purge_dlq_tasks_api_test.go | 6 +-- tests/schedule_migration_test.go | 32 ++++++------- tests/schedule_test.go | 4 +- tests/task_queue_stats_test.go | 2 +- tests/testcore/onebox.go | 8 ++-- tests/testcore/test_cluster_pool.go | 48 ++++++++++++------- tests/testcore/test_cluster_pool_test.go | 30 ++++++------ tests/testcore/test_env.go | 21 ++++---- tests/versioning_3_test.go | 2 +- tests/worker_deployment_version_test.go | 6 +-- tests/worker_registry_test.go | 2 +- tests/workflow_alias_search_attribute_test.go | 2 +- tests/workflow_reset_test.go | 4 +- 25 files changed, 112 insertions(+), 99 deletions(-) diff --git a/Makefile b/Makefile index 04dce8d5b23..eb7f4284800 100644 --- a/Makefile +++ b/Makefile @@ -480,11 +480,15 @@ build-tests: @printf $(COLOR) "Build tests..." @CGO_ENABLED=$(CGO_ENABLED) go test $(TEST_TAG_FLAG) -exec="true" -count=0 $(TEST_DIRS) -unit-test: clean-test-output +unit-test: clean-test-output unit-test-testcore @printf $(COLOR) "Run unit tests..." @CGO_ENABLED=$(CGO_ENABLED) go test $(UNIT_TEST_DIRS) $(COMPILED_TEST_ARGS) 2>&1 | tee -a test.log @$(MAKE) verify-test-log +unit-test-testcore: clean-test-output + @printf $(COLOR) "Run testcore unit tests..." + @CGO_ENABLED=$(CGO_ENABLED) go test ./tests/testcore/ -run "TestClusterPool" $(COMPILED_TEST_ARGS) 2>&1 | tee -a test.log + integration-test: clean-test-output @printf $(COLOR) "Run integration tests..." @CGO_ENABLED=$(CGO_ENABLED) go test $(INTEGRATION_TEST_DIRS) $(COMPILED_TEST_ARGS) 2>&1 | tee -a test.log diff --git a/docs/development/testing.md b/docs/development/testing.md index 92a412e681e..975e1e06d14 100644 --- a/docs/development/testing.md +++ b/docs/development/testing.md @@ -18,6 +18,7 @@ This document describes the project's testing setup, utilities and best practice - `TEMPORAL_TEST_LOG_FILE_LEVEL`: Minimum verbosity level written to the log file. Available levels: `debug` (default), `info`, `warn`, `error`, `fatal` - `TEMPORAL_TEST_OTEL_OUTPUT`: Enables OpenTelemetry (OTEL) trace output for failed tests to the provided file path. - `TEMPORAL_TEST_SHARED_CLUSTERS`: Number of shared clusters in the pool. Each can be used by multiple tests simultaneously. +- `TEMPORAL_TEST_SHARED_WORKER_CLUSTERS`: Number of shared clusters with system workers enabled in the pool. Each can be used by multiple tests simultaneously. - `TEMPORAL_TEST_DEDICATED_CLUSTERS`: Number of dedicated clusters in the pool. Each can be used by one test only at a time. - `TEMPORAL_TEST_TIMEOUT`: Sets the duration timeout per test (e.g., `90s` for 90 seconds). This can be overridden per-test using `testcore.WithTimeout()`. The timeout is multiplied by `debug.TimeoutMultiplier` when debugging. - `TEMPORAL_TEST_DATA_ENCODING`: Controls the encoding used for persistence DataBlobs. Available options: `proto3` (default) or `json`. diff --git a/tests/activity_api_batch_reset_test.go b/tests/activity_api_batch_reset_test.go index 0c2a58b2e8f..332e945bf5a 100644 --- a/tests/activity_api_batch_reset_test.go +++ b/tests/activity_api_batch_reset_test.go @@ -32,7 +32,7 @@ func TestActivityAPIBatchResetClientTestSuite(t *testing.T) { func newBatchResetEnv(t *testing.T) *testcore.TestEnv { return testcore.NewEnv( t, - testcore.WithWorkerService(), + testcore.WithWorkerService("batch operations"), // These tests intentionally start multiple batch operations in the same namespace. // The default per-namespace limit is 1, so raise it to the functional test limit. testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, testcore.ClientSuiteLimit), diff --git a/tests/activity_api_batch_unpause_test.go b/tests/activity_api_batch_unpause_test.go index fa4322a0177..e954f833421 100644 --- a/tests/activity_api_batch_unpause_test.go +++ b/tests/activity_api_batch_unpause_test.go @@ -92,7 +92,7 @@ func (s *ActivityApiBatchUnpauseClientTestSuite) createWorkflow(env *testcore.Te } func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Success() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch unpause activity operations require system worker batcher")) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() @@ -199,7 +199,7 @@ func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Succes } func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Failed() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch unpause activity operations require system worker batcher")) // neither activity type not "match all" is provided _, err := env.SdkClient().WorkflowService().StartBatchOperation(context.Background(), &workflowservice.StartBatchOperationRequest{ @@ -237,7 +237,7 @@ func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Failed // This is an end-to-end complement to the unit-level checkNamespace tests: it // exercises the full path from StartBatchOperation through the batcher worker. func (s *ActivityApiBatchUnpauseClientTestSuite) TestBatchTerminate_NamespaceIsolation() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch terminate namespace isolation test uses system worker batcher")) ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() diff --git a/tests/activity_api_batch_update_options_test.go b/tests/activity_api_batch_update_options_test.go index 3edc4d2d0a1..5c34ca96de6 100644 --- a/tests/activity_api_batch_update_options_test.go +++ b/tests/activity_api_batch_update_options_test.go @@ -46,7 +46,7 @@ func (s *ActivityAPIBatchUpdateOptionsSuite) createBatchUpdateOptionsWorkflow(en func (s *ActivityAPIBatchUpdateOptionsSuite) TestActivityBatchUpdateOptionsSuccess() { env := testcore.NewEnv(s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("batch operations"), testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, testcore.ClientSuiteLimit), ) @@ -194,7 +194,7 @@ func (s *ActivityAPIBatchUpdateOptionsSuite) TestActivityBatchUpdateOptionsSucce func (s *ActivityAPIBatchUpdateOptionsSuite) TestActivityBatchUpdateOptionsFailed() { env := testcore.NewEnv(s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("batch operations"), testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, testcore.ClientSuiteLimit), ) diff --git a/tests/advanced_visibility_test.go b/tests/advanced_visibility_test.go index 9b4f90f56b1..49077aa2aed 100644 --- a/tests/advanced_visibility_test.go +++ b/tests/advanced_visibility_test.go @@ -2411,7 +2411,7 @@ func (s *AdvancedVisibilitySuite) TestWorkerTaskReachability_Unversioned_InTaskQ } func (s *AdvancedVisibilitySuite) TestBuildIdScavenger_DeletesUnusedBuildId(enableUnifiedQueryConverter bool) { - env := s.newTestEnv(enableUnifiedQueryConverter, testcore.WithWorkerService()) + env := s.newTestEnv(enableUnifiedQueryConverter, testcore.WithWorkerService("build id scavenger workflow")) tq := s.T().Name() buildIdv0 := s.T().Name() + "-v0" buildIdv1 := s.T().Name() + "-v1" diff --git a/tests/chasm_test.go b/tests/chasm_test.go index d3e8af5219b..970b1d3b040 100644 --- a/tests/chasm_test.go +++ b/tests/chasm_test.go @@ -63,7 +63,7 @@ func newChasmTestEnv(t *testing.T, unified bool) chasmTestEnv { env := testcore.NewEnv( t, testcore.WithDedicatedCluster(), - testcore.WithWorkerService(), + testcore.WithWorkerService("delete namespace workflow"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.VisibilityEnableUnifiedQueryConverter, unified), testcore.WithDynamicConfig(dynamicconfig.DeleteNamespaceUseChasmDeleteExecution, true), diff --git a/tests/client_misc_test.go b/tests/client_misc_test.go index caf46b97230..44cbd305797 100644 --- a/tests/client_misc_test.go +++ b/tests/client_misc_test.go @@ -1063,7 +1063,7 @@ func (s *ClientMiscTestSuite) Test_StickyWorkerRestartWorkflowTask() { } func (s *ClientMiscTestSuite) TestBatchSignal() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch operations")) type myData struct { Stuff string @@ -1118,7 +1118,7 @@ func (s *ClientMiscTestSuite) TestBatchSignal() { } func (s *ClientMiscTestSuite) TestBatchReset() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch operations")) var count atomic.Int32 activityFn := func(ctx context.Context) (int32, error) { @@ -1182,7 +1182,7 @@ func (s *ClientMiscTestSuite) TestBatchReset() { } func (s *ClientMiscTestSuite) TestBatchResetByBuildId() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch operations")) tq := testcore.RandomizeStr(s.T().Name()) buildPrefix := uuid.NewString()[:6] + "-" buildIdv1 := buildPrefix + "v1" diff --git a/tests/dlq_test.go b/tests/dlq_test.go index ff78342df1d..ee2c43178c4 100644 --- a/tests/dlq_test.go +++ b/tests/dlq_test.go @@ -252,7 +252,7 @@ func (s *DLQSuite) TestReadArtificialDLQTasks() { // DLQ, this test then purges the DLQ and verifies that the task was deleted. // This test will then call DescribeDLQJob and CancelDLQJob api to verify. func (s *DLQSuite) TestPurgeRealWorkflow() { - env := s.newTestEnv(testcore.WithWorkerService()) + env := s.newTestEnv(testcore.WithWorkerService("dlq purge workflow")) _, dlqMessageID := s.executeDoomedWorkflow(env) @@ -281,7 +281,7 @@ func (s *DLQSuite) TestPurgeRealWorkflow() { // above test is more for testing specific CLI flags when reading from the DLQ. // This test will then call DescribeDLQJob and CancelDLQJob api to verify. func (s *DLQSuite) TestMergeRealWorkflow() { - env := s.newTestEnv(testcore.WithWorkerService()) + env := s.newTestEnv(testcore.WithWorkerService("dlq merge workflow")) // Verify that we can execute a normal workflow. run := s.executeWorkflow(env, "dlq-test-ok-workflow-id") @@ -324,7 +324,7 @@ func (s *DLQSuite) TestMergeRealWorkflow() { } func (s *DLQSuite) TestCancelRunningMerge() { - env := s.newTestEnv(testcore.WithWorkerService()) + env := s.newTestEnv(testcore.WithWorkerService("dlq merge workflow")) env.deleteBlockCh = make(chan any) // Execute several doomed workflows. diff --git a/tests/leakcheck/leak_test.go b/tests/leakcheck/leak_test.go index 7ed2583ddb0..6082421678e 100644 --- a/tests/leakcheck/leak_test.go +++ b/tests/leakcheck/leak_test.go @@ -150,8 +150,9 @@ func buildRunTeardownCluster(t *testing.T, leakCheck *objectleak.ObjectLeakCheck t.Run("cluster", func(t *testing.T) { env := testcore.NewEnv(t, testcore.WithDedicatedCluster(), - testcore.WithWorkerService(), - testcore.ForLeakChecker()) + testcore.WithWorkerService("leak checker needs worker service to exercise full server path"), + testcore.WithMTLS(), // Using a global option (MTLS) to satisfy the dedicated cluster guard + ) env.SdkWorker().RegisterWorkflow(smokeWorkflow) run, err := env.SdkClient().ExecuteWorkflow( diff --git a/tests/namespace_test.go b/tests/namespace_test.go index d581de7a27c..d0fe884369a 100644 --- a/tests/namespace_test.go +++ b/tests/namespace_test.go @@ -36,7 +36,7 @@ func TestNamespaceSuite(t *testing.T) { func (s *namespaceTestSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv { baseOpts := []testcore.TestOption{ - testcore.WithWorkerService(), + testcore.WithWorkerService("namespace deletion tests require the system worker service"), testcore.WithDynamicConfig(dynamicconfig.TransferProcessorUpdateAckInterval, 1*time.Second), testcore.WithDynamicConfig(dynamicconfig.VisibilityProcessorUpdateAckInterval, 1*time.Second), } diff --git a/tests/poller_scaling_test.go b/tests/poller_scaling_test.go index 10c0da48cf7..fc5acb00ad3 100644 --- a/tests/poller_scaling_test.go +++ b/tests/poller_scaling_test.go @@ -36,7 +36,7 @@ func TestPollerScalingFunctionalSuite(t *testing.T) { func (s *PollerScalingIntegSuite) setupEnv(opts ...testcore.TestOption) *testcore.TestEnv { opts = append([]testcore.TestOption{ - testcore.WithWorkerService(), + testcore.WithWorkerService("poller scaling test uses worker deployment system workflows"), // Force one partition so we can reliably see the backlog testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1), diff --git a/tests/purge_dlq_tasks_api_test.go b/tests/purge_dlq_tasks_api_test.go index 33fa2212ea1..776b6e04ed7 100644 --- a/tests/purge_dlq_tasks_api_test.go +++ b/tests/purge_dlq_tasks_api_test.go @@ -40,7 +40,7 @@ func (s *PurgeDLQTasksSuite) TestPurgeDLQTasks() { }{ { name: "HappyPath", - envOptions: []testcore.TestOption{testcore.WithWorkerService()}, + envOptions: []testcore.TestOption{testcore.WithWorkerService("purge DLQ workflow")}, }, { name: "MissingSourceCluster", @@ -58,7 +58,7 @@ func (s *PurgeDLQTasksSuite) TestPurgeDLQTasks() { }, { name: "QueueDoesNotExist", - envOptions: []testcore.TestOption{testcore.WithWorkerService()}, + envOptions: []testcore.TestOption{testcore.WithWorkerService("purge DLQ workflow")}, mutateRequest: func(request *adminservice.PurgeDLQTasksRequest) { request.DlqKey.TargetCluster = "does-not-exist" }, @@ -67,7 +67,7 @@ func (s *PurgeDLQTasksSuite) TestPurgeDLQTasks() { { name: "DeleteTasksUnavailableError", envOptions: []testcore.TestOption{ - testcore.WithWorkerService(), + testcore.WithWorkerService("purge DLQ workflow"), testcore.WithPersistenceFaultInjection(&config.FaultInjection{ Injector: func(target config.FaultInjectionTarget) error { if target.Store == config.QueueV2Name && diff --git a/tests/schedule_migration_test.go b/tests/schedule_migration_test.go index 7bac8af086d..efc1ec7ef0c 100644 --- a/tests/schedule_migration_test.go +++ b/tests/schedule_migration_test.go @@ -49,7 +49,7 @@ func TestScheduleMigrationTestSuite(t *testing.T) { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2AlreadyExists() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -291,7 +291,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1BlockedBySentine func (s *ScheduleMigrationTestSuite) TestScheduleMigrationDynamicConfig() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigration, true), ) @@ -401,7 +401,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationDynamicConfig() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV1ToV2() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -538,7 +538,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV1ToV2() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, false), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, false), @@ -745,7 +745,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1Idempotent() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, false), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, false), @@ -816,7 +816,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1Idempotent() { func (s *ScheduleMigrationTestSuite) TestCHASMScheduleDescribeAfterDisablingCreationAndMigration() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigration, true), @@ -918,7 +918,7 @@ func (s *ScheduleMigrationTestSuite) TestCHASMScheduleDescribeAfterDisablingCrea func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1RoutingFallback() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, true), @@ -1054,7 +1054,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1RoutingFallback( func (s *ScheduleMigrationTestSuite) TestScheduleUpdateAfterDelete() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, true), @@ -1165,7 +1165,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleUpdateAfterDelete() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV1ToV2WithClosedV2() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -1322,7 +1322,7 @@ func TestScheduleMigrationV1ToV2NoDuplicateRecentActions(t *testing.T) { // a CHASM sentinel (which would block the migration activity). env := testcore.NewEnv( t, - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithSdkWorker(), ) @@ -1470,7 +1470,7 @@ func TestScheduleMigrationDeferredWithRunningWorkflow(t *testing.T) { // (workflow-backed) schedule rather than a CHASM sentinel. env := testcore.NewEnv( t, - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigrationWithRunningWorkflows, false), ) @@ -1606,7 +1606,7 @@ func TestScheduleMigrationDeferredWithRunningWorkflow(t *testing.T) { func (s *ScheduleMigrationTestSuite) TestDeleteScheduleContextMetadata() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerSentinels, true), @@ -1810,7 +1810,7 @@ func (s *ScheduleMigrationTestSuite) TestDeleteScheduleContextMetadata() { func (s *ScheduleMigrationTestSuite) TestPatchScheduleContextMetadata() { env := testcore.NewEnv( s.T(), - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerRouting, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerSentinels, true), @@ -1973,7 +1973,7 @@ func (s *ScheduleMigrationTestSuite) TestPatchScheduleContextMetadata() { func TestScheduleMigration_StaleRunningDoesNotSkipPending(t *testing.T) { env := testcore.NewEnv( t, - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -2121,7 +2121,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationRolloutPercent() { t := s.T() env := testcore.NewEnv( t, - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigration, true), testcore.WithDynamicConfig(dynamicconfig.CHASMSchedulerMigrationRolloutPercent, 50), @@ -2247,7 +2247,7 @@ func TestScheduleMigration_NoRunningWorkflows_GeneratorStarts(t *testing.T) { env := testcore.NewEnv( t, - testcore.WithWorkerService(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) env.OverrideDynamicConfig(chasmscheduler.CurrentTweakables, tweakables) diff --git a/tests/schedule_test.go b/tests/schedule_test.go index 89043bea544..c62c65e0806 100644 --- a/tests/schedule_test.go +++ b/tests/schedule_test.go @@ -69,7 +69,7 @@ func scheduleCommonOpts(t *testing.T) []testcore.TestOption { } if strings.HasPrefix(t.Name(), "TestScheduleV1") { // only v1 needs the worker service - opts = append(opts, testcore.WithWorkerService()) + opts = append(opts, testcore.WithWorkerService("V1 scheduler")) } return opts } @@ -3596,7 +3596,7 @@ func testUpdateScheduleBlobSizeLimit(t *testing.T, newContext contextFactory) { func TestScheduleCreationRolloutPercent(t *testing.T) { opts := append(scheduleCommonOpts(t), // V1 worker is needed because at 50% rollout some schedules land on V1. - testcore.WithWorkerService(), + testcore.WithWorkerService("V1 scheduler"), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true), testcore.WithDynamicConfig(dynamicconfig.CHASMSchedulerCreationRolloutPercent, 50), ) diff --git a/tests/task_queue_stats_test.go b/tests/task_queue_stats_test.go index 3140ad66d7f..1f37391952d 100644 --- a/tests/task_queue_stats_test.go +++ b/tests/task_queue_stats_test.go @@ -67,7 +67,7 @@ func newTaskQueueStatsContext( extraOpts ...testcore.TestOption, ) *taskQueueStatsContext { opts := []testcore.TestOption{ - testcore.WithWorkerService(), + testcore.WithWorkerService("task queue stats tests use worker deployment system workflows"), testcore.WithDynamicConfig(dynamicconfig.EnableDeploymentVersions, true), testcore.WithDynamicConfig(dynamicconfig.FrontendEnableWorkerVersioningWorkflowAPIs, true), testcore.WithDynamicConfig(dynamicconfig.MatchingUseNewMatcher, usePriMatcher), diff --git a/tests/testcore/onebox.go b/tests/testcore/onebox.go index 7e94f0f6763..456c72a0e66 100644 --- a/tests/testcore/onebox.go +++ b/tests/testcore/onebox.go @@ -231,7 +231,9 @@ func newTemporal(t *testing.T, params *TemporalParams) *TemporalImpl { } // Override Nexus callback URL. This is parameterized on the frontend's HTTP address, // so it can't be overriden in the loop above. - impl.setNexusCallbackURL() + if len(params.HostsByProtocolByService) > 0 { + impl.setNexusCallbackURL() + } // Per-test overrides: cleaned up when the creating test finishes. for k, v := range params.DynamicConfigOverrides { impl.overrideDynamicConfigForTest(t, k, v) @@ -923,10 +925,6 @@ func sdkClientFactoryProvider( } func (c *TemporalImpl) setNexusCallbackURL() { - addrs := c.hostsByProtocolByService[httpProtocol][primitives.FrontendService].All - if len(addrs) == 0 { - return - } // Set Nexus callback URL with the cluster's HTTP address. This is a sensible default to avoid // users to need to manually set this. //nolint:revive // test callback endpoints are served by the local HTTP API in functional tests diff --git a/tests/testcore/test_cluster_pool.go b/tests/testcore/test_cluster_pool.go index 5e74fc94fd8..244e20d6233 100644 --- a/tests/testcore/test_cluster_pool.go +++ b/tests/testcore/test_cluster_pool.go @@ -30,8 +30,7 @@ func init() { sharedSize := max(1, runtime.GOMAXPROCS(0)/2) applyPoolSizeOverride(&sharedSize, "TEMPORAL_TEST_SHARED_CLUSTERS") - // System workers require significantly more memory, so default to a smaller pool. - sharedWorkerSize := max(1, runtime.GOMAXPROCS(0)/4) + sharedWorkerSize := max(1, runtime.GOMAXPROCS(0)/2) applyPoolSizeOverride(&sharedWorkerSize, "TEMPORAL_TEST_SHARED_WORKER_CLUSTERS") dedicatedSize := runtime.GOMAXPROCS(0) @@ -42,7 +41,7 @@ func init() { sharedWithWorker: newClusterPool(sharedWorkerSize, false, 0), dedicated: newClusterPool(dedicatedSize, true, 0), } - r.clusterCreatorFn = r.createCluster + r.createClusterFn = createCluster testClusterRouter = r } @@ -164,6 +163,21 @@ func (s *clusterPoolSlot) tearDownLocked(t *testing.T) { s.leaseCount = 0 } +// clusterRequest carries all the parameters needed to create a test cluster +// in the correct pool. +type clusterRequest struct { + dedicated bool + needWorkerService bool + dynamicConfig map[dynamicconfig.Key]any + clusterOpts []TestClusterOption +} + +// mustBeFresh returns true if the request requires a freshly created cluster, rather than an +// existing one in one of the cluster pools. +func (r clusterRequest) mustBeFresh() bool { return len(r.dynamicConfig) > 0 || len(r.clusterOpts) > 0 } + +func (r clusterRequest) needsDedicated() bool { return r.dedicated || r.mustBeFresh() } + // clusterRouter routes tests to shared/dedicated [clusterPool] or [suiteScopedCluster]s. type clusterRouter struct { shared *clusterPool @@ -171,9 +185,9 @@ type clusterRouter struct { dedicated *clusterPool suiteScoped sync.Map - // clusterCreatorFn is the function used to create new clusters. This allows us to inject a mock + // createClusterFn is the function used to create new clusters. This allows us to inject a mock // creation function in unit tests of this module to avoid spinning up clusters. - clusterCreatorFn func(*testing.T, map[dynamicconfig.Key]any, bool, []TestClusterOption) *FunctionalTestBase + createClusterFn func(*testing.T, map[dynamicconfig.Key]any, bool, []TestClusterOption) *FunctionalTestBase } // suiteScopedCluster owns one lazily created legacy suite cluster. @@ -209,29 +223,29 @@ func UseSuiteScopedCluster(t *testing.T) { }) } -func (p *clusterRouter) get(t *testing.T, dedicated bool, needWorkerService bool, dynamicConfig map[dynamicconfig.Key]any, clusterOpts []TestClusterOption) (tb *FunctionalTestBase) { +func (p *clusterRouter) get(t *testing.T, req clusterRequest) (tb *FunctionalTestBase) { defer func() { if tb != nil { tb.RegisterTest(t) } }() - if dedicated || len(dynamicConfig) > 0 || len(clusterOpts) > 0 { - return p.getDedicated(t, needWorkerService, dynamicConfig, clusterOpts) + if req.needsDedicated() { + return p.getDedicated(t, req) } if cluster := p.getSuiteScoped(t); cluster != nil { return cluster } - return p.getShared(t, needWorkerService) + return p.getShared(t, req.needWorkerService) } func (p *clusterRouter) getShared(t *testing.T, needWorkerService bool) *FunctionalTestBase { if needWorkerService { return p.sharedWithWorker.get(t, func() *FunctionalTestBase { - return p.clusterCreatorFn(t, nil, true, []TestClusterOption{withWorkerService(true)}) + return p.createClusterFn(t, nil, true, []TestClusterOption{withWorkerService(true)}) }) } return p.shared.get(t, func() *FunctionalTestBase { - return p.clusterCreatorFn(t, nil, true, nil) + return p.createClusterFn(t, nil, true, nil) }) } @@ -253,20 +267,20 @@ func (p *clusterRouter) getSuiteScoped(t *testing.T) *FunctionalTestBase { // TODO(stephan, #10580): remove this workaround once the proper cluster-pool fix lands. // Enable the worker service on suite-scoped clusters. The only current user (Versioning3) needs the system // worker for worker-deployment APIs. - suiteCluster.cluster = p.clusterCreatorFn(t, nil, true, []TestClusterOption{withWorkerService(true)}) + suiteCluster.cluster = p.createClusterFn(t, nil, true, []TestClusterOption{withWorkerService(true)}) }) suiteCluster.cluster.SetT(t) return suiteCluster.cluster } -func (p *clusterRouter) getDedicated(t *testing.T, needWorkerService bool, dynamicConfig map[dynamicconfig.Key]any, clusterOpts []TestClusterOption) *FunctionalTestBase { - if len(dynamicConfig) > 0 || len(clusterOpts) > 0 || needWorkerService { +func (p *clusterRouter) getDedicated(t *testing.T, req clusterRequest) *FunctionalTestBase { + if req.mustBeFresh() || req.needWorkerService { // Always create a new cluster in the following cases: // - custom configs are set, since they can't be shared across tests // - worker service is needed, since goroutines (system workers, matching partition managers) don't clean up between // tests and would accumulate in a long-lived pooled cluster. p.dedicated.reserveSlot(t) - cluster := p.clusterCreatorFn(t, dynamicConfig, false, append(clusterOpts, withWorkerService(needWorkerService))) + cluster := p.createClusterFn(t, req.dynamicConfig, false, append(req.clusterOpts, withWorkerService(req.needWorkerService))) // Register cleanup to tear down the cluster when the test completes. t.Cleanup(func() { @@ -280,11 +294,11 @@ func (p *clusterRouter) getDedicated(t *testing.T, needWorkerService bool, dynam // If no custom config or worker service is needed, reuse an existing cluster. return p.dedicated.get(t, func() *FunctionalTestBase { - return p.clusterCreatorFn(t, nil, false, nil) + return p.createClusterFn(t, nil, false, nil) }) } -func (p *clusterRouter) createCluster(t *testing.T, dynamicConfig map[dynamicconfig.Key]any, shared bool, clusterOpts []TestClusterOption) *FunctionalTestBase { +func createCluster(t *testing.T, dynamicConfig map[dynamicconfig.Key]any, shared bool, clusterOpts []TestClusterOption) *FunctionalTestBase { tbase := &FunctionalTestBase{} tbase.SetT(t) diff --git a/tests/testcore/test_cluster_pool_test.go b/tests/testcore/test_cluster_pool_test.go index a6f23cea207..d500eb3a29d 100644 --- a/tests/testcore/test_cluster_pool_test.go +++ b/tests/testcore/test_cluster_pool_test.go @@ -24,7 +24,7 @@ func newMockClusterPoolRouter(t *testing.T) (*clusterRouter, *[]clusterCreateFnA var calls []clusterCreateFnArgs mockClusterCreationFn := func(t *testing.T, dc map[dynamicconfig.Key]any, shared bool, opts []TestClusterOption) *FunctionalTestBase { - // Keep the worker service off unless explicitly enabled via WithWorkerService (this mirrors the actual createClusterFn implementation). + // Keep the worker service off unless explicitly enabled via WithWorkerService (this mirrors the actual createCluster implementation). baseOpts := []TestClusterOption{withWorkerService(false)} params := ApplyTestClusterOptions(append(baseOpts, opts...)) calls = append(calls, clusterCreateFnArgs{shared: shared, workerEnabled: params.EnableWorkerService}) @@ -35,7 +35,7 @@ func newMockClusterPoolRouter(t *testing.T) (*clusterRouter, *[]clusterCreateFnA shared: newClusterPool(2, false, 0), sharedWithWorker: newClusterPool(1, false, 0), // small pool to force reuse of the same cluster for multiple tests dedicated: newClusterPool(2, true, 0), - clusterCreatorFn: mockClusterCreationFn, + createClusterFn: mockClusterCreationFn, } return r, &calls } @@ -63,13 +63,13 @@ func TestClusterPool_MaxLeasesRecyclesOnNextAcquire(t *testing.T) { p := newClusterPool(1, false, 1) slot := p.allSlots[0] var created int - createCluster := func() *FunctionalTestBase { + createClusterFn := func() *FunctionalTestBase { created++ return &FunctionalTestBase{} } t.Run("uses cluster", func(t *testing.T) { - cluster := p.get(t, createCluster) + cluster := p.get(t, createClusterFn) require.Same(t, cluster, slot.cluster) require.Equal(t, 1, slot.activeLeases) require.Equal(t, 1, slot.leaseCount) @@ -83,7 +83,7 @@ func TestClusterPool_MaxLeasesRecyclesOnNextAcquire(t *testing.T) { // Lease-limit recycling happens on the next acquire after the prior lease releases. t.Run("recreates cluster", func(t *testing.T) { - cluster := p.get(t, createCluster) + cluster := p.get(t, createClusterFn) require.Same(t, cluster, slot.cluster) require.NotSame(t, firstCluster, cluster) require.Equal(t, 2, created) @@ -95,13 +95,13 @@ func TestClusterPool_MaxLeasesWaitsForActiveLeases(t *testing.T) { p := newClusterPool(1, false, 1) slot := p.allSlots[0] var created int - createCluster := func() *FunctionalTestBase { + createClusterFn := func() *FunctionalTestBase { created++ return &FunctionalTestBase{} } - activeCluster := p.get(t, createCluster) - concurrentCluster := p.get(t, createCluster) + activeCluster := p.get(t, createClusterFn) + concurrentCluster := p.get(t, createClusterFn) // Concurrent leases share the current cluster even after usage crosses maxLeases. require.Same(t, activeCluster, concurrentCluster) @@ -116,18 +116,18 @@ func TestClusterPool_PoisonedActiveClusterSwapsWithoutRecycling(t *testing.T) { p := newClusterPool(1, false, 1) slot := p.allSlots[0] var created int - createCluster := func() *FunctionalTestBase { + createClusterFn := func() *FunctionalTestBase { created++ return &FunctionalTestBase{ t: &sharedClusterT{name: t.Name()}, } } - poisonedCluster := p.get(t, createCluster) + poisonedCluster := p.get(t, createClusterFn) poisonedCluster.t.failed.Store(true) // Poison swaps the slot immediately, but the old active lease still has to release. - replacementCluster := p.get(t, createCluster) + replacementCluster := p.get(t, createClusterFn) require.NotSame(t, poisonedCluster, replacementCluster) require.Same(t, replacementCluster, slot.cluster) @@ -143,7 +143,7 @@ func TestClusterPool_SharedWorkerServiceRouting(t *testing.T) { var first, second *FunctionalTestBase t.Run("first call creates cluster", func(t *testing.T) { - first = router.get(t, false /* dedicated */, true /* workerService */, nil, nil) + first = router.get(t, clusterRequest{needWorkerService: true}) }) require.Equal(t, 1, router.sharedWithWorker.allSlots[0].leaseCount, "sharedWithWorker should have one lease") require.Equal(t, 0, router.shared.allSlots[0].leaseCount, "shared pool should be unused") @@ -151,7 +151,7 @@ func TestClusterPool_SharedWorkerServiceRouting(t *testing.T) { require.Equal(t, clusterCreateFnArgs{shared: true, workerEnabled: true}, (*calls)[0]) t.Run("second call reuses cluster", func(t *testing.T) { - second = router.get(t, false /* dedicated */, true /* workerService */, nil, nil) + second = router.get(t, clusterRequest{needWorkerService: true}) }) require.Same(t, first, second, "second call should return the same cluster, not allocate a new one") require.Len(t, *calls, 1, "no new cluster should have been created") @@ -163,7 +163,7 @@ func TestClusterPool_SharedWorkerServiceRouting(t *testing.T) { t.Run("worker service with dedicated uses dedicated pool", func(t *testing.T) { router, calls := newMockClusterPoolRouter(t) - router.get(t, true /* dedicated */, true /* workerService */, nil, nil) + router.get(t, clusterRequest{dedicated: true, needWorkerService: true}) require.Equal(t, 0, router.sharedWithWorker.allSlots[0].leaseCount, "sharedWithWorker pool should be unused") require.Len(t, *calls, 1) require.Equal(t, clusterCreateFnArgs{shared: false, workerEnabled: true}, (*calls)[0]) @@ -172,7 +172,7 @@ func TestClusterPool_SharedWorkerServiceRouting(t *testing.T) { t.Run("no worker service uses plain shared pool", func(t *testing.T) { router, calls := newMockClusterPoolRouter(t) - router.get(t, false /* dedicated */, false /* workerService */, nil, nil) + router.get(t, clusterRequest{}) require.Equal(t, 1, router.shared.allSlots[0].leaseCount, "shared pool should have one lease") require.Equal(t, 0, router.sharedWithWorker.allSlots[0].leaseCount, "sharedWithWorker pool should be unused") require.Len(t, *calls, 1) diff --git a/tests/testcore/test_env.go b/tests/testcore/test_env.go index b17e96e5925..5e358d801e9 100644 --- a/tests/testcore/test_env.go +++ b/tests/testcore/test_env.go @@ -114,17 +114,6 @@ func WithDedicatedCluster() TestOption { } } -// ForLeakChecker marks the test as intentionally using a dedicated cluster for -// leak measurement purposes, satisfying the [WithDedicatedCluster] guard without -// requiring an actual cluster-global side effect. -// -// NOTE: This option should NEVER be use anywhere outside of the object leak checker. -func ForLeakChecker() TestOption { - return func(o *testOptions) { - o.dedicatedReason = "leak checker" - } -} - // WithDisableTestloggerFailure disables the test logger's behavior of failing // the test when an error log matches a registered expectation (e.g. soft-assert // errors tagged with tag.FailedAssertion). Use for tests that intentionally @@ -168,7 +157,8 @@ func WithFxOptions(serviceName primitives.ServiceName, opts ...fx.Option) TestOp // If this is not coupled with any option that requires a dedicated cluster, // we will create a cluster that will be recycled by other tests that also // need the system worker service. -func WithWorkerService() TestOption { +// The string parameter serves as documentation regarding the usage of the worker service. +func WithWorkerService(_ string) TestOption { return func(o *testOptions) { o.needWorkerService = true } @@ -291,7 +281,12 @@ func NewEnv(t *testing.T, opts ...TestOption) *TestEnv { } // Obtain the test cluster from the router. - base := testClusterRouter.get(t, options.dedicatedCluster, options.needWorkerService, startupConfig, options.clusterOptions) + base := testClusterRouter.get(t, clusterRequest{ + dedicated: options.dedicatedCluster, + needWorkerService: options.needWorkerService, + dynamicConfig: startupConfig, + clusterOpts: options.clusterOptions, + }) cluster := base.GetTestCluster() // Create a dedicated namespace for the test to help with test isolation. diff --git a/tests/versioning_3_test.go b/tests/versioning_3_test.go index d28a75db3a2..ec0c966aed9 100644 --- a/tests/versioning_3_test.go +++ b/tests/versioning_3_test.go @@ -6712,7 +6712,7 @@ func (s *Versioning3Suite) TestPinnedCaN_FailedTransientNotificationRefiresDespi // made current again, then reset-by-build-ID resets the workflow before v2 usage // so the reset run resumes on v1. func (s *Versioning3Suite) TestPinnedCaN_ResetByBuildIDAfterRollback() { - env := s.setupEnv(testcore.WithWorkerService()) + env := s.setupEnv(testcore.WithWorkerService("batch operations")) tv := env.Tv() tv1 := tv.WithBuildIDNumber(1) diff --git a/tests/worker_deployment_version_test.go b/tests/worker_deployment_version_test.go index 4b8da3874f5..aa3e7874503 100644 --- a/tests/worker_deployment_version_test.go +++ b/tests/worker_deployment_version_test.go @@ -1875,7 +1875,7 @@ func (s *DeploymentVersionSuite) setAndCheckOverrideWithExpectedOutput(env *test func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetPinned_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. - testcore.WithWorkerService(), + testcore.WithWorkerService("worker-deployment version membership cache test"), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) @@ -2854,7 +2854,7 @@ func (s *DeploymentVersionSuite) makeAutoUpgradeOverride() *workflowpb.Versionin func (s *DeploymentVersionSuite) TestStartWorkflowExecution_WithPinnedOverride_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. - testcore.WithWorkerService(), + testcore.WithWorkerService("worker-deployment version membership cache test"), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) @@ -2906,7 +2906,7 @@ func (s *DeploymentVersionSuite) TestStartWorkflowExecution_WithUnpinnedOverride func (s *DeploymentVersionSuite) TestSignalWithStartWorkflowExecution_WithPinnedOverride_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. - testcore.WithWorkerService(), + testcore.WithWorkerService("worker-deployment version membership cache test"), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) diff --git a/tests/worker_registry_test.go b/tests/worker_registry_test.go index 968b9841e5d..6dab78419f6 100644 --- a/tests/worker_registry_test.go +++ b/tests/worker_registry_test.go @@ -27,7 +27,7 @@ func TestWorkerRegistryTestSuite(t *testing.T) { func (s *WorkerRegistryTestSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv { baseOpts := []testcore.TestOption{ - testcore.WithWorkerService(), + testcore.WithWorkerService("worker registry tests use system worker for heartbeat processing"), testcore.WithDynamicConfig(dynamicconfig.WorkerHeartbeatsEnabled, true), } return testcore.NewEnv(s.T(), append(baseOpts, opts...)...) diff --git a/tests/workflow_alias_search_attribute_test.go b/tests/workflow_alias_search_attribute_test.go index 86b4c781978..f7ccb58535a 100644 --- a/tests/workflow_alias_search_attribute_test.go +++ b/tests/workflow_alias_search_attribute_test.go @@ -32,7 +32,7 @@ func TestWorkflowAliasSearchAttributeTestSuite(t *testing.T) { func (s *WorkflowAliasSearchAttributeTestSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv { opts = append([]testcore.TestOption{ - testcore.WithWorkerService(), + testcore.WithWorkerService("workflow alias search attribute tests use worker deployment system workflows"), // Keep deployment versions short because worker-deployment system workflow IDs must fit into 255 characters. testcore.WithTestVars(func(tv *testvars.TestVars) *testvars.TestVars { diff --git a/tests/workflow_reset_test.go b/tests/workflow_reset_test.go index f18637d917f..d189f7826c1 100644 --- a/tests/workflow_reset_test.go +++ b/tests/workflow_reset_test.go @@ -215,7 +215,7 @@ func (s *WorkflowResetSuite) TestOriginalExecutionRunId() { // Test that the workflow options are updated when the workflow is reset. func (s *WorkflowResetSuite) TestResetWorkflowWithOptionsUpdate() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService("worker-deployment version registration")) deploymentName := "testing" buildID := "v.123" @@ -293,7 +293,7 @@ func (s *WorkflowResetSuite) TestResetWorkflowWithOptionsUpdate() { // Test batch reset operation with version update as post reset operation func (s *WorkflowResetSuite) TestBatchResetWithOptionsUpdate() { - env := testcore.NewEnv(s.T(), testcore.WithWorkerService()) + env := testcore.NewEnv(s.T(), testcore.WithWorkerService("batch operations")) deploymentName := "batch-testing" buildID := "v.456" From f26adfbd951f7c97bcdbbe62dc2417446e7bcfdf Mon Sep 17 00:00:00 2001 From: long-nt-tran Date: Tue, 7 Jul 2026 14:37:46 -0400 Subject: [PATCH 3/5] Fix test using onebox unnecessarily, revert onebox silent return --- tests/testcore/onebox.go | 4 +--- tests/testcore/test_cluster_pool_test.go | 20 ++++++++++---------- 2 files changed, 11 insertions(+), 13 deletions(-) diff --git a/tests/testcore/onebox.go b/tests/testcore/onebox.go index 456c72a0e66..26ed8ef7816 100644 --- a/tests/testcore/onebox.go +++ b/tests/testcore/onebox.go @@ -231,9 +231,7 @@ func newTemporal(t *testing.T, params *TemporalParams) *TemporalImpl { } // Override Nexus callback URL. This is parameterized on the frontend's HTTP address, // so it can't be overriden in the loop above. - if len(params.HostsByProtocolByService) > 0 { - impl.setNexusCallbackURL() - } + impl.setNexusCallbackURL() // Per-test overrides: cleaned up when the creating test finishes. for k, v := range params.DynamicConfigOverrides { impl.overrideDynamicConfigForTest(t, k, v) diff --git a/tests/testcore/test_cluster_pool_test.go b/tests/testcore/test_cluster_pool_test.go index d500eb3a29d..0912258a0e7 100644 --- a/tests/testcore/test_cluster_pool_test.go +++ b/tests/testcore/test_cluster_pool_test.go @@ -4,7 +4,6 @@ import ( "testing" "github.com/stretchr/testify/require" - "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/dynamicconfig" ) @@ -41,18 +40,19 @@ func newMockClusterPoolRouter(t *testing.T) (*clusterRouter, *[]clusterCreateFnA } func TestClusterPool_GlobalOverridesSurviveTestCleanup(t *testing.T) { - var dcClient *dynamicconfig.MemoryClient + dc := dynamicconfig.NewMemoryClient() - t.Run("create", func(t *testing.T) { - impl := newTemporal(t, &TemporalParams{ - ClusterMetadataConfig: &cluster.Config{}, - }) - dcClient = impl.dcClient + t.Run("apply", func(t *testing.T) { + // Apply global defaults the same way newTemporal does: via PartialOverrideValue + // without registering a t.Cleanup, so they persist beyond the test's lifetime. + for k, v := range defaultDynamicConfigOverrides { + dc.PartialOverrideValue(k, v) + } }) - // "create" subtest finished — its t.Cleanup has run - // but we expect global dynamic config overrides to still be in place. + // "apply" subtest finished - its t.Cleanup has run. + // Global overrides must still be in place. for k, v := range defaultDynamicConfigOverrides { - got := dcClient.GetValue(k) + got := dc.GetValue(k) require.NotEmpty(t, got, "key %s missing after cleanup", k) require.Equal(t, v, got[0].Value, "key %s wrong after cleanup", k) } From feecf3a838f131db613f40be5157d034ef260e0a Mon Sep 17 00:00:00 2001 From: long-nt-tran Date: Tue, 7 Jul 2026 15:35:14 -0400 Subject: [PATCH 4/5] Use global metric capture in leak test instead of MTLS --- tests/leakcheck/leak_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/leakcheck/leak_test.go b/tests/leakcheck/leak_test.go index 6082421678e..6ad3baf41cf 100644 --- a/tests/leakcheck/leak_test.go +++ b/tests/leakcheck/leak_test.go @@ -151,9 +151,12 @@ func buildRunTeardownCluster(t *testing.T, leakCheck *objectleak.ObjectLeakCheck env := testcore.NewEnv(t, testcore.WithDedicatedCluster(), testcore.WithWorkerService("leak checker needs worker service to exercise full server path"), - testcore.WithMTLS(), // Using a global option (MTLS) to satisfy the dedicated cluster guard ) + // Doing a no-op global metric capture to satisfy the dedicated cluster guard, otherwise + // we'll trigger a cluster misuse fatal. + env.StartGlobalMetricCapture() + env.SdkWorker().RegisterWorkflow(smokeWorkflow) run, err := env.SdkClient().ExecuteWorkflow( context.Background(), From c9763fb2af43585eb6996414d57e98231fb31976 Mon Sep 17 00:00:00 2001 From: long-nt-tran Date: Tue, 7 Jul 2026 15:53:41 -0400 Subject: [PATCH 5/5] Revert some unnecessary comment changes --- tests/schedule_migration_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/schedule_migration_test.go b/tests/schedule_migration_test.go index efc1ec7ef0c..4d69d2bb0d6 100644 --- a/tests/schedule_migration_test.go +++ b/tests/schedule_migration_test.go @@ -1322,7 +1322,7 @@ func TestScheduleMigrationV1ToV2NoDuplicateRecentActions(t *testing.T) { // a CHASM sentinel (which would block the migration activity). env := testcore.NewEnv( t, - testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), + testcore.WithWorkerService("V1 scheduler"), testcore.WithSdkWorker(), ) @@ -1470,7 +1470,7 @@ func TestScheduleMigrationDeferredWithRunningWorkflow(t *testing.T) { // (workflow-backed) schedule rather than a CHASM sentinel. env := testcore.NewEnv( t, - testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), + testcore.WithWorkerService("V1 scheduler"), testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigrationWithRunningWorkflows, false), ) @@ -1973,7 +1973,7 @@ func (s *ScheduleMigrationTestSuite) TestPatchScheduleContextMetadata() { func TestScheduleMigration_StaleRunningDoesNotSkipPending(t *testing.T) { env := testcore.NewEnv( t, - testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), + testcore.WithWorkerService("scheduler operations"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -2247,7 +2247,7 @@ func TestScheduleMigration_NoRunningWorkflows_GeneratorStarts(t *testing.T) { env := testcore.NewEnv( t, - testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), + testcore.WithWorkerService("scheduler operations"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) env.OverrideDynamicConfig(chasmscheduler.CurrentTweakables, tweakables)