diff --git a/CHANGELOG.md b/CHANGELOG.md index c4bad13..fbb3529 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,26 @@ 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. 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 workflow. The check runs before the body on every execution path (direct, diff --git a/src/context.rs b/src/context.rs index 9c59eed..6b20a4f 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 @@ -649,8 +656,36 @@ 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. + /// 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}; @@ -742,6 +777,41 @@ 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. 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. + // + // 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 + .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 // transaction committed but the run crashed before the system commit. @@ -752,8 +822,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 +941,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 +969,144 @@ 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(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); + } + 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. + /// `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, + 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)?; + 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(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. @@ -1803,6 +2005,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..4da7e0a 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::*; @@ -90,6 +103,29 @@ pub(crate) mod sealed { error: &str, serialization: &str, ) -> Result<()>; + + /// 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 + /// 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, + conn: &mut Self::Conn, + workflow_id: &str, + step_id: i32, + name: &str, + output: &str, + serialization: &str, + started_at_ms: i64, + ) -> Result; } } @@ -140,6 +176,12 @@ const COMPLETION_COLUMNS: &str = "workflow_id, step_id, output, error, serializa pub struct PgDataSource { pool: sqlx::PgPool, table: String, + /// 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")] @@ -176,7 +218,37 @@ impl PgDataSource { )) .execute(&pool) .await?; - Ok(Self { pool, table }) + Ok(Self { + 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. `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), + } } /// The pool this data source runs on. @@ -294,6 +366,41 @@ impl sealed::Backend for PgDataSource { .await?; Ok(()) } + + fn kind(&self) -> &DataSourceKind { + &self.kind + } + + 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 { + // 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) + .bind(output) + .bind(serialization) + .bind(started_at_ms) + .bind(chrono::Utc::now().timestamp_millis()) + .execute(conn) + .await?; + Ok(res.rows_affected() == 1) + } } /// A [`DataSource`] over a SQLite application database. @@ -312,6 +419,8 @@ impl sealed::Backend for PgDataSource { #[derive(Clone)] pub struct SqliteDataSource { pool: sqlx::SqlitePool, + /// What this data source points at — see [`PgDataSource`]'s `kind` field. + kind: DataSourceKind, } #[cfg(feature = "sqlite")] @@ -332,7 +441,22 @@ impl SqliteDataSource { ) .execute(&pool) .await?; - Ok(Self { pool }) + Ok(Self { + pool, + 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, + identity: crate::provider::ProviderIdentity, + ) -> Self { + Self { + pool, + kind: DataSourceKind::System(identity), + } } /// The pool this data source runs on. @@ -402,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) @@ -424,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) @@ -436,4 +562,37 @@ impl sealed::Backend for SqliteDataSource { .await?; Ok(()) } + + fn kind(&self) -> &DataSourceKind { + &self.kind + } + + 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 { + let res = 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(res.rows_affected() == 1) + } } 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 bb9bbed..b088e35 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(), } } @@ -202,6 +205,38 @@ 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. + /// + /// A **durare extension**. + /// + /// # 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(), self.identity.clone(), &self.schema) + } + /// 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 { @@ -387,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..f8ca6a0 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,46 @@ 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)] +pub struct ProviderIdentity(Arc); + +/// 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(Arc::new(IdentityMarker)) + } + + /// 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 +1280,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 6c0714e..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,9 +71,23 @@ impl SqliteProvider { Self { pool, serializer: Serializer::default(), + identity: crate::provider::ProviderIdentity::new(), } } + /// 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. 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). + pub fn system_datasource(&self) -> crate::SqliteDataSource { + crate::SqliteDataSource::system(self.pool.clone(), self.identity.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 @@ -162,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/src/transactions.rs b/src/transactions.rs index d3d79a9..8c0a1f3 100644 --- a/src/transactions.rs +++ b/src/transactions.rs @@ -138,10 +138,75 @@ //! 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::{DurableContext, DurableEngine, PostgresProvider, Result}; +//! # use std::sync::Arc; +//! # 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?; +//! +//! 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(()) } +//! ``` +//! +//! 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. +//! 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? +//! +//! | 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 //! diff --git a/tests/datasource.rs b/tests/datasource.rs index 6961d1e..45780e8 100644 --- a/tests/datasource.rs +++ b/tests/datasource.rs @@ -409,3 +409,367 @@ 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?; + // A body that redirects search_path must not redirect the + // 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) + }) + }) + .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(()) +} + +/// 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(()) +} + +/// 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() +}