Add healthy heartbeating activities to throughput_stress

scenarios/throughput_stress.go

@@ -69,6 +69,20 @@ const (
 	// PayloadDistributionJsonFlag is a JSON string (or @file) configuring a weighted
 	// activity payload-size distribution. See loadgen.PayloadConfig for details.
 	PayloadDistributionJsonFlag = "payload-distribution-json"
+	// HealthyHeartbeatActivitiesFlag is how many healthy heartbeating remote
+	// activities each internal iteration runs.
+	HealthyHeartbeatActivitiesFlag = "healthy-heartbeat-activities"
+	// HealthyHeartbeatDurationFlag is how long each healthy heartbeat activity runs.
+	HealthyHeartbeatDurationFlag = "healthy-heartbeat-duration"
+	// HealthyHeartbeatTimeoutFlag is the heartbeat timeout of each healthy heartbeat activity.
+	HealthyHeartbeatTimeoutFlag = "healthy-heartbeat-timeout"
+)
+
+const (
+	// DefaultHealthyHeartbeatActivities disables healthy heartbeat activities.
+	DefaultHealthyHeartbeatActivities = 0
+	DefaultHealthyHeartbeatDuration   = 30 * time.Second
+	DefaultHealthyHeartbeatTimeout    = 10 * time.Second
 )
 
 type tpsState struct {
@@ -117,6 +131,9 @@ type tpsConfig struct {
 	IncludeNexusStandaloneActivity            bool
 	IncludeNexusWorkflowActions               bool
 	Payload                                   *loadgen.PayloadConfig
+	HealthyHeartbeatActivities                int
+	HealthyHeartbeatDuration                  time.Duration
+	HealthyHeartbeatTimeout                   time.Duration
 }
 
 type tpsExecutor struct {
@@ -155,6 +172,9 @@ func init() {
 			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, update, and query a workflow (Go worker only).")
 			o.String(PayloadDistributionJsonFlag, "", "JSON payload-size distribution; use @<file> to read from a file.")
+			o.Int(HealthyHeartbeatActivitiesFlag, DefaultHealthyHeartbeatActivities, "Healthy heartbeating activities per internal iteration (Go worker only).")
+			o.Duration(HealthyHeartbeatDurationFlag, DefaultHealthyHeartbeatDuration, "How long each healthy heartbeat activity runs.")
+			o.Duration(HealthyHeartbeatTimeoutFlag, DefaultHealthyHeartbeatTimeout, "Heartbeat timeout of each healthy heartbeat activity.")
 		},
 		ExecutorFn: func() loadgen.Executor { return newThroughputStressExecutor() },
 	})
@@ -270,6 +290,23 @@ func (t *tpsExecutor) Configure(info loadgen.ScenarioInfo) error {
 		}
 	}
 
+	config.HealthyHeartbeatActivities = info.OptionInt(HealthyHeartbeatActivitiesFlag)
+	if config.HealthyHeartbeatActivities < 0 {
+		return fmt.Errorf("%s must be non-negative, got %d", HealthyHeartbeatActivitiesFlag, config.HealthyHeartbeatActivities)
+	}
+	config.HealthyHeartbeatDuration = info.OptionDuration(HealthyHeartbeatDurationFlag)
+	if config.HealthyHeartbeatDuration <= 0 {
+		return fmt.Errorf("%s must be positive, got %v", HealthyHeartbeatDurationFlag, config.HealthyHeartbeatDuration)
+	}
+	config.HealthyHeartbeatTimeout = info.OptionDuration(HealthyHeartbeatTimeoutFlag)
+	if config.HealthyHeartbeatTimeout <= 0 {
+		return fmt.Errorf("%s must be positive, got %v", HealthyHeartbeatTimeoutFlag, config.HealthyHeartbeatTimeout)
+	}
+
+	if config.HealthyHeartbeatDuration > time.Duration(1<<63-1)-config.HealthyHeartbeatTimeout {
+		return fmt.Errorf("%s plus %s exceeds the maximum duration", HealthyHeartbeatDurationFlag, HealthyHeartbeatTimeoutFlag)
+	}
+
 	t.config = config
 	t.rng = rand.New(rand.NewSource(config.RngSeed))
 	return nil
