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_unpause_test.go b/tests/activity_api_batch_unpause_test.go index 0b2f8c0b2a9..e954f833421 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("batch unpause activity operations require system worker batcher")) 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("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{ @@ -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("batch terminate namespace isolation test uses system worker batcher")) ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() diff --git a/tests/leakcheck/leak_test.go b/tests/leakcheck/leak_test.go index 169d0cec2f1..6ad3baf41cf 100644 --- a/tests/leakcheck/leak_test.go +++ b/tests/leakcheck/leak_test.go @@ -150,7 +150,12 @@ 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("leak checker needs worker service to exercise full server path"), + ) + + // 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( diff --git a/tests/namespace_test.go b/tests/namespace_test.go index 3ce173e87cf..d0fe884369a 100644 --- a/tests/namespace_test.go +++ b/tests/namespace_test.go @@ -31,7 +31,7 @@ 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 { diff --git a/tests/poller_scaling_test.go b/tests/poller_scaling_test.go index bde1c0dc9c1..fc5acb00ad3 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("poller scaling test uses worker deployment system workflows"), + // Force one partition so we can reliably see the backlog testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1), testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1), diff --git a/tests/schedule_migration_test.go b/tests/schedule_migration_test.go index f0efc1b4985..4d69d2bb0d6 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("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -291,6 +291,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1BlockedBySentine func (s *ScheduleMigrationTestSuite) TestScheduleMigrationDynamicConfig() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), 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("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -536,6 +538,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV1ToV2() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1() { env := testcore.NewEnv( s.T(), + 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), @@ -742,6 +745,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1Idempotent() { env := testcore.NewEnv( s.T(), + 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), @@ -812,6 +816,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1Idempotent() { func (s *ScheduleMigrationTestSuite) TestCHASMScheduleDescribeAfterDisablingCreationAndMigration() { env := testcore.NewEnv( s.T(), + 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), @@ -913,6 +918,7 @@ func (s *ScheduleMigrationTestSuite) TestCHASMScheduleDescribeAfterDisablingCrea func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1RoutingFallback() { env := testcore.NewEnv( s.T(), + 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), @@ -1048,6 +1054,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV2ToV1RoutingFallback( func (s *ScheduleMigrationTestSuite) TestScheduleUpdateAfterDelete() { env := testcore.NewEnv( s.T(), + 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), @@ -1158,6 +1165,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleUpdateAfterDelete() { func (s *ScheduleMigrationTestSuite) TestScheduleMigrationV1ToV2WithClosedV2() { env := testcore.NewEnv( s.T(), + testcore.WithWorkerService("schedule migration tests require system worker for v1 schedule management"), testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true), ) @@ -1598,6 +1606,7 @@ func TestScheduleMigrationDeferredWithRunningWorkflow(t *testing.T) { func (s *ScheduleMigrationTestSuite) TestDeleteScheduleContextMetadata() { env := testcore.NewEnv( s.T(), + 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), @@ -1801,6 +1810,7 @@ func (s *ScheduleMigrationTestSuite) TestDeleteScheduleContextMetadata() { func (s *ScheduleMigrationTestSuite) TestPatchScheduleContextMetadata() { env := testcore.NewEnv( s.T(), + 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), @@ -2111,6 +2121,7 @@ func (s *ScheduleMigrationTestSuite) TestScheduleMigrationRolloutPercent() { t := s.T() env := testcore.NewEnv( t, + 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), diff --git a/tests/task_queue_stats_test.go b/tests/task_queue_stats_test.go index 22a2fde478e..1f37391952d 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("task queue stats tests use worker deployment system workflows"), 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/test_cluster_pool.go b/tests/testcore/test_cluster_pool.go index 8211c7fe403..244e20d6233 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,34 @@ 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") + + sharedWorkerSize := max(1, runtime.GOMAXPROCS(0)/2) + 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.createClusterFn = createCluster + testClusterRouter = r } // clusterPool manages a fixed number of test [clusterPoolSlot]s. @@ -156,11 +163,31 @@ 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 - dedicated *clusterPool - suiteScoped sync.Map + shared *clusterPool + sharedWithWorker *clusterPool + dedicated *clusterPool + suiteScoped sync.Map + + // 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. + createClusterFn func(*testing.T, map[dynamicconfig.Key]any, bool, []TestClusterOption) *FunctionalTestBase } // suiteScopedCluster owns one lazily created legacy suite cluster. @@ -196,24 +223,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, req clusterRequest) (tb *FunctionalTestBase) { defer func() { if tb != nil { tb.RegisterTest(t) } }() - if dedicated || len(dynamicConfig) > 0 || len(clusterOpts) > 0 { - return p.getDedicated(t, dynamicConfig, clusterOpts) + if req.needsDedicated() { + return p.getDedicated(t, req) } if cluster := p.getSuiteScoped(t); cluster != nil { return cluster } - return p.getShared(t) + return p.getShared(t, req.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.createClusterFn(t, nil, true, []TestClusterOption{withWorkerService(true)}) + }) + } return p.shared.get(t, func() *FunctionalTestBase { - return p.createCluster(t, nil, true, nil) + return p.createClusterFn(t, nil, true, nil) }) } @@ -235,17 +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.createCluster(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, 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, 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.createCluster(t, dynamicConfig, false, clusterOpts) + 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() { @@ -257,13 +292,13 @@ 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.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 2243ed81385..0912258a0e7 100644 --- a/tests/testcore/test_cluster_pool_test.go +++ b/tests/testcore/test_cluster_pool_test.go @@ -4,23 +4,55 @@ import ( "testing" "github.com/stretchr/testify/require" - "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/dynamicconfig" ) -func TestGlobalOverridesSurviveTestCleanup(t *testing.T) { - var dcClient *dynamicconfig.MemoryClient +// 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 +} - t.Run("create", func(t *testing.T) { - impl := newTemporal(t, &TemporalParams{ - ClusterMetadataConfig: &cluster.Config{}, - }) - dcClient = impl.dcClient +// 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 createCluster 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), + createClusterFn: mockClusterCreationFn, + } + return r, &calls +} + +func TestClusterPool_GlobalOverridesSurviveTestCleanup(t *testing.T) { + dc := dynamicconfig.NewMemoryClient() + + 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) } @@ -31,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) @@ -51,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) @@ -63,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) @@ -84,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) @@ -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, 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") + 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, 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") + 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, 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]) + }) + + t.Run("no worker service uses plain shared pool", func(t *testing.T) { + router, calls := newMockClusterPoolRouter(t) + + 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) + 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..5e358d801e9 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 @@ -152,12 +153,14 @@ 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. +// The string parameter serves as documentation regarding the usage of the worker service. +func WithWorkerService(_ string) 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 +281,12 @@ 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, 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/worker_registry_test.go b/tests/worker_registry_test.go index 5f74681d4e0..6dab78419f6 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("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 5f11a2d7253..f7ccb58535a 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("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 { return tv.WithDeploymentSeries("alias-sa").WithBuildID("v1")