AI Model Prompting
You may want assistance from an AI model in building a DBOS application. To make sure your model has the latest information on how to use DBOS, provide it with this prompt.
You may also want to use the DBOS MCP server so your model can directly access your application's workflows and steps.
How To Use
First, use the click-to-copy button in the top right of the code block to copy the full prompt to your clipboard. Then, paste into your AI tool of choice (for example OpenAI's ChatGPT or Anthropic's Claude). This adds the prompt to your AI model's context, giving it up-to-date instructions on how to build an application with DBOS.
If you are using an AI-powered IDE, you can add this prompt to your project's context. For example:
- Claude Code: Add the prompt, or a link to it, to your CLAUDE.md file.
- Cursor: Add the prompt to your project rules.
- Zed: Copy the prompt to a file in your project, then use the
/filecommand to add the file to your context. - GitHub Copilot: Create a
.github/copilot-instructions.mdfile in your repository and add the prompt to it.
Prompt
# Build Reliable Applications With DBOS
## Guidelines
- Respond in a friendly and concise manner
- Ask clarifying questions when requirements are ambiguous
- Generate code in Golang using the DBOS library.
- You MUST import everything used in the code you generate
- You SHALL keep all code in a single file unless otherwise specified.
- DBOS does NOT stand for anything.
## Workflow Guidelines
Workflows provide durable execution so you can write programs that are resilient to any failure.
Workflows are comprised of steps, which are ordinary Golang functions called with dbos.RunAsStep.
When using DBOS workflows, you should call any function that performs complex operations or accesses external APIs or services as a step using dbos.RunAsStep.
If a workflow is interrupted for any reason (e.g., an executor restarts or crashes), when your program restarts the workflow automatically resumes execution from the last completed step.
- If asked to add DBOS to existing code, you MUST ask which function to make a workflow. Do NOT recommend any changes until they have told you what function to make a workflow. Do NOT make a function a workflow unless SPECIFICALLY requested.
- When making a function a workflow, you should make all functions it calls steps. Do NOT change the functions in any way.
- Do NOT make functions steps unless they are DIRECTLY called by a workflow.
- If the workflow function performs a non-deterministic action, you MUST move that action to its own function and make that function a step. Examples of non-deterministic actions include accessing an external API or service, accessing files on disk, generating a random number, of getting the current time.
- Do NOT start goroutines from workflows or use select in workflows. Instead, use DBOS's durable `dbos.Go` and `dbos.Select` functions which provide deterministic replay. For more complex parallel execution, use DBOS.RunWorkflow and DBOS queues.
- Do NOT range over a map to call steps or start workflows: Go map iteration order is random, which breaks determinism. Sort the keys first (e.g. `slices.Sorted(maps.Keys(m))`) and iterate over the sorted slice.
- DBOS workflows and steps should NOT have side effects in memory outside of their own scope. They can access global variables, but they should NOT create or update global variables or variables outside their scope.
- Do NOT call DBOS context methods (DBOS.Send, DBOS.Recv, DBOS.RunWorkflow, DBOS.RunAsTransaction, DBOS.Enqueue, DBOS.Go, DBOS.Sleep, DBOS.GetEvent, DBOS.CloseStream, handle.GetResult, or workflow/schedule management writes) from a step — they return an error. Calling one step function from another is fine (it runs inline as part of the enclosing step), and DBOS.SetEvent, DBOS.WriteStream, and read/list operations are allowed from steps.
## DBOS Lifecycle Guidelines
DBOS should be installed and imported from the `github.com/dbos-inc/dbos-transact-golang/dbos` package.
DBOS programs MUST have a main file (typically 'main.go') that creates all objects and workflow functions during startup.
Any DBOS program MUST create and launch a DBOS context in their main function.
All workflows must be registered BEFORE DBOS is launched.
Queues may be registered at any time, including after launch.
```go
func main() {
dbosContext, err := dbos.NewContext(context.Background(), dbos.Config{
AppName: "dbos-starter",
ApplicationVersion: "0.1.0",
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
})
if err != nil {
panic(fmt.Sprintf("Initializing DBOS failed: %v", err))
}
dbos.RegisterWorkflow(dbosContext, workflow)
err = dbos.Launch(dbosContext)
if err != nil {
panic(fmt.Sprintf("Launching DBOS failed: %v", err))
}
defer dbos.Shutdown(dbosContext, 5 * time.Second)
}
```
Here is an example main function using Gin:
```go
import (
"context"
"fmt"
"net/http"
"os"
"time"
"github.com/dbos-inc/dbos-transact-golang/dbos"
"github.com/gin-gonic/gin"
)
func main() {
dbosContext, err := dbos.NewContext(context.Background(), dbos.Config{
AppName: "dbos-starter",
ApplicationVersion: "0.1.0",
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
})
if err != nil {
panic(fmt.Sprintf("Initializing DBOS failed: %v", err))
}
dbos.RegisterWorkflow(dbosContext, workflow)
err = dbos.Launch(dbosContext)
if err != nil {
panic(fmt.Sprintf("Launching DBOS failed: %v", err))
}
defer dbos.Shutdown(dbosContext, 5 * time.Second)
r := gin.Default()
r.GET("/", func(c *gin.Context) {
handle, err := dbos.RunWorkflow(dbosContext, workflow, "")
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("Error in DBOS workflow: %v", err)})
return
}
result, err := handle.GetResult()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("Error in DBOS workflow: %v", err)})
return
}
c.JSON(http.StatusOK, gin.H{"result": result})
})
r.Run(":8080")
}
```
## Workflow and Steps Examples
Simple example:
```go showLineNumbers title="main.go"
package main
import (
"context"
"fmt"
"os"
"time"
"github.com/dbos-inc/dbos-transact-golang/dbos"
)
func workflow(ctx dbos.Context, _ string) (string, error) {
_, err := dbos.RunAsStep(ctx, stepOne)
if err != nil {
return "failure", err
}
_, err = dbos.RunAsStep(ctx, stepTwo)
if err != nil {
return "failure", err
}
return "success", err
}
func stepOne(ctx context.Context) (string, error) {
fmt.Println("Step one completed")
return "success", nil
}
func stepTwo(ctx context.Context) (string, error) {
fmt.Println("Step two completed")
return "success", nil
}
func main() {
dbosContext, err := dbos.NewContext(context.Background(), dbos.Config{
AppName: "dbos-starter",
ApplicationVersion: "0.1.0",
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
})
if err != nil {
panic(fmt.Sprintf("Initializing DBOS failed: %v", err))
}
dbos.RegisterWorkflow(dbosContext, workflow)
err = dbos.Launch(dbosContext)
if err != nil {
panic(fmt.Sprintf("Launching DBOS failed: %v", err))
}
defer dbos.Shutdown(dbosContext, 5 * time.Second)
handle, err := dbos.RunWorkflow(dbosContext, workflow, "")
if err != nil {
panic(fmt.Sprintf("Error in DBOS workflow: %v", err))
}
result, err := handle.GetResult()
if err != nil {
panic(fmt.Sprintf("Error in DBOS workflow: %v", err))
}
fmt.Println("Workflow result:", result)
}
```
Example with Gin:
```go showLineNumbers title="main.go"
package main
import (
"context"
"fmt"
"net/http"
"os"
"time"
"github.com/dbos-inc/dbos-transact-golang/dbos"
"github.com/gin-gonic/gin"
)
func workflow(ctx dbos.Context, _ string) (string, error) {
_, err := dbos.RunAsStep(ctx, stepOne)
if err != nil {
return "failure", err
}
for range 5 {
fmt.Println("Press Control + C to stop the app...")
dbos.Sleep(ctx, time.Second)
}
_, err = dbos.RunAsStep(ctx, stepTwo)
if err != nil {
return "failure", err
}
return "success", err
}
func stepOne(ctx context.Context) (string, error) {
fmt.Println("Step one completed")
return "success", nil
}
func stepTwo(ctx context.Context) (string, error) {
fmt.Println("Step two completed")
return "success", nil
}
func main() {
dbosContext, err := dbos.NewContext(context.Background(), dbos.Config{
AppName: "dbos-starter",
ApplicationVersion: "0.1.0",
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
})
if err != nil {
panic(fmt.Sprintf("Initializing DBOS failed: %v", err))
}
dbos.RegisterWorkflow(dbosContext, workflow)
err = dbos.Launch(dbosContext)
if err != nil {
panic(fmt.Sprintf("Launching DBOS failed: %v", err))
}
defer dbos.Shutdown(dbosContext, 5 * time.Second)
r := gin.Default()
r.GET("/", func(c *gin.Context) {
handle, err := dbos.RunWorkflow(dbosContext, workflow, "")
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("Error in DBOS workflow: %v", err)})
return
}
result, err := handle.GetResult()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("Error in DBOS workflow: %v", err)})
return
}
c.JSON(http.StatusOK, gin.H{"result": result})
})
r.Run(":8080")
}
```
Example with queues:
```go showLineNumbers title="main.go"
package main
import (
"context"
"fmt"
"net/http"
"os"
"time"
"github.com/dbos-inc/dbos-transact-golang/dbos"
"github.com/gin-gonic/gin"
)
func taskWorkflow(ctx dbos.Context, i int) (int, error) {
dbos.Sleep(ctx, 5*time.Second)
fmt.Printf("Task %d completed\n", i)
return i, nil
}
func queueWorkflow(ctx dbos.Context, queueName string) (int, error) {
fmt.Println("Enqueuing tasks")
queue, err := dbos.RetrieveQueue(ctx, queueName)
if err != nil {
return 0, err
}
handles := make([]dbos.WorkflowHandle[int], 10)
for i := range 10 {
handle, err := dbos.RunWorkflow(ctx, taskWorkflow, i, dbos.WithQueue(queue))
if err != nil {
return 0, err
}
handles[i] = handle
}
results := make([]int, 10)
for i, handle := range handles {
result, err := handle.GetResult()
if err != nil {
return 0, err
}
results[i] = result
}
fmt.Printf("Successfully completed %d tasks\n", len(results))
return len(results), nil
}
func main() {
dbosContext, err := dbos.NewContext(context.Background(), dbos.Config{
AppName: "dbos-starter",
ApplicationVersion: "0.1.0",
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
})
if err != nil {
panic(fmt.Sprintf("Initializing DBOS failed: %v", err))
}
dbos.RegisterWorkflow(dbosContext, queueWorkflow)
dbos.RegisterWorkflow(dbosContext, taskWorkflow)
err = dbos.Launch(dbosContext)
if err != nil {
panic(fmt.Sprintf("Launching DBOS failed: %v", err))
}
defer dbos.Shutdown(dbosContext, 5 * time.Second)
_, err = dbos.RegisterQueue(dbosContext, "queue")
if err != nil {
panic(fmt.Sprintf("Registering queue failed: %v", err))
}
r := gin.Default()
r.GET("/", func(c *gin.Context) {
handle, err := dbos.RunWorkflow(dbosContext, queueWorkflow, "queue")
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("Error in DBOS workflow: %v", err)})
return
}
result, err := handle.GetResult()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("Error in DBOS workflow: %v", err)})
return
}
c.JSON(http.StatusOK, gin.H{"result": result})
})
r.Run(":8080")
}
```
## Workflow Documentation
Workflows provide **durable execution** so you can write programs that are **resilient to any failure**.
Workflows are comprised of steps, which wrap ordinary Go functions.
If a workflow is interrupted for any reason (e.g., an executor restarts or crashes), when your program restarts the workflow automatically resumes execution from the last completed step.
To write a workflow, register a Go function with `RegisterWorkflow`.
Workflow registration must happen before launching the DBOS context with `dbos.Launch()`
The function's signature must match:
```go
type Workflow[P any, R any] func(ctx Context, input P) (R, error)
```
In other words, a workflow must take in a DBOS context and one other input of any serializable (json-encodable) type and must return one output of any serializable type and error.
For example:
```go
func stepOne(ctx context.Context) (string, error) {
fmt.Println("Step one completed")
return "success", nil
}
func stepTwo(ctx context.Context) (string, error) {
fmt.Println("Step two completed")
return "success", nil
}
func workflow(ctx dbos.Context, _ string) (string, error) {
_, err := dbos.RunAsStep(ctx, stepOne)
if err != nil {
return "failure", err
}
_, err = dbos.RunAsStep(ctx, stepTwo)
if err != nil {
return "failure", err
}
return "success", err
}
func main() {
... // Create the DBOS context
dbos.RegisterWorkflow(dbosContext, workflow)
... // Launch DBOS after registering all workflows
}
```
Call workflows with `RunWorkflow`.
This starts the workflow in the background and returns a workflow handle from which you can access information about the workflow or wait for it to complete and return its result.
Here's an example:
```go
func runWorkflowExample(dbosContext dbos.Context, input string) error {
handle, err := dbos.RunWorkflow(dbosContext, workflow, input)
if err != nil {
return err
}
result, err := handle.GetResult()
if err != nil {
return err
}
fmt.Println("Workflow result:", result)
return nil
}
```
### Workflow IDs and Idempotency
Every time you execute a workflow, that execution is assigned a unique ID, by default a UUID.
You can access this ID through `GetWorkflowID`, or from the handle's `GetWorkflowID` method.
Workflow IDs are useful for communicating with workflows and developing interactive workflows.
You can set the workflow ID of a workflow using `WithWorkflowID` when calling `RunWorkflow`.
Workflow IDs must be **globally unique** for your application.
An assigned workflow ID acts as an idempotency key: if a workflow is called multiple times with the same ID, it executes only once.
This is useful if your operations have side effects like making a payment or sending an email.
For example:
```go
func exampleWorkflow(ctx dbos.Context, input string) (string, error) {
workflowID, err := dbos.GetWorkflowID(ctx)
if err != nil {
return "", err
}
fmt.Printf("Running workflow with ID: %s\n", workflowID)
// ...
return "success", nil
}
func example(dbosContext dbos.Context, input string) error {
myID := "unique-workflow-id-123"
handle, err := dbos.RunWorkflow(dbosContext, exampleWorkflow, input, dbos.WithWorkflowID(myID))
if err != nil {
log.Fatal(err)
}
result, err := handle.GetResult()
if err != nil {
log.Fatal(err)
}
fmt.Println("Result:", result)
return nil
}
```
### Determinism
Workflows are in most respects normal Go functions.
They can have loops, branches, conditionals, and so on.
However, a workflow function must be **deterministic**: if called multiple times with the same inputs, it should invoke the same steps with the same inputs in the same order (given the same return values from those steps).
If you need to perform a non-deterministic operation like accessing the database, calling a third-party API, generating a random number, or getting the local time, you shouldn't do it directly in a workflow function.
Instead, you should do all non-deterministic operations in steps.
:::warning
Go's goroutine scheduler and `select` operation are non-deterministic. You should use them only inside steps, or use the durable `dbos.Go` and `dbos.Select` functions instead.
Go's map iteration order is also random. Don't call steps or start workflows while ranging over a map: sort the keys first (e.g. `slices.Sorted(maps.Keys(m))`) and iterate over the sorted slice.
:::
For example, **don't do this**:
```go
func exampleWorkflow(ctx dbos.Context, input string) (string, error) {
randomChoice := rand.Intn(2)
if randomChoice == 0 {
return dbos.RunAsStep(ctx, stepOne)
} else {
return dbos.RunAsStep(ctx, stepTwo)
}
}
```
Instead, do this:
```go
func generateChoice(ctx context.Context) (int, error) {
return rand.Intn(2), nil
}
func exampleWorkflow(ctx dbos.Context, input string) (string, error) {
randomChoice, err := dbos.RunAsStep(ctx, generateChoice)
if err != nil {
return "", err
}
if randomChoice == 0 {
return dbos.RunAsStep(ctx, stepOne)
} else {
return dbos.RunAsStep(ctx, stepTwo)
}
}
```
### Workflow Timeouts
You can set a timeout for a workflow using its input `Context`. Use `WithTimeout` to obtain a cancellable `Context`, as you would with a normal `context.Context`.
When the timeout expires, the workflow and all its children are cancelled. Cancelling a workflow sets its status to CANCELLED and preempts its execution at the beginning of its next step. You can detach a child workflow by passing it an uncancellable context, which you can obtain with `WithoutCancel`.
Timeouts are **start-to-completion**: if a workflow is enqueued, the timeout does not begin until the workflow is dequeued and starts execution. Also, timeouts are durable: they are stored in the database and persist across restarts, so workflows can have very long timeouts.
```go
func exampleWorkflow(ctx dbos.Context, input string) (string, error) {}
timeoutCtx, cancelFunc := dbos.WithTimeout(dbosCtx, 12*time.Hour)
handle, err := dbos.RunWorkflow(timeoutCtx, exampleWorkflow, "wait-for-cancel")
```
You can also manually cancel the workflow by calling its `cancel` function (or calling CancelWorkflow).
### Durable Sleep
You can use `Sleep` to put your workflow to sleep for any period of time.
This sleep is **durable**—DBOS saves the wakeup time in the database so that even if the workflow is interrupted and restarted multiple times while sleeping, it still wakes up on schedule.
Sleeping is useful for scheduling a workflow to run in the future (even days, weeks, or months from now).
For example:
```go
func exampleWorkflow(ctx dbos.Context, input struct {
TimeToSleep time.Duration
Task string
}) (string, error) {
// Sleep for the specified duration
_, err := dbos.Sleep(ctx, input.TimeToSleep)
if err != nil {
return "", err
}
// Execute the task after sleeping
result, err := dbos.RunAsStep(
ctx,
func(stepCtx context.Context) (string, error) {
return fmt.Sprintf("Completed: %s", input.Task), nil
},
)
if err != nil {
return "", err
}
return result, nil
}
```
### Concurrent Steps
DBOS provides durable `Go` and `Select` functions to run multiple steps concurrently within a workflow while preserving durability guarantees.
These are durable alternatives to Go's native goroutines and `select` statement.
`Go` launches a step asynchronously and returns a channel that will receive the result when the step completes.
`Select` waits for the first result from multiple concurrent steps.
```go
func workflow(ctx dbos.Context, _ string) (string, error) {
// Launch two concurrent steps
ch1, err := dbos.Go(ctx, func(ctx context.Context) (string, error) {
return queryServiceA(ctx)
})
if err != nil {
return "", err
}
ch2, err := dbos.Go(ctx, func(ctx context.Context) (string, error) {
return queryServiceB(ctx)
})
if err != nil {
return "", err
}
// Wait for the first result
result, err := dbos.Select(ctx, []<-chan dbos.StepOutcome[string]{ch1, ch2})
if err != nil {
return "", err
}
return result, nil
}
```
### Scheduled Workflows
You can schedule workflows to run on a cron expression.
Schedules are stored in the database and can be created, paused, resumed, and deleted at runtime.
Scheduled workflows are useful for running recurring tasks like data backups, report generation, or cleanup operations.
Scheduled workflows must accept a `dbos.ScheduledWorkflowInput`, which carries the cron tick time and a user-defined `Context` value attached to the schedule:
```go
type ScheduledWorkflowInput struct {
ScheduledTime time.Time `json:"scheduled_time"`
Context json.RawMessage `json:"context,omitempty"`
}
```
The `Context` field holds the raw JSON of the value set on the schedule; decode it inside the workflow with `dbos.DecodeScheduleContext[T](input)`:
```go
func DecodeScheduleContext[T any](input ScheduledWorkflowInput) (T, error)
```
Register the workflow normally, then create a schedule for it using `dbos.CreateSchedule` (or `dbos.ApplySchedules` to declaratively apply a set of schedules on start):
```go
func dailyBackup(ctx dbos.Context, input dbos.ScheduledWorkflowInput) (any, error) {
fmt.Printf("Running daily backup at: %s\n", input.ScheduledTime.Format(time.RFC3339))
... // Perform daily backup operations
return nil, nil
}
func main() {
dbosContext := ... // Initialize DBOS
dbos.RegisterWorkflow(dbosContext, dailyBackup)
err := dbos.Launch(dbosContext)
if err != nil {
log.Fatal(err)
}
// Schedule the workflow to run daily at 2:00 AM
err = dbos.CreateSchedule(dbosContext, dbos.ScheduleSpec{
ScheduleName: "daily-backup",
Workflow: dailyBackup,
Schedule: "0 0 2 * * *",
})
if err != nil {
log.Fatal(err)
}
}
```
Schedules can also be paused, resumed, deleted, backfilled, and triggered at runtime with `dbos.PauseSchedule`, `dbos.ResumeSchedule`, `dbos.DeleteSchedule`, `dbos.BackfillSchedule`, and `dbos.TriggerSchedule`. By default, scheduled invocations are enqueued on an internal queue; set the `QueueName` field of `dbos.ScheduleSpec` to route them to a declared queue for concurrency or rate-limit control.
### Workflow Versioning and Recovery
Because DBOS recovers workflows by re-executing them using information saved in the database, a workflow cannot safely be recovered if its code has changed since the workflow was started.
To guard against this, DBOS _versions_ applications and their workflows.
When DBOS is launched, it computes an application version from a hash of the application source code (this can be overridden through configuration).
All workflows are tagged with the application version on which they started.
When DBOS tries to recover workflows, it only recovers workflows whose version matches the current application version.
This prevents unsafe recovery of workflows that depend on different code.
You cannot change the version of a workflow, but you can use `ForkWorkflow` to restart a workflow from a specific step on a specific code version.
For more information on managing workflow recovery when self-hosting production DBOS applications, check out the guide.
### Workflow Attempts and Recovery
The `Attempts` field in `WorkflowStatus` tracks how many times a workflow has been executed.
- On first execution, `Attempts` is set to `1`.
- If the workflow is enqueued but not yet dequeued, `Attempts` is `0`.
- Each time the workflow is recovered (e.g., after a crash) or dequeued for execution, `Attempts` is incremented by `1`.
You can limit the number of attempts using `WithMaxRecoveryAttempts` when registering a workflow.
If `WithMaxRecoveryAttempts(n)` is set, the workflow may be attempted at most `n + 1` times (one initial execution plus `n` retries).
If this limit is exceeded, the workflow's status is set to `MAX_RECOVERY_ATTEMPTS_EXCEEDED` and it will no longer be recovered automatically.
You can use `ResumeWorkflow` to manually resume a workflow that has exceeded its maximum attempts after fixing the underlying issue.
```go
// Register a workflow that can be attempted at most 4 times (1 initial + 3 retries)
dbos.RegisterWorkflow(dbosContext, myWorkflow, dbos.WithMaxRecoveryAttempts(3))
```
## Steps
When using DBOS workflows, you should call any function that performs complex operations or accesses external APIs or services as a _step_.
If a workflow is interrupted, upon restart it automatically resumes execution from the **last completed step**.
You can use `RunAsStep` to call a function as a step.
For a function to be used as a step, it should return a serializable (json-encodable) value and an error and have this signature:
```go
type Step[R any] func(ctx context.Context) (R, error)
```
Here's a simple example:
```go
func generateRandomNumber(ctx context.Context) (int, error) {
return rand.Int(), nil
}
func workflowFunction(ctx dbos.Context, n int) (int, error) {
randomNumber, err := dbos.RunAsStep(
ctx,
generateRandomNumber,
dbos.WithStepName("generateRandomNumber"),
)
if err != nil {
return 0, err
}
return randomNumber, nil
}
```
You can pass arguments into a step by wrapping it in an anonymous function, like this:
```go
func generateRandomNumber(ctx context.Context, n int) (int, error) {
return rand.IntN(n), nil
}
func workflowFunction(ctx dbos.Context, n int) (int, error) {
randomNumber, err := dbos.RunAsStep(
ctx,
func(stepCtx context.Context) (int, error) {
return generateRandomNumber(stepCtx, n)
},
dbos.WithStepName("generateRandomNumber"),
)
if err != nil {
return 0, err
}
return randomNumber, nil
}
```
You should make a function a step if you're using it in a DBOS workflow and it performs a **nondeterministic** operation.
A nondeterministic operation is one that may return different outputs given the same inputs.
Common nondeterministic operations include:
- Accessing an external API or service, like serving a file from AWS S3, calling an external API like Stripe, or accessing an external data store like Elasticsearch.
- Accessing files on disk.
- Generating a random number.
- Getting the current time.
You **cannot** call, start, or enqueue workflows from within steps.
You also cannot call DBOS methods like `Send` or `Recv` from within steps.
These operations should be performed from workflow functions.
You can call one step from another step, but the called step becomes part of the calling step's execution rather than functioning as a separate step.
### Configurable Retries
You can optionally configure a step to automatically retry any error a set number of times with exponential backoff.
This is useful for automatically handling transient failures, like making requests to unreliable APIs.
Retries are configurable through step options that can be passed to `RunAsStep`.
Available retry configuration options include:
- `WithStepName` - Custom name for the step (default to the Go runtime reflection value)
- `WithStepMaxRetries` - Maximum number of times this step is automatically retried on failure (default 0)
- `WithStepMaxInterval` - Maximum delay between retries (default 5s)
- `WithStepBackoffFactor` - Exponential backoff multiplier between retries (default 2.0)
- `WithStepBaseInterval` - Initial delay between retries (default 100ms)
For example, let's configure this step to retry failures (such as if the site to be fetched is temporarily down) up to 10 times:
```go
func fetchStep(ctx context.Context, url string) (string, error) {
resp, err := http.Get(url)
if err != nil {
return "", err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return "", err
}
return string(body), nil
}
func fetchWorkflow(ctx dbos.Context, inputURL string) (string, error) {
return dbos.RunAsStep(
ctx,
func(stepCtx context.Context) (string, error) {
return fetchStep(stepCtx, inputURL)
},
dbos.WithStepName("fetchFunction"),
dbos.WithStepMaxRetries(10),
dbos.WithStepMaxInterval(30*time.Second),
dbos.WithStepBackoffFactor(2.0),
dbos.WithStepBaseInterval(500*time.Millisecond),
)
}
```
If a step exhausts all retry attempts, it returns an error to the calling workflow.
## Transactions & Datasources
A datasource is a handle to a database you own, over which DBOS can run durable transactions.
A transaction run through a datasource inside a workflow commits your application writes and the DBOS durability record atomically, so it executes exactly once even across crashes and recovery (stronger than a step, which is at-least-once).
Create a datasource with `NewDataSource`, passing a `*pgxpool.Pool` (Postgres/CockroachDB) or `*sql.DB` (SQLite).
It may be called at any time, before or after `Launch()`, and provisions a `transaction_completion` durability table in your database unless it already exists.
If the engine is the same handle as the DBOS system database (passed via `Config.SystemDBPool` or `Config.SQLiteSystemDB`), no `transaction_completion` table is created or managed: application writes and the DBOS checkpoint commit together in a single transaction (detection is by pointer identity, not connection string).
```go
func NewDataSource[E Engine](ctx Context, engine E, opts ...DataSourceOption) (*DataSource, error)
```
Options: `WithDataSourceName(name string)` (logs only, default "datasource"), `WithDataSourceSchema(schema string)` (schema for the durability table, default "dbos").
Run a transaction inside a workflow with `RunAsTransaction`.
The function receives a driver-agnostic `Tx` with `Exec`, `Query`, and `QueryRow` methods; DBOS commits if the function returns successfully and rolls back on error (do not call `Commit`/`Rollback` yourself).
```go
func RunAsTransaction[R any](ctx Context, ds *DataSource, fn Txn[R], opts ...StepOption) (R, error)
pool, _ := pgxpool.New(context.Background(), os.Getenv("APP_DATABASE_URL"))
ds, err := dbos.NewDataSource(dbosContext, pool, dbos.WithDataSourceName("app"))
// Inside a workflow:
n, err := dbos.RunAsTransaction(ctx, ds, func(txCtx context.Context, tx dbos.Tx) (int64, error) {
res, err := tx.Exec(txCtx, "INSERT INTO orders(item) VALUES ($1)", item)
if err != nil {
return 0, err
}
return res.RowsAffected()
}, dbos.WithStepMaxRetries(3))
```
Rules:
- `RunAsTransaction` must be called from within a workflow; it shares the workflow's step counter with `RunAsStep` and accepts the same step options.
- Serialization/deadlock conflicts are retried automatically with a fresh transaction; application errors follow the step retry policy.
- Nesting a `RunAsTransaction` inside another `RunAsTransaction` or inside a `RunAsStep` is rejected with an error — run transactions from workflow code.
## Workflow Communication
DBOS provides a few different ways to communicate with your workflows.
You can:
- Send messages to workflows
- Publish events from workflows for clients to read
- Stream values from workflows to clients
### Workflow Messaging and Notifications
You can send messages to a specific workflow.
This is useful for signaling a workflow or sending notifications to it while it's running.
<img src={require('@site/static/img/workflow-communication/workflow-messages.png').default} alt="DBOS Steps" width="750" className="custom-img"/>
### Send
```go
func Send[P any](ctx Client, destinationID string, message P, topic string, opts ...SendOption) error
```
You can call `Send()` to send a message to a workflow.
Messages can optionally be associated with a topic and are queued on the receiver per topic.
Pass `WithIdempotencyKey(key string)` to make a retried `Send` deliver at most once.
### Recv
```go
func Recv[R any](ctx Context, topic string, timeout time.Duration) (R, error)
```
Workflows can call `Recv()` to receive messages sent to them, optionally for a particular topic.
Each call to `Recv()` waits for and consumes the next message to arrive in the queue for the specified topic, returning an error if the wait times out.
If the topic is not specified, this method only receives messages sent without a topic.
### Messages Example
Messages are especially useful for sending notifications to a workflow.
For example, in an e-commerce application, the checkout workflow, after redirecting customers to a secure payments service, must wait for a notification from that service that the payment has finished processing.
To wait for this notification, the payments workflow uses `Recv()`, executing failure-handling code if the notification doesn't arrive in time:
```go
const PaymentStatusTopic = "payment_status"
func checkoutWorkflow(ctx dbos.Context, orderData OrderData) (string, error) {
// Process initial checkout steps...
// Wait for payment notification with a 5-minute timeout
notification, err := dbos.Recv[PaymentNotification](ctx, PaymentStatusTopic, 5*time.Minute)
if err != nil {
... // Handle timeout or other errors
}
// Handle the notification
if notification.Status == "completed" {
... // Handle the notification.
} else {
... // Handle a failure
}
}
```
A webhook waits for the payment processor to send the notification, then uses `Send()` to forward it to the workflow:
```go
func paymentWebhookHandler(w http.ResponseWriter, r *http.Request) {
// Parse the notification from the payment processor
notification := ...
// Retrieve the workflow ID from notification metadata
workflowID := ...
// Send the notification to the waiting workflow
err := dbos.Send(dbosContext, workflowID, notification, PaymentStatusTopic)
if err != nil {
http.Error(w, "Failed to send notification", http.StatusInternalServerError)
return
}
}
```
### Reliability Guarantees
All messages are persisted to the database, so if `Send` completes successfully, the destination workflow is guaranteed to be able to `Recv` it.
If you're sending a message from a workflow, DBOS guarantees exactly-once delivery.
### Workflow Events
Workflows can publish _events_, which are key-value pairs associated with the workflow.
They are useful for publishing information about the status of a workflow or to send a result to clients while the workflow is running.
<img src={require('@site/static/img/workflow-communication/workflow-events.png').default} alt="DBOS Steps" width="750" className="custom-img"/>
### SetEvent
```go
func SetEvent[P any](ctx Context, key string, message P) error
```
Any workflow can call `SetEvent` to publish a key-value pair, or update its value if has already been published.
### GetEvent
```go
func GetEvent[R any](ctx Client, targetWorkflowID, key string, timeout time.Duration) (R, error)
```
You can call `GetEvent` to retrieve the value published by a particular workflow ID for a particular key.
If the event does not yet exist, this call waits for it to be published, returning an error if the wait times out.
### Events Example
Events are especially useful for writing interactive workflows that communicate information to their caller.
For example, in an e-commerce application, the checkout workflow, after validating an order, directs the customer to a secure payments service to handle credit card processing.
To communicate the payments URL to the customer, it uses events.
The checkout workflow emits the payments URL using `SetEvent()`:
```go
const PaymentURLKey = "payment_url"
func checkoutWorkflow(ctx dbos.Context, orderData OrderData) (string, error) {
// Process order validation...
paymentsURL := ...
err := dbos.SetEvent(ctx, PaymentURLKey, paymentsURL)
if err != nil {
return "", fmt.Errorf("failed to set payment URL event: %w", err)
}
// Continue with checkout process...
}
```
The HTTP handler that originally started the workflow uses `GetEvent()` to await this URL, then redirects the customer to it:
```go
func webCheckoutHandler(dbosContext dbos.Context, w http.ResponseWriter, r *http.Request) {
orderData := parseOrderData(r) // Parse order from request
handle, err := dbos.RunWorkflow(dbosContext, checkoutWorkflow, orderData)
if err != nil {
http.Error(w, "Failed to start checkout", http.StatusInternalServerError)
return
}
// Wait up to 30 seconds for the payment URL event
url, err := dbos.GetEvent[string](dbosContext, handle.GetWorkflowID(), PaymentURLKey, 30*time.Second)
if err != nil {
// Handle a timeout
}
// Redirect the customer
}
```
### Reliability Guarantees
All events are persisted to the database, so the latest version of an event is always retrievable.
Additionally, if `GetEvent` is called in a workflow, the retrieved value is persisted in the database so workflow recovery can use that value, even if the event is later updated.
### Workflow Streaming
Workflows can stream data in real time to clients.
This is useful for streaming results from a long-running workflow or LLM call, or for monitoring and progress reporting.
#### Writing to Streams
```go
func WriteStream[P any](ctx Context, key string, value P) error
```
You can write values to a stream from a workflow or its steps.
A workflow may have any number of streams, each identified by a unique key.
When you are done writing to a stream, you should close it with `CloseStream`.
Otherwise, streams are automatically closed when the workflow terminates.
```go
func CloseStream(ctx Context, key string) error
```
DBOS streams are immutable and append-only.
Writes to a stream from a workflow happen exactly-once.
Writes to a stream from a step happen at-least-once; if a step fails and is retried, it may write to the stream multiple times.
#### Reading from Streams
```go
func ReadStream[R any](ctx Client, workflowID string, key string, opts ...ReadStreamOption) ([]R, bool, error)
```
You can read values from a stream from anywhere.
This function reads all values from a stream identified by a workflow ID and key.
It blocks until the stream is closed or the workflow becomes inactive (status is not `PENDING` or `ENQUEUED`).
It returns the values, whether the stream is closed, and any error.
To read without blocking, pass `WithReadStreamSnapshot()`, which returns as soon as all currently-available values have been drained; pair it with `WithReadStreamFromOffset(offset int)` to poll a stream incrementally.
You can also read from a stream asynchronously, which returns a channel:
```go
func ReadStreamAsync[R any](ctx Client, workflowID string, key string) (<-chan StreamValue[R], error)
```
```go
type StreamValue[R any] struct {
Value R // The stream value (zero value if error/closed)
Err error // Error if one occurred (nil otherwise)
Closed bool // Whether the stream is closed
}
```
#### Streaming Example
```go
func producerWorkflow(ctx dbos.Context, _ string) (string, error) {
err := dbos.WriteStream(ctx, "progress", "step 1 complete")
if err != nil {
return "", err
}
err = dbos.WriteStream(ctx, "progress", "step 2 complete")
if err != nil {
return "", err
}
err = dbos.CloseStream(ctx, "progress")
if err != nil {
return "", err
}
return "done", nil
}
// Blocking read
values, closed, err := dbos.ReadStream[string](ctx, workflowID, "progress")
// Async read: process values as they arrive
ch, err := dbos.ReadStreamAsync[string](ctx, workflowID, "progress")
if err != nil {
return err
}
for streamValue := range ch {
if streamValue.Err != nil {
return streamValue.Err
}
if streamValue.Closed {
break
}
fmt.Printf("Received: %s\n", streamValue.Value)
}
```
You can also read from a stream from outside a DBOS application with a DBOS Client using `ReadStream` or `ReadStreamAsync`.
## Cross-Language Interoperability
DBOS supports multiple languages (Python, TypeScript, Go, Java).
A client in one language can interact with workflows in another language using the **portable JSON** serialization format.
### Portable Workflows
Use `WithPortableWorkflow()` when calling `RunWorkflow` to serialize inputs, outputs, and errors in portable JSON:
```go
handle, err := dbos.RunWorkflow(dbosContext, processOrder, "order-123",
dbos.WithPortableWorkflow(),
)
```
### Portable Communication
Use portable options on `Send`, `SetEvent`, and `WriteStream` for cross-language messaging:
```go
// Send a message readable by any language
dbos.Send(ctx, "workflow-123",
map[string]any{"status": "complete"},
"updates",
dbos.WithPortableSend(),
)
// Set an event readable by any language
dbos.SetEvent(ctx, "progress",
map[string]any{"percent": 75},
dbos.WithPortableSetEvent(),
)
// Write to a stream readable by any language
dbos.WriteStream(ctx, "results",
map[string]any{"item": "processed"},
dbos.WithPortableWriteStream(),
)
```
### Enqueueing Cross-Language Workflows
To enqueue a workflow on an application written in another language from a Go client, pass `PortableWorkflowArgs` as the input (this automatically uses portable JSON):
```go
args := dbos.PortableWorkflowArgs{
PositionalArgs: []any{"order-123", 42},
}
handle, err := dbos.Enqueue[any](
client, "orders", "process_order", args,
dbos.WithEnqueueClassName("OrderProcessor"), // Required for Python/TS/Java targets
)
```
### Portable Errors
Return a `PortableWorkflowError` from a portable workflow to pass structured error info cross-language:
```go
return nil, &dbos.PortableWorkflowError{
Name: "ValidationError",
Message: "invalid input",
Code: 400,
Data: map[string]any{"field": "email"},
}
```
## Alerting
If you are using DBOS Conductor, you can register an alert handler to receive alerts when failure conditions are met.
The handler must be registered before calling `Launch()`. Only one handler is allowed per application.
```go
dbos.SetAlertHandler(dbosContext, func(ruleType string, message string, metadata map[string]string) {
slog.Warn(fmt.Sprintf("Alert received: %s - %s", ruleType, message))
for key, value := range metadata {
slog.Warn(fmt.Sprintf(" %s: %s", key, value))
}
})
```
The handler receives:
- **ruleType**: One of `WorkflowFailure`, `SlowQueue`, or `UnresponsiveApplication`.
- **message**: The alert message.
- **metadata**: Key-value string pairs with additional alert context.
## Queues
You can use queues to run many workflows at once with managed concurrency.
Queues provide _flow control_, letting you manage how many workflows run at once or how often workflows are started.
To create a queue, register it with `RegisterQueue`:
```go
queue, err := dbos.RegisterQueue(dbosContext, "example_queue")
```
`RegisterQueue` persists the queue's configuration to the system database.
It can be called at any time, including after `Launch()`, and the queue's configuration can be changed at runtime.
`RegisterQueue` returns a `Queue` handle.
Keep it somewhere your code can reach it (for example, a package-level variable)—enqueueing a workflow requires the handle, not the queue name.
If you only have the name, fetch the handle with `RetrieveQueue`.
You can then enqueue any workflow by passing the handle to `WithQueue` when calling `RunWorkflow`.
Enqueuing a function submits it for execution and returns a handle to it.
Queued tasks are started in first-in, first-out (FIFO) order.
```go
func processTask(ctx dbos.Context, task string) (string, error) {
// Process the task...
return fmt.Sprintf("Processed: %s", task), nil
}
func example(dbosContext dbos.Context, queue dbos.Queue) error {
// Enqueue a workflow
task := "example_task"
handle, err := dbos.RunWorkflow(dbosContext, processTask, task, dbos.WithQueue(queue))
if err != nil {
return err
}
// Get the result
result, err := handle.GetResult()
if err != nil {
return err
}
fmt.Println("Task result:", result)
return nil
}
```
### Queue Example
Here's an example of a workflow using a queue to process tasks concurrently:
```go
func taskWorkflow(ctx dbos.Context, task string) (string, error) {
// Process the task...
return fmt.Sprintf("Processed: %s", task), nil
}
func queueWorkflow(ctx dbos.Context, queueName string) ([]string, error) {
// Look up the queue handle by name
queue, err := dbos.RetrieveQueue(ctx, queueName)
if err != nil {
return nil, err
}
// Enqueue each task so all tasks are processed concurrently
tasks := []string{"task1", "task2", "task3", "task4", "task5"}
var handles []dbos.WorkflowHandle[string]
for _, task := range tasks {
handle, err := dbos.RunWorkflow(ctx, taskWorkflow, task, dbos.WithQueue(queue))
if err != nil {
return nil, fmt.Errorf("failed to enqueue task %s: %w", task, err)
}
handles = append(handles, handle)
}
// Wait for each task to complete and retrieve its result
var results []string
for i, handle := range handles {
result, err := handle.GetResult()
if err != nil {
return nil, fmt.Errorf("task %d failed: %w", i, err)
}
results = append(results, result)
}
return results, nil
}
func example(dbosContext dbos.Context) error {
handle, err := dbos.RunWorkflow(dbosContext, queueWorkflow, "example_queue")
if err != nil {
return err
}
results, err := handle.GetResult()
if err != nil {
return err
}
for _, result := range results {
fmt.Println(result)
}
return nil
}
```
### Enqueueing from Another Application
Often, you want to enqueue a workflow from outside your DBOS application.
For example, let's say you have an API server and a data processing service.
You're using DBOS to build a durable data pipeline in the data processing service.
When the API server receives a request, it should enqueue the data pipeline for execution on the data processing service.
You can use the DBOS Client to enqueue workflows from outside your DBOS application by connecting directly to your DBOS application's system database.
Since the DBOS Client is designed to be used from outside your DBOS application, workflow and queue metadata must be specified explicitly.
For example, this code enqueues the `dataPipeline` workflow on the `pipelineQueue` queue with a `ProcessInput` argument:
```go
type ProcessInput struct {
TaskID string
Data string
}
type ProcessOutput struct {
Result string
Status string
}
config := dbos.ClientConfig{
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
}
client, err := dbos.NewClient(context.Background(), config)
if err != nil {
log.Fatal(err)
}
defer dbos.Shutdown(client, 5*time.Second)
handle, err := dbos.Enqueue[ProcessOutput](
client,
"pipelineQueue",
"dataPipeline",
ProcessInput{TaskID: "task-123", Data: "data"},
)
if err != nil {
log.Fatal(err)
}
```
### Managing Concurrency
You can control how many workflows from a queue run simultaneously by configuring concurrency limits.
This helps prevent resource exhaustion when workflows consume significant memory or processing power.
#### Worker Concurrency
Worker concurrency sets the maximum number of workflows from a queue that can run concurrently on a single DBOS process.
This is particularly useful for resource-intensive workflows to avoid exhausting the resources of any process.
For example, this queue has a worker concurrency of 5, so each process will run at most 5 workflows from this queue simultaneously:
```go
queue, err := dbos.RegisterQueue(dbosContext, "example_queue", dbos.WithWorkerConcurrency(5))
```
#### Global Concurrency
Global concurrency limits the total number of workflows from a queue that can run concurrently across all DBOS processes in your application.
For example, this queue will have a maximum of 10 workflows running simultaneously across your entire application.
:::warning
Worker concurrency limits are recommended for most use cases.
Take care when using a global concurrency limit as any `PENDING` workflow on the queue counts toward the limit, including workflows from previous application versions
:::
```go
queue, err := dbos.RegisterQueue(dbosContext, "example_queue", dbos.WithGlobalConcurrency(10))
```
### Rate Limiting
You can set _rate limits_ for a queue, limiting the number of functions that it can start in a given period.
Rate limits are global across all DBOS processes using this queue.
For example, this queue has a limit of 100 workflows with a period of 60 seconds, so it may not start more than 100 workflows in 60 seconds:
```go
queue, err := dbos.RegisterQueue(dbosContext, "example_queue",
dbos.WithRateLimiter(&dbos.RateLimiter{
Limit: 100,
Period: 60 * time.Second, // 60 seconds
}))
```
Rate limits are especially useful when working with a rate-limited API, such as many LLM APIs.
### Reconfiguring Queues
Because queue configuration is persisted to the system database, you can change a queue's configuration at runtime without redeploying or restarting your workers.
Workers pick up the new configuration on their next polling iteration.
Use `RetrieveQueue` to fetch a queue, then call its `Set*` methods:
```go
queue, err := dbos.RetrieveQueue(dbosContext, "example_queue")
if err != nil {
return err
}
concurrency := 50
err = queue.SetGlobalConcurrency(dbosContext, &concurrency)
```
If your application calls `RegisterQueue` on startup, the next process to start can overwrite settings you applied at runtime via `Set*` methods.
Either update the `RegisterQueue` call to match the new configuration, or pass `WithQueueOnConflict(dbos.QueueConflictNeverUpdate)` to preserve the runtime changes.
### Deduplication
You can set a deduplication ID for an enqueued workflow using `WithDeduplicationID` when calling `RunWorkflow`.
At any given time, only one workflow with a specific deduplication ID can be enqueued in the specified queue.
If a workflow with a deduplication ID is currently enqueued or actively executing (status `ENQUEUED` or `PENDING`), subsequent workflow enqueue attempts with the same deduplication ID in the same queue will return an error.
Alternatively, use `WithDeduplicationPolicy(dbos.DeduplicationPolicyReturnExisting)` to instead return a handle to the existing workflow holding the deduplication ID.
For example, this is useful if you only want to have one workflow active at a time per user—set the deduplication ID to the user's ID.
**Example syntax:**
```go
func taskWorkflow(ctx dbos.Context, task string) (string, error) {
// Process the task...
return "completed", nil
}
func example(dbosContext dbos.Context, queue dbos.Queue) error {
task := "example_task"
deduplicationID := "user_12345" // Use user ID for deduplication
handle, err := dbos.RunWorkflow(
dbosContext, taskWorkflow, task,
dbos.WithQueue(queue),
dbos.WithDeduplicationID(deduplicationID))
if err != nil {
// Handle deduplication error or other failures
return fmt.Errorf("failed to enqueue workflow: %w", err)
}
result, err := handle.GetResult()
if err != nil {
return fmt.Errorf("workflow failed: %w", err)
}
fmt.Printf("Workflow completed: %s\n", result)
return nil
}
```
### Priority
You can set a priority for an enqueued workflow using `WithPriority` when calling `RunWorkflow`.
Workflows with the same priority are dequeued in **FIFO (first in, first out)** order. Priority values can range from `1` to `2,147,483,647`, where **a low number indicates a higher priority**.
If using priority, you must set `WithPriorityEnabled` on your queue.
:::tip
Workflows without assigned priorities have the highest priority and are dequeued before workflows with assigned priorities.
:::
To use priorities in a queue, you must enable it when creating the queue:
```go
queue, err := dbos.RegisterQueue(dbosContext, "example_queue", dbos.WithPriorityEnabled())
```
**Example syntax:**
```go
func taskWorkflow(ctx dbos.Context, task string) (string, error) {
// Process the task...
return "completed", nil
}
func example(dbosContext dbos.Context, queue dbos.Queue) error {
task := "example_task"
priority := uint(10) // Lower number = higher priority
handle, err := dbos.RunWorkflow(dbosContext, taskWorkflow, task,
dbos.WithQueue(queue),
dbos.WithPriority(priority))
if err != nil {
return err
}
result, err := handle.GetResult()
if err != nil {
return fmt.Errorf("workflow failed: %w", err)
}
fmt.Printf("Workflow completed: %s\n", result)
return nil
}
```
### Partitioned Queues
You can partition queues to distribute work across dynamically created queue partitions.
When you enqueue a workflow on a partitioned queue, you must supply a queue partition key.
In partitioned queues, all flow control (including concurrency and rate limits) is applied to individual partitions instead of the queue as a whole.
For example, to allow each user to run at most one task at a time:
```go
partitionedQueue, err := dbos.RegisterQueue(dbosContext, "user-tasks",
dbos.WithPartitionQueue(),
dbos.WithGlobalConcurrency(1),
)
// Enqueue workflows with partition keys
// Each user's tasks run with separate concurrency limits
handle, err := dbos.RunWorkflow(dbosContext, processTask, taskData,
dbos.WithQueue(partitionedQueue),
dbos.WithQueuePartitionKey(userID),
)
```
### Delayed Execution
You can delay an enqueued workflow's execution using `WithDelay`.
The workflow is initially placed in `DELAYED` status and does not execute.
After the delay expires, it transitions to `ENQUEUED` status and may be dequeued and executed.
```go
remindersQueue, err := dbos.RegisterQueue(dbosContext, "reminders")
if err != nil {
return err
}
// Send a reminder in one hour
handle, err := dbos.RunWorkflow(dbosContext, sendReminder, userID,
dbos.WithQueue(remindersQueue),
dbos.WithDelay(1 * time.Hour),
)
```
When enqueueing from a Client, use `WithEnqueueDelay` instead.
You can dynamically update or shorten the delay of a `DELAYED` workflow using `SetWorkflowDelay`:
```go
// Shorten the delay to 10 seconds from now
err := dbos.SetWorkflowDelay(ctx, handle.GetWorkflowID(), dbos.WithDelayDuration(10*time.Second))
// Or set an absolute deadline
err = dbos.SetWorkflowDelay(ctx, handle.GetWorkflowID(), dbos.WithDelayUntil(time.Now().Add(time.Minute)))
```
### Debouncing
Debouncing delays a workflow's execution until some time has passed since it was last called.
This is useful when rapid successive triggers should be coalesced into a single workflow execution.
```go
func processInput(ctx dbos.Context, input string) (string, error) {
fmt.Printf("Processing input: %s\n", input)
return "processed", nil
}
func main() {
dbosContext, _ := dbos.NewContext(context.Background(), dbos.Config{
AppName: "debounce-example",
ApplicationVersion: "0.1.0",
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
})
dbos.RegisterWorkflow(dbosContext, processInput)
// Create a debouncer with a maximum timeout of 30 seconds
debouncer, err := dbos.NewDebouncer(dbosContext, processInput,
dbos.WithDebouncerTimeout(30*time.Second))
if err != nil {
log.Fatal(err)
}
dbos.Launch(dbosContext)
defer dbos.Shutdown(dbosContext, 5*time.Second)
// Each call to Debounce pushes back the workflow start time by the delay.
// The workflow runs with the most recent input once the delay expires.
handle, err := debouncer.Debounce(dbosContext, "user-123", 5*time.Second, "first input")
if err != nil {
log.Fatal(err)
}
// If this call arrives within 5 seconds, the delay resets and the input updates
handle, err = debouncer.Debounce(dbosContext, "user-123", 5*time.Second, "updated input")
if err != nil {
log.Fatal(err)
}
result, err := handle.GetResult()
fmt.Println("Result:", result) // Processed with "updated input"
}
```
To debounce a workflow method of a configured instance (registered with `WithInstance`), pass the instance with `WithDebouncerInstance`:
```go
debouncer, err := dbos.NewDebouncer(ctx, slack.Send, dbos.WithDebouncerInstance(slack))
```
Debouncers can be created at any time, including after `Launch()`.
### ListenQueues
You can configure which queues the current DBOS process should listen to for workflow execution using `ListenQueues`.
By default, all registered queues are listened to. This allows multiple DBOS processes to share the same queues but listen to different subsets.
```go
dbos.RegisterQueue(ctx, "queue-1")
dbos.RegisterQueue(ctx, "queue-2")
dbos.RegisterQueue(ctx, "queue-3")
// This process only listens to queue-1 and queue-2.
dbos.ListenQueues(ctx, "queue-1", "queue-2")
dbos.Launch(ctx)
```
Queues are identified by name; each call to `ListenQueues` replaces the whole listen set (an empty set listens to every queue), and the set may be changed at any time, including after `Launch()`.
`ListenQueues` only controls what workflows are dequeued, not what workflows can be enqueued.
# Reference
DBOS has two interfaces: `Client` and `Context`, where `Context` embeds (extends) `Client`.
A `Client` connects to your application's system database and can enqueue and manage workflows, queues, and schedules from any process, including from outside a DBOS application (create one with `NewClient`).
A `Context` is at the center of a DBOS-enabled application: it is everything a `Client` is, plus workflow registration and durable execution (create one with `NewContext`).
`Context` extends Go's `context.Context` interface and carries essential state across workflow execution. Workflows and steps receive a new `Context` spun out of the root `Context` you manage. In addition, a `Context` can be used to set workflow timeouts.
DBOS operations are package-level functions whose first parameter tells you who can call them:
- A function taking a **`Client`** (e.g. `Enqueue`, `Send`, `ListWorkflows`, `RegisterQueue`, `CreateSchedule`) accepts a standalone client or any `Context`, because every `Context` **is** a `Client`.
- A function taking a **`Context`** (e.g. `RunWorkflow`, `RunAsStep`, `Recv`, `SetEvent`, `Sleep`) requires a DBOS context; workflow-scoped functions must receive the `Context` passed into the workflow function.
## Lifecycle
### Initialization
You can create a DBOS context using `NewContext`, which takes a `Config` object where `AppName` and one of `DatabaseURL`, `SystemDBPool`, or `SQLiteSystemDB` are mandatory.
```go
func NewContext(ctx context.Context, inputConfig Config) (Context, error)
```
```go
type Config struct {
AppName string // Application name for identification (required)
DatabaseURL string // Connection string to your system database. May be a PostgreSQL (postgres://...) or SQLite (sqlite:...) URL. Exactly one of DatabaseURL, SystemDBPool, or SQLiteSystemDB is required.
SystemDBPool *pgxpool.Pool // A custom Postgres/CockroachDB connection pool for your system database. Optional; takes precedence over DatabaseURL. Mutually exclusive with SQLiteSystemDB.
SQLiteSystemDB *sql.DB // A custom SQLite handle (e.g. from modernc.org/sqlite) to use as your system database. Optional; takes precedence over DatabaseURL. Mutually exclusive with SystemDBPool.
DatabaseSchema string // Database schema name (defaults to "dbos"; Postgres only)
Logger *slog.Logger // Custom logger instance (defaults to a new slog logger)
ConductorURL string // DBOS conductor service URL (optional)
ConductorAPIKey string // DBOS conductor API key (optional)
ConductorExecutorMetadata map[string]any // Metadata used to identify this executor on the Conductor dashboard (optional, must be JSON-serializable)
ApplicationVersion string // Application version (optional)
ExecutorID string // Executor ID (optional)
EnablePatching bool // Enable the patching system for Patch/DeprecatePatch (default: false)
Serializer Serializer[any] // Custom serializer for workflow inputs, outputs, and events (defaults to a JSON serializer)
SchedulerPollingInterval time.Duration // How often database-backed schedules are reconciled (default: 30s)
SystemDBStartupTimeout time.Duration // Maximum time for system database connection and migrations (default: 2 minutes)
}
```
For example:
```go
dbosContext, err := dbos.NewContext(context.Background(), dbos.Config{
AppName: "dbos-starter",
ApplicationVersion: "0.1.0",
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
})
if err != nil {
panic(err)
}
```
The newly created Context must be launched with `Launch()` before use and should be shut down with Shutdown() at program termination.
DBOS can back its system database with either Postgres (recommended for production; pass a `postgres://` `DatabaseURL` or a `*pgxpool.Pool` as `SystemDBPool`) or SQLite (useful for local development, testing, and single-node deployments; pass a `sqlite:` `DatabaseURL` or a `*sql.DB` as `SQLiteSystemDB`). SQLite support is not linked into your binary by default: to use SQLite, register the driver with one blank import anywhere in your binary: `import _ "github.com/dbos-inc/dbos-transact-golang/dbos/driver/sqlite"`. Without it, `NewContext` (or `NewClient`) fails at startup with an error naming this import. The import registers `modernc.org/sqlite`, a pure-Go driver requiring no cgo. SQLite `DatabaseURL` examples: `"sqlite:dbos.db"` (relative file) or `"sqlite:/var/lib/dbos.db"` (absolute file). `DatabaseSchema` applies to Postgres only.
### launch
```go
dbos.Launch(ctx Context) error
```
Launch the following resources managed by a `Context`:
- A system database connection pool
- A workflow scheduler
- A workflow queue runner
- (Optionally) a Conductor connection
In addition, `Launch()` may perform workflow recovery.
`Launch()` should be called by your program during startup before running any workflows.
### Shutdown
```go
func Shutdown(c Client, timeout time.Duration) error
```
Gracefully shutdown the DBOS runtime, waiting for workflows to complete and cleaning up resources. Accepts either a `Context` or a standalone `Client`. When you shutdown a `Context`, the underlying `context.Context` will be cancelled, which signals all DBOS resources they should stop executing, including workflows and steps. Shutting down a standalone client releases its system database connection pool.
**Parameters:**
- **timeout**: The time to wait for DBOS resources to gracefully terminate.
## Context management
### WithTimeout
```go
func WithTimeout(ctx Context, timeout time.Duration) (Context, context.CancelFunc)
```
`WithTimeout` returns a copy of the DBOS context with a timeout. The returned context will be canceled after the specified duration. See workflow timeouts for usage.
### WithoutCancel
```go
func WithoutCancel(ctx Context) Context
```
`WithoutCancel` returns a copy of the DBOS context that is not canceled when the parent context is canceled. This is useful to detach child workflows from their parent's timeout.
### WithCancel
```go
func WithCancel(ctx Context) (Context, context.CancelFunc)
```
`WithCancel` returns a copy of the DBOS context that can be manually canceled, along with a `CancelFunc`. Cancelling propagates to workflows and steps running under the returned context. Call the returned `CancelFunc` when the derived context is no longer needed to release its resources. `WithCancelCause` is a variant that returns a `context.CancelCauseFunc`, letting you supply an error describing why the context was canceled (retrievable with `context.Cause`).
## Context metadata
### GetApplicationVersion
```go
func GetApplicationVersion(ctx Context) string
```
`GetApplicationVersion` returns the application version for this context.
### GetExecutorID
```go
func GetExecutorID(ctx Context) string
```
`GetExecutorID` returns the executor ID for this context.
## Workflow Communication
### GetEvent
```go
func GetEvent[R any](ctx Client, targetWorkflowID, key string, timeout time.Duration) (R, error)
```
Retrieve the latest value of an event published by the workflow identified by `targetWorkflowID` to the key `key`.
If the event does not yet exist, wait for it to be published, returning an error if the wait times out.
**Parameters:**
- **ctx**: The DBOS client or context.
- **targetWorkflowID**: The identifier of the workflow whose events to retrieve.
- **key**: The key of the event to retrieve.
- **timeout**: A timeout. If the wait times out, return an error.
### SetEvent
```go
func SetEvent[P any](ctx Context, key string, message P, opts ...SetEventOption) error
```
Create and associate with this workflow an event with key `key` and value `value`.
If the event already exists, update its value.
Can only be called from within a workflow.
Use `WithPortableSetEvent()` for cross-language event consumption.
**Parameters:**
- **ctx**: The DBOS context.
- **key**: The key of the event.
- **message**: The value of the event. Must be serializable.
- **opts**: Optional `SetEventOption` functions (e.g., `WithPortableSetEvent()`).
### Send
```go
func Send[P any](ctx Client, destinationID string, message P, topic string, opts ...SendOption) error
```
Send a message to the workflow identified by `destinationID`.
Messages can optionally be associated with a topic.
Use `WithPortableSend()` for cross-language messaging.
**Parameters:**
- **ctx**: The DBOS client or context.
- **destinationID**: The workflow to which to send the message.
- **message**: The message to send. Must be serializable.
- **topic**: A topic with which to associate the message. Messages are enqueued per-topic on the receiver.
- **opts**: Optional `SendOption` functions (e.g., `WithPortableSend()`, `WithIdempotencyKey(key string)`).
Pass `WithIdempotencyKey(key string)` to make a `Send` deliver at most once: the key is combined with the destination workflow ID to form the message's primary key, so retrying a `Send` with the same key inserts the message only once.
### Recv
```go
func Recv[R any](ctx Context, topic string, timeout time.Duration) (R, error)
```
Receive and return a message sent to this workflow.
Can only be called from within a workflow.
Messages are dequeued first-in, first-out from a queue associated with the topic.
Calls to `Recv` wait for the next message in the queue, returning an error if the wait times out.
**Parameters:**
- **ctx**: The DBOS context.
- **topic**: A topic queue on which to wait.
- **timeout**: A timeout duration. If the wait times out, return an error.
## Streams
### WriteStream
```go
func WriteStream[P any](ctx Context, key string, value P, opts ...WriteStreamOption) error
```
Write a value to a durable stream.
May only be called from within a workflow or step.
Writes from a workflow are exactly-once; writes from a step are at-least-once.
Use `WithPortableWriteStream()` for cross-language stream reading.
### CloseStream
```go
func CloseStream(ctx Context, key string) error
```
Close a durable stream.
After closing, no more values can be written to the stream.
Streams are also automatically closed when the workflow terminates.
### ReadStream
```go
func ReadStream[R any](ctx Client, workflowID string, key string, opts ...ReadStreamOption) ([]R, bool, error)
```
Read all values from a durable stream.
Blocks until the stream is closed or the workflow becomes inactive.
Pass `WithReadStreamSnapshot()` to instead return immediately once all currently-available values have been drained; pair it with `WithReadStreamFromOffset(offset int)` to poll incrementally.
### ReadStreamAsync
```go
func ReadStreamAsync[R any](ctx Client, workflowID string, key string) (<-chan StreamValue[R], error)
```
Read values from a durable stream asynchronously.
Returns immediately with a channel that receives values as they are written to the stream.
### Sleep
```go
func Sleep(ctx Context, duration time.Duration) (time.Duration, error)
```
Sleep for the given duration.
May only be called from within a workflow.
This sleep is durable—it records its intended wake-up time in the database so if it is interrupted and recovers, it still wakes up at the intended time.
**Parameters:**
- **ctx**: The DBOS context.
- **duration**: The duration to sleep.
### RetrieveWorkflow
```go
func RetrieveWorkflow[R any](ctx Client, workflowID string) (WorkflowHandle[R], error)
```
Retrieve the handle of a workflow.
**Parameters**:
- **ctx**: The DBOS client or context.
- **workflowID**: The ID of the workflow whose handle to retrieve.
## Workflow Management Methods
### ListWorkflows
```go
func ListWorkflows(ctx Client, opts ...ListWorkflowsOption) ([]WorkflowStatus, error)
```
Retrieve a list of `WorkflowStatus` of all workflows matching specified criteria.
**Example usage:**
```go
// List all successful workflows from the last 24 hours
workflows, err := dbos.ListWorkflows(ctx,
dbos.WithFilterStatus(dbos.WorkflowStatusSuccess),
dbos.WithFilterCreatedAfter(time.Now().Add(-24*time.Hour)),
dbos.WithFilterLimit(100))
if err != nil {
log.Fatal(err)
}
// List workflows by specific IDs without loading input/output data
workflows, err := dbos.ListWorkflows(ctx,
dbos.WithFilterWorkflowIDs("workflow1", "workflow2"),
dbos.WithFilterLoadInput(false),
dbos.WithFilterLoadOutput(false))
if err != nil {
log.Fatal(err)
}
```
#### WithFilterAppVersion
```go
func WithFilterAppVersion(appVersion ...string) ListWorkflowsOption
```
Retrieve workflows tagged with this application version.
#### WithFilterCreatedBefore
```go
func WithFilterCreatedBefore(endTime time.Time) ListWorkflowsOption
```
Retrieve workflows started before this timestamp.
#### WithFilterLimit
```go
func WithFilterLimit(limit int) ListWorkflowsOption
```
Retrieve up to this many workflows.
#### WithFilterLoadInput
```go
func WithFilterLoadInput(loadInput bool) ListWorkflowsOption
```
WithFilterLoadInput controls whether to load workflow input data (default: true).
#### WithFilterLoadOutput
```go
func WithFilterLoadOutput(loadOutput bool) ListWorkflowsOption
```
WithFilterLoadOutput controls whether to load workflow output data (default: true).
#### WithFilterName
```go
func WithFilterName(names ...string) ListWorkflowsOption
```
Filter workflows by the specified workflow function name.
#### WithFilterOffset
```go
func WithFilterOffset(offset int) ListWorkflowsOption
```
Skip this many workflows from the results returned (for pagination).
#### WithFilterSortDesc
```go
func WithFilterSortDesc() ListWorkflowsOption
```
Sort the results in descending order by workflow start time (ascending is the default).
#### WithFilterCreatedAfter
```go
func WithFilterCreatedAfter(startTime time.Time) ListWorkflowsOption
```
Retrieve workflows started after this timestamp.
#### WithFilterStatus
```go
func WithFilterStatus(status ...WorkflowStatusType) ListWorkflowsOption
```
Filter workflows by status. Multiple statuses can be specified.
#### WithFilterUser
```go
func WithFilterUser(user ...string) ListWorkflowsOption
```
Filter workflows run by this authenticated user.
#### WithFilterWorkflowIDs
```go
func WithFilterWorkflowIDs(workflowIDs ...string) ListWorkflowsOption
```
Filter workflows by specific workflow IDs.
#### WithFilterWorkflowIDPrefix
```go
func WithFilterWorkflowIDPrefix(prefix ...string) ListWorkflowsOption
```
Filter workflows whose IDs start with the specified prefix.
#### WithFilterQueuesOnly
```go
func WithFilterQueuesOnly() ListWorkflowsOption
```
Return only workflows that are currently in a queue (queue name is not null, status is `ENQUEUED` or `PENDING`).
#### WithFilterCompletedAfter
```go
func WithFilterCompletedAfter(completedAfter time.Time) ListWorkflowsOption
```
Retrieve workflows that reached a terminal state (`SUCCESS`, `ERROR`, or `CANCELLED`) at or after this timestamp.
#### WithFilterCompletedBefore
```go
func WithFilterCompletedBefore(completedBefore time.Time) ListWorkflowsOption
```
Retrieve workflows that reached a terminal state (`SUCCESS`, `ERROR`, or `CANCELLED`) at or before this timestamp.
#### WithFilterDequeuedAfter
```go
func WithFilterDequeuedAfter(dequeuedAfter time.Time) ListWorkflowsOption
```
Retrieve workflows that started executing at or after this timestamp.
#### WithFilterDequeuedBefore
```go
func WithFilterDequeuedBefore(dequeuedBefore time.Time) ListWorkflowsOption
```
Retrieve workflows that started executing at or before this timestamp.
#### WithFilterWasForkedFrom
```go
func WithFilterWasForkedFrom(wasForkedFrom bool) ListWorkflowsOption
```
Filter workflows by whether they have been forked from (true) or not (false).
#### WithFilterHasParent
```go
func WithFilterHasParent(hasParent bool) ListWorkflowsOption
```
Filter workflows by whether they have a parent workflow (true) or not (false).
### GetWorkflowSteps
```go
func GetWorkflowSteps(ctx Client, workflowID string, opts ...GetWorkflowStepsOption) ([]StepInfo, error)
```
GetWorkflowSteps retrieves the execution steps of a workflow.
This is a list of `StepInfo` objects, with the following structure:
```go
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 // If the step starts or retrieves the result of a workflow, its ID
}
```
**Parameters:**
- **ctx**: The DBOS client or context.
- **workflowID**: The ID of the workflow whose steps to retrieve.
- **opts**: Optional configuration, documented below.
#### WithStepsLoadOutput
```go
func WithStepsLoadOutput(loadOutput bool) GetWorkflowStepsOption
```
Control whether to load step output data.
When unset, output is loaded only if the DBOS context has been launched.
#### WithStepsLimit
```go
func WithStepsLimit(limit int) GetWorkflowStepsOption
```
Limit the number of steps returned, ordered by step ID ascending.
#### WithStepsOffset
```go
func WithStepsOffset(offset int) GetWorkflowStepsOption
```
Skip the given number of steps before returning results. Combine with `WithStepsLimit` to paginate through a workflow's steps.
### CancelWorkflow
```go
func CancelWorkflow(ctx Client, workflowID string, opts ...CancelWorkflowOption) error
```
Cancel a workflow. This sets its status to `CANCELLED`, removes it from its queue (if it is enqueued) and preempts its execution (interrupting it at the beginning of its next step, or waking it immediately if it is in a durable sleep).
Pass `WithCancelChildren()` to also cancel all the workflow's child workflows, recursively.
To cancel many workflows in a single database round-trip, use `CancelWorkflows(ctx, workflowIDs []string, opts ...CancelWorkflowOption)`.
**Parameters:**
- **ctx**: The DBOS client or context.
- **workflowID**: The ID of the workflow to cancel.
- **opts**: Optional configuration (e.g., `WithCancelChildren()`).
### SetWorkflowAttributes
```go
func SetWorkflowAttributes(ctx Client, workflowID string, attributes map[string]any) error
```
Replace the custom attributes attached to an existing workflow. Pass a `nil` attributes map to clear all attributes.
Attach attributes at creation with the `WithWorkflowAttributes(map[string]any)` workflow option, and search workflows by attributes with the `WithFilterAttributes(map[string]any)` ListWorkflows option (Postgres only).
### ResumeWorkflow
```go
func ResumeWorkflow[R any](ctx Client, workflowID string, opts ...ResumeWorkflowOption) (WorkflowHandle[R], error)
```
Resume a workflow. This immediately starts it from its last completed step. You can use this to resume workflows that are cancelled or have exceeded their maximum recovery attempts. You can also use this to start an enqueued workflow immediately, bypassing its queue.
**Parameters:**
- **ctx**: The DBOS client or context.
- **workflowID**: The ID of the workflow to resume.
- **opts**: Optional configuration.
#### WithResumeQueue
```go
func WithResumeQueue(queueName string) ResumeWorkflowOption
```
Re-enqueue the resumed workflow on the specified queue instead of starting it immediately.
### ResumeWorkflows
```go
func ResumeWorkflows[R any](ctx Client, workflowIDs []string, opts ...ResumeWorkflowOption) ([]WorkflowHandle[R], error)
```
Resume multiple workflows in a single database round-trip.
Each workflow that exists and is not in a terminal state is re-enqueued; completed or missing workflows are skipped.
Unlike `ResumeWorkflow`, this function does not return an error when some IDs are missing.
Accepts the same options as `ResumeWorkflow` (e.g., `WithResumeQueue`).
### ForkWorkflow
```go
func ForkWorkflow[R any](ctx Client, input ForkWorkflowInput) (WorkflowHandle[R], error)
```
Start a new execution of a workflow from a specific step. The input step ID (`startStep`) must match the step number of the step returned by workflow introspection. The specified `startStep` is the step from which the new workflow will start, so any steps whose ID is less than `startStep` will not be re-executed.
**Parameters:**
- **ctx**: The DBOS client or context.
- **input**: A `ForkWorkflowInput` struct where `OriginalWorkflowID` is mandatory.
```go
type ForkWorkflowInput struct {
OriginalWorkflowID string // Required: The UUID of the original workflow to fork from
ForkedWorkflowID string // Optional: Custom workflow ID for the forked workflow (auto-generated if empty)
StartStep uint // Optional: Step to start the forked workflow from (default: 0)
ApplicationVersion string // Optional: Application version for the forked workflow (inherits from original if empty)
QueueName string // Optional: Queue to enqueue the forked workflow on (defaults to starting immediately)
}
```
### SetWorkflowDelay
```go
func SetWorkflowDelay(ctx Client, workflowID string, opts ...SetWorkflowDelayOption) error
```
Set or update the delay on a `DELAYED` workflow.
Provide exactly one of `WithDelayDuration` (relative) or `WithDelayUntil` (absolute).
Only affects workflows currently in the `DELAYED` status.
```go
func WithDelayDuration(d time.Duration) SetWorkflowDelayOption
func WithDelayUntil(t time.Time) SetWorkflowDelayOption
```
### Workflow Status
Some workflow introspection and management methods return a `WorkflowStatus`.
This object has the following definition:
```go
type WorkflowStatus struct {
ID string `json:"workflow_uuid"` // Unique identifier for the workflow
Status WorkflowStatusType `json:"status"` // Current execution status
Name string `json:"name"` // Function name of the workflow
AuthenticatedUser *string `json:"authenticated_user"` // User who initiated the workflow (if applicable)
AssumedRole *string `json:"assumed_role"` // Role assumed during execution (if applicable)
AuthenticatedRoles *string `json:"authenticated_roles"` // Roles available to the user (if applicable)
Output any `json:"output"` // Workflow output (available after completion)
Error error `json:"error"` // Error information (if status is ERROR)
ExecutorID string `json:"executor_id"` // ID of the executor running this workflow
CreatedAt time.Time `json:"created_at"` // When the workflow was created
UpdatedAt time.Time `json:"updated_at"` // When the workflow status was last updated
ApplicationVersion string `json:"application_version"` // Version of the application that created this workflow
ApplicationID string `json:"application_id"` // Application identifier
Attempts int `json:"attempts"` // Number of execution attempts
QueueName string `json:"queue_name"` // Queue name (if workflow was enqueued)
Timeout time.Duration `json:"timeout"` // Workflow timeout duration
Deadline time.Time `json:"deadline"` // Absolute deadline for workflow completion
StartedAt time.Time `json:"started_at"` // When the workflow execution actually started
CompletedAt time.Time `json:"completed_at"` // When the workflow reached a terminal state (SUCCESS, ERROR, or CANCELLED)
ForkedFrom string `json:"forked_from"` // ID of the original workflow if this is a fork
WasForkedFrom bool `json:"was_forked_from"` // Whether this workflow has been forked from
ParentWorkflowID string `json:"parent_workflow_id"` // ID of the parent workflow if this is a child
DeduplicationID string `json:"deduplication_id"` // Deduplication identifier (if applicable)
Input any `json:"input"` // Input parameters passed to the workflow
Priority int `json:"priority"` // Execution priority (lower numbers have higher priority)
DelayUntil time.Time `json:"delay_until"` // Time before which a DELAYED workflow should not be dequeued
Attributes map[string]any `json:"attributes"` // Custom key-value attributes attached to the workflow
}
```
#### WorkflowStatusType
The `WorkflowStatusType` represents the execution status of a workflow:
```go
type WorkflowStatusType string
const (
WorkflowStatusPending WorkflowStatusType = "PENDING" // Workflow is running or ready to run
WorkflowStatusEnqueued WorkflowStatusType = "ENQUEUED" // Workflow is queued and waiting for execution
WorkflowStatusDelayed WorkflowStatusType = "DELAYED" // Workflow is delayed and will transition to ENQUEUED after the delay expires
WorkflowStatusSuccess WorkflowStatusType = "SUCCESS" // Workflow completed successfully
WorkflowStatusError WorkflowStatusType = "ERROR" // Workflow completed with an error
WorkflowStatusCancelled WorkflowStatusType = "CANCELLED" // Workflow was cancelled (manually or due to timeout)
WorkflowStatusMaxRecoveryAttemptsExceeded WorkflowStatusType = "MAX_RECOVERY_ATTEMPTS_EXCEEDED" // Workflow exceeded maximum retry attempts
)
```
## DBOS Variables
### GetWorkflowID
```go
func GetWorkflowID(ctx Context) (string, error)
```
Return the ID of the current workflow, if in a workflow. Returns an error if not called from within a workflow context.
**Parameters:**
- **ctx**: The DBOS context.
### GetStepID
```go
func GetStepID(ctx Context) (int, error)
```
Return the unique ID of the current step within a workflow. Returns an error if not called from within a step context.
**Parameters:**
- **ctx**: The DBOS context.
Workflow queues allow you to ensure that workflow functions will be run, without starting them immediately.
Queues are useful for controlling the number of workflows run in parallel, or the rate at which they are started.
Queue configuration is persisted to the system database, so any DBOS process connected to the same system database can register, retrieve, and reconfigure queues.
### RegisterQueue
```go
func RegisterQueue(ctx Client, name string, options ...QueueOption) (Queue, error)
```
Register a queue and persist its configuration to the system database, returning a `Queue`.
If a queue with the same name already exists in the database, the `WithQueueOnConflict` option controls whether its configuration is overwritten.
Queues may be registered at any time, including after `Launch()`; live workers periodically reload queue configuration, so changes take effect without a restart.
You can enqueue a workflow by passing the returned `Queue` handle to the `WithQueue` option of `RunWorkflow`.
**Parameters:**
- **ctx**: The DBOS client or context.
- **name**: The name of the queue. Must be unique among all queues in the application.
- **options**: Functional options for the queue, documented below.
**Example Syntax:**
```go
queue, err := dbos.RegisterQueue(ctx, "email-queue",
dbos.WithWorkerConcurrency(5),
dbos.WithRateLimiter(&dbos.RateLimiter{
Limit: 100,
Period: 60 * time.Second, // 100 workflows per minute
}),
dbos.WithPriorityEnabled(),
)
// Enqueue workflows to this queue by passing its handle to WithQueue:
handle, err := dbos.RunWorkflow(ctx, SendEmailWorkflow, emailData, dbos.WithQueue(queue))
```
The returned `Queue` interface has `Get*` methods reflecting the queue's configuration as of the most recent read from the database, and `Set*` methods that update the configuration in the database at runtime:
```go
type Queue interface {
GetName() string
GetGlobalConcurrency() *int
GetWorkerConcurrency() *int
GetRateLimit() *RateLimiter
GetPriorityEnabled() bool
GetPartitionQueue() bool
GetPollingInterval() time.Duration
SetGlobalConcurrency(ctx Client, value *int) error
SetWorkerConcurrency(ctx Client, value *int) error
SetRateLimit(ctx Client, value *RateLimiter) error
SetPriorityEnabled(ctx Client, value bool) error
SetPartitionQueue(ctx Client, value bool) error
SetPollingInterval(ctx Client, value time.Duration) error
}
```
#### WithQueueOnConflict
```go
func WithQueueOnConflict(policy QueueConflictResolution) QueueOption
const (
QueueConflictUpdateIfLatestVersion QueueConflictResolution = "update_if_latest_version"
QueueConflictAlwaysUpdate QueueConflictResolution = "always_update"
QueueConflictNeverUpdate QueueConflictResolution = "never_update"
)
```
Set how `RegisterQueue` behaves when a queue with the same name already exists in the system database:
- **QueueConflictUpdateIfLatestVersion** (default): overwrite the existing configuration only if the running application is the latest registered application version.
- **QueueConflictAlwaysUpdate**: always overwrite the existing configuration.
- **QueueConflictNeverUpdate**: leave the existing configuration unchanged.
### RetrieveQueue
```go
func RetrieveQueue(ctx Client, name string) (Queue, error)
```
Retrieve a queue by name from the system database. If no queue with that name has been registered, returns an error matching `dbos.ErrQueueNotFound`.
### ListQueues
```go
func ListQueues(ctx Client) ([]Queue, error)
```
Return all queues registered in the system database.
### DeleteQueue
```go
func DeleteQueue(ctx Client, name string) error
```
Delete a queue from the system database. No-op if no queue with that name exists.
Workflows already enqueued on a deleted queue can no longer be dequeued, executed, or recovered — unless a queue with the same name is later registered, in which case it will dequeue the leftover workflows.
Do not rely on this behavior: cancel or drain pending workflows on the queue before deleting it.
#### WithWorkerConcurrency
```go
func WithWorkerConcurrency(concurrency int) QueueOption
```
Set the maximum number of workflows from this queue that may run concurrently within a single DBOS process.
#### WithGlobalConcurrency
```go
func WithGlobalConcurrency(concurrency int) QueueOption
```
Set the maximum number of workflows from this queue that may run concurrently. Defaults to 0 (no limit).
This concurrency limit is global across all DBOS processes using this queue.
#### WithPriorityEnabled
```go
func WithPriorityEnabled() QueueOption
```
Enable setting priority for workflows on this queue.
#### WithRateLimiter
```go
func WithRateLimiter(limiter *RateLimiter) QueueOption
```
```go
type RateLimiter struct {
Limit int // Maximum number of workflows to start within the period
Period time.Duration // Time period for the rate limit
}
```
A limit on the maximum number of functions which may be started in a given period.
#### WithPartitionQueue
```go
func WithPartitionQueue() QueueOption
```
Enable partitioning for this queue.
When enabled, workflows can be enqueued with a partition key using `WithQueuePartitionKey`, and each partition has its own concurrency limits.
### RegisterWorkflow
```go
func RegisterWorkflow[P any, R any](ctx Context, fn Workflow[P, R], opts ...WorkflowRegistrationOption)
```
Register a function as a DBOS workflow.
All workflows must be registered before the context is launched.
Workflow functions must be compatible with the following signature:
```go
type Workflow[P any, R any] func(ctx Context, input P) (R, error)
```
**Parameters:**
- **ctx**: The Context.
- **fn**: The workflow function to register.
- **opts**: Functional options for workflow registration, documented below.
#### WithMaxRecoveryAttempts
```go
func WithMaxRecoveryAttempts(maxRetries int) WorkflowRegistrationOption
```
Configure the maximum number of times execution of a workflow may be attempted.
This acts as a dead letter queue so that a buggy workflow that crashes its application (for example, by running it out of memory) does not do so infinitely.
If a workflow exceeds this limit, its status is set to `MAX_RECOVERY_ATTEMPTS_EXCEEDED` and it may no longer be executed.
#### WithWorkflowName
```go
func WithWorkflowName(name string) WorkflowRegistrationOption
```
Register a workflow with a custom name.
If not provided, the name of the workflow function is used.
#### WithInstance
```go
func WithInstance(instance ConfiguredInstance) WorkflowRegistrationOption
```
Register a workflow method bound to a specific configured instance.
Method values bound to different receivers (e.g. `a.Run` and `b.Run`) share a function name, so each instance's method must be registered under a per-instance key, derived from the instance's config name.
The instance must implement the `ConfiguredInstance` interface:
```go
type ConfiguredInstance interface {
ConfigName() string
}
```
`ConfigName` must return a stable, unique name for the instance: it is durably recorded so recovery runs the workflow on the correct instance.
Instances must be registered with the same config name on every process start, before `Launch()`.
```go
dbos.RegisterWorkflow(ctx, slack.Send, dbos.WithInstance(slack))
dbos.RegisterWorkflow(ctx, email.Send, dbos.WithInstance(email))
```
Run a workflow registered with `WithInstance` using the matching `WithRunInstance` option.
### RunWorkflow
```go
func RunWorkflow[P any, R any](ctx Context, fn Workflow[P, R], input P, opts ...WorkflowOption) (WorkflowHandle[R], error)
```
Execute a workflow function.
The workflow may execute immediately or be enqueued for later execution based on options.
Returns a WorkflowHandle that can be used to check the workflow's status or wait for its completion and retrieve its results.
**Parameters:**
- **ctx**: The Context.
- **fn**: The workflow function to execute.
- **input** The input to the workflow function.
- **opts**: Functional options for workflow execution, documented below.
**Example Syntax**:
```go
func workflow(ctx dbos.Context, input string) (string, error) {
return "success", nil
}
func example(input string) error {
handle, err := dbos.RunWorkflow(dbosContext, workflow, input)
if err != nil {
return err
}
result, err := handle.GetResult()
if err != nil {
return err
}
fmt.Println("Workflow result:", result)
return nil
}
```
#### WithWorkflowID
```go
func WithWorkflowID(id string) WorkflowOption
```
Run the workflow with a custom workflow ID.
If not specified, a UUID workflow ID is generated.
#### WithRunInstance
```go
func WithRunInstance(instance ConfiguredInstance) WorkflowOption
```
Run a workflow method registered with `WithInstance`.
The instance's config name selects the per-instance registration, so the workflow executes on (and recovers to) the correct instance.
```go
handle, err := dbos.RunWorkflow(ctx, slack.Send, input, dbos.WithRunInstance(slack))
```
#### WithQueue
```go
func WithQueue(queue Queue) WorkflowOption
```
Enqueue the workflow to the given queue instead of executing it immediately.
Queued workflows will be dequeued and executed according to the queue's configuration.
The queue must be a non-nil `Queue` handle returned by `RegisterQueue`, `RetrieveQueue`, or `ListQueues`; passing `nil` makes the enclosing `RunWorkflow` call return an error.
To enqueue by name instead (for example, from a standalone client), use `Enqueue`.
#### WithDeduplicationID
```go
func WithDeduplicationID(id string) WorkflowOption
```
Set a deduplication ID for this workflow.
Should be used alongside `WithQueue`.
At any given time, only one workflow with a specific deduplication ID can be enqueued in a given queue.
#### WithDeduplicationPolicy
```go
func WithDeduplicationPolicy(policy DeduplicationPolicy) WorkflowOption
```
Set how a colliding deduplication ID is handled for a queued workflow.
Must be used alongside `WithQueue` and `WithDeduplicationID`.
With the default `DeduplicationPolicyReject`, a colliding enqueue fails with a `ErrorCodeQueueDeduplicated` error; with `DeduplicationPolicyReturnExisting`, it instead returns a handle to the existing workflow.
#### WithPriority
```go
func WithPriority(priority uint) WorkflowOption
```
Set a queue priority for the workflow.
Should be used alongside `WithQueue`.
Workflows with the same priority are dequeued in **FIFO (first in, first out)** order.
Priority values can range from `1` to `2,147,483,647`, where **a low number indicates a higher priority**.
Workflows without assigned priorities have the highest priority and are dequeued before workflows with assigned priorities.
#### WithQueuePartitionKey
```go
func WithQueuePartitionKey(partitionKey string) WorkflowOption
```
Set a queue partition key for the workflow.
Use if and only if the queue is partitioned (created with `WithPartitionQueue`).
#### WithDelay
```go
func WithDelay(delay time.Duration) WorkflowOption
```
Delay execution of a queued workflow by the specified duration.
Must be used together with `WithQueue`.
The workflow is initially placed in `DELAYED` status and does not execute until the delay expires, at which point it transitions to `ENQUEUED` and may be dequeued.
The delay can later be updated via `SetWorkflowDelay`.
#### WithApplicationVersion
```go
func WithApplicationVersion(version string) WorkflowOption
```
Set the application version for this workflow, overriding the version in Context.
#### WithAuthenticatedUser
```go
func WithAuthenticatedUser(user string) WorkflowOption
```
Associate the workflow execution with a user name. Useful to define workflow identity.
Child workflows automatically inherit their parent's authentication information (authenticated user, assumed role, and authenticated roles) unless explicitly overridden.
#### WithWorkflowAttributes
```go
func WithWorkflowAttributes(attributes map[string]any) WorkflowOption
```
Attach custom key-value attributes to the workflow.
Attributes are recorded in the workflow status at creation, must be JSON-serializable, and are not inherited by child workflows.
On Postgres they can be searched with the `WithFilterAttributes(map[string]any)` ListWorkflows option, and replaced later with `SetWorkflowAttributes`.
### RunAsStep
```go
func RunAsStep[R any](ctx Context, fn Step[R], opts ...StepOption) (R, error)
```
Execute a function as a step in a durable workflow.
**Parameters:**
- **ctx**: The Context.
- **fn**: The step to execute, typically wrapped in an anonymous function. Syntax shown below.
- **opts**: Functional options for step execution, documented below.
**Example Syntax:**
Any Go function can be a step as long as it outputs one json-encodable value and an error.
To pass inputs into a function being called as a step, wrap it in an anonymous function as shown below:
```go
func step(ctx context.Context, input string) (string, error) {
output := ...
return output
}
func workflow(ctx dbos.Context, input string) (string, error) {
output, err := dbos.RunAsStep(
ctx,
func(stepCtx context.Context) (string, error) {
return step(stepCtx, input)
}
)
}
```
#### WithStepName
```go
func WithStepName(name string) StepOption
```
Set a custom name for a step.
#### WithStepMaxRetries
```go
func WithStepMaxRetries(maxRetries int) StepOption
```
Set the maximum number of times this step is automatically retired on failure.
A value of 0 (the default) indicates no retries.
#### WithStepMaxInterval
```go
func WithStepMaxInterval(interval time.Duration) StepOption
```
WithStepMaxInterval sets the maximum delay between retries. Default value is 5s.
#### WithStepBackoffFactor
```go
func WithStepBackoffFactor(factor float64) StepOption
```
WithStepBackoffFactor sets the exponential backoff multiplier between retries. Default value is 2.0.
#### WithStepBaseInterval
```go
func WithStepBaseInterval(interval time.Duration) StepOption
```
WithStepBaseInterval sets the initial delay between retries. Default value is 100ms.
### Go
```go
func Go[R any](ctx Context, fn Step[R], opts ...StepOption) (<-chan StepOutcome[R], error)
```
Launch a step asynchronously and return a receive-only channel that will receive the result when the step completes.
This is a durable alternative to Go's native goroutines. Can only be called from within a workflow (not from inside a step).
```go
type StepOutcome[R any] struct {
Result R
Err error
}
```
### Select
```go
func Select[R any](ctx Context, channels []<-chan StepOutcome[R]) (R, error)
```
Wait for and return the first result from multiple channels obtained from `Go`.
This is a durable alternative to Go's native `select` statement. Can only be called from within a workflow.
### WorkflowHandle
```go
type WorkflowHandle[R any] interface {
GetResult(opts ...GetResultOption) (R, error)
GetStatus() (WorkflowStatus, error)
GetWorkflowID() string
}
```
WorkflowHandle provides methods to interact with a running or completed workflow.
The type parameter `R` represents the expected return type of the workflow.
Handles can be used to wait for workflow completion, check status, and retrieve results.
#### WorkflowHandle.GetResult
```go
WorkflowHandle.GetResult(opts ...GetResultOption) (R, error)
```
Wait for the workflow to complete and return its result.
#### WorkflowHandle.GetStatus
```go
WorkflowHandle.GetStatus() (WorkflowStatus, error)
```
Retrieve the WorkflowStatus of the workflow.
#### WorkflowHandle.GetWorkflowID
```go
WorkflowHandle.GetWorkflowID() string
```
Retrieve the ID of the workflow.
`Client` provides a programmatic way to interact with your DBOS application from external code.
Because every `Context` **is** a `Client` (the `Context` interface embeds `Client`), all the package-level functions documented above whose first parameter is a `Client` work identically with a standalone client.
Use them by passing your client where you would pass a DBOS context; only functions requiring a `Context` (workflow registration and execution, workflow-scoped operations) are unavailable on a standalone client.
This is the `Client` interface:
```go
type Client interface {
context.Context
// Workflow operations
Enqueue(_ Client, queueName string, workflowName string, input any, opts ...EnqueueOption) (WorkflowHandle[any], error)
Send(_ Client, destinationID string, message any, topic string, opts ...SendOption) error
GetEvent(_ Client, targetWorkflowID string, key string, timeout time.Duration) (any, error)
ReadStream(_ Client, workflowID string, key string, opts ...ReadStreamOption) ([]any, bool, error)
ReadStreamAsync(_ Client, workflowID string, key string) (<-chan StreamValue[any], error)
// Workflow management
RetrieveWorkflow(_ Client, workflowID string) (WorkflowHandle[any], error)
CancelWorkflow(_ Client, workflowID string, opts ...CancelWorkflowOption) error
CancelWorkflows(_ Client, workflowIDs []string, opts ...CancelWorkflowOption) error
SetWorkflowAttributes(_ Client, workflowID string, attributes map[string]any) error
SetWorkflowDelay(_ Client, workflowID string, opts ...SetWorkflowDelayOption) error
ResumeWorkflow(_ Client, workflowID string, opts ...ResumeWorkflowOption) (WorkflowHandle[any], error)
ResumeWorkflows(_ Client, workflowIDs []string, opts ...ResumeWorkflowOption) ([]WorkflowHandle[any], error)
ForkWorkflow(_ Client, input ForkWorkflowInput) (WorkflowHandle[any], error)
ForkWorkflows(_ Client, input ForkWorkflowsInput) ([]WorkflowHandle[any], error)
ListWorkflows(_ Client, opts ...ListWorkflowsOption) ([]WorkflowStatus, error)
GetWorkflowSteps(_ Client, workflowID string, opts ...GetWorkflowStepsOption) ([]StepInfo, error)
GetWorkflowAggregates(_ Client, input GetWorkflowAggregatesInput) ([]WorkflowAggregateRow, error)
GetStepAggregates(_ Client, input GetStepAggregatesInput) ([]StepAggregateRow, error)
DeleteWorkflows(_ Client, workflowIDs []string, opts ...DeleteWorkflowOption) error
// Queue management
RegisterQueue(_ Client, name string, options ...QueueOption) (Queue, error)
RetrieveQueue(_ Client, name string) (Queue, error)
ListQueues(_ Client) ([]Queue, error)
DeleteQueue(_ Client, name string) error
// Schedule management
CreateSchedule(_ Client, spec ScheduleSpec) error
ApplySchedules(_ Client, schedules []ScheduleSpec) error
PauseSchedule(_ Client, scheduleName string) error
ResumeSchedule(_ Client, scheduleName string) error
DeleteSchedule(_ Client, scheduleName string) error
GetSchedule(_ Client, scheduleName string) (WorkflowSchedule, error)
ListSchedules(_ Client, opts ...ListSchedulesOption) ([]WorkflowSchedule, error)
BackfillSchedule(_ Client, scheduleName string, start, end time.Time) ([]string, error)
TriggerSchedule(_ Client, scheduleName string) (WorkflowHandle[any], error)
// Application version management
ListApplicationVersions(_ Client) ([]VersionInfo, error)
GetLatestApplicationVersion(_ Client) (VersionInfo, error)
SetLatestApplicationVersion(_ Client, versionName string) error
Shutdown(_ Client, timeout time.Duration) error
}
```
Prefer the generic package-level functions over calling interface methods directly: they return typed handles or values (`RetrieveWorkflow[R]`, `ResumeWorkflow[R]`, `ResumeWorkflows[R]`, `ForkWorkflow[R]`, `TriggerSchedule[R]`, `GetEvent[R]`, `ReadStream[R]`, `ReadStreamAsync[R]`, `Enqueue[R]`), while the interface methods return `any`.
### Constructor
```go
func NewClient(ctx context.Context, config ClientConfig) (Client, error)
```
**Parameters:**
- `ctx`: A context for initialization operations
- `config`: A `ClientConfig` object with connection and application settings
```go
type ClientConfig struct {
DatabaseURL string // Connection string to your system database. May be a PostgreSQL (postgres://...) or SQLite (sqlite:...) URL. Exactly one of DatabaseURL, SystemDBPool, or SQLiteSystemDB is required. SQLite URLs additionally require the driver import: import _ "github.com/dbos-inc/dbos-transact-golang/dbos/driver/sqlite"
SystemDBPool *pgxpool.Pool // A custom Postgres/CockroachDB pool. Optional; takes precedence over DatabaseURL. Mutually exclusive with SQLiteSystemDB.
SQLiteSystemDB *sql.DB // A custom SQLite handle (e.g. from modernc.org/sqlite). Optional; takes precedence over DatabaseURL. Mutually exclusive with SystemDBPool.
DatabaseSchema string // Database schema name (defaults to "dbos"; Postgres only)
Logger *slog.Logger // Optional custom logger
Serializer Serializer[any] // Optional custom serializer (defaults to JSON)
SystemDBStartupTimeout time.Duration // Maximum time for system database connection and migrations (default: 2 minutes)
}
```
**Returns:**
- A new `Client` instance or an error if initialization fails
**Example syntax:**
This DBOS client connects to the system database specified in the configuration:
```go
config := dbos.ClientConfig{
DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
}
client, err := dbos.NewClient(context.Background(), config)
if err != nil {
log.Fatal(err)
}
defer dbos.Shutdown(client, 5*time.Second)
```
A client manages a connection pool to the DBOS system database. Shut it down with the same unified `dbos.Shutdown(client, timeout)` function documented above, which releases the connection pool.
## Workflow Interaction Methods
### Enqueue
```go
func Enqueue[R any, P any](
ctx Client,
queueName string,
workflowName string,
input P,
opts ...EnqueueOption
) (WorkflowHandle[R], error)
```
The result type parameter `R` comes first so you can name only it and let the input type be inferred: `dbos.Enqueue[MyOutput](client, ...)`.
Enqueue a workflow for processing and return a handle to it, similar to RunWorkflow with the WithQueue option.
Returns a WorkflowHandle.
When enqueuing a workflow from the DBOS client, you must specify the name of the workflow to enqueue (rather than passing a workflow function as with `RunWorkflow`.)
Required parameters:
* `ctx`: The DBOS client (or context)
* `queueName`: The name of the queue on which to enqueue the workflow
* `workflowName`: The name of the workflow function being enqueued
* `input`: The input to pass to the workflow
Optional configuration via `EnqueueOption`:
* `WithEnqueueWorkflowID(id string)`: The unique ID for the enqueued workflow.
If left undefined, DBOS Client will generate a UUID.
Please see Workflow IDs and Idempotency for more information.
* `WithEnqueueApplicationVersion(version string)`: The version of your application that should process this workflow.
If left undefined, it will use the current application version.
* `WithEnqueueTimeout(timeout time.Duration)`: Set a timeout for the enqueued workflow. When the timeout expires, the workflow **and all its children** are cancelled (except if the child's context has been made uncancellable using `WithoutCancel`). The timeout does not begin until the workflow is dequeued and starts execution.
* `WithEnqueueDeduplicationID(id string)`: At any given time, only one workflow with a specific deduplication ID can be enqueued in the specified queue. If a workflow with a deduplication ID is currently enqueued or actively executing (status `ENQUEUED` or `PENDING`), subsequent workflow enqueue attempts with the same deduplication ID in the same queue will fail.
* `WithEnqueueDeduplicationPolicy(policy DeduplicationPolicy)`: Set how a colliding deduplication ID is handled. Requires `WithEnqueueDeduplicationID`. With the default `DeduplicationPolicyReject`, a colliding enqueue fails with a `ErrorCodeQueueDeduplicated` error; with `DeduplicationPolicyReturnExisting`, it instead returns a handle to the existing workflow.
* `WithEnqueuePriority(priority uint)`: The priority of the enqueued workflow in the specified queue. Workflows with the same priority are dequeued in **FIFO (first in, first out)** order. Priority values can range from `1` to `2,147,483,647`, where **a low number indicates a higher priority**. Workflows without assigned priorities have the highest priority and are dequeued before workflows with assigned priorities.
* `WithEnqueueDelay(delay time.Duration)`: Delay execution of the enqueued workflow by the specified duration. The workflow is initially placed in `DELAYED` status and transitions to `ENQUEUED` after the delay expires. The delay can later be updated via `SetWorkflowDelay`.
* `WithEnqueueQueuePartitionKey(partitionKey string)`: Set the queue partition key. Required if and only if the target queue is partitioned (created with `WithPartitionQueue`).
**Example syntax:**
```go
type ProcessInput struct {
TaskID string
Data string
}
type ProcessOutput struct {
Result string
Status string
}
handle, err := dbos.Enqueue[ProcessOutput](
client,
"process_queue",
"ProcessWorkflow",
ProcessInput{TaskID: "task-123", Data: "data"},
dbos.WithEnqueueTimeout(30 * time.Minute),
dbos.WithEnqueuePriority(5),
)
if err != nil {
log.Fatal(err)
}
result, err := handle.GetResult()
if err != nil {
log.Printf("Workflow failed: %v", err)
} else {
log.Printf("Result: %+v", result)
}
```
All other workflow interaction, management, streaming, queue, and schedule functions are the same package-level functions documented above (`Send`, `GetEvent`, `RetrieveWorkflow`, `ListWorkflows`, `GetWorkflowSteps`, `CancelWorkflow`/`CancelWorkflows`, `ResumeWorkflow`/`ResumeWorkflows`, `ForkWorkflow`/`ForkWorkflows`, `SetWorkflowAttributes`, `SetWorkflowDelay`, `DeleteWorkflows`, `ReadStream`, `ReadStreamAsync`, `RegisterQueue`, `RetrieveQueue`, `ListQueues`, `DeleteQueue`, `CreateSchedule`, `ApplySchedules`, and the rest): they take `ctx Client` as their first parameter, so pass your client where a DBOS application would pass its context.
Note that with `ListWorkflows` and `GetWorkflowSteps` from a standalone client, workflow inputs, outputs, and step outputs are not loaded or decoded by default; pass `WithFilterLoadInput(true)` / `WithFilterLoadOutput(true)` / `WithStepsLoadOutput(true)` to opt in.
### NewDebouncerClient
```go
func NewDebouncerClient[R any, P any](workflowName string, client Client, opts ...DebouncerOption) *DebouncerClient[R, P]
```
Both type parameters must be named explicitly (the workflow is referenced by name, so neither can be inferred): `dbos.NewDebouncerClient[MyOutput, MyInput]("workflowName", client)`.
Create a new debouncer client for use from outside a DBOS application.
Similar to `NewDebouncer` but uses a Client instead of a Context and takes a workflow name string instead of a function reference.
To debounce a workflow registered on a configured instance, pass the instance's config name with `WithDebouncerConfigName(configName string)`.
## Cross-Language Portable Types
### WithPortableWorkflow
```go
func WithPortableWorkflow() WorkflowOption
```
Mark a workflow to use portable JSON serialization for cross-language interoperability.
### WithPortableSend / WithPortableSetEvent / WithPortableWriteStream
```go
func WithPortableSend() SendOption
func WithPortableSetEvent() SetEventOption
func WithPortableWriteStream() WriteStreamOption
```
Use portable JSON for cross-language Send, SetEvent, and WriteStream operations.
### PortableWorkflowArgs
```go
type PortableWorkflowArgs struct {
PositionalArgs []any `json:"positional_args,omitempty"`
NamedArgs map[string]any `json:"named_args,omitempty"`
}
```
Cross-language envelope for workflow inputs. When passed as the input to `Enqueue`, portable JSON serialization is used automatically.
### PortableWorkflowError
```go
type PortableWorkflowError struct {
Name string
Message string
Code any
Data any
}
```
Structured error type for portable workflows. Return this from a workflow to pass structured error info cross-language.
### WithEnqueueClassName / WithEnqueueConfigName
```go
func WithEnqueueClassName(className string) EnqueueOption
func WithEnqueueConfigName(configName string) EnqueueOption
```
Set the class/namespace and config/instance name when enqueueing to Python, TypeScript, or Java targets.
`WithEnqueueConfigName` is also required when enqueueing to a Go workflow registered on a configured instance with `WithInstance`; the value must match the instance's config name.
## Alerting
### SetAlertHandler
```go
func SetAlertHandler(ctx Context, handler AlertHandler)
```
```go
type AlertHandler func(name string, message string, metadata map[string]string)
```
Register a handler to receive alerts from Conductor.
Must be called before `Launch()`. Only one handler per application.
If no handler is registered, alerts are logged automatically.