Send timer cancelation message

For the Redis backend, timer cancelation message will also remove future scheduled events. For other backends this keeps the current behavior.

Author: cschleiden

.vscode/launch.json

@@ -45,7 +45,7 @@
       "type": "go",
       "request": "launch",
       "mode": "debug",
-      "program": "${workspaceFolder}/samples/cancellation/cancellation.go"
+      "program": "${workspaceFolder}/samples/cancelation/cancelation.go"
     },
     {
       "name": "Launch subworkflow sample",

backend/redis/events.go

@@ -30,6 +30,15 @@ func addEventToStream(ctx context.Context, rdb redis.UniversalClient, streamKey
 	return &msgID, nil
 }
 
+// KEYS[1] - future event zset key
+// KEYS[2] - future event key
+// ARGV[1] - timestamp
+// ARGV[2] - event payload
+var addFutureEventCmd = redis.NewScript(`
+	redis.call("ZADD", KEYS[1], ARGV[1], KEYS[2])
+	redis.call("SET", KEYS[2], ARGV[2])
+`)
+
 func addFutureEvent(ctx context.Context, rdb redis.UniversalClient, instance *core.WorkflowInstance, event *history.Event) error {
 	futureEvent := &futureEvent{
 		Instance: instance,
@@ -41,12 +50,32 @@ func addFutureEvent(ctx context.Context, rdb redis.UniversalClient, instance *co
 		return err
 	}
 
-	if err := rdb.ZAdd(ctx, futureEventsKey(), &redis.Z{
-		Member: eventData,
-		Score:  float64(event.VisibleAt.Unix()),
-	}).Err(); err != nil {
+	if err := addFutureEventCmd.Run(
+		ctx,
+		rdb,
+		[]string{futureEventsKey(), futureEventKey(instance.InstanceID, event.ScheduleEventID)},
+		event.VisibleAt.Unix(),
+		string(eventData),
+	).Err(); err != nil && err != redis.Nil {
 		return fmt.Errorf("adding future event: %w", err)
 	}
 
 	return nil
 }
+
+// KEYS[1] - future event zset key
+// KEYS[2] - future event key
+var removeFutureEventCmd = redis.NewScript(`
+	redis.call("ZREM", KEYS[1], KEYS[2])
+	redis.call("DEL", KEYS[2])
+`)
+
+func removeFutureEvent(ctx context.Context, rdb redis.UniversalClient, instance *core.WorkflowInstance, event *history.Event) error {
+	key := futureEventKey(instance.InstanceID, event.ScheduleEventID)
+
+	if err := removeFutureEventCmd.Run(ctx, rdb, []string{futureEventsKey(), key}).Err(); err != nil && err != redis.Nil {
+		return fmt.Errorf("removing future event: %w", err)
+	}
+
+	return nil
+}

backend/redis/keys.go

@@ -23,3 +23,7 @@ func historyKey(instanceID string) string {
 func futureEventsKey() string {
 	return "future-events"
 }
+
+func futureEventKey(instanceID string, scheduleEventID int64) string {
+	return fmt.Sprintf("future-event:%v:%v", instanceID, scheduleEventID)
+}

backend/redis/workflow.go

@@ -25,11 +25,11 @@ type futureEvent struct {
 // KEYS[1] - future event set key
 // ARGV[1] - current timestamp for zrange
 var futureEventsCmd = redis.NewScript(`
-	local events = redis.call("ZRANGEBYSCORE", KEYS[1], "-inf", ARGV[1])
+	local events = redis.call("ZRANGE", KEYS[1], "-inf", ARGV[1], "BYSCORE")
 	if events ~= false and #events ~= 0 then
 		redis.call("ZREMRANGEBYSCORE", KEYS[1], "-inf", ARGV[1])
+		return redis.call("MGET", unpack(events))
 	end
-	return events
 `)
 
 func (rb *redisBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, error) {
@@ -38,43 +38,45 @@ func (rb *redisBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, er
 	nowStr := strconv.Itoa(int(now))
 
 	result, err := futureEventsCmd.Run(ctx, rb.rdb, []string{futureEventsKey()}, nowStr).Result()
-	if err != nil {
+	if err != nil && err != redis.Nil {
 		return nil, fmt.Errorf("checking future events: %w", err)
 	}
 
-	for _, eventR := range result.([]interface{}) {
-		eventStr := eventR.(string)
-		var futureEvent futureEvent
-		if err := json.Unmarshal([]byte(eventStr), &futureEvent); err != nil {
-			return nil, fmt.Errorf("unmarshaling event: %w", err)
-		}
+	if result != nil {
+		for _, eventR := range result.([]interface{}) {
+			eventStr := eventR.(string)
+			var futureEvent futureEvent
+			if err := json.Unmarshal([]byte(eventStr), &futureEvent); err != nil {
+				return nil, fmt.Errorf("unmarshaling event: %w", err)
+			}
 
-		instanceState, err := readInstance(ctx, rb.rdb, futureEvent.Instance.InstanceID)
-		if err != nil {
-			if err == backend.ErrInstanceNotFound {
-				rb.options.Logger.Debug("Ignoring future event for non-existing instance", "instance_id", futureEvent.Instance.InstanceID, "event_id", futureEvent.Event.ID)
-				continue
-			} else {
-				return nil, fmt.Errorf("reading instance: %w", err)
+			instanceState, err := readInstance(ctx, rb.rdb, futureEvent.Instance.InstanceID)
+			if err != nil {
+				if err == backend.ErrInstanceNotFound {
+					rb.options.Logger.Debug("Ignoring future event for non-existing instance", "instance_id", futureEvent.Instance.InstanceID, "event_id", futureEvent.Event.ID)
+					continue
+				} else {
+					return nil, fmt.Errorf("reading instance: %w", err)
+				}
 			}
-		}
 
-		if instanceState.State != backend.WorkflowStateActive {
-			rb.options.Logger.Debug("Ignoring future event for already completed instance", "instance_id", futureEvent.Instance.InstanceID, "event_id", futureEvent.Event.ID)
-			continue
-		}
+			if instanceState.State != backend.WorkflowStateActive {
+				rb.options.Logger.Debug("Ignoring future event for already completed instance", "instance_id", futureEvent.Instance.InstanceID, "event_id", futureEvent.Event.ID)
+				continue
+			}
 
-		msgID, err := addEventToStream(ctx, rb.rdb, pendingEventsKey(futureEvent.Instance.InstanceID), futureEvent.Event)
-		if err != nil {
-			return nil, fmt.Errorf("adding future event to stream: %w", err)
-		}
+			msgID, err := addEventToStream(ctx, rb.rdb, pendingEventsKey(futureEvent.Instance.InstanceID), futureEvent.Event)
+			if err != nil {
+				return nil, fmt.Errorf("adding future event to stream: %w", err)
+			}
 
-		// Instance now has at least one pending event, try to queue task
-		if _, err := rb.workflowQueue.Enqueue(ctx, futureEvent.Instance.InstanceID, &workflowTaskData{
-			LastPendingEventMessageID: *msgID,
-		}); err != nil {
-			if err != taskqueue.ErrTaskAlreadyInQueue {
-				return nil, fmt.Errorf("queueing workflow task: %w", err)
+			// Instance now has at least one pending event, try to queue task
+			if _, err := rb.workflowQueue.Enqueue(ctx, futureEvent.Instance.InstanceID, &workflowTaskData{
+				LastPendingEventMessageID: *msgID,
+			}); err != nil {
+				if err != taskqueue.ErrTaskAlreadyInQueue {
+					return nil, fmt.Errorf("queueing workflow task: %w", err)
+				}
 			}
 		}
 	}
@@ -171,7 +173,11 @@ func (rb *redisBackend) CompleteWorkflowTask(ctx context.Context, taskID string,
 
 		// TODO: use pipelines
 		for _, event := range events {
-			if event.VisibleAt != nil {
+			if event.Type == history.EventType_TimerCanceled {
+				if err := removeFutureEvent(ctx, rb.rdb, targetInstance, &event); err != nil {
+					return err
+				}
+			} else if event.VisibleAt != nil {
 				// Add future events
 				if err := addFutureEvent(ctx, rb.rdb, targetInstance, &event); err != nil {
 					return err

internal/command/command.go

@@ -95,15 +95,15 @@ func NewScheduleTimerCommand(id int64, at time.Time) Command {
 }
 
 type CancelTimerCommandAttr struct {
-	TimerID int
+	TimerScheduleEventID int64
 }
 
-func NewCancelTimerCommand(id int64, timerID int) Command {
+func NewCancelTimerCommand(id int64, timerID int64) Command {
 	return Command{
 		ID:   id,
 		Type: CommandType_CancelTimer,
 		Attr: &CancelTimerCommandAttr{
-			TimerID: timerID,
+			TimerScheduleEventID: timerID,
 		},
 	}
 }

internal/history/history.go

@@ -29,6 +29,7 @@ const (
 
 	EventType_TimerScheduled
 	EventType_TimerFired
+	EventType_TimerCanceled
 
 	EventType_SignalReceived
 

internal/history/serialization.go

@@ -60,6 +60,8 @@ func DeserializeAttributes(eventType EventType, attributes []byte) (attr interfa
 		attr = &TimerScheduledAttributes{}
 	case EventType_TimerFired:
 		attr = &TimerFiredAttributes{}
+	case EventType_TimerCanceled:
+		attr = &TimerCanceledAttributes{}
 
 	case EventType_SubWorkflowScheduled:
 		attr = &SubWorkflowScheduledAttributes{}

internal/history/timer_canceled.go

@@ -0,0 +1,3 @@
+package history
+
+type TimerCanceledAttributes struct{}

internal/history/workflow_cancelled.go renamed to internal/history/workflow_canceled.go

@@ -11,7 +11,7 @@ type Channel[T any] interface {
 
 	Receive(ctx Context) (v T, ok bool)
 
-	ReceiveNonblocking(ctx Context) (v T, ok bool)
+	ReceiveNonBlocking(ctx Context) (v T, ok bool)
 
 	Close()
 }
@@ -21,6 +21,8 @@ type ChannelInternal[T any] interface {
 
 	ReceiveNonBlocking(ctx Context) (v T, ok bool)
 
+	// AddReceiveCallback adds a callback that is called once when a value is sent to the channel. This is similar
+	// to the blocking `Receive` method, but is not blocking a coroutine.
 	AddReceiveCallback(cb func(v T, ok bool))
 }
 
@@ -144,7 +146,7 @@ func (c *channel[T]) Receive(ctx Context) (v T, ok bool) {
 	}
 }
 
-func (c *channel[T]) ReceiveNonblocking(ctx Context) (T, bool) {
+func (c *channel[T]) ReceiveNonBlocking(ctx Context) (T, bool) {
 	if v, ok, rok := c.tryReceive(); rok {
 		return v, ok
 	}
@@ -226,15 +228,6 @@ func (c *channel[T]) AddReceiveCallback(cb func(v T, ok bool)) {
 	c.receivers = append(c.receivers, cb)
 }
 
-func (c *channel[T]) ReceiveNonBlocking(ctx Context) (T, bool) {
-	if v, ok, rok := c.tryReceive(); rok {
-		return v, ok
-	}
-
-	var z T
-	return z, false
-}
-
 func (c *channel[T]) Closed() bool {
 	return c.closed
 }