minco_plugin_audit/
journal.rs

1//! Transactional audit-intent journal and explicit bounded relay.
2
3use crate::{AuditLedgerError, AuditLedgerWriter, AuditRecordV2, MAX_AUDIT_BATCH_RECORDS};
4use async_trait::async_trait;
5use chrono::{DateTime, TimeDelta, Utc};
6use std::{collections::BTreeMap, sync::Arc};
7use tokio::sync::RwLock;
8use uuid::Uuid;
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum AuditJournalStatus {
12    Pending,
13    Claimed,
14    Failed,
15    Quarantined,
16}
17
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub struct AuditJournalEntry {
20    pub record: AuditRecordV2,
21    pub status: AuditJournalStatus,
22    pub attempt_count: u32,
23    pub encoded_bytes: usize,
24    pub available_at: DateTime<Utc>,
25    pub claimed_by: Option<String>,
26    pub claim_expires_at: Option<DateTime<Utc>>,
27    pub failure_code: Option<String>,
28}
29
30impl AuditJournalEntry {
31    pub fn pending(record: AuditRecordV2) -> Result<Self, AuditLedgerError> {
32        let encoded_bytes = record.validate()?;
33        Ok(Self {
34            record,
35            status: AuditJournalStatus::Pending,
36            attempt_count: 0,
37            encoded_bytes,
38            available_at: Utc::now(),
39            claimed_by: None,
40            claim_expires_at: None,
41            failure_code: None,
42        })
43    }
44}
45
46#[async_trait]
47pub trait AuditJournalStore: Send + Sync + std::fmt::Debug {
48    async fn enqueue(&self, entry: AuditJournalEntry) -> Result<(), AuditLedgerError>;
49
50    async fn claim_pending(
51        &self,
52        worker_id: &str,
53        limit: usize,
54        claim_expires_at: DateTime<Utc>,
55    ) -> Result<Vec<AuditJournalEntry>, AuditLedgerError>;
56
57    async fn mark_delivered(
58        &self,
59        event_ids: &[Uuid],
60        worker_id: &str,
61    ) -> Result<(), AuditLedgerError>;
62
63    async fn mark_retry(
64        &self,
65        event_ids: &[Uuid],
66        worker_id: &str,
67        failure_code: &str,
68        retry_at: DateTime<Utc>,
69    ) -> Result<(), AuditLedgerError>;
70
71    async fn quarantine(
72        &self,
73        event_ids: &[Uuid],
74        worker_id: &str,
75        failure_code: &str,
76    ) -> Result<(), AuditLedgerError>;
77
78    async fn recover_expired_claims(&self, now: DateTime<Utc>) -> Result<usize, AuditLedgerError>;
79}
80
81#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
82pub struct AuditRelayReport {
83    pub claimed: usize,
84    pub inserted: usize,
85    pub duplicates: usize,
86    pub retried: usize,
87    pub quarantined: usize,
88}
89
90#[derive(Debug, Clone)]
91pub struct AuditRelay {
92    journal: Arc<dyn AuditJournalStore>,
93    ledger: Arc<dyn AuditLedgerWriter>,
94}
95
96impl AuditRelay {
97    pub fn new(journal: Arc<dyn AuditJournalStore>, ledger: Arc<dyn AuditLedgerWriter>) -> Self {
98        Self { journal, ledger }
99    }
100
101    /// Executes one explicit bounded delivery pass. Minco never schedules it.
102    pub async fn dispatch_once(
103        &self,
104        worker_id: &str,
105        limit: usize,
106        lease: TimeDelta,
107    ) -> Result<AuditRelayReport, AuditLedgerError> {
108        validate_claim(worker_id, limit, lease)?;
109        let now = Utc::now();
110        self.journal.recover_expired_claims(now).await?;
111        let entries = self
112            .journal
113            .claim_pending(worker_id, limit, now + lease)
114            .await?;
115        let mut report = AuditRelayReport {
116            claimed: entries.len(),
117            ..AuditRelayReport::default()
118        };
119        if entries.is_empty() {
120            return Ok(report);
121        }
122        let event_ids = entries
123            .iter()
124            .map(|entry| entry.record.event_id)
125            .collect::<Vec<_>>();
126        let records = entries
127            .into_iter()
128            .map(|entry| entry.record)
129            .collect::<Vec<_>>();
130        match self.ledger.append_batch(&records).await {
131            Ok(appended) => {
132                self.journal.mark_delivered(&event_ids, worker_id).await?;
133                report.inserted = appended.inserted;
134                report.duplicates = appended.duplicates;
135            }
136            Err(error) if error.is_permanent() => {
137                self.journal
138                    .quarantine(&event_ids, worker_id, error.stable_code())
139                    .await?;
140                report.quarantined = event_ids.len();
141            }
142            Err(error) => {
143                self.journal
144                    .mark_retry(
145                        &event_ids,
146                        worker_id,
147                        error.stable_code(),
148                        Utc::now() + TimeDelta::seconds(30),
149                    )
150                    .await?;
151                report.retried = event_ids.len();
152            }
153        }
154        Ok(report)
155    }
156}
157
158fn validate_claim(worker_id: &str, limit: usize, lease: TimeDelta) -> Result<(), AuditLedgerError> {
159    if worker_id.trim().is_empty()
160        || worker_id.len() > 128
161        || worker_id.chars().any(char::is_control)
162        || limit == 0
163        || limit > MAX_AUDIT_BATCH_RECORDS
164        || lease <= TimeDelta::zero()
165        || lease > TimeDelta::hours(1)
166    {
167        return Err(AuditLedgerError::InvalidJournalClaim);
168    }
169    Ok(())
170}
171
172fn validate_transition(event_ids: &[Uuid], worker_id: &str) -> Result<(), AuditLedgerError> {
173    if event_ids.is_empty()
174        || event_ids.len() > MAX_AUDIT_BATCH_RECORDS
175        || worker_id.trim().is_empty()
176    {
177        return Err(AuditLedgerError::InvalidJournalClaim);
178    }
179    Ok(())
180}
181
182#[derive(Debug, Default)]
183pub struct MemoryAuditJournal {
184    entries: RwLock<BTreeMap<Uuid, AuditJournalEntry>>,
185}
186
187impl MemoryAuditJournal {
188    pub async fn entries(&self) -> Vec<AuditJournalEntry> {
189        self.entries.read().await.values().cloned().collect()
190    }
191}
192
193#[async_trait]
194impl AuditJournalStore for MemoryAuditJournal {
195    async fn enqueue(&self, entry: AuditJournalEntry) -> Result<(), AuditLedgerError> {
196        if entry.status != AuditJournalStatus::Pending
197            || entry.encoded_bytes != entry.record.validate()?
198        {
199            return Err(AuditLedgerError::InvalidJournalEntry);
200        }
201        let mut entries = self.entries.write().await;
202        let result = match entries.get(&entry.record.event_id) {
203            Some(existing) if existing.record == entry.record => Ok(()),
204            Some(_) => Err(AuditLedgerError::EventConflict(entry.record.event_id)),
205            None => {
206                entries.insert(entry.record.event_id, entry);
207                Ok(())
208            }
209        };
210        drop(entries);
211        result
212    }
213
214    async fn claim_pending(
215        &self,
216        worker_id: &str,
217        limit: usize,
218        claim_expires_at: DateTime<Utc>,
219    ) -> Result<Vec<AuditJournalEntry>, AuditLedgerError> {
220        let lease = claim_expires_at - Utc::now();
221        validate_claim(worker_id, limit, lease)?;
222        let now = Utc::now();
223        let mut entries = self.entries.write().await;
224        let ids = entries
225            .values()
226            .filter(|entry| {
227                matches!(
228                    entry.status,
229                    AuditJournalStatus::Pending | AuditJournalStatus::Failed
230                ) && entry.available_at <= now
231            })
232            .take(limit)
233            .map(|entry| entry.record.event_id)
234            .collect::<Vec<_>>();
235        let mut claimed = Vec::with_capacity(ids.len());
236        for id in ids {
237            let entry = entries.get_mut(&id).expect("selected memory entry");
238            entry.status = AuditJournalStatus::Claimed;
239            entry.attempt_count = entry.attempt_count.saturating_add(1);
240            entry.claimed_by = Some(worker_id.into());
241            entry.claim_expires_at = Some(claim_expires_at);
242            claimed.push(entry.clone());
243        }
244        drop(entries);
245        Ok(claimed)
246    }
247
248    async fn mark_delivered(
249        &self,
250        event_ids: &[Uuid],
251        worker_id: &str,
252    ) -> Result<(), AuditLedgerError> {
253        validate_transition(event_ids, worker_id)?;
254        let mut entries = self.entries.write().await;
255        require_claims(&entries, event_ids, worker_id)?;
256        for event_id in event_ids {
257            entries.remove(event_id);
258        }
259        drop(entries);
260        Ok(())
261    }
262
263    async fn mark_retry(
264        &self,
265        event_ids: &[Uuid],
266        worker_id: &str,
267        failure_code: &str,
268        retry_at: DateTime<Utc>,
269    ) -> Result<(), AuditLedgerError> {
270        validate_transition(event_ids, worker_id)?;
271        validate_failure_code(failure_code)?;
272        let mut entries = self.entries.write().await;
273        require_claims(&entries, event_ids, worker_id)?;
274        for event_id in event_ids {
275            let entry = entries.get_mut(event_id).expect("validated memory claim");
276            entry.status = AuditJournalStatus::Failed;
277            entry.available_at = retry_at;
278            entry.claimed_by = None;
279            entry.claim_expires_at = None;
280            entry.failure_code = Some(failure_code.into());
281        }
282        drop(entries);
283        Ok(())
284    }
285
286    async fn quarantine(
287        &self,
288        event_ids: &[Uuid],
289        worker_id: &str,
290        failure_code: &str,
291    ) -> Result<(), AuditLedgerError> {
292        validate_transition(event_ids, worker_id)?;
293        validate_failure_code(failure_code)?;
294        let mut entries = self.entries.write().await;
295        require_claims(&entries, event_ids, worker_id)?;
296        for event_id in event_ids {
297            let entry = entries.get_mut(event_id).expect("validated memory claim");
298            entry.status = AuditJournalStatus::Quarantined;
299            entry.claimed_by = None;
300            entry.claim_expires_at = None;
301            entry.failure_code = Some(failure_code.into());
302        }
303        drop(entries);
304        Ok(())
305    }
306
307    async fn recover_expired_claims(&self, now: DateTime<Utc>) -> Result<usize, AuditLedgerError> {
308        let mut entries = self.entries.write().await;
309        let mut recovered = 0;
310        for entry in entries.values_mut() {
311            if entry.status == AuditJournalStatus::Claimed
312                && entry.claim_expires_at.is_some_and(|expires| expires <= now)
313            {
314                entry.status = AuditJournalStatus::Failed;
315                entry.available_at = now;
316                entry.claimed_by = None;
317                entry.claim_expires_at = None;
318                entry.failure_code = Some("AUDIT-CLAIM-EXPIRED".into());
319                recovered += 1;
320            }
321        }
322        drop(entries);
323        Ok(recovered)
324    }
325}
326
327fn require_claims(
328    entries: &BTreeMap<Uuid, AuditJournalEntry>,
329    event_ids: &[Uuid],
330    worker_id: &str,
331) -> Result<(), AuditLedgerError> {
332    if event_ids.iter().all(|event_id| {
333        entries.get(event_id).is_some_and(|entry| {
334            entry.status == AuditJournalStatus::Claimed
335                && entry.claimed_by.as_deref() == Some(worker_id)
336        })
337    }) {
338        Ok(())
339    } else {
340        Err(AuditLedgerError::JournalClaimLost)
341    }
342}
343
344fn validate_failure_code(value: &str) -> Result<(), AuditLedgerError> {
345    if value.is_empty()
346        || value.len() > 128
347        || !value
348            .bytes()
349            .all(|byte| byte.is_ascii_uppercase() || byte.is_ascii_digit() || byte == b'-')
350    {
351        Err(AuditLedgerError::InvalidJournalEntry)
352    } else {
353        Ok(())
354    }
355}
356
357#[cfg(test)]
358mod tests {
359    use super::*;
360    use crate::{
361        AuditActor, AuditLedgerError, AuditLedgerWriter, AuditResourceRef, MemoryAuditLedger,
362    };
363    use async_trait::async_trait;
364
365    fn record() -> AuditRecordV2 {
366        AuditRecordV2::new(
367            "tenant",
368            "order.created",
369            AuditResourceRef::new("order", "one"),
370            AuditActor::human("subject"),
371            "placeOrder",
372            Uuid::now_v7(),
373        )
374    }
375
376    #[derive(Debug)]
377    struct FailingLedger(AuditLedgerError);
378
379    #[async_trait]
380    impl AuditLedgerWriter for FailingLedger {
381        async fn append_batch(
382            &self,
383            _records: &[AuditRecordV2],
384        ) -> Result<crate::AuditAppendReport, AuditLedgerError> {
385            match self.0 {
386                AuditLedgerError::Infrastructure => Err(AuditLedgerError::Infrastructure),
387                _ => Err(AuditLedgerError::InvalidRecord("invalid".into())),
388            }
389        }
390    }
391
392    #[tokio::test]
393    async fn relay_deletes_only_after_idempotent_ledger_commit() {
394        let journal = Arc::new(MemoryAuditJournal::default());
395        let ledger = Arc::new(MemoryAuditLedger::default());
396        let action = record();
397        journal
398            .enqueue(AuditJournalEntry::pending(action).unwrap())
399            .await
400            .unwrap();
401        let report = AuditRelay::new(journal.clone(), ledger)
402            .dispatch_once("worker", 10, TimeDelta::minutes(1))
403            .await
404            .unwrap();
405        assert_eq!(report.inserted, 1);
406        assert!(journal.entries().await.is_empty());
407    }
408
409    #[tokio::test]
410    async fn transient_failure_retries_and_permanent_failure_quarantines() {
411        for (error, expected) in [
412            (AuditLedgerError::Infrastructure, AuditJournalStatus::Failed),
413            (
414                AuditLedgerError::InvalidRecord("invalid".into()),
415                AuditJournalStatus::Quarantined,
416            ),
417        ] {
418            let journal = Arc::new(MemoryAuditJournal::default());
419            journal
420                .enqueue(AuditJournalEntry::pending(record()).unwrap())
421                .await
422                .unwrap();
423            AuditRelay::new(journal.clone(), Arc::new(FailingLedger(error)))
424                .dispatch_once("worker", 10, TimeDelta::minutes(1))
425                .await
426                .unwrap();
427            assert_eq!(journal.entries().await[0].status, expected);
428        }
429    }
430}
minco_plugin_audit/journal.rs

minco_plugin_audit/
journal.rs
