distributed/repository/
inbox.rs

1//! Consumer inbox — the optional consumer-side effect fence.
2//!
3//! The inbox is the consumer-side complement to the producer outbox: an optional
4//! durable receipt that lets a consumer get **effectively-once local database
5//! effects** on top of at-least-once transport delivery. It is *not* a read-model
6//! feature (it replaces the removed `read_model_processed_messages`, which wrongly
7//! coupled delivery dedupe to the projection contract — see
8//! `specs/consumer-inbox-design.md`).
9//!
10//! An [`InboxReceipt`] is a participant in the transactional commit batch
11//! (alongside aggregate events, outbox rows, read-model write plans, and
12//! snapshots). The relational stores write it to an operational `consumer_inbox`
13//! table in the **same transaction** as everything else in the batch — that
14//! atomicity is what makes the fence real: handler effects and the receipt land
15//! together or not at all.
16//!
17//! Default consumers stay idempotent (a replayed projection re-converges); the
18//! inbox is the opt-in pattern for when exactly-once local effects are worth the
19//! cost. It only fences *local transactional* effects — external side effects
20//! (HTTP calls, etc.) still require handler idempotency.
21
22use std::time::SystemTime;
23
24use super::RepositoryError;
25
26/// A single consumer/message processing receipt.
27///
28/// Identified by `(consumer, message_id)`; committed atomically with the
29/// consumer's other writes so a redelivery of the same message is a no-op.
30#[derive(Clone, Debug, PartialEq, Eq)]
31pub struct InboxReceipt {
32    /// The logical consumer (e.g. a projection or service name) — the dedupe scope.
33    pub consumer: String,
34    /// The transport message's stable id.
35    pub message_id: String,
36    /// When the receipt was created (advisory). The relational backends stamp the
37    /// stored `processed_at` server-side, so this value is not currently persisted;
38    /// it exists for in-process use and forward compatibility with retention/prune.
39    pub processed_at: SystemTime,
40}
41
42impl InboxReceipt {
43    /// Create a receipt for `(consumer, message_id)`, stamped now.
44    pub fn new(consumer: impl Into<String>, message_id: impl Into<String>) -> Self {
45        Self {
46            consumer: consumer.into(),
47            message_id: message_id.into(),
48            processed_at: crate::time::now(),
49        }
50    }
51
52    /// The `(consumer, message_id)` dedupe key.
53    pub fn key(&self) -> (&str, &str) {
54        (&self.consumer, &self.message_id)
55    }
56
57    /// Reject an empty `consumer` or `message_id` so every backend behaves
58    /// identically (the relational `CHECK` constraints are a backstop).
59    pub fn validate(&self) -> Result<(), RepositoryError> {
60        if self.consumer.is_empty() || self.message_id.is_empty() {
61            return Err(RepositoryError::InvalidInboxReceipt {
62                consumer: self.consumer.clone(),
63                message_id: self.message_id.clone(),
64            });
65        }
66        Ok(())
67    }
68}
69
70/// Outcome of committing an [`InboxReceipt`].
71///
72/// A `Duplicate` is **not** an error: the message was already processed, so the
73/// consumer treats it as success (acks) without re-running effects.
74#[derive(Clone, Copy, Debug, PartialEq, Eq)]
75pub enum InboxOutcome {
76    /// The receipt was newly recorded (first time this message was processed).
77    Processed,
78    /// A receipt for `(consumer, message_id)` already existed; effects were skipped.
79    Duplicate,
80}
81
82impl InboxOutcome {
83    /// Whether this is the first processing (vs a deduplicated replay).
84    pub fn is_processed(self) -> bool {
85        matches!(self, InboxOutcome::Processed)
86    }
87
88    /// Whether the message was already processed (a deduplicated replay).
89    pub fn is_duplicate(self) -> bool {
90        matches!(self, InboxOutcome::Duplicate)
91    }
92}
93
94#[cfg(test)]
95mod tests {
96    use super::*;
97
98    #[test]
99    fn receipt_key_is_consumer_and_message_id() {
100        let r = InboxReceipt::new("projections", "evt-1");
101        assert_eq!(r.key(), ("projections", "evt-1"));
102        assert_eq!(r.consumer, "projections");
103        assert_eq!(r.message_id, "evt-1");
104    }
105
106    #[test]
107    fn outcome_predicates() {
108        assert!(InboxOutcome::Processed.is_processed());
109        assert!(!InboxOutcome::Processed.is_duplicate());
110        assert!(InboxOutcome::Duplicate.is_duplicate());
111        assert!(!InboxOutcome::Duplicate.is_processed());
112    }
113
114    #[test]
115    fn validate_rejects_empty_fields() {
116        assert!(InboxReceipt::new("c", "m").validate().is_ok());
117        assert!(matches!(
118            InboxReceipt::new("", "m").validate(),
119            Err(RepositoryError::InvalidInboxReceipt { .. })
120        ));
121        assert!(matches!(
122            InboxReceipt::new("c", "").validate(),
123            Err(RepositoryError::InvalidInboxReceipt { .. })
124        ));
125    }
126}
distributed/repository/inbox.rs

distributed/repository/
inbox.rs
