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::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    Configuration(Box<crate::client::ClientError>),
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 RetryableError for LocalDeliveryError {
115    fn is_retryable(&self) -> bool {
116        match self {
117            Self::Storage(error) => error.is_retryable(),
118            Self::NetworkFailure(error) => xmtp_api::preflight::failure(error.as_ref())
119                .is_some_and(RetryableError::is_retryable),
120            _ => false,
121        }
122    }
123}
xmtp_mls/subscriptions/local_delivery/types.rs

xmtp_mls/subscriptions/local_delivery/
types.rs
