Fix timer schedule/cancel race

Author: cschleiden

backend/redis/events_future.go

@@ -3,6 +3,7 @@ package redis
 import (
 	"context"
 	"fmt"
+	"log"
 	"strconv"
 	"time"
 
@@ -23,24 +24,25 @@ import (
 // Note: this does not work with Redis Cluster since not all keys are passed into the script.
 var futureEventsCmd = redis.NewScript(`
 	-- Find events which should become visible now
-	local events = redis.call("ZRANGE", KEYS[1], "-inf", ARGV[1], "BYSCORE")
+	local now = ARGV[1]
+	local events = redis.call("ZRANGE", KEYS[1], "-inf", now, "BYSCORE")
 	for i = 1, #events do
 		local instanceSegment = redis.call("HGET", events[i], "instance")
 
-		-- Add event to pending event stream
-		local eventData = redis.call("HGET", events[i], "event")
-		local pending_events_key = "pending-events:" .. instanceSegment
-		redis.call("XADD", pending_events_key, "*", "event", eventData)
-
-		-- Try to queue workflow task
-		local already_queued = redis.call("SADD", KEYS[3], instanceSegment)
-		if already_queued ~= 0 then
+		-- Try to queue workflow task. If a workflow task is already queued, ignore this event for now.
+		local added = redis.call("SADD", KEYS[3], instanceSegment)
+		if added == 1 then
 			redis.call("XADD", KEYS[2], "*", "id", instanceSegment, "data", "")
-		end
 
-		-- Delete event hash data
-		redis.call("DEL", events[i])
-		redis.call("ZREM", KEYS[1], events[i])
+			-- Add event to pending event stream
+			local eventData = redis.call("HGET", events[i], "event")
+			local pending_events_key = "pending-events:" .. instanceSegment
+			redis.call("XADD", pending_events_key, "*", "event", eventData)
+
+			-- Delete event hash data
+			redis.call("DEL", events[i])
+			redis.call("ZREM", KEYS[1], events[i])
+		end
 	end
 
 	return #events
@@ -52,12 +54,14 @@ func scheduleFutureEvents(ctx context.Context, rb *redisBackend) error {
 
 	queueKeys := rb.workflowQueue.Keys()
 
-	if _, err := futureEventsCmd.Run(ctx, rb.rdb, []string{
+	if r, err := futureEventsCmd.Run(ctx, rb.rdb, []string{
 		futureEventsKey(),
 		queueKeys.StreamKey,
 		queueKeys.SetKey,
 	}, nowStr).Result(); err != nil && err != redis.Nil {
 		return fmt.Errorf("checking future events: %w", err)
+	} else {
+		log.Println("Scheduled", r.(int64), "future events")
 	}
 
 	return nil

backend/redis/scripts/complete_workflow_task.lua

@@ -91,12 +91,16 @@ local timersToCancel = tonumber(getArgv())
 for i = 1, timersToCancel do
     local futureEventKey = getKey()
 
-    redis.call("ZREM", futureEventZSetKey, futureEventKey)
-    -- remove payload
-    local eventId = redis.call("HGET", futureEventKey, "id")
-    redis.call("HDEL", payloadHashKey, eventId)
-    -- remove event hash
-    redis.call("DEL", futureEventKey)
+    local eventRemoved = redis.call("ZREM", futureEventZSetKey, futureEventKey)
+    -- Event might've become visible while this task was being processed, in that
+    -- case it would be already removed from futureEventZSetKey
+    if eventRemoved == 1 then
+        -- remove payload
+        local eventId = redis.call("HGET", futureEventKey, "id")
+        redis.call("HDEL", payloadHashKey, eventId)
+        -- remove event hash
+        redis.call("DEL", futureEventKey)
+    end
 end
 
 -- Schedule timers