Merge pull request #105 from cschleiden/fix-redis-get-history

Honor `lastSequenceID` parameter when retrieving history from redis backend

Author: cschleiden

backend/redis/events.go

@@ -31,20 +31,21 @@ func addEventToStreamP(ctx context.Context, p redis.Pipeliner, streamKey string,
 // ARGV[1] - event data as serialized strings
 var addEventsToStreamCmd = redis.NewScript(`
 	local msgID = ""
-	for i = 1, #ARGV do
-		msgID = redis.call("XADD", KEYS[1], "*", "event", ARGV[i])
+	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 {
+func addEventsToHistoryStreamP(ctx context.Context, p redis.Pipeliner, streamKey string, events []history.Event) error {
 	eventsData := make([]string, 0)
 	for _, event := range events {
 		eventData, err := json.Marshal(event)
 		if err != nil {
 			return err
 		}
 
+		eventsData = append(eventsData, historyID(event.SequenceID))
 		eventsData = append(eventsData, string(eventData))
 	}
 

backend/redis/instance.go

@@ -58,7 +58,13 @@ func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, instance *wo
 }
 
 func (rb *redisBackend) GetWorkflowInstanceHistory(ctx context.Context, instance *core.WorkflowInstance, lastSequenceID *int64) ([]history.Event, error) {
-	msgs, err := rb.rdb.XRange(ctx, historyKey(instance.InstanceID), "-", "+").Result()
+	start := "-"
+
+	if lastSequenceID != nil {
+		start = "(" + historyID(*lastSequenceID)
+	}
+
+	msgs, err := rb.rdb.XRange(ctx, historyKey(instance.InstanceID), start, "+").Result()
 	if err != nil {
 		return nil, err
 	}

backend/redis/keys.go

@@ -20,6 +20,10 @@ func historyKey(instanceID string) string {
 	return fmt.Sprintf("history:%v", instanceID)
 }
 
+func historyID(sequenceID int64) string {
+	return fmt.Sprintf("%v-0", sequenceID)
+}
+
 func futureEventsKey() string {
 	return "future-events"
 }

backend/redis/queue.go

@@ -3,7 +3,6 @@ package redis
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"time"
 
@@ -30,8 +29,6 @@ type TaskItem[T any] struct {
 	Data T
 }
 
-var errTaskAlreadyInQueue = errors.New("task already in queue")
-
 type KeyInfo struct {
 	StreamKey string
 	SetKey    string
@@ -80,7 +77,7 @@ func (q *taskQueue[T]) Keys() KeyInfo {
 	}
 }
 
-// KEYS[1] = stream
+// KEYS[1] = set
 // KEYS[2] = stream
 // ARGV[1] = caller provided id of the task
 // ARGV[2] = additional data to store with the task

backend/redis/queue_test.go

@@ -73,7 +73,7 @@ func Test_TaskQueue(t *testing.T) {
 				_, err = client.Pipelined(ctx, func(p redis.Pipeliner) error {
 					return q.Enqueue(ctx, p, "t1", nil)
 				})
-				require.Error(t, errTaskAlreadyInQueue, err)
+				require.NoError(t, err)
 
 				task, err := q.Dequeue(ctx, client, lockTimeout, blockTimeout)
 				require.NoError(t, err)

backend/redis/signal.go

@@ -25,17 +25,11 @@ func (rb *redisBackend) SignalWorkflow(ctx context.Context, instanceID string, e
 	))
 	defer span.End()
 
-	if _, err = rb.rdb.Pipelined(ctx, func(p redis.Pipeliner) error {
-		if err := addEventToStreamP(ctx, p, pendingEventsKey(instanceID), &event); err != nil {
+	if _, err = rb.rdb.TxPipelined(ctx, func(p redis.Pipeliner) error {
+		if err := rb.addWorkflowInstanceEventP(ctx, p, instanceState.Instance, &event); err != nil {
 			return fmt.Errorf("adding event to stream: %w", err)
 		}
 
-		if err := rb.workflowQueue.Enqueue(ctx, p, instanceID, nil); err != nil {
-			if err != errTaskAlreadyInQueue {
-				return fmt.Errorf("queueing workflow task: %w", err)
-			}
-		}
-
 		return nil
 	}); err != nil {
 		return err

backend/redis/workflow.go

@@ -104,7 +104,8 @@ func (rb *redisBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, er
 
 	if instanceState.State == backend.WorkflowStateFinished {
 		l := rb.Logger().With(
-			"task_id", instanceTask.ID,
+			"task_id", instanceTask.TaskID,
+			"id", instanceTask.ID,
 			"instance_id", instanceState.Instance.InstanceID)
 
 		// This should never happen. For now, log information and then panic.
@@ -153,8 +154,8 @@ var removePendingEventsCmd = redis.NewScript(`
 var requeueInstanceCmd = redis.NewScript(`
 	local pending_events = redis.call("XLEN", KEYS[1])
 	if pending_events > 0 then
-		local already_queued = redis.call("SADD", KEYS[3], ARGV[1])
-		if already_queued ~= 0 then
+		local added = redis.call("SADD", KEYS[3], ARGV[1])
+		if added == 1 then
 			redis.call("XADD", KEYS[2], "*", "id", ARGV[1], "data", "")
 		end
 	end
@@ -181,7 +182,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 	p := rb.rdb.TxPipeline()
 
 	// Add executed events to the history
-	if err := addEventsToStreamP(ctx, p, historyKey(instance.InstanceID), executedEvents); err != nil {
+	if err := addEventsToHistoryStreamP(ctx, p, historyKey(instance.InstanceID), executedEvents); err != nil {
 		return fmt.Errorf("serializing : %w", err)
 	}
 
@@ -192,7 +193,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		}
 	}
 
-	// Scheduler timers
+	// Schedule timers
 	for _, timerEvent := range timerEvents {
 		if err := addFutureEventP(ctx, p, instance, &timerEvent); err != nil {
 			return err
@@ -220,12 +221,10 @@ func (rb *redisBackend) CompleteWorkflowTask(
 			}
 		}
 
-		// If any pending message was added, try to queue workflow task
+		// Try to queue workflow task
 		if targetInstance != instance {
 			if err := rb.workflowQueue.Enqueue(ctx, p, targetInstance.InstanceID, nil); err != nil {
-				if err != errTaskAlreadyInQueue {
-					return fmt.Errorf("adding instance to locked instances set: %w", err)
-				}
+				return fmt.Errorf("enqueuing workflow task: %w", err)
 			}
 		}
 	}

internal/workflow/executor.go

@@ -97,12 +97,12 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 
 	logger := e.logger.With("task_id", t.ID, "instance_id", t.WorkflowInstance.InstanceID)
 
-	logger.Debug("Executing workflow task")
+	logger.Debug("Executing workflow task", "task_last_sequence_id", t.LastSequenceID)
 
 	skipNewEvents := false
 
 	if t.LastSequenceID > e.lastSequenceID {
-		logger.Debug("Task has newer history than current state, fetching and replaying history", "task_sequence_id", t.LastSequenceID, "sequence_id", e.lastSequenceID)
+		logger.Debug("Task has newer history than current state, fetching and replaying history", "task_sequence_id", t.LastSequenceID, "local_sequence_id", e.lastSequenceID)
 
 		h, err := e.historyProvider.GetWorkflowInstanceHistory(ctx, t.WorkflowInstance, &e.lastSequenceID)
 		if err != nil {
@@ -188,6 +188,10 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 func (e *executor) replayHistory(history []history.Event) error {
 	e.workflowState.SetReplaying(true)
 	for _, event := range history {
+		if event.SequenceID < e.lastSequenceID {
+			panic("history has older events than current state")
+		}
+
 		if err := e.executeEvent(event); err != nil {
 			return err
 		}
@@ -216,6 +220,8 @@ func (e *executor) executeNewEvents(newEvents []history.Event) ([]history.Event,
 
 func (e *executor) Close() {
 	if e.workflow != nil {
+		e.logger.Debug("Stopping workflow executor", "instance_id", e.workflowState.Instance().InstanceID)
+
 		// End workflow if running to prevent leaking goroutines
 		e.workflow.Close()
 	}