xmtp_mls/subscriptions/local_delivery/
types.rs

1use thiserror::Error;
2use xmtp_common::{ErrorCode, RetryableError, time::Duration};
3use xmtp_db::StorageError;
4
5pub use xmtp_db::delivery::DeliveryFilter as LocalDeliveryFilter;
6
7use crate::subscriptions::{incoming::stream_api::RecoveryFailure, policy::StreamPolicy};
8
9const LEASE_RENEWAL_DIVISOR: u32 = 3;
10
11/// Bounds local reads and maintains exclusive default-consumer ownership.
12#[derive(Debug, Clone, Copy)]
13pub(crate) struct LocalDeliveryConfig {
14    /// Maximum candidates fetched in one local read.
15    pub batch_size: u32,
16    /// Maximum candidate bytes, checked before message blobs are loaded.
17    pub max_bytes: u64,
18    /// Maximum wait before checking cross-process writes and missed local wakes.
19    pub poll_interval: Duration,
20    /// Time before another process can take over an unrenewed default consumer.
21    pub lease_duration: Duration,
22    /// Renewal interval; must be shorter than the lease duration.
23    pub renew_interval: Duration,
24}
25
26impl Default for LocalDeliveryConfig {
27    fn default() -> Self {
28        Self::from(&StreamPolicy::default())
29    }
30}
31
32impl From<&StreamPolicy> for LocalDeliveryConfig {
33    fn from(settings: &StreamPolicy) -> Self {
34        Self {
35            batch_size: settings.max_local_read_rows,
36            max_bytes: settings.max_local_read_bytes,
37            poll_interval: settings.active_database_poll_interval,
38            lease_duration: settings.default_consumer_lease_duration,
39            renew_interval: settings.default_consumer_lease_duration / LEASE_RENEWAL_DIVISOR,
40        }
41    }
42}
43
44impl LocalDeliveryConfig {
45    pub(super) fn validate(self) -> Result<(), LocalDeliveryError> {
46        if self.batch_size == 0
47            || self.max_bytes == 0
48            || self.poll_interval.is_zero()
49            || self.renew_interval.is_zero()
50            || self.renew_interval >= self.lease_duration
51        {
52            return Err(LocalDeliveryError::InvalidConfiguration);
53        }
54        self.lease_until(0)?;
55        Ok(())
56    }
57
58    pub(super) fn lease_until(self, now: i64) -> Result<i64, LocalDeliveryError> {
59        now.checked_add(self.lease_duration_ns()?)
60            .ok_or(LocalDeliveryError::InvalidConfiguration)
61    }
62
63    pub(super) fn lease_duration_ns(self) -> Result<i64, LocalDeliveryError> {
64        i64::try_from(self.lease_duration.as_nanos())
65            .map_err(|_| LocalDeliveryError::InvalidConfiguration)
66    }
67}
68
69#[derive(Debug, Error, ErrorCode)]
70pub enum LocalDeliveryError {
71    /// Preserve the first terminal failure for every pending token.
72    #[error(transparent)]
73    #[error_code(inherit)]
74    SessionFailure(std::sync::Arc<LocalDeliveryError>),
75    /// The retained item has no decoded message. Not retryable.
76    #[error("The delivery item has no enriched message")]
77    EnrichedMessageUnavailable,
78    /// This stream exhausted its network retry budget. Not retryable; a new stream may retry.
79    #[error("The message reader exhausted its network recovery budget after {attempts} attempts: {}", source.as_ref().map(|error| error.to_string()).unwrap_or_else(|| "the recovery deadline elapsed without a transport error".into()))]
80    NetworkRecoveryExhausted {
81        attempts: u32,
82        #[source]
83        source: Option<std::sync::Arc<crate::subscriptions::incoming::IncomingError>>,
84    },
85    /// A terminal network error stopped this stream. A transient configuration
86    /// admission failure may be retried by opening a new reader.
87    #[error("The message reader network failed: {0}")]
88    NetworkFailure(#[source] std::sync::Arc<crate::subscriptions::incoming::IncomingError>),
89    /// The client's server configuration changed. Not retryable on this client.
90    #[error(transparent)]
91    #[error_code(inherit)]
92    Blocked(#[from] crate::server_configuration::BlockedConnection),
93    /// Database receipt, lease, cursor, or acknowledgement failure. May be retryable.
94    #[error(transparent)]
95    #[error_code(inherit)]
96    Storage(#[from] StorageError),
97    /// The callback failed or its token was dropped before acknowledgement. Not retryable.
98    #[error("The delivered message was not acknowledged")]
99    AcknowledgementRejected,
100    /// A previous acknowledgement write failed. Reopen to retry delivery. Not retryable.
101    #[error("Delivery stopped after acknowledgement persistence failed")]
102    AcknowledgementFailed,
103    /// Scope, filters, or retained content changed before dispatch. Reselect without acknowledgement.
104    #[error("The queued message belongs to an old delivery selection")]
105    SelectionChanged,
106    /// The batch or timing settings cannot maintain a valid consumer lease. Not retryable.
107    #[error("The local delivery configuration is invalid")]
108    InvalidConfiguration,
109    /// This reader has been closed. Not retryable.
110    #[error("The local message reader is closed")]
111    Closed,
112}
113
114impl From<RecoveryFailure> for LocalDeliveryError {
115    fn from(failure: RecoveryFailure) -> Self {
116        match failure {
117            RecoveryFailure::Exhausted { attempts, source } => {
118                Self::NetworkRecoveryExhausted { attempts, source }
119            }
120            RecoveryFailure::Terminal(error) => Self::NetworkFailure(error),
121        }
122    }
123}
124
125impl crate::worker::NeedsDbReconnect for LocalDeliveryError {
126    #[deny(clippy::wildcard_enum_match_arm)]
127    fn needs_db_reconnect(&self) -> bool {
128        match self {
129            Self::SessionFailure(error) => error.needs_db_reconnect(),
130            Self::NetworkRecoveryExhausted { source, .. } => source
131                .as_ref()
132                .is_some_and(|error| error.needs_db_reconnect()),
133            Self::NetworkFailure(error) => error.needs_db_reconnect(),
134            Self::Storage(error) => error.db_needs_connection(),
135            Self::EnrichedMessageUnavailable
136            | Self::Blocked(..)
137            | Self::AcknowledgementRejected
138            | Self::AcknowledgementFailed
139            | Self::SelectionChanged
140            | Self::InvalidConfiguration
141            | Self::Closed => false,
142        }
143    }
144}
145
146impl RetryableError for LocalDeliveryError {
147    fn is_retryable(&self) -> bool {
148        match self {
149            Self::Storage(error) => error.is_retryable(),
150            Self::NetworkFailure(error) => xmtp_api::preflight::failure(error.as_ref())
151                .is_some_and(RetryableError::is_retryable),
152            _ => false,
153        }
154    }
155
156    fn rate_limit_delay(&self) -> Option<xmtp_common::time::Duration> {
157        match self {
158            Self::NetworkRecoveryExhausted {
159                source: Some(error),
160                ..
161            }
162            | Self::NetworkFailure(error) => error.rate_limit_delay(),
163            _ => None,
164        }
165    }
166}
xmtp_mls/subscriptions/local_delivery/types.rs

xmtp_mls/subscriptions/local_delivery/
types.rs
