DBOS "client context" (#53)

This PR adds workflow management features usable by a "client context",
that is, a DBOSContext object that is initialized but not launched. The
benefits of doing so is being able to interact with DBOS workflows
without having the process dequeue tasks, recover workflows, or run an
admin server.

Features added:
- Resume workflow
- Fork workflow
- Enqueue workflow by name
- Add ability to dismiss output/input from list workflows

Because the registry is indexed by workflow FQN, we need to keep another
map from workflow "custom" name to workflow FQN. This should have been
tested in PR that added custom workflow names.

Author: maxdml

dbos/client_test.go

@@ -54,7 +54,7 @@ func processConfig(inputConfig *Config) (*Config, error) {
 
 	// Load defaults
 	if dbosConfig.Logger == nil {
-		dbosConfig.Logger = slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))
+		dbosConfig.Logger = slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
 	}
 
 	return dbosConfig, nil
@@ -76,15 +76,21 @@ type DBOSContext interface {
 	// Workflow operations
 	RunAsStep(_ DBOSContext, fn StepFunc) (any, error)                                                            // Execute a function as a durable step within a workflow
 	RunAsWorkflow(_ DBOSContext, fn WorkflowFunc, input any, opts ...WorkflowOption) (WorkflowHandle[any], error) // Start a new workflow execution
-	Send(_ DBOSContext, input WorkflowSendInputInternal) error                                                    // Send a message to another workflow
+	Send(_ DBOSContext, input WorkflowSendInput) error                                                            // Send a message to another workflow
 	Recv(_ DBOSContext, input WorkflowRecvInput) (any, error)                                                     // Receive a message sent to this workflow
 	SetEvent(_ DBOSContext, input WorkflowSetEventInput) error                                                    // Set a key-value event for this workflow
 	GetEvent(_ DBOSContext, input WorkflowGetEventInput) (any, error)                                             // Get a key-value event from a target workflow
 	Sleep(duration time.Duration) (time.Duration, error)                                                          // Durable sleep that survives workflow recovery
 	GetWorkflowID() (string, error)                                                                               // Get the current workflow ID (only available within workflows)
+	GetStepID() (int, error)                                                                                      // Get the current step ID (only available within workflows)
 
 	// Workflow management
-	RetrieveWorkflow(_ DBOSContext, workflowID string) (WorkflowHandle[any], error) // Get a handle to an existing workflow
+	RetrieveWorkflow(_ DBOSContext, workflowID string) (WorkflowHandle[any], error)   // Get a handle to an existing workflow
+	Enqueue(_ DBOSContext, params EnqueueOptions) (WorkflowHandle[any], error)        // Enqueue a new workflow with parameters
+	CancelWorkflow(workflowID string) error                                           // Cancel a workflow by setting its status to CANCELLED
+	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(opts ...ListWorkflowsOption) ([]WorkflowStatus, error)              // List workflows based on filtering criteria
 
 	// Accessors
 	GetApplicationVersion() string // Get the application version for this context
@@ -114,8 +120,9 @@ type dbosContext struct {
 	workflowsWg *sync.WaitGroup
 
 	// Workflow registry
-	workflowRegistry map[string]workflowRegistryEntry
-	workflowRegMutex *sync.RWMutex
+	workflowRegistry        map[string]workflowRegistryEntry
+	workflowRegMutex        *sync.RWMutex
+	workflowCustomNametoFQN sync.Map // Maps fully qualified workflow names to custom names. Usefor when client enqueues a workflow by name because registry is indexed by FQN.
 
 	// Workflow scheduler
 	workflowScheduler *cron.Cron

dbos/dbos.go

@@ -199,7 +199,14 @@ func (qr *queueRunner) run(ctx *dbosContext) {
 			}
 			for _, workflow := range dequeuedWorkflows {
 				// Find the workflow in the registry
-				registeredWorkflow, exists := ctx.workflowRegistry[workflow.name]
+
+				wfName, ok := ctx.workflowCustomNametoFQN.Load(workflow.name)
+				if !ok {
+					ctx.logger.Error("Workflow not found in registry", "workflow_name", workflow.name)
+					continue
+				}
+
+				registeredWorkflow, exists := ctx.workflowRegistry[wfName.(string)]
 				if !exists {
 					ctx.logger.Error("workflow function not found in registry", "workflow_name", workflow.name)
 					continue

dbos/queue.go

@@ -11,6 +11,7 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 		status:             []WorkflowStatusType{WorkflowStatusPending},
 		executorIDs:        executorIDs,
 		applicationVersion: ctx.applicationVersion,
+		loadInput:          true,
 	})
 	if err != nil {
 		return nil, err
@@ -36,7 +37,13 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 			continue
 		}
 
-		registeredWorkflow, exists := ctx.workflowRegistry[workflow.Name]
+		wfName, ok := ctx.workflowCustomNametoFQN.Load(workflow.Name)
+		if !ok {
+			ctx.logger.Error("Workflow not found in registry", "workflow_name", workflow.Name)
+			continue
+		}
+
+		registeredWorkflow, exists := ctx.workflowRegistry[wfName.(string)]
 		if !exists {
 			ctx.logger.Error("Workflow function not found in registry", "workflow_id", workflow.ID, "name", workflow.Name)
 			continue

dbos/recovery.go

@@ -103,9 +103,9 @@ func TestWorkflowEncoding(t *testing.T) {
 		}
 
 		// Test results from ListWorkflows
-		workflows, err := executor.(*dbosContext).systemDB.listWorkflows(context.Background(), listWorkflowsDBInput{
-			workflowIDs: []string{directHandle.GetWorkflowID()},
-		})
+		workflows, err := ListWorkflows(executor, WithWorkflowIDs(
+			[]string{directHandle.GetWorkflowID()},
+		))
 		if err != nil {
 			t.Fatalf("failed to list workflows: %v", err)
 		}
@@ -220,9 +220,9 @@ func TestWorkflowEncoding(t *testing.T) {
 		}
 
 		// Test results from ListWorkflows
-		workflows, err := executor.(*dbosContext).systemDB.listWorkflows(context.Background(), listWorkflowsDBInput{
-			workflowIDs: []string{directHandle.GetWorkflowID()},
-		})
+		workflows, err := ListWorkflows(executor, WithWorkflowIDs(
+			[]string{directHandle.GetWorkflowID()},
+		))
 		if err != nil {
 			t.Fatalf("failed to list workflows: %v", err)
 		}
@@ -317,7 +317,7 @@ func setEventUserDefinedTypeWorkflow(ctx DBOSContext, input string) (string, err
 		},
 	}
 
-	err := SetEvent(ctx, WorkflowSetEventInputGeneric[UserDefinedEventData]{Key: input, Message: eventData})
+	err := SetEvent(ctx, GenericWorkflowSetEventInput[UserDefinedEventData]{Key: input, Message: eventData})
 	if err != nil {
 		return "", err
 	}
@@ -394,7 +394,7 @@ func sendUserDefinedTypeWorkflow(ctx DBOSContext, destinationID string) (string,
 
 	// Send should automatically register this type with gob
 	// Note the explicit type parameter since compiler cannot infer UserDefinedEventData from string input
-	err := Send(ctx, WorkflowSendInput[UserDefinedEventData]{
+	err := Send(ctx, GenericWorkflowSendInput[UserDefinedEventData]{
 		DestinationID: destinationID,
 		Topic:         "user-defined-topic",
 		Message:       sendData,