Toward executor objects #44

maxdml · 2025-07-30T20:55:54Z

This PR:

Encapsulate the state of a "DBOS executor" in an interface exported by the package. This must be typeless, hence:
Preserves compile-time type checking by exposing package-level methods that accept a DBOS executor interface
Attempts to keep the programming interface reasonable by creating a DBOSContext type which holds both DBOS executor functionalities and extends the native context.Context interface.

The mirror functions are made with the same signature so it is more intuitive for users when mocking.

Few things to note/improve/think about

Workflow functions now require our DBOSContext instead of a context.Context. This is because all operations need a context. Steps on the other hand can accept a normal context.Context if they wish to, and upcast it to DBOSContext if they need to do DBOS stuff.
DBOSContext is local to a process. This means the user cannot expect a context to: 1) be passed to the workflow at recovery 2) be passed to a queued task by the queue runner. This is quite un-intuitive w.r.t native Golang context.Context
SetEvent / Send types are not greatly named
We should find a way for library specific contexts to play together. Right now a gin handler looks like: func checkoutEndpoint(c *gin.Context, dbosCtx dbos.DBOSContext, logger *logrus.Logger) {

Another thing this PR does, on the path of improving the step interface, is allowing users to pass step parameters through the context, using the WithValue method. We'll have to improve on this as well: right now WithValue does nothing if the provided interface is not a concrete dbosContext (our internal struct implementing DBOSContext), so a user cannot rely on this in their tests if they mock DBOSContext.

Immediate next PRs:

Move logger, queue registry in DBOSContext
Tighten usage of context to manage resources like queue runner
Add tests running workflows inside goroutines
Improve step UX

DBOSContext exposes all the DBOS methods an end-user is expected to write in their code:

type DBOSContext interface {
	context.Context

	// Context Lifecycle
	Launch() error
	Shutdown()

	// Workflow operations
	RunAsStep(_ DBOSContext, fn StepFunc, input ...any) (any, error)
	RunAsWorkflow(_ DBOSContext, fn WorkflowFunc, input any, opts ...WorkflowOption) (WorkflowHandle[any], error)
	Send(_ DBOSContext, input WorkflowSendInputInternal) error
	Recv(_ DBOSContext, input WorkflowRecvInput) (any, error)
	SetEvent(_ DBOSContext, input WorkflowSetEventInput) error
	GetEvent(_ DBOSContext, input WorkflowGetEventInput) (any, error)
	Sleep(duration time.Duration) (time.Duration, error)
	GetWorkflowID() (string, error)

	// Workflow management
	RetrieveWorkflow(_ DBOSContext, workflowID string) (WorkflowHandle[any], error)

	// Accessors
	GetApplicationVersion() string
	GetExecutorID() string
	GetApplicationID() string
}

New library usage:

	dbosContext, err := dbos.NewDBOSContext(dbos.Config{
		AppName:     "widget_store_go",
		DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
	})
	if err != nil {
		logger.WithError(err).Fatal("DBOS initialization failed")
	}
	dbos.RegisterWorkflow(dbosContext, checkoutWorkflow)
	dbos.RegisterWorkflow(dbosContext, dispatchOrderWorkflow)

	err = dbosContext.Launch()
	if err != nil {
		logger.WithError(err).Fatal("DBOS service start failed")
	}
	defer dbosContext.Shutdown()

Starting workflows:

	handle, err := dbos.RunAsWorkflow(dbosCtx, checkoutWorkflow, "", dbos.WithWorkflowID(idempotencyKey))

The testing can be done as:

	dbosContextMock := mocks.NewMockDBOSContext(t)

	// Test running the wrapped workflow
	t.Run("Payment fails", func(t *testing.T) {
		wfID := "test-workflow-id"

		// Set expectations on what DBOS stuff that happens within the workflow
		dbosContextMock.On("GetWorkflowID").Return(wfID, nil)
		dbosContextMock.On("RunAsStep", dbosContextMock, mock.Anything, mock.Anything).Return(1, nil).Once()
		dbosContextMock.On("RunAsStep", dbosContextMock, mock.Anything, mock.Anything).Return(false, nil).Once()
		dbosContextMock.On("RunAsStep", dbosContextMock, mock.Anything, mock.Anything).Return("", nil).Once()
		dbosContextMock.On("SetEvent", dbosContextMock, mock.Anything).Return(nil).Once()

		res, err := checkoutWorkflow(dbosContextMock, "")
		if err != nil {
			t.Fatalf("checkout workflow failed: %v", err)
		}
		if res != "" {
			t.Fatalf("expected empty result, got %s", res)
		}

		dbosContextMock.AssertExpectations(t)
	})

kraftp

This is getting better. Really want to see what it looks like in code. One thing that jumps out is that we're now passing name into RunAsStep and RunAsWorkflow. That's clunky and probably not necessary--even if the type-erased function has the wrong name, we can extract the right name in the wrapper and pass it through the context.

maxdml · 2025-07-31T16:26:52Z

This is getting better. Really want to see what it looks like in code. One thing that jumps out is that we're now passing name into RunAsStep and RunAsWorkflow. That's clunky and probably not necessary--even if the type-erased function has the wrong name, we can extract the right name in the wrapper and pass it through the context.

Correct and I've been looking into doing exactly this. However, setting it in the context means using the Value method which means the user would.. have to mock it when passing a mocked DBOSContext. And when calling a package-level DBOS function, the user should only have to mock the mirror interface method.

The solution is likely to pass it in the options -- thus allowing user to optionally set workflow and step names, which we planned to do eventually.

This works well for workflows but, won't work with steps if we want to remove functional step options for a revisited step interface accepting a variadic any argument. Looking into it :)

kraftp · 2025-07-31T16:39:47Z

This is getting better. Really want to see what it looks like in code. One thing that jumps out is that we're now passing name into RunAsStep and RunAsWorkflow. That's clunky and probably not necessary--even if the type-erased function has the wrong name, we can extract the right name in the wrapper and pass it through the context.

Correct and I've been looking into doing exactly this. However, setting it in the context means using the Value method which means the user would.. have to mock it when passing a mocked DBOSContext. And when calling a package-level DBOS function, the user should only have to mock the mirror interface method.

The solution is likely to pass it in the options -- thus allowing user to optionally set workflow and step names, which we planned to do eventually.

This works well for workflows but, won't work with steps if we want to remove functional step options for a revisited step interface accepting a variadic any argument. Looking into it :)

There has to be a workaround for this--worst case, an in-memory map from type-erased functions to names.

maxdml · 2025-07-31T17:04:44Z

There has to be a workaround for this--worst case, an in-memory map from type-erased functions to names.

Yup, that's one option, but it won't work when we move all the functional options to the context. Anything the users -- or us, inside the Generic function -- set through the context has to go through values like ctx.WithStepParams and has to be mocked by the interface provider

kraftp · 2025-07-31T18:22:59Z

There has to be a workaround for this--worst case, an in-memory map from type-erased functions to names.

Yup, that's one option, but it won't work when we move all the functional options to the context. Anything the users -- or us, inside the Generic function -- set through the context has to go through values like ctx.WithStepParams and has to be mocked by the interface provider

I'm sure there's a way to circumvent that, but we don't need to figure it out right now--let's get this interface built out, then fix steps, then we'll clean up the details.

maxdml · 2025-07-31T23:06:25Z

There has to be a workaround for this--worst case, an in-memory map from type-erased functions to names.

Yup, that's one option, but it won't work when we move all the functional options to the context. Anything the users -- or us, inside the Generic function -- set through the context has to go through values like ctx.WithStepParams and has to be mocked by the interface provider

I'm sure there's a way to circumvent that, but we don't need to figure it out right now--let's get this interface built out, then fix steps, then we'll clean up the details.

Fair enough -- the topics are sufficiently intertwined though that I took the time to find a basic solution that we will improve on. Specifically:

Expose a WithValue method used to create a new DBOSContext with a new context value. This is in any case something Go users will expect from a context.
Use that to allow users to pass in the step name and other step parameters
In the generic RunAsStep, store the mapping type erased fqn -> user provided fqn
In the interface RunAsStep, figure out which one to use (user provided step name or the one we stored)

This seems to work well and I can now almost call steps without parameters.

Workflow names are passed through a new functional option.

I did have to fix the way I implemented the Generic RunAsWorkflow. To return a typed handle to the user, this unfortunately as to create a new goroutine to pass in the workflow outcome from an untyped channel to a typed one. Maybe there's way to improve on that too.

Overall the solution is coming together and I am now updating the tests.

… do not have the right concrete type

…all interface method directly during recovery / scheduled wfs

…cancellable context

maxdml · 2025-08-01T23:58:32Z

dbos/dbos.go

+	// FIXME: cancellation now has to go through the DBOSContext
+	ctx, cancel := context.WithCancel(c.ctx)
+	c.queueRunnerCtx = ctx
+	c.queueRunnerCancelFunc = cancel
+	c.queueRunnerDone = make(chan struct{})


In a next PR I'll tighten up the usage of DBOSContext to manage its owned resources

maxdml · 2025-08-02T00:01:41Z

dbos/queue.go

-		getLogger().Warn("NewWorkflowQueue called after DBOS initialization, dynamic registration is not supported")
-		return WorkflowQueue{}
-	}
+	// TODO: Add runtime check for post-initialization registration if needed


