Fix continue as new race condition for sql backends

Author: cschleiden

backend/mysql/mysql.go

@@ -332,16 +332,17 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, err
 			FROM instances i
 			INNER JOIN pending_events pe ON i.instance_id = pe.instance_id
 			WHERE
-				i.completed_at IS NULL
+				state = ? AND i.completed_at IS NULL
 				AND (pe.visible_at IS NULL OR pe.visible_at <= ?)
 				AND (i.locked_until IS NULL OR i.locked_until < ?)
 				AND (i.sticky_until IS NULL OR i.sticky_until < ? OR i.worker = ?)
 			LIMIT 1
 			FOR UPDATE OF i SKIP LOCKED`,
-		now,          // event.visible_at
-		now,          // locked_until
-		now,          // sticky_until
-		b.workerName, // worker
+		core.WorkflowInstanceStateActive, // state
+		now,                              // event.visible_at
+		now,                              // locked_until
+		now,                              // sticky_until
+		b.workerName,                     // worker
 	)
 
 	var id int
@@ -484,7 +485,7 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 
 	// Unlock instance, but keep it sticky to the current worker
 	var completedAt *time.Time
-	if state == core.WorkflowInstanceStateFinished {
+	if state == core.WorkflowInstanceStateContinuedAsNew || state == core.WorkflowInstanceStateFinished {
 		t := time.Now()
 		completedAt = &t
 	}

backend/sqlite/sqlite.go

@@ -288,7 +288,7 @@ func (sb *sqliteBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, e
 					WHERE
 						(locked_until IS NULL OR locked_until < ?)
 						AND (sticky_until IS NULL OR sticky_until < ? OR worker = ?)
-						AND completed_at IS NULL
+						AND state = ? AND i.completed_at IS NULL
 						AND EXISTS (
 							SELECT 1
 								FROM pending_events
@@ -298,10 +298,11 @@ func (sb *sqliteBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, e
 			) RETURNING id, execution_id, parent_instance_id, parent_execution_id, parent_schedule_event_id, metadata, sticky_until`,
 		now.Add(sb.options.WorkflowLockTimeout), // new locked_until
 		sb.workerName,
-		now,           // locked_until
-		now,           // sticky_until
-		sb.workerName, // worker
-		now,           // event.visible_at
+		now,                              // locked_until
+		now,                              // sticky_until
+		sb.workerName,                    // worker
+		core.WorkflowInstanceStateActive, // state
+		now,                              // pending_event.visible_at
 	)
 
 	var instanceID, executionID string
@@ -383,7 +384,7 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 	defer tx.Rollback()
 
 	var completedAt *time.Time
-	if state == core.WorkflowInstanceStateFinished || state == core.WorkflowInstanceStateContinuedAsNew {
+	if state == core.WorkflowInstanceStateContinuedAsNew || state == core.WorkflowInstanceStateFinished {
 		t := time.Now()
 		completedAt = &t
 	}