From a45c80ff225c51932342f403457a8cb4a757353a Mon Sep 17 00:00:00 2001 From: lilydoar Date: Tue, 20 Oct 2026 12:04:18 -0700 Subject: [PATCH 1/4] Add healthy heartbeating activities to throughput_stress throughput_stress only heartbeats in its retry scenario, where the first attempt deliberately times out. Nothing exercises activities that stay alive by heartbeating, so load runs cover the heartbeat failure path but not sustained successful heartbeats. healthy-heartbeat-activities adds that many remote activities to each internal iteration. Each heartbeats for healthy-heartbeat-duration with a healthy-heartbeat-timeout heartbeat timeout and a single attempt, so a heartbeat timeout fails the run. Off by default. Uses the Go worker's existing heartbeat activity, so no worker change is needed. --- scenarios/throughput_stress.go | 54 +++++++++ scenarios/throughput_stress_heartbeat_test.go | 103 ++++++++++++++++++ 2 files changed, 157 insertions(+) create mode 100644 scenarios/throughput_stress_heartbeat_test.go diff --git a/scenarios/throughput_stress.go b/scenarios/throughput_stress.go index 8ae3e8f5..f6f890bf 100644 --- a/scenarios/throughput_stress.go +++ b/scenarios/throughput_stress.go @@ -69,6 +69,13 @@ 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. Default 0 disables them. + 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" ) type tpsState struct { @@ -117,6 +124,9 @@ type tpsConfig struct { IncludeNexusStandaloneActivity bool IncludeNexusWorkflowActions bool Payload *loadgen.PayloadConfig + HealthyHeartbeatActivities int + HealthyHeartbeatDuration time.Duration + HealthyHeartbeatTimeout time.Duration } type tpsExecutor struct { @@ -155,6 +165,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 @ to read from a file.") + o.Int(HealthyHeartbeatActivitiesFlag, 0, "Healthy heartbeating activities per internal iteration (Go worker only).") + o.Duration(HealthyHeartbeatDurationFlag, 30*time.Second, "How long each healthy heartbeat activity runs.") + o.Duration(HealthyHeartbeatTimeoutFlag, 10*time.Second, "Heartbeat timeout of each healthy heartbeat activity.") }, ExecutorFn: func() loadgen.Executor { return newThroughputStressExecutor() }, }) @@ -270,6 +283,19 @@ 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) + } + t.config = config t.rng = rand.New(rand.NewSource(config.RngSeed)) return nil @@ -709,6 +735,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 +1209,29 @@ func (t *tpsExecutor) createStandaloneNexusOperationAction(input *NexusOperation }), DefaultRemoteActivity) } +// healthyHeartbeatActivity heartbeats for its whole duration and is not +// retried, so a heartbeat timeout on a healthy activity fails the run. It +// heartbeats more often than the SDK's heartbeat send window, so that window +// sets the RPC rate. +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{ diff --git a/scenarios/throughput_stress_heartbeat_test.go b/scenarios/throughput_stress_heartbeat_test.go new file mode 100644 index 00000000..b441d3e9 --- /dev/null +++ b/scenarios/throughput_stress_heartbeat_test.go @@ -0,0 +1,103 @@ +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) +} From 81218eeef84a96752f92fc21d0edbf1383b4231a Mon Sep 17 00:00:00 2001 From: lilydoar Date: Tue, 20 Oct 2026 13:27:45 -0700 Subject: [PATCH 2/4] Name the healthy heartbeat option defaults --- scenarios/throughput_stress.go | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/scenarios/throughput_stress.go b/scenarios/throughput_stress.go index f6f890bf..6dbb9cb4 100644 --- a/scenarios/throughput_stress.go +++ b/scenarios/throughput_stress.go @@ -70,7 +70,7 @@ const ( // 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. Default 0 disables them. + // activities each internal iteration runs. HealthyHeartbeatActivitiesFlag = "healthy-heartbeat-activities" // HealthyHeartbeatDurationFlag is how long each healthy heartbeat activity runs. HealthyHeartbeatDurationFlag = "healthy-heartbeat-duration" @@ -78,6 +78,13 @@ const ( HealthyHeartbeatTimeoutFlag = "healthy-heartbeat-timeout" ) +const ( + // DefaultHealthyHeartbeatActivities disables healthy heartbeat activities. + DefaultHealthyHeartbeatActivities = 0 + DefaultHealthyHeartbeatDuration = 30 * time.Second + DefaultHealthyHeartbeatTimeout = 10 * time.Second +) + type tpsState struct { // CompletedIterations is the number of iteration that have been completed. CompletedIterations int `json:"completedIterations"` @@ -165,9 +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 @ to read from a file.") - o.Int(HealthyHeartbeatActivitiesFlag, 0, "Healthy heartbeating activities per internal iteration (Go worker only).") - o.Duration(HealthyHeartbeatDurationFlag, 30*time.Second, "How long each healthy heartbeat activity runs.") - o.Duration(HealthyHeartbeatTimeoutFlag, 10*time.Second, "Heartbeat timeout of each healthy heartbeat activity.") + 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() }, }) From 8a3280ec6262c94d83c9f2edf90299b98a772699 Mon Sep 17 00:00:00 2001 From: lilydoar Date: Mon, 26 Oct 2026 14:11:40 -0700 Subject: [PATCH 3/4] Reject overflowing healthy-heartbeat activity deadlines --- scenarios/throughput_stress.go | 9 ++++--- scenarios/throughput_stress_heartbeat_test.go | 26 +++++++++++++++++++ 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/scenarios/throughput_stress.go b/scenarios/throughput_stress.go index 6dbb9cb4..bdfefa5e 100644 --- a/scenarios/throughput_stress.go +++ b/scenarios/throughput_stress.go @@ -303,6 +303,10 @@ func (t *tpsExecutor) Configure(info loadgen.ScenarioInfo) error { 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 @@ -1216,10 +1220,7 @@ func (t *tpsExecutor) createStandaloneNexusOperationAction(input *NexusOperation }), DefaultRemoteActivity) } -// healthyHeartbeatActivity heartbeats for its whole duration and is not -// retried, so a heartbeat timeout on a healthy activity fails the run. It -// heartbeats more often than the SDK's heartbeat send window, so that window -// sets the RPC rate. +// A missed heartbeat must fail the load. func healthyHeartbeatActivity(duration, heartbeatTimeout time.Duration) *Action { return &Action{ Variant: &Action_ExecActivity{ diff --git a/scenarios/throughput_stress_heartbeat_test.go b/scenarios/throughput_stress_heartbeat_test.go index b441d3e9..39aa3577 100644 --- a/scenarios/throughput_stress_heartbeat_test.go +++ b/scenarios/throughput_stress_heartbeat_test.go @@ -101,3 +101,29 @@ func TestThroughputStressHealthyHeartbeatKeepsActivityAlive(t *testing.T) { }, 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()) + } + }) + } +} From 3ee96632c4095d741fd10ba696ec0b8b58a952a0 Mon Sep 17 00:00:00 2001 From: lilydoar Date: Tue, 27 Oct 2026 13:58:10 -0700 Subject: [PATCH 4/4] Clarify why healthy heartbeat activities do not retry --- scenarios/throughput_stress.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scenarios/throughput_stress.go b/scenarios/throughput_stress.go index bdfefa5e..150b61bd 100644 --- a/scenarios/throughput_stress.go +++ b/scenarios/throughput_stress.go @@ -1220,7 +1220,7 @@ func (t *tpsExecutor) createStandaloneNexusOperationAction(input *NexusOperation }), DefaultRemoteActivity) } -// A missed heartbeat must fail the load. +// Keep heartbeat timeouts observable rather than masking them with activity retries. func healthyHeartbeatActivity(duration, heartbeatTimeout time.Duration) *Action { return &Action{ Variant: &Action_ExecActivity{