es_entity/operation/savepoint.rs
1//! Savepoint-scoped operations for per-item isolation inside a single transaction.
2
3use sqlx::{Acquire, Transaction};
4
5use crate::{clock::ClockHandle, db};
6
7use super::{AtomicOperation, hooks};
8
9/// An [`AtomicOperation`] scoped to a database `SAVEPOINT` inside a parent [`DbOp`].
10///
11/// Created by [`DbOp::with_savepoint`] / [`DbOp::begin_savepoint`]. Statements
12/// executed through it run inside the savepoint, so they can be undone with
13/// [`rollback`](Self::rollback) without poisoning — or ending — the parent
14/// transaction. This is what makes a "loop over N items in one transaction, but
15/// isolate each item's failure" pattern possible: one `COMMIT` (one WAL flush)
16/// for the whole batch, while a failing item unwinds only its own writes.
17///
18/// # Hooks are staged, not executed
19///
20/// Commit hooks registered on a `SavepointOp` — including the ones repositories
21/// register internally, e.g. via `post_persist_hook` — are **staged** in a
22/// private buffer rather than added to the parent operation:
23///
24/// - [`release`](Self::release) issues `RELEASE SAVEPOINT` and then folds the
25/// staged hooks into the parent's buffer through the ordinary
26/// registration/merge path, exactly as if they had been registered on the
27/// parent directly.
28/// - [`rollback`](Self::rollback) issues `ROLLBACK TO SAVEPOINT` and drops the
29/// staged hooks. A rolled-back item therefore contributes zero hook state to
30/// match its zero database state — no phantom event publishes, no inserts
31/// referencing rows that no longer exist.
32///
33/// No hook's [`pre_commit`] / [`post_commit`] ever runs at savepoint boundaries.
34/// They run once, at the parent's [`commit`](DbOp::commit), over the final merged
35/// hook set — so `post_commit` still only fires after a durable `COMMIT`, and
36/// [`on_rollback`] still only fires when the whole transaction is gone.
37///
38/// [`DbOp`]: super::DbOp
39/// [`DbOp::with_savepoint`]: super::DbOp::with_savepoint
40/// [`DbOp::begin_savepoint`]: super::DbOp::begin_savepoint
41/// [`pre_commit`]: hooks::CommitHook::pre_commit
42/// [`post_commit`]: hooks::CommitHook::post_commit
43/// [`on_rollback`]: hooks::CommitHook::on_rollback
44pub struct SavepointOp<'t> {
45 tx: Transaction<'t, db::Db>,
46 clock: ClockHandle,
47 now: Option<chrono::DateTime<chrono::Utc>>,
48 /// Hooks registered while the savepoint is open. Folded into
49 /// `parent_hooks` on release, dropped on rollback.
50 staged: hooks::CommitHooks,
51 /// The parent operation's hook buffer. Held as a `&mut` to a *disjoint*
52 /// field of the parent (the nested transaction borrows its `tx` field), so
53 /// the savepoint can fold into it without the parent's hooks ever leaving
54 /// the parent.
55 parent_hooks: &'t mut Option<hooks::CommitHooks>,
56}
57
58impl<'t> SavepointOp<'t> {
59 pub(super) async fn begin(
60 tx: &'t mut Transaction<'_, db::Db>,
61 clock: ClockHandle,
62 now: Option<chrono::DateTime<chrono::Utc>>,
63 parent_hooks: &'t mut Option<hooks::CommitHooks>,
64 ) -> Result<Self, sqlx::Error> {
65 Ok(Self {
66 tx: tx.begin().await?,
67 clock,
68 now,
69 staged: hooks::CommitHooks::new(),
70 parent_hooks,
71 })
72 }
73
74 /// Releases the savepoint, keeping this scope's work.
75 ///
76 /// Issues `RELEASE SAVEPOINT`, then folds the staged commit hooks into the
77 /// parent operation's buffer via the normal registration/merge path — so a
78 /// mergeable hook type accumulates across savepoints exactly as it would
79 /// have on the parent, and a non-mergeable one lands at its own position in
80 /// release order.
81 ///
82 /// If the `RELEASE` itself fails the staged hooks are dropped and the error
83 /// is returned: the parent transaction is in an indeterminate state and the
84 /// caller must abandon it rather than commit.
85 pub async fn release(self) -> Result<(), sqlx::Error> {
86 let Self {
87 tx,
88 staged,
89 parent_hooks,
90 ..
91 } = self;
92 tx.commit().await?;
93 parent_hooks
94 .as_mut()
95 .expect("no hooks")
96 .absorb_staged(staged);
97 Ok(())
98 }
99
100 /// Rolls back to the savepoint, discarding this scope's work.
101 ///
102 /// Issues `ROLLBACK TO SAVEPOINT` and drops the staged commit hooks. The
103 /// parent transaction stays alive and usable — including after an error
104 /// that would otherwise have poisoned it.
105 ///
106 /// Dropping a `SavepointOp` without calling either `release` or `rollback`
107 /// has the same database effect (sqlx queues the rollback on the
108 /// connection) and likewise discards the staged hooks.
109 pub async fn rollback(self) -> Result<(), sqlx::Error> {
110 self.tx.rollback().await
111 }
112}
113
114impl AtomicOperation for SavepointOp<'_> {
115 fn maybe_now(&self) -> Option<chrono::DateTime<chrono::Utc>> {
116 self.now
117 }
118
119 fn clock(&self) -> &ClockHandle {
120 &self.clock
121 }
122
123 fn connection(&mut self) -> &mut db::Connection {
124 self.tx.connection()
125 }
126
127 fn add_commit_hook<H: hooks::CommitHook>(&mut self, hook: H) -> Result<(), H> {
128 self.staged.add(hook);
129 Ok(())
130 }
131
132 /// Reads the staged buffer first, falling back to the parent's.
133 ///
134 /// While a savepoint is open the two are not yet merged, so a hook type
135 /// present in *both* reports only the staged instance here — they become
136 /// one hook when the savepoint is released.
137 fn commit_hook<H: hooks::CommitHook>(&self) -> Option<&H> {
138 self.staged
139 .get_last::<H>()
140 .or_else(|| self.parent_hooks.as_ref()?.get_last::<H>())
141 }
142
143 fn supports_hooks(&self) -> bool {
144 true
145 }
146}