Exercise Nexus workflow messaging in throughput stress

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.

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

loadgen/generic_executor_test.go

@@ -37,14 +37,18 @@ 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{
 		MetricsHandler: client.MetricsNopHandler,
 		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)
 			},
 		})