Standalone activity heartbeating #8730

dandavison · 2025-12-01T23:02:43Z

What changed?

Implement heartbeating for CHASM activities (standalone activity)

Why?

Required feature

How did you test it?

built
added new functional test(s)

Note

Implements heartbeating for standalone activities, adding heartbeat timer tasks, token validation, API wiring, and comprehensive tests for timeout/retry behavior.

Activity (core):
- Add RecordHeartbeat with task-token validation, recording details, and scheduling HeartbeatTimeoutTask; heartbeat response includes CancelRequested.
- Introduce WithToken[T] wrapper and require it for HandleCompleted/Failed/Canceled; validate via new ValidateActivityTaskToken.
- Schedule heartbeat timer on TransitionStarted; compute timers from attempt start time; rename helper to getOrCreateLastHeartbeat.
- Add heartbeat-timeout failure creation and handling in TransitionTimedOut.
Task Executors:
- New heartbeatTimeoutTaskExecutor with precise validation (attempt match, latest hb) and execution (retry or heartbeat timeout).
- Wire executor in library.Tasks() and register via fx.
APIs (History/Frontend):
- Route RecordActivityTaskHeartbeat, RespondActivityTask{Completed,Failed,Canceled} to CHASM component when token has ComponentRef, passing WithToken.
Protobuf/Config:
- Extend proto/v1/tasks.proto with HeartbeatTimeoutTask (+ generated helpers).
Tests:
- Add functional tests covering heartbeats: invalid/stale tokens, cancelRequested flag, details on retry, heartbeat-driven timeout and retries, and heartbeats preventing timeout.

^{Written by Cursor Bugbot for commit 988cc7f. This will update automatically on new commits. Configure here.}

dandavison · 2025-12-01T23:05:57Z

chasm/lib/activity/activity.go

+func (a *Activity) HandleStarted(
+	ctx chasm.MutableContext,
+	request *historyservice.RecordActivityTaskStartedRequest,
+) (*historyservice.RecordActivityTaskStartedResponse, error) {


ignore: formatting only

Looks like @fretz12 and you have different ideas on how to format these signatures. I personally like the style in this PR better.

I like this formatting better too, but ideally we should have the linter take care of it.
I'm running gofumpt locally, it seems to do a good job... maybe introduce it as part of the lint-code target?

dandavison · 2025-12-01T23:06:03Z

chasm/lib/activity/activity.go

+	ctx chasm.Context,
+	key chasm.EntityKey,
+	response *historyservice.RecordActivityTaskStartedResponse,
+) error {


ignore: formatting only

dandavison · 2025-12-01T23:06:42Z

chasm/lib/activity/activity.go

+func (a *Activity) handleCancellationRequested(
+	ctx chasm.MutableContext,
+	req *activitypb.CancelActivityExecutionRequest,
+) (*activitypb.CancelActivityExecutionResponse, error) {


ignore: formatting only

bergundy · 2025-12-02T00:58:26Z

chasm/lib/activity/activity.go

+func (a *Activity) HandleStarted(
+	ctx chasm.MutableContext,
+	request *historyservice.RecordActivityTaskStartedRequest,
+) (*historyservice.RecordActivityTaskStartedResponse, error) {


Looks like @fretz12 and you have different ideas on how to format these signatures. I personally like the style in this PR better.

bergundy · 2025-12-02T01:06:13Z

chasm/lib/activity/activity_tasks.go

+}
+
+// Validate validates a HeartbeatTimeoutTask.
+func (e *heartbeatTimeoutTaskExecutor) Validate(


You will want to validate that the schedule time in the attributes is still relevant here not in the execute function. It should deterministic function of the last heartbeat time and the attempt start time (hbDeadline below should be equal to the schedule time).

See reply below

bergundy · 2025-12-02T01:07:11Z

chasm/lib/activity/activity_tasks.go

+	hbTimeout := activity.GetHeartbeatTimeout().AsDuration()
+	attemptStartTime := attempt.GetStartedTime().AsTime()
+	lastHbTime := lastHb.GetRecordedTime().AsTime() // could be from a previous attempt
+	// No heartbeats in the attempt so far is equivalent to a heartbeat having been sent at attempt
+	// start time.
+	hbDeadline := util.MaxTime(lastHbTime, attemptStartTime).Add(hbTimeout)
+
+	if ctx.Now(activity).Before(hbDeadline) {
+		// Deadline has not expired; schedule a new task.
+		ctx.AddTask(
+			activity,
+			chasm.TaskAttributes{
+				ScheduledTime: hbDeadline,
+			},
+			&activitypb.HeartbeatTimeoutTask{
+				Attempt: attempt.GetCount(),
+			},
+		)
+		return nil
+	}


This shouldn't happen here. Validate in the Validate function. Schedule a new task any time a heartbeat is received.

I currently still think that the design I've proposed is preferable:

I've added a high water mark mechanism to make the tasks become invalid on execution.

Discussed offline -- design B incurs extra persistence writes that aren't balanced out by the changes to numbers of logical tasks created. I've switched the PR over to the design that @bergundy proposes.

bergundy · 2025-12-02T01:10:08Z

chasm/lib/activity/library.go

 			l.activityDispatchTaskExecutor,
 			l.activityDispatchTaskExecutor,
 		),
+		// TODO(dan): why are the task names "FooTimer" but "FooTimeoutTask" in the struct names?


We agreed not to add Task to the task names. Not sure if we want to use TaskExecutor vs. just Executor for the struct names but that is not critical and can change easily. The string names cannot be changed easily OTOH because they affect how tasks are represented in persistence.

bergundy · 2025-12-02T01:14:47Z

chasm/lib/activity/activity.go

+		Details:      input.Request.HeartbeatRequest.GetDetails(),
 	})
