Merge pull request #138 from cschleiden/signal-after-completion-fix

Do not reset workflow instance state when receiving signal after comp…

Author: cschleiden

backend/mysql/mysql.go

@@ -356,10 +356,11 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, err
 	}
 
 	t := &task.Workflow{
-		ID:               wfi.InstanceID,
-		WorkflowInstance: wfi,
-		Metadata:         metadata,
-		NewEvents:        []history.Event{},
+		ID:                    wfi.InstanceID,
+		WorkflowInstance:      wfi,
+		WorkflowInstanceState: core.WorkflowInstanceStateActive,
+		Metadata:              metadata,
+		NewEvents:             []history.Event{},
 	}
 
 	// Get new events

backend/sqlite/sqlite.go

@@ -287,10 +287,11 @@ func (sb *sqliteBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, e
 	}
 
 	t := &task.Workflow{
-		ID:               wfi.InstanceID,
-		WorkflowInstance: wfi,
-		Metadata:         metadata,
-		NewEvents:        []history.Event{},
+		ID:                    wfi.InstanceID,
+		WorkflowInstance:      wfi,
+		WorkflowInstanceState: core.WorkflowInstanceStateActive,
+		Metadata:              metadata,
+		NewEvents:             []history.Event{},
 	}
 
 	// Get new events

backend/test/e2e.go

@@ -170,6 +170,13 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 
 				err = c.SignalWorkflow(ctx, instance.InstanceID, "signal", nil)
 				require.NoError(t, err)
+				_, err = client.GetWorkflowResult[int](ctx, c, instance, time.Millisecond*5)
+				require.NoError(t, err)
+
+				err = c.SignalWorkflow(ctx, instance.InstanceID, "signal", nil)
+				require.NoError(t, err)
+				_, err = client.GetWorkflowResult[int](ctx, c, instance, time.Millisecond*5)
+				require.NoError(t, err)
 			},
 		},
 		{

internal/metrickeys/metrickeys.go

@@ -28,5 +28,7 @@ const (
 	// Reason for evicting an entry from the workflow instance cache
 	EvictionReason = "reason"
 
+	SubWorkflow = "subworkflow"
+
 	ActivityName = "activity"
 )

internal/worker/workflow.go

@@ -89,6 +89,7 @@ func (ww *WorkflowWorker) runPoll(ctx context.Context) {
 			}
 
 			if task != nil {
+				ww.wg.Add(1)
 				ww.workflowTaskQueue <- task
 			}
 		}
@@ -109,7 +110,6 @@ func (ww *WorkflowWorker) runDispatcher() {
 
 		t := t
 
-		ww.wg.Add(1)
 		go func() {
 			defer ww.wg.Done()
 
@@ -131,18 +131,25 @@ func (ww *WorkflowWorker) handle(ctx context.Context, t *task.Workflow) {
 	ww.backend.Metrics().Distribution(metrickeys.WorkflowTaskDelay, metrics.Tags{}, float64(timeInQueue/time.Millisecond))
 
 	timer := metrics.Timer(ww.backend.Metrics(), metrickeys.WorkflowTaskProcessed, metrics.Tags{})
-	defer timer.Stop()
 
 	result, err := ww.handleTask(ctx, t)
 	if err != nil {
 		ww.logger.Panic("could not handle workflow task", "error", err)
 	}
 
+	// Only record the time spent in the workflow code
+	timer.Stop()
+
 	state := core.WorkflowInstanceStateActive
 	if result.Completed {
 		state = core.WorkflowInstanceStateFinished
 
-		ww.backend.Metrics().Counter(metrickeys.WorkflowInstanceFinished, metrics.Tags{}, 1)
+		if t.WorkflowInstanceState != state {
+			// If the workflow is now finished, record
+			ww.backend.Metrics().Counter(metrickeys.WorkflowInstanceFinished, metrics.Tags{
+				metrickeys.SubWorkflow: fmt.Sprint(t.WorkflowInstance.SubWorkflow()),
+			}, 1)
+		}
 	}
 
 	ww.backend.Metrics().Counter(metrickeys.ActivityTaskScheduled, metrics.Tags{}, int64(len(result.ActivityEvents)))

internal/workflow/executor.go

@@ -100,15 +100,17 @@ 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.
+		// This could happen if signals are delivered after the workflow is finished
 		logger.Error("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
+		return &ExecutionResult{
+			Completed: true,
+		}, nil
 	}
 
 	skipNewEvents := false