diff --git a/devserver/devserver.go b/devserver/devserver.go index f81f62da1..c0bc3c8a4 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,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) -} 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 df2126950..b2735514b 100644 --- a/internal/workertest/env.go +++ b/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 } 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..26cb8f928 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/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, 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,