Add standalone-activity support to throughput_stress (RE-353)

README.md

@@ -370,6 +370,13 @@ The throughput_stress scenario can generate Nexus load if the scenario is starte
   go run ./cmd/omes run-scenario-with-worker --scenario throughput_stress --language go --option nexus-endpoint=my-nexus-endpoint --run-id default-run-id
   ```
 
+#### Standalone activities
+
+The throughput_stress scenario can generate standalone-activity load (activities started outside
+any workflow context via `StartActivityExecution`) with `--option include-standalone-activity=true`.
+This requires server support for standalone activities (dynamic config `activity.enableStandalone`).
+Implemented for the Go, Python, TypeScript, .NET, Java, and Ruby workers.
+
 ### Fuzzer
 
 The fuzzer scenario makes use of the kitchen sink workflow (see below) to exercise a wide

clioptions/worker.go

@@ -51,6 +51,6 @@ func (m *WorkerOptions) FlagSetWithPrefix(prefix string) *pflag.FlagSet {
 	m.fs.IntVar(&m.ActivityPollerAutoscaleMax, prefix+"activity-poller-autoscale-max", 0, "Max for activity poller autoscaling (overrides max-concurrent-activity-pollers")
 	m.fs.IntVar(&m.WorkflowPollerAutoscaleMax, prefix+"workflow-poller-autoscale-max", 0, "Max for workflow poller autoscaling (overrides max-concurrent-workflow-pollers")
 	m.fs.Float64Var(&m.WorkerActivitiesPerSecond, prefix+"activities-per-second", 0, "Per-worker activity rate limit")
-	m.fs.BoolVar(&m.ErrOnUnimplemented, prefix+"err-on-unimplemented", false, "Fail on unimplemented actions (currently this only applies to concurrent client actions)")
+	m.fs.BoolVar(&m.ErrOnUnimplemented, prefix+"err-on-unimplemented", false, "Fail on unimplemented actions (e.g. concurrent client actions and standalone Nexus/activity operations) instead of skipping them")
 	return m.fs
 }

internal/workertest/env.go

@@ -68,6 +68,22 @@ type TestResult struct {
 	ObservedLogs *observer.ObservedLogs
 }
 
+// runExecutorConfig holds per-run options for RunExecutorTest.
+type runExecutorConfig struct {
+	errOnUnimplemented bool
+}
+
+type RunExecutorOption func(*runExecutorConfig)
+
+// WithErrOnUnimplemented starts the worker with --err-on-unimplemented, so that
+// actions a worker does not support fail loudly instead of being skipped. Use
+// it for tests that assert a feature is rejected by a given SDK.
+func WithErrOnUnimplemented() RunExecutorOption {
+	return func(c *runExecutorConfig) {
+		c.errOnUnimplemented = true
+	}
+}
+
 type TestEnvironment struct {
 	testEnvConfig
 	devServer         *devserver.Server
@@ -182,7 +198,12 @@ func (env *TestEnvironment) RunExecutorTest(
 	executor loadgen.Executor,
 	scenarioInfo loadgen.ScenarioInfo,
 	sdk clioptions.Language,
+	opts ...RunExecutorOption,
 ) (TestResult, error) {
+	var cfg runExecutorConfig
+	for _, opt := range opts {
+		opt(&cfg)
+	}
 	testLogger := zaptest.NewLogger(t).Core()
 	observeLogger, observedLogs := observer.New(zap.DebugLevel)
 	logger := zap.New(zapcore.NewTee(testLogger, observeLogger)).Sugar()
@@ -203,7 +224,7 @@ func (env *TestEnvironment) RunExecutorTest(
 	scenarioInfo.Namespace = testNamespace
 
 	taskQueueName := loadgen.TaskQueueForRun(scenarioInfo.RunID)
-	workerShutdownCh := env.workerPool.startWorker(testCtx, logger, sdk, taskQueueName, scenarioInfo)
+	workerShutdownCh := env.workerPool.startWorker(testCtx, logger, sdk, taskQueueName, scenarioInfo, cfg.errOnUnimplemented)
 
 	execErr := executor.Run(testCtx, scenarioInfo)
 

internal/workertest/workerpool.go

@@ -93,6 +93,7 @@ func (w *workerPool) startWorker(
 	sdk clioptions.Language,
 	taskQueueName string,
 	scenarioInfo loadgen.ScenarioInfo,
+	errOnUnimplemented bool,
 ) <-chan error {
 	workerDone := make(chan error, 1)
 
@@ -117,6 +118,9 @@ func (w *workerPool) startWorker(
 		}
 		runner.ClientOptions.FlagSet().Set("server-address", w.env.DevServerAddress())
 		runner.ClientOptions.FlagSet().Set("namespace", testNamespace)
+		if errOnUnimplemented {
+			runner.WorkerOptions.FlagSet().Set("worker-err-on-unimplemented", "true")
+		}
 		workerDone <- runner.Run(ctx, baseDir)
 	}()
 

loadgen/kitchen_sink_executor_test.go

@@ -72,6 +72,8 @@ func TestKitchenSink(t *testing.T) {
 		"nexusoperation.enableStandalone": true,
 		// Standalone Nexus system callbacks require CHASM callbacks.
 		"history.enableCHASMCallbacks": true,
+		// Enable StartActivityExecution for the standalone-activity subtest.
+		"activity.enableStandalone": true,
 	}))
 
 	// Default workflow execution timeout for tests
@@ -1034,6 +1036,33 @@ func TestKitchenSink(t *testing.T) {
 				ActivityTaskCompleted`),
 			expectedUnsupportedErrs: standaloneNexusUnsupportedSDKs,
 		},
