Skip to content

[testcore] Shared cluster pool for clusters that require WorkerService #10905

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
long-nt-tran wants to merge 5 commits into
base: main
Choose a base branch
from
Open

[testcore] Shared cluster pool for clusters that require WorkerService #10905

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
bdd1214
[testcore] Shared cluster pool for clusters that require WorkerService
long-nt-tran Jul 1, 2026
5a001a3
Address PR feedback
long-nt-tran Jul 7, 2026
f26adfb
Fix test using onebox unnecessarily, revert onebox silent return
long-nt-tran Jul 7, 2026
feecf3a
Use global metric capture in leak test instead of MTLS
long-nt-tran Jul 7, 2026
c9763fb
Revert some unnecessary comment changes
long-nt-tran Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions docs/development/testing.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
Expand Down
9 changes: 4 additions & 5 deletions tests/activity_api_batch_unpause_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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{
Expand Down Expand Up @@ -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()

Expand Down
7 changes: 6 additions & 1 deletion tests/leakcheck/leak_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Comment on lines +156 to +158

@long-nt-tran long-nt-tran Jul 8, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@stephanos using the WithMTLS() option causes more leaks, i spent a few tokens:

WithMTLS() is the cause. When the cluster tears down with mTLS enabled, the TLS connection becomes invalid. The sharedNamespaceWorker heartbeat RPC fails with a TLS error instead of a clean connection reset. TLS errors trigger a different
  retry classification in the grpc middleware — the goroutine parks in waitRetryBackoff for a longer, more deterministic backoff window. By the time goleak samples the stacks, the goroutine is reliably mid-retry.

  Without mTLS, the connection closes more cleanly — the goroutine either finishes the heartbeat or quickly returns to the idle (*sharedNamespaceWorker).run select loop, which IS in the ignore list. That race was nearly always won before.

  waitRetryBackoff still needs to be in the ignore list regardless — it's a transient state of the already-documented "SDK worker goroutines not fully stopped" leak.

I didn't want to add more ignores, doing this for now. But can also see a case for adding MTLS + ignore options, and then work to address those in a future PR. LMK if that's preferable

@stephanos stephanos Jul 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

StartGlobalMetricCapture is good for now 👍 let's leave mTLS cleanup for after we cleaned up the current batch


env.SdkWorker().RegisterWorkflow(smokeWorkflow)
run, err := env.SdkClient().ExecuteWorkflow(
Expand Down
2 changes: 1 addition & 1 deletion tests/namespace_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
5 changes: 3 additions & 2 deletions tests/poller_scaling_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

@stephanos stephanos Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we delete UseSuiteScopedCluster now actually?

@long-nt-tran long-nt-tran Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we still need it until we split versioning_3 suite

@stephanos stephanos Jul 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, good call!

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),
Expand Down
15 changes: 13 additions & 2 deletions tests/schedule_migration_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
)

Expand Down Expand Up @@ -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),
)
Expand Down Expand Up @@ -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),
)

Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
)

Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down
6 changes: 3 additions & 3 deletions tests/task_queue_stats_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)
})
}
}
Expand Down
Loading
Loading