outbox_core/idempotency/
storage.rs

1//! Producer-side token reservation backend.
2//!
3//! [`IdempotencyStorageProvider`] is the narrow async contract an external
4//! store (Redis, another SQL table, an in-memory set for tests) must satisfy
5//! so [`OutboxService`](crate::service::OutboxService) can reject duplicate
6//! events at insert time. [`NoIdempotency`] is the trivial "always succeeds"
7//! implementation used when no such check is required.
8
9use crate::error::OutboxError;
10use crate::object::IdempotencyToken;
11use async_trait::async_trait;
12
13/// Async contract for reserving an idempotency token before an event is
14/// written.
15///
16/// The implementation is expected to perform an atomic "claim if absent"
17/// operation (e.g. `SET NX` in Redis) so concurrent producers cannot both
18/// reserve the same token.
19#[cfg_attr(test, mockall::automock)]
20#[async_trait]
21pub trait IdempotencyStorageProvider {
22    /// Attempts to reserve `token` as unused.
23    ///
24    /// Returns `true` if the token has been reserved by this call and the
25    /// event may proceed, or `false` if the token was already taken — in
26    /// which case [`OutboxService::add_event`](crate::service::OutboxService::add_event)
27    /// surfaces a [`DuplicateEvent`](OutboxError::DuplicateEvent).
28    ///
29    /// # Errors
30    ///
31    /// Returns an [`OutboxError`] — typically
32    /// [`InfrastructureError`](OutboxError::InfrastructureError) — when the
33    /// reservation backend itself fails (connection lost, timeout, etc.).
34    /// The error is propagated to the caller without inserting the event.
35    async fn try_reserve(&self, token: &IdempotencyToken) -> Result<bool, OutboxError>;
36}
37
38/// No-op [`IdempotencyStorageProvider`] that accepts every token.
39///
40/// Used when no external reservation backend is configured — i.e. when
41/// [`OutboxService::new`](crate::service::OutboxService::new) is called
42/// instead of
43/// [`with_idempotency`](crate::service::OutboxService::with_idempotency).
44/// Tokens produced by the strategy are still stored on the event row, so
45/// downstream consumers can deduplicate on their side if they need to.
46pub struct NoIdempotency;
47#[async_trait]
48impl IdempotencyStorageProvider for NoIdempotency {
49    /// Always returns `Ok(true)` — the token is considered "reserved"
50    /// unconditionally.
51    async fn try_reserve(&self, _token: &IdempotencyToken) -> Result<bool, OutboxError> {
52        Ok(true)
53    }
54}
55
56#[cfg(test)]
57#[allow(clippy::unwrap_used)]
58mod tests {
59    use super::*;
60    use rstest::rstest;
61
62    #[rstest]
63    #[case::non_empty("any-token")]
64    #[case::empty("")]
65    #[case::unicode("token-🦀")]
66    #[tokio::test]
67    async fn no_idempotency_returns_ok_true_for_any_token(#[case] raw: &str) {
68        let provider = NoIdempotency;
69        let token = IdempotencyToken::new(raw.to_string());
70        assert!(provider.try_reserve(&token).await.unwrap());
71    }
72
73    #[rstest]
74    #[tokio::test]
75    async fn no_idempotency_returns_ok_true_across_repeated_calls_with_same_token() {
76        let provider = NoIdempotency;
77        let token = IdempotencyToken::new("same".into());
78        assert!(provider.try_reserve(&token).await.unwrap());
79        assert!(provider.try_reserve(&token).await.unwrap());
80    }
81}
outbox_core/idempotency/storage.rs

outbox_core/idempotency/
storage.rs
