From 34d8268a03c56def6a93b7f99b7b51aa0e26db19 Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Wed, 5 Aug 2026 20:19:56 -0700 Subject: [PATCH 1/2] feat(transactions): durable transactions on a separate application database 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. --- CHANGELOG.md | 16 ++ src/context.rs | 359 ++++++++++++++++++++++++++++++++++++ src/datasource.rs | 439 ++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 10 + src/postgres.rs | 2 +- src/transactions.rs | 51 ++++- tests/datasource.rs | 411 +++++++++++++++++++++++++++++++++++++++++ 7 files changed, 1286 insertions(+), 2 deletions(-) create mode 100644 src/datasource.rs create mode 100644 tests/datasource.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 08715cc..c4bad13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,22 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). the denial under the cross-SDK `DBOSNotAuthorizedError` envelope name. A declaration naming an unregistered workflow is rejected at launch/build. Documented in the `security` guide's new Authorization section. +- Durable transactions on a separate application database: + `ctx.transaction_on(&ds, name, |conn| …)` (and `transaction_on_with` for + isolation/read-only/retry options) runs a body against your own database + through a `PgDataSource` or `SqliteDataSource` over your `sqlx` pool. The + body's writes commit atomically with a `transaction_completion` witness + row in your database (table shape matches the Go SDK's; created on + construction, schema configurable on Postgres), then the step checkpoint + commits to the system database; recovery replays checkpoint-first, then + the witness row, so the body runs exactly once even across a crash + between the two commits. The body receives the backend's native `sqlx` + connection (`&mut PgConnection` / `&mut SqliteConnection`), so existing + queries, `sqlx` macros, and DAOs work unchanged; on Postgres a raw + `COMMIT`/`ROLLBACK` inside the body is detected and refused. Permanent + failures are mirrored into the witness table; conflicts retry on fresh + transactions without consuming the application retry budget. The + `DataSource` trait is sealed. Documented in the `transactions` guide. ### Documentation - HTTP triggering recipe: a runnable axum example diff --git a/src/context.rs b/src/context.rs index c979205..3a4e267 100644 --- a/src/context.rs +++ b/src/context.rs @@ -631,6 +631,354 @@ impl DurableContext { out } + /// Run a durable transaction on a **separate application database**. + /// + /// [`transaction`](Self::transaction) commits the body's SQL and the step + /// checkpoint together — but only in the *system* database. This runs the + /// body against your own database through a + /// [`PgDataSource`](crate::PgDataSource) or + /// [`SqliteDataSource`](crate::SqliteDataSource), keeping the same + /// exactly-once guarantee with a two-commit protocol: the body's writes + /// and a `transaction_completion` witness row commit atomically on the + /// application database, then the ordinary checkpoint is written to the + /// system database. Recovery replays in layers — checkpoint first, then + /// the completion row (a crash between the two commits) — and re-runs the + /// body only when neither exists. + /// + /// The body receives the backend's **native `sqlx` connection** + /// (`&mut sqlx::PgConnection` / `&mut sqlx::SqliteConnection`), so + /// existing queries, `sqlx` macros, and data-access helpers work + /// unchanged. durare owns the transaction: there is no commit method on a + /// plain connection, and on Postgres a raw `COMMIT`/`ROLLBACK` statement + /// smuggled through SQL is detected and fails the step. Like + /// [`transaction`](Self::transaction), the body must be re-runnable + /// (`Fn`): a serialization conflict or deadlock restarts it on a fresh + /// transaction. + /// + /// If your application database *is* the system database, prefer + /// [`transaction`](Self::transaction) — one commit instead of two. + /// + /// ```no_run + /// # use durare::{DurableContext, PgDataSource, Result}; + /// # async fn ex(ctx: DurableContext, ds: PgDataSource) -> Result<()> { + /// let total: i64 = ctx + /// .transaction_on(&ds, "record-order", |conn| Box::pin(async move { + /// sqlx::query("INSERT INTO orders(item) VALUES ($1)") + /// .bind("widget") + /// .execute(&mut *conn) + /// .await?; + /// let n = sqlx::query_scalar("SELECT count(*) FROM orders") + /// .fetch_one(&mut *conn) + /// .await?; + /// Ok(n) + /// })) + /// .await?; + /// # let _ = total; + /// # Ok(()) } + /// ``` + #[cfg(any(feature = "postgres", feature = "sqlite"))] + pub async fn transaction_on(&self, ds: &DS, name: &str, f: F) -> Result + where + DS: crate::datasource::DataSource, + T: Serialize + DeserializeOwned + 'static, + F: for<'c> Fn(&'c mut DS::Conn) -> Pin> + Send + 'c>> + + Send + + Sync + + 'static, + { + self.transaction_on_with(ds, TransactionOptions::new(name), f) + .await + } + + /// Like [`transaction_on`](Self::transaction_on) but with explicit + /// [`TransactionOptions`] — isolation level (advisory on SQLite), + /// read-only, and the application-error retry policy. Conflicts and + /// transient database errors are retried on a fresh transaction + /// regardless, without consuming the `max_retries` budget; once that + /// budget is exhausted the failure is recorded in **both** databases, so a + /// replay returns the same error without re-running the body. + #[cfg(any(feature = "postgres", feature = "sqlite"))] + pub async fn transaction_on_with( + &self, + ds: &DS, + opts: TransactionOptions, + f: F, + ) -> Result + where + DS: crate::datasource::DataSource, + T: Serialize + DeserializeOwned + 'static, + F: for<'c> Fn(&'c mut DS::Conn) -> Pin> + Send + 'c>> + + Send + + Sync + + 'static, + { + // Same nesting guard as `transaction_with`: a body that opens another + // transaction (on either database) is refused up front. + 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); + + let seq = self.next_seq(); + let span = self.op_span("transaction", &opts.name, seq); + let out = self + .run_datasource_transaction(ds, &opts, &f, seq) + .instrument(span.clone()) + .await; + span.record("otel.status_code", if out.is_ok() { "OK" } else { "ERROR" }); + out + } + + /// The two-commit protocol behind [`transaction_on`](Self::transaction_on): + /// layered replay, then fresh execution under the same two-loop retry + /// structure as the single-database transactional step. + #[cfg(any(feature = "postgres", feature = "sqlite"))] + async fn run_datasource_transaction( + &self, + ds: &DS, + opts: &TransactionOptions, + f: &F, + seq: i32, + ) -> Result + where + DS: crate::datasource::DataSource, + T: Serialize + DeserializeOwned + 'static, + F: for<'c> Fn(&'c mut DS::Conn) -> Pin> + Send + 'c>> + + Send + + Sync + + 'static, + { + // Layer 1: the system-database checkpoint — a completed run. + if let Some(stored) = self.replay_or_guard::(seq, &opts.name).await? { + return Ok(stored); + } + let started = chrono::Utc::now().timestamp_millis(); + + // Layer 2: a completion row without a checkpoint — the application + // transaction committed but the run crashed before the system commit. + // Replay the stored outcome without re-running the body. + if let Some(row) = ds.fetch_completion(&self.workflow_id, seq).await? { + return self + .replay_completion_row(seq, &opts.name, row, started) + .await; + } + + let ser = self.provider.serializer(); + + // OUTER loop: the user-facing retry policy for application errors, + // mirroring the single-database transactional step. Conflicts are + // handled by the inner loop and don't count against this budget. + let mut user_attempt: u32 = 0; + let body_err = loop { + // INNER loop: one committed attempt, or an application error + // surfaced to the outer loop. A serialization/deadlock conflict or + // transient DB error rolls back and retries on a fresh transaction + // — unbounded (until it clears or the workflow is cancelled). + let mut conflict_attempt: u32 = 0; + let outcome = loop { + match self.datasource_attempt(ds, opts, f, seq, &ser).await { + Ok(DsAttempt::Committed(value)) => break Ok(value), + // Another execution committed this step first: its row is + // the canonical outcome — replay it. + Ok(DsAttempt::AlreadyCompleted) => { + let row = ds + .fetch_completion(&self.workflow_id, seq) + .await? + .ok_or_else(|| { + Error::app( + "transaction_completion row vanished after a duplicate insert", + ) + })?; + return self + .replay_completion_row(seq, &opts.name, row, started) + .await; + } + Err(e) if e.is_tx_conflict() || e.is_retryable() => { + self.datasource_conflict_wait(conflict_attempt).await?; + conflict_attempt = conflict_attempt.saturating_add(1); + } + Err(e) => break Err(e), + } + }; + match outcome { + Ok(value) => { + // Second commit: checkpoint into the system database. The + // application transaction is already durable, so a racing + // writer's canonical outcome wins if there is one. + let stored = self + .provider + .record_step_result( + &self.workflow_id, + seq, + &opts.name, + value, + None, + Some(started), + ) + .await?; + return outcome_value(stored); + } + Err(e) if opts.should_user_retry(&e, user_attempt) => { + let delay = opts.user_retry_backoff(user_attempt); + tracing::warn!( + step = %opts.name, + attempt = user_attempt + 1, + error = %e, + "transaction failed; retrying after backoff" + ); + tokio::time::sleep(delay).await; + user_attempt += 1; + } + Err(e) => break e, + } + }; + + // Mirror the permanent failure into the application database (the + // body's transaction rolled back, so this is a standalone insert), + // written before the system-database record to keep the + // layer-1-then-layer-2 recovery order. Best-effort: the system + // database remains the source of truth. + let encoded = crate::serialize::encode_error(&ser, &body_err); + if let Err(mirror_err) = ds + .insert_failure(&self.workflow_id, seq, &encoded, ser.name()) + .await + { + tracing::warn!( + step = %opts.name, + error = %mirror_err, + "failed to mirror the transaction failure into the application database" + ); + } + self.record_failure(seq, &opts.name, body_err, Some(started)) + .await + } + + /// One fresh application-database attempt: begin, run the body, write the + /// completion row, commit — all atomic. Begins a fresh transaction on + /// every call so a closed/aborted one never leaks into a retry. + #[cfg(any(feature = "postgres", feature = "sqlite"))] + async fn datasource_attempt( + &self, + ds: &DS, + opts: &TransactionOptions, + f: &F, + seq: i32, + ser: &crate::serialize::Serializer, + ) -> Result + where + DS: crate::datasource::DataSource, + T: Serialize + DeserializeOwned + 'static, + F: for<'c> Fn(&'c mut DS::Conn) -> Pin> + Send + 'c>> + + Send + + Sync + + 'static, + { + let mut tx = ds.begin(opts.isolation, opts.read_only).await?; + let fingerprint = ds.tx_fingerprint(&mut *tx).await?; + match f(&mut *tx).await { + Ok(v) => { + let value = serde_json::to_value(v)?; + // A body that ended our transaction via raw SQL would make the + // completion row commit separately from the writes it + // witnesses — detect and refuse instead of breaking atomicity. + if let Some(expected) = &fingerprint { + if ds.tx_fingerprint(&mut *tx).await?.as_ref() != Some(expected) { + let _ = ds.rollback(tx).await; + return Err(Error::app( + "the transaction body terminated the surrounding database \ + transaction (a raw COMMIT or ROLLBACK?), so its writes cannot \ + be committed atomically with the durability record", + )); + } + } + let encoded = ser.encode(&value)?; + if !ds + .insert_completion( + &mut *tx, + &self.workflow_id, + seq, + Some(&encoded), + None, + ser.name(), + ) + .await? + { + let _ = ds.rollback(tx).await; + return Ok(DsAttempt::AlreadyCompleted); + } + ds.commit(tx).await?; + Ok(DsAttempt::Committed(value)) + } + Err(e) => { + let _ = ds.rollback(tx).await; + Err(e) + } + } + } + + /// Replay a layer-2 completion row: backfill the system-database + /// checkpoint from it, then surface the stored outcome — the recorded + /// output, or the recorded failure as its reconstructed error. + #[cfg(any(feature = "postgres", feature = "sqlite"))] + async fn replay_completion_row( + &self, + seq: i32, + name: &str, + row: crate::datasource::CompletionRow, + started: i64, + ) -> Result { + tracing::Span::current().record("dbos.step.replayed", true); + if let Some(err_text) = row.error.as_deref() { + let stored = self + .provider + .record_step_result( + &self.workflow_id, + seq, + name, + Value::Null, + Some(err_text), + Some(started), + ) + .await?; + return outcome_value(stored); + } + let output = row + .output + .as_deref() + .ok_or_else(|| Error::app("transaction completion row has neither output nor error"))?; + let ser = self.provider.serializer(); + let value = crate::serialize::decode(&ser, row.serialization.as_deref(), output)?; + let stored = self + .provider + .record_step_result(&self.workflow_id, seq, name, value, None, Some(started)) + .await?; + outcome_value(stored) + } + + /// Back off after an application-database conflict, bailing out if the + /// workflow has been cancelled — so a transaction stuck on contention or a + /// transient outage keeps retrying until it clears or the workflow is + /// actually cancelled. + #[cfg(any(feature = "postgres", feature = "sqlite"))] + async fn datasource_conflict_wait(&self, attempt: u32) -> Result<()> { + if let Some(status) = self.provider.get_workflow_status(&self.workflow_id).await? { + if status.status == STATUS_CANCELLED { + return Err(Error::Cancelled(self.workflow_id.clone())); + } + } + let ms = (1u64 << attempt.min(10)).min(1000); + tokio::time::sleep(std::time::Duration::from_millis(ms)).await; + Ok(()) + } + /// Race several async `branches` and return the `(index, value)` of the first /// to complete — a **durable** select. /// @@ -1474,6 +1822,17 @@ const LISTEN_NOTIFY_BACKSTOP: Duration = Duration::from_secs(5); /// read back consistently. const PATCH_PREFIX: &str = "DBOS.patch-"; +/// Outcome of one application-database attempt in the two-commit protocol +/// behind [`DurableContext::transaction_on`]. +#[cfg(any(feature = "postgres", feature = "sqlite"))] +enum DsAttempt { + /// The body ran and its transaction committed; carries the JSON output. + Committed(Value), + /// A completion row already existed — another execution committed this + /// step first, and its row is the canonical outcome. + AlreadyCompleted, +} + /// Turn a recorded step outcome into the typed value a step returns: a recorded /// output is deserialized; a recorded failure is surfaced as its reconstructed /// error (so a replayed failed step returns the same error without re-running). diff --git a/src/datasource.rs b/src/datasource.rs new file mode 100644 index 0000000..3e95949 --- /dev/null +++ b/src/datasource.rs @@ -0,0 +1,439 @@ +//! Data sources: durable transactions on a **separate application database** +//! (`DurableContext::transaction_on`). Not rendered — this module is private; +//! the user-facing docs live on [`PgDataSource`]/[`SqliteDataSource`], +//! `transaction_on`, and the `transactions` guide. +//! +//! The protocol splits durability into two commits with a witness row: the +//! application transaction commits the user's writes plus a +//! `transaction_completion` row atomically, then the ordinary step checkpoint +//! is written to the system database. Recovery replays in layers — checkpoint +//! first, completion row second (the crash window between the commits) — and +//! only runs the body when neither exists. The table shape matches the Go +//! SDK's (`transaction_completion`, `step_id`); Python (`datasource_outputs`) +//! and TypeScript (`function_num`) diverge from Go and from each other, so +//! there is no cross-SDK contract to hold — Go, our parity anchor, wins. + +#[cfg(feature = "postgres")] +use crate::error::Error; +use crate::error::Result; +use crate::tx::IsolationLevel; +use async_trait::async_trait; +use std::ops::DerefMut; + +/// One row of the `transaction_completion` table: the witness that the +/// application transaction for `(workflow_id, step_id)` committed (output set) +/// or permanently failed (error set). +pub struct CompletionRow { + pub(crate) output: Option, + pub(crate) error: Option, + pub(crate) serialization: Option, +} + +pub(crate) mod sealed { + use super::*; + + /// The per-backend surface of a data source. Sealed: the protocol in + /// `DurableContext::transaction_on` is written once against this trait, and + /// the set of backends is closed (like every DBOS SDK's), so it is not + /// implementable outside the crate. + #[async_trait] + pub trait Backend: Send + Sync { + /// The native connection type handed to a transaction body. + type Conn: Send; + /// The in-progress native transaction; derefs to [`Self::Conn`]. + type NativeTx: DerefMut + Send; + + /// Begin a transaction on the application pool at `isolation` + /// (advisory on SQLite, which is always serializable). + async fn begin(&self, isolation: IsolationLevel, read_only: bool) + -> Result; + + /// Commit `tx`. + async fn commit(&self, tx: Self::NativeTx) -> Result<()>; + + /// Roll back `tx` (best-effort; dropping also rolls back). + async fn rollback(&self, tx: Self::NativeTx) -> Result<()>; + + /// An identifier of the connection's *current* transaction, used to + /// detect a body that terminated the surrounding transaction via raw + /// SQL. `None` when the backend cannot cheaply provide one (SQLite). + async fn tx_fingerprint(&self, conn: &mut Self::Conn) -> Result>; + + /// Read the completion row for `(workflow_id, step_id)`, if any. + async fn fetch_completion( + &self, + workflow_id: &str, + step_id: i32, + ) -> Result>; + + /// Insert the completion row inside the caller's transaction. Returns + /// `false` when a row already exists (another execution committed this + /// step first — the caller rolls back and replays the canonical row). + #[allow(clippy::too_many_arguments)] + async fn insert_completion( + &self, + conn: &mut Self::Conn, + workflow_id: &str, + step_id: i32, + output: Option<&str>, + error: Option<&str>, + serialization: &str, + ) -> Result; + + /// Mirror a permanent failure into the completion table, outside any + /// transaction (the body's transaction rolled back). Idempotent: an + /// existing row is left untouched. + async fn insert_failure( + &self, + workflow_id: &str, + step_id: i32, + error: &str, + serialization: &str, + ) -> Result<()>; + } +} + +/// A handle to an application database that +/// [`transaction_on`](crate::DurableContext::transaction_on) runs durable +/// transactions against. +/// +/// Implemented by [`PgDataSource`] and [`SqliteDataSource`]. The trait is +/// **sealed** — the backend set is closed, like every DBOS SDK's — but usable +/// as a bound for helpers generic over the backend. The associated `Conn` type +/// (from the sealed supertrait) is the native `sqlx` connection a transaction +/// body receives: `sqlx::PgConnection` for [`PgDataSource`], +/// `sqlx::SqliteConnection` for [`SqliteDataSource`]. +pub trait DataSource: sealed::Backend {} + +const COMPLETION_COLUMNS: &str = "workflow_id, step_id, output, error, serialization, created_at"; + +/// A [`DataSource`] over a Postgres application database. +/// +/// Pass it to +/// [`transaction_on`](crate::DurableContext::transaction_on) to run a durable +/// transaction whose body receives the native `&mut sqlx::PgConnection` — +/// existing queries, `sqlx` macros, and data-access helpers work unchanged. +/// +/// Constructing one ensures the completion table exists: +/// `"".transaction_completion` (schema `dbos` by default), creating +/// the schema and table if missing — so the pool's role needs `CREATE` +/// privileges once, or create the table ahead of time from your own +/// migrations: +/// +/// ```sql +/// CREATE SCHEMA IF NOT EXISTS "dbos"; +/// CREATE TABLE IF NOT EXISTS "dbos".transaction_completion ( +/// workflow_id TEXT NOT NULL, +/// step_id INT NOT NULL, +/// output TEXT, +/// error TEXT, +/// serialization TEXT, +/// created_at BIGINT NOT NULL DEFAULT (EXTRACT(EPOCH FROM now())*1000)::bigint, +/// PRIMARY KEY (workflow_id, step_id) +/// ); +/// ``` +/// +/// The shape matches the Go DBOS SDK's, so a Go worker sharing this +/// application database reads the same rows. +#[cfg(feature = "postgres")] +#[derive(Clone)] +pub struct PgDataSource { + pool: sqlx::PgPool, + table: String, +} + +#[cfg(feature = "postgres")] +impl PgDataSource { + /// Create a data source over `pool`, keeping the completion table under + /// the default `dbos` schema. + pub async fn new(pool: sqlx::PgPool) -> Result { + Self::with_schema(pool, "dbos").await + } + + /// Like [`new`](Self::new), with the completion table under `schema` + /// instead of `dbos`. The name must be a plain identifier + /// (`[A-Za-z_][A-Za-z0-9_]*`). + pub async fn with_schema(pool: sqlx::PgPool, schema: &str) -> Result { + if !crate::postgres::is_plain_identifier(schema) { + return Err(Error::app(format!( + "invalid Postgres schema name {schema:?}: must match [A-Za-z_][A-Za-z0-9_]*" + ))); + } + sqlx::query(&format!("CREATE SCHEMA IF NOT EXISTS \"{schema}\"")) + .execute(&pool) + .await?; + let table = format!("\"{schema}\".transaction_completion"); + sqlx::query(&format!( + "CREATE TABLE IF NOT EXISTS {table} ( + workflow_id TEXT NOT NULL, + step_id INT NOT NULL, + output TEXT, + error TEXT, + serialization TEXT, + created_at BIGINT NOT NULL DEFAULT (EXTRACT(EPOCH FROM now())*1000)::bigint, + PRIMARY KEY (workflow_id, step_id) + )" + )) + .execute(&pool) + .await?; + Ok(Self { pool, table }) + } + + /// The pool this data source runs on. + pub fn pool(&self) -> &sqlx::PgPool { + &self.pool + } +} + +#[cfg(feature = "postgres")] +impl DataSource for PgDataSource {} + +#[cfg(feature = "postgres")] +#[async_trait] +impl sealed::Backend for PgDataSource { + type Conn = sqlx::PgConnection; + type NativeTx = sqlx::Transaction<'static, sqlx::Postgres>; + + async fn begin(&self, isolation: IsolationLevel, read_only: bool) -> Result { + let mut tx = self.pool.begin().await?; + // `SET TRANSACTION` must come before any query in the tx. + if isolation != IsolationLevel::ReadCommitted || read_only { + let mut stmt = format!("SET TRANSACTION ISOLATION LEVEL {}", isolation.pg_sql()); + if read_only { + stmt.push_str(" READ ONLY"); + } + sqlx::query(&stmt).execute(&mut *tx).await?; + } + Ok(tx) + } + + async fn commit(&self, tx: Self::NativeTx) -> Result<()> { + tx.commit().await?; + Ok(()) + } + + async fn rollback(&self, tx: Self::NativeTx) -> Result<()> { + tx.rollback().await?; + Ok(()) + } + + async fn tx_fingerprint(&self, conn: &mut Self::Conn) -> Result> { + // txid_current() is stable for the life of one transaction, so a body + // that ran COMMIT/ROLLBACK via raw SQL lands in a new transaction and + // the fingerprint changes. + let id: i64 = sqlx::query_scalar("SELECT txid_current()::bigint") + .fetch_one(conn) + .await?; + Ok(Some(id.to_string())) + } + + async fn fetch_completion( + &self, + workflow_id: &str, + step_id: i32, + ) -> Result> { + use sqlx::Row as _; + let row = sqlx::query(&format!( + "SELECT output, error, serialization FROM {} WHERE workflow_id = $1 AND step_id = $2", + self.table + )) + .bind(workflow_id) + .bind(step_id) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| CompletionRow { + output: r.get("output"), + error: r.get("error"), + serialization: r.get("serialization"), + })) + } + + async fn insert_completion( + &self, + conn: &mut Self::Conn, + workflow_id: &str, + step_id: i32, + output: Option<&str>, + error: Option<&str>, + serialization: &str, + ) -> Result { + let res = sqlx::query(&format!( + "INSERT INTO {} ({COMPLETION_COLUMNS}) VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (workflow_id, step_id) DO NOTHING", + self.table + )) + .bind(workflow_id) + .bind(step_id) + .bind(output) + .bind(error) + .bind(serialization) + .bind(chrono::Utc::now().timestamp_millis()) + .execute(conn) + .await?; + Ok(res.rows_affected() == 1) + } + + async fn insert_failure( + &self, + workflow_id: &str, + step_id: i32, + error: &str, + serialization: &str, + ) -> Result<()> { + sqlx::query(&format!( + "INSERT INTO {} ({COMPLETION_COLUMNS}) VALUES ($1, $2, NULL, $3, $4, $5) + ON CONFLICT (workflow_id, step_id) DO NOTHING", + self.table + )) + .bind(workflow_id) + .bind(step_id) + .bind(error) + .bind(serialization) + .bind(chrono::Utc::now().timestamp_millis()) + .execute(&self.pool) + .await?; + Ok(()) + } +} + +/// A [`DataSource`] over a SQLite application database. +/// +/// Pass it to +/// [`transaction_on`](crate::DurableContext::transaction_on) to run a durable +/// transaction whose body receives the native `&mut sqlx::SqliteConnection`. +/// +/// Constructing one ensures the (unqualified) `transaction_completion` table +/// exists, with the same columns as [`PgDataSource`]'s. +/// +/// SQLite runs every transaction serializably, so the isolation level in +/// [`TransactionOptions`](crate::TransactionOptions) is advisory here; +/// busy/locked contention is retried like a serialization conflict. +#[cfg(feature = "sqlite")] +#[derive(Clone)] +pub struct SqliteDataSource { + pool: sqlx::SqlitePool, +} + +#[cfg(feature = "sqlite")] +impl SqliteDataSource { + /// Create a data source over `pool`, creating the completion table if + /// missing. + pub async fn new(pool: sqlx::SqlitePool) -> Result { + sqlx::query( + "CREATE TABLE IF NOT EXISTS transaction_completion ( + workflow_id TEXT NOT NULL, + step_id INTEGER NOT NULL, + output TEXT, + error TEXT, + serialization TEXT, + created_at INTEGER NOT NULL DEFAULT (CAST(strftime('%s','now') AS INTEGER) * 1000), + PRIMARY KEY (workflow_id, step_id) + )", + ) + .execute(&pool) + .await?; + Ok(Self { pool }) + } + + /// The pool this data source runs on. + pub fn pool(&self) -> &sqlx::SqlitePool { + &self.pool + } +} + +#[cfg(feature = "sqlite")] +impl DataSource for SqliteDataSource {} + +#[cfg(feature = "sqlite")] +#[async_trait] +impl sealed::Backend for SqliteDataSource { + type Conn = sqlx::SqliteConnection; + type NativeTx = sqlx::Transaction<'static, sqlx::Sqlite>; + + async fn begin(&self, _isolation: IsolationLevel, _read_only: bool) -> Result { + // SQLite is always serializable; isolation/read-only are advisory. + Ok(self.pool.begin().await?) + } + + async fn commit(&self, tx: Self::NativeTx) -> Result<()> { + tx.commit().await?; + Ok(()) + } + + async fn rollback(&self, tx: Self::NativeTx) -> Result<()> { + tx.rollback().await?; + Ok(()) + } + + async fn tx_fingerprint(&self, _conn: &mut Self::Conn) -> Result> { + // No cheap transaction identifier on SQLite; the guard is documented + // as Postgres-only. + Ok(None) + } + + async fn fetch_completion( + &self, + workflow_id: &str, + step_id: i32, + ) -> Result> { + use sqlx::Row as _; + let row = sqlx::query( + "SELECT output, error, serialization FROM transaction_completion + WHERE workflow_id = ? AND step_id = ?", + ) + .bind(workflow_id) + .bind(step_id) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| CompletionRow { + output: r.get("output"), + error: r.get("error"), + serialization: r.get("serialization"), + })) + } + + async fn insert_completion( + &self, + conn: &mut Self::Conn, + workflow_id: &str, + step_id: i32, + output: Option<&str>, + error: Option<&str>, + serialization: &str, + ) -> Result { + let res = sqlx::query(&format!( + "INSERT OR IGNORE INTO transaction_completion ({COMPLETION_COLUMNS}) + VALUES (?, ?, ?, ?, ?, ?)" + )) + .bind(workflow_id) + .bind(step_id) + .bind(output) + .bind(error) + .bind(serialization) + .bind(chrono::Utc::now().timestamp_millis()) + .execute(conn) + .await?; + Ok(res.rows_affected() == 1) + } + + async fn insert_failure( + &self, + workflow_id: &str, + step_id: i32, + error: &str, + serialization: &str, + ) -> Result<()> { + sqlx::query(&format!( + "INSERT OR IGNORE INTO transaction_completion ({COMPLETION_COLUMNS}) + VALUES (?, ?, NULL, ?, ?, ?)" + )) + .bind(workflow_id) + .bind(step_id) + .bind(error) + .bind(serialization) + .bind(chrono::Utc::now().timestamp_millis()) + .execute(&self.pool) + .await?; + Ok(()) + } +} diff --git a/src/lib.rs b/src/lib.rs index 1ad96dd..631908f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -177,6 +177,8 @@ mod client; #[cfg(feature = "conductor")] mod conductor; mod context; +#[cfg(any(feature = "postgres", feature = "sqlite"))] +mod datasource; mod debounce; mod engine; mod error; @@ -200,6 +202,14 @@ pub use client::Client; #[cfg_attr(docsrs, doc(cfg(feature = "conductor")))] pub use conductor::{AlertHandler, Conductor, ConductorConfig}; pub use context::{AuthContext, DurableContext, RetryPredicate, StepOptions}; +#[cfg(any(feature = "postgres", feature = "sqlite"))] +pub use datasource::DataSource; +#[cfg(feature = "postgres")] +#[cfg_attr(docsrs, doc(cfg(feature = "postgres")))] +pub use datasource::PgDataSource; +#[cfg(feature = "sqlite")] +#[cfg_attr(docsrs, doc(cfg(feature = "sqlite")))] +pub use datasource::SqliteDataSource; pub use debounce::{Debouncer, DebouncerClient}; /// Macro plumbing referenced by `#[durare::workflow]`; not public API. #[doc(hidden)] diff --git a/src/postgres.rs b/src/postgres.rs index ba7ce35..bb9bbed 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -109,7 +109,7 @@ const DEFAULT_SCHEMA: &str = "dbos"; /// A schema name safe to interpolate into `CREATE SCHEMA` and `search_path` /// without quoting games: `[A-Za-z_][A-Za-z0-9_]*`. -fn is_plain_identifier(s: &str) -> bool { +pub(crate) fn is_plain_identifier(s: &str) -> bool { let mut chars = s.chars(); matches!(chars.next(), Some(c) if c.is_ascii_alphabetic() || c == '_') && chars.all(|c| c.is_ascii_alphanumeric() || c == '_') diff --git a/src/transactions.rs b/src/transactions.rs index c0ed5fb..d3d79a9 100644 --- a/src/transactions.rs +++ b/src/transactions.rs @@ -66,7 +66,8 @@ //! commits. On replay, the recorded outcome is read back and returned without //! running the body at all. One consequence of the single-transaction model: //! the tables you touch must live in the **same database** as the `dbos` -//! system schema. +//! system schema. For tables in a database of their own, see [a separate +//! application database](#a-separate-application-database) below. //! //! Transactions require a SQL backend — [`PostgresProvider`] or //! [`SqliteProvider`]. On [`InMemoryProvider`] they return an error. @@ -96,6 +97,54 @@ //! body on a new transaction with exponential backoff, and only the final //! outcome is checkpointed. //! +//! # A separate application database +//! +//! When your tables live in their own database — not the one holding the +//! `dbos` schema — a single commit can no longer cover both the writes and +//! the checkpoint. [`DurableContext::transaction_on`] keeps the exactly-once +//! guarantee anyway, with a **two-commit protocol**: construct a +//! [`PgDataSource`] or [`SqliteDataSource`](crate::SqliteDataSource) over +//! your own `sqlx` pool, and the body's writes commit atomically **with a +//! witness row** in a `transaction_completion` table that durare creates in +//! your database; the ordinary checkpoint follows as a second commit to the +//! system database. Recovery replays in layers — checkpoint first, then the +//! witness row (covering a crash between the two commits) — so the body still +//! runs exactly once. +//! +//! Unlike [`Tx`], the body receives the backend's **native `sqlx` +//! connection** (`&mut sqlx::PgConnection` / `&mut sqlx::SqliteConnection`), +//! so existing queries, compile-time-checked `sqlx` macros, and data-access +//! helpers written against `sqlx` work unchanged: +//! +//! ```no_run +//! # use durare::{DurableContext, PgDataSource, Result}; +//! # async fn ex(ctx: DurableContext, ds: PgDataSource) -> Result<()> { +//! let n: i64 = ctx +//! .transaction_on(&ds, "record-order", |conn| Box::pin(async move { +//! sqlx::query("INSERT INTO orders(item) VALUES ($1)") +//! .bind("widget") +//! .execute(&mut *conn) +//! .await?; +//! Ok(1) +//! })) +//! .await?; +//! # let _ = n; +//! # Ok(()) } +//! ``` +//! +//! The trade for the native connection: the body is committed to one backend +//! at the call site, where a [`Tx`] body runs on either. Failure semantics, +//! isolation, and the retry policy match [`transaction`] — with the failure +//! also mirrored into the witness table, so your database is self-describing. +//! durare owns the transaction: the connection has no commit method, and on +//! Postgres a raw `COMMIT`/`ROLLBACK` smuggled through SQL is detected and +//! fails the step. If the "separate" database is actually the system +//! database, prefer [`transaction`] — one commit instead of two. +//! +//! [`transaction`]: crate::DurableContext::transaction +//! [`DurableContext::transaction_on`]: crate::DurableContext::transaction_on +//! [`PgDataSource`]: crate::PgDataSource +//! //! # Writing the SQL //! //! The [`Tx`] API is dialect-agnostic: write `?` placeholders (rewritten to diff --git a/tests/datasource.rs b/tests/datasource.rs new file mode 100644 index 0000000..6961d1e --- /dev/null +++ b/tests/datasource.rs @@ -0,0 +1,411 @@ +//! Durable transactions on a separate application database +//! (`ctx.transaction_on`): the body's writes and a `transaction_completion` +//! witness row commit atomically on the app database, the checkpoint goes to +//! the system database, and recovery replays in layers — checkpoint first, +//! then the completion row — so the body runs exactly once even across the +//! crash window between the two commits. + +use durare::{ + DurableContext, DurableEngine, Error, InMemoryProvider, Result, Serializer, SqliteDataSource, + TransactionOptions, WorkflowOptions, +}; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; + +mod common; + +/// A fresh SQLite application database (a real file — the pool may open more +/// than one connection) with an `orders` table, plus its `SqliteDataSource`. +async fn sqlite_app_db() -> (SqliteDataSource, sqlx::SqlitePool) { + let mut path = std::env::temp_dir(); + path.push(format!("durare-ds-{}.db", uuid::Uuid::new_v4())); + let pool = sqlx::sqlite::SqlitePool::connect(&format!("sqlite://{}?mode=rwc", path.display())) + .await + .expect("open sqlite app db"); + sqlx::query("CREATE TABLE orders (item TEXT NOT NULL)") + .execute(&pool) + .await + .expect("create app table"); + let ds = SqliteDataSource::new(pool.clone()) + .await + .expect("create datasource"); + (ds, pool) +} + +async fn order_count(pool: &sqlx::SqlitePool) -> i64 { + sqlx::query_scalar("SELECT COUNT(*) FROM orders") + .fetch_one(pool) + .await + .unwrap() +} + +/// Happy path: the body's insert and the completion row commit together; the +/// output round-trips; starting the same workflow id again replays without +/// running the body. +#[tokio::test] +async fn commits_writes_with_witness_row_exactly_once() -> Result<()> { + let (ds, pool) = sqlite_app_db().await; + let runs = Arc::new(AtomicU32::new(0)); + + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + let (wf_ds, wf_runs) = (ds.clone(), runs.clone()); + engine.register("order", move |ctx: DurableContext, item: String| { + let (ds, runs) = (wf_ds.clone(), wf_runs.clone()); + async move { + ctx.transaction_on(&ds, "record-order", move |conn| { + let (item, runs) = (item.clone(), runs.clone()); + Box::pin(async move { + runs.fetch_add(1, Ordering::SeqCst); + sqlx::query("INSERT INTO orders(item) VALUES (?)") + .bind(&item) + .execute(&mut *conn) + .await?; + let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM orders") + .fetch_one(&mut *conn) + .await?; + Ok(n) + }) + }) + .await + } + }); + engine.launch().await?; + + let n: i64 = engine + .start::( + "order", + "widget".into(), + WorkflowOptions::with_id("ds-once"), + ) + .await? + .await?; + assert_eq!(n, 1); + assert_eq!(order_count(&pool).await, 1); + + // The witness row is in the application database, keyed by workflow+step. + let (output, error): (Option, Option) = sqlx::query_as( + "SELECT output, error FROM transaction_completion WHERE workflow_id = 'ds-once' AND step_id = 0", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert!(output.is_some(), "success row stores the output"); + assert!(error.is_none()); + + // Starting the same workflow id again returns the recorded result without + // re-running anything. + let again: i64 = engine + .start::( + "order", + "widget".into(), + WorkflowOptions::with_id("ds-once"), + ) + .await? + .await?; + assert_eq!(again, 1); + assert_eq!(runs.load(Ordering::SeqCst), 1, "body ran exactly once"); + assert_eq!(order_count(&pool).await, 1); + Ok(()) +} + +/// The crash window: the application transaction committed (completion row +/// present) but the system checkpoint was never written. The next execution +/// replays the stored output without running the body — layer-2 recovery. +#[tokio::test] +async fn completion_row_replays_without_rerunning_the_body() -> Result<()> { + let (ds, pool) = sqlite_app_db().await; + let runs = Arc::new(AtomicU32::new(0)); + + // A prior incarnation committed the app transaction for step 0 of + // "ds-window" and crashed before the checkpoint: plant its witness row, + // encoded the way the engine's serializer would have written it. + let ser = Serializer::Json; + sqlx::query( + "INSERT INTO transaction_completion (workflow_id, step_id, output, error, serialization, created_at) + VALUES ('ds-window', 0, ?, NULL, ?, 0)", + ) + .bind(ser.encode(&serde_json::json!(41))?) + .bind(ser.name()) + .execute(&pool) + .await + .unwrap(); + + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + let (wf_ds, wf_runs) = (ds.clone(), runs.clone()); + engine.register("window", move |ctx: DurableContext, (): ()| { + let (ds, runs) = (wf_ds.clone(), wf_runs.clone()); + async move { + let n: i64 = ctx + .transaction_on(&ds, "record-order", move |conn| { + let runs = runs.clone(); + Box::pin(async move { + runs.fetch_add(1, Ordering::SeqCst); + sqlx::query("INSERT INTO orders(item) VALUES ('should-not-run')") + .execute(&mut *conn) + .await?; + Ok(0) + }) + }) + .await?; + Ok::<_, Error>(n) + } + }); + engine.launch().await?; + + let n: i64 = engine + .start::<(), i64>("window", (), WorkflowOptions::with_id("ds-window")) + .await? + .await?; + assert_eq!(n, 41, "the stored output replayed"); + assert_eq!(runs.load(Ordering::SeqCst), 0, "the body never ran"); + assert_eq!(order_count(&pool).await, 0, "no new writes"); + Ok(()) +} + +/// A permanent failure rolls back the body's writes, mirrors the error into +/// the completion table, and replays as the same error: a second workflow +/// given the first one's mirrored row surfaces the error without running. +#[tokio::test] +async fn failure_rolls_back_mirrors_and_replays() -> Result<()> { + let (ds, pool) = sqlite_app_db().await; + let runs = Arc::new(AtomicU32::new(0)); + + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + let (wf_ds, wf_runs) = (ds.clone(), runs.clone()); + engine.register("doomed", move |ctx: DurableContext, (): ()| { + let (ds, runs) = (wf_ds.clone(), wf_runs.clone()); + async move { + ctx.transaction_on(&ds, "doomed-tx", move |conn| { + let runs = runs.clone(); + Box::pin(async move { + runs.fetch_add(1, Ordering::SeqCst); + sqlx::query("INSERT INTO orders(item) VALUES ('rolled-back')") + .execute(&mut *conn) + .await?; + Err::(Error::app("boom")) + }) + }) + .await + } + }); + engine.launch().await?; + + let err = engine + .start::<(), i64>("doomed", (), WorkflowOptions::with_id("ds-fail")) + .await? + .await + .expect_err("body error propagates"); + assert!(err.to_string().contains("boom"), "{err}"); + assert_eq!(order_count(&pool).await, 0, "the insert rolled back"); + + // The failure was mirrored into the application database. + let (output, error): (Option, Option) = sqlx::query_as( + "SELECT output, error FROM transaction_completion WHERE workflow_id = 'ds-fail' AND step_id = 0", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert!(output.is_none()); + let error = error.expect("failure row stores the error"); + + // Layer-2 error replay: plant that row under a fresh workflow id whose + // body would succeed — the stored failure wins and the body never runs. + sqlx::query( + "INSERT INTO transaction_completion (workflow_id, step_id, output, error, serialization, created_at) + VALUES ('ds-fail-replay', 0, NULL, ?, ?, 0)", + ) + .bind(&error) + .bind(Serializer::Json.name()) + .execute(&pool) + .await + .unwrap(); + let before = runs.load(Ordering::SeqCst); + let err = engine + .start::<(), i64>("doomed", (), WorkflowOptions::with_id("ds-fail-replay")) + .await? + .await + .expect_err("stored failure replays"); + assert!(err.to_string().contains("boom"), "{err}"); + assert_eq!(runs.load(Ordering::SeqCst), before, "the body never re-ran"); + Ok(()) +} + +/// The application-error retry policy: the body re-runs on a fresh +/// transaction per attempt, failed attempts leave no writes behind, and only +/// the final success commits. +#[tokio::test] +async fn retry_policy_reruns_on_fresh_transactions() -> Result<()> { + let (ds, pool) = sqlite_app_db().await; + let attempts = Arc::new(AtomicU32::new(0)); + + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + let (wf_ds, wf_attempts) = (ds.clone(), attempts.clone()); + engine.register("flaky", move |ctx: DurableContext, (): ()| { + let (ds, attempts) = (wf_ds.clone(), wf_attempts.clone()); + async move { + let opts = TransactionOptions::new("flaky-tx") + .max_retries(3) + .base_interval(std::time::Duration::from_millis(1)); + ctx.transaction_on_with(&ds, opts, move |conn| { + let attempts = attempts.clone(); + Box::pin(async move { + let n = attempts.fetch_add(1, Ordering::SeqCst); + sqlx::query("INSERT INTO orders(item) VALUES ('attempt')") + .execute(&mut *conn) + .await?; + if n < 2 { + return Err(Error::app("transient")); + } + Ok(()) + }) + }) + .await + } + }); + engine.launch().await?; + + engine + .start::<(), ()>("flaky", (), WorkflowOptions::with_id("ds-flaky")) + .await? + .await?; + assert_eq!( + attempts.load(Ordering::SeqCst), + 3, + "two failures, then success" + ); + assert_eq!( + order_count(&pool).await, + 1, + "rolled-back attempts left nothing" + ); + Ok(()) +} + +/// Nesting is refused up front, like the single-database transactional step. +#[tokio::test] +async fn nested_transaction_is_rejected() -> Result<()> { + let (ds, _pool) = sqlite_app_db().await; + + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + let wf_ds = ds.clone(); + engine.register("nested", move |ctx: DurableContext, (): ()| { + let ds = wf_ds.clone(); + async move { + let inner_ctx = ctx.clone(); + let inner_ds = ds.clone(); + ctx.transaction_on(&ds, "outer", move |_conn| { + let (ctx, ds) = (inner_ctx.clone(), inner_ds.clone()); + Box::pin(async move { + ctx.transaction_on(&ds, "inner", |_conn| Box::pin(async move { Ok(()) })) + .await + }) + }) + .await + } + }); + engine.launch().await?; + + let err = engine + .start::<(), ()>("nested", (), WorkflowOptions::with_id("ds-nested")) + .await? + .await + .expect_err("nested transaction"); + assert!( + err.to_string() + .contains("cannot start a transaction inside another transaction"), + "{err}" + ); + Ok(()) +} + +/// Postgres end to end in a hermetic database: schema-qualified witness table, +/// custom schema support, hostile schema names rejected, and the raw-COMMIT +/// guard. +#[tokio::test] +async fn pg_datasource_end_to_end() -> Result<()> { + use durare::PgDataSource; + + let Some(base) = std::env::var("DATABASE_URL").ok().filter(|s| !s.is_empty()) else { + eprintln!("skipping pg_datasource_end_to_end: DATABASE_URL unset"); + return Ok(()); + }; + let (admin, url, dbname) = common::hermetic_pg_db(&base, "durare_ds").await; + let pool = sqlx::postgres::PgPool::connect(&url).await.unwrap(); + sqlx::query("CREATE TABLE orders (item TEXT NOT NULL)") + .execute(&pool) + .await + .unwrap(); + + // A hostile schema name is rejected before any SQL runs. + let hostile = PgDataSource::with_schema(pool.clone(), "bad\"; DROP TABLE orders; --").await; + assert!(hostile.is_err(), "hostile schema name must be rejected"); + + // A custom (plain-identifier) schema works and holds the witness table. + let ds = PgDataSource::with_schema(pool.clone(), "app_durable").await?; + + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + let wf_ds = ds.clone(); + engine.register("order", move |ctx: DurableContext, (): ()| { + let ds = wf_ds.clone(); + async move { + ctx.transaction_on(&ds, "record-order", |conn| { + Box::pin(async move { + sqlx::query("INSERT INTO orders(item) VALUES ($1)") + .bind("widget") + .execute(&mut *conn) + .await?; + let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM orders") + .fetch_one(&mut *conn) + .await?; + Ok(n) + }) + }) + .await + } + }); + let guard_ds = ds.clone(); + engine.register("escape", move |ctx: DurableContext, (): ()| { + let ds = guard_ds.clone(); + async move { + ctx.transaction_on(&ds, "escape-tx", |conn| { + Box::pin(async move { + // A body must not end durare's transaction; this one does. + sqlx::query("COMMIT").execute(&mut *conn).await?; + Ok(0i64) + }) + }) + .await + } + }); + engine.launch().await?; + + let n: i64 = engine + .start::<(), i64>("order", (), WorkflowOptions::with_id("pg-ds-1")) + .await? + .await?; + assert_eq!(n, 1); + let witnesses: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM \"app_durable\".transaction_completion WHERE workflow_id = 'pg-ds-1'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(witnesses, 1, "witness row lives in the custom schema"); + + // The raw-COMMIT guard: detected, refused, surfaced as the step's error. + let err = engine + .start::<(), i64>("escape", (), WorkflowOptions::with_id("pg-ds-escape")) + .await? + .await + .expect_err("raw COMMIT inside the body"); + assert!( + err.to_string() + .contains("terminated the surrounding database transaction"), + "{err}" + ); + + engine.shutdown(std::time::Duration::from_secs(5)).await?; + pool.close().await; + common::drop_hermetic_pg_db(&admin, &dbname).await; + Ok(()) +} From 1564182a96bf7fe848a54ff09f47fb8354dea4fe Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Thu, 6 Aug 2026 12:38:25 -0700 Subject: [PATCH 2/2] refactor(context): extract the transaction re-entry guard into a helper 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. --- src/context.rs | 55 +++++++++++++++++++++----------------------------- 1 file changed, 23 insertions(+), 32 deletions(-) diff --git a/src/context.rs b/src/context.rs index 3a4e267..9c59eed 100644 --- a/src/context.rs +++ b/src/context.rs @@ -199,6 +199,17 @@ impl DurableContext { self.seq.fetch_add(1, Ordering::Relaxed) } + /// Set the in-transaction flag, refusing a nested transaction (it would + /// deadlock on the outer's write lock). The guard clears the flag on drop. + fn begin_transaction(&self) -> Result> { + if self.in_transaction.swap(true, Ordering::SeqCst) { + return Err(Error::app( + "cannot start a transaction inside another transaction", + )); + } + Ok(TxFlagGuard(&self.in_transaction)) + } + /// The span covering one durable operation (a step or a transaction), /// carrying the DBOS trace attributes (see the /// [`observability`](crate::observability) guide). Created inside the @@ -587,24 +598,7 @@ impl DurableContext { + Sync + 'static, { - // Reject a transaction nested inside another: the inner would open a - // separate connection and block forever on the write lock the outer - // already holds (a deadlock). A context clone captured in an outer body - // shares this flag, so a nested `ctx.transaction` is caught here rather - // than hanging. Checked before consuming a step slot. - if self.in_transaction.swap(true, Ordering::SeqCst) { - return Err(Error::app( - "cannot start a transaction inside another transaction", - )); - } - // Clear the flag on every exit path (including the `?` below). - 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); + let _guard = self.begin_transaction()?; let seq = self.next_seq(); let span = self.op_span("transaction", &opts.name, seq); @@ -712,20 +706,7 @@ impl DurableContext { + Sync + 'static, { - // Same nesting guard as `transaction_with`: a body that opens another - // transaction (on either database) is refused up front. - 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); + let _guard = self.begin_transaction()?; let seq = self.next_seq(); let span = self.op_span("transaction", &opts.name, seq); @@ -1822,6 +1803,16 @@ const LISTEN_NOTIFY_BACKSTOP: Duration = Duration::from_secs(5); /// read back consistently. const PATCH_PREFIX: &str = "DBOS.patch-"; +/// Clears the in-transaction flag on drop (see +/// [`DurableContext::begin_transaction`]). +struct TxFlagGuard<'a>(&'a AtomicBool); + +impl Drop for TxFlagGuard<'_> { + fn drop(&mut self) { + self.0.store(false, Ordering::SeqCst); + } +} + /// Outcome of one application-database attempt in the two-commit protocol /// behind [`DurableContext::transaction_on`]. #[cfg(any(feature = "postgres", feature = "sqlite"))]