xmtp_mls_core/
policy.rs

1//! Client limits for incoming work and local message readers.
2
3use std::time::Duration;
4use xmtp_configuration::*;
5use xmtp_db::incoming_envelope::{IncomingLimits, NetworkEntityKind, PendingBudget};
6use xmtp_proto::types::IncomingBatchLimits;
7
8/// Bounds apply to encoded items and row metadata, not to the whole database.
9#[derive(Debug, Clone)]
10pub struct StreamPolicy {
11    /// Maximum rows committed in one durable admission transaction.
12    pub max_admission_rows: u32,
13    /// Maximum encoded bytes committed in one admission transaction.
14    pub max_admission_bytes: u64,
15    /// Maximum rows accepted from one network fetch, before admission chunks.
16    pub max_fetched_rows: u32,
17    /// Maximum bytes accepted from one network fetch; independent of per-kind budgets.
18    pub max_fetched_bytes: u64,
19    /// Total retained group-envelope budget across all group topics.
20    pub group_pending: PendingBudget,
21    /// Total retained Welcome budget, separate from group and identity budgets.
22    pub welcome_pending: PendingBudget,
23    /// Total retained identity-envelope budget across identity topics.
24    pub identity_pending: PendingBudget,
25    /// Per-topic row limit, applied alongside the total kind budget.
26    pub max_pending_rows_per_topic: u64,
27    /// Per-topic byte limit; rejected chunks do not advance F.
28    pub max_pending_bytes_per_topic: u64,
29    /// Maximum concurrent dependency and fixed-target requests.
30    pub max_dependency_requests: usize,
31    /// Maximum stored candidates loaded by one local-delivery read.
32    pub max_local_read_rows: u32,
33    /// Maximum candidate bytes checked before loading message blobs.
34    pub max_local_read_bytes: u64,
35    /// Bound on stream-first receipt waits. Explicit sync queries without this wait.
36    pub receiver_fallback_interval: Duration,
37    /// Poll interval for cross-process progress and missed local wake events.
38    pub active_database_poll_interval: Duration,
39    /// Default-reader ownership expires after this interval without renewal.
40    pub default_consumer_lease_duration: Duration,
41    /// Healthy-primary wait before an absent exact identity reference is rejected.
42    pub identity_reference_wait: Duration,
43    /// One budget for target capture, receipt, processing, and required send follow-up work.
44    pub barrier_timeout: Duration,
45    /// Interval between full rescans of blocked Welcome rows.
46    pub blocked_welcome_rescan_interval: Duration,
47    /// First delay before retrying a source that failed with a permanent error.
48    pub permanent_retry_initial: Duration,
49    /// Longest delay between retries of a repeatedly failing source.
50    pub permanent_retry_max: Duration,
51}
52
53impl Default for StreamPolicy {
54    fn default() -> Self {
55        Self {
56            max_admission_rows: STREAM_BATCH_ROWS,
57            max_admission_bytes: STREAM_BATCH_BYTES,
58            max_fetched_rows: STREAM_BATCH_ROWS,
59            max_fetched_bytes: STREAM_BATCH_BYTES,
60            group_pending: PendingBudget {
61                rows: STREAM_GROUP_ROWS,
62                bytes: STREAM_GROUP_BYTES,
63            },
64            welcome_pending: PendingBudget {
65                rows: STREAM_WELCOME_ROWS,
66                bytes: STREAM_WELCOME_BYTES,
67            },
68            identity_pending: PendingBudget {
69                rows: STREAM_IDENTITY_ROWS,
70                bytes: STREAM_IDENTITY_BYTES,
71            },
72            max_pending_rows_per_topic: STREAM_TOPIC_ROWS,
73            max_pending_bytes_per_topic: STREAM_TOPIC_BYTES,
74            max_dependency_requests: IDENTITY_DEPENDENCY_CONCURRENCY,
75            max_local_read_rows: STREAM_LOCAL_READ_ROWS,
76            max_local_read_bytes: STREAM_LOCAL_READ_BYTES,
77            receiver_fallback_interval: RECEIVER_FALLBACK_INTERVAL,
78            active_database_poll_interval: ACTIVE_DATABASE_POLL_INTERVAL,
79            default_consumer_lease_duration: DEFAULT_CONSUMER_LEASE_DURATION,
80            identity_reference_wait: IDENTITY_REFERENCE_WAIT,
81            barrier_timeout: STREAM_BARRIER_TIMEOUT,
82            blocked_welcome_rescan_interval: STREAM_BLOCKED_WELCOME_RESCAN_INTERVAL,
83            permanent_retry_initial: STREAM_PERMANENT_RETRY_INITIAL,
84            permanent_retry_max: STREAM_PERMANENT_RETRY_MAX,
85        }
86    }
87}
88
89impl StreamPolicy {
90    /// The bounds of one fetched page.
91    pub fn fetched_limits(&self) -> IncomingBatchLimits {
92        IncomingBatchLimits {
93            max_rows: usize::try_from(self.max_fetched_rows).unwrap_or(usize::MAX),
94            max_bytes: usize::try_from(self.max_fetched_bytes).unwrap_or(usize::MAX),
95        }
96    }
97
98    /// The incoming limits for one entity kind.
99    pub fn incoming_limits(&self, kind: NetworkEntityKind) -> IncomingLimits {
100        IncomingLimits {
101            batch: PendingBudget {
102                rows: u64::from(self.max_admission_rows),
103                bytes: self.max_admission_bytes,
104            },
105            topic: PendingBudget {
106                rows: self.max_pending_rows_per_topic,
107                bytes: self.max_pending_bytes_per_topic,
108            },
109            kind: match kind {
110                NetworkEntityKind::Group => self.group_pending,
111                NetworkEntityKind::Welcome => self.welcome_pending,
112                NetworkEntityKind::Identity => self.identity_pending,
113            },
114        }
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use super::*;
121
122    impl StreamPolicy {
123        fn validate(&self) -> Result<(), &'static str> {
124            let positive = [
125                ("max_admission_rows", u64::from(self.max_admission_rows)),
126                ("max_admission_bytes", self.max_admission_bytes),
127                ("max_fetched_rows", u64::from(self.max_fetched_rows)),
128                ("max_fetched_bytes", self.max_fetched_bytes),
129                ("group_pending.rows", self.group_pending.rows),
130                ("group_pending.bytes", self.group_pending.bytes),
131                ("welcome_pending.rows", self.welcome_pending.rows),
132                ("welcome_pending.bytes", self.welcome_pending.bytes),
133                ("identity_pending.rows", self.identity_pending.rows),
134                ("identity_pending.bytes", self.identity_pending.bytes),
135                (
136                    "max_pending_rows_per_topic",
137                    self.max_pending_rows_per_topic,
138                ),
139                (
140                    "max_pending_bytes_per_topic",
141                    self.max_pending_bytes_per_topic,
142                ),
143                (
144                    "max_dependency_requests",
145                    self.max_dependency_requests as u64,
146                ),
147                ("max_local_read_rows", u64::from(self.max_local_read_rows)),
148                ("max_local_read_bytes", self.max_local_read_bytes),
149            ];
150            for (name, value) in positive {
151                if value == 0 || value > i64::MAX as u64 {
152                    return Err(name);
153                }
154            }
155            for (name, value) in [
156                (
157                    "receiver_fallback_interval",
158                    self.receiver_fallback_interval,
159                ),
160                (
161                    "active_database_poll_interval",
162                    self.active_database_poll_interval,
163                ),
164                (
165                    "default_consumer_lease_duration",
166                    self.default_consumer_lease_duration,
167                ),
168                ("identity_reference_wait", self.identity_reference_wait),
169                ("barrier_timeout", self.barrier_timeout),
170                (
171                    "blocked_welcome_rescan_interval",
172                    self.blocked_welcome_rescan_interval,
173                ),
174                ("permanent_retry_initial", self.permanent_retry_initial),
175                ("permanent_retry_max", self.permanent_retry_max),
176            ] {
177                // The same timer values must work on native and browser clients.
178                if value.is_zero() || value.as_millis() > i32::MAX as u128 {
179                    return Err(name);
180                }
181            }
182            if self.active_database_poll_interval >= self.default_consumer_lease_duration {
183                return Err("default_consumer_lease_duration");
184            }
185            if self.permanent_retry_initial > self.permanent_retry_max {
186                return Err("permanent_retry_max");
187            }
188            if self.identity_reference_wait
189                <= Duration::from_millis(BACKEND_DEFAULT_STATEMENT_TIMEOUT_MS)
190            {
191                return Err("identity_reference_wait");
192            }
193            Ok(())
194        }
195    }
196
197    #[xmtp_common::test(unwrap_try = true)]
198    fn internal_limits_fit_storage_and_timer_ranges() {
199        StreamPolicy::default().validate()?;
200    }
201}
xmtp_mls_core/policy.rs

xmtp_mls_core/
policy.rs