-	return nil, nil
+	return &historyservice.RecordActivityTaskHeartbeatResponse{


Just before returning here you want to generate a new heartbeat task if the heartbeat timeout is set.

See reply above; I currently still think the proposed design is preferable.

This is done now -- switched back to your design.

bergundy · 2025-12-02T01:17:20Z

tests/standalone_activity_test.go

+				require.Error(t, err)
+				statusErr := serviceerror.ToStatus(err)
+				require.NotNil(t, statusErr)
+				require.Equal(t, codes.InvalidArgument, statusErr.Code())


Alternatively, you can use require.ErrorAs(err, &invalidArgumentErr)

Leaving as is for now

fretz12 · 2025-12-02T00:20:45Z

chasm/lib/activity/validator.go

+) error {
+	if a.Status != activitystatepb.ACTIVITY_EXECUTION_STATUS_STARTED &&
+		a.Status != activitystatepb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED {
+		return serviceerror.NewNotFound("activity task not found")


Wrong error message? Also NotFound doesn't feel like the right error type, even though I get it's checking if a token "matches". Perhaps FailPrecondition?

I used NotFound because it'd what the workflow implementation uses (see IsActivityTaskNotFoundForToken)

fretz12 · 2025-12-02T00:21:29Z

chasm/lib/activity/validator.go

+		return err
+	}
+	if token.Attempt != attempt.GetCount() {
+		return serviceerror.NewNotFound("activity task not found")


Same comment as above

I used NotFound because it'd what the workflow implementation uses (see IsActivityTaskNotFoundForToken)

fretz12 · 2025-12-02T00:28:55Z

chasm/lib/activity/activity.go

+	if err := ValidateActivityTaskToken(ctx, a, input.Token); err != nil {
+		return nil, err
+	}
 	a.LastHeartbeat = chasm.NewDataField(ctx, &activitypb.ActivityHeartbeatState{


@bergundy is there any performance pentalty if we create a new field on every hearbeat?

No, no penalty.

fretz12 · 2025-12-02T00:31:22Z

chasm/lib/activity/activity.go

 	a.LastHeartbeat = chasm.NewDataField(ctx, &activitypb.ActivityHeartbeatState{
 		RecordedTime: timestamppb.New(ctx.Now(a)),
-		Details:      details,
+		Details:      input.Request.HeartbeatRequest.GetDetails(),


Suggested change

Details: input.Request.HeartbeatRequest.GetDetails(),

Details: input.Request.GetHeartbeatRequest().GetDetails(),

Thanks, taken

fretz12 · 2025-12-02T01:00:43Z

chasm/lib/activity/activity_tasks.go

+func (e *heartbeatTimeoutTaskExecutor) Execute(
+	ctx chasm.MutableContext,
+	activity *Activity,
+	taskAttrs chasm.TaskAttributes,


nit: taskAttrs, task are unused

Thanks, marked as _ unused parameters

fretz12 · 2025-12-02T01:09:03Z

chasm/lib/activity/activity_tasks.go

+		ctx.AddTask(
+			activity,
+			chasm.TaskAttributes{
+				ScheduledTime: hbDeadline,


If hbDeadline happens to be attemptStartTime + hbTimeout (i.e., no hearbeat recorded yet), won't the next timer basically pop the same time as the current one being executed?

That should not happen because of this condition

if ctx.Now(activity).Before(hbDeadline)

the heartbeat task will not execute before time hbDeadline since the first task is scheduled for hbTimeout.

chasm/lib/activity/activity.go

cursor · 2025-12-09T00:06:16Z

chasm/lib/activity/proto/v1/activity_state.proto


    // Set if activity cancelation was requested.
-    ActivityCancelState cancel_state = 11;
+    ActivityCancelState cancel_state = 12;


Bug: Proto field numbers changed breaking wire format compatibility

The new field last_heartbeat_task_scheduled_time was inserted at field number 7, causing all subsequent fields to be renumbered: retry_policy moved from 7→8, status from 8→9, schedule_time from 9→10, priority from 10→11, and cancel_state from 11→12. This breaks protobuf wire format backward compatibility. If any ActivityState messages were previously serialized and stored, deserializing them with the new schema will cause data corruption (e.g., the old retry_policy data would be interpreted as last_heartbeat_task_scheduled_time). New fields should be added with the next available field number (12) without renumbering existing fields.

Additional Locations (1)

chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go#L172-L188

dandavison · 2025-12-09T00:07:39Z

chasm/lib/activity/activity.go

+	if err := TransitionStarted.Apply(a, ctx, nil); err != nil {
+		return nil, err
+	}
+


The movement here was required to fix a bug that I encountered when implementing the other proposed design. We needed attempt.StartedTime to be available in TransitionStarted so that we could use it in scheduling the first heartbeat task, as well as the base for the start-to-close timeout. Before this PR we were using an ad-hoc ctx.Now() that was close to but not equal to StartedTime.

Reverting it doesn't cause a test failure with the hreartbeating design in this PR, but it's better to set the field state before calling the transition function.

Reverting this would now cause a test failure since the PR is back to design A.

dandavison · 2025-12-09T00:10:33Z

chasm/lib/activity/statemachine.go

 	activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
 	func(a *Activity, ctx chasm.MutableContext, _ any) error {
+		attempt := a.LastAttempt.Get(ctx)
+		startTime := attempt.GetStartedTime().AsTime()


This is where we use StartedTime that I referred to above; it wasn't possible before because we were calling the transition function before setting the attempt field state.

Reverting this doesn't cause a test failure in the current PR. But it did with the alternative design, and in any case it's better to use StartedTime rather than an arbitrary timestamp that's close to it in time.

Reverting this would now cause a test failure since the PR is back to design A.

This reverts commit 534685045cdb01abd7d6191126c25204cc6493ec.

This reverts commit 3796405.

dandavison requested review from a team as code owners December 1, 2025 23:02

dandavison requested review from bergundy and fretz12 December 1, 2025 23:03

dandavison force-pushed the heartbeat branch 2 times, most recently from 0527002 to 6ffe4da Compare December 1, 2025 23:22

dandavison commented Dec 1, 2025

View reviewed changes

bergundy reviewed Dec 2, 2025

View reviewed changes

fretz12 reviewed Dec 2, 2025

View reviewed changes

cursor bot reviewed Dec 5, 2025

View reviewed changes

chasm/lib/activity/activity.go Show resolved Hide resolved

dandavison force-pushed the saa-schedule-to-close-bug-2 branch 3 times, most recently from 81b2516 to c8e5800 Compare December 6, 2025 00:24

Base automatically changed from saa-schedule-to-close-bug-2 to standalone-activity December 6, 2025 00:31

dandavison force-pushed the heartbeat branch from 7c0893a to c37d16a Compare December 8, 2025 18:30

dandavison changed the base branch from standalone-activity to standalone-activity-describe-and-get-outcome December 8, 2025 18:31

cursor bot reviewed Dec 8, 2025

View reviewed changes

chasm/lib/activity/activity.go Show resolved Hide resolved

dandavison force-pushed the standalone-activity-describe-and-get-outcome branch from 938ff87 to 4ba7671 Compare December 8, 2025 21:38

dandavison force-pushed the heartbeat branch from e3f87a9 to 5b95a60 Compare December 8, 2025 23:22

Base automatically changed from standalone-activity-describe-and-get-outcome to standalone-activity December 8, 2025 23:39

dandavison added 2 commits December 8, 2025 18:52

Activity heartbeating

bbf3c34

startTime bug fix

e3e5777

dandavison force-pushed the heartbeat branch from 5b95a60 to 77b6f1c Compare December 8, 2025 23:52

cursor bot reviewed Dec 9, 2025

View reviewed changes

dandavison commented Dec 9, 2025

View reviewed changes

dandavison added 5 commits December 8, 2025 22:01

Respond to upstream

b003aa4

Alternative heartbeating implementation

aa5155a

Revert "Alternative heartbeating implementation"

16192e2

This reverts commit 534685045cdb01abd7d6191126c25204cc6493ec.

Add hwm

3796405

Revert "Add hwm"

d0836a6

This reverts commit 3796405.

Alternative heartbeating implementation

3ba3273

dandavison force-pushed the heartbeat branch from 77b6f1c to 3ba3273 Compare December 9, 2025 03:05

dandavison requested review from bergundy and fretz12 December 10, 2025 02:24

dandavison added 2 commits December 10, 2025 12:43

Address linter complaints

e968ad0

Doc-comment edits

988cc7f

	Details: input.Request.HeartbeatRequest.GetDetails(),
	Details: input.Request.GetHeartbeatRequest().GetDetails(),

Standalone activity heartbeating #8730

Are you sure you want to change the base?

Standalone activity heartbeating #8730

Conversation

dandavison commented Dec 1, 2025 • edited by cursor bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changed?

Why?

How did you test it?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

cursor bot Dec 9, 2025

Choose a reason for hiding this comment

Bug: Proto field numbers changed breaking wire format compatibility

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dandavison commented Dec 1, 2025 •

edited by cursor bot

Loading