test conflicting execution of workflows / tasks of the same ID

maxdml · maxdml · commit 83b81487f441 · 2025-08-10T18:26:04.000-07:00
diff --git a/dbos/queues_test.go b/dbos/queues_test.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -23,6 +24,7 @@ This suite tests
 [x] worker concurrency (2 at a time across two "workers")
 [x] worker concurrency X recovery
 [x] rate limiter
+[x] conflicting workflow on different queues
 [] queue deduplication
 [] queue priority
 [x] queued workflow times out
@@ -48,6 +50,8 @@ func TestWorkflowQueues(t *testing.T) {
 
 	queue := NewWorkflowQueue(dbosCtx, "test-queue")
 	dlqEnqueueQueue := NewWorkflowQueue(dbosCtx, "test-successive-enqueue-queue")
+	conflictQueue1 := NewWorkflowQueue(dbosCtx, "conflict-queue-1")
+	conflictQueue2 := NewWorkflowQueue(dbosCtx, "conflict-queue-2")
 
 	dlqStartEvent := NewEvent()
 	dlqCompleteEvent := NewEvent()
@@ -172,14 +176,15 @@ func TestWorkflowQueues(t *testing.T) {
 		}
 	})
 
-	/* TODO: we will move queue registry in the new interface in a subsequent PR
 	t.Run("DynamicRegistration", func(t *testing.T) {
-		q := NewWorkflowQueue("dynamic-queue")
-		if len(q.name) > 0 {
-			t.Fatalf("expected nil queue for dynamic registration after DBOS initialization, got %v", q)
-		}
+		// Attempting to register a queue after launch should panic
+		defer func() {
+			if r := recover(); r == nil {
+				t.Fatal("expected panic from queue registration after launch but got none")
+			}
+		}()
+		NewWorkflowQueue(dbosCtx, "dynamic-queue")
 	})
-	*/
 
 	t.Run("QueueWorkflowDLQ", func(t *testing.T) {
 		workflowID := "blocking-workflow-test"
@@ -255,6 +260,52 @@ func TestWorkflowQueues(t *testing.T) {
 			t.Fatal("expected queue entries to be cleaned up after successive enqueues test")
 		}
 	})
