Discard events for finished instances

Author: cschleiden

backend/redis/workflow.go

@@ -100,23 +100,6 @@ func (rb *redisBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, er
 		newEvents = append(newEvents, event)
 	}
 
-	if instanceState.State == core.WorkflowInstanceStateFinished {
-		l := rb.Logger().With(
-			"task_id", instanceTask.TaskID,
-			"id", instanceTask.ID,
-			"instance_id", instanceState.Instance.InstanceID)
-
-		// This should never happen. For now, log information and then panic.
-		l.Error("got workflow task for finished workflow instance")
-
-		// Log events that lead to this task
-		for _, event := range newEvents {
-			l.Error("pending_event", "id", event.ID, "event_type", event.Type.String(), "schedule_event_id", event.ScheduleEventID)
-		}
-
-		panic("Dequeued already finished workflow instance task")
-	}
-
 	return &task.Workflow{
 		ID:                    instanceTask.TaskID,
 		WorkflowInstance:      instanceState.Instance,

backend/test/e2e.go

@@ -155,6 +155,23 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 				require.Equal(t, 7, r)
 			},
 		},
+		{
+			name: "Signal_after_completion",
+			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
+				wf := func(ctx workflow.Context) error {
+					return nil
+				}
+				register(t, ctx, w, []interface{}{wf}, nil)
+
+				// Run workflow to completion
+				instance := runWorkflow(t, ctx, c, wf)
+				_, err := client.GetWorkflowResult[int](ctx, c, instance, time.Second*20)
+				require.NoError(t, err)
+
+				err = c.SignalWorkflow(ctx, instance.InstanceID, "signal", nil)
+				require.NoError(t, err)
+			},
+		},
 		{
 			name: "SubWorkflow_Simple",
 			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {

internal/workflow/executor.go

@@ -99,6 +99,18 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 
 	logger.Debug("Executing workflow task", "task_last_sequence_id", t.LastSequenceID)
 
+	if t.WorkflowInstanceState == core.WorkflowInstanceStateFinished {
+		// This should never happen. For now, log information and then panic.
+		logger.Debug("Received workflow task for finished workflow instance, discarding events")
+
+		// Log events that caused this task to be scheduled
+		for _, event := range t.NewEvents {
+			logger.Debug("Discarded event:", "id", event.ID, "event_type", event.Type.String(), "schedule_event_id", event.ScheduleEventID)
+		}
+
+		return &ExecutionResult{}, nil
+	}
+
 	skipNewEvents := false
 
 	if t.LastSequenceID > e.lastSequenceID {