Durable transactions on a separate application database - #153
Conversation
…tabase All three reference SDKs support durable transactions against an application database of the user's own, while durare only offered the single-database transactional step. This closes that gap with ctx.transaction_on(&ds, name, |conn| ...) over a PgDataSource or SqliteDataSource built on the user's sqlx pool. A single commit cannot span two databases, so the call uses a two-commit protocol: the body's writes and a transaction_completion witness row commit atomically on the application database, then the ordinary step checkpoint commits to the system database. Recovery replays in layers - checkpoint first, then the witness row (covering a crash between the two commits) - and runs the body only when neither exists, so the body runs exactly once. Permanent failures are mirrored into the witness table before the system record, keeping the recovery order; conflicts and transient errors retry on fresh transactions without consuming the application retry budget, the same two-loop structure as the single-database step. The body receives the backend's native sqlx connection rather than a wrapper, so existing queries, sqlx macros, and data-access helpers work unchanged; the DataSource trait is sealed (the backend set is closed, as in every DBOS SDK). Since commit/rollback stay with the engine, a plain connection cannot end the transaction - and on Postgres a raw COMMIT or ROLLBACK smuggled through SQL is caught by a txid fingerprint check and fails the step instead of silently splitting the writes from their witness. The witness table matches the Go SDK's shape (transaction_completion, step_id; schema-qualified "dbos" on Postgres, configurable, unqualified on SQLite). There is no cross-SDK contract to hold: Python names the table datasource_outputs and TypeScript renames the step column, so the table is language-local everywhere and Go wins as the parity anchor. Closes #148.
| if self.in_transaction.swap(true, Ordering::SeqCst) { | ||
| return Err(Error::app( | ||
| "cannot start a transaction inside another transaction", | ||
| )); | ||
| } | ||
| struct ResetOnDrop<'a>(&'a AtomicBool); | ||
| impl Drop for ResetOnDrop<'_> { | ||
| fn drop(&mut self) { | ||
| self.0.store(false, Ordering::SeqCst); | ||
| } | ||
| } | ||
| let _reset = ResetOnDrop(&self.in_transaction); |
There was a problem hiding this comment.
tiny improvement: this could be extracted out in something like a fn begin_transaction(&self) -> Result<ResetOnDrop> helper since it's the same code as transaction_with
There was a problem hiding this comment.
I see, applied in 1564182, both methods are down to one line
|
This LGTM. The only thing I'd mention, possibly as a future (micro?)-optimization, is to shortcut into a single commit if we're in the same database. I know callers can use the regular |
Both transaction_with and transaction_on_with carried the same 14-line nesting guard (flag swap + a function-local reset-on-drop struct). Pull it into begin_transaction() returning a TxFlagGuard, so the flag protocol has one definition. Suggested in review on #153.
I see, filed a follow-up issue #155 for this.. Maybe we can have the engine hand out the datasource, something like |
|
@assiotis I'll just merge this one anyway. thanks for the review |
2 participants
Closes #148.
All three reference SDKs support durable transactions against an application database of the user's own; durare only offered the single-database transactional step (
ctx.transaction). This PR addsctx.transaction_on(&ds, name, |conn| …)— andtransaction_on_withfor isolation/read-only/retry options — over aPgDataSourceorSqliteDataSourcebuilt on the user's own sqlx pool.Protocol
A single commit cannot span two databases, so the call splits durability into two commits bridged by a witness row:
transaction_completionrow (workflow_id, step_id, output, serialization) commit atomically on the user's database.operation_outputs.Recovery replays in layers: a system-database checkpoint wins outright; a witness row without a checkpoint (a crash landed between the two commits) replays the stored output without re-running the body and backfills the checkpoint; only when neither exists does the body run. Permanent failures are mirrored into the witness table before the system record, preserving the recovery order. Conflicts and transient database errors retry on fresh transactions without consuming the application
max_retriesbudget — the same two-loop structure as the single-database transactional step.Design decisions
&mut sqlx::PgConnection/&mut sqlx::SqliteConnection, so existing queries, compile-time-checked sqlx macros, and DAOs work unchanged. TheDataSourcetrait is sealed — the backend set is closed, as in every DBOS SDK (Go'sEngineis a closed type-set union; Python and TS ship concrete per-engine classes). Python and TS hand the body their ecosystem's native handle (SQLAlchemy session, ORM tx client); sqlx'sExecutoris the Rust equivalent. Discussed with @assiotis's proposal on [Parity] Durable transactions against a separate application database (DataSource / transaction_completion model) #148 — this follows that direction.COMMIT/ROLLBACKsmuggled through SQL is caught by atxid_current()fingerprint check and fails the step instead of silently splitting the writes from their witness. No reference SDK detects this — Go'sTxeven exposesCommit/Rollbackto the body.transaction_completionwhile Python usesdatasource_outputs, and TS renamesstep_id→function_num. The table is language-local in every SDK, so durare matches Go (name consensus + parity anchor); a Go worker sharing the application database reads the same rows. Schema isdbosby default, configurable viawith_schema(plain-identifier-validated), unqualified on SQLite.ctx.transaction— the two-commit protocol is still correct there, just one commit dearer.Tests
tests/datasource.rs(6): witness row + exactly-once + idempotent restart; a planted completion row replays without running the body (the crash window, and equivalently a foreign-written row); failure rollback + mirror + error replay; the application retry policy re-runs on fresh transactions with failed attempts leaving no writes; nesting rejected; hermetic-Postgres end-to-end covering the custom schema, hostile schema-name rejection, and the raw-COMMIT guard.Docs: new "A separate application database" section in the
transactionsguide, full rustdoc on both datasource types andtransaction_on, CHANGELOG entry under Unreleased.