diff --git a/docs/throughput-stress.md b/docs/throughput-stress.md index 33e32cb8..1659f20f 100644 --- a/docs/throughput-stress.md +++ b/docs/throughput-stress.md @@ -90,15 +90,15 @@ is any. Asking for `include-standalone-nexus=true` while Nexus is off is a contradiction, and fails the run. -## Nexus operation with a standalone activity +## Nexus operation actions -`include-nexus-standalone-activity` adds a standalone activity backed Nexus operation. -It is driven two ways each iteration: As an in-workflow Nexus operation, and — when standalone Nexus is part of the run — -as a standalone Nexus operation. +### `include-nexus-workflow-actions` -This is **opt-in and off by default**; pass `--option include-nexus-standalone-activity=true`. -It requires `nexus-enabled` and also needs server support for standalone activities and activity -completion callbacks (dynamic config `activity.enableStandalone` and `activity.enableCallbacks`) and a -Nexus callback URL; if those are off the operation fails clearly rather than being skipped. +The workflow actions start their target workflow with signal-with-start, update it, then send an +ordinary signal that completes it. This requires Nexus update callback support. -Currently only supported and run by Go workers. +### `include-nexus-standalone-activity` + +The standalone activity action is driven two ways each iteration: as an in-workflow Nexus operation +and, when standalone Nexus is part of the run, as a standalone Nexus operation. This requires +standalone activity and callback support. diff --git a/loadgen/generic_executor.go b/loadgen/generic_executor.go index 3e656c2b..6a641596 100644 --- a/loadgen/generic_executor.go +++ b/loadgen/generic_executor.go @@ -154,24 +154,26 @@ func (g *genericRun) Run(ctx context.Context) error { // cancellation while the run is healthy. stopping := iterErr != nil && ctx.Err() != nil && errors.Is(iterErr, context.Canceled) + switch { + case stopping: + g.logger.Debugf("Iteration %v abandoned: run is stopping", run.Iteration) + case iterErr == nil: + run.Duration = elapsed + g.completed.Add(1) + if g.config.OnCompletion != nil { + g.config.OnCompletion(ctx, run) + } + default: + g.failed.Add(1) + if g.config.OnIterationFailure != nil { + g.config.OnIterationFailure(ctx, run, iterErr) + } + } + + // Notify the waiter after callbacks finish so Run cannot return before they do. select { case <-ctx.Done(): case doneCh <- err: - switch { - case stopping: - g.logger.Debugf("Iteration %v abandoned: run is stopping", run.Iteration) - case iterErr == nil: - run.Duration = elapsed - g.completed.Add(1) - if g.config.OnCompletion != nil { - g.config.OnCompletion(ctx, run) - } - default: - g.failed.Add(1) - if g.config.OnIterationFailure != nil { - g.config.OnIterationFailure(ctx, run, iterErr) - } - } } }() diff --git a/loadgen/generic_executor_test.go b/loadgen/generic_executor_test.go index 1c2fae4f..5f5ddd07 100644 --- a/loadgen/generic_executor_test.go +++ b/loadgen/generic_executor_test.go @@ -37,6 +37,10 @@ func (i *iterationTracker) assertSeen(t *testing.T, iterations int) { } func execute(executor *GenericExecutor, runConfig RunConfiguration) error { + return executeContext(context.Background(), executor, runConfig) +} + +func executeContext(ctx context.Context, executor *GenericExecutor, runConfig RunConfiguration) error { logger := zap.Must(zap.NewDevelopment()) defer logger.Sync() info := ScenarioInfo{ @@ -44,7 +48,7 @@ func execute(executor *GenericExecutor, runConfig RunConfiguration) error { Logger: logger.Sugar(), Configuration: runConfig, } - return executor.Run(context.Background(), info) + return executor.Run(ctx, info) } func TestRunHappyPathIterations(t *testing.T) { @@ -62,6 +66,44 @@ func TestRunHappyPathIterations(t *testing.T) { }) } +func TestRunWaitsForOnCompletion(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + callbackStarted := make(chan struct{}) + releaseCallback := make(chan struct{}) + runDone := make(chan error, 1) + + go func() { + runDone <- execute(&GenericExecutor{ + Execute: func(ctx context.Context, run *Run) error { + return nil + }}, + RunConfiguration{ + Iterations: 1, + OnCompletion: func(ctx context.Context, run *Run) { + close(callbackStarted) + <-releaseCallback + }, + }, + ) + }() + + <-callbackStarted + + synctest.Wait() + + returned := false + select { + case <-runDone: + returned = true + default: + } + close(releaseCallback) + + require.False(t, returned, "executor returned before OnCompletion finished") + require.NoError(t, <-runDone) + }) +} + func TestRunFailIterations(t *testing.T) { synctest.Test(t, func(t *testing.T) { tracker := newIterationTracker() @@ -278,6 +320,38 @@ func TestRunContinueOnIterationFailure(t *testing.T) { }) } +func TestRunReportsNonCancellationFailureAfterCancellation(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + failureReported := make(chan struct{}, 1) + err := executeContext(ctx, &GenericExecutor{ + Execute: func(ctx context.Context, run *Run) error { + cancel() + return errors.New("deliberate fail from test") + }, + }, RunConfiguration{ + Iterations: 1, + ContinueOnIterationFailure: true, + OnIterationFailure: func(ctx context.Context, run *Run, err error) { + failureReported <- struct{}{} + }, + }) + require.Error(t, err) + + synctest.Wait() + + reported := false + select { + case <-failureReported: + reported = true + default: + } + require.True(t, reported, "non-cancellation error should be reported as a failure") + }) +} + // TestRunStoppedIterationsAreNotCountedAsFailures pins that iterations abandoned // by a caller stopping the run are left out of the tallies, so a clean stop is // not reported as a burst of failures. @@ -291,7 +365,7 @@ func TestRunStoppedIterationsAreNotCountedAsFailures(t *testing.T) { defer cancel() var inFlight int - executor := &GenericExecutor{ + err := executeContext(ctx, &GenericExecutor{ Execute: func(ctx context.Context, run *Run) error { mu.Lock() inFlight++ @@ -306,27 +380,19 @@ func TestRunStoppedIterationsAreNotCountedAsFailures(t *testing.T) { <-ctx.Done() return ctx.Err() }, - } - - logger := zap.Must(zap.NewDevelopment()) - defer logger.Sync() - err := executor.Run(ctx, ScenarioInfo{ - MetricsHandler: client.MetricsNopHandler, - Logger: logger.Sugar(), - Configuration: RunConfiguration{ - Iterations: 100, - MaxConcurrent: concurrent, - ContinueOnIterationFailure: true, - OnCompletion: func(ctx context.Context, run *Run) { - mu.Lock() - defer mu.Unlock() - completed = append(completed, run.Iteration) - }, - OnIterationFailure: func(ctx context.Context, run *Run, err error) { - mu.Lock() - defer mu.Unlock() - failed = append(failed, run.Iteration) - }, + }, RunConfiguration{ + Iterations: 100, + MaxConcurrent: concurrent, + ContinueOnIterationFailure: true, + OnCompletion: func(ctx context.Context, run *Run) { + mu.Lock() + defer mu.Unlock() + completed = append(completed, run.Iteration) + }, + OnIterationFailure: func(ctx context.Context, run *Run, err error) { + mu.Lock() + defer mu.Unlock() + failed = append(failed, run.Iteration) }, }) diff --git a/scenarios/throughput_stress.go b/scenarios/throughput_stress.go index fbff4ffd..c6cb7be4 100644 --- a/scenarios/throughput_stress.go +++ b/scenarios/throughput_stress.go @@ -67,6 +67,9 @@ const ( // Opt-in and off by default (only the Go worker implements the operation); requires Nexus load // (nexus-enabled) and server support for standalone activities + activity completion callbacks. IncludeNexusStandaloneActivityFlag = "include-nexus-standalone-activity" + // IncludeNexusWorkflowActionsFlag enables Nexus operations that signal and update a workflow. + // Opt-in and off by default; requires Nexus load (nexus-enabled). + IncludeNexusWorkflowActionsFlag = "include-nexus-workflow-actions" // PayloadDistributionJsonFlag is a JSON string (or @file) configuring a weighted // activity payload-size distribution. See loadgen.PayloadConfig for details. PayloadDistributionJsonFlag = "payload-distribution-json" @@ -109,6 +112,7 @@ type tpsConfig struct { IncludeStandaloneActivity bool IncludeStandaloneActivityOperatorCommands bool IncludeNexusStandaloneActivity bool + IncludeNexusWorkflowActions bool Payload *loadgen.PayloadConfig } @@ -119,6 +123,8 @@ type tpsExecutor struct { isResuming bool runID string rng *rand.Rand + // onActionsCreated observes generated actions and may be called concurrently by iteration goroutines. + onActionsCreated func([]*ActionSet) } var _ loadgen.Resumable = (*tpsExecutor)(nil) @@ -148,6 +154,7 @@ func init() { return c.Namespace.GetStandaloneActivityOperatorCommands() }) o.Bool(IncludeNexusStandaloneActivityFlag, false, "Include a Nexus operation that starts a standalone activity (Go worker only).") + o.Bool(IncludeNexusWorkflowActionsFlag, false, "Include Nexus operations that signal and update a workflow (Go worker only).") o.String(PayloadDistributionJsonFlag, "", "JSON payload-size distribution; use @ to read from a file.") }, ExecutorFn: func() loadgen.Executor { return newThroughputStressExecutor() }, @@ -155,7 +162,10 @@ func init() { } func newThroughputStressExecutor() *tpsExecutor { - return &tpsExecutor{state: &tpsState{}} + return &tpsExecutor{ + state: &tpsState{}, + onActionsCreated: func([]*ActionSet) {}, + } } // Snapshot returns a snapshot of the current state. @@ -248,6 +258,10 @@ func (t *tpsExecutor) Configure(info loadgen.ScenarioInfo) error { if config.IncludeNexusStandaloneActivity && !config.NexusEnabled { return fmt.Errorf("%s requires %s", IncludeNexusStandaloneActivityFlag, NexusEnabledFlag) } + config.IncludeNexusWorkflowActions = info.OptionBool(IncludeNexusWorkflowActionsFlag) + if config.IncludeNexusWorkflowActions && !config.NexusEnabled { + return fmt.Errorf("%s requires %s", IncludeNexusWorkflowActionsFlag, NexusEnabledFlag) + } if payloadStr := info.OptionString(PayloadDistributionJsonFlag); payloadStr != "" { config.Payload, err = loadgen.ParseAndValidatePayloadConfig(payloadStr) @@ -287,6 +301,7 @@ func (t *tpsExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) error // Standalone operations are part of Nexus load, so they go with it. t.config.IncludeStandaloneNexus = false t.config.IncludeNexusStandaloneActivity = false + t.config.IncludeNexusWorkflowActions = false } else { info.Logger.Infof("Using nexus endpoint %q", nexus.Endpoint) } @@ -364,7 +379,9 @@ func (t *tpsExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) error // // NOTE: No client actions (e.g. Signal) are defined; however, client action activities are. // That means these client actions are sent from the activity worker instead of Omes. - options.Params.WorkflowInput.InitialActions = t.createActions(run) + actions := t.createActions(run) + options.Params.WorkflowInput.InitialActions = actions + t.onActionsCreated(actions) return nil }, @@ -521,6 +538,7 @@ func (t *tpsExecutor) createActionsChunk( // Create actions for the current chunk for i := 0; i < itersPerChunk; i++ { + iterationIndex := t.internalIterationIndex(run, remainingInternalIters, i) syncActions := []*Action{ PayloadActivity(t.samplePayloadSize(rng), t.samplePayloadSize(rng), DefaultLocalActivity), PayloadActivity(0, t.samplePayloadSize(rng), DefaultLocalActivity), @@ -616,6 +634,10 @@ func (t *tpsExecutor) createActionsChunk( ) } } + if t.config.IncludeNexusWorkflowActions { + nexusWorkflowID := fmt.Sprintf("%s/nexus-workflow-%d", run.DefaultStartWorkflowOptions().ID, iterationIndex) + asyncActions = append(asyncActions, t.createNexusWorkflowActionSequence(nexusWorkflowID)) + } } // Add standalone activities, if configured. @@ -623,11 +645,10 @@ func (t *tpsExecutor) createActionsChunk( asyncActions = append(asyncActions, t.createStandaloneActivityAction(loadgen.TaskQueueForRun(run.RunID), rng)) } if t.config.IncludeStandaloneActivityOperatorCommands { - commandOrdinal := t.internalIterationIndex(run, remainingInternalIters, i) asyncActions = append(asyncActions, t.createStandaloneActivityOperatorCommandsAction( loadgen.TaskQueueForRun(run.RunID), - commandOrdinal, + iterationIndex, ), ) } @@ -896,22 +917,26 @@ func (t *tpsExecutor) createNexusAttachCallbacksAction() *Action { {Variant: &Action_NestedActionSet{ NestedActionSet: &ActionSet{Concurrent: true, Actions: fanout}, }}, - {Variant: &Action_SendSignal{ - SendSignal: &SendSignalAction{ - WorkflowId: handlerWfID, - SignalName: "do_actions_signal", - Args: []*common.Payload{ConvertToPayload(&DoSignal_DoSignalActions{ - Variant: &DoSignal_DoSignalActions_DoActions{ - DoActions: SingleActionSet(NewEmptyReturnResultAction()), - }, - })}, - AwaitableChoice: &AwaitableChoice{ - // The operation futures below are the correctness gate. Do not fail if an - // active/passive transition replays this after the handler has completed. - Condition: &AwaitableChoice_Abandon{Abandon: &emptypb.Empty{}}, - }, + NexusOperation(&ExecuteNexusOperation{ + Endpoint: t.config.NexusEndpoint, + Input: &NexusOperationRequest{ + Action: &NexusOperationRequest_WorkflowAction{WorkflowAction: &NexusWorkflowAction{ + WorkflowId: handlerWfID, + Action: &NexusWorkflowAction_Signal{Signal: &DoSignal{ + Variant: &DoSignal_DoSignalActions_{DoSignalActions: &DoSignal_DoSignalActions{ + Variant: &DoSignal_DoSignalActions_DoActions{ + DoActions: SingleActionSet(NewEmptyReturnResultAction()), + }, + }}, + }}, + }}, }, - }}, + AwaitableChoice: &AwaitableChoice{ + // The operation futures below are the correctness gate. Do not fail if an + // active/passive transition replays this after the handler has completed. + Condition: &AwaitableChoice_Abandon{Abandon: &emptypb.Empty{}}, + }, + }), {Variant: &Action_AwaitPendingActions{ AwaitPendingActions: &AwaitPendingActions{}, }}, @@ -932,6 +957,78 @@ func (t *tpsExecutor) createNexusStandaloneActivityAction() *Action { }) } +// createNexusWorkflowActionSequence exercises workflow messaging through one Nexus target. +func (t *tpsExecutor) createNexusWorkflowActionSequence(workflowID string) *Action { + return &Action{Variant: &Action_NestedActionSet{NestedActionSet: &ActionSet{Actions: []*Action{ + t.createNexusSignalWithStartAction(workflowID, &WorkflowInput{}), + t.createNexusUpdateAction(workflowID), + t.createNexusSignalAction(workflowID), + }}}} +} + +func (t *tpsExecutor) createNexusSignalAction(workflowID string) *Action { + return NexusOperation(&ExecuteNexusOperation{ + Endpoint: t.config.NexusEndpoint, + Input: &NexusOperationRequest{ + Action: &NexusOperationRequest_WorkflowAction{WorkflowAction: &NexusWorkflowAction{ + WorkflowId: workflowID, + Action: &NexusWorkflowAction_Signal{Signal: &DoSignal{ + Variant: &DoSignal_DoSignalActions_{DoSignalActions: &DoSignal_DoSignalActions{ + Variant: &DoSignal_DoSignalActions_DoActionsInMain{ + DoActionsInMain: SingleActionSet(NewEmptyReturnResultAction()), + }, + }}, + }}, + }}, + }, + ExpectedOutput: ConvertToPayload(workflowID), + }) +} + +func (t *tpsExecutor) createNexusSignalWithStartAction(workflowID string, workflowInput *WorkflowInput) *Action { + return NexusOperation(&ExecuteNexusOperation{ + Endpoint: t.config.NexusEndpoint, + Input: &NexusOperationRequest{ + Action: &NexusOperationRequest_WorkflowAction{WorkflowAction: &NexusWorkflowAction{ + WorkflowId: workflowID, + StartOptions: &NexusWorkflowStartOptions{WorkflowInput: workflowInput}, + Action: &NexusWorkflowAction_Signal{Signal: &DoSignal{ + Variant: &DoSignal_DoSignalActions_{DoSignalActions: &DoSignal_DoSignalActions{ + Variant: &DoSignal_DoSignalActions_DoActions{ + DoActions: SingleActionSet(NewTimerAction(time.Millisecond)), + }, + }}, + WithStart: true, + }}, + }}, + }, + ExpectedOutput: ConvertToPayload(workflowID), + }) +} + +func (t *tpsExecutor) createNexusUpdateAction(workflowID string) *Action { + return NexusOperation(&ExecuteNexusOperation{ + Endpoint: t.config.NexusEndpoint, + Input: &NexusOperationRequest{ + Action: &NexusOperationRequest_WorkflowAction{WorkflowAction: &NexusWorkflowAction{ + WorkflowId: workflowID, + Action: &NexusWorkflowAction_Update{Update: &DoUpdate{ + Variant: &DoUpdate_DoActions{DoActions: &DoActionsUpdate{ + Variant: &DoActionsUpdate_DoActions{DoActions: SingleActionSet( + // The update handler's return value is itself encoded by the data converter + // before the server forwards it to the Nexus completion callback, so the value + // is wrapped twice here: the caller decodes the outer layer and compares the + // inner Payload against ExecuteNexusOperation.expected_output. + NewReturnResultAction(ConvertToPayload(ConvertToPayload(workflowID))), + )}, + }}, + }}, + }}, + }, + ExpectedOutput: ConvertToPayload(ConvertToPayload(workflowID)), + }) +} + func (t *tpsExecutor) createStandaloneNexusOperationAction(input *NexusOperationRequest) *Action { return ClientActivity(ClientActions(&ClientAction{ Variant: &ClientAction_DoStandaloneNexusOperation{ diff --git a/scenarios/throughput_stress_test.go b/scenarios/throughput_stress_test.go index a7f0b0c7..2485d1b3 100644 --- a/scenarios/throughput_stress_test.go +++ b/scenarios/throughput_stress_test.go @@ -3,6 +3,7 @@ package scenarios import ( "fmt" "math/rand" + "strings" "testing" "time" @@ -15,6 +16,7 @@ import ( "go.temporal.io/api/workflowservice/v1" "go.temporal.io/sdk/converter" "go.uber.org/zap" + "google.golang.org/protobuf/proto" ) func TestThroughputStress(t *testing.T) { @@ -147,11 +149,9 @@ func TestThroughputStressFeatureAutoEnable(t *testing.T) { require.Equal(t, 1, executor.Snapshot().(tpsState).CompletedIterations) } -func TestThroughputStressNexusStandaloneActivity(t *testing.T) { +func TestThroughputStressNexusStandaloneActivityActions(t *testing.T) { t.Parallel() - 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, @@ -160,95 +160,125 @@ func TestThroughputStressNexusStandaloneActivity(t *testing.T) { "history.enableCHASMCallbacks": true, })) env := workertest.SetupTestEnvironment(t, - workertest.WithExecutorTimeout(1*time.Minute), + workertest.WithExecutorTimeout(time.Minute), workertest.WithDevServer(server)) + type actionCounts struct { + inWorkflow int + standalone int + } + actionsCreated := make(chan []*ks.ActionSet, 1) + exec := newThroughputStressExecutor() + exec.onActionsCreated = func(actions []*ks.ActionSet) { + actionsCreated <- actions + } scenarioInfo := loadgen.ScenarioInfo{ - RunID: runID, - Configuration: loadgen.RunConfiguration{ - Iterations: 1, - }, + RunID: "tps-nsa-actions", + Configuration: loadgen.RunConfiguration{Iterations: 1}, Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{ IterFlag: "1", ContinueAsNewAfterIterFlag: "1", SleepTimeFlag: "1ms", VisibilityVerificationTimeoutFlag: "10s", NexusEnabledFlag: "true", + IncludeStandaloneNexusFlag: "true", IncludeNexusStandaloneActivityFlag: "true", }), } - executor := newThroughputStressExecutor() - _, err := env.RunExecutorTest(t, executor, scenarioInfo, clioptions.LangGo) - require.NoError(t, err, "Executor should complete successfully with nexus standalone activity enabled") - - require.True(t, executor.config.IncludeNexusStandaloneActivity, - "nexus standalone activity should be enabled") - require.Equal(t, 1, executor.Snapshot().(tpsState).CompletedIterations) -} - -func TestThroughputStressNexusStandaloneActivityActions(t *testing.T) { - t.Parallel() - - exec := newThroughputStressExecutor() - exec.config = &tpsConfig{ - InternalIterations: 1, - ContinueAsNewAfterIter: 0, - NexusEnabled: true, - NexusEndpoint: "test-endpoint", - IncludeStandaloneNexus: true, - IncludeNexusStandaloneActivity: true, - SleepTime: time.Millisecond, - RngSeed: 1, - } - exec.rng = rand.New(rand.NewSource(1)) + _, err := env.RunExecutorTest(t, exec, scenarioInfo, clioptions.LangGo) + require.NoError(t, err) - run := (&loadgen.ScenarioInfo{ - RunID: "tps-nsa-actions", - ExecutionID: "exec", - Logger: zap.NewNop().Sugar(), - }).NewRun(0) - - var inWorkflow, standalone bool - var walk func(actions []*ks.Action) - walk = func(actions []*ks.Action) { - for _, a := range actions { - if op := a.GetNexusOperation(); op.GetInput().GetStartActivity() != nil { - inWorkflow = true + var counts actionCounts + for _, actionSet := range <-actionsCreated { + walkActions(actionSet.GetActions(), func(action *ks.Action) { + if action.GetNexusOperation().GetInput().GetStartActivity() != nil { + counts.inWorkflow++ } // Find the nested standalone-Nexus client action. - if seq := a.GetExecActivity().GetClient().GetClientSequence(); seq != nil { - for _, set := range seq.GetActionSets() { - for _, ca := range set.GetActions() { - if sn := ca.GetDoStandaloneNexusOperation().GetOperation(); sn.GetInput().GetStartActivity() != nil { - standalone = true + if sequence := action.GetExecActivity().GetClient().GetClientSequence(); sequence != nil { + for _, clientActionSet := range sequence.GetActionSets() { + for _, clientAction := range clientActionSet.GetActions() { + operation := clientAction.GetDoStandaloneNexusOperation().GetOperation() + if operation.GetInput().GetStartActivity() != nil { + counts.standalone++ } } } } - if nested := a.GetNestedActionSet(); nested != nil { - walk(nested.GetActions()) - } - } - } - for _, set := range exec.createActions(run) { - walk(set.GetActions()) + }) } - - require.True(t, inWorkflow, - `expected an in-workflow Nexus start-activity action`) - require.True(t, standalone, - `expected a standalone Nexus start-activity client action`) + require.Equal(t, actionCounts{inWorkflow: 1, standalone: 1}, counts) } func TestThroughputStressNexusAttachSignalIsFireAndForget(t *testing.T) { actions := (&tpsExecutor{config: &tpsConfig{NexusEndpoint: "test-endpoint"}}). createNexusAttachCallbacksAction().GetNestedActionSet().GetActions() - require.NotNil(t, actions[1].GetSendSignal().GetAwaitableChoice().GetAbandon()) + require.NotNil(t, actions[1].GetNexusOperation().GetAwaitableChoice().GetAbandon()) require.NotNil(t, actions[2].GetAwaitPendingActions()) } +func TestThroughputStressNexusWorkflowActions(t *testing.T) { + t.Parallel() + + server := workertest.StartDevServer(t, workertest.WithDynamicConfig(map[string]any{ + "history.enableChasm": true, + "history.enableCHASMCallbacks": true, + "history.enableCHASMSignalBacklinks": true, + "history.enableUpdateCallbacks": true, + })) + env := workertest.SetupTestEnvironment(t, + workertest.WithExecutorTimeout(time.Minute), + workertest.WithDevServer(server)) + + actionsCreated := make(chan []*ks.ActionSet, 1) + exec := newThroughputStressExecutor() + exec.onActionsCreated = func(actions []*ks.ActionSet) { + actionsCreated <- actions + } + scenarioInfo := loadgen.ScenarioInfo{ + RunID: "nexus-workflow-actions-signal-with-start", + Configuration: loadgen.RunConfiguration{Iterations: 1}, + Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{ + IterFlag: "1", + NexusEnabledFlag: "true", + IncludeNexusWorkflowActionsFlag: "true", + SleepTimeFlag: "1ms", + VisibilityVerificationTimeoutFlag: "10s", + }), + } + + _, err := env.RunExecutorTest(t, exec, scenarioInfo, clioptions.LangGo) + require.NoError(t, err, scenarioInfo.RunID) + + var workflowActions []*ks.NexusWorkflowAction + for _, actionSet := range <-actionsCreated { + walkActions(actionSet.GetActions(), func(action *ks.Action) { + workflowAction := action.GetNexusOperation().GetInput().GetWorkflowAction() + if strings.Contains(workflowAction.GetWorkflowId(), "/nexus-workflow-") { + workflowActions = append(workflowActions, workflowAction) + } + }) + } + require.Len(t, workflowActions, 3) + + signalWithStartAction := workflowActions[0] + require.True(t, signalWithStartAction.GetSignal().GetWithStart()) + require.True(t, proto.Equal(&ks.WorkflowInput{}, signalWithStartAction.GetStartOptions().GetWorkflowInput())) + + updateAction := workflowActions[1] + require.NotNil(t, updateAction.GetUpdate()) + + signalAction := workflowActions[2] + require.NotNil(t, signalAction.GetSignal()) + require.False(t, signalAction.GetSignal().GetWithStart()) + require.True(t, proto.Equal( + ks.SingleActionSet(ks.NewEmptyReturnResultAction()), + signalAction.GetSignal().GetDoSignalActions().GetDoActionsInMain(), + )) +} + func TestThroughputStressConfigurePayload(t *testing.T) { t.Parallel() @@ -361,6 +391,25 @@ func TestThroughputStressConfigureExplicitStandaloneNexusRequiresNexusEnabled(t require.Contains(t, err.Error(), NexusEnabledFlag) } +func TestThroughputStressConfigureNexusWorkflowActionsRequireNexusEnabled(t *testing.T) { + t.Parallel() + + options, err := loadgen.GetScenario("throughput_stress").ResolveOptions(map[string]string{ + IncludeNexusWorkflowActionsFlag: "true", + NexusEnabledFlag: "false", + }) + require.NoError(t, err) + + err = newThroughputStressExecutor().Configure(loadgen.ScenarioInfo{ + RunID: "tps-nexus-workflow-action", + Options: options, + }) + + require.Error(t, err) + require.Contains(t, err.Error(), IncludeNexusWorkflowActionsFlag) + require.Contains(t, err.Error(), NexusEnabledFlag) +} + func TestThroughputStressConfigureInvalidPayload(t *testing.T) { t.Parallel() @@ -468,6 +517,15 @@ func TestThroughputStressOperatorCommandsAcrossRunsAndContinueAsNew(t *testing.T }, commandTypes) } +func walkActions(actions []*ks.Action, visit func(*ks.Action)) { + for _, action := range actions { + visit(action) + if nested := action.GetNestedActionSet(); nested != nil { + walkActions(nested.GetActions(), visit) + } + } +} + func standaloneActivityOperatorCommandsInConcurrentGroups(actions []*ks.Action) ( commands []*ks.DoStandaloneActivityOperatorCommands, ) {