Allow canceling recurring reminders after execution for non-internal actors (dapr#7337)

* Allow canceling recurring reminders after execution for non-internal actors

Internal actors have the ability to cancel a recurring reminder by responding to an execution with the special error `ErrReminderCanceled` after a reminder is executed.

However, the capability was never exposed to "external" actors.

This PR adds the ability for an actor to stop recurring reminders (and timers) after they are executed also for "external" actors. This is possible by including the header `X-DaprReminderCancel` with a truthy value in the response when a reminder or timer is executed.

This will require support in SDKs, but it allows something that is particularly helpful (and today can be buggy, see dapr#6666)

Signed-off-by: ItalyPaleAle <[email protected]>

* Lint

Signed-off-by: ItalyPaleAle <[email protected]>

* Lint

Signed-off-by: ItalyPaleAle <[email protected]>

---------

Signed-off-by: ItalyPaleAle <[email protected]>
Co-authored-by: Mukundan Sundararajan <[email protected]>
Co-authored-by: Dapr Bot <[email protected]>

Author: 3 people

pkg/actors/actors.go

@@ -56,6 +56,7 @@ import (
 	eventqueue "github.com/dapr/kit/events/queue"
 	"github.com/dapr/kit/logger"
 	"github.com/dapr/kit/ptr"
+	"github.com/dapr/kit/utils"
 )
 
 const (
@@ -676,6 +677,11 @@ func (a *actorsRuntime) callLocalActor(ctx context.Context, req *internalv1pb.In
 		return res, actorerrors.NewActorError(res)
 	}
 
+	// Allow stopping a recurring reminder or timer
+	if v := res.GetHeaders()["X-Daprremindercancel"]; v != nil && len(v.GetValues()) > 0 && utils.IsTruthy(v.GetValues()[0]) {
+		return res, ErrReminderCanceled
+	}
+
 	return res, nil
 }
 

tests/integration/suite/actors/reminders/basic.go

@@ -45,7 +45,8 @@ type basic struct {
 	daprd *daprd.Daprd
 	place *placement.Placement
 
-	methodcalled atomic.Int64
+	reminderCalled     atomic.Int64
+	stopReminderCalled atomic.Int64
 }
 
 func (b *basic) Setup(t *testing.T) []framework.Option {
@@ -57,7 +58,12 @@ func (b *basic) Setup(t *testing.T) []framework.Option {
 		w.WriteHeader(http.StatusOK)
 	})
 	handler.HandleFunc("/actors/myactortype/myactorid/method/remind/remindermethod", func(w http.ResponseWriter, r *http.Request) {
-		b.methodcalled.Add(1)
+		b.reminderCalled.Add(1)
+	})
+	handler.HandleFunc("/actors/myactortype/myactorid/method/remind/stopreminder", func(w http.ResponseWriter, r *http.Request) {
+		b.stopReminderCalled.Add(1)
+		w.Header().Set("X-DaprReminderCancel", "true")
+		w.WriteHeader(http.StatusOK)
 	})
 	handler.HandleFunc("/actors/myactortype/myactorid/method/foo", func(w http.ResponseWriter, r *http.Request) {})
 
@@ -82,30 +88,19 @@ func (b *basic) Run(t *testing.T, ctx context.Context) {
 
 	daprdURL := "http://localhost:" + strconv.Itoa(b.daprd.HTTPPort()) + "/v1.0/actors/myactortype/myactorid"
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, daprdURL+"/method/foo", nil)
-	require.NoError(t, err)
-
-	require.EventuallyWithT(t, func(c *assert.CollectT) {
-		resp, rErr := client.Do(req)
-		//nolint:testifylint
-		if assert.NoError(c, rErr) {
-			assert.NoError(c, resp.Body.Close())
-			assert.Equal(c, http.StatusOK, resp.StatusCode)
-		}
-	}, time.Second*10, time.Millisecond*10, "actor not ready in time")
-
-	body := `{"dueTime": "0ms"}`
-	req, err = http.NewRequestWithContext(ctx, http.MethodPost, daprdURL+"/reminders/remindermethod", strings.NewReader(body))
-	require.NoError(t, err)
-
-	resp, err := client.Do(req)
-	require.NoError(t, err)
-	require.NoError(t, resp.Body.Close())
-	assert.Equal(t, http.StatusNoContent, resp.StatusCode)
-
-	assert.Eventually(t, func() bool {
-		return b.methodcalled.Load() == 1
-	}, time.Second*3, time.Millisecond*10)
+	t.Run("actor ready", func(t *testing.T) {
+		req, err := http.NewRequestWithContext(ctx, http.MethodPost, daprdURL+"/method/foo", nil)
+		require.NoError(t, err)
+
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			resp, rErr := client.Do(req)
+			//nolint:testifylint
+			if assert.NoError(c, rErr) {
+				assert.NoError(c, resp.Body.Close())
+				assert.Equal(c, http.StatusOK, resp.StatusCode)
+			}
+		}, 10*time.Second, 10*time.Millisecond, "actor not ready in time")
+	})
 
 	conn, err := grpc.DialContext(ctx, b.daprd.GRPCAddress(),
 		grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock(),
@@ -114,15 +109,57 @@ func (b *basic) Run(t *testing.T, ctx context.Context) {
 	t.Cleanup(func() { require.NoError(t, conn.Close()) })
 	gclient := rtv1.NewDaprClient(conn)
 
-	_, err = gclient.RegisterActorReminder(ctx, &rtv1.RegisterActorReminderRequest{
-		ActorType: "myactortype",
-		ActorId:   "myactorid",
-		Name:      "remindermethod",
-		DueTime:   "0ms",
+	t.Run("schedule reminder via HTTP", func(t *testing.T) {
+		const body = `{"dueTime": "0ms"}`
+		var (
+			req  *http.Request
+			resp *http.Response
+		)
+		req, err = http.NewRequestWithContext(ctx, http.MethodPost, daprdURL+"/reminders/remindermethod", strings.NewReader(body))
+		require.NoError(t, err)
+
+		resp, err = client.Do(req)
+		require.NoError(t, err)
+		require.NoError(t, resp.Body.Close())
+		assert.Equal(t, http.StatusNoContent, resp.StatusCode)
+
+		assert.Eventually(t, func() bool {
+			return b.reminderCalled.Load() == 1
+		}, 3*time.Second, 10*time.Millisecond)
 	})
-	require.NoError(t, err)
 
-	assert.Eventually(t, func() bool {
-		return b.methodcalled.Load() == 2
-	}, time.Second*3, time.Millisecond*10)
+	t.Run("schedule reminder via gRPC", func(t *testing.T) {
+		_, err = gclient.RegisterActorReminder(ctx, &rtv1.RegisterActorReminderRequest{
+			ActorType: "myactortype",
+			ActorId:   "myactorid",
+			Name:      "remindermethod",
+			DueTime:   "0ms",
+		})
+		require.NoError(t, err)
+
+		assert.Eventually(t, func() bool {
+			return b.reminderCalled.Load() == 2
+		}, 3*time.Second, 10*time.Millisecond)
+	})
+
+	t.Run("cancel recurring reminder", func(t *testing.T) {
+		// Register a reminder that repeats every second
+		_, err = gclient.RegisterActorReminder(ctx, &rtv1.RegisterActorReminderRequest{
+			ActorType: "myactortype",
+			ActorId:   "myactorid",
+			Name:      "stopreminder",
+			DueTime:   "0s",
+			Period:    "1s",
+		})
+		require.NoError(t, err)
+
+		// Should be invoked once
+		assert.Eventually(t, func() bool {
+			return b.stopReminderCalled.Load() == 1
+		}, 3*time.Second, 10*time.Millisecond)
+
+		// After 2s, should not have been invoked more
+		time.Sleep(2 * time.Second)
+		assert.Equal(t, int64(1), b.stopReminderCalled.Load())
+	})
 }