+
+	t.Run("ConflictingWorkflowOnDifferentQueues", func(t *testing.T) {
+		workflowID := "conflicting-workflow-id"
+
+		// Enqueue the same workflow ID on the first queue
+		handle1, err := RunAsWorkflow(dbosCtx, queueWorkflow, "test-input-1", WithQueue(conflictQueue1.Name), WithWorkflowID(workflowID))
+		if err != nil {
+			t.Fatalf("failed to enqueue workflow on first queue: %v", err)
+		}
+
+		// Get the result from the first workflow to ensure it completes
+		result1, err := handle1.GetResult()
+		if err != nil {
+			t.Fatalf("failed to get result from first workflow: %v", err)
+		}
+		if result1 != "test-input-1" {
+			t.Fatalf("expected 'test-input-1', got %v", result1)
+		}
+
+		// Now try to enqueue the same workflow ID on a different queue
+		// This should trigger a ConflictingWorkflowError
+		_, err = RunAsWorkflow(dbosCtx, queueWorkflow, "test-input-2", WithQueue(conflictQueue2.Name), WithWorkflowID(workflowID))
+		if err == nil {
+			t.Fatal("expected ConflictingWorkflowError when enqueueing same workflow ID on different queue, but got none")
+		}
+
+		// Check that it's the correct error type
+		dbosErr, ok := err.(*DBOSError)
+		if !ok {
+			t.Fatalf("expected error to be of type *DBOSError, got %T", err)
+		}
+
+		if dbosErr.Code != ConflictingWorkflowError {
+			t.Fatalf("expected error code to be ConflictingWorkflowError, got %v", dbosErr.Code)
+		}
+
+		// Check that the error message contains queue information
+		expectedMsgPart := "different queue"
+		if !strings.Contains(err.Error(), expectedMsgPart) {
+			t.Fatalf("expected error message to contain '%s', got '%s'", expectedMsgPart, err.Error())
+		}
+
+		if !queueEntriesAreCleanedUp(dbosCtx) {
+			t.Fatal("expected queue entries to be cleaned up after conflicting workflow test")
+		}
+	})
 }
 
 func TestQueueRecovery(t *testing.T) {
diff --git a/dbos/workflows_test.go b/dbos/workflows_test.go
@@ -5,7 +5,7 @@ Test workflow and steps features
 [x] Wrapping various golang methods in DBOS workflows
 [x] workflow idempotency
 [x] workflow DLQ
-[] workflow conflicting name
+[x] workflow conflicting name
 [] workflow timeouts & deadlines (including child workflows)
 */
 
@@ -1902,6 +1902,153 @@ func getEventIdempotencyWorkflow(ctx DBOSContext, input setEventWorkflowInput) (
 	return result, nil
 }
 
+// Test workflows and steps for parameter mismatch validation
+func conflictWorkflowA(dbosCtx DBOSContext, input string) (string, error) {
+	return RunAsStep(dbosCtx, func(ctx context.Context) (string, error) {
+		return conflictStepA(ctx)
+	})
+}
+
+func conflictWorkflowB(dbosCtx DBOSContext, input string) (string, error) {
+	return RunAsStep(dbosCtx, func(ctx context.Context) (string, error) {
+		return conflictStepB(ctx)
+	})
+}
+
+func conflictStepA(_ context.Context) (string, error) {
+	return "step-a-result", nil
+}
+
+func conflictStepB(_ context.Context) (string, error) {
+	return "step-b-result", nil
+}
+
+func workflowWithMultipleSteps(dbosCtx DBOSContext, input string) (string, error) {
+	// First step
+	result1, err := RunAsStep(dbosCtx, func(ctx context.Context) (string, error) {
+		return conflictStepA(ctx)
+	})
+	if err != nil {
+		return "", err
+	}
+
+	// Second step - this is where we'll test step name conflicts
+	result2, err := RunAsStep(dbosCtx, func(ctx context.Context) (string, error) {
+		return conflictStepB(ctx)
+	})
+	if err != nil {
+		return "", err
+	}
+
+	return result1 + "-" + result2, nil
+}
+
+func TestWorkflowExecutionMismatch(t *testing.T) {
+	dbosCtx := setupDBOS(t, true, true)
+
+	// Register workflows for testing
+	RegisterWorkflow(dbosCtx, conflictWorkflowA)
+	RegisterWorkflow(dbosCtx, conflictWorkflowB)
+	RegisterWorkflow(dbosCtx, workflowWithMultipleSteps)
+
+	t.Run("WorkflowNameConflict", func(t *testing.T) {
+		workflowID := uuid.NewString()
+
+		// First, run conflictWorkflowA with a specific workflow ID
+		handle1, err := RunAsWorkflow(dbosCtx, conflictWorkflowA, "test-input", WithWorkflowID(workflowID))
+		if err != nil {
+			t.Fatalf("failed to start first workflow: %v", err)
+		}
+
+		// Get the result to ensure it completes
+		result1, err := handle1.GetResult()
+		if err != nil {
+			t.Fatalf("failed to get result from first workflow: %v", err)
+		}
+		if result1 != "step-a-result" {
+			t.Fatalf("expected 'step-a-result', got '%s'", result1)
+		}
+
+		// Now try to run conflictWorkflowB with the same workflow ID
+		// This should return a ConflictingWorkflowError
+		_, err = RunAsWorkflow(dbosCtx, conflictWorkflowB, "test-input", WithWorkflowID(workflowID))
+		if err == nil {
+			t.Fatal("expected ConflictingWorkflowError when running different workflow with same ID, but got none")
+		}
+
+		// Check that it's the correct error type
+		dbosErr, ok := err.(*DBOSError)
+		if !ok {
+			t.Fatalf("expected error to be of type *DBOSError, got %T", err)
+		}
+
+		if dbosErr.Code != ConflictingWorkflowError {
+			t.Fatalf("expected error code to be ConflictingWorkflowError, got %v", dbosErr.Code)
+		}
+
+		// Check that the error message contains the workflow names
+		expectedMsgPart := "Workflow already exists with a different name"
+		if !strings.Contains(err.Error(), expectedMsgPart) {
+			t.Fatalf("expected error message to contain '%s', got '%s'", expectedMsgPart, err.Error())
+		}
+	})
+
+	t.Run("StepNameConflict", func(t *testing.T) {
+		// This test simulates a scenario where a workflow is recovered but
+		// the step implementation has changed, causing a step name mismatch
+
+		// First, start a workflow and let it complete partially
+		handle1, err := RunAsWorkflow(dbosCtx, workflowWithMultipleSteps, "test-input")
+		if err != nil {
+			t.Fatalf("failed to start workflow: %v", err)
+		}
+
+		// Complete the workflow
+		result, err := handle1.GetResult()
+		if err != nil {
+			t.Fatalf("failed to get result from workflow: %v", err)
+		}
+		if result != "step-a-result-step-b-result" {
+			t.Fatalf("expected 'step-a-result-step-b-result', got '%s'", result)
+		}
+
+		// Now simulate what happens if we try to check operation execution
+		// with a different step name for the same step ID
+		workflowID := handle1.GetWorkflowID()
+
+		// This directly tests the CheckOperationExecution method with mismatched step name
+		// We'll check step ID 0 (first step) but with wrong step name
+		wrongStepName := "wrong-step-name"
+		_, err = dbosCtx.(*dbosContext).systemDB.CheckOperationExecution(dbosCtx, checkOperationExecutionDBInput{
+			workflowID: workflowID,
+			stepID:     0,
+			stepName:   wrongStepName,
+		})
+
+		if err == nil {
+			t.Fatal("expected UnexpectedStep error when checking operation with wrong step name, but got none")
+		}
+
+		// Check that it's the correct error type
+		dbosErr, ok := err.(*DBOSError)
+		if !ok {
+			t.Fatalf("expected error to be of type *DBOSError, got %T", err)
+		}
+
+		if dbosErr.Code != UnexpectedStep {
+			t.Fatalf("expected error code to be UnexpectedStep, got %v", dbosErr.Code)
+		}
+
+		// Check that the error message contains step information
+		if !strings.Contains(err.Error(), "Check that your workflow is deterministic") {
+			t.Fatalf("expected error message to contain 'Check that your workflow is deterministic', got '%s'", err.Error())
+		}
+		if !strings.Contains(err.Error(), wrongStepName) {
+			t.Fatalf("expected error message to contain wrong step name '%s', got '%s'", wrongStepName, err.Error())
+		}
+	})
+}
+
 func TestSetGetEvent(t *testing.T) {
 	dbosCtx := setupDBOS(t, true, true)
 