+		{
+			name: "ExecActivity/Client/StandaloneActivity",
+			testInput: &TestInput{
+				WorkflowInput: &WorkflowInput{
+					InitialActions: ListActionSet(
+						ClientActivity(
+							ClientActions(&ClientAction{
+								Variant: &ClientAction_DoStandaloneActivity{
+									DoStandaloneActivity: &DoStandaloneActivity{
+										Activity: &ExecuteActivityAction{
+											ActivityType:        &ExecuteActivityAction_Noop{},
+											StartToCloseTimeout: &durationpb.Duration{Seconds: 5},
+											// TaskQueue filled by PrepareTestInput
+										},
+									},
+								},
+							}),
+							DefaultRemoteActivity,
+						),
+					),
+				},
+			},
+			historyMatcher: PartialHistoryMatcher(`
+				ActivityTaskScheduled {"activityType":{"name":"client"}}
+				ActivityTaskStarted
+				ActivityTaskCompleted`),
+		},
 		{
 			name: "UnsupportedAction",
 			testInput: &TestInput{
@@ -1100,6 +1129,10 @@ func testForSDK(
 	nexusEndpoint, err := env.CreateNexusEndpoint(t.Context(), scenarioInfo.RunID)
 	require.NoError(t, err, "Failed to create Nexus endpoint")
 
+	// Task queue this run's worker polls; standalone activities target it so the
+	// worker picks them up.
+	runTaskQueue := TaskQueueForRun(scenarioInfo.RunID)
+
 	executor := &KitchenSinkExecutor{
 		TestInput: tc.testInput,
 		PrepareTestInput: func(_ context.Context, _ ScenarioInfo, input *TestInput) error {
@@ -1115,6 +1148,9 @@ func testForSDK(
 									if sno := ca.GetDoStandaloneNexusOperation(); sno != nil && sno.Endpoint == "" {
 										sno.Endpoint = nexusEndpoint
 									}
+									if sa := ca.GetDoStandaloneActivity(); sa.GetActivity() != nil && sa.GetActivity().TaskQueue == "" {
+										sa.GetActivity().TaskQueue = runTaskQueue
+									}
 								}
 							}
 						}
@@ -1148,7 +1184,9 @@ func testUnsupportedFeature(
 		executor: executor,
 		sdk:      sdk,
 	}
-	_, execErr := env.RunExecutorTest(t, testExecutor, scenarioInfo, sdk)
+	// Run the worker in strict mode so unsupported actions fail loudly rather
+	// than being skipped, which is what this test asserts.
+	_, execErr := env.RunExecutorTest(t, testExecutor, scenarioInfo, sdk, WithErrOnUnimplemented())
 
 	require.Errorf(t, execErr, "SDK %s should fail for unsupported feature", sdk)
 	require.NotEmptyf(t, expectedErr, "invalid test case: expectedUnsupportedErrs must be set for SDK %s if the feature is unsupported", sdk)

loadgen/kitchensink/client_action_executor.go

@@ -131,6 +131,8 @@ func (e *ClientActionsExecutor) executeClientAction(ctx context.Context, action
 		return err
 	} else if sano := action.GetDoStandaloneNexusOperation(); sano != nil {
 		return e.executeStandaloneNexusOperation(ctx, sano)
+	} else if sa := action.GetDoStandaloneActivity(); sa != nil {
+		return e.executeStandaloneActivity(ctx, sa)
 	} else {
 		return fmt.Errorf("client action must be set")
 	}
@@ -225,3 +227,26 @@ func (e *ClientActionsExecutor) executeStandaloneNexusOperation(ctx context.Cont
 	}
 	return nil
 }
+
+func (e *ClientActionsExecutor) executeStandaloneActivity(ctx context.Context, sa *DoStandaloneActivity) error {
+	act := sa.GetActivity()
+	if act == nil {
+		return fmt.Errorf("DoStandaloneActivity.activity is required")
+	}
+
+	actType, args := ActivityNameAndArgs(act)
+
+	handle, err := e.Client.ExecuteActivity(ctx, client.StartActivityOptions{
+		ID:                     fmt.Sprintf("standalone-activity-%s-%s", e.WorkflowOptions.ID, uuid.NewString()),
+		TaskQueue:              act.TaskQueue,
+		ScheduleToCloseTimeout: act.ScheduleToCloseTimeout.AsDuration(),
+		ScheduleToStartTimeout: act.ScheduleToStartTimeout.AsDuration(),
+		StartToCloseTimeout:    act.StartToCloseTimeout.AsDuration(),
+		HeartbeatTimeout:       act.HeartbeatTimeout.AsDuration(),
+		RetryPolicy:            ConvertFromPBRetryPolicy(act.RetryPolicy),
+	}, actType, args...)
+	if err != nil {
+		return fmt.Errorf("failed to start standalone activity: %w", err)
+	}
+	return handle.Get(ctx, nil)
+}

loadgen/kitchensink/helpers.go

@@ -7,13 +7,58 @@ import (
 	"go.temporal.io/api/common/v1"
 	"go.temporal.io/api/failure/v1"
 	"go.temporal.io/sdk/converter"
+	"go.temporal.io/sdk/temporal"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
 // Using human-readable JSON encoding for payloads to aid with debugging.
 var jsonPayloadConverter = converter.NewProtoJSONPayloadConverter()
 
+// ActivityNameAndArgs maps an ExecuteActivityAction's activity variant to the
+// registered activity name and its args. Shared by the workflow-scheduled path
+// (worker launchActivity) and the standalone-activity client path so the two
+// stay in sync. Unrecognized variants fall back to "noop".
+func ActivityNameAndArgs(act *ExecuteActivityAction) (string, []any) {
+	if delay := act.GetDelay(); delay != nil {
+		return "delay", []any{delay.AsDuration()}
+	} else if payload := act.GetPayload(); payload != nil {
+		inputData := make([]byte, payload.BytesToReceive)
+		for i := range inputData {
+			inputData[i] = byte(i % 256)
+		}
+		return "payload", []any{inputData, payload.BytesToReturn}
+	} else if client := act.GetClient(); client != nil {
+		return "client", []any{client}
+	} else if retryable := act.GetRetryableError(); retryable != nil {
+		return "retryable_error", []any{retryable}
+	} else if timeout := act.GetTimeout(); timeout != nil {
+		return "timeout", []any{timeout}
+	} else if heartbeat := act.GetHeartbeat(); heartbeat != nil {
+		return "heartbeat", []any{heartbeat}
+	}
+	return "noop", nil
+}
+
+// ConvertFromPBRetryPolicy converts a proto RetryPolicy into an SDK RetryPolicy.
+func ConvertFromPBRetryPolicy(retryPolicy *common.RetryPolicy) *temporal.RetryPolicy {
+	if retryPolicy == nil {
+		return nil
+	}
+	p := temporal.RetryPolicy{
+		BackoffCoefficient:     retryPolicy.BackoffCoefficient,
+		MaximumAttempts:        retryPolicy.MaximumAttempts,
+		NonRetryableErrorTypes: retryPolicy.NonRetryableErrorTypes,
+	}
+	if v := retryPolicy.MaximumInterval; v != nil {
+		p.MaximumInterval = v.AsDuration()
+	}
+	if v := retryPolicy.InitialInterval; v != nil {
+		p.InitialInterval = v.AsDuration()
+	}
+	return &p
+}
+
 type ActionFactory[T any] func(*T) *Action
 
 func SingleActionSet(actions ...*Action) *ActionSet {