cratestack_sqlx/audit/
sink.rs

1//! Post-commit fan-out of already-persisted [`AuditEvent`]s to the
2//! runtime's installed [`cratestack_core::AuditSink`] (cratestack#473).
3
4use cratestack_core::AuditEvent;
5
6use crate::descriptor::SqlxRuntime;
7
8/// Fan a batch of already-committed [`AuditEvent`]s out to the
9/// runtime's installed [`cratestack_core::AuditSink`].
10///
11/// `pub`, not `pub(crate)` (cratestack#534): every `run()` call site in
12/// this crate calls it internally, right after its own `tx.commit()`
13/// succeeds — that usage is unaffected. What's new is that a caller
14/// composing `run_in_tx` writes across a transaction *they* own can
15/// now call this too, once *their* commit succeeds, passing the
16/// `AuditEvent`s each call's [`super::RunInTxOutcome`] handed back.
17/// Nothing about the dispatch itself changed: it is still a plain
18/// sequential fan-out with no DB I/O of its own (see below) — only who
19/// is allowed to invoke it changed.
20///
21/// **Deliberately called after `tx.commit()`, never before or from
22/// inside the transaction.** Two reasons:
23///
24/// 1. **No double-write of the source of truth.** The only DB write is
25///    [`super::enqueue_audit_event`], run once, in-transaction, before
26///    this is ever reached. This function performs no DB I/O — it only
27///    invokes `AuditSink::record`, an out-of-band, best-effort
28///    projection. There is exactly one write to `cratestack_audit` per
29///    event either way; this cannot cause a second one.
30/// 2. **The transaction must not wait on downstream I/O.** A sink can
31///    be a Kafka publish, a Redis command, or an HTTP webhook — any of
32///    which can be slow or hang. Running that call while still holding
33///    the mutation's row locks would turn an unrelated outage (the
34///    Kafka broker is down) into a long-held Postgres lock, which is
35///    far worse than a late or dropped downstream projection. Waiting
36///    for commit also guarantees a sink is only ever invoked for
37///    events that actually happened — a rolled-back transaction never
38///    reaches this call, so the sink can't observe a mutation the
39///    database itself discarded.
40///
41/// Errors are logged, not propagated: by the time this runs the
42/// mutation already committed, so failing the caller's request over a
43/// downstream sink hiccup would be strictly worse than a best-effort
44/// delivery. This mirrors `run()`'s existing `let _ =
45/// self.runtime.drain_event_outbox().await;` treatment of its own
46/// post-commit, best-effort fan-out.
47///
48/// **Still not called from any `run_in_tx` variant — that remains a
49/// deliberate omission, not an oversight (cratestack#534).** `run_in_tx`
50/// hands the transaction back to the caller uncommitted, so this crate
51/// has no reliable "after commit" point of its own to run at — same
52/// reason `run_in_tx` never drains the event outbox on its own either
53/// (see `crate::query::write::create`'s doc comment, and
54/// [`crate::SqlxRuntime::drain_event_outbox`] for that mechanism's own
55/// equivalent, now-public opt-in). What changed is that a `run_in_tx`
56/// caller now genuinely *can* opt in: every `run_in_tx` variant returns
57/// a [`super::RunInTxOutcome`] carrying the `AuditEvent`(s) it built and
58/// already persisted, and this function is `pub` so the caller can pass
59/// them straight through — after their own `tx.commit()` succeeds, never
60/// before. The generated `Cratestack::dispatch_audit_sink` method is the
61/// ergonomic surface for that; this free function is what it forwards
62/// to. Skipping the call (or forgetting it) means exactly what option
63/// (c) in cratestack#534 describes: `cratestack_audit` still gets the
64/// row, but the installed `AuditSink` observes nothing for that
65/// transaction — silently, same as before this fix, just opt-out now
66/// instead of impossible-to-opt-into.
67/// `crates/cratestack-pg/tests/banking_chained_audit_tx.rs` is the shape
68/// this closes: two `run_in_tx` writes chained in one caller-managed
69/// transaction, both audited, both now observable by a sink the caller
70/// dispatches to once after their single `tx.commit()`.
71///
72/// **Dispatch is sequential, not concurrent**, and that amplifies with
73/// batch size: the `for` loop below `.await`s each `AuditSink::record`
74/// call one at a time, so an `update_many`/`delete_many`/`batch_*` call
75/// touching N rows makes N sequential post-commit sink calls before the
76/// response returns — a slow sink's added latency is per-row, not
77/// per-request. Deliberately not parallelised here: concurrent
78/// dispatch's ordering guarantees and per-event error semantics are a
79/// design question of their own, not a cleanup.
80pub async fn dispatch_audit_sink(runtime: &SqlxRuntime, events: &[AuditEvent]) {
81    for event in events {
82        if let Err(error) = runtime.audit_sink().record(event).await {
83            tracing::warn!(
84                error = %error,
85                event_id = %event.event_id,
86                model = %event.model,
87                operation = event.operation.as_str(),
88                "audit sink failed to record event",
89            );
90        }
91    }
92}
cratestack_sqlx/audit/sink.rs

cratestack_sqlx/audit/
sink.rs
