From 20c462b0cfbd3ff24cd618f96b168efd6521b9cd Mon Sep 17 00:00:00 2001 From: Kundan <281732484+ks-temporal@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:26:26 -0700 Subject: [PATCH] feat(activity): add `activity reset --clear-heartbeat-details` Resurrects the `--reset-heartbeats` flag removed from `activity reset` in PR #1092, as a new flag `--clear-heartbeat-details`. It controls the existing `reset_heartbeat` field on both the single-activity request and the batch operation, and the proto field is unchanged as `ResetHeartbeat`. resets via `--query` always cleared heartbeat details, while the single-activity did not, as the default is to not clear the details. Both paths are now controlled by this flag. Rename `#reset-heartbeats` anchor to `#clear-heartbeat-details` for consistency. Tests cover the flag on both single activity and batch paths. --- internal/temporalcli/commands.activity.go | 3 +- .../temporalcli/commands.activity_test.go | 166 ++++++++++++++++++ internal/temporalcli/commands.gen.go | 6 +- internal/temporalcli/commands.yaml | 13 +- 4 files changed, 181 insertions(+), 7 deletions(-) diff --git a/internal/temporalcli/commands.activity.go b/internal/temporalcli/commands.activity.go index cb572995a..62d7eef5b 100644 --- a/internal/temporalcli/commands.activity.go +++ b/internal/temporalcli/commands.activity.go @@ -1184,6 +1184,7 @@ func (c *TemporalActivityResetCommand) run(cctx *CommandContext, args []string) RunId: c.RunId, Identity: c.Parent.Identity, KeepPaused: c.KeepPaused, + ResetHeartbeat: c.ClearHeartbeatDetails, Jitter: durationpb.New(c.Jitter.Duration()), RestoreOriginalOptions: c.RestoreOriginalOptions, ResourceId: resourceID, @@ -1206,7 +1207,7 @@ func (c *TemporalActivityResetCommand) run(cctx *CommandContext, args []string) } else { // batch operation resetActivitiesOperation := &batch.BatchOperationResetActivities{ Identity: c.Parent.Identity, - ResetHeartbeat: true, + ResetHeartbeat: c.ClearHeartbeatDetails, KeepPaused: c.KeepPaused, Jitter: durationpb.New(c.Jitter.Duration()), RestoreOriginalOptions: c.RestoreOriginalOptions, diff --git a/internal/temporalcli/commands.activity_test.go b/internal/temporalcli/commands.activity_test.go index 3a29644a3..0f670d49b 100644 --- a/internal/temporalcli/commands.activity_test.go +++ b/internal/temporalcli/commands.activity_test.go @@ -10,6 +10,7 @@ import ( "time" "github.com/google/uuid" + "go.temporal.io/api/common/v1" "go.temporal.io/api/enums/v1" "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" @@ -672,8 +673,173 @@ func (s *SharedServerSuite) TestActivityReset_ActivityIdWithQueryStartsBatch() { s.True(startedBatch.Load(), "a batch operation should have been started") } +// TestActivityReset_ClearHeartbeatDetailsFlag asserts --clear-heartbeat-details +// maps onto the request's ResetHeartbeat field for a single activity, and that +// heartbeat details are left alone when the flag is absent. +func (s *SharedServerSuite) TestActivityReset_ClearHeartbeatDetailsFlag() { + run := s.waitActivityStarted() + + var lastRequestLock sync.Mutex + var resetRequest *workflowservice.ResetActivityExecutionRequest + s.CommandHarness.Options.AdditionalClientGRPCDialOptions = append( + s.CommandHarness.Options.AdditionalClientGRPCDialOptions, + grpc.WithChainUnaryInterceptor(func( + ctx context.Context, + method string, req, reply any, + cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption, + ) error { + lastRequestLock.Lock() + if r, ok := req.(*workflowservice.ResetActivityExecutionRequest); ok { + resetRequest = r + } + lastRequestLock.Unlock() + return invoker(ctx, method, req, reply, cc, opts...) + }), + ) + + tests := []struct { + name string + extra []string + expected bool + }{ + {name: "flag omitted", expected: false}, + {name: "flag set", extra: []string{"--clear-heartbeat-details"}, expected: true}, + } + for _, tc := range tests { + lastRequestLock.Lock() + resetRequest = nil + lastRequestLock.Unlock() + + args := append([]string{"--activity-id", activityId}, tc.extra...) + res := sendActivityCommand("reset", run, s, args...) + s.NoError(res.Err, tc.name) + + lastRequestLock.Lock() + req := resetRequest + lastRequestLock.Unlock() + s.NotNil(req, tc.name) + s.Equal(tc.expected, req.GetResetHeartbeat(), tc.name) + } +} + +// TestActivityReset_ClearHeartbeatDetailsFlagBatch is the batch (--query) +// counterpart: the flag drives ResetHeartbeat on the batch operation, which is +// false unless asked for. +func (s *SharedServerSuite) TestActivityReset_ClearHeartbeatDetailsFlagBatch() { + run := s.waitActivityStarted() + + var lastRequestLock sync.Mutex + var startBatchRequest *workflowservice.StartBatchOperationRequest + s.CommandHarness.Options.AdditionalClientGRPCDialOptions = append( + s.CommandHarness.Options.AdditionalClientGRPCDialOptions, + grpc.WithChainUnaryInterceptor(func( + ctx context.Context, + method string, req, reply any, + cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption, + ) error { + lastRequestLock.Lock() + if r, ok := req.(*workflowservice.StartBatchOperationRequest); ok { + startBatchRequest = r + } + lastRequestLock.Unlock() + return invoker(ctx, method, req, reply, cc, opts...) + }), + ) + + tests := []struct { + name string + extra []string + expected bool + }{ + {name: "flag omitted", expected: false}, + {name: "flag set", extra: []string{"--clear-heartbeat-details"}, expected: true}, + } + for _, tc := range tests { + lastRequestLock.Lock() + startBatchRequest = nil + lastRequestLock.Unlock() + + args := []string{ + "activity", "reset", + "--query", fmt.Sprintf("WorkflowId = '%s'", run.GetID()), + "--yes", + "--address", s.Address(), + } + res := s.Execute(append(args, tc.extra...)...) + s.NoError(res.Err, tc.name) + + lastRequestLock.Lock() + req := startBatchRequest + lastRequestLock.Unlock() + s.NotNil(req, tc.name) + s.Equal(tc.expected, req.GetResetActivitiesOperation().GetResetHeartbeat(), tc.name) + } +} + +// TestActivityReset_ClearHeartbeatDetailsEffect checks the end-to-end effect on +// the server: the recorded heartbeat details survive a plain reset and are gone +// after a reset with --clear-heartbeat-details. +func (s *SharedServerSuite) TestActivityReset_ClearHeartbeatDetailsEffect() { + var recordHeartbeat atomic.Bool + recordHeartbeat.Store(true) + s.Worker().OnDevActivity(func(ctx context.Context, a any) (any, error) { + // Only the first attempt records heartbeat details; later attempts fail + // without heartbeating, so nothing re-populates what a reset clears. + if recordHeartbeat.Load() { + activity.RecordHeartbeat(ctx, "heartbeat-details") + } + return nil, fmt.Errorf("intentional failure to keep the activity retrying") + }) + s.Worker().OnDevWorkflow(func(ctx workflow.Context, a any) (any, error) { + // A long retry interval parks the activity in retry backoff, where a + // reset applies immediately. + ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + ActivityID: activityId, + StartToCloseTimeout: 1 * time.Minute, + RetryPolicy: &temporal.RetryPolicy{ + InitialInterval: 30 * time.Second, + MaximumAttempts: 0, + }, + }) + var res any + err := workflow.ExecuteActivity(ctx, DevActivity).Get(ctx, &res) + return res, err + }) + + run := waitWorkflowStarted(s) + s.Eventually(func() bool { + return len(s.pendingActivityHeartbeatDetails(run)) > 0 + }, 10*time.Second, 200*time.Millisecond, "activity should have recorded heartbeat details") + recordHeartbeat.Store(false) + + // Without the flag, the reset leaves the heartbeat details in place. + res := sendActivityCommand("reset", run, s, "--activity-id", activityId) + s.NoError(res.Err) + s.Never(func() bool { + return len(s.pendingActivityHeartbeatDetails(run)) == 0 + }, 2*time.Second, 200*time.Millisecond, "heartbeat details should survive a plain reset") + + // With the flag, they are cleared. + res = sendActivityCommand("reset", run, s, "--activity-id", activityId, "--clear-heartbeat-details") + s.NoError(res.Err) + s.Eventually(func() bool { + return len(s.pendingActivityHeartbeatDetails(run)) == 0 + }, 10*time.Second, 200*time.Millisecond, "--clear-heartbeat-details should clear the heartbeat details") +} + // Test helpers +func (s *SharedServerSuite) pendingActivityHeartbeatDetails(run client.WorkflowRun) []*common.Payload { + resp, err := s.Client.DescribeWorkflowExecution(s.Context, run.GetID(), run.GetRunID()) + s.NoError(err) + for _, act := range resp.GetPendingActivities() { + if act.GetActivityId() == activityId { + return act.GetHeartbeatDetails().GetPayloads() + } + } + return nil +} + func (s *SharedServerSuite) waitActivityStarted() client.WorkflowRun { s.Worker().OnDevActivity(func(ctx context.Context, a any) (any, error) { time.Sleep(0xFFFF * time.Hour) diff --git a/internal/temporalcli/commands.gen.go b/internal/temporalcli/commands.gen.go index 0165f6e30..32c303f20 100644 --- a/internal/temporalcli/commands.gen.go +++ b/internal/temporalcli/commands.gen.go @@ -871,6 +871,7 @@ type TemporalActivityResetCommand struct { SingleActivityOrBatchOptions ActivityId string KeepPaused bool + ClearHeartbeatDetails bool Jitter cliext.FlagDuration RestoreOriginalOptions bool } @@ -882,13 +883,14 @@ func NewTemporalActivityResetCommand(cctx *CommandContext, parent *TemporalActiv s.Command.Use = "reset [flags]" s.Command.Short = "Reset an Activity" if hasHighlighting { - s.Command.Long = "Reset an activity.\nThis restarts the activity as if it were first being scheduled: the\nattempt count returns to one, its per-attempt timeouts are re-armed, and\nits heartbeat details are cleared.\n\nIf the activity may be executing (i.e. it has not yet timed out), the\nreset will take effect the next time it fails, heartbeats, or times out.\nIf is waiting for a retry (i.e. has failed or timed out), the reset\nwill apply immediately.\n\nIf the activity is already paused, it will be unpaused by default.\nYou can specify \x1b[1m--keep-paused\x1b[0m to prevent this.\n\nIf the activity is paused and the \x1b[1m--keep-paused\x1b[0m flag is not provided,\nit will be unpaused. If the activity is paused and the \x1b[1m--keep-paused\x1b[0m\nflag is provided, it will stay paused.\n\nEither \x1b[1m--activity-id\x1b[0m (with \x1b[1m--workflow-id\x1b[0m for a workflow Activity, or\nalone for a standalone Activity) or \x1b[1m--query\x1b[0m must be specified.\n\n### Resetting activities that heartbeat {#reset-heartbeats}\n\nActivities that heartbeat will receive a\nCanceled failure the next time\nthey heartbeat after a reset.\n\nIf, in your Activity, you need to do any cleanup when an Activity is\nreset, handle this error and then re-throw it when you've cleaned up.\n\nReset always clears the heartbeat details.\n\nSpecify the Activity and Workflow IDs:\n\n\x1b[1mtemporal activity reset \\\n --activity-id YourActivityId \\\n --workflow-id YourWorkflowId \\\n --keep-paused\x1b[0m\n\nWorkflow Activities can be reset in bulk with a visibility query list filter:\n\n\x1b[1mtemporal activity reset \\\n --query 'WorkflowType=\"YourWorkflow\"'\x1b[0m\n\nThe \x1b[1m--query\x1b[0m flag currently applies only to Workflow Activities.\n\nOmit \x1b[1m--workflow-id\x1b[0m to target a Standalone Activity by Activity ID\nand optional Run ID." + s.Command.Long = "Reset an activity.\nThis restarts the activity as if it were first being scheduled: the\nattempt count returns to one and its per-attempt timeouts are re-armed.\nUse \x1b[1m--clear-heartbeat-details\x1b[0m to also clear its\nheartbeat details.\n\nIf the activity may be executing (i.e. it has not yet timed out), the\nreset will take effect the next time it fails, heartbeats, or times out.\nIf is waiting for a retry (i.e. has failed or timed out), the reset\nwill apply immediately.\n\nIf the activity is already paused, it will be unpaused by default.\nYou can specify \x1b[1m--keep-paused\x1b[0m to prevent this.\n\nIf the activity is paused and the \x1b[1m--keep-paused\x1b[0m flag is not provided,\nit will be unpaused. If the activity is paused and the \x1b[1m--keep-paused\x1b[0m\nflag is provided, it will stay paused.\n\nEither \x1b[1m--activity-id\x1b[0m (with \x1b[1m--workflow-id\x1b[0m for a workflow Activity, or\nalone for a standalone Activity) or \x1b[1m--query\x1b[0m must be specified.\n\n### Resetting activities that heartbeat {#clear-heartbeat-details}\n\nActivities that heartbeat will receive a\nCanceled failure the next time\nthey heartbeat after a reset.\n\nIf, in your Activity, you need to do any cleanup when an Activity is\nreset, handle this error and then re-throw it when you've cleaned up.\n\nUse \x1b[1m--clear-heartbeat-details\x1b[0m to clear the heartbeat details as part of\nthe reset.\n\nSpecify the Activity and Workflow IDs:\n\n\x1b[1mtemporal activity reset \\\n --activity-id YourActivityId \\\n --workflow-id YourWorkflowId \\\n --keep-paused\x1b[0m\n\nWorkflow Activities can be reset in bulk with a visibility query list filter:\n\n\x1b[1mtemporal activity reset \\\n --query 'WorkflowType=\"YourWorkflow\"'\x1b[0m\n\nThe \x1b[1m--query\x1b[0m flag currently applies only to Workflow Activities.\n\nOmit \x1b[1m--workflow-id\x1b[0m to target a Standalone Activity by Activity ID\nand optional Run ID." } else { - s.Command.Long = "Reset an activity.\nThis restarts the activity as if it were first being scheduled: the\nattempt count returns to one, its per-attempt timeouts are re-armed, and\nits heartbeat details are cleared.\n\nIf the activity may be executing (i.e. it has not yet timed out), the\nreset will take effect the next time it fails, heartbeats, or times out.\nIf is waiting for a retry (i.e. has failed or timed out), the reset\nwill apply immediately.\n\nIf the activity is already paused, it will be unpaused by default.\nYou can specify `--keep-paused` to prevent this.\n\nIf the activity is paused and the `--keep-paused` flag is not provided,\nit will be unpaused. If the activity is paused and the `--keep-paused`\nflag is provided, it will stay paused.\n\nEither `--activity-id` (with `--workflow-id` for a workflow Activity, or\nalone for a standalone Activity) or `--query` must be specified.\n\n### Resetting activities that heartbeat {#reset-heartbeats}\n\nActivities that heartbeat will receive a\nCanceled failure the next time\nthey heartbeat after a reset.\n\nIf, in your Activity, you need to do any cleanup when an Activity is\nreset, handle this error and then re-throw it when you've cleaned up.\n\nReset always clears the heartbeat details.\n\nSpecify the Activity and Workflow IDs:\n\n```\ntemporal activity reset \\\n --activity-id YourActivityId \\\n --workflow-id YourWorkflowId \\\n --keep-paused\n```\n\nWorkflow Activities can be reset in bulk with a visibility query list filter:\n\n```\ntemporal activity reset \\\n --query 'WorkflowType=\"YourWorkflow\"'\n```\n\nThe `--query` flag currently applies only to Workflow Activities.\n\nOmit `--workflow-id` to target a Standalone Activity by Activity ID\nand optional Run ID." + s.Command.Long = "Reset an activity.\nThis restarts the activity as if it were first being scheduled: the\nattempt count returns to one and its per-attempt timeouts are re-armed.\nUse `--clear-heartbeat-details` to also clear its\nheartbeat details.\n\nIf the activity may be executing (i.e. it has not yet timed out), the\nreset will take effect the next time it fails, heartbeats, or times out.\nIf is waiting for a retry (i.e. has failed or timed out), the reset\nwill apply immediately.\n\nIf the activity is already paused, it will be unpaused by default.\nYou can specify `--keep-paused` to prevent this.\n\nIf the activity is paused and the `--keep-paused` flag is not provided,\nit will be unpaused. If the activity is paused and the `--keep-paused`\nflag is provided, it will stay paused.\n\nEither `--activity-id` (with `--workflow-id` for a workflow Activity, or\nalone for a standalone Activity) or `--query` must be specified.\n\n### Resetting activities that heartbeat {#clear-heartbeat-details}\n\nActivities that heartbeat will receive a\nCanceled failure the next time\nthey heartbeat after a reset.\n\nIf, in your Activity, you need to do any cleanup when an Activity is\nreset, handle this error and then re-throw it when you've cleaned up.\n\nUse `--clear-heartbeat-details` to clear the heartbeat details as part of\nthe reset.\n\nSpecify the Activity and Workflow IDs:\n\n```\ntemporal activity reset \\\n --activity-id YourActivityId \\\n --workflow-id YourWorkflowId \\\n --keep-paused\n```\n\nWorkflow Activities can be reset in bulk with a visibility query list filter:\n\n```\ntemporal activity reset \\\n --query 'WorkflowType=\"YourWorkflow\"'\n```\n\nThe `--query` flag currently applies only to Workflow Activities.\n\nOmit `--workflow-id` to target a Standalone Activity by Activity ID\nand optional Run ID." } s.Command.Args = cobra.NoArgs s.Command.Flags().StringVarP(&s.ActivityId, "activity-id", "a", "", "The Activity ID to reset. Mutually exclusive with `--query`. Set `--workflow-id` to target a workflow Activity, or omit it to target a standalone Activity (the latest run unless `--run-id` is set).") s.Command.Flags().BoolVar(&s.KeepPaused, "keep-paused", false, "If the activity was paused, it will stay paused.") + s.Command.Flags().BoolVar(&s.ClearHeartbeatDetails, "clear-heartbeat-details", false, "Clear the Activity's heartbeat details.") s.Jitter = 0 s.Command.Flags().Var(&s.Jitter, "jitter", "The activity will reset at random a time within the specified duration. Can only be used with --query.") s.Command.Flags().BoolVar(&s.RestoreOriginalOptions, "restore-original-options", false, "Restore the original options of the activity.") diff --git a/internal/temporalcli/commands.yaml b/internal/temporalcli/commands.yaml index 6d24a8a00..2a4368f85 100644 --- a/internal/temporalcli/commands.yaml +++ b/internal/temporalcli/commands.yaml @@ -637,8 +637,9 @@ commands: description: | Reset an activity. This restarts the activity as if it were first being scheduled: the - attempt count returns to one, its per-attempt timeouts are re-armed, and - its [heartbeat details](#reset-heartbeats) are cleared. + attempt count returns to one and its per-attempt timeouts are re-armed. + Use `--clear-heartbeat-details` to also clear its + [heartbeat details](#clear-heartbeat-details). If the activity may be executing (i.e. it has not yet timed out), the reset will take effect the next time it fails, heartbeats, or times out. @@ -655,7 +656,7 @@ commands: Either `--activity-id` (with `--workflow-id` for a workflow Activity, or alone for a standalone Activity) or `--query` must be specified. - ### Resetting activities that heartbeat {#reset-heartbeats} + ### Resetting activities that heartbeat {#clear-heartbeat-details} Activities that heartbeat will receive a [Canceled failure](/references/failures#cancelled-failure) the next time @@ -664,7 +665,8 @@ commands: If, in your Activity, you need to do any cleanup when an Activity is reset, handle this error and then re-throw it when you've cleaned up. - Reset always clears the heartbeat details. + Use `--clear-heartbeat-details` to clear the heartbeat details as part of + the reset. Specify the Activity and Workflow IDs: @@ -697,6 +699,9 @@ commands: - name: keep-paused type: bool description: If the activity was paused, it will stay paused. + - name: clear-heartbeat-details + type: bool + description: Clear the Activity's heartbeat details. - name: jitter type: duration description: |