aster_forge_audit/
db_writer.rs

1//! Explicitly owned audit persistence using the product-neutral DB store.
2
3use std::sync::Arc;
4use std::time::Duration;
5
6use aster_forge_db::{AuditLogCreate, DbError, create_audit_log_requests};
7use aster_forge_runtime::{
8    BufferFlushError, BufferStopReport, BufferedBatchConfig, BufferedWriterEvent,
9    BufferedWriterSnapshot, BufferedWriterState, ManagedBatchWriter, OverflowPolicy,
10    RecordAdmission, RecordError, RuntimeComponentBundle, RuntimeComponentBundleRegistration,
11    RuntimeComponentKind, RuntimeComponentRegistry, StartupPhaseFailurePolicy, WriteFailure,
12    WriteFailureClass, runtime_component,
13};
14use sea_orm::DatabaseConnection;
15use tokio::sync::mpsc;
16use tokio::time::Instant;
17
18use crate::{AUDIT_LOGS_COMPONENT, AUDIT_MANAGER_COMPONENT, AUDIT_MANAGER_FLUSH_SHUTDOWN_PHASE};
19
20/// Default queue capacity; an additional single overflow record can be pending.
21pub const DEFAULT_AUDIT_LOG_QUEUE_CAPACITY: usize = 4096;
22/// Default immediate-flush threshold.
23pub const DEFAULT_AUDIT_LOG_BATCH_SIZE: usize = 100;
24/// Default partial-batch delay.
25pub const DEFAULT_AUDIT_LOG_DELAYED_FLUSH_AFTER: Duration = Duration::from_secs(1);
26/// Explicit writer startup phase, before product lifecycle audit hooks.
27pub const AUDIT_WRITER_START_PHASE: &str = "audit_writer_start";
28
29/// Product-neutral audit buffering configuration.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct AuditLogBufferConfig {
32    /// Maximum unacknowledged queued records, including an in-flight batch; zero becomes one.
33    pub queue_capacity: usize,
34    /// Immediate-flush threshold and maximum transaction chunk size; zero becomes one.
35    pub batch_size: usize,
36    /// Maximum wait from first queued record before a partial batch is attempted.
37    pub delayed_flush_after: Duration,
38}
39
40impl AuditLogBufferConfig {
41    /// Creates bounded audit buffering settings.
42    #[must_use]
43    pub const fn new(
44        queue_capacity: usize,
45        batch_size: usize,
46        delayed_flush_after: Duration,
47    ) -> Self {
48        Self {
49            queue_capacity,
50            batch_size,
51            delayed_flush_after,
52        }
53    }
54}
55
56impl Default for AuditLogBufferConfig {
57    fn default() -> Self {
58        Self::new(
59            DEFAULT_AUDIT_LOG_QUEUE_CAPACITY,
60            DEFAULT_AUDIT_LOG_BATCH_SIZE,
61            DEFAULT_AUDIT_LOG_DELAYED_FLUSH_AFTER,
62        )
63    }
64}
65
66/// Caller-owned audit writer. Construction starts no workers, timers, or database operations.
67///
68/// Queued admission acknowledges memory only. Mandatory audit should use the DB store in the
69/// business transaction, or explicitly await `flush` and map failures at the product boundary.
70/// A failed or interrupted write remains available through `unconfirmed`; never replay an unknown
71/// commit without reconciling it first. Stop/join this writer before closing the database.
72pub struct AuditWriter {
73    writer: ManagedBatchWriter<AuditLogCreate, DbError>,
74}
75
76impl AuditWriter {
77    /// Uses default bounds and a bounded direct-overflow lane.
78    #[must_use]
79    pub fn new(db: DatabaseConnection) -> Self {
80        Self::with_config(
81            db,
82            AuditLogBufferConfig::default(),
83            OverflowPolicy::DirectWrite,
84            None,
85        )
86    }
87
88    /// Configures overflow and optional bounded event delivery. Observer code is never invoked
89    /// by this writer; a slow, panicking, or absent event consumer cannot block persistence.
90    #[must_use]
91    pub fn with_config(
92        db: DatabaseConnection,
93        config: AuditLogBufferConfig,
94        overflow: OverflowPolicy,
95        events: Option<mpsc::Sender<BufferedWriterEvent>>,
96    ) -> Self {
97        let writer = ManagedBatchWriter::new(
98            BufferedBatchConfig::new(
99                config.queue_capacity,
100                config.batch_size,
101                config.delayed_flush_after,
102            ),
103            overflow,
104            events,
105            move |requests: Vec<AuditLogCreate>| {
106                let db = db.clone();
107                async move {
108                    // A single transaction per chunk makes failure atomic and exposes uncertain
109                    // commit outcomes. Disable internal retries: the owner decides whether to retry.
110                    let retry = aster_forge_db::retry::RetryConfig {
111                        max_retries: 0,
112                        ..aster_forge_db::retry::RetryConfig::deadlock()
113                    };
114                    aster_forge_db::transaction::with_transaction_retry(
115                        &db,
116                        &retry,
117                        move |txn| {
118                            let requests = requests.clone();
119                            Box::pin(async move { create_audit_log_requests(txn, requests).await })
120                        },
121                        DbError::is_retryable,
122                    )
123                    .await
124                    .map_err(classify_failure)
125                }
126            },
127        );
128        Self { writer }
129    }
130
131    /// Starts the owned worker exactly once; normally called by the runtime component.
132    ///
133    /// # Errors
134    /// Returns the current state when already started, draining, or stopped.
135    pub async fn start(&self) -> Result<(), BufferedWriterState> {
136        self.writer.start().await
137    }
138    /// Admits an audit record. `Queued` acknowledges memory only; `DirectPersisted` acknowledges
139    /// just this overflow record. A cancelled direct waiter does not revoke accepted admission.
140    ///
141    /// # Errors
142    /// Rejected records are returned. Direct failures retain the accepted record in this writer.
143    pub async fn record(
144        &self,
145        request: AuditLogCreate,
146    ) -> Result<RecordAdmission, RecordError<AuditLogCreate, DbError>> {
147        self.writer.record(request).await
148    }
149    /// Atomically closes admission while keeping accepted records available for drain.
150    /// Call after all product producers and the server-shutdown audit hook have finished.
151    pub fn begin_drain(&self) {
152        self.writer.begin_drain();
153    }
154    /// Awaits persistence of records accepted before this future's first poll, without closing
155    /// admission. Batch and direct writes are acknowledged separately; later records and their
156    /// failures do not extend this scope. An expired deadline schedules no new forced write,
157    /// while an already acknowledged scope succeeds immediately, even with an expired deadline.
158    /// Failed records stay in memory; cancelling this future leaves persistence owned by the worker.
159    ///
160    /// # Errors
161    /// Returns lifecycle, original classified persistence, or deadline errors.
162    pub async fn flush(&self, deadline: Instant) -> Result<(), BufferFlushError<DbError>> {
163        self.writer.flush(deadline).await
164    }
165    /// Closes admission, attempts drain, then cancels and joins the worker before returning.
166    /// Inspect remaining counts/failures before deciding product exit behavior. Repeated and
167    /// concurrent calls are safe; database handles must remain open until this completes.
168    pub async fn stop(&self, deadline: Instant) -> BufferStopReport<DbError> {
169        self.writer.stop(deadline).await
170    }
171    /// Explicitly resumes only retry-safe failures; unknown commits and permanent failures stay
172    /// blocked. Does not restart a stopped writer or silently replay retained records.
173    pub fn retry_failed(&self) {
174        self.writer.retry_failed();
175    }
176    /// Returns per-instance state, acknowledgment counters, and original classified failures.
177    #[must_use]
178    pub fn snapshot(&self) -> BufferedWriterSnapshot<DbError> {
179        self.writer.snapshot()
180    }
181    /// Copies unacknowledged records for product-controlled reconciliation. Stop the writer
182    /// first for a stable recovery snapshot; these records must not be assumed absent from DB.
183    #[must_use]
184    pub fn unconfirmed(&self) -> Vec<AuditLogCreate> {
185        self.writer.unconfirmed()
186    }
187}
188
189fn classify_failure(error: DbError) -> WriteFailure<DbError> {
190    let class = if error.commit_outcome_is_unknown() {
191        WriteFailureClass::UnknownCommit
192    } else if error.is_retryable() {
193        WriteFailureClass::Retryable
194    } else {
195        WriteFailureClass::Permanent
196    };
197    WriteFailure::new(class, error)
198}
199
200/// Registers explicit writer startup and drain/stop on the existing audit graph.
201///
202/// Register this bundle BEFORE lifecycle audit hooks so startup can record through the writer.
203/// `audit_manager` stops after `audit_logs` (and its producer/mail dependencies); database close
204/// must depend on `AUDIT_MANAGER_COMPONENT`. The timeout belongs to the writer, so the runtime
205/// must not externally time out this phase before it joins its worker.
206/// Drain failures are reported as failed runtime phases, independently of lifecycle-hook policy.
207/// An unrepresentable timeout stops immediately and reports a configuration error after joining.
208pub fn audit_writer_component(
209    writer: Arc<AuditWriter>,
210    drain_timeout: Duration,
211) -> RuntimeComponentBundleRegistration<impl RuntimeComponentBundle> {
212    runtime_component(move |registry: &mut RuntimeComponentRegistry| {
213        let startup_writer = writer.clone();
214        registry
215            .component(AUDIT_MANAGER_COMPONENT)
216            .kind(RuntimeComponentKind::Product)
217            .depends_on(AUDIT_LOGS_COMPONENT)
218            .startup(
219                AUDIT_WRITER_START_PHASE,
220                StartupPhaseFailurePolicy::Required,
221                move || {
222                    let writer = startup_writer.clone();
223                    async move {
224                        writer
225                            .start()
226                            .await
227                            .map_err(|state| format!("audit writer cannot start from {state:?}"))
228                    }
229                },
230            )
231            .shutdown(AUDIT_MANAGER_FLUSH_SHUTDOWN_PHASE, None, move || {
232                let writer = writer.clone();
233                async move { stop_writer(&writer, drain_timeout).await }
234            });
235    })
236}
237
238// The component's timeout is internal: dropping an externally timed phase could close the
239// database while its worker is still alive. Even an unrepresentable deadline must stop/join.
240async fn stop_writer(writer: &AuditWriter, timeout: Duration) -> Result<(), String> {
241    let now = Instant::now();
242    let deadline = now.checked_add(timeout);
243    let report = writer.stop(deadline.unwrap_or(now)).await;
244    let remaining = report.snapshot.queued + report.snapshot.direct_pending;
245    if deadline.is_none() {
246        return Err(format!(
247            "audit drain timeout exceeds clock range; remaining={remaining}; worker joined, database dependency may close"
248        ));
249    }
250    if remaining == 0 {
251        return Ok(());
252    }
253    Err(format!(
254        "audit drain incomplete: remaining={remaining}, deadline_elapsed={}, batch_failure={:?}, direct_failure={:?}; worker joined, database dependency may close; reconcile retained records before replay",
255        report.deadline_elapsed, report.snapshot.batch_failure, report.snapshot.direct_failure
256    ))
257}
258
259/// Composes owned writer startup, product lifecycle hooks, and mandatory shutdown drain.
260///
261/// Unlike a manual collection of hooks, this bundle always registers writer startup first.
262/// Lifecycle records keep their product-defined semantics and error policy. With
263/// `mail-outbox-dependency`, mail drain finishes before the shutdown record is admitted.
264/// Do not also register `audit_component`/`audit_manager_component`, which own the same phase.
265pub fn audit_component_with_writer<T, StartFn, StartFut, ShutdownFn, ShutdownFut>(
266    resources: T,
267    writer: Arc<AuditWriter>,
268    drain_timeout: Duration,
269    record_server_start: StartFn,
270    record_server_shutdown: ShutdownFn,
271) -> RuntimeComponentBundleRegistration<impl RuntimeComponentBundle>
272where
273    T: Clone + Send + 'static,
274    StartFn: FnOnce(T) -> StartFut + Send + Sync + 'static,
275    StartFut: std::future::Future<Output = Result<(), String>> + Send + 'static,
276    ShutdownFn: FnOnce(T) -> ShutdownFut + Send + Sync + 'static,
277    ShutdownFut: std::future::Future<Output = Result<(), String>> + Send + 'static,
278{
279    audit_component_with_writer_after(
280        resources,
281        writer,
282        drain_timeout,
283        crate::DEFAULT_SERVER_SHUTDOWN_AUDIT_DEPENDENCIES,
284        record_server_start,
285        record_server_shutdown,
286    )
287}
288
289/// Owned audit bundle with explicit producer dependencies, for example a follower with tasks
290/// but no mail outbox. Writer startup still precedes product hooks; stop follows every declared
291/// producer. Use this rather than registering placeholder components to satisfy default edges.
292pub fn audit_component_with_writer_after<T, StartFn, StartFut, ShutdownFn, ShutdownFut>(
293    resources: T,
294    writer: Arc<AuditWriter>,
295    drain_timeout: Duration,
296    shutdown_dependencies: &'static [&'static str],
297    record_server_start: StartFn,
298    record_server_shutdown: ShutdownFn,
299) -> RuntimeComponentBundleRegistration<impl RuntimeComponentBundle>
300where
301    T: Clone + Send + 'static,
302    StartFn: FnOnce(T) -> StartFut + Send + Sync + 'static,
303    StartFut: std::future::Future<Output = Result<(), String>> + Send + 'static,
304    ShutdownFn: FnOnce(T) -> ShutdownFut + Send + Sync + 'static,
305    ShutdownFut: std::future::Future<Output = Result<(), String>> + Send + 'static,
306{
307    let owner = audit_writer_component(writer, drain_timeout);
308    let start = crate::server_start_audit_component(resources.clone(), record_server_start);
309    let shutdown = crate::server_shutdown_audit_component_after(
310        resources,
311        shutdown_dependencies,
312        record_server_shutdown,
313    );
314    runtime_component(move |registry: &mut RuntimeComponentRegistry| {
315        registry
316            .register_bundle(owner)
317            .register_bundle(start)
318            .register_bundle(shutdown);
319    })
320}
321
322#[cfg(test)]
323mod tests {
324    use super::*;
325    use aster_forge_db::DatabaseErrorKind;
326    use aster_forge_runtime::WriteFailureCause;
327
328    #[test]
329    fn retains_native_classification_and_unknown_commit_has_priority() {
330        let retry = classify_failure(DbError::database_operation_classified(
331            "deadlock",
332            DatabaseErrorKind::Deadlock,
333        ));
334        assert_eq!(retry.class, WriteFailureClass::Retryable);
335        let unknown = classify_failure(DbError::commit_outcome_unknown(
336            "response lost",
337            Some(DatabaseErrorKind::Deadlock),
338        ));
339        assert_eq!(unknown.class, WriteFailureClass::UnknownCommit);
340        let WriteFailureCause::Persistence(error) = unknown.cause else {
341            panic!("original error missing")
342        };
343        assert!(error.commit_outcome_is_unknown());
344        assert_eq!(
345            error.database_error_kind(),
346            Some(DatabaseErrorKind::Deadlock)
347        );
348        assert_eq!(
349            classify_failure(DbError::non_retryable("invalid record")).class,
350            WriteFailureClass::Permanent
351        );
352    }
353}
aster_forge_audit/db_writer.rs

aster_forge_audit/
db_writer.rs
