feat(activity): report activity reset in the heartbeat RPC response

.github/workflows/release-please.yml

@@ -0,0 +1,27 @@
+name: release-please
+on:
+  push:
+    branches:
+      - master
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: release-please
+  cancel-in-progress: false
+
+jobs:
+  release-please:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: write
+      pull-requests: write
+    steps:
+      - name: Update release
+        uses: googleapis/release-please-action@45996ed1f6d02564a971a2fa1b5860e934307cf7 # v5.0.0
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          target-branch: master

.release-please-manifest.json

@@ -0,0 +1,3 @@
+{
+  ".": "6.0.0"
+}

release-please-config.json

@@ -0,0 +1,9 @@
+{
+  "$schema": "https://raw.githubusercontent.com/googleapis/release-please/main/schemas/config.json",
+  "release-type": "go",
+  "include-component-in-tag": false,
+  "pull-request-header": "Release pull request. Merge it to create the tag and the GitHub release.",
+  "packages": {
+    ".": {}
+  }
+}

rpc.go

@@ -38,10 +38,11 @@ type RecordHeartbeatRequest struct {
 	Details   []byte `json:"details"`
 }
 
-// RecordHeartbeatResponse sent back to the worker to indicate that activity was canceled.
+// RecordHeartbeatResponse sent back to the worker to indicate that the activity was canceled, paused or reset.
 type RecordHeartbeatResponse struct {
 	Canceled bool `json:"canceled"`
 	Paused   bool `json:"paused"`
+	Reset    bool `json:"reset"`
 }
 
 // RecordActivityHeartbeat records heartbeat for an activity.
@@ -73,12 +74,16 @@ func (r *rpc) RecordActivityHeartbeat(in RecordHeartbeatRequest, out *RecordHear
 
 	activity.RecordHeartbeat(ctx, details)
 
-	if stderr.Is(context.Cause(ctx), activity.ErrActivityPaused) {
+	cause := context.Cause(ctx)
+	switch {
+	case stderr.Is(cause, activity.ErrActivityPaused):
 		*out = RecordHeartbeatResponse{Paused: true}
-		return nil
+	case stderr.Is(cause, activity.ErrActivityReset):
+		*out = RecordHeartbeatResponse{Reset: true}
+	default:
+		*out = RecordHeartbeatResponse{Canceled: ctx.Err() != nil}
 	}
 
-	*out = RecordHeartbeatResponse{Canceled: ctx.Err() != nil}
 	return nil
 }
 

tests/general/hp_test.go

@@ -19,6 +19,7 @@ import (
 	"go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/history/v1"
 	"go.temporal.io/api/operatorservice/v1"
+	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/client"
 )
 
@@ -549,6 +550,44 @@ func Test_ActivityHeartbeatProto(t *testing.T) {
 	wg.Wait()
 }
 
+func Test_ActivityResetProto(t *testing.T) {
+	stopCh := make(chan struct{}, 1)
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	s := helpers.NewTestServer(t, stopCh, wg, "../configs/.rr-proto.yaml")
+
+	w, err := s.Client.ExecuteWorkflow(
+		context.Background(),
+		client.StartWorkflowOptions{
+			TaskQueue: "default",
+		},
+		"ActivityResetWorkflow",
+	)
+	assert.NoError(t, err)
+
+	// let the activity start and record its first heartbeat
+	time.Sleep(time.Second * 2)
+
+	_, err = s.Client.WorkflowService().ResetActivity(context.Background(), &workflowservice.ResetActivityRequest{
+		Namespace: "default",
+		Execution: &common.WorkflowExecution{
+			WorkflowId: w.GetID(),
+			RunId:      w.GetRunID(),
+		},
+		Activity: &workflowservice.ResetActivityRequest_Type{
+			Type: "HeartBeatActivity.untilReset",
+		},
+	})
+	require.NoError(t, err)
+
+	var result string
+	require.NoError(t, w.Get(context.Background(), &result))
+	assert.Equal(t, `Temporal\Exception\Client\ActivityResetException`, result)
+
+	stopCh <- struct{}{}
+	wg.Wait()
+}
+
 func Test_BinaryPayloadProto(t *testing.T) {
 	stopCh := make(chan struct{}, 1)
 	wg := &sync.WaitGroup{}

tests/php_test_files/src/Activity/HeartBeatActivity.php

@@ -5,6 +5,7 @@
 use Temporal\Activity;
 use Temporal\Activity\ActivityInterface;
 use Temporal\Activity\ActivityMethod;
+use Temporal\Exception\Client\ActivityCompletionException;
 use Temporal\Roadrunner\Internal\Error;
 
 #[ActivityInterface(prefix: "HeartBeatActivity.")]
@@ -55,4 +56,19 @@ public function failedActivity(
 
         return 'OK!';
     }
+
+    #[ActivityMethod]
+    public function untilReset(): string
+    {
+        try {
+            for ($i = 0; $i < 20; $i++) {
+                Activity::heartbeat(['value' => $i]);
+                sleep(1);
+            }
+        } catch (ActivityCompletionException $e) {
+            return $e::class;
+        }
+
+        return 'NOT_RESET';
+    }
 }

tests/php_test_files/src/Workflow/ActivityResetWorkflow.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Workflow;
+
+use Temporal\Activity\ActivityOptions;
+use Temporal\Tests\Activity\HeartBeatActivity;
+use Temporal\Workflow;
+use Temporal\Workflow\WorkflowMethod;
+
+#[Workflow\WorkflowInterface]
+class ActivityResetWorkflow
+{
+    #[WorkflowMethod(name: 'ActivityResetWorkflow')]
+    public function handler(): iterable
+    {
+        $act = Workflow::newActivityStub(
+            HeartBeatActivity::class,
+            ActivityOptions::new()
+                ->withStartToCloseTimeout(30)
+                ->withHeartbeatTimeout(2)
+        );
+
+        return yield $act->untilReset();
+    }
+}