Merge branch 'main' into 93

Author: AayushSaini101

dbos/admin_server.go

@@ -412,7 +412,7 @@ func newAdminServer(ctx *dbosContext, port int) *adminServer {
 	mux.HandleFunc(_WORKFLOW_STEPS_PATTERN, func(w http.ResponseWriter, r *http.Request) {
 		workflowID := r.PathValue("id")
 
-		steps, err := ctx.systemDB.getWorkflowSteps(ctx, workflowID)
+		steps, err := GetWorkflowSteps(ctx, workflowID)
 		if err != nil {
 			ctx.logger.Error("Failed to list workflow steps", "workflow_id", workflowID, "error", err)
 			http.Error(w, fmt.Sprintf("Failed to list steps: %v", err), http.StatusInternalServerError)

dbos/conductor.go

@@ -719,8 +719,8 @@ func (c *Conductor) handleListStepsRequest(data []byte, requestID string) error
 	}
 	c.logger.Debug("Handling list steps request", "request", req)
 
-	// Get workflow steps using the existing systemDB method
-	steps, err := c.dbosCtx.systemDB.getWorkflowSteps(c.dbosCtx, req.WorkflowID)
+	// Get workflow steps using the public GetWorkflowSteps method
+	steps, err := GetWorkflowSteps(c.dbosCtx, req.WorkflowID)
 	if err != nil {
 		c.logger.Error("Failed to list workflow steps", "workflow_id", req.WorkflowID, "error", err)
 		errorMsg := fmt.Sprintf("failed to list workflow steps: %v", err)

dbos/conductor_protocol.go

@@ -190,8 +190,8 @@ type listStepsConductorResponse struct {
 	Output *[]workflowStepsConductorResponseBody `json:"output,omitempty"`
 }
 
-// formatWorkflowStepsResponseBody converts stepInfo to workflowStepsConductorResponseBody for the conductor protocol
-func formatWorkflowStepsResponseBody(step stepInfo) workflowStepsConductorResponseBody {
+// formatWorkflowStepsResponseBody converts StepInfo to workflowStepsConductorResponseBody for the conductor protocol
+func formatWorkflowStepsResponseBody(step StepInfo) workflowStepsConductorResponseBody {
 	output := workflowStepsConductorResponseBody{
 		FunctionID:   step.StepID,
 		FunctionName: step.StepName,

dbos/dbos.go

@@ -123,6 +123,7 @@ type DBOSContext interface {
 	ResumeWorkflow(_ DBOSContext, workflowID string) (WorkflowHandle[any], error)                                         // Resume a cancelled workflow
 	ForkWorkflow(_ DBOSContext, input ForkWorkflowInput) (WorkflowHandle[any], error)                                     // Fork a workflow from a specific step
 	ListWorkflows(_ DBOSContext, opts ...ListWorkflowsOption) ([]WorkflowStatus, error)                                   // List workflows based on filtering criteria
+	GetWorkflowSteps(_ DBOSContext, workflowID string) ([]StepInfo, error)                                                // Get the execution steps of a workflow
 
 	// Accessors
 	GetApplicationVersion() string // Get the application version for this context
@@ -349,7 +350,6 @@ func NewDBOSContext(inputConfig Config) (DBOSContext, error) {
 
 	// Initialize the queue runner and register DBOS internal queue
 	initExecutor.queueRunner = newQueueRunner(initExecutor.logger)
-	NewWorkflowQueue(initExecutor, _DBOS_INTERNAL_QUEUE_NAME)
 
 	// Initialize conductor if API key is provided
 	if config.ConductorAPIKey != "" {
@@ -404,6 +404,7 @@ func (c *dbosContext) Launch() error {
 	}
 
 	// Start the queue runner in a goroutine
+	NewWorkflowQueue(c, _DBOS_INTERNAL_QUEUE_NAME)
 	go func() {
 		c.queueRunner.run(c)
 	}()

dbos/migrations/000001_initial_dbos_schema.up.sql

@@ -7,7 +7,7 @@ CREATE SCHEMA IF NOT EXISTS dbos;
 CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
 
 -- Create workflow_status table
-CREATE TABLE dbos.workflow_status (
+CREATE TABLE IF NOT EXISTS dbos.workflow_status (
     workflow_uuid TEXT PRIMARY KEY,
     status TEXT,
     name TEXT,
@@ -34,17 +34,27 @@ CREATE TABLE dbos.workflow_status (
 );
 
 -- Create indexes for workflow_status
-CREATE INDEX workflow_status_created_at_index ON dbos.workflow_status (created_at);
-CREATE INDEX workflow_status_executor_id_index ON dbos.workflow_status (executor_id);
-CREATE INDEX workflow_status_status_index ON dbos.workflow_status (status);
+CREATE INDEX IF NOT EXISTS workflow_status_created_at_index ON dbos.workflow_status (created_at);
+CREATE INDEX IF NOT EXISTS workflow_status_executor_id_index ON dbos.workflow_status (executor_id);
+CREATE INDEX IF NOT EXISTS workflow_status_status_index ON dbos.workflow_status (status);
 
 -- Create unique constraint for queue_name and deduplication_id
-ALTER TABLE dbos.workflow_status 
-ADD CONSTRAINT uq_workflow_status_queue_name_dedup_id 
-UNIQUE (queue_name, deduplication_id);
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM information_schema.table_constraints
+        WHERE constraint_name = 'uq_workflow_status_queue_name_dedup_id'
+        AND table_name = 'workflow_status'
+        AND table_schema = 'dbos'
+    ) THEN
+        ALTER TABLE dbos.workflow_status
+        ADD CONSTRAINT uq_workflow_status_queue_name_dedup_id
+        UNIQUE (queue_name, deduplication_id);
+    END IF;
+END $$;
 
 -- Create operation_outputs table
-CREATE TABLE dbos.operation_outputs (
+CREATE TABLE IF NOT EXISTS dbos.operation_outputs (
     workflow_uuid TEXT NOT NULL,
     function_id INTEGER NOT NULL,
     function_name TEXT NOT NULL DEFAULT '',
@@ -56,7 +66,7 @@ CREATE TABLE dbos.operation_outputs (
         ON UPDATE CASCADE ON DELETE CASCADE
 );
 
-CREATE TABLE dbos.notifications (
+CREATE TABLE IF NOT EXISTS dbos.notifications (
     destination_uuid TEXT NOT NULL,
     topic TEXT,
     message TEXT NOT NULL,
@@ -66,7 +76,7 @@ CREATE TABLE dbos.notifications (
         ON UPDATE CASCADE ON DELETE CASCADE
 );
 -- Create index for notifications
-CREATE INDEX idx_workflow_topic ON dbos.notifications (destination_uuid, topic);
+CREATE INDEX IF NOT EXISTS idx_workflow_topic ON dbos.notifications (destination_uuid, topic);
 
 -- Create notification function
 CREATE OR REPLACE FUNCTION dbos.notifications_function() RETURNS TRIGGER AS $$
@@ -79,12 +89,22 @@ END;
 $$ LANGUAGE plpgsql;
 
 -- Create notification trigger
-CREATE TRIGGER dbos_notifications_trigger
-AFTER INSERT ON dbos.notifications
-FOR EACH ROW EXECUTE FUNCTION dbos.notifications_function();
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM information_schema.triggers
+        WHERE trigger_name = 'dbos_notifications_trigger'
+        AND event_object_table = 'notifications'
+        AND event_object_schema = 'dbos'
+    ) THEN
+        CREATE TRIGGER dbos_notifications_trigger
+        AFTER INSERT ON dbos.notifications
+        FOR EACH ROW EXECUTE FUNCTION dbos.notifications_function();
+    END IF;
+END $$;
 
 -- Create workflow_events table
-CREATE TABLE dbos.workflow_events (
+CREATE TABLE IF NOT EXISTS dbos.workflow_events (
     workflow_uuid TEXT NOT NULL,
     key TEXT NOT NULL,
     value TEXT NOT NULL,
@@ -104,6 +124,16 @@ END;
 $$ LANGUAGE plpgsql;
 
 -- Create events trigger
-CREATE TRIGGER dbos_workflow_events_trigger
-AFTER INSERT ON dbos.workflow_events
-FOR EACH ROW EXECUTE FUNCTION dbos.workflow_events_function();
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM information_schema.triggers
+        WHERE trigger_name = 'dbos_workflow_events_trigger'
+        AND event_object_table = 'workflow_events'
+        AND event_object_schema = 'dbos'
+    ) THEN
+        CREATE TRIGGER dbos_workflow_events_trigger
+        AFTER INSERT ON dbos.workflow_events
+        FOR EACH ROW EXECUTE FUNCTION dbos.workflow_events_function();
+    END IF;
+END $$;

dbos/queues_test.go

@@ -173,7 +173,7 @@ func TestWorkflowQueues(t *testing.T) {
 		assert.Equal(t, "test-input", res)
 
 		// List steps: the workflow should have 1 step
-		steps, err := dbosCtx.(*dbosContext).systemDB.getWorkflowSteps(dbosCtx, handle.GetWorkflowID())
+		steps, err := GetWorkflowSteps(dbosCtx, handle.GetWorkflowID())
 		require.NoError(t, err)
 		assert.Len(t, steps, 1)
 		assert.Equal(t, 0, steps[0].StepID)
@@ -207,7 +207,7 @@ func TestWorkflowQueues(t *testing.T) {
 		assert.Equal(t, expectedResult, res)
 
 		// List steps: the workflow should have 2 steps (Start the child and GetResult)
-		steps, err := dbosCtx.(*dbosContext).systemDB.getWorkflowSteps(dbosCtx, handle.GetWorkflowID())
+		steps, err := GetWorkflowSteps(dbosCtx, handle.GetWorkflowID())
 		require.NoError(t, err)
 		assert.Len(t, steps, 2)
 		assert.Equal(t, runtime.FuncForPC(reflect.ValueOf(queueWorkflow).Pointer()).Name(), steps[0].StepName)
@@ -230,7 +230,7 @@ func TestWorkflowQueues(t *testing.T) {
 		assert.Equal(t, expectedResult, res)
 
 		// List steps: the workflow should have 2 steps (Start the child and GetResult)
-		steps, err := dbosCtx.(*dbosContext).systemDB.getWorkflowSteps(dbosCtx, handle.GetWorkflowID())
+		steps, err := GetWorkflowSteps(dbosCtx, handle.GetWorkflowID())
 		require.NoError(t, err)
 		assert.Len(t, steps, 2)
 		assert.Equal(t, runtime.FuncForPC(reflect.ValueOf(queueWorkflow).Pointer()).Name(), steps[0].StepName)
@@ -253,7 +253,7 @@ func TestWorkflowQueues(t *testing.T) {
 		assert.Equal(t, expectedResult, res)
 
 		// List steps: the workflow should have 2 steps (Start the child and GetResult)
-		steps, err := dbosCtx.(*dbosContext).systemDB.getWorkflowSteps(dbosCtx, handle.GetWorkflowID())
+		steps, err := GetWorkflowSteps(dbosCtx, handle.GetWorkflowID())
 		require.NoError(t, err)
 		assert.Len(t, steps, 2)
 		assert.Equal(t, "custom-name", steps[0].StepName)
@@ -279,7 +279,7 @@ func TestWorkflowQueues(t *testing.T) {
 
 		// Check that the parent workflow (the one we ran directly) has 2 steps:
 		// one for enqueueing the child and one for calling GetResult
-		steps, err := dbosCtx.(*dbosContext).systemDB.getWorkflowSteps(dbosCtx, handle.GetWorkflowID())
+		steps, err := GetWorkflowSteps(dbosCtx, handle.GetWorkflowID())
 		require.NoError(t, err)
 		assert.Len(t, steps, 2)
 		assert.Equal(t, runtime.FuncForPC(reflect.ValueOf(queueWorkflow).Pointer()).Name(), steps[0].StepName)

dbos/serialization_test.go

@@ -114,7 +114,7 @@ func TestWorkflowEncoding(t *testing.T) {
 		assert.Equal(t, "workflow error: step error", workflow.Error.Error())
 
 		// Test results from GetWorkflowSteps
-		steps, err := executor.(*dbosContext).systemDB.getWorkflowSteps(context.Background(), directHandle.GetWorkflowID())
+		steps, err := GetWorkflowSteps(executor, directHandle.GetWorkflowID())
 		require.NoError(t, err)
 		require.Len(t, steps, 1)
 		step := steps[0]
@@ -175,7 +175,7 @@ func TestWorkflowEncoding(t *testing.T) {
 		assert.Equal(t, "processed by encodingStepStruct", workflowOutput.B)
 
 		// Test results from GetWorkflowSteps
-		steps, err := executor.(*dbosContext).systemDB.getWorkflowSteps(context.Background(), directHandle.GetWorkflowID())
+		steps, err := GetWorkflowSteps(executor, directHandle.GetWorkflowID())
 		require.NoError(t, err)
 		require.Len(t, steps, 1)
 		step := steps[0]

dbos/system_database.go

@@ -47,7 +47,7 @@ type systemDatabase interface {
 	// Steps
 	recordOperationResult(ctx context.Context, input recordOperationResultDBInput) error
 	checkOperationExecution(ctx context.Context, input checkOperationExecutionDBInput) (*recordedResult, error)
-	getWorkflowSteps(ctx context.Context, workflowID string) ([]stepInfo, error)
+	getWorkflowSteps(ctx context.Context, workflowID string) ([]StepInfo, error)
 
 	// Communication (special steps)
 	send(ctx context.Context, input WorkflowSendInput) error
@@ -140,28 +140,35 @@ const (
 
 func runMigrations(databaseURL string) error {
 	// Change the driver to pgx5
-	databaseURL = "pgx5://" + strings.TrimPrefix(databaseURL, "postgres://")
-
-	// Create migration source from embedded files
-	d, err := iofs.New(migrationFiles, "migrations")
-	if err != nil {
-		return newInitializationError(fmt.Sprintf("failed to create migration source: %v", err))
-	}
-
-	// Add custom migration table name to avoid conflicts with user migrations
-	// Parse the URL to properly determine where to add the query parameter
 	parsedURL, err := url.Parse(databaseURL)
 	if err != nil {
 		return newInitializationError(fmt.Sprintf("failed to parse database URL: %v", err))
 	}
+	// Handle various PostgreSQL URL schemes
+	switch parsedURL.Scheme {
+	case "postgres", "postgresql":
+		parsedURL.Scheme = "pgx5"
+	case "pgx5":
+		// Already in correct format
+	default:
+		return newInitializationError(fmt.Sprintf("unsupported database URL scheme: %s", parsedURL.Scheme))
+	}
+	databaseURL = parsedURL.String()
 
+	// Add custom migration table name to avoid conflicts with user migrations
 	// Check if query parameters already exist
 	separator := "?"
 	if parsedURL.RawQuery != "" {
 		separator = "&"
 	}
 	databaseURL += separator + "x-migrations-table=" + _DBOS_MIGRATION_TABLE
 
+	// Create migration source from embedded files
+	d, err := iofs.New(migrationFiles, "migrations")
+	if err != nil {
+		return newInitializationError(fmt.Sprintf("failed to create migration source: %v", err))
+	}
+
 	// Create migrator
 	m, err := migrate.NewWithSourceInstance("iofs", d, databaseURL)
 	if err != nil {
@@ -1326,15 +1333,16 @@ func (s *sysDB) checkOperationExecution(ctx context.Context, input checkOperatio
 	return result, nil
 }
 
-type stepInfo struct {
-	StepID          int
-	StepName        string
-	Output          any
-	Error           error
-	ChildWorkflowID string
+// StepInfo contains information about a workflow step execution.
+type StepInfo struct {
+	StepID          int    // The sequential ID of the step within the workflow
+	StepName        string // The name of the step function
+	Output          any    // The output returned by the step (if any)
+	Error           error  // The error returned by the step (if any)
+	ChildWorkflowID string // The ID of a child workflow spawned by this step (if applicable)
 }
 
-func (s *sysDB) getWorkflowSteps(ctx context.Context, workflowID string) ([]stepInfo, error) {
+func (s *sysDB) getWorkflowSteps(ctx context.Context, workflowID string) ([]StepInfo, error) {
 	query := `SELECT function_id, function_name, output, error, child_workflow_id
 			  FROM dbos.operation_outputs
 			  WHERE workflow_uuid = $1
@@ -1346,9 +1354,9 @@ func (s *sysDB) getWorkflowSteps(ctx context.Context, workflowID string) ([]step
 	}
 	defer rows.Close()
 
-	var steps []stepInfo
+	var steps []StepInfo
 	for rows.Next() {
-		var step stepInfo
+		var step StepInfo
 		var outputString *string
 		var errorString *string
 		var childWorkflowID *string
@@ -2078,6 +2086,10 @@ func (s *sysDB) dequeueWorkflows(ctx context.Context, input dequeueWorkflowsInpu
 		}
 	}
 
+	if maxTasks <= 0 {
+		return nil, nil
+	}
+
 	// Build the query to select workflows for dequeueing
 	// Use SKIP LOCKED when no global concurrency is set to avoid blocking,
 	// otherwise use NOWAIT to ensure consistent view across processes

dbos/workflow.go

@@ -2015,7 +2015,7 @@ func (c *dbosContext) ListWorkflows(_ DBOSContext, opts ...ListWorkflowsOption)
 	// Call the context method to list workflows
 	workflows, err := c.systemDB.listWorkflows(c, dbInput)
 	if err != nil {
-		return nil, fmt.Errorf("failed to list workflows: %w", err)
+		return nil, err
 	}
 
 	return workflows, nil
@@ -2062,3 +2062,32 @@ func ListWorkflows(ctx DBOSContext, opts ...ListWorkflowsOption) ([]WorkflowStat
 	}
 	return ctx.ListWorkflows(ctx, opts...)
 }
+
+func (c *dbosContext) GetWorkflowSteps(_ DBOSContext, workflowID string) ([]StepInfo, error) {
+	return c.systemDB.getWorkflowSteps(c, workflowID)
+}
+
+// GetWorkflowSteps retrieves the execution steps of a workflow.
+// Returns a list of step information including step IDs, names, outputs, errors, and child workflow IDs.
+//
+// Parameters:
+//   - ctx: DBOS context for the operation
+//   - workflowID: The unique identifier of the workflow
+//
+// Returns a slice of StepInfo structs containing information about each executed step.
+//
+// Example:
+//
+//	steps, err := dbos.GetWorkflowSteps(ctx, "workflow-id")
+//	if err != nil {
+//	    log.Fatal(err)
+//	}
+//	for _, step := range steps {
+//	    log.Printf("Step %d: %s", step.StepID, step.StepName)
+//	}
+func GetWorkflowSteps(ctx DBOSContext, workflowID string) ([]StepInfo, error) {
+	if ctx == nil {
+		return nil, errors.New("ctx cannot be nil")
+	}
+	return ctx.GetWorkflowSteps(ctx, workflowID)
+}