Handle conflicting executions (dbos-inc#139)

If a workflow observes the returned error is a `DBOSError` with code
`ConflictingIDError`, fall back to `awaitWorkflowResult` in the workflow
goroutine.

Note that we cannot distinguish this error if the user handles it (e.g.,
during `setEvent`) and wraps it with some other error message.

Author: maxdml

dbos/system_database.go

@@ -1189,17 +1189,16 @@ func (s *sysDB) recordOperationResult(ctx context.Context, input recordOperation
 		return fmt.Errorf("failed to serialize output: %w", err)
 	}
 
-	var commandTag pgconn.CommandTag
 	if input.tx != nil {
-		commandTag, err = input.tx.Exec(ctx, query,
+		_, err = input.tx.Exec(ctx, query,
 			input.workflowID,
 			input.stepID,
 			outputString,
 			errorString,
 			input.stepName,
 		)
 	} else {
-		commandTag, err = s.pool.Exec(ctx, query,
+		_, err = s.pool.Exec(ctx, query,
 			input.workflowID,
 			input.stepID,
 			outputString,
@@ -1208,12 +1207,6 @@ func (s *sysDB) recordOperationResult(ctx context.Context, input recordOperation
 		)
 	}
 
-	/*
-		s.logger.Debug("RecordOperationResult CommandTag", "command_tag", commandTag)
-		s.logger.Debug("RecordOperationResult Rows affected", "rows_affected", commandTag.RowsAffected())
-		s.logger.Debug("RecordOperationResult SQL", "sql", commandTag.String())
-	*/
-
 	if err != nil {
 		s.logger.Error("RecordOperationResult Error occurred", "error", err)
 		if pgErr, ok := err.(*pgconn.PgError); ok && pgErr.Code == _PG_ERROR_UNIQUE_VIOLATION {
@@ -1222,10 +1215,6 @@ func (s *sysDB) recordOperationResult(ctx context.Context, input recordOperation
 		return err
 	}
 
-	if commandTag.RowsAffected() == 0 {
-		s.logger.Warn("RecordOperationResult No rows were affected by the insert")
-	}
-
 	return nil
 }
 
@@ -1778,7 +1767,7 @@ func (s *sysDB) send(ctx context.Context, input WorkflowSendInput) error {
 
 		err = s.recordOperationResult(ctx, recordInput)
 		if err != nil {
-			return fmt.Errorf("failed to record operation result: %w", err)
+			return err
 		}
 	}
 
@@ -1932,7 +1921,7 @@ func (s *sysDB) recv(ctx context.Context, input recvInput) (any, error) {
 	}
 	err = s.recordOperationResult(ctx, recordInput)
 	if err != nil {
-		return nil, fmt.Errorf("failed to record operation result: %w", err)
+		return nil, err
 	}
 
 	if err := tx.Commit(ctx); err != nil {
@@ -2013,7 +2002,7 @@ func (s *sysDB) setEvent(ctx context.Context, input WorkflowSetEventInput) error
 
 	err = s.recordOperationResult(ctx, recordInput)
 	if err != nil {
-		return fmt.Errorf("failed to record operation result: %w", err)
+		return err
 	}
 
 	// Commit transaction
@@ -2145,7 +2134,7 @@ func (s *sysDB) getEvent(ctx context.Context, input getEventInput) (any, error)
 
 		err = s.recordOperationResult(ctx, recordInput)
 		if err != nil {
-			return nil, fmt.Errorf("failed to record operation result: %w", err)
+			return nil, err
 		}
 	}
 

dbos/workflow.go

@@ -826,35 +826,44 @@ func (c *dbosContext) RunWorkflow(_ DBOSContext, fn WorkflowFunc, input any, opt
 	c.workflowsWg.Add(1)
 	go func() {
 		defer c.workflowsWg.Done()
-		result, err := fn(workflowCtx, input)
-		status := WorkflowStatusSuccess
 
-		// If an error occurred, set the status to error
-		if err != nil {
-			status = WorkflowStatusError
-		}
+		var result any
+		var err error
 
-		// If the afterFunc has started, the workflow was cancelled and the status should be set to cancelled
-		if stopFunc != nil && !stopFunc() {
-			c.logger.Info("Workflow was cancelled. Waiting for cancel function to complete", "workflow_id", workflowID)
-			// Wait for the cancel function to complete
-			// Note this must happen before we write on the outcome channel (and signal the handler's GetResult)
-			<-cancelFuncCompleted
-			// Set the status to cancelled and move on so we still record the outcome in the DB
-			status = WorkflowStatusCancelled
-		}
+		result, err = fn(workflowCtx, input)
 
-		recordErr := c.systemDB.updateWorkflowOutcome(uncancellableCtx, updateWorkflowOutcomeDBInput{
-			workflowID: workflowID,
-			status:     status,
-			err:        err,
-			output:     result,
-		})
-		if recordErr != nil {
-			c.logger.Error("Error recording workflow outcome", "workflow_id", workflowID, "error", recordErr)
-			outcomeChan <- workflowOutcome[any]{result: nil, err: recordErr}
-			close(outcomeChan)
-			return
+		// Handle DBOS ID conflict errors by waiting workflow result
+		var dbosErr *DBOSError
+		if errors.As(err, &dbosErr) && dbosErr.Code == ConflictingIDError {
+			c.logger.Warn("Workflow ID conflict detected. Waiting for existing workflow to complete", "workflow_id", workflowID)
+			result, err = c.systemDB.awaitWorkflowResult(uncancellableCtx, workflowID)
+		} else {
+			status := WorkflowStatusSuccess
+
+			// If an error occurred, set the status to error
+			if err != nil {
+				status = WorkflowStatusError
+			}
+
+			// If the afterFunc has started, the workflow was cancelled and the status should be set to cancelled
+			if stopFunc != nil && !stopFunc() {
+				c.logger.Info("Workflow was cancelled. Waiting for cancel function to complete", "workflow_id", workflowID)
+				<-cancelFuncCompleted // Wait for the cancel function to complete
+				status = WorkflowStatusCancelled
+			}
+
+			recordErr := c.systemDB.updateWorkflowOutcome(uncancellableCtx, updateWorkflowOutcomeDBInput{
+				workflowID: workflowID,
+				status:     status,
+				err:        err,
+				output:     result,
+			})
+			if recordErr != nil {
+				c.logger.Error("Error recording workflow outcome", "workflow_id", workflowID, "error", recordErr)
+				outcomeChan <- workflowOutcome[any]{result: nil, err: recordErr}
+				close(outcomeChan)
+				return
+			}
 		}
 		outcomeChan <- workflowOutcome[any]{result: result, err: err}
 		close(outcomeChan)

dbos/workflows_test.go

@@ -1736,14 +1736,11 @@ func TestSendRecv(t *testing.T) {
 	})
 
 	t.Run("TestConcurrentRecvs", func(t *testing.T) {
-		// Test concurrent receivers - only 1 should timeout, others should get errors
+		// Test concurrent receivers - all should return valid results
 		receiveTopic := "concurrent-recv-topic"
 
-		// Start multiple concurrent receive workflows - no messages will be sent
+		// Start multiple concurrent receive workflows
 		numReceivers := 5
-		var wg sync.WaitGroup
-		results := make(chan string, numReceivers)
-		errors := make(chan error, numReceivers)
 		receiverHandles := make([]WorkflowHandle[string], numReceivers)
 
 		// Start all receivers - they will signal when ready and wait for coordination
@@ -1768,56 +1765,12 @@ func TestSendRecv(t *testing.T) {
 		// Now unblock all receivers simultaneously so they race to the Recv call
 		concurrentRecvStartEvent.Set()
 
-		// Collect results from all receivers concurrently
-		// Only 1 should timeout (winner of the CV), others should get errors
-		wg.Add(numReceivers)
+		// Collect results from all receivers
 		for i := range numReceivers {
-			go func(index int) {
-				defer wg.Done()
-				result, err := receiverHandles[index].GetResult()
-				if err != nil {
-					errors <- err
-				} else {
-					results <- result
-				}
-			}(i)
-		}
-
-		wg.Wait()
-		close(results)
-		close(errors)
-
-		// Count timeout results and errors
-		timeoutCount := 0
-		errorCount := 0
-
-		for result := range results {
-			if result == "" {
-				// Empty string indicates a timeout - only 1 receiver should get this
-				timeoutCount++
-			}
+			result, err := receiverHandles[i].GetResult()
+			require.NoError(t, err, "receiver %d should not error", i)
+			require.Equal(t, result, "", "receiver %d should have an empty string result", i)
 		}
-
-		for err := range errors {
-			t.Logf("Receiver error (expected): %v", err)
-
-			// Check that the error is of the expected type
-			dbosErr, ok := err.(*DBOSError)
-			require.True(t, ok, "expected error to be of type *DBOSError, got %T", err)
-			require.Equal(t, ConflictingIDError, dbosErr.Code, "expected error code to be ConflictingIDError, got %v", dbosErr.Code)
-			require.Equal(t, "concurrent-recv-wfid", dbosErr.WorkflowID, "expected workflow ID to be 'concurrent-recv-wfid', got %s", dbosErr.WorkflowID)
-			require.True(t, dbosErr.IsBase, "expected error to have IsBase=true")
-			require.Contains(t, dbosErr.Message, "Conflicting workflow ID concurrent-recv-wfid", "expected error message to contain conflicting workflow ID")
-
-			errorCount++
-		}
-
-		// Verify that exactly 1 receiver timed out and 4 got errors
-		assert.Equal(t, 1, timeoutCount, "expected exactly 1 receiver to timeout")
-		assert.Equal(t, 4, errorCount, "expected exactly 4 receivers to get errors")
-
-		// Ensure total results match expected
-		assert.Equal(t, numReceivers, timeoutCount+errorCount, "expected total results to equal number of receivers")
 	})
 
 	t.Run("durableSleep", func(t *testing.T) {