feat: add Go function to DBOSContext for executing steps in Go routines with result channel

Author: af-md

dbos/dbos.go

@@ -111,6 +111,7 @@ type DBOSContext interface {
 	// Workflow operations
 	RunAsStep(_ DBOSContext, fn StepFunc, opts ...StepOption) (any, error)                                      // Execute a function as a durable step within a workflow
 	RunWorkflow(_ DBOSContext, fn WorkflowFunc, input any, opts ...WorkflowOption) (WorkflowHandle[any], error) // Start a new workflow execution
+	Go(_ DBOSContext, fn StepFunc, opts ...StepOption) (chan StepOutcome[any], error)                           // Starts a step inside a Go routine and returns a channel to receive the result
 	Send(_ DBOSContext, destinationID string, message any, topic string) error                                  // Send a message to another workflow
 	Recv(_ DBOSContext, topic string, timeout time.Duration) (any, error)                                       // Receive a message sent to this workflow
 	SetEvent(_ DBOSContext, key string, message any) error                                                      // Set a key-value event for this workflow

dbos/workflow.go

@@ -1015,7 +1015,8 @@ func WithNextStepID(stepID int) StepOption {
 }
 
 // StepOutcome holds the result and error from a step execution
-type stepOutcome[R any] struct {
+// This struct is returned as part of a channel from the Go function when running the step inside a Go routine
+type StepOutcome[R any] struct {
 	result R
 	err    error
 }
@@ -1222,19 +1223,56 @@ func (c *dbosContext) RunAsStep(_ DBOSContext, fn StepFunc, opts ...StepOption)
 // Go runs a step inside a Go routine and returns a channel to receive the result.
 // Go generates a deterministic step ID for the step before running the step in a routine, since routines are not deterministic.
 // The step ID is used to track the steps within the same workflow and use the step ID to perform recovery.
-// The folliwing examples shows how to use Go: 
-//
-//	resultChan, err := dbos.Go(ctx, func(ctx context.Context) (string, error) {
-//		return "Hello, World!", nil
-//	})
-//
-//	resultChan := <-resultChan // wait for the channel to receive
-//	if resultChan.err != nil {
-//		// Handle error
-//	}
-//  result := resultChan.result
-//
-func Go[R any](ctx DBOSContext, fn Step[R], opts ...StepOption) (chan stepOutcome[R], error) {
+// The folliwing examples shows how to use Go:
+//
+//		resultChan, err := dbos.Go(ctx, func(ctx context.Context) (string, error) {
+//			return "Hello, World!", nil
+//		})
+//
+//		resultChan := <-resultChan // wait for the channel to receive
+//		if resultChan.err != nil {
+//			// Handle error
+//		}
+//	 result := resultChan.result
+func Go[R any](ctx DBOSContext, fn Step[R], opts ...StepOption) (chan StepOutcome[R], error) {
+	if ctx == nil {
+		// is this the correct return here?
+		return *new(chan StepOutcome[R]), newStepExecutionError("", "", "ctx cannot be nil")
+	}
+
+	if fn == nil {
+		return *new(chan StepOutcome[R]), newStepExecutionError("", "", "step function cannot be nil")
+	}
+
+	// Append WithStepName option to ensure the step name is set. This will not erase a user-provided step name
+	stepName := runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name()
+	opts = append(opts, WithStepName(stepName))
+
+	// Type-erase the function
+	typeErasedFn := StepFunc(func(ctx context.Context) (any, error) { return fn(ctx) })
+
+	result, err := ctx.Go(ctx, typeErasedFn, opts...)
+	// Step function could return a nil result
+	if result == nil {
+		return *new(chan StepOutcome[R]), err
+	}
+
+	// Otherwise type-check and cast the result
+	outcome := <-result
+	typedResult, ok := outcome.result.(R)
+	if !ok {
+		return *new(chan StepOutcome[R]), fmt.Errorf("unexpected result type: expected %T, got %T", *new(R), result)
+	}
+	outcomeChan := make(chan StepOutcome[R], 1)
+	defer close(outcomeChan)
+	outcomeChan <- StepOutcome[R]{
+		result: typedResult,
+		err:    outcome.err,
+	}
+	return outcomeChan, nil
+}
+
+func (c *dbosContext) Go(ctx DBOSContext, fn StepFunc, opts ...StepOption) (chan StepOutcome[any], error) {
 	// create a determistic step ID
 	stepName := runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name()
 	wfState, ok := ctx.Value(workflowStateKey).(*workflowState)
@@ -1244,11 +1282,11 @@ func Go[R any](ctx DBOSContext, fn Step[R], opts ...StepOption) (chan stepOutcom
 	stepID := wfState.nextStepID()
 	opts = append(opts, WithNextStepID(stepID))
 
-	// run step inside a Go routine by passing a stepID 
-	result := make(chan stepOutcome[R], 1)
+	// run step inside a Go routine by passing a stepID
+	result := make(chan StepOutcome[any], 1)
 	go func() {
-		res, err := RunAsStep(ctx, fn, opts...)
-		result <- stepOutcome[R]{
+		res, err := ctx.RunAsStep(ctx, fn, opts...)
+		result <- StepOutcome[any]{
 			result: res,
 			err:    err,
 		}

dbos/workflows_test.go

@@ -902,7 +902,7 @@ func TestGoRunningStepsInsideGoRoutines(t *testing.T) {
 		const numSteps = 100
 		results := make(chan string, numSteps)
 		errors := make(chan error, numSteps)
-		var resultChans []<-chan stepOutcome[string]
+		var resultChans []<-chan StepOutcome[string]
 
 		goWorkflow := func(dbosCtx DBOSContext, input string) (string, error) {
 			for range numSteps {