@@ -709,6 +746,11 @@ func (t *tpsExecutor) createActionsChunk(
 			)
 		}
 
+		for range t.config.HealthyHeartbeatActivities {
+			asyncActions = append(asyncActions,
+				healthyHeartbeatActivity(t.config.HealthyHeartbeatDuration, t.config.HealthyHeartbeatTimeout))
+		}
+
 		// Add sleep activities, if configured.
 		// It simulates custom traffic patterns by generating activities that sleep
 		// for a specified duration, with optional priority and fairness keys.
@@ -1178,6 +1220,26 @@ func (t *tpsExecutor) createStandaloneNexusOperationAction(input *NexusOperation
 	}), DefaultRemoteActivity)
 }
 
+// Keep heartbeat timeouts observable rather than masking them with activity retries.
+func healthyHeartbeatActivity(duration, heartbeatTimeout time.Duration) *Action {
+	return &Action{
+		Variant: &Action_ExecActivity{
+			ExecActivity: &ExecuteActivityAction{
+				ActivityType: &ExecuteActivityAction_Heartbeat{
+					Heartbeat: &ExecuteActivityAction_HeartbeatTimeoutActivity{
+						SuccessDuration:   durationpb.New(duration),
+						HeartbeatInterval: durationpb.New(heartbeatTimeout / 4),
+					},
+				},
+				StartToCloseTimeout: durationpb.New(duration + heartbeatTimeout),
+				HeartbeatTimeout:    durationpb.New(heartbeatTimeout),
+				RetryPolicy:         &common.RetryPolicy{MaximumAttempts: 1},
+				Locality:            &ExecuteActivityAction_Remote{Remote: &RemoteActivityOptions{}},
+			},
+		},
+	}
+}
+
 func (t *tpsExecutor) createStandaloneActivityAction(taskQueue string, rng *rand.Rand) *Action {
 	size := int32(t.samplePayloadSize(rng))
 	return ClientActivity(ClientActions(&ClientAction{

scenarios/throughput_stress_heartbeat_test.go

@@ -0,0 +1,129 @@
+package scenarios
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/temporalio/omes/clioptions"
+	"github.com/temporalio/omes/internal/workertest"
+	"github.com/temporalio/omes/loadgen"
+	ks "github.com/temporalio/omes/loadgen/kitchensink"
+	"go.uber.org/zap"
+)
+
+func healthyHeartbeatActions(t *testing.T, options map[string]string) []*ks.ExecuteActivityAction {
+	t.Helper()
+	executor := newThroughputStressExecutor()
+	info := loadgen.ScenarioInfo{
+		RunID:       "tps-healthy-heartbeat",
+		ExecutionID: "exec",
+		Logger:      zap.NewNop().Sugar(),
+		Options:     loadgen.MustResolveScenarioOptions("throughput_stress", options),
+	}
+	require.NoError(t, executor.Configure(info))
+
+	var activities []*ks.ExecuteActivityAction
+	walkActions(executor.createActions(info.NewRun(1))[0].GetActions(), func(action *ks.Action) {
+		activity := action.GetExecActivity()
+		if heartbeat := activity.GetHeartbeat(); heartbeat != nil && heartbeat.GetFailAttempts() == 0 {
+			activities = append(activities, activity)
+		}
+	})
+	return activities
+}
+
+func TestThroughputStressHealthyHeartbeatActivities(t *testing.T) {
+	t.Parallel()
+
+	t.Run("off by default", func(t *testing.T) {
+		require.Empty(t, healthyHeartbeatActions(t, map[string]string{NexusEnabledFlag: "false"}))
+	})
+
+	t.Run("configured per internal iteration", func(t *testing.T) {
+		activities := healthyHeartbeatActions(t, map[string]string{
+			NexusEnabledFlag:               "false",
+			IterFlag:                       "3",
+			HealthyHeartbeatActivitiesFlag: "2",
+			HealthyHeartbeatDurationFlag:   "20s",
+			HealthyHeartbeatTimeoutFlag:    "8s",
+		})
+		require.Len(t, activities, 6)
+		for _, activity := range activities {
+			require.Equal(t, 20*time.Second, activity.GetHeartbeat().GetSuccessDuration().AsDuration())
+			require.Equal(t, 2*time.Second, activity.GetHeartbeat().GetHeartbeatInterval().AsDuration())
+			require.Equal(t, 8*time.Second, activity.GetHeartbeatTimeout().AsDuration())
+			require.Equal(t, 28*time.Second, activity.GetStartToCloseTimeout().AsDuration())
+			require.Equal(t, int32(1), activity.GetRetryPolicy().GetMaximumAttempts())
+			require.NotNil(t, activity.GetRemote())
+		}
+	})
+}
+
+func TestThroughputStressConfigureInvalidHealthyHeartbeat(t *testing.T) {
+	t.Parallel()
+
+	for flag, value := range map[string]string{
+		HealthyHeartbeatActivitiesFlag: "-1",
+		HealthyHeartbeatDurationFlag:   "0s",
+		HealthyHeartbeatTimeoutFlag:    "0s",
+	} {
+		t.Run(flag, func(t *testing.T) {
+			err := newThroughputStressExecutor().Configure(loadgen.ScenarioInfo{
+				RunID:   "tps-invalid-healthy-heartbeat",
+				Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{flag: value}),
+			})
+			require.ErrorContains(t, err, flag)
+		})
+	}
+}
+
+// The activity outlives its heartbeat timeout and is not retried, so the run
+// only succeeds if the Go worker keeps heartbeating it.
+func TestThroughputStressHealthyHeartbeatKeepsActivityAlive(t *testing.T) {
+	t.Parallel()
+
+	runID := fmt.Sprintf("tps-healthy-heartbeat-%d", time.Now().UnixNano())
+	env := workertest.SetupTestEnvironment(t, workertest.WithExecutorTimeout(time.Minute))
+	_, err := env.RunExecutorTest(t, newThroughputStressExecutor(), loadgen.ScenarioInfo{
+		RunID:         runID,
+		Configuration: loadgen.RunConfiguration{Iterations: 1},
+		Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{
+			IterFlag:                          "1",
+			NexusEnabledFlag:                  "false",
+			SleepTimeFlag:                     "1ms",
+			VisibilityVerificationTimeoutFlag: "10s",
+			HealthyHeartbeatActivitiesFlag:    "1",
+			HealthyHeartbeatDurationFlag:      "6s",
+			HealthyHeartbeatTimeoutFlag:       "3s",
+		}),
+	}, clioptions.LangGo)
+	require.NoError(t, err)
+}
+
+func TestThroughputStressHealthyHeartbeatDurationOverflow(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		duration time.Duration
+		invalid  bool
+	}{
+		{name: "largest representable sum", duration: time.Duration(1<<63-1) - time.Second},
+		{name: "sum overflows", duration: time.Duration(1<<63 - 1), invalid: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			info := loadgen.ScenarioInfo{Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{
+				NexusEnabledFlag:             "false",
+				HealthyHeartbeatDurationFlag: tc.duration.String(),
+				HealthyHeartbeatTimeoutFlag:  time.Second.String(),
+			})}
+			err := newThroughputStressExecutor().Configure(info)
+			if tc.invalid {
+				require.ErrorContains(t, err, HealthyHeartbeatDurationFlag)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, time.Duration(1<<63-1), healthyHeartbeatActivity(tc.duration, time.Second).GetExecActivity().GetStartToCloseTimeout().AsDuration())
+			}
+		})
+	}
+}