Will add this in a next PR + prevention of post Launch wf registration

maxdml · 2025-08-02T00:02:43Z

dbos/queue.go

 					}
 				}

+				// XXX this demonstrate why contexts cannot be used globally -- the task does not inherit the context used in the program that enqueued it


here ctx is the root context created by the user.

maxdml · 2025-08-02T00:14:05Z

dbos/system_database.go


 	// Queues
-	DequeueWorkflows(ctx context.Context, queue WorkflowQueue) ([]dequeuedWorkflow, error)
+	DequeueWorkflows(ctx context.Context, queue WorkflowQueue, executorID, applicationVersion string) ([]dequeuedWorkflow, error)


will cleanup inputs (for a struct) in later PRs

maxdml · 2025-08-02T00:15:46Z

dbos/system_database.go

+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+		}
+


AwaitWorkflowResult should be cancellable -- work for another PR but placing this here in the meantime.

maxdml · 2025-08-02T00:16:58Z

dbos/system_database.go

-	destinationID string
-	message       any
-	topic         string
+type WorkflowSendInputInternal struct {


Terrible name that must be made public for the DBOSContext to be implementable (DBOSContext.Send must take a typeless input)

maxdml · 2025-08-02T00:18:35Z

dbos/utils_test.go

-	err = Launch()
-	if err != nil {
-		t.Fatalf("failed to launch DBOS instance: %v", err)
-	}


Most tests don't need Launch(). Only useful for queues, recovery thread, and notifications

maxdml · 2025-08-02T00:32:10Z

dbos/workflow.go

+// RegisterWorkflow is generically typed, allowing us to register the workflow input and output types for gob encoding
+// The registered workflow is wrapped in a typed-erased wrapper which performs runtime type checks and conversions
+// To execute the workflow, use DBOSContext.RunAsWorkflow
+func RegisterWorkflow[P any, R any](ctx DBOSContext, fn GenericWorkflowFunc[P, R], opts ...workflowRegistrationOption) {


One drawback of not wrapping anymore and returning the wrapped function, is that we can't automatically inject, at runtime, the registration options (today only max retries), when the user calls RunAsWorkflow.

This is very annoying because it means that, if we want the users to not have to pass it for every workflow, we must add registration option to the registry and allow RunAsWorkflow, to, at runtime, fetch the registration option (from the registry).

The registry is keyed by the provided workflow name, here the FQN -- maybe a user provided name in the future.

Now, consider that only the package-level RunAsWorkflow nows the FQN. The interface version receives a typeless wrapper that has another name. The registry is attached to the interface, so the package-level object cannot make any assumption about it.

The way we solve this is by adding a private withWorkflowName option to the WorkflowOptions. The package-level RunAsWorkflow injects that option, such that the Interface version of RunAsWorkflow can always retrieve its registered name from the parameters and get its register-time options. The wrappers we create during registration (for recovery and scheduled workflows) does the same (they could inject the the max retry option, but this way we have a unified way to get the name from the interface method.)

maxdml · 2025-08-02T00:37:34Z

dbos/workflow.go

+	// Create a typed channel for the user to get a typed handle
+	if handle, ok := handle.(*workflowHandle[any]); ok {
+		typedOutcomeChan := make(chan workflowOutcome[R], 1)
+
+		go func() {
+			defer close(typedOutcomeChan)
+			outcome := <-handle.outcomeChan
+
+			resultErr := outcome.err
+			var typedResult R
+			if typedRes, ok := outcome.result.(R); ok {
+				typedResult = typedRes
+			} else { // This should never happen
+				typedResult = *new(R)
+				typeErr := fmt.Errorf("unexpected result type: expected %T, got %T", *new(R), outcome.result)
+				resultErr = errors.Join(resultErr, typeErr)
+			}
+
+			typedOutcomeChan <- workflowOutcome[R]{
+				result: typedResult,
+				err:    resultErr,
+			}
+		}()
+
+		typedHandle := &workflowHandle[R]{
+			workflowID:  handle.workflowID,
+			outcomeChan: typedOutcomeChan,
+			dbosContext: handle.dbosContext,
+		}
+
+		return typedHandle, nil
+	}


We cannot return a generically typed handle directly from the interface RunAsWorkflow. This is because the outcome channel itself is typed. This solution has an intermediate goroutine pipe the untyped result from the interface RunAsWorfklow to a typed channel, so we can create & return a typed handle to the user. Would love to be suggested alternative ideas.

maxdml · 2025-08-02T00:39:02Z

dbos/workflow.go


 	// Check if we are within a workflow (and thus a child workflow)
-	parentWorkflowState, ok := ctx.Value(workflowStateKey).(*workflowState)
+	parentWorkflowState, ok := c.Value(workflowStateKey).(*workflowState)


I am still in the process of figuring of whether this will work during child workflows recovery. Because the recovery thread will use the root context, which today doesn't not have a workflowStateKey. The recovery thread might be able to reconstruct this, however.

maxdml · 2025-08-02T00:41:59Z

dbos/workflow.go

+type StepFunc func(ctx context.Context, input any) (any, error)
+type GenericStepFunc[P any, R any] func(ctx context.Context, input P) (R, error)
+
+const StepParamsKey DBOSContextKey = "stepParams"


This is how users can set step parameters, without passing functional options nor a new input parameter.

maxdml · 2025-08-02T00:48:03Z

dbos/workflow.go

-func WithMaxInterval(maxInterval time.Duration) stepOption {
-	return func(p *StepParams) {
-		p.MaxInterval = maxInterval
+var typeErasedStepNameToStepName = make(map[string]string)


Again some pain:

The user might or might not provide a custom step name

The package level method knows the true name of the function

The interface method sees only the typed erased wrapper name

We want package-level and interface methods to have the same signature

No functional options, no "step params" to smooth the step UX

--> We use an in-memory map to pass the function name to the interface method, which will use it for a default name if the user didn't provide any.

maxdml · 2025-08-02T00:48:45Z

dbos/workflow.go

 }

-type WorkflowSetEventInput[R any] struct {
+type WorkflowSetEventInputGeneric[R any] struct {


maxdml · 2025-08-02T00:50:06Z

dbos/workflows_test.go

+	stepCtx := WithValue(dbosCtx, StepParamsKey, &StepParams{
+		MaxRetries:   5,
+		BaseInterval: 1 * time.Millisecond,
+		MaxInterval:  10 * time.Millisecond,
+	})

-var (
-	stepWithinAStepWf = WithWorkflow(stepWithinAStepWorkflow)
-	stepRetryWf       = WithWorkflow(stepRetryWorkflow)
-)
+	return RunAsStep(stepCtx, stepRetryAlwaysFailsStep, input)


step params in use

maxdml added 5 commits July 28, 2025 22:27

wip

8d5a2ca

update tests

b66a2aa

wip

e346712

new interface

8200442

cleanup interface + RunAsWorkflow

897ba1a

kraftp reviewed Jul 31, 2025

View reviewed changes

maxdml added 19 commits July 31, 2025 16:07

fix RunAsWorkflow and work on new step interface

e1eb063

remove register functions from interface -- simply does nothing if we…

c4f0c60

… do not have the right concrete type

fix

589b6d3

fix and nits

38417e2

WIP: first pass at updating the tests

b6eb296

shutdown logs + check if launched

6c332ce

cleanup

d0f4272

fix step output handling + retrieval of registration-time options + c…

c996d7a

…all interface method directly during recovery / scheduled wfs

fix qrunner context for now

4a3fcd2

update tests

fb97d3d

simpler

c3ea87b

fix bug in WithValue

7a54132

comment

e4a9806

identify uncancellable AwaitWorkflowResult -- will have to pass it a …

ab75aaf

…cancellable context

add defensive type check

60808e3

wrapper should use provided context

81c2fde

typo

5149aa5

fix small race in test + check recovery handles == original handles

cebb49b

nit

2a77ad8

Revert changes to step interface for now

aad3c37

maxdml commented Aug 1, 2025

View reviewed changes

maxdml marked this pull request as ready for review August 1, 2025 23:58