cratestack_sqlx/
isolated_run.rs

1//! [`SqlxRuntime::run_isolated`]: the retry loop an `@isolation`
2//! procedure's generated `invoke_with_db` runs its authorization and body
3//! in (docs/design/procedure-isolation.md §5, GHSA-r67q-4qqq-g9gm).
4
5use std::future::Future;
6use std::sync::Arc;
7use std::time::Duration;
8
9use cratestack_core::{CratestackError, DbErrorInfo, TransactionAbort, TransactionIsolation};
10
11use crate::audit::dispatch_audit_sink;
12use crate::bound::BoundTx;
13use crate::descriptor::SqlxRuntime;
14use crate::error::cratestack_error_from_sqlx;
15use crate::retriable::retriable_sqlstate;
16use crate::sqlx;
17use crate::transaction::Tx;
18
19/// Public detail of the error an exhausted retry budget produces.
20const ISOLATION_CONFLICT_DETAIL: &str =
21    "transaction could not be completed because of concurrent updates; retry the request";
22
23enum Attempt<T> {
24    Committed(T),
25    Retry(&'static str),
26    Failed(CratestackError),
27}
28
29impl SqlxRuntime {
30    /// The transaction this runtime is bound to, when it is the
31    /// per-attempt runtime of an `@isolation` procedure.
32    pub(crate) fn bound(&self) -> Option<&BoundTx> {
33        self.bound.as_deref()
34    }
35
36    /// Retries an `@isolation` procedure gets after a serialization
37    /// failure (`40001`) or detected deadlock (`40P01`) before the
38    /// request fails with `409 TRANSACTION_ABORTED`. Default 3; `0`
39    /// disables retry.
40    pub fn with_isolation_max_retries(mut self, max_retries: u32) -> Self {
41        self.isolation_max_retries = max_retries;
42        self
43    }
44
45    /// Run `body` inside one transaction begun at `isolation`, committing
46    /// on `Ok`. `body` receives (by value, so the future it returns can own
47    /// it) a runtime bound to that transaction: every
48    /// builder, `transaction()` and `@authorize` probe made through it
49    /// runs on the transaction. On `40001`/`40P01` — from a statement,
50    /// from `COMMIT`, or observed by any operation during the attempt even
51    /// if `body` swallowed it — the attempt is rolled back and `body` runs
52    /// again after a short backoff, up to the configured retry budget,
53    /// then fails with `TransactionAborted` (409 `TRANSACTION_ABORTED`).
54    /// `AuditSink` fan-out and the
55    /// outbox drain the attempt's writes requested happen once, after the
56    /// committed attempt.
57    ///
58    /// A `TransactionAborted` returned *by `body`* — another transaction's
59    /// exhausted retries, propagated — is not retried (it is final) and is
60    /// marked [`AbortOwnership::Propagated`](cratestack_core::AbortOwnership)
61    /// so no dispatch can claim it as its own; only this loop's own
62    /// exhaustion is returned as `Exhausted`.
63    ///
64    /// On a runtime that is already bound to an attempt (a nested
65    /// `@isolation` call), `body` runs once as a savepoint of that attempt
66    /// instead of in a new transaction, and the outermost attempt owns the
67    /// retries (`bound::join_bound`, docs/design/procedure-isolation.md
68    /// §7.1).
69    ///
70    /// Generated code calls this; it is public only so the facade crates
71    /// can reach it.
72    #[doc(hidden)]
73    pub async fn run_isolated<F, Fut, T>(
74        &self,
75        isolation: TransactionIsolation,
76        mut body: F,
77    ) -> Result<T, CratestackError>
78    where
79        F: FnMut(SqlxRuntime) -> Fut,
80        Fut: Future<Output = Result<T, CratestackError>>,
81    {
82        if let Some(bound) = self.bound() {
83            return crate::bound::join_bound(self, bound, isolation, body).await;
84        }
85        let mut attempt = 0u32;
86        loop {
87            attempt += 1;
88            let begin = format!("BEGIN ISOLATION LEVEL {}", isolation.as_sql());
89            // `AssertSqlSafe`: `as_sql()` is one of three `&'static str`
90            // literals; nothing request-derived reaches this statement.
91            let tx = self
92                .pool()
93                .begin_with(sqlx::AssertSqlSafe(begin))
94                .await
95                .map_err(cratestack_error_from_sqlx)?;
96            let bound = Arc::new(BoundTx::new(Tx::new(tx), isolation));
97            let mut runtime = self.clone();
98            runtime.bound = Some(bound.clone());
99            let result = body(runtime)
100                .await
101                .map_err(CratestackError::propagate_transaction_abort);
102            match finish_attempt(&bound, result).await {
103                Attempt::Committed(value) => {
104                    let deferred = bound.take_deferred();
105                    if deferred.drain_outbox {
106                        let _ = self.drain_event_outbox().await;
107                    }
108                    dispatch_audit_sink(self, &deferred.audit_events).await;
109                    deferred.publish().await;
110                    return Ok(value);
111                }
112                Attempt::Retry(sqlstate) if attempt <= self.isolation_max_retries => {
113                    tracing::debug!(
114                        target: "cratestack",
115                        cratestack_isolation = isolation.as_sql(),
116                        cratestack_sqlstate = sqlstate,
117                        cratestack_attempt = attempt,
118                        "retrying @isolation transaction",
119                    );
120                    backoff(attempt).await;
121                }
122                Attempt::Retry(sqlstate) => {
123                    return Err(CratestackError::TransactionAborted(
124                        TransactionAbort::exhausted(DbErrorInfo {
125                            detail: ISOLATION_CONFLICT_DETAIL.to_owned(),
126                            sqlstate: Some(sqlstate.to_owned()),
127                            constraint: None,
128                        }),
129                    ));
130                }
131                Attempt::Failed(error) => return Err(error),
132            }
133        }
134    }
135}
136
137async fn finish_attempt<T>(bound: &BoundTx, result: Result<T, CratestackError>) -> Attempt<T> {
138    let Some(tx) = bound.take().await else {
139        return Attempt::Failed(CratestackError::Internal(
140            "@isolation transaction missing at commit".to_owned(),
141        ));
142    };
143    let tx = tx.into_inner();
144    let taint = bound.tainted();
145    // A body that returned `Ok` over a transaction `db.transaction(..)`
146    // could not close cleanly has not succeeded: its `COMMIT` would be a
147    // silent `ROLLBACK` (aborted) or a no-op (already ended).
148    let result = match (result, bound.take_poison()) {
149        (Ok(_), Some(poison)) => Err(poison),
150        (result, _) => result,
151    };
152    match result {
153        Ok(value) if taint.is_none() => match tx.commit().await {
154            Ok(()) => Attempt::Committed(value),
155            Err(error) => {
156                let error = cratestack_error_from_sqlx(error);
157                match retriable_sqlstate(&error) {
158                    Some(sqlstate) => Attempt::Retry(sqlstate),
159                    None => Attempt::Failed(error),
160                }
161            }
162        },
163        Ok(_) => {
164            let _ = tx.rollback().await;
165            Attempt::Retry(taint.unwrap_or("40001"))
166        }
167        Err(error) => {
168            let _ = tx.rollback().await;
169            match retriable_sqlstate(&error).or(taint) {
170                Some(sqlstate) => Attempt::Retry(sqlstate),
171                None => Attempt::Failed(error),
172            }
173        }
174    }
175}
176
177/// `2ms × 2^(retry-1)`, capped at 64 ms, plus up to as much again of
178/// jitter so two contenders that just collided do not retry in lockstep.
179async fn backoff(retry: u32) {
180    let base_ms = (2u64 << retry.saturating_sub(1).min(5)).min(64);
181    let nanos = std::time::SystemTime::now()
182        .duration_since(std::time::UNIX_EPOCH)
183        .map(|elapsed| u64::from(elapsed.subsec_nanos()))
184        .unwrap_or(0);
185    let jitter = Duration::from_nanos(nanos % (base_ms * 1_000_000 + 1));
186    tokio::time::sleep(Duration::from_millis(base_ms) + jitter).await;
187}
cratestack_sqlx/isolated_run.rs

cratestack_sqlx/
isolated_run.rs
