Merge pull request #292 from cschleiden/redis-fix-active-key

Use correct key for checking whether an instance exists

Author: cschleiden

backend/redis/scripts/complete_workflow_task.lua

@@ -52,6 +52,38 @@ for i = 1, executedEvents do
     lastSequenceId = tonumber(sequenceId)
 end
 
+-- Remove executed pending events
+local lastPendingEventMessageId = getArgv()
+redis.call("XTRIM", pendingEventsKey, "MINID", lastPendingEventMessageId)
+redis.call("XDEL", pendingEventsKey, lastPendingEventMessageId)
+
+-- Update instance state
+local now = getArgv()
+local state = tonumber(getArgv())
+
+-- State constants
+local ContinuedAsNew = tonumber(getArgv())
+local Finished = tonumber(getArgv())
+
+instance["state"] = state
+
+-- If workflow instance finished, remove active execution
+local activeInstanceExecutionKey = getKey()
+if state == ContinuedAsNew or state == Finished then
+    -- Remove active execution
+    redis.call("DEL", activeInstanceExecutionKey)
+
+    instance["completed_at"] = now
+
+    redis.call("SREM", activeInstancesKey, instanceSegment)
+end
+
+if lastSequenceId > 0 then
+    instance["last_sequence_id"] = lastSequenceId
+end
+
+redis.call("SET", instanceKey, cjson.encode(instance))
+
 -- Remove canceled timers
 local timersToCancel = tonumber(getArgv())
 for i = 1, timersToCancel do
@@ -80,6 +112,20 @@ for i = 1, timersToSchedule do
 	storePayload(eventId, payloadData)
 end
 
+-- Schedule activities
+local activities = tonumber(getArgv())
+local activitySetKey = getKey()
+local activityStreamKey = getKey()
+for i = 1, activities do
+    local activityId = getArgv()
+    local activityData = getArgv()
+
+    local added = redis.call("SADD", activitySetKey, activityId)
+	if added == 1 then
+		redis.call("XADD", activityStreamKey, "*", "id", activityId, "data", activityData)
+	end
+end
+
 -- Send events to other workflow instances
 local otherWorkflowInstances = tonumber(getArgv())
 for i = 1, otherWorkflowInstances do
@@ -102,7 +148,7 @@ for i = 1, otherWorkflowInstances do
         local conflictEventPayloadData = getArgv()
 
         -- Does the instance exist already?
-        local instanceExists = redis.call("EXISTS", targetActiveInstanceExecutionState)
+        local instanceExists = redis.call("EXISTS", targetActiveInstanceExecutionKey)
         if instanceExists == 1 then
             redis.call("XADD", pendingEventsKey, "*", "event", conflictEventData)
             storePayload(conflictEventId, conflictEventPayloadData)
@@ -147,61 +193,14 @@ for i = 1, otherWorkflowInstances do
     end
 end
 
--- Update instance state
-local now = getArgv()
-local state = tonumber(getArgv())
-
--- State constants
-local ContinuedAsNew = tonumber(getArgv())
-local Finished = tonumber(getArgv())
-
-instance["state"] = state
-
--- If workflow instance finished, remove active execution
-local activeInstanceExecutionKey = getKey()
-if state == ContinuedAsNew or state == Finished then
-    -- Remove active execution
-    redis.call("DEL", activeInstanceExecutionKey)
-
-    instance["completed_at"] = now
-
-    redis.call("SREM", activeInstancesKey, instanceSegment)
-end
-
-if lastSequenceId > 0 then
-    instance["last_sequence_id"] = lastSequenceId
-end
-
-redis.call("SET", instanceKey, cjson.encode(instance))
-
--- Schedule activities
-local activities = tonumber(getArgv())
-local activitySetKey = getKey()
-local activityStreamKey = getKey()
-for i = 1, activities do
-    local activityId = getArgv()
-    local activityData = getArgv()
-
-    local added = redis.call("SADD", activitySetKey, activityId)
-	if added == 1 then
-		redis.call("XADD", activityStreamKey, "*", "id", activityId, "data", activityData)
-	end
-end
-
--- Remove executed pending events
-local lastPendingEventMessageId = getArgv()
-redis.call("XTRIM", pendingEventsKey, "MINID", lastPendingEventMessageId)
-redis.call("XDEL", pendingEventsKey, lastPendingEventMessageId)
-
--- Complete workflow task and unlock instance
+-- Complete workflow task and mark instance task as completed
 local taskId = getArgv()
 local groupName = getArgv()
 local task = redis.call("XRANGE", workflowStreamKey, taskId, taskId)
 if #task ~= 0 then
     local id = task[1][2][2]
     redis.call("SREM", workflowSetKey, id)
     redis.call("XACK", workflowStreamKey, groupName, taskId)
