xmtp_db/encrypted_store/
stream_storage.rs

1//! Shared transaction and error types for durable client streams.
2
3use diesel::{Connection, SqliteConnection, connection::TransactionManager};
4use thiserror::Error;
5use xmtp_common::{ErrorCode, RetryableError};
6
7use crate::{ConnectionExt, StorageError};
8
9/// The independent capacity check that rejected an otherwise valid admission.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum BudgetScope {
12    /// One fetched or admitted batch.
13    Batch,
14    /// All pending rows for one topic.
15    Topic,
16    /// All pending rows of one log kind; dependency kinds keep separate capacity.
17    Kind,
18}
19
20/// Stable storage failures that preserve receipt, processing, and delivery invariants.
21#[derive(Debug, Error, ErrorCode)]
22pub enum StreamStorageError {
23    /// The batch does not have valid, strictly increasing envelope IDs.
24    #[error("The ordered envelope batch is invalid")]
25    InvalidBatch,
26    /// The source omitted a prefix not yet stored in this database.
27    #[error("The source starts at {after}, beyond received position {received}")]
28    MissingPrefix { after: u64, received: u64 },
29    /// Existing progress was not written by ordered admission.
30    #[error("Network progress has no proven received prefix")]
31    UninitializedNetworkProgress,
32    /// Pending work exceeds its row or byte budget. Retry after processing drains it.
33    #[error("The {scope:?} pending budget is full")]
34    Capacity { scope: BudgetScope },
35    /// Another processor changed the pending head. Reload state before retrying.
36    #[error("The pending envelope is no longer the topic head")]
37    HeadChanged,
38    /// Installing this welcome would rewind or replace processed state.
39    #[error("The join anchor does not advance processed progress")]
40    StaleJoinAnchor,
41    /// Another default consumer holds an unexpired lease.
42    #[error("A default message consumer is already active")]
43    AlreadyActive,
44    /// The lease expired or a new consumer acquired ownership.
45    #[error("The default message consumer no longer owns delivery")]
46    NotCurrentOwner,
47    /// The cursor was issued before restore or by another database.
48    #[error("The delivery cursor belongs to another database")]
49    ForeignCursor,
50    /// The persistent local message counter cannot allocate another number.
51    #[error("The delivery sequence allocator is exhausted")]
52    DeliveryExhausted,
53    /// The next retained message exceeds the local read byte limit. Not retryable.
54    #[error("The next local message needs {bytes} bytes, above the {limit} byte limit")]
55    LocalReadCapacity { bytes: u64, limit: u64 },
56    /// The cursor is ahead of local history or the lease interval is empty.
57    #[error("The delivery cursor or lease is invalid")]
58    InvalidDeliveryPosition,
59}
60
61impl RetryableError for StreamStorageError {
62    fn is_retryable(&self) -> bool {
63        matches!(self, Self::Capacity { .. } | Self::HeadChanged)
64    }
65}
66
67/// Acquire the database writer before reading mutable stream state.
68/// Nested calls use a savepoint under the writer already held by the caller.
69pub(crate) fn stream_transaction<C, T>(
70    connection: &C,
71    work: impl FnOnce(&mut SqliteConnection) -> Result<T, StorageError>,
72) -> Result<T, StorageError>
73where
74    C: ConnectionExt,
75{
76    connection.raw_query(|conn| {
77        let nested =
78            <SqliteConnection as Connection>::TransactionManager::transaction_manager_status_mut(
79                conn,
80            )
81            .transaction_depth()?
82            .is_some();
83        Ok(if nested {
84            conn.transaction(work)
85        } else {
86            conn.immediate_transaction(work)
87        })
88    })?
89}
xmtp_db/encrypted_store/stream_storage.rs

xmtp_db/encrypted_store/
stream_storage.rs
