coven_replication/sync/store/
error.rs

1use coven_protocol::objects::StoreObjectError;
2use coven_protocol::store_commit::{StoreBatchCommitRef, StoreDeviceId};
3
4#[derive(Debug, thiserror::Error)]
5pub enum BlobPreparationCleanupError {
6    #[error("prepared blob spool is absent: {}", path.display())]
7    MissingSpool { path: std::path::PathBuf },
8    #[error("prepared blob file: {0}")]
9    File(#[from] coven_foundation::atomic_file::FileError),
10}
11
12#[derive(Debug)]
13pub struct BlobPreparationRollback {
14    operation: Box<StoreError>,
15    cleanup: Vec<BlobPreparationCleanupError>,
16}
17
18impl BlobPreparationRollback {
19    pub(crate) fn new(operation: StoreError, cleanup: Vec<BlobPreparationCleanupError>) -> Self {
20        Self {
21            operation: Box::new(operation),
22            cleanup,
23        }
24    }
25}
26
27impl std::fmt::Display for BlobPreparationRollback {
28    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29        write!(formatter, "blob preparation failed: {}", self.operation)?;
30        for cleanup in &self.cleanup {
31            write!(formatter, "; cleanup failed: {cleanup}")?;
32        }
33        Ok(())
34    }
35}
36
37impl std::error::Error for BlobPreparationRollback {
38    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
39        Some(self.operation.as_ref())
40    }
41}
42
43#[derive(Debug)]
44pub enum StorePreparationError {
45    Database(coven_database::DbError),
46    Gate(String),
47    AssetScan(String),
48    AssetScanFile(coven_foundation::store_dir::LocalBlobStoreError),
49    AssetUpload(String),
50    Storage {
51        operation: &'static str,
52        source: coven_protocol::objects::StorageError,
53    },
54    LocalUserBlob {
55        namespace: String,
56        id: String,
57    },
58    MissingPreparedBlob {
59        namespace: String,
60        id: String,
61    },
62}
63
64impl std::fmt::Display for StorePreparationError {
65    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
66        match self {
67            Self::Database(error) => write!(f, "database error: {error}"),
68            Self::Gate(error) => write!(f, "gate error: {error}"),
69            Self::AssetScan(error) => write!(f, "asset scan error: {error}"),
70            Self::AssetScanFile(error) => write!(f, "asset scan error: {error}"),
71            Self::AssetUpload(error) => write!(f, "asset upload error: {error}"),
72            Self::Storage { operation, source } => write!(f, "{operation}: {source}"),
73            Self::LocalUserBlob { namespace, id } => {
74                write!(
75                    f,
76                    "user-provided blob {namespace}/{id} still has a local external ref"
77                )
78            }
79            Self::MissingPreparedBlob { namespace, id } => {
80                write!(
81                    f,
82                    "blob {namespace}/{id} has no prepared exact publication object"
83                )
84            }
85        }
86    }
87}
88
89impl std::error::Error for StorePreparationError {
90    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
91        match self {
92            Self::Database(source) => Some(source),
93            Self::Storage { source, .. } => Some(source),
94            Self::AssetScanFile(source) => Some(source),
95            Self::Gate(_)
96            | Self::AssetScan(_)
97            | Self::AssetUpload(_)
98            | Self::LocalUserBlob { .. }
99            | Self::MissingPreparedBlob { .. } => None,
100        }
101    }
102}
103
104#[derive(Debug, thiserror::Error)]
105pub enum StoreError {
106    #[error("database: {0}")]
107    Database(#[from] coven_database::DbError),
108    #[error("local file: {0}")]
109    File(#[from] coven_foundation::atomic_file::FileError),
110    #[error("inspect host blob source {}: {source}", path.display())]
111    InspectBlobSource {
112        path: std::path::PathBuf,
113        #[source]
114        source: std::io::Error,
115    },
116    #[error("blob cache: {0}")]
117    BlobCache(#[from] crate::sync::BlobCacheError),
118    #[error("{0}")]
119    Object(#[from] StoreObjectError),
120    #[error("Store protocol: {0}")]
121    Protocol(#[from] coven_protocol::store_commit::StoreProtocolError),
122    #[error("Store JSON: {0}")]
123    Json(#[from] serde_json::Error),
124    #[error("Store changeset: {0}")]
125    Changeset(#[from] coven_database::ChangesetError),
126    #[error("Store writer authorization: {0}")]
127    WriterAuthorization(#[source] Box<crate::sync::store::StoreWriterAuthorizationError>),
128    #[error("Store sync cycle: {0}")]
129    SyncCycle(#[source] Box<crate::sync::cycle::SyncCycleFailure>),
130    #[error("Store membership chain: {0}")]
131    AnchoredChain(#[source] Box<crate::sync::store::AnchoredChainError>),
132    #[error("Store protocol root: {0}")]
133    ProtocolRoot(#[source] Box<crate::sync::store::protocol_root::StoreProtocolRootError>),
134    #[error("Store audience package: {0}")]
135    AudiencePackage(#[from] coven_protocol::audience_package::AudiencePackageError),
136    #[error("Store blob path: {0}")]
137    BlobPath(#[from] coven_foundation::store_dir::PathTokenError),
138    #[error("Store remote object: {0}")]
139    RemoteObject(#[from] coven_protocol::remote_object::RemoteObjectRecordError),
140    #[error("Store blob locator: {0}")]
141    BlobLocator(#[from] coven_protocol::blob::locator::BlobLocatorError),
142    #[error("Store prepared commit: {0}")]
143    PreparedCommit(#[source] coven_protocol::prepared_commit::PreparedCommitError),
144    #[error("Store membership preparation: {0}")]
145    MembershipPreparation(
146        #[source] coven_protocol::membership_mutation::MembershipPreparationError,
147    ),
148    #[error("Store keyring: {0}")]
149    Keyring(#[source] Box<crate::sync::store::MembershipMutationError>),
150    #[error("Store row routing key: {0}")]
151    RowRoutingKey(#[from] coven_protocol::circle::RowRoutingKeyError),
152    #[error("Store Circle package: {0}")]
153    CirclePackage(#[source] Box<crate::sync::store::CirclePackageReadError>),
154    #[error("Store protocol state {key:?} is absent")]
155    MissingState { key: &'static str },
156    #[error("Store protocol state {key:?} is invalid: {reason}")]
157    InvalidState { key: &'static str, reason: String },
158    #[error("outbound Store row is invalid: {0}")]
159    InvalidOutbound(String),
160    /// Another writer took the activation slot between this operation's
161    /// preparation and its publication, so the candidate had to be re-prepared
162    /// and **nothing was persisted**. Re-derive from durable state and run the
163    /// operation again; it is the ordinary outcome of two writers racing, not a
164    /// damaged store.
165    #[error("another writer activated first; this Store operation persisted nothing")]
166    ActivationConflict,
167    #[error("outbound Store preparation failed: {0}")]
168    Preparation(#[source] StorePreparationError),
169    #[error("{0}")]
170    BlobPreparationRollback(#[from] BlobPreparationRollback),
171    #[error("blob preparation cleanup: {0}")]
172    BlobPreparationCleanup(#[from] BlobPreparationCleanupError),
173    #[error("outbound blob {namespace}/{id} is local and cannot be published")]
174    LocalUserBlob { namespace: String, id: String },
175    #[error("outbound blob {namespace}/{id} is absent from storage")]
176    MissingBlob { namespace: String, id: String },
177    #[error("checking outbound blob {namespace}/{id}: {source}")]
178    BlobStorage {
179        namespace: String,
180        id: String,
181        source: coven_protocol::objects::StorageError,
182    },
183    #[error("Store pull: {0}")]
184    Pull(#[from] crate::sync::store::pull::StorePullError),
185    #[error("Store sequence {current} has no representable successor")]
186    SequenceExhausted { current: u64 },
187    #[error("published Store write count has no representable successor")]
188    PublishCountExhausted,
189    /// Preparation failed AND recording that write's blocked status failed, so
190    /// the write is not marked blocked. Carries both failures rather than
191    /// reporting one and describing the other.
192    #[error("write {write_id} was not marked blocked ({status}) after it failed to prepare ({preparation})")]
193    WriteBlockNotRecorded {
194        write_id: coven_protocol::write::WriteId,
195        preparation: Box<StoreError>,
196        status: coven_database::DbError,
197    },
198    #[error("Store author {device_id} was excluded before candidate activation")]
199    AuthorExcluded { device_id: StoreDeviceId },
200    #[error("Merge announcement selected {actual:?}, not candidate {expected:?}")]
201    MergeAnnouncementOccupied {
202        expected: Box<StoreBatchCommitRef>,
203        actual: Box<StoreBatchCommitRef>,
204    },
205    #[error("{0}")]
206    CirclePublicationBlocked(coven_protocol::circle::CirclePublicationBlocked),
207}
208
209impl StoreError {
210    /// A retained prepared object that opens to different bytes is invalid
211    /// outbound state, not a provider failure.
212    pub(crate) fn prepared_object(error: coven_protocol::objects::StorageError) -> Self {
213        match error {
214            coven_protocol::objects::StorageError::PreparedObjectMismatch(key) => {
215                Self::InvalidOutbound(format!(
216                    "prepared exact object {key} differs from its signed bytes"
217                ))
218            }
219            error => StoreObjectError::from(error).into(),
220        }
221    }
222
223    pub(crate) fn write_block(&self) -> Option<coven_protocol::write::WriteBlock> {
224        match self {
225            Self::Database(_)
226            | Self::File(_)
227            | Self::InspectBlobSource { .. }
228            | Self::BlobCache(_)
229            | Self::BlobPreparationRollback(_)
230            | Self::BlobPreparationCleanup(_)
231            | Self::WriteBlockNotRecorded { .. }
232            | Self::BlobStorage { .. }
233            // Nothing was persisted and the caller re-runs the operation, so this
234            // blocks no writer.
235            | Self::ActivationConflict
236            | Self::Pull(_)
237            | Self::SyncCycle(_) => None,
238            Self::MergeAnnouncementOccupied { .. }
239            | Self::SequenceExhausted { .. }
240            | Self::PublishCountExhausted
241            | Self::AuthorExcluded { .. } => Some(coven_protocol::write::WriteBlock::InvalidProtocolState {
242                reason: self.to_string(),
243            }),
244            Self::CirclePublicationBlocked(
245                coven_protocol::circle::CirclePublicationBlocked::RotationRequired {
246                    circle_id,
247                    removed_members,
248                },
249            ) => Some(coven_protocol::write::WriteBlock::RotationRequired {
250                circle_id: *circle_id,
251                removed_members: removed_members.clone(),
252            }),
253            Self::Object(StoreObjectError::Storage(_)) => None,
254            Self::MissingBlob { namespace, id } => Some(coven_protocol::write::WriteBlock::MissingBlob {
255                namespace: namespace.clone(),
256                id: id.clone(),
257            }),
258            Self::LocalUserBlob { namespace, id } => Some(coven_protocol::write::WriteBlock::LocalUserBlob {
259                namespace: namespace.clone(),
260                id: id.clone(),
261            }),
262            Self::MissingState { key } => Some(coven_protocol::write::WriteBlock::InvalidProtocolState {
263                reason: format!("Store protocol state {key:?} is absent"),
264            }),
265            Self::InvalidState { key, reason } => Some(coven_protocol::write::WriteBlock::InvalidProtocolState {
266                reason: format!("Store protocol state {key:?} is invalid: {reason}"),
267            }),
268            Self::InvalidOutbound(_)
269            | Self::Object(_)
270            | Self::Protocol(_)
271            | Self::Json(_)
272            | Self::Changeset(_)
273            | Self::WriterAuthorization(_)
274            | Self::AnchoredChain(_)
275            | Self::ProtocolRoot(_)
276            | Self::AudiencePackage(_)
277            | Self::BlobPath(_)
278            | Self::RemoteObject(_)
279            | Self::BlobLocator(_)
280            | Self::PreparedCommit(_)
281            | Self::MembershipPreparation(_)
282            | Self::Keyring(_)
283            | Self::RowRoutingKey(_)
284            | Self::CirclePackage(_) => {
285                Some(coven_protocol::write::WriteBlock::InvalidPackage {
286                    reason: self.to_string(),
287                })
288            }
289            Self::Preparation(StorePreparationError::LocalUserBlob { namespace, id }) => {
290                Some(coven_protocol::write::WriteBlock::LocalUserBlob {
291                    namespace: namespace.clone(),
292                    id: id.clone(),
293                })
294            }
295            Self::Preparation(StorePreparationError::MissingPreparedBlob { namespace, id }) => {
296                Some(coven_protocol::write::WriteBlock::MissingBlob {
297                    namespace: namespace.clone(),
298                    id: id.clone(),
299                })
300            }
301            Self::Preparation(StorePreparationError::Gate(_))
302            | Self::Preparation(StorePreparationError::AssetScan(_))
303            | Self::Preparation(StorePreparationError::AssetScanFile(_))
304            | Self::Preparation(StorePreparationError::Database(_)) => {
305                Some(coven_protocol::write::WriteBlock::InvalidPackage {
306                    reason: self.to_string(),
307                })
308            }
309            Self::Preparation(StorePreparationError::AssetUpload(_))
310            | Self::Preparation(StorePreparationError::Storage { .. }) => None,
311        }
312    }
313}
314
315impl From<crate::sync::store::CirclePackageReadError> for StoreError {
316    fn from(error: crate::sync::store::CirclePackageReadError) -> Self {
317        Self::CirclePackage(Box::new(error))
318    }
319}
320
321impl From<coven_protocol::prepared_commit::PreparedCommitError> for StoreError {
322    fn from(error: coven_protocol::prepared_commit::PreparedCommitError) -> Self {
323        StoreError::PreparedCommit(error)
324    }
325}
326
327impl From<coven_protocol::membership_mutation::MembershipPreparationError> for StoreError {
328    fn from(error: coven_protocol::membership_mutation::MembershipPreparationError) -> Self {
329        StoreError::MembershipPreparation(error)
330    }
331}
332
333impl From<crate::sync::store::MembershipMutationError> for StoreError {
334    fn from(error: crate::sync::store::MembershipMutationError) -> Self {
335        Self::Keyring(Box::new(error))
336    }
337}
338
339impl From<crate::sync::store::StoreWriterAuthorizationError> for StoreError {
340    fn from(error: crate::sync::store::StoreWriterAuthorizationError) -> Self {
341        Self::WriterAuthorization(Box::new(error))
342    }
343}
344
345impl From<crate::sync::cycle::SyncCycleFailure> for StoreError {
346    fn from(error: crate::sync::cycle::SyncCycleFailure) -> Self {
347        Self::SyncCycle(Box::new(error))
348    }
349}
350
351impl From<crate::sync::store::AnchoredChainError> for StoreError {
352    fn from(error: crate::sync::store::AnchoredChainError) -> Self {
353        Self::AnchoredChain(Box::new(error))
354    }
355}
356
357impl From<crate::sync::store::protocol_root::StoreProtocolRootError> for StoreError {
358    fn from(error: crate::sync::store::protocol_root::StoreProtocolRootError) -> Self {
359        Self::ProtocolRoot(Box::new(error))
360    }
361}
coven_replication/sync/store/error.rs

coven_replication/sync/store/
error.rs
