Merge pull request #293 from cschleiden/redis-atomic-create-instance

Refactor workflow instance creation

Author: cschleiden

backend/redis/diagnostics.go

@@ -52,6 +52,10 @@ func (rb *redisBackend) GetWorkflowInstances(ctx context.Context, afterInstanceI
 		instanceKeys = append(instanceKeys, instanceKeyFromSegment(instanceSegment))
 	}
 
+	if len(instanceKeys) == 0 {
+		return nil, nil
+	}
+
 	instances, err := rb.rdb.MGet(ctx, instanceKeys...).Result()
 	if err != nil {
 		return nil, fmt.Errorf("getting instances: %w", err)

backend/redis/instance.go

@@ -15,41 +15,60 @@ import (
 )
 
 func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
-	activeInstance, err := readActiveInstanceExecution(ctx, rb.rdb, instance.InstanceID)
-	if err != nil && err != backend.ErrInstanceNotFound {
-		return err
-	}
+	keyInfo := rb.workflowQueue.Keys()
 
-	if activeInstance != nil {
-		return backend.ErrInstanceAlreadyExists
+	instanceState, err := json.Marshal(&instanceState{
+		Instance:  instance,
+		State:     core.WorkflowInstanceStateActive,
+		Metadata:  event.Attributes.(*history.ExecutionStartedAttributes).Metadata,
+		CreatedAt: time.Now(),
+	})
+	if err != nil {
+		return fmt.Errorf("marshaling instance state: %w", err)
 	}
 
-	p := rb.rdb.TxPipeline()
-
-	if err := createInstanceP(ctx, p, instance, event.Attributes.(*history.ExecutionStartedAttributes).Metadata); err != nil {
-		return err
+	activeInstance, err := json.Marshal(instance)
+	if err != nil {
+		return fmt.Errorf("marshaling instance: %w", err)
 	}
 
-	// Create event stream with initial event
-	if err := addEventPayloadsP(ctx, p, instance, []*history.Event{event}); err != nil {
-		return fmt.Errorf("adding event payloads: %w", err)
+	eventData, err := marshalEventWithoutAttributes(event)
+	if err != nil {
+		return fmt.Errorf("marshaling event: %w", err)
 	}
 
-	if err := addEventToStreamP(ctx, p, pendingEventsKey(instance), event); err != nil {
-		return fmt.Errorf("adding event to stream: %w", err)
+	payloadData, err := json.Marshal(event.Attributes)
+	if err != nil {
+		return fmt.Errorf("marshaling event payload: %w", err)
 	}
 
-	// Queue workflow instance task
-	if err := rb.workflowQueue.Enqueue(ctx, p, instanceSegment(instance), nil); err != nil {
-		return fmt.Errorf("queueing workflow task: %w", err)
-	}
+	_, err = createWorkflowInstanceCmd.Run(ctx, rb.rdb, []string{
+		instanceKey(instance),
+		activeInstanceExecutionKey(instance.InstanceID),
+		pendingEventsKey(instance),
+		payloadKey(instance),
+		instancesActive(),
+		keyInfo.SetKey,
+		keyInfo.StreamKey,
+	},
+		instanceSegment(instance),
+		string(instanceState),
+		string(activeInstance),
+		event.ID,
+		eventData,
+		payloadData,
+	).Result()
+
+	if err != nil {
+		if _, ok := err.(redis.Error); ok {
+			if err.Error() == "ERR InstanceAlreadyExists" {
+				return backend.ErrInstanceAlreadyExists
+			}
+		}
 
-	if _, err := p.Exec(ctx); err != nil {
 		return fmt.Errorf("creating workflow instance: %w", err)
 	}
 
-	rb.options.Logger.Debug("Created new workflow instance")
-
 	return nil
 }
 

backend/redis/redis.go

@@ -24,7 +24,10 @@ var _ backend.Backend = (*redisBackend)(nil)
 //go:embed scripts/*.lua
 var luaScripts embed.FS
 
-var completeWorkflowTaskCmd *redis.Script
+var (
+	createWorkflowInstanceCmd *redis.Script
+	completeWorkflowTaskCmd   *redis.Script
+)
 
 func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (*redisBackend, error) {
 	workflowQueue, err := newTaskQueue[any](client, "workflows")
@@ -75,7 +78,8 @@ func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (
 	// Load all Lua scripts
 
 	cmdMapping := map[string]**redis.Script{
-		"complete_workflow_task.lua": &completeWorkflowTaskCmd,
+		"create_workflow_instance.lua": &createWorkflowInstanceCmd,
+		"complete_workflow_task.lua":   &completeWorkflowTaskCmd,
 	}
 
 	for scriptFile, cmd := range cmdMapping {

backend/redis/scripts/create_workflow_instance.lua

@@ -0,0 +1,59 @@
+local keyIdx = 1
+local argvIdx = 1
+
+local getKey = function()
+    local key = KEYS[keyIdx]
+    keyIdx = keyIdx + 1
+    return key
+end
+
+local getArgv = function()
+    local argv = ARGV[argvIdx]
+    argvIdx = argvIdx + 1
+    return argv
+end
+
+local instanceKey = getKey()
+local activeInstanceExecutionKey = getKey()
+local pendingEventsKey = getKey()
+local payloadHashKey = getKey()
+
+local instancesActiveKey = getKey()
+
+local workflowSetKey = getKey()
+local workflowStreamKey = getKey()
+
+local instanceSegment = getArgv()
+
+-- Is there an existing instance with active execution?
+local instanceExists = redis.call("EXISTS", activeInstanceExecutionKey)
+if instanceExists == 1 then
+  return redis.error_reply("ERR InstanceAlreadyExists")
+end
+
+-- Create new instance
+local instanceState = getArgv()
+redis.call("SETNX", instanceKey, instanceState)
+
+-- Set active execution
+local activeInstanceExecutionState = getArgv()
+redis.call("SET", activeInstanceExecutionKey, activeInstanceExecutionState)
+
+-- Track active instance
+redis.call("SADD", instancesActiveKey, instanceSegment)
+
+-- add initial event & payload
+local eventId = getArgv()
+local eventData = getArgv()
+redis.call("XADD", pendingEventsKey, "*", "event", eventData)
+
+local payload = getArgv()
+redis.pcall("HSETNX", payloadHashKey, eventId, payload)
+
+-- queue workflow task
+local added = redis.call("SADD", workflowSetKey, instanceSegment)
+if added == 1 then
+    redis.call("XADD", workflowStreamKey, "*", "id", instanceSegment, "data", "")
+end
+
+return true

client/client.go

@@ -85,7 +85,12 @@ func (c *Client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 		return nil, fmt.Errorf("creating workflow instance: %w", err)
 	}
 
-	c.backend.Logger().Debug("Created workflow instance", log.InstanceIDKey, wfi.InstanceID, log.ExecutionIDKey, wfi.ExecutionID)
+	c.backend.Logger().Debug(
+		"Created workflow instance",
+		log.InstanceIDKey, wfi.InstanceID,
+		log.ExecutionIDKey, wfi.ExecutionID,
+		log.WorkflowNameKey, workflowName,
+	)
 
 	c.backend.Metrics().Counter(metrickeys.WorkflowInstanceCreated, metrics.Tags{}, 1)