-
     redis.call("XDEL", workflowStreamKey, taskId)
 end
 

backend/redis/workflow.go

@@ -132,6 +132,15 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		args = append(args, event.ID, historyID(event.SequenceID), eventData, payloadData, event.SequenceID)
 	}
 
+	// Remove executed pending events
+	lastPendingEventMessageID := task.CustomData.(string)
+	args = append(args, lastPendingEventMessageID)
+
+	// Update instance state and update active execution
+	nowStr := time.Now().Format(time.RFC3339)
+	args = append(args, string(nowStr), int(state), int(core.WorkflowInstanceStateContinuedAsNew), int(core.WorkflowInstanceStateFinished))
+	keys = append(keys, activeInstanceExecutionKey(instance.InstanceID))
+
 	// Remove canceled timers
 	timersToCancel := make([]*history.Event, 0)
 	for _, event := range executedEvents {
@@ -163,14 +172,30 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		keys = append(keys, futureEventKey(instance, timerEvent.ScheduleEventID))
 	}
 
+	// Schedule activities
+	args = append(args, len(activityEvents))
+	activityQueueKeys := rb.activityQueue.Keys()
+	keys = append(keys, activityQueueKeys.SetKey, activityQueueKeys.StreamKey)
+	for _, activityEvent := range activityEvents {
+		activityData, err := json.Marshal(&activityData{
+			Instance: instance,
+			ID:       activityEvent.ID,
+			Event:    activityEvent,
+		})
+		if err != nil {
+			return fmt.Errorf("marshaling activity data: %w", err)
+		}
+		args = append(args, activityEvent.ID, activityData)
+	}
+
 	// Send new workflow events to the respective streams
 	groupedEvents := history.EventsByWorkflowInstance(workflowEvents)
 	args = append(args, len(groupedEvents))
 	for targetInstance, events := range groupedEvents {
 		keys = append(keys, instanceKey(&targetInstance), activeInstanceExecutionKey(targetInstance.InstanceID))
 		args = append(args, instanceSegment(&targetInstance), targetInstance.InstanceID)
 
-		// Are we creating a new sub-workflow instance?
+		// Are we creating a new workflow instance?
 		m := events[0]
 		createNewInstance := m.HistoryEvent.Type == history.EventType_WorkflowExecutionStarted
 		args = append(args, createNewInstance)
@@ -218,31 +243,6 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		}
 	}
 
-	// Update instance state and update active execution
-	nowStr := time.Now().Format(time.RFC3339)
-	args = append(args, string(nowStr), int(state), int(core.WorkflowInstanceStateContinuedAsNew), int(core.WorkflowInstanceStateFinished))
-	keys = append(keys, activeInstanceExecutionKey(instance.InstanceID))
-
-	// Store activity data
-	args = append(args, len(activityEvents))
-	activityQueueKeys := rb.activityQueue.Keys()
-	keys = append(keys, activityQueueKeys.SetKey, activityQueueKeys.StreamKey)
-	for _, activityEvent := range activityEvents {
-		activityData, err := json.Marshal(&activityData{
-			Instance: instance,
-			ID:       activityEvent.ID,
-			Event:    activityEvent,
-		})
-		if err != nil {
-			return fmt.Errorf("marshaling activity data: %w", err)
-		}
-		args = append(args, activityEvent.ID, activityData)
-	}
-
-	// Remove executed pending events
-	lastPendingEventMessageID := task.CustomData.(string)
-	args = append(args, lastPendingEventMessageID)
-
 	// Complete workflow task and unlock instance.
 	args = append(args, task.ID, rb.workflowQueue.groupName)