From d73351cd6134f122a79da68b4d6103d97a7b7f03 Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Thu, 6 Aug 2026 21:06:57 -0700 Subject: [PATCH 01/11] feat(transactions): single-commit fast path via provider system_datasource (closes #155) --- CHANGELOG.md | 11 +++ src/context.rs | 153 +++++++++++++++++++++++++++-- src/datasource.rs | 125 +++++++++++++++++++++++- src/postgres.rs | 30 ++++++ src/sqlite.rs | 13 +++ src/transactions.rs | 30 +++++- tests/datasource.rs | 227 ++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 576 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c4bad13..25fa7b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,17 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### Added +- Single-commit fast path for application tables in the system database: + `PostgresProvider::system_datasource()` / `SqliteProvider::system_datasource()` + return a data source over the provider's own pool, so `transaction_on` + commits the body's writes and the step checkpoint in one transaction — the + same guarantee as `ctx.transaction`, with no witness table — while the body + keeps the native `sqlx` connection and its full type support (`jsonb`, + arrays, `uuid`, …) that the portable `Param` set can't express. Sameness is + established by construction (the provider hands out its own pool), never by + detection: a user-constructed `PgDataSource` always uses the two-commit + protocol, which stays correct on any database. + - Declarative role-based authorization: `require_roles(name, roles)` on the engine and builder declares the roles a caller must hold to invoke a workflow. The check runs before the body on every execution path (direct, diff --git a/src/context.rs b/src/context.rs index 9c59eed..8aa3fe9 100644 --- a/src/context.rs +++ b/src/context.rs @@ -649,8 +649,15 @@ impl DurableContext { /// (`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. + /// If your application tables live **in the system database**, get the + /// data source from the provider instead — + /// `PostgresProvider::system_datasource` / + /// `SqliteProvider::system_datasource`. Sameness is then known by + /// construction, and this call takes a **single-commit fast path**: the + /// body's writes and the checkpoint commit in one transaction (no witness + /// row, no crash window) while the body keeps the native connection — + /// unlike [`transaction`](Self::transaction), whose + /// [`Param`](crate::Param) bindings cover only a small portable type set. /// /// ```no_run /// # use durare::{DurableContext, PgDataSource, Result}; @@ -742,6 +749,16 @@ impl DurableContext { return Ok(stored); } let started = chrono::Utc::now().timestamp_millis(); + let ser = self.provider.serializer(); + + // A system data source runs on the system database's own pool, so one + // commit can cover the body's writes and the checkpoint — no witness + // row, no crash window, no layer 2. + if ds.is_system() { + return self + .run_system_datasource_transaction(ds, opts, f, seq, &ser, started) + .await; + } // Layer 2: a completion row without a checkpoint — the application // transaction committed but the run crashed before the system commit. @@ -752,8 +769,6 @@ impl DurableContext { .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. @@ -873,11 +888,7 @@ impl DurableContext { 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", - )); + return Err(Error::app(TX_TERMINATED_MSG)); } } let encoded = ser.encode(&value)?; @@ -905,6 +916,122 @@ impl DurableContext { } } + /// The single-commit fast path behind [`transaction_on`](Self::transaction_on) + /// for a **system** data source (one built by a provider's + /// `system_datasource`): the pool is the system database's own, so the + /// step checkpoint commits inside the body's transaction — same guarantee + /// as [`transaction`](Self::transaction), no witness row, no crash window. + /// Same two-loop retry structure as the two-commit path. + #[cfg(any(feature = "postgres", feature = "sqlite"))] + async fn run_system_datasource_transaction( + &self, + ds: &DS, + opts: &TransactionOptions, + f: &F, + seq: i32, + ser: &crate::serialize::Serializer, + started: i64, + ) -> 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 user_attempt: u32 = 0; + let body_err = loop { + let mut conflict_attempt: u32 = 0; + let outcome = loop { + match self + .system_datasource_attempt(ds, opts, f, seq, ser, started) + .await + { + Ok(value) => break Ok(value), + 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) => return Ok(serde_json::from_value(value)?), + 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, + } + }; + // No witness table on the fast path: the failure is recorded in the + // system database only, like the single-database transactional step. + self.record_failure(seq, &opts.name, body_err, Some(started)) + .await + } + + /// One fast-path attempt: begin on the system pool, run the body, insert + /// the `operation_outputs` checkpoint in the same transaction, commit. + #[cfg(any(feature = "postgres", feature = "sqlite"))] + async fn system_datasource_attempt( + &self, + ds: &DS, + opts: &TransactionOptions, + f: &F, + seq: i32, + ser: &crate::serialize::Serializer, + started: i64, + ) -> 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)?; + // Ending our transaction via raw SQL would split the writes + // from their checkpoint — detect and refuse. + 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(TX_TERMINATED_MSG)); + } + } + let encoded = ser.encode(&value)?; + ds.insert_checkpoint( + &mut *tx, + &self.workflow_id, + seq, + &opts.name, + &encoded, + ser.name(), + started, + ) + .await?; + ds.commit(tx).await?; + Ok(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. @@ -1803,6 +1930,14 @@ const LISTEN_NOTIFY_BACKSTOP: Duration = Duration::from_secs(5); /// read back consistently. const PATCH_PREFIX: &str = "DBOS.patch-"; +/// Error for a `transaction_on` body that ended durare's database transaction +/// via raw SQL, which would split the writes from their durability record. +#[cfg(any(feature = "postgres", feature = "sqlite"))] +const TX_TERMINATED_MSG: &str = + "the transaction body terminated the surrounding database transaction (a raw \ + COMMIT or ROLLBACK?), so its writes cannot be committed atomically with the \ + durability record"; + /// Clears the in-transaction flag on drop (see /// [`DurableContext::begin_transaction`]). struct TxFlagGuard<'a>(&'a AtomicBool); diff --git a/src/datasource.rs b/src/datasource.rs index 3e95949..9234019 100644 --- a/src/datasource.rs +++ b/src/datasource.rs @@ -90,6 +90,28 @@ pub(crate) mod sealed { error: &str, serialization: &str, ) -> Result<()>; + + /// Whether this data source runs on the system database's own pool + /// (built by a provider's `system_datasource`), enabling the + /// single-commit fast path: the checkpoint commits with the body's + /// writes, no completion row needed. + fn is_system(&self) -> bool; + + /// Insert the step checkpoint into `operation_outputs` on the caller's + /// transaction — the fast-path equivalent of the completion row plus + /// the system commit, in one. Only valid on a system data source, + /// whose pool resolves the unqualified system tables. + #[allow(clippy::too_many_arguments)] + async fn insert_checkpoint( + &self, + conn: &mut Self::Conn, + workflow_id: &str, + step_id: i32, + name: &str, + output: &str, + serialization: &str, + started_at_ms: i64, + ) -> Result<()>; } } @@ -140,6 +162,10 @@ const COMPLETION_COLUMNS: &str = "workflow_id, step_id, output, error, serializa pub struct PgDataSource { pool: sqlx::PgPool, table: String, + /// True when built by `PostgresProvider::system_datasource` — the pool is + /// the system database's own, so `transaction_on` takes the single-commit + /// fast path and the completion table is never used (or created). + system: bool, } #[cfg(feature = "postgres")] @@ -176,7 +202,22 @@ impl PgDataSource { )) .execute(&pool) .await?; - Ok(Self { pool, table }) + Ok(Self { + pool, + table, + system: false, + }) + } + + /// A data source over the system database's own pool (see + /// `PostgresProvider::system_datasource`). Creates nothing: the fast path + /// never touches a completion table. + pub(crate) fn system(pool: sqlx::PgPool) -> Self { + Self { + pool, + table: "transaction_completion".to_string(), + system: true, + } } /// The pool this data source runs on. @@ -294,6 +335,41 @@ impl sealed::Backend for PgDataSource { .await?; Ok(()) } + + fn is_system(&self) -> bool { + self.system + } + + async fn insert_checkpoint( + &self, + conn: &mut Self::Conn, + workflow_id: &str, + step_id: i32, + name: &str, + output: &str, + serialization: &str, + started_at_ms: i64, + ) -> Result<()> { + // Unqualified: the system pool's per-connection search_path resolves + // the system schema, same as the provider's own queries. + sqlx::query( + "INSERT INTO operation_outputs + (workflow_uuid, function_id, function_name, output, serialization, + started_at_epoch_ms, completed_at_epoch_ms) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (workflow_uuid, function_id) DO NOTHING", + ) + .bind(workflow_id) + .bind(step_id) + .bind(name) + .bind(output) + .bind(serialization) + .bind(started_at_ms) + .bind(chrono::Utc::now().timestamp_millis()) + .execute(conn) + .await?; + Ok(()) + } } /// A [`DataSource`] over a SQLite application database. @@ -312,6 +388,9 @@ impl sealed::Backend for PgDataSource { #[derive(Clone)] pub struct SqliteDataSource { pool: sqlx::SqlitePool, + /// True when built by `SqliteProvider::system_datasource` — see + /// [`PgDataSource`]'s `system` field. + system: bool, } #[cfg(feature = "sqlite")] @@ -332,7 +411,16 @@ impl SqliteDataSource { ) .execute(&pool) .await?; - Ok(Self { pool }) + Ok(Self { + pool, + system: false, + }) + } + + /// A data source over the system database's own pool (see + /// `SqliteProvider::system_datasource`). Creates nothing. + pub(crate) fn system(pool: sqlx::SqlitePool) -> Self { + Self { pool, system: true } } /// The pool this data source runs on. @@ -436,4 +524,37 @@ impl sealed::Backend for SqliteDataSource { .await?; Ok(()) } + + fn is_system(&self) -> bool { + self.system + } + + async fn insert_checkpoint( + &self, + conn: &mut Self::Conn, + workflow_id: &str, + step_id: i32, + name: &str, + output: &str, + serialization: &str, + started_at_ms: i64, + ) -> Result<()> { + sqlx::query( + "INSERT INTO operation_outputs + (workflow_uuid, function_id, function_name, output, serialization, + started_at_epoch_ms, completed_at_epoch_ms) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (workflow_uuid, function_id) DO NOTHING", + ) + .bind(workflow_id) + .bind(step_id) + .bind(name) + .bind(output) + .bind(serialization) + .bind(started_at_ms) + .bind(chrono::Utc::now().timestamp_millis()) + .execute(conn) + .await?; + Ok(()) + } } diff --git a/src/postgres.rs b/src/postgres.rs index bb9bbed..fc82bb9 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -202,6 +202,36 @@ impl PostgresProvider { &self.pool } + /// A [`PgDataSource`](crate::PgDataSource) over this provider's own pool, + /// for application tables that live **in the system database**. + /// + /// Because the pool is known to be the system database's, + /// [`transaction_on`](crate::DurableContext::transaction_on) takes a + /// **single-commit fast path**: the body's writes and the step checkpoint + /// commit in one transaction — the same guarantee as + /// [`transaction`](crate::DurableContext::transaction), but the body gets + /// the native `&mut sqlx::PgConnection` (full Postgres type support: + /// `jsonb`, arrays, `uuid`, …) instead of the `Param`-limited [`Tx`] + /// wrapper. No completion table is used or created. + /// + /// # Trust note + /// + /// The connection handed to a transaction body is unrestricted by design: + /// it can address the system tables, because the application owns the + /// database and the credential — not the SDK — is the real protection + /// boundary (the same is true of [`Tx`], whose SQL text is equally + /// unfiltered). What *is* guarded is the silent accident: a stray + /// `COMMIT`/`ROLLBACK` in the body is detected and fails the step. Also + /// note these connections start with `search_path` set to the system + /// schema, so **unqualified** DDL in a body (`CREATE TABLE orders …`) + /// lands in that schema next to the system tables — qualify your table + /// names if you care where they live. + /// + /// [`Tx`]: crate::Tx + pub fn system_datasource(&self) -> crate::PgDataSource { + crate::PgDataSource::system(self.pool.clone()) + } + /// Choose the format new values are encoded with. Use [`Serializer::Portable`] /// when this database is shared with DBOS workers in other languages. pub fn with_serializer(mut self, serializer: Serializer) -> Self { diff --git a/src/sqlite.rs b/src/sqlite.rs index 6c0714e..f4d83f2 100644 --- a/src/sqlite.rs +++ b/src/sqlite.rs @@ -72,6 +72,19 @@ impl SqliteProvider { } } + /// A [`SqliteDataSource`](crate::SqliteDataSource) over this provider's + /// own pool, for application tables that live **in the system database**. + /// [`transaction_on`](crate::DurableContext::transaction_on) then takes + /// the single-commit fast path — writes and checkpoint in one transaction, + /// with the body on the native `&mut sqlx::SqliteConnection`; no + /// completion table is used or created. See + /// `PostgresProvider::system_datasource` for the full story, including + /// the trust note (the connection is unrestricted by design — the + /// application owns the database). + pub fn system_datasource(&self) -> crate::SqliteDataSource { + crate::SqliteDataSource::system(self.pool.clone()) + } + /// Back off before the next attempt of the unbounded transaction-conflict /// retry loop, unless the workflow has been cancelled — in which case return /// [`Error::Cancelled`] so an operator can stop a transaction wedged on a diff --git a/src/transactions.rs b/src/transactions.rs index d3d79a9..a6c1a43 100644 --- a/src/transactions.rs +++ b/src/transactions.rs @@ -138,8 +138,34 @@ //! 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. +//! fails the step. +//! +//! ## Application tables in the system database +//! +//! When your tables share the database that holds the `dbos` schema, ask the +//! **provider** for the data source instead of building one yourself: +//! +//! ```no_run +//! # use durare::{DurableEngine, PostgresProvider, Result}; +//! # use std::sync::Arc; +//! # async fn ex(url: &str) -> Result<()> { +//! let provider = PostgresProvider::connect(url).await?; +//! let ds = provider.system_datasource(); // provider's own pool — no guessing +//! let engine = DurableEngine::new(Arc::new(provider)).await?; +//! # let _ = (ds, engine); +//! # Ok(()) } +//! ``` +//! +//! Because that data source is built from the provider's own pool, sameness +//! is true by construction (never detected or guessed), and `transaction_on` +//! takes a **single-commit fast path**: the body's writes and the step +//! checkpoint commit in one transaction — the same guarantee as +//! [`transaction`], with no witness table at all — while the body keeps the +//! native connection and its full type support (`jsonb`, arrays, `uuid`, …) +//! that [`Param`]'s portable set can't express. A user-constructed +//! [`PgDataSource`] never takes the fast path, even if its pool happens to +//! point at the system database: a wrong "same database" guess would break +//! atomicity, so the shortcut is reserved for the case that can't be wrong. //! //! [`transaction`]: crate::DurableContext::transaction //! [`DurableContext::transaction_on`]: crate::DurableContext::transaction_on diff --git a/tests/datasource.rs b/tests/datasource.rs index 6961d1e..cb094e2 100644 --- a/tests/datasource.rs +++ b/tests/datasource.rs @@ -409,3 +409,230 @@ async fn pg_datasource_end_to_end() -> Result<()> { common::drop_hermetic_pg_db(&admin, &dbname).await; Ok(()) } + +/// The system-datasource fast path: writes and checkpoint in one commit on the +/// system database's own pool, no witness table, native connection, replay +/// from the ordinary checkpoint. +#[tokio::test] +async fn system_datasource_single_commit_without_witness_table() -> Result<()> { + use durare::SqliteProvider; + + let mut path = std::env::temp_dir(); + path.push(format!("durare-sysds-{}.db", uuid::Uuid::new_v4())); + let url = format!("sqlite://{}", path.display()); + let provider = SqliteProvider::connect(&url).await?; + let ds = provider.system_datasource(); + let pool = ds.pool().clone(); + let runs = Arc::new(AtomicU32::new(0)); + + let mut engine = DurableEngine::new(Arc::new(provider)).await?; + sqlx::query("CREATE TABLE sys_orders (item TEXT NOT NULL)") + .execute(&pool) + .await + .unwrap(); + + let (wf_ds, wf_runs) = (ds.clone(), runs.clone()); + engine.register("sys-order", move |ctx: DurableContext, (): ()| { + let (ds, runs) = (wf_ds.clone(), wf_runs.clone()); + async move { + ctx.transaction_on(&ds, "sys-tx", move |conn| { + let runs = runs.clone(); + Box::pin(async move { + runs.fetch_add(1, Ordering::SeqCst); + sqlx::query("INSERT INTO sys_orders(item) VALUES ('gear')") + .execute(&mut *conn) + .await?; + let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sys_orders") + .fetch_one(&mut *conn) + .await?; + Ok(n) + }) + }) + .await + } + }); + engine.launch().await?; + + let n: i64 = engine + .start::<(), i64>("sys-order", (), WorkflowOptions::with_id("sysds-1")) + .await? + .await?; + assert_eq!(n, 1); + + // No witness table was ever created — the fast path doesn't need one. + let witness_tables: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'transaction_completion'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(witness_tables, 0, "fast path creates no witness table"); + + // The checkpoint committed with the writes, as an ordinary step row. + let checkpoints: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM operation_outputs + WHERE workflow_uuid = 'sysds-1' AND function_name = 'sys-tx' AND output IS NOT NULL", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(checkpoints, 1); + + // Replay: same workflow id returns the recorded result, body untouched. + let again: i64 = engine + .start::<(), i64>("sys-order", (), WorkflowOptions::with_id("sysds-1")) + .await? + .await?; + assert_eq!(again, 1); + assert_eq!(runs.load(Ordering::SeqCst), 1, "body ran exactly once"); + Ok(()) +} + +/// Fast-path failure: the body's writes roll back with nothing half-committed, +/// the error is recorded as an ordinary step failure, and a replay returns the +/// same error without re-running the body. +#[tokio::test] +async fn system_datasource_failure_rolls_back_and_replays() -> Result<()> { + use durare::SqliteProvider; + + let mut path = std::env::temp_dir(); + path.push(format!("durare-sysds-{}.db", uuid::Uuid::new_v4())); + let url = format!("sqlite://{}", path.display()); + let provider = SqliteProvider::connect(&url).await?; + let ds = provider.system_datasource(); + let pool = ds.pool().clone(); + let runs = Arc::new(AtomicU32::new(0)); + + let mut engine = DurableEngine::new(Arc::new(provider)).await?; + sqlx::query("CREATE TABLE sys_orders (item TEXT NOT NULL)") + .execute(&pool) + .await + .unwrap(); + + let (wf_ds, wf_runs) = (ds.clone(), runs.clone()); + engine.register("sys-doomed", move |ctx: DurableContext, (): ()| { + let (ds, runs) = (wf_ds.clone(), wf_runs.clone()); + async move { + ctx.transaction_on(&ds, "sys-doomed-tx", move |conn| { + let runs = runs.clone(); + Box::pin(async move { + runs.fetch_add(1, Ordering::SeqCst); + sqlx::query("INSERT INTO sys_orders(item) VALUES ('rolled-back')") + .execute(&mut *conn) + .await?; + Err::(Error::app("sys-boom")) + }) + }) + .await + } + }); + engine.launch().await?; + + let err = engine + .start::<(), i64>("sys-doomed", (), WorkflowOptions::with_id("sysds-fail")) + .await? + .await + .expect_err("body error propagates"); + assert!(err.to_string().contains("sys-boom"), "{err}"); + let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sys_orders") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(rows, 0, "the insert rolled back"); + + let before = runs.load(Ordering::SeqCst); + let err = engine + .start::<(), i64>("sys-doomed", (), WorkflowOptions::with_id("sysds-fail")) + .await? + .await + .expect_err("recorded failure replays"); + assert!(err.to_string().contains("sys-boom"), "{err}"); + assert_eq!(runs.load(Ordering::SeqCst), before, "body never re-ran"); + Ok(()) +} + +/// Postgres fast path, end to end: the motivating case — rich Postgres types +/// (jsonb, bigint arrays) bound natively in a single-commit durable +/// transaction — plus no witness table anywhere in the database. +#[tokio::test] +async fn pg_system_datasource_rich_types_single_commit() -> Result<()> { + use durare::PostgresProvider; + + let Some(base) = std::env::var("DATABASE_URL").ok().filter(|s| !s.is_empty()) else { + eprintln!("skipping pg_system_datasource_rich_types_single_commit: DATABASE_URL unset"); + return Ok(()); + }; + let (admin, url, dbname) = common::hermetic_pg_db(&base, "durare_sysds").await; + let provider = PostgresProvider::connect(&url).await?; + let ds = provider.system_datasource(); + let pool = ds.pool().clone(); + + let mut engine = DurableEngine::new(Arc::new(provider)).await?; + sqlx::query("CREATE TABLE sys_orders (meta JSONB NOT NULL, tags BIGINT[] NOT NULL)") + .execute(&pool) + .await + .unwrap(); + + let wf_ds = ds.clone(); + engine.register("sys-rich", move |ctx: DurableContext, (): ()| { + let ds = wf_ds.clone(); + async move { + ctx.transaction_on(&ds, "sys-rich-tx", |conn| { + Box::pin(async move { + // Types Param cannot express, bound natively. + sqlx::query("INSERT INTO sys_orders(meta, tags) VALUES ($1, $2)") + .bind(serde_json::json!({"reason": "fee", "amount": 42})) + .bind(vec![7i64, 11, 13]) + .execute(&mut *conn) + .await?; + let amount: i64 = + sqlx::query_scalar("SELECT (meta->>'amount')::bigint FROM sys_orders") + .fetch_one(&mut *conn) + .await?; + Ok(amount) + }) + }) + .await + } + }); + engine.launch().await?; + + let amount: i64 = engine + .start::<(), i64>("sys-rich", (), WorkflowOptions::with_id("pg-sysds-1")) + .await? + .await?; + assert_eq!( + amount, 42, + "jsonb round-tripped through the native connection" + ); + + let tags: Vec = sqlx::query_scalar("SELECT tags FROM sys_orders") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(tags, vec![7, 11, 13], "array round-tripped"); + + // No witness table anywhere: the fast path never creates one. + let witness_tables: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM information_schema.tables WHERE table_name = 'transaction_completion'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(witness_tables, 0, "fast path creates no witness table"); + + // The checkpoint is an ordinary step row, committed with the writes. + let checkpoints: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM operation_outputs + WHERE workflow_uuid = 'pg-sysds-1' AND function_name = 'sys-rich-tx' AND output IS NOT NULL", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(checkpoints, 1); + + engine.shutdown(std::time::Duration::from_secs(5)).await?; + pool.close().await; + common::drop_hermetic_pg_db(&admin, &dbname).await; + Ok(()) +} From 4bc639f9dd427fb4ec0ef88f3c028240c34948a8 Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Thu, 6 Aug 2026 22:05:40 -0700 Subject: [PATCH 02/11] docs(transactions): decision table, extension labels, and qualified-name guidance for the fast path --- src/context.rs | 7 +++++++ src/postgres.rs | 2 ++ src/sqlite.rs | 2 +- src/transactions.rs | 31 ++++++++++++++++++++++++++++--- 4 files changed, 38 insertions(+), 4 deletions(-) diff --git a/src/context.rs b/src/context.rs index 8aa3fe9..f24c50d 100644 --- a/src/context.rs +++ b/src/context.rs @@ -541,6 +541,13 @@ impl DurableContext { /// step. Requires a SQL backend (Postgres or SQLite); on the in-memory /// backend it returns an error. /// + /// This is the default transactional step: the body stays portable across + /// backends. When a step's types outgrow [`Param`](crate::Param) (`jsonb`, + /// arrays, `uuid`, …) or it should reuse sqlx-typed helpers, switch that + /// step to [`transaction_on`](Self::transaction_on) — see the + /// [`transactions`](crate::transactions) guide's "Which transaction API?" + /// table. + /// /// The body receives a [`Tx`] and returns a boxed future — `Box::pin(async /// move { … })`, mirroring sqlx's own transaction closures. SQL is written /// with `?` placeholders (rewritten to `$1, $2, …` for Postgres) and bound diff --git a/src/postgres.rs b/src/postgres.rs index fc82bb9..1576c0c 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -214,6 +214,8 @@ impl PostgresProvider { /// `jsonb`, arrays, `uuid`, …) instead of the `Param`-limited [`Tx`] /// wrapper. No completion table is used or created. /// + /// A **durare extension**. + /// /// # Trust note /// /// The connection handed to a transaction body is unrestricted by design: diff --git a/src/sqlite.rs b/src/sqlite.rs index f4d83f2..fa9b001 100644 --- a/src/sqlite.rs +++ b/src/sqlite.rs @@ -77,7 +77,7 @@ impl SqliteProvider { /// [`transaction_on`](crate::DurableContext::transaction_on) then takes /// the single-commit fast path — writes and checkpoint in one transaction, /// with the body on the native `&mut sqlx::SqliteConnection`; no - /// completion table is used or created. See + /// completion table is used or created. A durare extension; see /// `PostgresProvider::system_datasource` for the full story, including /// the trust note (the connection is unrestricted by design — the /// application owns the database). diff --git a/src/transactions.rs b/src/transactions.rs index a6c1a43..d38d2c4 100644 --- a/src/transactions.rs +++ b/src/transactions.rs @@ -146,13 +146,24 @@ //! **provider** for the data source instead of building one yourself: //! //! ```no_run -//! # use durare::{DurableEngine, PostgresProvider, Result}; +//! # use durare::{DurableContext, DurableEngine, PostgresProvider, Result}; //! # use std::sync::Arc; -//! # async fn ex(url: &str) -> Result<()> { +//! # async fn ex(ctx: DurableContext, url: &str) -> Result<()> { //! let provider = PostgresProvider::connect(url).await?; //! let ds = provider.system_datasource(); // provider's own pool — no guessing //! let engine = DurableEngine::new(Arc::new(provider)).await?; -//! # let _ = (ds, engine); +//! +//! ctx.transaction_on(&ds, "audit", |conn| Box::pin(async move { +//! // Qualify table names in fast-path bodies: this pool's search_path +//! // points at the system schema. +//! sqlx::query("INSERT INTO public.audit_log(entry) VALUES ($1)") +//! .bind(serde_json::json!({"kind": "transfer"})) // jsonb, natively +//! .execute(&mut *conn) +//! .await?; +//! Ok(()) +//! })) +//! .await?; +//! # let _ = engine; //! # Ok(()) } //! ``` //! @@ -166,8 +177,22 @@ //! [`PgDataSource`] never takes the fast path, even if its pool happens to //! point at the system database: a wrong "same database" guess would break //! atomicity, so the shortcut is reserved for the case that can't be wrong. +//! This is a durare extension. +//! +//! # Which transaction API? +//! +//! | Your situation | Use | Body receives | Commits | +//! |---|---|---|---| +//! | Simple types, tables in the system database | [`transaction`] | [`Tx`] — portable `?` SQL | 1 | +//! | Rich types or existing sqlx code, tables in the system database | [`transaction_on`] + `system_datasource()` | native connection | 1 | +//! | Tables in a separate database | [`transaction_on`] + [`PgDataSource`] | native connection | 2 | +//! +//! Rule of thumb: start with [`transaction`] — it keeps the body portable +//! across backends. Switch a step to [`transaction_on`] when its types +//! outgrow [`Param`] or it should reuse sqlx-typed helpers. //! //! [`transaction`]: crate::DurableContext::transaction +//! [`transaction_on`]: crate::DurableContext::transaction_on //! [`DurableContext::transaction_on`]: crate::DurableContext::transaction_on //! [`PgDataSource`]: crate::PgDataSource //! From c659c0b0f2d64d2a4a9fd729e1ef53f9396fb47f Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 14:07:32 -0700 Subject: [PATCH 03/11] feat(transactions): bind system data sources to their provider identity --- src/context.rs | 29 ++++++++++++++++++---- src/datasource.rs | 60 ++++++++++++++++++++++++++++----------------- src/lib.rs | 8 +++--- src/postgres.rs | 9 ++++++- src/provider.rs | 43 ++++++++++++++++++++++++++++++++ src/sqlite.rs | 9 ++++++- tests/datasource.rs | 50 +++++++++++++++++++++++++++++++++++++ 7 files changed, 175 insertions(+), 33 deletions(-) diff --git a/src/context.rs b/src/context.rs index f24c50d..b8b660e 100644 --- a/src/context.rs +++ b/src/context.rs @@ -760,11 +760,30 @@ impl DurableContext { // A system data source runs on the system database's own pool, so one // commit can cover the body's writes and the checkpoint — no witness - // row, no crash window, no layer 2. - if ds.is_system() { - return self - .run_system_datasource_transaction(ds, opts, f, seq, &ser, started) - .await; + // row, no crash window, no layer 2. But only under the engine whose + // provider minted it: the fast path writes the checkpoint through the + // data source's pool, which is only this workflow's system database if + // the identities match. A mismatch is a wiring bug — fail loudly + // rather than splitting the checkpoint from the status row. + match ds.kind() { + crate::datasource::DataSourceKind::System(identity) + if self + .provider + .provider_identity() + .is_some_and(|own| identity.matches(own)) => + { + return self + .run_system_datasource_transaction(ds, opts, f, seq, &ser, started) + .await; + } + crate::datasource::DataSourceKind::System(_) => { + return Err(Error::app( + "this system data source was minted by a different provider than the \ + one this workflow runs on; use system_datasource() from this \ + engine's own provider (or an external data source)", + )); + } + crate::datasource::DataSourceKind::External => {} } // Layer 2: a completion row without a checkpoint — the application diff --git a/src/datasource.rs b/src/datasource.rs index 9234019..f18cb50 100644 --- a/src/datasource.rs +++ b/src/datasource.rs @@ -29,6 +29,19 @@ pub struct CompletionRow { pub(crate) serialization: Option, } +/// What a data source points at, as far as durability is concerned. +#[derive(Clone)] +pub enum DataSourceKind { + /// A user-owned application database: the two-commit protocol with a + /// witness row applies. + External, + /// The system database itself, minted by the identified provider's + /// `system_datasource`. The single-commit fast path applies — but only + /// under an engine whose provider carries the *same* identity; any other + /// engine rejects the data source rather than misrouting its checkpoint. + System(crate::provider::ProviderIdentity), +} + pub(crate) mod sealed { use super::*; @@ -91,11 +104,10 @@ pub(crate) mod sealed { serialization: &str, ) -> Result<()>; - /// Whether this data source runs on the system database's own pool - /// (built by a provider's `system_datasource`), enabling the - /// single-commit fast path: the checkpoint commits with the body's - /// writes, no completion row needed. - fn is_system(&self) -> bool; + /// What this data source points at — external application database, + /// or the system database of an identified provider (see + /// [`DataSourceKind`]). + fn kind(&self) -> &DataSourceKind; /// Insert the step checkpoint into `operation_outputs` on the caller's /// transaction — the fast-path equivalent of the completion row plus @@ -162,10 +174,9 @@ const COMPLETION_COLUMNS: &str = "workflow_id, step_id, output, error, serializa pub struct PgDataSource { pool: sqlx::PgPool, table: String, - /// True when built by `PostgresProvider::system_datasource` — the pool is - /// the system database's own, so `transaction_on` takes the single-commit - /// fast path and the completion table is never used (or created). - system: bool, + /// What this data source points at; `System` enables the single-commit + /// fast path, bound to the minting provider's identity. + kind: DataSourceKind, } #[cfg(feature = "postgres")] @@ -205,18 +216,18 @@ impl PgDataSource { Ok(Self { pool, table, - system: false, + kind: DataSourceKind::External, }) } /// A data source over the system database's own pool (see /// `PostgresProvider::system_datasource`). Creates nothing: the fast path /// never touches a completion table. - pub(crate) fn system(pool: sqlx::PgPool) -> Self { + pub(crate) fn system(pool: sqlx::PgPool, identity: crate::provider::ProviderIdentity) -> Self { Self { pool, table: "transaction_completion".to_string(), - system: true, + kind: DataSourceKind::System(identity), } } @@ -336,8 +347,8 @@ impl sealed::Backend for PgDataSource { Ok(()) } - fn is_system(&self) -> bool { - self.system + fn kind(&self) -> &DataSourceKind { + &self.kind } async fn insert_checkpoint( @@ -388,9 +399,8 @@ impl sealed::Backend for PgDataSource { #[derive(Clone)] pub struct SqliteDataSource { pool: sqlx::SqlitePool, - /// True when built by `SqliteProvider::system_datasource` — see - /// [`PgDataSource`]'s `system` field. - system: bool, + /// What this data source points at — see [`PgDataSource`]'s `kind` field. + kind: DataSourceKind, } #[cfg(feature = "sqlite")] @@ -413,14 +423,20 @@ impl SqliteDataSource { .await?; Ok(Self { pool, - system: false, + kind: DataSourceKind::External, }) } /// A data source over the system database's own pool (see /// `SqliteProvider::system_datasource`). Creates nothing. - pub(crate) fn system(pool: sqlx::SqlitePool) -> Self { - Self { pool, system: true } + pub(crate) fn system( + pool: sqlx::SqlitePool, + identity: crate::provider::ProviderIdentity, + ) -> Self { + Self { + pool, + kind: DataSourceKind::System(identity), + } } /// The pool this data source runs on. @@ -525,8 +541,8 @@ impl sealed::Backend for SqliteDataSource { Ok(()) } - fn is_system(&self) -> bool { - self.system + fn kind(&self) -> &DataSourceKind { + &self.kind } async fn insert_checkpoint( diff --git a/src/lib.rs b/src/lib.rs index 631908f..683393a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -230,10 +230,10 @@ pub use memory::InMemoryProvider; pub use postgres::PostgresProvider; pub use provider::{ is_terminal, ChangeWait, DequeueRequest, ExportedWorkflow, ForkParams, ListFilter, - NotificationInsert, StateProvider, StepAggregate, StepAggregateQuery, StepInfo, VersionInfo, - WorkflowAggregate, WorkflowAggregateQuery, WorkflowStatus, STATUS_CANCELLED, STATUS_DELAYED, - STATUS_ENQUEUED, STATUS_ERROR, STATUS_MAX_RECOVERY_ATTEMPTS_EXCEEDED, STATUS_PENDING, - STATUS_SUCCESS, + NotificationInsert, ProviderIdentity, StateProvider, StepAggregate, StepAggregateQuery, + StepInfo, VersionInfo, WorkflowAggregate, WorkflowAggregateQuery, WorkflowStatus, + STATUS_CANCELLED, STATUS_DELAYED, STATUS_ENQUEUED, STATUS_ERROR, + STATUS_MAX_RECOVERY_ATTEMPTS_EXCEEDED, STATUS_PENDING, STATUS_SUCCESS, }; pub use queue::{RateLimiter, WorkflowQueue}; pub use schedule::{ diff --git a/src/postgres.rs b/src/postgres.rs index 1576c0c..a2fd5ad 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -132,6 +132,8 @@ pub struct PostgresProvider { notify_hub: Arc, /// Cancels the background listener task when the provider is dropped. listener_token: CancellationToken, + /// This instance's identity; binds system data sources to it. + identity: crate::provider::ProviderIdentity, /// Ensures the listener task is spawned at most once (on the first `init`). listener_started: AtomicBool, } @@ -193,6 +195,7 @@ impl PostgresProvider { notify_hub: Arc::new(NotifyHub::default()), listener_token: CancellationToken::new(), listener_started: AtomicBool::new(false), + identity: crate::provider::ProviderIdentity::new(), } } @@ -231,7 +234,7 @@ impl PostgresProvider { /// /// [`Tx`]: crate::Tx pub fn system_datasource(&self) -> crate::PgDataSource { - crate::PgDataSource::system(self.pool.clone()) + crate::PgDataSource::system(self.pool.clone(), self.identity.clone()) } /// Choose the format new values are encoded with. Use [`Serializer::Portable`] @@ -419,6 +422,10 @@ fn row_to_status(serializer: &Serializer, row: &sqlx::postgres::PgRow) -> Workfl #[async_trait] impl StateProvider for PostgresProvider { + fn provider_identity(&self) -> Option<&crate::provider::ProviderIdentity> { + Some(&self.identity) + } + async fn ping(&self) -> Result<()> { // One round trip proves reachability and that the dbos system schema // is migrated: `_sqlx_migrations` (in this pool's search_path schema) diff --git a/src/provider.rs b/src/provider.rs index e175110..5714697 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -6,6 +6,7 @@ use chrono::{DateTime, Utc}; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; +use std::sync::Arc; use std::time::Duration; /// Map a `workflow_status` insert failure to a typed deduplication error when it @@ -1207,6 +1208,39 @@ pub struct DequeueRequest { /// The pluggable durable-state backend. /// /// This is the single seam that decouples the runtime from storage. v0.1 ships a +/// An opaque identity for one provider *instance*: two handles match only if +/// they were cloned from the same original. Compared by pointer, so a match is +/// unforgeable — there is no way to construct an identity equal to another's. +/// +/// Used to bind a system data source (`system_datasource`) to the provider +/// that minted it, so +/// [`transaction_on`](crate::DurableContext::transaction_on) takes its +/// single-commit fast path only against that provider's own database and +/// rejects a system data source from a different engine. +#[derive(Clone, Default)] +pub struct ProviderIdentity(Arc); + +#[derive(Default)] +struct IdentityMarker; + +impl ProviderIdentity { + /// A fresh identity, equal only to its own clones. + pub fn new() -> Self { + Self::default() + } + + /// Whether `other` was cloned from the same original as `self`. + pub fn matches(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.0, &other.0) + } +} + +impl std::fmt::Debug for ProviderIdentity { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "ProviderIdentity({:p})", Arc::as_ptr(&self.0)) + } +} + /// Postgres implementation and an in-memory one; a DynamoDB / Aurora DSQL /// implementation can be added later **without touching the engine**. /// @@ -1239,6 +1273,15 @@ pub trait StateProvider: Send + Sync { crate::serialize::Serializer::Json } + /// This provider instance's [`ProviderIdentity`], if it issues one. Used to + /// verify that a system data source was minted by *this* provider before + /// the single-commit fast path is taken. The default (`None`) is fail-safe: + /// a provider without an identity never matches, so the fast path is never + /// wrongly taken against it. + fn provider_identity(&self) -> Option<&ProviderIdentity> { + None + } + /// Whether this backend pushes change signals (Postgres `LISTEN`/`NOTIFY`), /// so a blocked `recv`/`get_event` is woken as soon as the row it waits for /// is written rather than only by polling. Callers that get `true` can wait diff --git a/src/sqlite.rs b/src/sqlite.rs index fa9b001..f8d59eb 100644 --- a/src/sqlite.rs +++ b/src/sqlite.rs @@ -40,6 +40,8 @@ pub struct SqliteProvider { /// Format used when *encoding* stored values; decoding follows each row's /// recorded format. See [`crate::Serializer`]. serializer: Serializer, + /// This instance's identity; binds system data sources to it. + identity: crate::provider::ProviderIdentity, } impl SqliteProvider { @@ -69,6 +71,7 @@ impl SqliteProvider { Self { pool, serializer: Serializer::default(), + identity: crate::provider::ProviderIdentity::new(), } } @@ -82,7 +85,7 @@ impl SqliteProvider { /// the trust note (the connection is unrestricted by design — the /// application owns the database). pub fn system_datasource(&self) -> crate::SqliteDataSource { - crate::SqliteDataSource::system(self.pool.clone()) + crate::SqliteDataSource::system(self.pool.clone(), self.identity.clone()) } /// Back off before the next attempt of the unbounded transaction-conflict @@ -175,6 +178,10 @@ fn row_to_status(serializer: &Serializer, row: &sqlx::sqlite::SqliteRow) -> Work #[async_trait] impl StateProvider for SqliteProvider { + fn provider_identity(&self) -> Option<&crate::provider::ProviderIdentity> { + Some(&self.identity) + } + async fn ping(&self) -> Result<()> { // One round trip proves reachability and that the dbos system schema // is migrated (see the Postgres provider for the rule; a newer schema diff --git a/tests/datasource.rs b/tests/datasource.rs index cb094e2..3562e12 100644 --- a/tests/datasource.rs +++ b/tests/datasource.rs @@ -636,3 +636,53 @@ async fn pg_system_datasource_rich_types_single_commit() -> Result<()> { common::drop_hermetic_pg_db(&admin, &dbname).await; Ok(()) } + +/// A system data source is bound to the provider that minted it: used under a +/// different engine, the fast path is refused instead of misrouting the +/// checkpoint into the wrong system database. +#[tokio::test] +async fn foreign_system_datasource_is_rejected() -> Result<()> { + use durare::SqliteProvider; + + let mut path_a = std::env::temp_dir(); + path_a.push(format!("durare-sysds-a-{}.db", uuid::Uuid::new_v4())); + let provider_a = SqliteProvider::connect(&format!("sqlite://{}", path_a.display())).await?; + let foreign_ds = provider_a.system_datasource(); + // Engine A exists so provider A's database is real and migrated. + let _engine_a = DurableEngine::new(Arc::new(provider_a)).await?; + + let mut path_b = std::env::temp_dir(); + path_b.push(format!("durare-sysds-b-{}.db", uuid::Uuid::new_v4())); + let provider_b = SqliteProvider::connect(&format!("sqlite://{}", path_b.display())).await?; + let runs = Arc::new(AtomicU32::new(0)); + + let mut engine_b = DurableEngine::new(Arc::new(provider_b)).await?; + let (wf_ds, wf_runs) = (foreign_ds.clone(), runs.clone()); + engine_b.register("misrouted", move |ctx: DurableContext, (): ()| { + let (ds, runs) = (wf_ds.clone(), wf_runs.clone()); + async move { + ctx.transaction_on(&ds, "misrouted-tx", move |conn| { + let runs = runs.clone(); + Box::pin(async move { + runs.fetch_add(1, Ordering::SeqCst); + sqlx::query("SELECT 1").fetch_one(&mut *conn).await?; + Ok(()) + }) + }) + .await + } + }); + engine_b.launch().await?; + + let err = engine_b + .start::<(), ()>("misrouted", (), WorkflowOptions::with_id("sysds-foreign")) + .await? + .await + .expect_err("foreign system datasource must be refused"); + assert!( + err.to_string().contains("minted by a different provider"), + "{err}" + ); + assert_eq!(runs.load(Ordering::SeqCst), 0, "the body never ran"); + Ok(()) +} From 013a07d3a909c597f85d30d4ea3a28df851cdd56 Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 14:09:37 -0700 Subject: [PATCH 04/11] fix(transactions): roll back and replay the canonical outcome when the fast-path checkpoint already exists --- src/context.rs | 48 +++++++++++++++++++-------- src/datasource.rs | 20 ++++++------ tests/datasource.rs | 79 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 125 insertions(+), 22 deletions(-) diff --git a/src/context.rs b/src/context.rs index b8b660e..c1ff3ca 100644 --- a/src/context.rs +++ b/src/context.rs @@ -974,7 +974,17 @@ impl DurableContext { .system_datasource_attempt(ds, opts, f, seq, ser, started) .await { - Ok(value) => break Ok(value), + Ok(DsAttempt::Committed(value)) => break Ok(value), + // Another execution checkpointed this step first; its + // recorded outcome is canonical — replay it. + Ok(DsAttempt::AlreadyCompleted) => { + return self + .replay_or_guard::(seq, &opts.name) + .await? + .ok_or_else(|| { + Error::app("checkpoint row vanished after a duplicate insert") + }); + } Err(e) if e.is_tx_conflict() || e.is_retryable() => { self.datasource_conflict_wait(conflict_attempt).await?; conflict_attempt = conflict_attempt.saturating_add(1); @@ -1006,6 +1016,9 @@ impl DurableContext { /// One fast-path attempt: begin on the system pool, run the body, insert /// the `operation_outputs` checkpoint in the same transaction, commit. + /// `AlreadyCompleted` means another execution checkpointed this step + /// first: this attempt rolled back — its writes discarded — and the + /// caller replays the canonical outcome. #[cfg(any(feature = "postgres", feature = "sqlite"))] async fn system_datasource_attempt( &self, @@ -1015,7 +1028,7 @@ impl DurableContext { seq: i32, ser: &crate::serialize::Serializer, started: i64, - ) -> Result + ) -> Result where DS: crate::datasource::DataSource, T: Serialize + DeserializeOwned + 'static, @@ -1038,18 +1051,27 @@ impl DurableContext { } } let encoded = ser.encode(&value)?; - ds.insert_checkpoint( - &mut *tx, - &self.workflow_id, - seq, - &opts.name, - &encoded, - ser.name(), - started, - ) - .await?; + if !ds + .insert_checkpoint( + &mut *tx, + &self.workflow_id, + seq, + &opts.name, + &encoded, + ser.name(), + started, + ) + .await? + { + // Another execution already checkpointed this step. Roll + // back — discarding this attempt's writes keeps the step + // exactly-once even under duplicate execution — and let + // the caller replay the canonical outcome. + let _ = ds.rollback(tx).await; + return Ok(DsAttempt::AlreadyCompleted); + } ds.commit(tx).await?; - Ok(value) + Ok(DsAttempt::Committed(value)) } Err(e) => { let _ = ds.rollback(tx).await; diff --git a/src/datasource.rs b/src/datasource.rs index f18cb50..24081d0 100644 --- a/src/datasource.rs +++ b/src/datasource.rs @@ -111,8 +111,10 @@ pub(crate) mod sealed { /// Insert the step checkpoint into `operation_outputs` on the caller's /// transaction — the fast-path equivalent of the completion row plus - /// the system commit, in one. Only valid on a system data source, - /// whose pool resolves the unqualified system tables. + /// the system commit, in one. Only valid on a system data source. + /// Returns `false` when a checkpoint already exists (another execution + /// committed this step first — the caller rolls back, discarding this + /// attempt's writes, and replays the canonical outcome). #[allow(clippy::too_many_arguments)] async fn insert_checkpoint( &self, @@ -123,7 +125,7 @@ pub(crate) mod sealed { output: &str, serialization: &str, started_at_ms: i64, - ) -> Result<()>; + ) -> Result; } } @@ -360,10 +362,10 @@ impl sealed::Backend for PgDataSource { output: &str, serialization: &str, started_at_ms: i64, - ) -> Result<()> { + ) -> Result { // Unqualified: the system pool's per-connection search_path resolves // the system schema, same as the provider's own queries. - sqlx::query( + let res = sqlx::query( "INSERT INTO operation_outputs (workflow_uuid, function_id, function_name, output, serialization, started_at_epoch_ms, completed_at_epoch_ms) @@ -379,7 +381,7 @@ impl sealed::Backend for PgDataSource { .bind(chrono::Utc::now().timestamp_millis()) .execute(conn) .await?; - Ok(()) + Ok(res.rows_affected() == 1) } } @@ -554,8 +556,8 @@ impl sealed::Backend for SqliteDataSource { output: &str, serialization: &str, started_at_ms: i64, - ) -> Result<()> { - sqlx::query( + ) -> Result { + let res = sqlx::query( "INSERT INTO operation_outputs (workflow_uuid, function_id, function_name, output, serialization, started_at_epoch_ms, completed_at_epoch_ms) @@ -571,6 +573,6 @@ impl sealed::Backend for SqliteDataSource { .bind(chrono::Utc::now().timestamp_millis()) .execute(conn) .await?; - Ok(()) + Ok(res.rows_affected() == 1) } } diff --git a/tests/datasource.rs b/tests/datasource.rs index 3562e12..3bdb86b 100644 --- a/tests/datasource.rs +++ b/tests/datasource.rs @@ -686,3 +686,82 @@ async fn foreign_system_datasource_is_rejected() -> Result<()> { assert_eq!(runs.load(Ordering::SeqCst), 0, "the body never ran"); Ok(()) } + +/// A duplicate execution that loses the checkpoint race has its writes rolled +/// back and the canonical outcome replayed — the fast path is exactly-once +/// even under double execution. The rival checkpoint is planted from inside +/// the body over a separate connection, landing in the exact window between +/// the body's writes and this attempt's checkpoint insert. +#[tokio::test] +async fn duplicate_fast_path_execution_rolls_back_and_replays() -> Result<()> { + use durare::SqliteProvider; + + let mut path = std::env::temp_dir(); + path.push(format!("durare-sysds-dup-{}.db", uuid::Uuid::new_v4())); + let url = format!("sqlite://{}", path.display()); + let provider = SqliteProvider::connect(&url).await?; + let ds = provider.system_datasource(); + let pool = ds.pool().clone(); + let runs = Arc::new(AtomicU32::new(0)); + + let mut engine = DurableEngine::new(Arc::new(provider)).await?; + sqlx::query("CREATE TABLE sys_orders (item TEXT NOT NULL)") + .execute(&pool) + .await + .unwrap(); + + let (wf_ds, wf_runs, rival_pool) = (ds.clone(), runs.clone(), pool.clone()); + engine.register("sys-dup", move |ctx: DurableContext, (): ()| { + let (ds, runs, rival_pool) = (wf_ds.clone(), wf_runs.clone(), rival_pool.clone()); + let wf_id = ctx.workflow_id().to_string(); + async move { + ctx.transaction_on(&ds, "dup-tx", move |conn| { + let (runs, rival_pool, wf_id) = (runs.clone(), rival_pool.clone(), wf_id.clone()); + Box::pin(async move { + runs.fetch_add(1, Ordering::SeqCst); + // The "other executor" commits this step's checkpoint on a + // separate connection, right in the race window. + let ser = Serializer::Json; + sqlx::query( + "INSERT INTO operation_outputs + (workflow_uuid, function_id, function_name, output, serialization, + started_at_epoch_ms, completed_at_epoch_ms) + VALUES (?, 0, 'dup-tx', ?, ?, 0, 0)", + ) + .bind(&wf_id) + .bind(ser.encode(&serde_json::json!(99))?) + .bind(ser.name()) + .execute(&rival_pool) + .await?; + // This attempt's own write — must be rolled back. + sqlx::query("INSERT INTO sys_orders(item) VALUES ('duplicate')") + .execute(&mut *conn) + .await?; + Ok(1i64) + }) + }) + .await + } + }); + engine.launch().await?; + + let n: i64 = engine + .start::<(), i64>("sys-dup", (), WorkflowOptions::with_id("sysds-dup")) + .await? + .await?; + assert_eq!(n, 99, "the rival's canonical outcome replayed, not ours"); + assert_eq!(runs.load(Ordering::SeqCst), 1, "body ran once"); + assert_eq!( + order_count_in(&pool, "sys_orders").await, + 0, + "the losing attempt's writes rolled back" + ); + Ok(()) +} + +async fn order_count_in(pool: &sqlx::SqlitePool, table: &str) -> i64 { + sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {table}")) + .fetch_one(pool) + .await + .unwrap() +} From e9607d4ed8f6b9f89af5e556d7dfc83ad65419a9 Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 14:10:49 -0700 Subject: [PATCH 05/11] fix(transactions): schema-qualify the fast-path checkpoint insert --- src/datasource.rs | 36 +++++++++++++++++++++++++++--------- src/postgres.rs | 2 +- tests/datasource.rs | 5 +++++ 3 files changed, 33 insertions(+), 10 deletions(-) diff --git a/src/datasource.rs b/src/datasource.rs index 24081d0..16f928c 100644 --- a/src/datasource.rs +++ b/src/datasource.rs @@ -179,6 +179,9 @@ pub struct PgDataSource { /// What this data source points at; `System` enables the single-commit /// fast path, bound to the minting provider's identity. kind: DataSourceKind, + /// Where the fast path writes its checkpoint: schema-qualified when the + /// minting provider's schema is known, immune to `search_path` changes. + checkpoint_table: String, } #[cfg(feature = "postgres")] @@ -219,16 +222,31 @@ impl PgDataSource { pool, table, kind: DataSourceKind::External, + checkpoint_table: "operation_outputs".to_string(), }) } /// A data source over the system database's own pool (see /// `PostgresProvider::system_datasource`). Creates nothing: the fast path - /// never touches a completion table. - pub(crate) fn system(pool: sqlx::PgPool, identity: crate::provider::ProviderIdentity) -> Self { + /// never touches a completion table. `schema` is the provider's system + /// schema, used to fully qualify the checkpoint insert so nothing the + /// body does to `search_path` can redirect it; empty (a `from_pool` + /// provider) falls back to search_path resolution, the documented + /// contract for caller-owned pools. + pub(crate) fn system( + pool: sqlx::PgPool, + identity: crate::provider::ProviderIdentity, + schema: &str, + ) -> Self { + let checkpoint_table = if schema.is_empty() { + "operation_outputs".to_string() + } else { + format!("\"{schema}\".operation_outputs") + }; Self { pool, table: "transaction_completion".to_string(), + checkpoint_table, kind: DataSourceKind::System(identity), } } @@ -363,15 +381,15 @@ impl sealed::Backend for PgDataSource { serialization: &str, started_at_ms: i64, ) -> Result { - // Unqualified: the system pool's per-connection search_path resolves - // the system schema, same as the provider's own queries. - let res = sqlx::query( - "INSERT INTO operation_outputs - (workflow_uuid, function_id, function_name, output, serialization, - started_at_epoch_ms, completed_at_epoch_ms) + // Fully qualified (when the minting provider's schema is known), so + // nothing the body does to `search_path` can redirect the checkpoint. + let res = sqlx::query(&format!( + "INSERT INTO {} (workflow_uuid, function_id, function_name, output, serialization, + started_at_epoch_ms, completed_at_epoch_ms) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (workflow_uuid, function_id) DO NOTHING", - ) + self.checkpoint_table + )) .bind(workflow_id) .bind(step_id) .bind(name) diff --git a/src/postgres.rs b/src/postgres.rs index a2fd5ad..b088e35 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -234,7 +234,7 @@ impl PostgresProvider { /// /// [`Tx`]: crate::Tx pub fn system_datasource(&self) -> crate::PgDataSource { - crate::PgDataSource::system(self.pool.clone(), self.identity.clone()) + crate::PgDataSource::system(self.pool.clone(), self.identity.clone(), &self.schema) } /// Choose the format new values are encoded with. Use [`Serializer::Portable`] diff --git a/tests/datasource.rs b/tests/datasource.rs index 3bdb86b..57b2f7d 100644 --- a/tests/datasource.rs +++ b/tests/datasource.rs @@ -589,6 +589,11 @@ async fn pg_system_datasource_rich_types_single_commit() -> Result<()> { sqlx::query_scalar("SELECT (meta->>'amount')::bigint FROM sys_orders") .fetch_one(&mut *conn) .await?; + // A body that redirects search_path must not redirect the + // checkpoint: the fast-path insert is schema-qualified. + sqlx::query("SET search_path TO public") + .execute(&mut *conn) + .await?; Ok(amount) }) }) From 3459487b9f47fa4fe78363f96625d062d5c5787d Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 14:16:23 -0700 Subject: [PATCH 06/11] docs(transactions): the data source is part of the workflow's deterministic contract --- src/context.rs | 21 +++++++++++++++++++++ src/transactions.rs | 16 +++++++++++++++- 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/src/context.rs b/src/context.rs index c1ff3ca..c1557c5 100644 --- a/src/context.rs +++ b/src/context.rs @@ -665,6 +665,27 @@ impl DurableContext { /// row, no crash window) while the body keeps the native connection — /// unlike [`transaction`](Self::transaction), whose /// [`Param`](crate::Param) bindings cover only a small portable type set. + /// A system data source is bound to the provider that minted it; used + /// under a different engine it is rejected rather than misrouting its + /// checkpoint. + /// + /// # The data source is part of the workflow's contract + /// + /// Which database `ds` points at is invisible to the engine — it cannot + /// tell a right database from a wrong one, and running against the wrong + /// one **succeeds silently**. Two rules keep that from biting: + /// + /// - **Derive `ds` from the workflow's input**, deterministically (e.g. + /// look the tenant up in a map keyed by an input field) — never from + /// ambient state that can disagree with the input, and never captured + /// once at registration for all runs. + /// - **Keep the wiring stable across executions**, exactly like the + /// workflow's code: recovery looks for the witness row in whatever + /// database `ds` points at *now*, so repointing it while runs are + /// in flight (e.g. migrating a tenant's data mid-run) strands the + /// witness and re-runs the body. Drain in-flight workflows before + /// moving a database, or move the `transaction_completion` table with + /// the data. /// /// ```no_run /// # use durare::{DurableContext, PgDataSource, Result}; diff --git a/src/transactions.rs b/src/transactions.rs index d38d2c4..8c0a1f3 100644 --- a/src/transactions.rs +++ b/src/transactions.rs @@ -177,7 +177,21 @@ //! [`PgDataSource`] never takes the fast path, even if its pool happens to //! point at the system database: a wrong "same database" guess would break //! atomicity, so the shortcut is reserved for the case that can't be wrong. -//! This is a durare extension. +//! For the same reason, a system data source used under a *different* engine +//! is rejected rather than misrouting its checkpoint. This is a durare +//! extension. +//! +//! ## The data source is part of the workflow's contract +//! +//! The engine cannot tell a right database from a wrong one — a body run +//! against the wrong tenant's database **succeeds silently**, and recovery +//! looks for the witness row in whatever database the data source points at +//! *now*. So treat the wiring like the workflow's code: derive the data +//! source deterministically from the workflow's input (a lookup keyed by an +//! input field, not a value captured once at registration), and keep it +//! pointing at the same database for the life of every run — drain in-flight +//! workflows before migrating a database, or move `transaction_completion` +//! along with the data. //! //! # Which transaction API? //! From 6a16f62899f00c562c3b38c5e788344670cb3887 Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 14:16:23 -0700 Subject: [PATCH 07/11] test(transactions): keep the search_path sabotage transaction-local --- tests/datasource.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/datasource.rs b/tests/datasource.rs index 57b2f7d..45780e8 100644 --- a/tests/datasource.rs +++ b/tests/datasource.rs @@ -590,8 +590,11 @@ async fn pg_system_datasource_rich_types_single_commit() -> Result<()> { .fetch_one(&mut *conn) .await?; // A body that redirects search_path must not redirect the - // checkpoint: the fast-path insert is schema-qualified. - sqlx::query("SET search_path TO public") + // checkpoint: the fast-path insert (which runs after the + // body, inside this same transaction) is schema-qualified. + // LOCAL so the redirect dies with the transaction instead + // of poisoning the pooled connection for later users. + sqlx::query("SET LOCAL search_path TO public") .execute(&mut *conn) .await?; Ok(amount) From 0c1f63aa61d5ea7619cccbf6cb7fcf00d30f146a Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 15:06:21 -0700 Subject: [PATCH 08/11] fix(transactions): suppress only the key conflict on SQLite datasource inserts --- src/datasource.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/datasource.rs b/src/datasource.rs index 16f928c..4da7e0a 100644 --- a/src/datasource.rs +++ b/src/datasource.rs @@ -526,8 +526,9 @@ impl sealed::Backend for SqliteDataSource { serialization: &str, ) -> Result { let res = sqlx::query(&format!( - "INSERT OR IGNORE INTO transaction_completion ({COMPLETION_COLUMNS}) - VALUES (?, ?, ?, ?, ?, ?)" + "INSERT INTO transaction_completion ({COMPLETION_COLUMNS}) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (workflow_id, step_id) DO NOTHING" )) .bind(workflow_id) .bind(step_id) @@ -548,8 +549,9 @@ impl sealed::Backend for SqliteDataSource { serialization: &str, ) -> Result<()> { sqlx::query(&format!( - "INSERT OR IGNORE INTO transaction_completion ({COMPLETION_COLUMNS}) - VALUES (?, ?, NULL, ?, ?, ?)" + "INSERT INTO transaction_completion ({COMPLETION_COLUMNS}) + VALUES (?, ?, NULL, ?, ?, ?) + ON CONFLICT (workflow_id, step_id) DO NOTHING" )) .bind(workflow_id) .bind(step_id) From 6638e8105d295ceac0170c9d0f14bbaf22d90ffa Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 15:06:42 -0700 Subject: [PATCH 09/11] docs(context): pin the replay-before-identity-check ordering --- src/context.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/context.rs b/src/context.rs index c1557c5..6b20a4f 100644 --- a/src/context.rs +++ b/src/context.rs @@ -786,6 +786,12 @@ impl DurableContext { // data source's pool, which is only this workflow's system database if // the identities match. A mismatch is a wiring bug — fail loudly // rather than splitting the checkpoint from the status row. + // + // This check deliberately sits AFTER the layer-1 replay above: a step + // that already completed replays from its checkpoint even when the + // wiring is now foreign, so recovering finished work is never hostage + // to a configuration change — only a fresh execution is rejected. Do + // not hoist the (cheaper) match above the replay read. match ds.kind() { crate::datasource::DataSourceKind::System(identity) if self From 01b1779ba923aeb90ced363e91527653f214d5fd Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 15:07:35 -0700 Subject: [PATCH 10/11] refactor(provider): make ProviderIdentity construction explicit and pin the Arc allocation invariant --- src/provider.rs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/src/provider.rs b/src/provider.rs index 5714697..f8ca6a0 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -1217,16 +1217,23 @@ pub struct DequeueRequest { /// [`transaction_on`](crate::DurableContext::transaction_on) takes its /// single-commit fast path only against that provider's own database and /// rejects a system data source from a different engine. -#[derive(Clone, Default)] +#[derive(Clone)] pub struct ProviderIdentity(Arc); -#[derive(Default)] +/// INVARIANT: this must stay behind an `Arc`. Identity is the address of the +/// per-instance `ArcInner` allocation — `Arc::new` allocates one even for a +/// zero-sized value, so every identity is distinct and clones compare equal. +/// A `Box`/`Rc`-of-static "simplification" would give every zero-sized +/// instance the same dangling address and make ALL identities match. struct IdentityMarker; +// No `Default`: `default()` conventionally yields one canonical value, but a +// fresh identity is unique by design — two `default()` calls would not match. +#[allow(clippy::new_without_default)] impl ProviderIdentity { /// A fresh identity, equal only to its own clones. pub fn new() -> Self { - Self::default() + Self(Arc::new(IdentityMarker)) } /// Whether `other` was cloned from the same original as `self`. From ef95eea6cdfded31af77063fa502faa1400bdaa4 Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Fri, 7 Aug 2026 15:08:04 -0700 Subject: [PATCH 11/11] docs(changelog): cover the identity binding, duplicate replay, and qualified checkpoint --- CHANGELOG.md | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 25fa7b4..fbb3529 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,7 +17,16 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). arrays, `uuid`, …) that the portable `Param` set can't express. Sameness is established by construction (the provider hands out its own pool), never by detection: a user-constructed `PgDataSource` always uses the two-commit - protocol, which stays correct on any database. + protocol, which stays correct on any database. A system data source is + bound to the provider instance that minted it via the new `ProviderIdentity` + token (`StateProvider::provider_identity`, a defaulted method — custom + providers are unaffected); used under a different engine it is rejected + with an error instead of misrouting its checkpoint. The fast path's + checkpoint insert is schema-qualified, so nothing a body does to + `search_path` can redirect it, and a duplicate execution that loses the + checkpoint race is rolled back — its writes discarded — and replays the + canonical outcome, keeping the step exactly-once even under double + execution. - Declarative role-based authorization: `require_roles(name, roles)` on the engine and builder declares the roles a caller must hold to invoke a