Transactions & Datasources

A datasource is a handle to a database you own, over which DBOS can run durable transactions. When you run a transaction through a datasource inside a workflow, your application writes and the DBOS durability record commit atomically in your database, so the transaction executes exactly once even if your program crashes and the workflow is recovered.

This is a stronger guarantee than a step provides: a step that writes to a database may re-execute (and re-commit) if the process crashes after the write but before the step is checkpointed.

Creating a Datasource

Create a datasource with NewDataSource, passing a connection pool to your database: a *pgxpool.Pool for Postgres or CockroachDB, or a *sql.DB for SQLite.

import "github.com/jackc/pgx/v5/pgxpool"

pool, err := pgxpool.New(context.Background(), os.Getenv("APP_DATABASE_URL"))
if err != nil {
    log.Fatal(err)
}
ds, err := dbos.NewDataSource(dbosContext, pool, dbos.WithDataSourceName("app"))
if err != nil {
    log.Fatal(err)
}

NewDataSource may be called at any time, before or after Launch(). It provisions a transaction_completion durability table in your database (in the dbos schema by default, configurable with WithDataSourceSchema) unless the table already exists. If you prefer to manage DDL yourself, pre-create the table in your own migrations and connect with a role that only needs SELECT, INSERT on it.

If you pass the same engine handle you gave DBOS as its system database (via Config.SystemDBPool or Config.SqliteSystemDB), no transaction_completion table is created or managed at all: your application writes and the DBOS checkpoint commit together in a single transaction. See Sharing the System Database Engine.

Running Transactions

Inside a workflow, run a transaction with RunAsTransaction. Your function receives a Tx on which to run queries; DBOS commits the transaction if the function returns successfully and rolls it back if it returns an error.

func checkoutWorkflow(ctx dbos.DBOSContext, item string) (int64, error) {
    // This transaction runs exactly once, even across crashes and recovery.
    orderID, err := dbos.RunAsTransaction(ctx, ds, func(txCtx context.Context, tx dbos.Tx) (int64, error) {
        var id int64
        err := tx.QueryRow(txCtx, "INSERT INTO orders(item) VALUES ($1) RETURNING id", item).Scan(&id)
        return id, err
    })
    if err != nil {
        return 0, err
    }

    // Transactions and steps share the workflow's step counter and can be interleaved.
    _, err = dbos.RunAsStep(ctx, func(stepCtx context.Context) (string, error) {
        return sendConfirmation(stepCtx, orderID)
    })
    return orderID, err
}

RunAsTransaction must be called from within a workflow. It accepts the same options as RunAsStep (WithStepName, WithStepMaxRetries, and so on). Serialization and deadlock conflicts are retried automatically with a fresh transaction; errors returned by your function follow your step retry policy.

tip

If your application tables live in the same database that hosts the DBOS system schema (for example, the shared-engine setup), you can atomically write application data and enqueue a workflow in one transaction by calling the dbos.enqueue_workflow PL/pgSQL function from within it.

Guarantees

• A RunAsTransaction called at the top level of a workflow is exactly-once: the application writes and the durability record commit atomically, and recovery replays the recorded output instead of re-running the function.
• A RunAsTransaction nested inside a RunAsStep is allowed but downgraded to the step's at-least-once guarantee, because the enclosing step owns the durability boundary.
• Nesting a RunAsTransaction inside another RunAsTransaction is rejected with an error.

See the datasources reference for details on durability, recovery, and permissions.