From 6ec33f529de96e5dc679addc78ca61152ef931f4 Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Thu, 3 Sep 2026 16:12:47 -0700 Subject: [PATCH 01/11] Run kitchen sink SDK tests in isolated namespaces --- devserver/devserver.go | 23 ++----- internal/workertest/env.go | 67 ++++++++++++++----- internal/workertest/serverpool.go | 92 +++++++++++++++++++++++++++ internal/workertest/workerpool.go | 2 +- loadgen/kitchen_sink_executor_test.go | 17 ++++- 5 files changed, 165 insertions(+), 36 deletions(-) create mode 100644 internal/workertest/serverpool.go diff --git a/devserver/devserver.go b/devserver/devserver.go index f81f62da1..5c362c9d5 100644 --- a/devserver/devserver.go +++ b/devserver/devserver.go @@ -206,7 +206,7 @@ func Start(ctx context.Context, opts Options) (*Server, error) { } success = true - if err := s.registerNamespace(ctx, opts.Namespace); err != nil { + if err := s.RegisterNamespace(ctx, opts.Namespace); err != nil { _ = s.Stop() return nil, fmt.Errorf("devserver: register namespace %q: %w", opts.Namespace, err) } @@ -219,6 +219,11 @@ func (s *Server) FrontendHostPort() string { return s.frontend } +// RegisterNamespace registers a namespace if it does not already exist. +func (s *Server) RegisterNamespace(ctx context.Context, namespace string) error { + return registerNamespace(ctx, s.frontend, namespace) +} + // Ports returns the service ports allocated for the running server. func (s *Server) Ports() Ports { return s.ports @@ -254,19 +259,3 @@ func classifyExitErr(err error) error { } return err } - -func (s *Server) registerNamespace(ctx context.Context, namespace string) error { - ctx, cancel := context.WithCancelCause(ctx) - defer cancel(nil) - - go func() { - select { - case procErr := <-s.done: - s.done <- procErr - cancel(fmt.Errorf("process exited before namespace registration completed: %w", procErr)) - case <-ctx.Done(): - } - }() - - return registerNamespace(ctx, s.frontend, namespace) -} diff --git a/internal/workertest/env.go b/internal/workertest/env.go index df2126950..856e17dbf 100644 --- a/internal/workertest/env.go +++ b/internal/workertest/env.go @@ -37,6 +37,8 @@ type testEnvConfig struct { executorTimeout time.Duration runID string dynamicConfig map[string]any + namespace string + devServerGroup string } type TestEnvOption func(*testEnvConfig) @@ -64,6 +66,21 @@ func WithDynamicConfig(values map[string]any) TestEnvOption { } } +// WithNamespace configures the namespace used by the test environment. +func WithNamespace(namespace string) TestEnvOption { + return func(c *testEnvConfig) { + c.namespace = namespace + } +} + +// WithDevServerGroup configures the test environment to share a dev server +// with other environments created by the same test using the same group. +func WithDevServerGroup(group string) TestEnvOption { + return func(c *testEnvConfig) { + c.devServerGroup = group + } +} + type TestResult struct { ObservedLogs *observer.ObservedLogs } @@ -106,13 +123,12 @@ func (env *TestEnvironment) NexusEndpointName() string { return env.nexusEndpointName } +func (env *TestEnvironment) Namespace() string { + return env.namespace +} + func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment { - cfg := testEnvConfig{ - executorTimeout: defaultTestRunTimeout, - } - for _, opt := range opts { - opt(&cfg) - } + cfg := newTestEnvConfig(opts) serverRef, err := versions.Get("SERVER_VERSION") require.NoError(t, err) @@ -120,18 +136,31 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment testLogger := zaptest.NewLogger(t) serverLogger := testLogger.Named("devserver").Sugar() - server, err := devserver.Start(t.Context(), devserver.Options{ + serverOpts := devserver.Options{ Ref: serverRef, Namespace: testNamespace, DynamicConfigValues: cfg.dynamicConfig, Output: workerctl.NewLogWriter(serverLogger), Logger: serverLogger, - }) + } + + var server *devserver.Server + var cleanupServer func() + if cfg.devServerGroup == "" { + server, err = devserver.Start(t.Context(), serverOpts) + cleanupServer = func() { _ = server.Stop() } + } else { + server, cleanupServer, err = groupedTestDevServers.acquire(t, cfg, serverOpts) + } require.NoError(t, err, "Failed to start dev server") + t.Cleanup(cleanupServer) + if cfg.namespace != testNamespace { + require.NoError(t, server.RegisterNamespace(t.Context(), cfg.namespace), "Failed to register namespace") + } temporalClient, err := client.Dial(client.Options{ HostPort: server.FrontendHostPort(), - Namespace: testNamespace, + Namespace: cfg.namespace, }) require.NoError(t, err, "Failed to create Temporal client") @@ -159,13 +188,21 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment return env } +func newTestEnvConfig(opts []TestEnvOption) testEnvConfig { + cfg := testEnvConfig{ + executorTimeout: defaultTestRunTimeout, + namespace: testNamespace, + } + for _, opt := range opts { + opt(&cfg) + } + return cfg +} + func (env *TestEnvironment) cleanup() { if env.temporalClient != nil { env.temporalClient.Close() } - if env.devServer != nil { - env.devServer.Stop() - } env.workerPool.cleanup() } @@ -179,7 +216,7 @@ func (env *TestEnvironment) CreateNexusEndpoint(ctx context.Context, runID strin Target: &nexus.EndpointTarget{ Variant: &nexus.EndpointTarget_Worker_{ Worker: &nexus.EndpointTarget_Worker{ - Namespace: testNamespace, + Namespace: env.namespace, TaskQueue: taskQueueName, }, }, @@ -221,11 +258,11 @@ func (env *TestEnvironment) RunExecutorTest( scenarioInfo.Logger = logger.Named("executor") scenarioInfo.MetricsHandler = client.MetricsNopHandler scenarioInfo.Client = env.temporalClient - scenarioInfo.Namespace = testNamespace + scenarioInfo.Namespace = env.namespace // Resolve any capability-gated feature options against the dev server, so // integration tests exercise the same code path as the CLI. - if err := loadgen.ResolveFeatureOptions(testCtx, env.temporalClient, testNamespace, scenarioInfo.Options, logger); err != nil { + if err := loadgen.ResolveFeatureOptions(testCtx, env.temporalClient, env.namespace, scenarioInfo.Options, logger); err != nil { return TestResult{ObservedLogs: observedLogs}, err } diff --git a/internal/workertest/serverpool.go b/internal/workertest/serverpool.go new file mode 100644 index 000000000..dbf634afd --- /dev/null +++ b/internal/workertest/serverpool.go @@ -0,0 +1,92 @@ +package workertest + +import ( + "fmt" + "maps" + "reflect" + "sync" + "testing" + + "github.com/temporalio/omes/devserver" +) + +type pooledTestDevServer struct { + server *devserver.Server + dynamicConfig map[string]any + namespaces map[string]struct{} + test *testing.T + refs int +} + +type testDevServerPool struct { + sync.Mutex + servers map[string]*pooledTestDevServer +} + +var groupedTestDevServers = testDevServerPool{ + servers: make(map[string]*pooledTestDevServer), +} + +func (p *testDevServerPool) acquire( + t *testing.T, + cfg testEnvConfig, + opts devserver.Options, +) (*devserver.Server, func(), error) { + p.Lock() + defer p.Unlock() + + if pooled := p.servers[cfg.devServerGroup]; pooled != nil { + if pooled.test != t { + return nil, nil, fmt.Errorf( + "dev server group %q already belongs to another test", + cfg.devServerGroup, + ) + } + if !reflect.DeepEqual(pooled.dynamicConfig, cfg.dynamicConfig) { + return nil, nil, fmt.Errorf( + "dev server group %q already uses different dynamic config", + cfg.devServerGroup, + ) + } + if _, exists := pooled.namespaces[cfg.namespace]; exists { + return nil, nil, fmt.Errorf( + "dev server group %q already uses namespace %q", + cfg.devServerGroup, + cfg.namespace, + ) + } + pooled.namespaces[cfg.namespace] = struct{}{} + pooled.refs++ + return pooled.server, func() { p.release(cfg.devServerGroup, pooled.server) }, nil + } + + server, err := devserver.Start(t.Context(), opts) + if err != nil { + return nil, nil, err + } + p.servers[cfg.devServerGroup] = &pooledTestDevServer{ + server: server, + dynamicConfig: maps.Clone(cfg.dynamicConfig), + namespaces: map[string]struct{}{cfg.namespace: {}}, + test: t, + refs: 1, + } + return server, func() { p.release(cfg.devServerGroup, server) }, nil +} + +func (p *testDevServerPool) release(id string, server *devserver.Server) { + p.Lock() + pooled := p.servers[id] + if pooled == nil || pooled.server != server { + p.Unlock() + return + } + pooled.refs-- + if pooled.refs > 0 { + p.Unlock() + return + } + delete(p.servers, id) + p.Unlock() + _ = server.Stop() +} diff --git a/internal/workertest/workerpool.go b/internal/workertest/workerpool.go index 6967241f9..adf2a791f 100644 --- a/internal/workertest/workerpool.go +++ b/internal/workertest/workerpool.go @@ -117,7 +117,7 @@ func (w *workerPool) startWorker( }, } runner.ClientOptions.FlagSet().Set("server-address", w.env.DevServerAddress()) - runner.ClientOptions.FlagSet().Set("namespace", testNamespace) + runner.ClientOptions.FlagSet().Set("namespace", w.env.Namespace()) if errOnUnimplemented { runner.WorkerOptions.FlagSet().Set("worker-err-on-unimplemented", "true") } diff --git a/loadgen/kitchen_sink_executor_test.go b/loadgen/kitchen_sink_executor_test.go index ce044c825..21e09852c 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/loadgen/kitchen_sink_executor_test.go @@ -75,7 +75,7 @@ func TestKitchenSink(t *testing.T) { if os.Getenv("CI") != "" && onlySDK == "" { t.Skip("Skipping kitchensink test in CI without specific SDK set") } - env := SetupTestEnvironment(t, WithDynamicConfig(map[string]any{ + dynamicConfig := map[string]any{ // Enable StartNexusOperationExecution for the standalone-nexus subtests. "nexusoperation.enableStandalone": true, // Standalone Nexus system callbacks require CHASM callbacks. @@ -83,7 +83,18 @@ func TestKitchenSink(t *testing.T) { // Enable StartActivityExecution for the standalone-activity subtest. "activity.enableStandalone": true, "history.enableStandaloneActivityOperatorCommands": true, - })) + } + testEnvironments := make(map[clioptions.Language]*TestEnvironment) + for _, sdk := range sdks { + if onlySDK != "" && string(sdk) != onlySDK { + continue + } + testEnvironments[sdk] = SetupTestEnvironment(t, + WithDevServerGroup(t.Name()), + WithNamespace(fmt.Sprintf("default.%s", sdk)), + WithDynamicConfig(dynamicConfig), + ) + } // Default workflow execution timeout for tests defaultWorkflowTimeout := 30 * time.Second @@ -1142,7 +1153,7 @@ func TestKitchenSink(t *testing.T) { } t.Run(string(sdk), func(t *testing.T) { t.Parallel() - testForSDK(t, tc, sdk, env, defaultWorkflowTimeout) + testForSDK(t, tc, sdk, testEnvironments[sdk], defaultWorkflowTimeout) }) } }) From 375f0d65083b62dd5cec98e7d0d44562a8baaaa1 Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Thu, 3 Sep 2026 16:16:33 -0700 Subject: [PATCH 02/11] Update devserver.go --- devserver/devserver.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/devserver/devserver.go b/devserver/devserver.go index 5c362c9d5..7648cf001 100644 --- a/devserver/devserver.go +++ b/devserver/devserver.go @@ -221,6 +221,18 @@ func (s *Server) FrontendHostPort() string { // RegisterNamespace registers a namespace if it does not already exist. func (s *Server) RegisterNamespace(ctx context.Context, namespace string) error { + ctx, cancel := context.WithCancelCause(ctx) + defer cancel(nil) + + go func() { + select { + case procErr := <-s.done: + s.done <- procErr + cancel(fmt.Errorf("process exited before namespace registration completed: %w", procErr)) + case <-ctx.Done(): + } + }() + return registerNamespace(ctx, s.frontend, namespace) } From 711cf934a0b2f241d5b8bd2eb96786a1222faf93 Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Thu, 3 Sep 2026 16:20:52 -0700 Subject: [PATCH 03/11] Update env.go --- internal/workertest/env.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/workertest/env.go b/internal/workertest/env.go index 856e17dbf..b35c909b3 100644 --- a/internal/workertest/env.go +++ b/internal/workertest/env.go @@ -26,7 +26,7 @@ import ( ) const ( - testNamespace = "default" + defaultTestNamespace = "default" defaultTestRunTimeout = 30 * time.Second workerBuildTimeout = 1 * time.Minute workerShutdownTimeout = 5 * time.Second @@ -138,7 +138,7 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment serverLogger := testLogger.Named("devserver").Sugar() serverOpts := devserver.Options{ Ref: serverRef, - Namespace: testNamespace, + Namespace: defaultTestNamespace, DynamicConfigValues: cfg.dynamicConfig, Output: workerctl.NewLogWriter(serverLogger), Logger: serverLogger, @@ -154,7 +154,7 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment } require.NoError(t, err, "Failed to start dev server") t.Cleanup(cleanupServer) - if cfg.namespace != testNamespace { + if cfg.namespace != defaultTestNamespace { require.NoError(t, server.RegisterNamespace(t.Context(), cfg.namespace), "Failed to register namespace") } @@ -191,7 +191,7 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment func newTestEnvConfig(opts []TestEnvOption) testEnvConfig { cfg := testEnvConfig{ executorTimeout: defaultTestRunTimeout, - namespace: testNamespace, + namespace: defaultTestNamespace, } for _, opt := range opts { opt(&cfg) From cc5ba41770724caec33db057cabbe6c5e8459d53 Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Thu, 3 Sep 2026 16:23:41 -0700 Subject: [PATCH 04/11] Simplify test dev server grouping --- internal/workertest/env.go | 37 ++++++------- internal/workertest/serverpool.go | 80 ++++++++++++--------------- loadgen/kitchen_sink_executor_test.go | 11 ++-- 3 files changed, 55 insertions(+), 73 deletions(-) diff --git a/internal/workertest/env.go b/internal/workertest/env.go index b35c909b3..6c00883ec 100644 --- a/internal/workertest/env.go +++ b/internal/workertest/env.go @@ -128,7 +128,13 @@ func (env *TestEnvironment) Namespace() string { } func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment { - cfg := newTestEnvConfig(opts) + cfg := testEnvConfig{ + executorTimeout: defaultTestRunTimeout, + namespace: defaultTestNamespace, + } + for _, opt := range opts { + opt(&cfg) + } serverRef, err := versions.Get("SERVER_VERSION") require.NoError(t, err) @@ -138,25 +144,27 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment serverLogger := testLogger.Named("devserver").Sugar() serverOpts := devserver.Options{ Ref: serverRef, - Namespace: defaultTestNamespace, + Namespace: cfg.namespace, DynamicConfigValues: cfg.dynamicConfig, Output: workerctl.NewLogWriter(serverLogger), Logger: serverLogger, } var server *devserver.Server - var cleanupServer func() if cfg.devServerGroup == "" { server, err = devserver.Start(t.Context(), serverOpts) - cleanupServer = func() { _ = server.Stop() } + if err == nil { + t.Cleanup(func() { _ = server.Stop() }) + } } else { - server, cleanupServer, err = groupedTestDevServers.acquire(t, cfg, serverOpts) + server, err = groupedTestDevServers.acquire( + t, + cfg.devServerGroup, + cfg.namespace, + serverOpts, + ) } require.NoError(t, err, "Failed to start dev server") - t.Cleanup(cleanupServer) - if cfg.namespace != defaultTestNamespace { - require.NoError(t, server.RegisterNamespace(t.Context(), cfg.namespace), "Failed to register namespace") - } temporalClient, err := client.Dial(client.Options{ HostPort: server.FrontendHostPort(), @@ -188,17 +196,6 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment return env } -func newTestEnvConfig(opts []TestEnvOption) testEnvConfig { - cfg := testEnvConfig{ - executorTimeout: defaultTestRunTimeout, - namespace: defaultTestNamespace, - } - for _, opt := range opts { - opt(&cfg) - } - return cfg -} - func (env *TestEnvironment) cleanup() { if env.temporalClient != nil { env.temporalClient.Close() diff --git a/internal/workertest/serverpool.go b/internal/workertest/serverpool.go index dbf634afd..81e64d308 100644 --- a/internal/workertest/serverpool.go +++ b/internal/workertest/serverpool.go @@ -14,79 +14,67 @@ type pooledTestDevServer struct { server *devserver.Server dynamicConfig map[string]any namespaces map[string]struct{} - test *testing.T - refs int +} + +type testDevServerGroupKey struct { + test *testing.T + group string } type testDevServerPool struct { sync.Mutex - servers map[string]*pooledTestDevServer + servers map[testDevServerGroupKey]*pooledTestDevServer } var groupedTestDevServers = testDevServerPool{ - servers: make(map[string]*pooledTestDevServer), + servers: make(map[testDevServerGroupKey]*pooledTestDevServer), } func (p *testDevServerPool) acquire( t *testing.T, - cfg testEnvConfig, + group string, + namespace string, opts devserver.Options, -) (*devserver.Server, func(), error) { +) (*devserver.Server, error) { p.Lock() defer p.Unlock() - if pooled := p.servers[cfg.devServerGroup]; pooled != nil { - if pooled.test != t { - return nil, nil, fmt.Errorf( - "dev server group %q already belongs to another test", - cfg.devServerGroup, - ) - } - if !reflect.DeepEqual(pooled.dynamicConfig, cfg.dynamicConfig) { - return nil, nil, fmt.Errorf( + key := testDevServerGroupKey{test: t, group: group} + if pooled := p.servers[key]; pooled != nil { + if !maps.EqualFunc(pooled.dynamicConfig, opts.DynamicConfigValues, reflect.DeepEqual) { + return nil, fmt.Errorf( "dev server group %q already uses different dynamic config", - cfg.devServerGroup, + group, ) } - if _, exists := pooled.namespaces[cfg.namespace]; exists { - return nil, nil, fmt.Errorf( + if _, exists := pooled.namespaces[namespace]; exists { + return nil, fmt.Errorf( "dev server group %q already uses namespace %q", - cfg.devServerGroup, - cfg.namespace, + group, + namespace, ) } - pooled.namespaces[cfg.namespace] = struct{}{} - pooled.refs++ - return pooled.server, func() { p.release(cfg.devServerGroup, pooled.server) }, nil + if err := pooled.server.RegisterNamespace(t.Context(), namespace); err != nil { + return nil, err + } + pooled.namespaces[namespace] = struct{}{} + return pooled.server, nil } server, err := devserver.Start(t.Context(), opts) if err != nil { - return nil, nil, err + return nil, err } - p.servers[cfg.devServerGroup] = &pooledTestDevServer{ + p.servers[key] = &pooledTestDevServer{ server: server, - dynamicConfig: maps.Clone(cfg.dynamicConfig), - namespaces: map[string]struct{}{cfg.namespace: {}}, - test: t, - refs: 1, + dynamicConfig: maps.Clone(opts.DynamicConfigValues), + namespaces: map[string]struct{}{namespace: {}}, } - return server, func() { p.release(cfg.devServerGroup, server) }, nil -} - -func (p *testDevServerPool) release(id string, server *devserver.Server) { - p.Lock() - pooled := p.servers[id] - if pooled == nil || pooled.server != server { + t.Cleanup(func() { + p.Lock() + delete(p.servers, key) p.Unlock() - return - } - pooled.refs-- - if pooled.refs > 0 { - p.Unlock() - return - } - delete(p.servers, id) - p.Unlock() - _ = server.Stop() + _ = server.Stop() + }) + return server, nil } diff --git a/loadgen/kitchen_sink_executor_test.go b/loadgen/kitchen_sink_executor_test.go index 21e09852c..7092865fc 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/loadgen/kitchen_sink_executor_test.go @@ -87,10 +87,10 @@ func TestKitchenSink(t *testing.T) { testEnvironments := make(map[clioptions.Language]*TestEnvironment) for _, sdk := range sdks { if onlySDK != "" && string(sdk) != onlySDK { - continue + continue // not using t.Skip as it's too noisy } testEnvironments[sdk] = SetupTestEnvironment(t, - WithDevServerGroup(t.Name()), + WithDevServerGroup("kitchen-sink"), WithNamespace(fmt.Sprintf("default.%s", sdk)), WithDynamicConfig(dynamicConfig), ) @@ -1147,13 +1147,10 @@ func TestKitchenSink(t *testing.T) { } input.WorkflowInput.InitialActions = append(input.WorkflowInput.InitialActions, ListActionSet(NewEmptyReturnResultAction())...) - for _, sdk := range sdks { - if onlySDK != "" && string(sdk) != onlySDK { - continue // not using t.Skip as it's too noisy - } + for sdk, env := range testEnvironments { t.Run(string(sdk), func(t *testing.T) { t.Parallel() - testForSDK(t, tc, sdk, testEnvironments[sdk], defaultWorkflowTimeout) + testForSDK(t, tc, sdk, env, defaultWorkflowTimeout) }) } }) From e34e4c46f3d66b80df85fbd557daafc28024577a Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Fri, 4 Sep 2026 12:05:56 -0700 Subject: [PATCH 05/11] Address isolated namespace review feedback --- internal/workertest/env.go | 2 +- internal/workertest/serverpool.go | 92 +++++++++++++++++---------- loadgen/kitchen_sink_executor_test.go | 12 ++-- 3 files changed, 68 insertions(+), 38 deletions(-) diff --git a/internal/workertest/env.go b/internal/workertest/env.go index 6c00883ec..cc31deb99 100644 --- a/internal/workertest/env.go +++ b/internal/workertest/env.go @@ -164,7 +164,7 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment serverOpts, ) } - require.NoError(t, err, "Failed to start dev server") + require.NoError(t, err, "Failed to acquire dev server") temporalClient, err := client.Dial(client.Options{ HostPort: server.FrontendHostPort(), diff --git a/internal/workertest/serverpool.go b/internal/workertest/serverpool.go index 81e64d308..0ac9f844c 100644 --- a/internal/workertest/serverpool.go +++ b/internal/workertest/serverpool.go @@ -11,7 +11,10 @@ import ( ) type pooledTestDevServer struct { + mu sync.Mutex + start sync.Once server *devserver.Server + err error dynamicConfig map[string]any namespaces map[string]struct{} } @@ -36,45 +39,70 @@ func (p *testDevServerPool) acquire( namespace string, opts devserver.Options, ) (*devserver.Server, error) { + key := testDevServerGroupKey{test: t, group: group} + p.Lock() - defer p.Unlock() + pooled := p.servers[key] + if pooled == nil { + pooled = &pooledTestDevServer{} + p.servers[key] = pooled + } + p.Unlock() - key := testDevServerGroupKey{test: t, group: group} - if pooled := p.servers[key]; pooled != nil { - if !maps.EqualFunc(pooled.dynamicConfig, opts.DynamicConfigValues, reflect.DeepEqual) { - return nil, fmt.Errorf( - "dev server group %q already uses different dynamic config", - group, - ) - } - if _, exists := pooled.namespaces[namespace]; exists { - return nil, fmt.Errorf( - "dev server group %q already uses namespace %q", - group, - namespace, - ) + started := false + pooled.start.Do(func() { + started = true + pooled.server, pooled.err = devserver.Start(t.Context(), opts) + if pooled.err != nil { + return } - if err := pooled.server.RegisterNamespace(t.Context(), namespace); err != nil { - return nil, err + pooled.dynamicConfig = maps.Clone(opts.DynamicConfigValues) + pooled.namespaces = map[string]struct{}{namespace: {}} + t.Cleanup(func() { + p.Lock() + delete(p.servers, key) + p.Unlock() + _ = pooled.server.Stop() + }) + }) + if pooled.err != nil { + p.Lock() + if p.servers[key] == pooled { + delete(p.servers, key) } - pooled.namespaces[namespace] = struct{}{} + p.Unlock() + return nil, pooled.err + } + if started { return pooled.server, nil } + return pooled.acquireNamespace(t, group, namespace, opts.DynamicConfigValues) +} - server, err := devserver.Start(t.Context(), opts) - if err != nil { - return nil, err +func (p *pooledTestDevServer) acquireNamespace( + t *testing.T, + group string, + namespace string, + dynamicConfig map[string]any, +) (*devserver.Server, error) { + p.mu.Lock() + defer p.mu.Unlock() + if !maps.EqualFunc(p.dynamicConfig, dynamicConfig, reflect.DeepEqual) { + return nil, fmt.Errorf( + "dev server group %q already uses different dynamic config", + group, + ) } - p.servers[key] = &pooledTestDevServer{ - server: server, - dynamicConfig: maps.Clone(opts.DynamicConfigValues), - namespaces: map[string]struct{}{namespace: {}}, + if _, exists := p.namespaces[namespace]; exists { + return nil, fmt.Errorf( + "dev server group %q already uses namespace %q", + group, + namespace, + ) } - t.Cleanup(func() { - p.Lock() - delete(p.servers, key) - p.Unlock() - _ = server.Stop() - }) - return server, nil + if err := p.server.RegisterNamespace(t.Context(), namespace); err != nil { + return nil, err + } + p.namespaces[namespace] = struct{}{} + return p.server, nil } diff --git a/loadgen/kitchen_sink_executor_test.go b/loadgen/kitchen_sink_executor_test.go index 7092865fc..bcab3f8bb 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/loadgen/kitchen_sink_executor_test.go @@ -23,8 +23,6 @@ import ( "google.golang.org/protobuf/types/known/emptypb" ) -const namespace = "default" - var ( sdks = []clioptions.Language{ clioptions.LangGo, @@ -1147,7 +1145,11 @@ func TestKitchenSink(t *testing.T) { } input.WorkflowInput.InitialActions = append(input.WorkflowInput.InitialActions, ListActionSet(NewEmptyReturnResultAction())...) - for sdk, env := range testEnvironments { + for _, sdk := range sdks { + env, ok := testEnvironments[sdk] + if !ok { + continue + } t.Run(string(sdk), func(t *testing.T) { t.Parallel() testForSDK(t, tc, sdk, env, defaultWorkflowTimeout) @@ -1306,7 +1308,7 @@ func testSupportedFeature( _, execErr := env.RunExecutorTest(t, testExecutor, scenarioInfo, sdk) taskQueueName := TaskQueueForRun(scenarioInfo.RunID) - historyEvents, historyErr := getWorkflowHistory(t, taskQueueName, env.TemporalClient()) + historyEvents, historyErr := getWorkflowHistory(t, taskQueueName, env.TemporalClient(), env.Namespace()) if execErr != nil { if len(historyEvents) > 0 { t.Logf("History events for debugging:") @@ -1348,7 +1350,7 @@ func (w *kitchenSinkTestWrapper) Run(ctx context.Context, info ScenarioInfo) err return w.executor.Run(ctx, info) } -func getWorkflowHistory(t *testing.T, taskQueueName string, temporalClient client.Client) ([]*history.HistoryEvent, error) { +func getWorkflowHistory(t *testing.T, taskQueueName string, temporalClient client.Client, namespace string) ([]*history.HistoryEvent, error) { executions, err := temporalClient.ListWorkflow(t.Context(), &workflowservice.ListWorkflowExecutionsRequest{ Namespace: namespace, From c57bb9fbeb731a8535e11b006cc93b51d9301a5d Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Fri, 4 Sep 2026 12:09:25 -0700 Subject: [PATCH 06/11] Reorder workflow history helper parameters --- loadgen/kitchen_sink_executor_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/loadgen/kitchen_sink_executor_test.go b/loadgen/kitchen_sink_executor_test.go index bcab3f8bb..1bd9de3f2 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/loadgen/kitchen_sink_executor_test.go @@ -1308,7 +1308,7 @@ func testSupportedFeature( _, execErr := env.RunExecutorTest(t, testExecutor, scenarioInfo, sdk) taskQueueName := TaskQueueForRun(scenarioInfo.RunID) - historyEvents, historyErr := getWorkflowHistory(t, taskQueueName, env.TemporalClient(), env.Namespace()) + historyEvents, historyErr := getWorkflowHistory(t, env.TemporalClient(), env.Namespace(), taskQueueName) if execErr != nil { if len(historyEvents) > 0 { t.Logf("History events for debugging:") @@ -1350,7 +1350,7 @@ func (w *kitchenSinkTestWrapper) Run(ctx context.Context, info ScenarioInfo) err return w.executor.Run(ctx, info) } -func getWorkflowHistory(t *testing.T, taskQueueName string, temporalClient client.Client, namespace string) ([]*history.HistoryEvent, error) { +func getWorkflowHistory(t *testing.T, temporalClient client.Client, namespace, taskQueueName string) ([]*history.HistoryEvent, error) { executions, err := temporalClient.ListWorkflow(t.Context(), &workflowservice.ListWorkflowExecutionsRequest{ Namespace: namespace, From da5843287ec6eb195e2660562b51188d2af5b84f Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Fri, 4 Sep 2026 12:22:34 -0700 Subject: [PATCH 07/11] Simplify shared dev server setup --- devserver/config_test.go | 21 +++++ devserver/devserver.go | 21 ++--- devserver/namespace.go | 3 + internal/workertest/env.go | 45 ++++++----- internal/workertest/serverpool.go | 108 -------------------------- loadgen/kitchen_sink_executor_test.go | 21 ++--- 6 files changed, 71 insertions(+), 148 deletions(-) delete mode 100644 internal/workertest/serverpool.go diff --git a/devserver/config_test.go b/devserver/config_test.go index e44a90b77..e9e38255e 100644 --- a/devserver/config_test.go +++ b/devserver/config_test.go @@ -1,10 +1,12 @@ package devserver import ( + "context" "errors" "os" "path/filepath" "testing" + "time" "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" @@ -36,6 +38,25 @@ func TestServerPorts(t *testing.T) { require.Equal(t, ports, server.Ports()) } +func TestRegisterNamespaceReportsExitedProcessAfterStop(t *testing.T) { + processErr := errors.New("boom") + exited := make(chan struct{}) + close(exited) + server := &Server{ + frontend: "127.0.0.1:1", + workDir: t.TempDir(), + cancel: func() {}, + exited: exited, + exitErr: processErr, + } + require.ErrorIs(t, server.Stop(), processErr) + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + err := server.RegisterNamespace(ctx, "test") + require.ErrorContains(t, err, "process exited before namespace registration completed: boom") +} + func TestDefaultOutputDirUsesRepoRoot(t *testing.T) { root := t.TempDir() require.NoError(t, os.WriteFile(filepath.Join(root, "go.mod"), []byte("module test\n"), 0644)) diff --git a/devserver/devserver.go b/devserver/devserver.go index 7648cf001..435ba89d3 100644 --- a/devserver/devserver.go +++ b/devserver/devserver.go @@ -104,7 +104,8 @@ type Server struct { logger *zap.SugaredLogger workDir string cancel context.CancelFunc - done chan error + exited chan struct{} + exitErr error stopOnce sync.Once stopErr error } @@ -193,17 +194,18 @@ func Start(ctx context.Context, opts Options) (*Server, error) { return nil, fmt.Errorf("devserver: start: %w", err) } - done := make(chan error, 1) - go func() { done <- cmd.Wait() }() - s := &Server{ frontend: frontendAddr, ports: serverPorts, logger: opts.Logger, workDir: workDir, cancel: cancel, - done: done, + exited: make(chan struct{}), } + go func() { + s.exitErr = cmd.Wait() + close(s.exited) + }() success = true if err := s.RegisterNamespace(ctx, opts.Namespace); err != nil { @@ -226,9 +228,8 @@ func (s *Server) RegisterNamespace(ctx context.Context, namespace string) error go func() { select { - case procErr := <-s.done: - s.done <- procErr - cancel(fmt.Errorf("process exited before namespace registration completed: %w", procErr)) + case <-s.exited: + cancel(fmt.Errorf("process exited before namespace registration completed: %w", s.exitErr)) case <-ctx.Done(): } }() @@ -247,9 +248,9 @@ func (s *Server) Ports() Ports { func (s *Server) Stop() error { s.stopOnce.Do(func() { s.cancel() - err := <-s.done + <-s.exited _ = os.RemoveAll(s.workDir) - s.stopErr = classifyExitErr(err) + s.stopErr = classifyExitErr(s.exitErr) }) return s.stopErr } diff --git a/devserver/namespace.go b/devserver/namespace.go index 770c1c624..9834d6fb7 100644 --- a/devserver/namespace.go +++ b/devserver/namespace.go @@ -36,6 +36,9 @@ func registerNamespace(ctx context.Context, frontendAddr, namespace string) erro if err == nil || isAlreadyExists(err) { return nil } + if ctx.Err() != nil { + return context.Cause(ctx) + } if !isRetryable(err) { return err } diff --git a/internal/workertest/env.go b/internal/workertest/env.go index cc31deb99..4ce2b7c8a 100644 --- a/internal/workertest/env.go +++ b/internal/workertest/env.go @@ -38,7 +38,7 @@ type testEnvConfig struct { runID string dynamicConfig map[string]any namespace string - devServerGroup string + sharedDevServer *devserver.Server } type TestEnvOption func(*testEnvConfig) @@ -73,11 +73,10 @@ func WithNamespace(namespace string) TestEnvOption { } } -// WithDevServerGroup configures the test environment to share a dev server -// with other environments created by the same test using the same group. -func WithDevServerGroup(group string) TestEnvOption { +// WithDevServer configures the test environment to use an existing dev server. +func WithDevServer(server *devserver.Server) TestEnvOption { return func(c *testEnvConfig) { - c.devServerGroup = group + c.sharedDevServer = server } } @@ -127,7 +126,7 @@ func (env *TestEnvironment) Namespace() string { return env.namespace } -func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment { +func newTestEnvConfig(opts ...TestEnvOption) testEnvConfig { cfg := testEnvConfig{ executorTimeout: defaultTestRunTimeout, namespace: defaultTestNamespace, @@ -135,6 +134,12 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment for _, opt := range opts { opt(&cfg) } + return cfg +} + +// StartDevServer starts a dev server and stops it during test cleanup. +func StartDevServer(t *testing.T, opts ...TestEnvOption) *devserver.Server { + cfg := newTestEnvConfig(opts...) serverRef, err := versions.Get("SERVER_VERSION") require.NoError(t, err) @@ -142,29 +147,27 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment testLogger := zaptest.NewLogger(t) serverLogger := testLogger.Named("devserver").Sugar() - serverOpts := devserver.Options{ + server, err := devserver.Start(t.Context(), devserver.Options{ Ref: serverRef, Namespace: cfg.namespace, DynamicConfigValues: cfg.dynamicConfig, Output: workerctl.NewLogWriter(serverLogger), Logger: serverLogger, - } + }) + require.NoError(t, err, "Failed to start dev server") + t.Cleanup(func() { _ = server.Stop() }) + return server +} - var server *devserver.Server - if cfg.devServerGroup == "" { - server, err = devserver.Start(t.Context(), serverOpts) - if err == nil { - t.Cleanup(func() { _ = server.Stop() }) - } +func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment { + cfg := newTestEnvConfig(opts...) + server := cfg.sharedDevServer + if server == nil { + server = StartDevServer(t, opts...) } else { - server, err = groupedTestDevServers.acquire( - t, - cfg.devServerGroup, - cfg.namespace, - serverOpts, - ) + err := server.RegisterNamespace(t.Context(), cfg.namespace) + require.NoError(t, err, "Failed to register namespace") } - require.NoError(t, err, "Failed to acquire dev server") temporalClient, err := client.Dial(client.Options{ HostPort: server.FrontendHostPort(), diff --git a/internal/workertest/serverpool.go b/internal/workertest/serverpool.go deleted file mode 100644 index 0ac9f844c..000000000 --- a/internal/workertest/serverpool.go +++ /dev/null @@ -1,108 +0,0 @@ -package workertest - -import ( - "fmt" - "maps" - "reflect" - "sync" - "testing" - - "github.com/temporalio/omes/devserver" -) - -type pooledTestDevServer struct { - mu sync.Mutex - start sync.Once - server *devserver.Server - err error - dynamicConfig map[string]any - namespaces map[string]struct{} -} - -type testDevServerGroupKey struct { - test *testing.T - group string -} - -type testDevServerPool struct { - sync.Mutex - servers map[testDevServerGroupKey]*pooledTestDevServer -} - -var groupedTestDevServers = testDevServerPool{ - servers: make(map[testDevServerGroupKey]*pooledTestDevServer), -} - -func (p *testDevServerPool) acquire( - t *testing.T, - group string, - namespace string, - opts devserver.Options, -) (*devserver.Server, error) { - key := testDevServerGroupKey{test: t, group: group} - - p.Lock() - pooled := p.servers[key] - if pooled == nil { - pooled = &pooledTestDevServer{} - p.servers[key] = pooled - } - p.Unlock() - - started := false - pooled.start.Do(func() { - started = true - pooled.server, pooled.err = devserver.Start(t.Context(), opts) - if pooled.err != nil { - return - } - pooled.dynamicConfig = maps.Clone(opts.DynamicConfigValues) - pooled.namespaces = map[string]struct{}{namespace: {}} - t.Cleanup(func() { - p.Lock() - delete(p.servers, key) - p.Unlock() - _ = pooled.server.Stop() - }) - }) - if pooled.err != nil { - p.Lock() - if p.servers[key] == pooled { - delete(p.servers, key) - } - p.Unlock() - return nil, pooled.err - } - if started { - return pooled.server, nil - } - return pooled.acquireNamespace(t, group, namespace, opts.DynamicConfigValues) -} - -func (p *pooledTestDevServer) acquireNamespace( - t *testing.T, - group string, - namespace string, - dynamicConfig map[string]any, -) (*devserver.Server, error) { - p.mu.Lock() - defer p.mu.Unlock() - if !maps.EqualFunc(p.dynamicConfig, dynamicConfig, reflect.DeepEqual) { - return nil, fmt.Errorf( - "dev server group %q already uses different dynamic config", - group, - ) - } - if _, exists := p.namespaces[namespace]; exists { - return nil, fmt.Errorf( - "dev server group %q already uses namespace %q", - group, - namespace, - ) - } - if err := p.server.RegisterNamespace(t.Context(), namespace); err != nil { - return nil, err - } - p.namespaces[namespace] = struct{}{} - return p.server, nil -} diff --git a/loadgen/kitchen_sink_executor_test.go b/loadgen/kitchen_sink_executor_test.go index 1bd9de3f2..75198f3aa 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/loadgen/kitchen_sink_executor_test.go @@ -82,15 +82,21 @@ func TestKitchenSink(t *testing.T) { "activity.enableStandalone": true, "history.enableStandaloneActivityOperatorCommands": true, } - testEnvironments := make(map[clioptions.Language]*TestEnvironment) + var enabledSDKs []clioptions.Language for _, sdk := range sdks { if onlySDK != "" && string(sdk) != onlySDK { continue // not using t.Skip as it's too noisy } + enabledSDKs = append(enabledSDKs, sdk) + } + require.NotEmpty(t, enabledSDKs, "SDK=%q matches no known SDK", onlySDK) + + server := StartDevServer(t, WithDynamicConfig(dynamicConfig)) + testEnvironments := make(map[clioptions.Language]*TestEnvironment, len(enabledSDKs)) + for _, sdk := range enabledSDKs { testEnvironments[sdk] = SetupTestEnvironment(t, - WithDevServerGroup("kitchen-sink"), - WithNamespace(fmt.Sprintf("default.%s", sdk)), - WithDynamicConfig(dynamicConfig), + WithDevServer(server), + WithNamespace(fmt.Sprintf("kitchensink-%s", sdk)), ) } @@ -1145,11 +1151,8 @@ func TestKitchenSink(t *testing.T) { } input.WorkflowInput.InitialActions = append(input.WorkflowInput.InitialActions, ListActionSet(NewEmptyReturnResultAction())...) - for _, sdk := range sdks { - env, ok := testEnvironments[sdk] - if !ok { - continue - } + for _, sdk := range enabledSDKs { + env := testEnvironments[sdk] t.Run(string(sdk), func(t *testing.T) { t.Parallel() testForSDK(t, tc, sdk, env, defaultWorkflowTimeout) From 2a303828e3f5582d0f3ccde5ddc1b9a89554c60e Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Fri, 4 Sep 2026 12:23:35 -0700 Subject: [PATCH 08/11] Remove namespace registration regression test --- devserver/config_test.go | 21 --------------------- 1 file changed, 21 deletions(-) diff --git a/devserver/config_test.go b/devserver/config_test.go index e9e38255e..e44a90b77 100644 --- a/devserver/config_test.go +++ b/devserver/config_test.go @@ -1,12 +1,10 @@ package devserver import ( - "context" "errors" "os" "path/filepath" "testing" - "time" "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" @@ -38,25 +36,6 @@ func TestServerPorts(t *testing.T) { require.Equal(t, ports, server.Ports()) } -func TestRegisterNamespaceReportsExitedProcessAfterStop(t *testing.T) { - processErr := errors.New("boom") - exited := make(chan struct{}) - close(exited) - server := &Server{ - frontend: "127.0.0.1:1", - workDir: t.TempDir(), - cancel: func() {}, - exited: exited, - exitErr: processErr, - } - require.ErrorIs(t, server.Stop(), processErr) - - ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) - defer cancel() - err := server.RegisterNamespace(ctx, "test") - require.ErrorContains(t, err, "process exited before namespace registration completed: boom") -} - func TestDefaultOutputDirUsesRepoRoot(t *testing.T) { root := t.TempDir() require.NoError(t, os.WriteFile(filepath.Join(root, "go.mod"), []byte("module test\n"), 0644)) From efb30281ca2a38217b670baf07b8767fb394e211 Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Fri, 4 Sep 2026 12:29:23 -0700 Subject: [PATCH 09/11] Update kitchen_sink_executor_test.go --- loadgen/kitchen_sink_executor_test.go | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/loadgen/kitchen_sink_executor_test.go b/loadgen/kitchen_sink_executor_test.go index 75198f3aa..40f1d86b1 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/loadgen/kitchen_sink_executor_test.go @@ -73,15 +73,7 @@ func TestKitchenSink(t *testing.T) { if os.Getenv("CI") != "" && onlySDK == "" { t.Skip("Skipping kitchensink test in CI without specific SDK set") } - dynamicConfig := map[string]any{ - // Enable StartNexusOperationExecution for the standalone-nexus subtests. - "nexusoperation.enableStandalone": true, - // Standalone Nexus system callbacks require CHASM callbacks. - "history.enableCHASMCallbacks": true, - // Enable StartActivityExecution for the standalone-activity subtest. - "activity.enableStandalone": true, - "history.enableStandaloneActivityOperatorCommands": true, - } + var enabledSDKs []clioptions.Language for _, sdk := range sdks { if onlySDK != "" && string(sdk) != onlySDK { @@ -91,7 +83,16 @@ func TestKitchenSink(t *testing.T) { } require.NotEmpty(t, enabledSDKs, "SDK=%q matches no known SDK", onlySDK) - server := StartDevServer(t, WithDynamicConfig(dynamicConfig)) + server := StartDevServer(t, WithDynamicConfig(map[string]any{ + // Enable StartNexusOperationExecution for the standalone-nexus subtests. + "nexusoperation.enableStandalone": true, + // Standalone Nexus system callbacks require CHASM callbacks. + "history.enableCHASMCallbacks": true, + // Enable StartActivityExecution for the standalone-activity subtest. + "activity.enableStandalone": true, + "history.enableStandaloneActivityOperatorCommands": true, + })) + testEnvironments := make(map[clioptions.Language]*TestEnvironment, len(enabledSDKs)) for _, sdk := range enabledSDKs { testEnvironments[sdk] = SetupTestEnvironment(t, From 9ee370f5e988412c8620992f5c34bd1e6c2b084e Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Fri, 4 Sep 2026 12:37:04 -0700 Subject: [PATCH 10/11] Split dev server and environment options --- internal/workertest/env.go | 27 +++++++++++++++++++------- scenarios/throughput_stress_test.go | 30 +++++++++++++++-------------- 2 files changed, 36 insertions(+), 21 deletions(-) diff --git a/internal/workertest/env.go b/internal/workertest/env.go index 4ce2b7c8a..fb8cf5bb4 100644 --- a/internal/workertest/env.go +++ b/internal/workertest/env.go @@ -33,10 +33,16 @@ const ( ) // Functional options configuration +type devServerConfig struct { + dynamicConfig map[string]any +} + +// DevServerOption configures a test dev server. +type DevServerOption func(*devServerConfig) + type testEnvConfig struct { executorTimeout time.Duration runID string - dynamicConfig map[string]any namespace string sharedDevServer *devserver.Server } @@ -60,8 +66,8 @@ func WithNexusEndpoint(runID string) TestEnvOption { // WithDynamicConfig passes dynamic-config overrides to the dev server, on top // of the test-friendly defaults that devserver applies. Use this for opt-in // feature gates that a specific test needs. -func WithDynamicConfig(values map[string]any) TestEnvOption { - return func(c *testEnvConfig) { +func WithDynamicConfig(values map[string]any) DevServerOption { + return func(c *devServerConfig) { c.dynamicConfig = values } } @@ -138,8 +144,15 @@ func newTestEnvConfig(opts ...TestEnvOption) testEnvConfig { } // StartDevServer starts a dev server and stops it during test cleanup. -func StartDevServer(t *testing.T, opts ...TestEnvOption) *devserver.Server { - cfg := newTestEnvConfig(opts...) +func StartDevServer(t *testing.T, opts ...DevServerOption) *devserver.Server { + return startDevServer(t, defaultTestNamespace, opts...) +} + +func startDevServer(t *testing.T, namespace string, opts ...DevServerOption) *devserver.Server { + var cfg devServerConfig + for _, opt := range opts { + opt(&cfg) + } serverRef, err := versions.Get("SERVER_VERSION") require.NoError(t, err) @@ -149,7 +162,7 @@ func StartDevServer(t *testing.T, opts ...TestEnvOption) *devserver.Server { serverLogger := testLogger.Named("devserver").Sugar() server, err := devserver.Start(t.Context(), devserver.Options{ Ref: serverRef, - Namespace: cfg.namespace, + Namespace: namespace, DynamicConfigValues: cfg.dynamicConfig, Output: workerctl.NewLogWriter(serverLogger), Logger: serverLogger, @@ -163,7 +176,7 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment cfg := newTestEnvConfig(opts...) server := cfg.sharedDevServer if server == nil { - server = StartDevServer(t, opts...) + server = startDevServer(t, cfg.namespace) } else { err := server.RegisterNamespace(t.Context(), cfg.namespace) require.NoError(t, err, "Failed to register namespace") diff --git a/scenarios/throughput_stress_test.go b/scenarios/throughput_stress_test.go index 47ac0c27c..0620638b0 100644 --- a/scenarios/throughput_stress_test.go +++ b/scenarios/throughput_stress_test.go @@ -93,16 +93,17 @@ func TestThroughputStressFeatureAutoEnable(t *testing.T) { // No WithNexusEndpoint here: the scenario creates its own endpoint for the run, // which is part of the path under test. Pre-creating one would collide on the // same NexusEndpointForRun name. + server := workertest.StartDevServer(t, workertest.WithDynamicConfig(map[string]any{ + // Gates the capabilities the feature options read. + "activity.enableStandalone": true, + "history.enableStandaloneActivityOperatorCommands": true, + "nexusoperation.enableStandalone": true, + // Standalone Nexus system callbacks require CHASM callbacks. + "history.enableCHASMCallbacks": true, + })) env := workertest.SetupTestEnvironment(t, workertest.WithExecutorTimeout(1*time.Minute), - workertest.WithDynamicConfig(map[string]any{ - // Gates the capabilities the feature options read. - "activity.enableStandalone": true, - "history.enableStandaloneActivityOperatorCommands": true, - "nexusoperation.enableStandalone": true, - // Standalone Nexus system callbacks require CHASM callbacks. - "history.enableCHASMCallbacks": true, - })) + workertest.WithDevServer(server)) // Without the namespace advertising standalone activities there is nothing for // the feature option to resolve from, and the rest of this test proves nothing. @@ -152,14 +153,15 @@ func TestThroughputStressNexusStandaloneActivity(t *testing.T) { runID := fmt.Sprintf("tps-nsa-%d", time.Now().Unix()) // Enable the activity-backed operation and standalone-Nexus completion path. + server := workertest.StartDevServer(t, workertest.WithDynamicConfig(map[string]any{ + "activity.enableStandalone": true, + "activity.enableCallbacks": true, + "nexusoperation.enableStandalone": true, + "history.enableCHASMCallbacks": true, + })) env := workertest.SetupTestEnvironment(t, workertest.WithExecutorTimeout(1*time.Minute), - workertest.WithDynamicConfig(map[string]any{ - "activity.enableStandalone": true, - "activity.enableCallbacks": true, - "nexusoperation.enableStandalone": true, - "history.enableCHASMCallbacks": true, - })) + workertest.WithDevServer(server)) scenarioInfo := loadgen.ScenarioInfo{ RunID: runID, From 85e3fd877831f7b47d43a1bd1852ffcfdb6fce49 Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Fri, 4 Sep 2026 12:54:10 -0700 Subject: [PATCH 11/11] Address remaining namespace review feedback --- devserver/devserver.go | 8 ++++++-- internal/workertest/env.go | 6 ++++-- loadgen/kitchen_sink_executor_test.go | 14 ++++++++++++-- 3 files changed, 22 insertions(+), 6 deletions(-) diff --git a/devserver/devserver.go b/devserver/devserver.go index 435ba89d3..c0bc3c8a4 100644 --- a/devserver/devserver.go +++ b/devserver/devserver.go @@ -229,7 +229,11 @@ func (s *Server) RegisterNamespace(ctx context.Context, namespace string) error go func() { select { case <-s.exited: - cancel(fmt.Errorf("process exited before namespace registration completed: %w", s.exitErr)) + err := errors.New("process exited before namespace registration completed") + if s.exitErr != nil { + err = fmt.Errorf("%w: %w", err, s.exitErr) + } + cancel(err) case <-ctx.Done(): } }() @@ -260,7 +264,7 @@ func (s *Server) Stop() error { // non-zero exit codes that didn't come from a signal, and SIGKILL that // fires when the server fails to exit within WaitDelay. func classifyExitErr(err error) error { - if err == nil { + if err == nil || errors.Is(err, context.Canceled) { return nil } var exitErr *exec.ExitError diff --git a/internal/workertest/env.go b/internal/workertest/env.go index fb8cf5bb4..b2735514b 100644 --- a/internal/workertest/env.go +++ b/internal/workertest/env.go @@ -168,7 +168,9 @@ func startDevServer(t *testing.T, namespace string, opts ...DevServerOption) *de Logger: serverLogger, }) require.NoError(t, err, "Failed to start dev server") - t.Cleanup(func() { _ = server.Stop() }) + t.Cleanup(func() { + require.NoError(t, server.Stop(), "Dev server exited abnormally") + }) return server } @@ -179,7 +181,7 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment server = startDevServer(t, cfg.namespace) } else { err := server.RegisterNamespace(t.Context(), cfg.namespace) - require.NoError(t, err, "Failed to register namespace") + require.NoErrorf(t, err, "Failed to register namespace %q", cfg.namespace) } temporalClient, err := client.Dial(client.Options{ diff --git a/loadgen/kitchen_sink_executor_test.go b/loadgen/kitchen_sink_executor_test.go index 40f1d86b1..26cb8f928 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/loadgen/kitchen_sink_executor_test.go @@ -19,6 +19,7 @@ import ( "go.temporal.io/api/history/v1" "go.temporal.io/api/workflowservice/v1" "go.temporal.io/sdk/client" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/emptypb" ) @@ -74,9 +75,14 @@ func TestKitchenSink(t *testing.T) { t.Skip("Skipping kitchensink test in CI without specific SDK set") } + var only clioptions.Language + if onlySDK != "" { + require.NoError(t, only.Set(onlySDK), "invalid SDK env var") + } + var enabledSDKs []clioptions.Language for _, sdk := range sdks { - if onlySDK != "" && string(sdk) != onlySDK { + if only != "" && sdk != only { continue // not using t.Skip as it's too noisy } enabledSDKs = append(enabledSDKs, sdk) @@ -1234,8 +1240,12 @@ func testForSDK( // worker picks them up. runTaskQueue := TaskQueueForRun(scenarioInfo.RunID) + // PrepareTestInput fills the per-run endpoint and task queue into the input in + // place, so work on a copy before making further per-run changes. + testInput := proto.Clone(tc.testInput).(*TestInput) + executor := &KitchenSinkExecutor{ - TestInput: tc.testInput, + TestInput: testInput, PrepareTestInput: func(_ context.Context, _ ScenarioInfo, input *TestInput) error { if input.WorkflowInput != nil { for _, actionSet := range input.WorkflowInput.InitialActions {