1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
use std::marker::PhantomData;
use es_entity::hooks::{CommitHook, HookOperation, PreCommitRet};
use serde::{Serialize, de::DeserializeOwned};
use tokio::sync::{broadcast, mpsc};
use crate::out::event::{PersistentDelivery, PersistentOutboxEvent};
use crate::out::gap_fill::GapFillRequest;
use crate::out::post_persist_hook::PostPersistHooks;
use crate::sequence::EventSequence;
use crate::tables::MailboxTables;
pub struct PersistEvents<P, Tables>
where
P: Serialize + DeserializeOwned + Send + Sync + 'static + Unpin,
Tables: MailboxTables,
{
sender: broadcast::Sender<PersistentDelivery<P>>,
/// Reports the committed batch's `(min, max)` to the debounced notifier.
notifier_tx: mpsc::UnboundedSender<(EventSequence, EventSequence)>,
/// Reports sequences this operation allocated but failed to commit to
/// the [`GapFiller`](crate::out::gap_fill::GapFiller): see
/// [`CommitHook::on_rollback`] below and the own-failure branches in
/// `pre_commit`.
abandoned_tx: mpsc::UnboundedSender<GapFillRequest>,
pre_commit_events: Vec<P>,
/// Persisted events, stashed chunk-by-chunk *as* `pre_commit` runs (not
/// at its end): if a later chunk, a later hook, or the COMMIT itself
/// fails, this holds exactly the sequences the rolled-back transaction
/// allocated — which is what `on_rollback` (or the own-failure
/// branches) reports for compensation. Consumed (emptied) by
/// `post_commit` on the success path.
post_commit_events: Vec<PersistentOutboxEvent<P>>,
batch_size: usize,
/// Snapshot of the outbox's registered post-persist hooks, taken when
/// this commit hook is constructed (i.e. at the operation's first
/// publish). Merged publishes keep the first snapshot.
post_persist_hooks: PostPersistHooks<P>,
/// Set on the force-execute path: `post_commit` never runs, so the
/// persist statement must carry the in-tx NOTIFY. Also suppresses
/// own-failure compensation — on that path the caller's bare transaction
/// is still open when `pre_commit` returns and its fate is unknowable
/// (it may savepoint-recover and commit), so those sequences are the
/// proof-gated backstop's responsibility.
///
/// Narrower scope than the name once implied: with es-entity's
/// re-entrant hook registration, this only fires for an operation that
/// genuinely has no commit pass to join (a bare `sqlx::Transaction`).
/// A repost from inside another outbox's [`PostPersistHook`] no longer
/// takes this path — `add_commit_hook` now succeeds there, so the
/// repost joins the enclosing commit pass and gets the full
/// `post_commit`/`on_rollback` lifecycle instead.
///
/// [`PostPersistHook`]: crate::out::PostPersistHook
notify_in_tx: bool,
_phantom: PhantomData<Tables>,
}
impl<P, Tables> PersistEvents<P, Tables>
where
P: Serialize + DeserializeOwned + Send + Sync + 'static + Unpin,
Tables: MailboxTables,
{
pub fn new(
sender: broadcast::Sender<PersistentDelivery<P>>,
notifier_tx: mpsc::UnboundedSender<(EventSequence, EventSequence)>,
abandoned_tx: mpsc::UnboundedSender<GapFillRequest>,
events: impl IntoIterator<Item = impl Into<P>>,
batch_size: usize,
post_persist_hooks: PostPersistHooks<P>,
) -> Self {
Self {
sender,
notifier_tx,
abandoned_tx,
pre_commit_events: events.into_iter().map(Into::into).collect(),
post_commit_events: Vec::new(),
batch_size,
post_persist_hooks,
notify_in_tx: false,
_phantom: PhantomData,
}
}
/// Use the in-transaction NOTIFY persist variant — for the force-execute
/// path only (see the `notify_in_tx` field doc above for its narrowed
/// scope: genuinely hookless operations, not reposts).
pub(crate) fn with_in_tx_notify(mut self) -> Self {
self.notify_in_tx = true;
self
}
/// Events buffered on this hook, awaiting persistence at commit.
/// Backs the [`Outbox::cursor`](crate::out::Outbox::cursor) read API.
pub(crate) fn pending(&self) -> &[P] {
&self.pre_commit_events
}
/// Own-failure compensation: report the sequences persisted so far to
/// the [`GapFiller`](crate::out::gap_fill::GapFiller) before
/// `pre_commit` returns its error. The transaction is still open here
/// (its rollback follows once the error propagates), so the report
/// must stay a channel send — the GapFiller's insert parks briefly
/// on the dying transaction's speculative-insertion locks and resolves
/// when the rollback lands; awaiting that insert inline would deadlock
/// on our own transaction. Skipped on the force-execute path
/// (`notify_in_tx`), where the caller's transaction may yet recover
/// and commit.
fn report_own_failure(&mut self) {
if self.notify_in_tx {
return;
}
let abandoned: Vec<EventSequence> = self
.post_commit_events
.drain(..)
.map(|event| event.sequence)
.collect();
if !abandoned.is_empty() {
let _ = self.abandoned_tx.send(GapFillRequest::Abandoned(abandoned));
}
}
}
impl<P, Tables> CommitHook for PersistEvents<P, Tables>
where
P: Serialize + DeserializeOwned + Send + Sync + 'static + Unpin,
Tables: MailboxTables,
{
async fn pre_commit(
mut self,
mut op: HookOperation<'_>,
) -> Result<PreCommitRet<'_, Self>, sqlx::Error> {
let batch_size = self.batch_size.max(1);
let events = std::mem::take(&mut self.pre_commit_events);
self.post_commit_events.reserve(events.len());
let mut events = events.into_iter();
loop {
let chunk: Vec<P> = events.by_ref().take(batch_size).collect();
if chunk.is_empty() {
break;
}
let persist = if self.notify_in_tx {
Tables::persist_events_notifying(&mut op, chunk.into_iter()).await
} else {
Tables::persist_events(&mut op, chunk.into_iter()).await
};
let persisted_chunk = match persist {
Ok(chunk) => chunk,
Err(error) => {
// Earlier chunks were persisted and are now doomed
// with the transaction (the failing statement's own
// sequences are unknowable — burned inside the
// aborted statement; the backstop proves and fills
// them).
self.report_own_failure();
return Err(error);
}
};
// Stash before running the post-persist hooks: a hook error
// rolls the transaction back with these sequences already
// allocated, and compensation must know about them.
let chunk_start = self.post_commit_events.len();
self.post_commit_events.extend(persisted_chunk);
for hook in self.post_persist_hooks.iter() {
if let Err(error) = hook
.on_persisted(&mut op, &self.post_commit_events[chunk_start..])
.await
{
self.report_own_failure();
return Err(error);
}
}
}
PreCommitRet::ok(self, op)
}
fn post_commit(mut self) {
let post_commit_events = std::mem::take(&mut self.post_commit_events);
let batch_range = match (post_commit_events.first(), post_commit_events.last()) {
(Some(first), Some(last)) => Some((first.sequence, last.sequence)),
_ => None,
};
for event in post_commit_events {
let _ = self.sender.send(PersistentDelivery::from(Ok(event)));
}
if let Some(range) = batch_range {
let _ = self.notifier_tx.send(range);
}
}
/// Rollback compensation (the reactive tier of gap filling).
///
/// es-entity fires this when the commit failed after our `pre_commit`
/// had completed — a later hook's `pre_commit` errored (the
/// transaction is already rolled back when this runs, so the
/// GapFiller's insert contends with nothing) or the COMMIT itself
/// failed (the transaction is over server-side either way; the
/// GapFiller's `ON CONFLICT DO NOTHING` insert is idempotent against
/// a commit that actually landed). The stashed sequences are this
/// process's own abandoned allocations; reporting them to the
/// [`GapFiller`](crate::out::gap_fill::GapFiller) fills their
/// placeholders in milliseconds instead of leaving downstream
/// listeners stalled until the grace-gated backstop proves them lost.
///
/// Signal-only per the trait contract: the DB work happens on the
/// GapFiller task. Failures *inside* our own `pre_commit` never
/// reach here (the failing hook is consumed by its own call) — those
/// report from the error branches in `pre_commit` itself.
fn on_rollback(mut self) {
let abandoned: Vec<EventSequence> = self
.post_commit_events
.drain(..)
.map(|event| event.sequence)
.collect();
if !abandoned.is_empty() {
let _ = self.abandoned_tx.send(GapFillRequest::Abandoned(abandoned));
}
}
fn merge(&mut self, other: &mut Self) -> bool {
self.pre_commit_events.append(&mut other.pre_commit_events);
true
}
}