Merge branch 'main' into fix_redis_diag_panic

Author: decibelcooper

backend/monoprocess/monoprocess.go

@@ -17,7 +17,6 @@ type monoprocessBackend struct {
 
 	workflowSignal chan struct{}
 	activitySignal chan struct{}
-	signalTimeout  time.Duration
 
 	logger *slog.Logger
 }
@@ -106,7 +105,9 @@ func (b *monoprocessBackend) CompleteWorkflowTask(
 		// Note that the worker will be notified even if the timer event gets
 		// cancelled. This is ok, because the poller will simply find no task
 		// and continue.
-		time.AfterFunc(attr.At.Sub(time.Now()), func() { b.notifyWorkflowWorker(context.Background()) })
+		time.AfterFunc(time.Until(attr.At), func() {
+			b.notifyWorkflowWorker(ctx)
+		})
 	}
 
 	b.notifyWorkflowWorker(ctx)

backend/mysql/mysql.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cschleiden/go-workflows/backend/metrics"
 	"github.com/cschleiden/go-workflows/core"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
+	"github.com/cschleiden/go-workflows/internal/workflowerrors"
 	"github.com/cschleiden/go-workflows/workflow"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/google/uuid"
@@ -146,7 +147,7 @@ func (b *mysqlBackend) CreateWorkflowInstance(ctx context.Context, instance *wor
 	defer tx.Rollback()
 
 	// Create workflow instance
-	if err := createInstance(ctx, tx, instance, event.Attributes.(*history.ExecutionStartedAttributes).Metadata, false); err != nil {
+	if err := createInstance(ctx, tx, instance, event.Attributes.(*history.ExecutionStartedAttributes).Metadata); err != nil {
 		return err
 	}
 
@@ -304,7 +305,13 @@ func (b *mysqlBackend) GetWorkflowInstanceState(ctx context.Context, instance *w
 	return state, nil
 }
 
-func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, metadata *workflow.Metadata, ignoreDuplicate bool) error {
+func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, metadata *workflow.Metadata) error {
+	// Check for existing instance
+	if err := tx.QueryRowContext(ctx, "SELECT 1 FROM `instances` WHERE instance_id = ? AND state = ? LIMIT 1", wfi.InstanceID, core.WorkflowInstanceStateActive).
+		Scan(new(int)); err != sql.ErrNoRows {
+		return backend.ErrInstanceAlreadyExists
+	}
+
 	var parentInstanceID, parentExecutionID *string
 	var parentEventID *int64
 	if wfi.SubWorkflow() {
@@ -318,9 +325,9 @@ func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, met
 		return fmt.Errorf("marshaling metadata: %w", err)
 	}
 
-	res, err := tx.ExecContext(
+	_, err = tx.ExecContext(
 		ctx,
-		"INSERT IGNORE INTO `instances` (instance_id, execution_id, parent_instance_id, parent_execution_id, parent_schedule_event_id, metadata, state) VALUES (?, ?, ?, ?, ?, ?, ?)",
+		"INSERT INTO `instances` (instance_id, execution_id, parent_instance_id, parent_execution_id, parent_schedule_event_id, metadata, state) VALUES (?, ?, ?, ?, ?, ?, ?)",
 		wfi.InstanceID,
 		wfi.ExecutionID,
 		parentInstanceID,
@@ -333,17 +340,6 @@ func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, met
 		return fmt.Errorf("inserting workflow instance: %w", err)
 	}
 
-	if !ignoreDuplicate {
-		rows, err := res.RowsAffected()
-		if err != nil {
-			return err
-		}
-
-		if rows != 1 {
-			return backend.ErrInstanceAlreadyExists
-		}
-	}
-
 	return nil
 }
 
@@ -624,23 +620,33 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 	groupedEvents := history.EventsByWorkflowInstance(workflowEvents)
 
 	for targetInstance, events := range groupedEvents {
-		for _, m := range events {
-			if m.HistoryEvent.Type == history.EventType_WorkflowExecutionStarted {
-				a := m.HistoryEvent.Attributes.(*history.ExecutionStartedAttributes)
-				// Create new instance
-				if err := createInstance(ctx, tx, m.WorkflowInstance, a.Metadata, true); err != nil {
-					return err
+		// Are we creating a new sub-workflow instance?
+		m := events[0]
+		if m.HistoryEvent.Type == history.EventType_WorkflowExecutionStarted {
+			a := m.HistoryEvent.Attributes.(*history.ExecutionStartedAttributes)
+			// Create new instance
+			if err := createInstance(ctx, tx, m.WorkflowInstance, a.Metadata); err != nil {
+				if err == backend.ErrInstanceAlreadyExists {
+					if err := insertPendingEvents(ctx, tx, instance, []*history.Event{
+						history.NewPendingEvent(time.Now(), history.EventType_SubWorkflowFailed, &history.SubWorkflowFailedAttributes{
+							Error: workflowerrors.FromError(backend.ErrInstanceAlreadyExists),
+						}, history.ScheduleEventID(m.WorkflowInstance.ParentEventID)),
+					}); err != nil {
+						return fmt.Errorf("inserting sub-workflow failed event: %w", err)
+					}
+
+					continue
 				}
 
-				break
+				return fmt.Errorf("creating sub-workflow instance: %w", err)
 			}
 		}
 
+		// Insert pending events for target instance
 		historyEvents := []*history.Event{}
 		for _, m := range events {
 			historyEvents = append(historyEvents, m.HistoryEvent)
 		}
-
 		if err := insertPendingEvents(ctx, tx, &targetInstance, historyEvents); err != nil {
 			return fmt.Errorf("inserting messages: %w", err)
 		}

backend/redis/delete.go

@@ -11,11 +11,13 @@ import (
 // KEYS[1] - instance key
 // KEYS[2] - pending events key
 // KEYS[3] - history key
-// KEYS[4] - instances-by-creation key
+// KEYS[4] - payload key
+// KEYS[5] - active-instance-execution key
+// KEYS[6] - instances-by-creation key
 // ARGV[1] - instance segment
 var deleteCmd = redis.NewScript(
-	`redis.call("DEL", KEYS[1], KEYS[2], KEYS[3])
-	return redis.call("ZREM", KEYS[4], ARGV[1])`)
+	`redis.call("DEL", KEYS[1], KEYS[2], KEYS[3], KEYS[4], KEYS[5])
+	return redis.call("ZREM", KEYS[6], ARGV[1])`)
 
 // deleteInstance deletes an instance from Redis. It does not attempt to remove any future events or pending
 // workflow tasks. It's assumed that the instance is in the finished state.
@@ -26,6 +28,8 @@ func deleteInstance(ctx context.Context, rdb redis.UniversalClient, instance *co
 		instanceKey(instance),
 		pendingEventsKey(instance),
 		historyKey(instance),
+		payloadKey(instance),
+		activeInstanceExecutionKey(instance.InstanceID),
 		instancesByCreation(),
 	}, instanceSegment(instance)).Err(); err != nil {
 		return fmt.Errorf("failed to delete instance: %w", err)

backend/redis/events.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"strconv"
 
 	"github.com/cschleiden/go-workflows/backend/history"
 	"github.com/cschleiden/go-workflows/core"
@@ -73,89 +72,3 @@ func addEventToStreamP(ctx context.Context, p redis.Pipeliner, streamKey string,
 		},
 	}).Err()
 }
-
-// addEventsToStream adds the given events to the given event stream. If successful, the message id of the last event added
-// is returned
-// KEYS[1] - stream key
-// ARGV[1] - event data as serialized strings
-var addEventsToStreamCmd = redis.NewScript(`
-	local msgID = ""
-	for i = 1, #ARGV, 2 do
-		msgID = redis.call("XADD", KEYS[1], ARGV[i], "event", ARGV[i + 1])
-	end
-	return msgID
-`)
-
-func addEventsToStreamP(ctx context.Context, p redis.Pipeliner, streamKey string, events []*history.Event) error {
-	eventsData := make([]string, 0)
-	for _, event := range events {
-		eventData, err := marshalEventWithoutAttributes(event)
-		if err != nil {
-			return err
-		}
-
-		// log.Println("addEventsToHistoryStreamP:", event.SequenceID, string(eventData))
-
-		eventsData = append(eventsData, historyID(event.SequenceID))
-		eventsData = append(eventsData, string(eventData))
-	}
-
-	addEventsToStreamCmd.Run(ctx, p, []string{streamKey}, eventsData)
-
-	return nil
-}
-
-// Adds an event to be delivered in the future. Not cluster-safe.
-// KEYS[1] - future event zset key
-// KEYS[2] - future event key
-// KEYS[3] - instance payload key
-// ARGV[1] - timestamp
-// ARGV[2] - Instance segment
-// ARGV[3] - event id
-// ARGV[4] - event data
-// ARGV[5] - event payload
-var addFutureEventCmd = redis.NewScript(`
-	redis.call("ZADD", KEYS[1], ARGV[1], KEYS[2])
-	redis.call("HSET", KEYS[2], "instance", ARGV[2], "id", ARGV[3], "event", ARGV[4])
-	redis.call("HSETNX", KEYS[3], ARGV[3], ARGV[5])
-	return 0
-`)
-
-func addFutureEventP(ctx context.Context, p redis.Pipeliner, instance *core.WorkflowInstance, event *history.Event) error {
-	eventData, err := marshalEventWithoutAttributes(event)
-	if err != nil {
-		return err
-	}
-
-	payloadEventData, err := json.Marshal(event.Attributes)
-	if err != nil {
-		return err
-	}
-
-	return addFutureEventCmd.Run(
-		ctx, p,
-		[]string{futureEventsKey(), futureEventKey(instance, event.ScheduleEventID), payloadKey(instance)},
-		strconv.FormatInt(event.VisibleAt.UnixMilli(), 10),
-		instanceSegment(instance),
-		event.ID,
-		string(eventData),
-		string(payloadEventData),
-	).Err()
-}
-
-// Remove a scheduled future event. Not cluster-safe.
-// KEYS[1] - future event zset key
-// KEYS[2] - future event key
-// KEYS[3] - instance payload key
-var removeFutureEventCmd = redis.NewScript(`
-	redis.call("ZREM", KEYS[1], KEYS[2])
-	local eventID = redis.call("HGET", KEYS[2], "id")
-	redis.call("HDEL", KEYS[3], eventID)
-	return redis.call("DEL", KEYS[2])
-`)
-
-// removeFutureEvent removes a scheduled future event for the given event. Events are associated via their ScheduleEventID
-func removeFutureEventP(ctx context.Context, p redis.Pipeliner, instance *core.WorkflowInstance, event *history.Event) {
-	key := futureEventKey(instance, event.ScheduleEventID)
-	removeFutureEventCmd.Run(ctx, p, []string{futureEventsKey(), key, payloadKey(instance)})
-}

backend/redis/events_future.go

@@ -0,0 +1,64 @@
+package redis
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	redis "github.com/redis/go-redis/v9"
+)
+
+// Find all due future events. For each event:
+// - Look up event data
+// - Add to pending event stream for workflow instance
+// - Try to queue workflow task for workflow instance
+// - Remove event from future event set and delete event data
+//
+// KEYS[1] - future event set key
+// KEYS[2] - workflow task queue stream
+// KEYS[3] - workflow task queue set
+// ARGV[1] - current timestamp for zrange
+//
+// 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")
+	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
+			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])
+	end
+
+	return #events
+`)
+
+func scheduleFutureEvents(ctx context.Context, rb *redisBackend) error {
+	now := time.Now().UnixMilli()
+	nowStr := strconv.FormatInt(now, 10)
+
+	queueKeys := rb.workflowQueue.Keys()
+
+	if _, 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)
+	}
+
+	return nil
+}

backend/redis/expire.go

@@ -17,6 +17,7 @@ import (
 // KEYS[3] - instance key
 // KEYS[4] - pending events key
 // KEYS[5] - history key
+// KEYS[6] - payload key
 // ARGV[1] - current timestamp
 // ARGV[2] - expiration time in seconds
 // ARGV[3] - expiration timestamp in unix milliseconds
@@ -55,6 +56,7 @@ func setWorkflowInstanceExpiration(ctx context.Context, rdb redis.UniversalClien
 		instanceKey(instance),
 		pendingEventsKey(instance),
 		historyKey(instance),
+		payloadKey(instance),
 	},
 		nowStr,
 		expiration.Seconds(),