Struct DbOp 

pub struct DbOp<'c> { /* private fields */ }

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

impl<'c> DbOp<'c>

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.

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.

pub fn with_time(self, time: DateTime<Utc>) -> DbOpWithTime<'c>

Transitions to a DbOpWithTime with the given time cached.

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.

pub async fn with_db_time(self) -> Result<DbOpWithTime<'c>, Error>

Transitions to a DbOpWithTime using the database time.

Priority order:

1. Cached time if present
2. Manual clock time if the clock is manual
3. Database time via SELECT NOW()

pub fn maybe_now(&self) -> Option<DateTime<Utc>>

Returns the optionally cached chrono::DateTime

pub async fn begin(&mut self) -> Result<DbOp<'_>, Error>

Begins a nested transaction.

pub async fn with_savepoint<T, E, F>( &mut self, f: F, ) -> Result<Result<T, E>, Error>

where F: AsyncFnOnce(&mut SavepointOp<'_>) -> Result<T, E>,

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.

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.

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_commit fails → the transaction is rolled back first, then the earlier (already-pre_committed) hooks are notified.
• The COMMIT itself 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).

pub fn tx_mut(&mut self) -> &mut Transaction<'c, Db>

Gets a mutable handle to the inner transaction

Trait Implementations

impl<'o> AtomicOperation for DbOp<'o>

fn maybe_now(&self) -> Option<DateTime<Utc>>

Function for querying when the operation is taking place - if it is cached.

fn clock(&self) -> &ClockHandle

Returns the clock handle for time operations.

fn connection(&mut self) -> &mut Connection

Returns the raw underlying connection. The desired way to represent this would actually be as a GAT:

fn add_commit_hook<H: CommitHook>(&mut self, hook: H) -> Result<(), H>

Registers a commit hook that will run pre_commit before and post_commit after the transaction commits. Returns Ok(()) if the hook was registered, Err(hook) if hooks are not supported.

fn commit_hook<H: CommitHook>(&self) -> Option<&H>

Typed shared access to the currently-accumulating commit hook of type H, if this operation supports commit hooks and one is registered. Returns the hook a subsequent add_commit_hook::<H> call would merge into.

fn supports_hooks(&self) -> bool

Whether this operation supports commit hooks.

fn as_executor(&mut self) -> OneTimeExecutor<'_, &mut Connection>

Returns the sqlx::Executor implementation that statements should be executed through.