pub struct DbOp<'c> { /* private fields */ }Expand description
Default return type of the derived EsRepo::begin_op().
Used as a wrapper of a sqlx::Transaction but can also cache the time at which the
transaction is taking place.
When a manual clock is provided, the transaction will automatically cache that clock’s time, enabling deterministic testing. This cached time will be used in all time-dependent operations.
Implementations§
Source§impl<'c> DbOp<'c>
impl<'c> DbOp<'c>
Sourcepub async fn init(pool: &Pool) -> Result<DbOp<'static>, Error>
pub async fn init(pool: &Pool) -> Result<DbOp<'static>, Error>
Initializes a transaction using the global clock.
Delegates to init_with_clock using the global clock handle.
Sourcepub async fn init_with_clock(
pool: &Pool,
clock: &ClockHandle,
) -> Result<DbOp<'static>, Error>
pub async fn init_with_clock( pool: &Pool, clock: &ClockHandle, ) -> Result<DbOp<'static>, Error>
Initializes a transaction with the specified clock.
If the clock is manual, its current time will be cached in the transaction.
Sourcepub fn with_time(self, time: DateTime<Utc>) -> DbOpWithTime<'c>
pub fn with_time(self, time: DateTime<Utc>) -> DbOpWithTime<'c>
Transitions to a DbOpWithTime with the given time cached.
Sourcepub fn with_clock_time(self) -> DbOpWithTime<'c>
pub fn with_clock_time(self) -> DbOpWithTime<'c>
Transitions to a DbOpWithTime using the clock.
Uses cached time if present, otherwise uses the clock’s current time.
Sourcepub async fn with_db_time(self) -> Result<DbOpWithTime<'c>, Error>
pub async fn with_db_time(self) -> Result<DbOpWithTime<'c>, Error>
Transitions to a DbOpWithTime using the database time.
Priority order:
- Cached time if present
- Manual clock time if the clock is manual
- Database time via
SELECT NOW()
Sourcepub fn maybe_now(&self) -> Option<DateTime<Utc>>
pub fn maybe_now(&self) -> Option<DateTime<Utc>>
Returns the optionally cached chrono::DateTime
Sourcepub async fn with_savepoint<T, E, F>(
&mut self,
f: F,
) -> Result<Result<T, E>, Error>
pub async fn with_savepoint<T, E, F>( &mut self, f: F, ) -> Result<Result<T, E>, Error>
Runs f inside a SAVEPOINT, keeping its work on Ok and undoing it on Err.
This is the building block for processing a batch of items in one
transaction — one COMMIT, one WAL flush — while still isolating each
item’s failure. An item that errors unwinds only its own writes and
staged commit hooks; the transaction stays usable, so the loop continues
and its healthy items still commit.
§Two layers of Result
- The outer
Err(sqlx::Error)means the savepoint machinery itself failed (or the error was never savepoint-recoverable, e.g. the connection died). The parent operation is in an indeterminate state: abandon it, don’t commit. - The inner
Err(E)is the item’s own failure, already rolled back cleanly. Record the outcome and keep going.
If the closure fails and the rollback fails, the rollback error is
returned as the outer Err and the item’s error is dropped — the
poisoned-transaction signal is what the caller must act on.
§Collecting per-item outcomes
The closure may borrow from its environment, but host-side mutations do
not unwind with the savepoint. Return the item’s verdict through
Ok/Err and record it outside, where the outcome is authoritative:
let mut op = DbOp::init(&pool).await?;
let mut outcomes = Vec::with_capacity(items.len());
for item in items {
// `?` here: infra failure — abandon the whole batch.
let res = op
.with_savepoint(async |op| self.process_in_op(op, item).await)
.await?;
outcomes.push(match res {
Ok(()) => Outcome::Complete,
Err(e) => Outcome::Retry(e),
});
}
op.commit().await?;See SavepointOp for how commit hooks are staged and folded in.
Kept as an inherent method so existing call sites need no import; the
behaviour lives in SavepointOperation::with_savepoint, which every
AtomicOperation gets.
Sourcepub async fn begin_savepoint(&mut self) -> Result<SavepointOp<'_>, Error>
pub async fn begin_savepoint(&mut self) -> Result<SavepointOp<'_>, Error>
Begins a SAVEPOINT scope explicitly.
The escape hatch for when with_savepoint’s
closure form doesn’t fit — the returned SavepointOp must be finished
with release or
rollback. Dropping it rolls back.
Sourcepub async fn commit(self) -> Result<(), Error>
pub async fn commit(self) -> Result<(), Error>
Commits the inner transaction.
On the failure paths the commit hooks’ on_rollback runs after the
transaction is definitively gone, so hook-side compensation never
contends with the dying transaction’s own locks:
- A later hook’s
pre_commitfails → the transaction is rolled back first, then the earlier (already-pre_committed) hooks are notified. - The
COMMITitself fails → the transaction is over server-side either way, so the hooks are notified directly (their side effects must be idempotent against a possibly-landed commit).
Sourcepub fn tx_mut(&mut self) -> &mut Transaction<'c, Db>
pub fn tx_mut(&mut self) -> &mut Transaction<'c, Db>
Gets a mutable handle to the inner transaction
Trait Implementations§
Source§impl<'o> AtomicOperation for DbOp<'o>
impl<'o> AtomicOperation for DbOp<'o>
Source§fn savepoint_parts(&mut self) -> (&mut Connection, HookSlot<'_>)
fn savepoint_parts(&mut self) -> (&mut Connection, HookSlot<'_>)
tx and commit_hooks are disjoint fields, so both can be borrowed
mutably in one expression — the borrow split that a pair of &mut self
accessors could not express, which is the whole reason this method
returns both halves at once.
Source§fn maybe_now(&self) -> Option<DateTime<Utc>>
fn maybe_now(&self) -> Option<DateTime<Utc>>
Source§fn clock(&self) -> &ClockHandle
fn clock(&self) -> &ClockHandle
Source§fn connection(&mut self) -> &mut Connection
fn connection(&mut self) -> &mut Connection
Source§fn add_commit_hook<H: CommitHook>(&mut self, hook: H) -> Result<(), H>
fn add_commit_hook<H: CommitHook>(&mut self, hook: H) -> Result<(), H>
Source§fn commit_hook<H: CommitHook>(&self) -> Option<&H>
fn commit_hook<H: CommitHook>(&self) -> Option<&H>
H,
if this operation supports commit hooks and one is registered.
Returns the hook a subsequent add_commit_hook::<H> call would merge into.Source§fn supports_hooks(&self) -> bool
fn supports_hooks(&self) -> bool
Source§fn as_executor(&mut self) -> OneTimeExecutor<'_, &mut Connection>
fn as_executor(&mut self) -> OneTimeExecutor<'_, &mut Connection>
sqlx::Executor implementation that statements should be
executed through. Read moreAuto Trait Implementations§
impl<'c> !RefUnwindSafe for DbOp<'c>
impl<'c> !Sync for DbOp<'c>
impl<'c> !UnwindSafe for DbOp<'c>
impl<'c> Freeze for DbOp<'c>
impl<'c> Send for DbOp<'c>
impl<'c> Unpin for DbOp<'c>
impl<'c> UnsafeUnpin for DbOp<'c>
Blanket Implementations§
Source§impl<T> BatchIsolation for Twhere
T: SavepointOperation + ?Sized,
impl<T> BatchIsolation for Twhere
T: SavepointOperation + ?Sized,
Source§fn run_isolated<'a, T, V, E, F>(
&'a mut self,
items: &'a [T],
f: F,
) -> impl Future<Output = Result<Vec<Result<V, E>>, Error>> + 'awhere
T: 'a,
V: 'a,
E: 'a,
F: AsyncFnOnce(&mut SavepointOp<'_>, &T) -> Result<V, E> + Clone + Sync + 'a,
fn run_isolated<'a, T, V, E, F>(
&'a mut self,
items: &'a [T],
f: F,
) -> impl Future<Output = Result<Vec<Result<V, E>>, Error>> + 'awhere
T: 'a,
V: 'a,
E: 'a,
F: AsyncFnOnce(&mut SavepointOp<'_>, &T) -> Result<V, E> + Clone + Sync + 'a,
Source§fn run_bisected<'a, T, E, F>(
&'a mut self,
items: &'a [T],
budget: BisectBudget,
f: F,
) -> impl Future<Output = Result<BisectOutcomes<E>, Error>> + 'awhere
T: 'a,
E: Error + 'static,
F: AsyncFnOnce(&mut SavepointOp<'_>, &[T]) -> Result<(), E> + Clone + Sync + 'a,
fn run_bisected<'a, T, E, F>(
&'a mut self,
items: &'a [T],
budget: BisectBudget,
f: F,
) -> impl Future<Output = Result<BisectOutcomes<E>, Error>> + 'awhere
T: 'a,
E: Error + 'static,
F: AsyncFnOnce(&mut SavepointOp<'_>, &[T]) -> Result<(), E> + Clone + Sync + 'a,
Source§fn run_bisected_with<'a, T, E, F, P>(
&'a mut self,
items: &'a [T],
budget: BisectBudget,
policy: TransientPolicy<P>,
f: F,
) -> impl Future<Output = Result<BisectOutcomes<E>, Error>> + 'a
fn run_bisected_with<'a, T, E, F, P>( &'a mut self, items: &'a [T], budget: BisectBudget, policy: TransientPolicy<P>, f: F, ) -> impl Future<Output = Result<BisectOutcomes<E>, Error>> + 'a
run_bisected with a caller-supplied notion of
which failures are transient. Read moreSource§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
impl<T> BorrowMut<T> for Twhere
T: ?Sized,
Source§fn borrow_mut(&mut self) -> &mut T
fn borrow_mut(&mut self) -> &mut T
Source§impl<T> Instrument for T
impl<T> Instrument for T
Source§fn instrument(self, span: Span) -> Instrumented<Self> ⓘ
fn instrument(self, span: Span) -> Instrumented<Self> ⓘ
Source§fn in_current_span(self) -> Instrumented<Self> ⓘ
fn in_current_span(self) -> Instrumented<Self> ⓘ
Source§impl<T> IntoEither for T
impl<T> IntoEither for T
Source§fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ
fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ
self into a Left variant of Either<Self, Self>
if into_left is true.
Converts self into a Right variant of Either<Self, Self>
otherwise. Read moreSource§fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
self into a Left variant of Either<Self, Self>
if into_left(&self) returns true.
Converts self into a Right variant of Either<Self, Self>
otherwise. Read moreSource§impl<T> SavepointOperation for Twhere
T: AtomicOperation + ?Sized,
impl<T> SavepointOperation for Twhere
T: AtomicOperation + ?Sized,
Source§fn with_savepoint<T, E, F>(
&mut self,
f: F,
) -> impl Future<Output = Result<Result<T, E>, Error>>
fn with_savepoint<T, E, F>( &mut self, f: F, ) -> impl Future<Output = Result<Result<T, E>, Error>>
f inside a SAVEPOINT, keeping its work on Ok and undoing it on
Err — see DbOp::with_savepoint for the
full contract, including the two layers of Result. Read moreSource§fn begin_savepoint(
&mut self,
) -> impl Future<Output = Result<SavepointOp<'_>, Error>> + Send
fn begin_savepoint( &mut self, ) -> impl Future<Output = Result<SavepointOp<'_>, Error>> + Send
SAVEPOINT scope explicitly — see
DbOp::begin_savepoint. Must be finished
with release or
rollback; dropping it rolls back. Read more