Run kitchen sink SDK tests in isolated namespaces

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,20 +194,21 @@ 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 {
+	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 +221,26 @@ 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 {
+	ctx, cancel := context.WithCancelCause(ctx)
+	defer cancel(nil)
+
+	go func() {
+		select {
+		case <-s.exited:
+			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():
+		}
+	}()
+
+	return registerNamespace(ctx, s.frontend, namespace)
+}
+
 // Ports returns the service ports allocated for the running server.
 func (s *Server) Ports() Ports {
 	return s.ports
@@ -230,9 +252,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
 }
@@ -242,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
@@ -254,19 +276,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)
-}

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
 		}

internal/workertest/env.go

@@ -26,17 +26,25 @@ import (
 )
 
 const (
-	testNamespace         = "default"
+	defaultTestNamespace  = "default"
 	defaultTestRunTimeout = 30 * time.Second
 	workerBuildTimeout    = 1 * time.Minute
 	workerShutdownTimeout = 5 * time.Second
 )
 
 // 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
 }
 
 type TestEnvOption func(*testEnvConfig)
@@ -58,12 +66,26 @@ 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
 	}
 }
 
+// WithNamespace configures the namespace used by the test environment.
+func WithNamespace(namespace string) TestEnvOption {
+	return func(c *testEnvConfig) {
+		c.namespace = namespace
+	}
+}
+
+// WithDevServer configures the test environment to use an existing dev server.
+func WithDevServer(server *devserver.Server) TestEnvOption {
+	return func(c *testEnvConfig) {
+		c.sharedDevServer = server
+	}
+}
+
 type TestResult struct {
 	ObservedLogs *observer.ObservedLogs
 }
@@ -106,13 +128,31 @@ func (env *TestEnvironment) NexusEndpointName() string {
 	return env.nexusEndpointName
 }
 
-func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment {
+func (env *TestEnvironment) Namespace() string {
+	return env.namespace
+}
+
+func newTestEnvConfig(opts ...TestEnvOption) testEnvConfig {
 	cfg := testEnvConfig{
 		executorTimeout: defaultTestRunTimeout,
+		namespace:       defaultTestNamespace,
 	}
 	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 ...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)
@@ -122,16 +162,31 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment
 	serverLogger := testLogger.Named("devserver").Sugar()
 	server, err := devserver.Start(t.Context(), devserver.Options{
 		Ref:                 serverRef,
-		Namespace:           testNamespace,
+		Namespace:           namespace,
 		DynamicConfigValues: cfg.dynamicConfig,
 		Output:              workerctl.NewLogWriter(serverLogger),
 		Logger:              serverLogger,
 	})
 	require.NoError(t, err, "Failed to start dev server")
+	t.Cleanup(func() {
+		require.NoError(t, server.Stop(), "Dev server exited abnormally")
+	})
+	return server
+}
+
+func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment {
+	cfg := newTestEnvConfig(opts...)
+	server := cfg.sharedDevServer
+	if server == nil {
+		server = startDevServer(t, cfg.namespace)
+	} else {
+		err := server.RegisterNamespace(t.Context(), cfg.namespace)
+		require.NoErrorf(t, err, "Failed to register namespace %q", cfg.namespace)
+	}
 
 	temporalClient, err := client.Dial(client.Options{
 		HostPort:  server.FrontendHostPort(),
-		Namespace: testNamespace,
+		Namespace: cfg.namespace,
 	})
 	require.NoError(t, err, "Failed to create Temporal client")
 
@@ -163,9 +218,6 @@ func (env *TestEnvironment) cleanup() {
 	if env.temporalClient != nil {
 		env.temporalClient.Close()
 	}
-	if env.devServer != nil {
-		env.devServer.Stop()
-	}
 	env.workerPool.cleanup()
 }
 
@@ -179,7 +231,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 +273,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
 	}
 

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")
 		}

loadgen/kitchen_sink_executor_test.go

@@ -19,12 +19,11 @@ 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"
 )
 
-const namespace = "default"
-
 var (
 	sdks = []clioptions.Language{
 		clioptions.LangGo,
@@ -75,7 +74,22 @@ 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{
+
+	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 only != "" && sdk != only {
+			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(map[string]any{
 		// Enable StartNexusOperationExecution for the standalone-nexus subtests.
 		"nexusoperation.enableStandalone": true,
 		// Standalone Nexus system callbacks require CHASM callbacks.
@@ -85,6 +99,14 @@ func TestKitchenSink(t *testing.T) {
 		"history.enableStandaloneActivityOperatorCommands": true,
 	}))
 
+	testEnvironments := make(map[clioptions.Language]*TestEnvironment, len(enabledSDKs))
+	for _, sdk := range enabledSDKs {
+		testEnvironments[sdk] = SetupTestEnvironment(t,
+			WithDevServer(server),
+			WithNamespace(fmt.Sprintf("kitchensink-%s", sdk)),
+		)
+	}
+
 	// Default workflow execution timeout for tests
 	defaultWorkflowTimeout := 30 * time.Second
 
@@ -1136,10 +1158,8 @@ 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 := range enabledSDKs {
+				env := testEnvironments[sdk]
 				t.Run(string(sdk), func(t *testing.T) {
 					t.Parallel()
 					testForSDK(t, tc, sdk, env, defaultWorkflowTimeout)
@@ -1220,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 {
@@ -1298,7 +1322,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, env.TemporalClient(), env.Namespace(), taskQueueName)
 	if execErr != nil {
 		if len(historyEvents) > 0 {
 			t.Logf("History events for debugging:")
@@ -1340,7 +1364,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, temporalClient client.Client, namespace, taskQueueName string) ([]*history.HistoryEvent, error) {
 	executions, err := temporalClient.ListWorkflow(t.Context(),
 		&workflowservice.ListWorkflowExecutionsRequest{
 			Namespace: namespace,