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 publication waits for accepted history: {0:?}")]
186    PublicationHeld(Vec<crate::sync::store::pull::HeldStorePosition>),
187    #[error("Store publication response was uncertain ({publication}); verifying its outcome failed: {verification}")]
188    PublicationSettlement {
189        publication: coven_protocol::objects::StorageError,
190        #[source]
191        verification: Box<StoreError>,
192    },
193    #[error("Store sequence {current} has no representable successor")]
194    SequenceExhausted { current: u64 },
195    #[error("published Store write count has no representable successor")]
196    PublishCountExhausted,
197    /// A write operation failed AND recording its blocked status failed, so
198    /// the write is not marked blocked. Carries both failures rather than
199    /// reporting one and describing the other.
200    #[error("write {write_id} was not marked blocked ({status}) after its operation failed ({operation})")]
201    WriteBlockNotRecorded {
202        write_id: coven_protocol::write::WriteId,
203        #[source]
204        operation: Box<StoreError>,
205        status: coven_database::DbError,
206    },
207    #[error("Store author {device_id} was excluded before candidate activation")]
208    AuthorExcluded { device_id: StoreDeviceId },
209    #[error("Merge announcement selected {actual:?}, not candidate {expected:?}")]
210    MergeAnnouncementOccupied {
211        expected: Box<StoreBatchCommitRef>,
212        actual: Box<StoreBatchCommitRef>,
213    },
214    #[error("{0}")]
215    CirclePublicationBlocked(coven_protocol::circle::CirclePublicationBlocked),
216}
217
218impl StoreError {
219    /// A retained prepared object that opens to different bytes is invalid
220    /// outbound state, not a provider failure.
221    pub(crate) fn prepared_object(error: coven_protocol::objects::StorageError) -> Self {
222        match error {
223            coven_protocol::objects::StorageError::PreparedObjectMismatch(key) => {
224                Self::InvalidOutbound(format!(
225                    "prepared exact object {key} differs from its signed bytes"
226                ))
227            }
228            error => StoreObjectError::from(error).into(),
229        }
230    }
231
232    pub(crate) fn write_block(
233        &self,
234        attempted_write: &coven_protocol::write::WriteId,
235    ) -> Option<(
236        coven_protocol::write::WriteId,
237        coven_protocol::write::WriteBlock,
238    )> {
239        let mut cause: &(dyn std::error::Error + 'static) = self;
240        loop {
241            if let Some(database) = cause.downcast_ref::<coven_database::DbError>() {
242                if let Some(conflict) = database.write_rebase_conflict() {
243                    return Some((
244                        conflict.write_id.clone(),
245                        coven_protocol::write::WriteBlock::RebaseConflict(conflict.clone()),
246                    ));
247                }
248                // The database owner distinguishes its operation failure from
249                // secondary cleanup failures; keep that decision at its owner.
250                break;
251            }
252            let Some(source) = cause.source() else {
253                break;
254            };
255            cause = source;
256        }
257        let block = match self {
258            Self::Preparation(StorePreparationError::Database(_)) => Some(
259                coven_protocol::write::WriteBlock::InvalidPackage { reason: self.to_string() },
260            ),
261            Self::Database(_)
262            | Self::File(_)
263            | Self::InspectBlobSource { .. }
264            | Self::BlobCache(_)
265            | Self::BlobPreparationRollback(_)
266            | Self::BlobPreparationCleanup(_)
267            | Self::WriteBlockNotRecorded { .. }
268            | Self::BlobStorage { .. }
269            // Nothing was persisted and the caller re-runs the operation, so this
270            // blocks no writer.
271            | Self::ActivationConflict
272            | Self::Pull(_)
273            | Self::PublicationHeld(_)
274            | Self::PublicationSettlement { .. }
275            | Self::SyncCycle(_) => None,
276            Self::MergeAnnouncementOccupied { .. }
277            | Self::SequenceExhausted { .. }
278            | Self::PublishCountExhausted
279            | Self::AuthorExcluded { .. } => Some(coven_protocol::write::WriteBlock::InvalidProtocolState {
280                reason: self.to_string(),
281            }),
282            Self::CirclePublicationBlocked(
283                coven_protocol::circle::CirclePublicationBlocked::RotationRequired {
284                    circle_id,
285                    removed_members,
286                },
287            ) => Some(coven_protocol::write::WriteBlock::RotationRequired {
288                circle_id: *circle_id,
289                removed_members: removed_members.clone(),
290            }),
291            Self::Object(StoreObjectError::Storage(_)) => None,
292            Self::MissingBlob { namespace, id } => Some(coven_protocol::write::WriteBlock::MissingBlob {
293                namespace: namespace.clone(),
294                id: id.clone(),
295            }),
296            Self::LocalUserBlob { namespace, id } => Some(coven_protocol::write::WriteBlock::LocalUserBlob {
297                namespace: namespace.clone(),
298                id: id.clone(),
299            }),
300            Self::MissingState { key } => Some(coven_protocol::write::WriteBlock::InvalidProtocolState {
301                reason: format!("Store protocol state {key:?} is absent"),
302            }),
303            Self::InvalidState { key, reason } => Some(coven_protocol::write::WriteBlock::InvalidProtocolState {
304                reason: format!("Store protocol state {key:?} is invalid: {reason}"),
305            }),
306            Self::InvalidOutbound(_)
307            | Self::Object(_)
308            | Self::Protocol(_)
309            | Self::Json(_)
310            | Self::Changeset(_)
311            | Self::WriterAuthorization(_)
312            | Self::AnchoredChain(_)
313            | Self::ProtocolRoot(_)
314            | Self::AudiencePackage(_)
315            | Self::BlobPath(_)
316            | Self::RemoteObject(_)
317            | Self::BlobLocator(_)
318            | Self::PreparedCommit(_)
319            | Self::MembershipPreparation(_)
320            | Self::Keyring(_)
321            | Self::RowRoutingKey(_)
322            | Self::CirclePackage(_) => {
323                Some(coven_protocol::write::WriteBlock::InvalidPackage {
324                    reason: self.to_string(),
325                })
326            }
327            Self::Preparation(StorePreparationError::LocalUserBlob { namespace, id }) => {
328                Some(coven_protocol::write::WriteBlock::LocalUserBlob {
329                    namespace: namespace.clone(),
330                    id: id.clone(),
331                })
332            }
333            Self::Preparation(StorePreparationError::MissingPreparedBlob { namespace, id }) => {
334                Some(coven_protocol::write::WriteBlock::MissingBlob {
335                    namespace: namespace.clone(),
336                    id: id.clone(),
337                })
338            }
339            Self::Preparation(StorePreparationError::Gate(_))
340            | Self::Preparation(StorePreparationError::AssetScan(_))
341            | Self::Preparation(StorePreparationError::AssetScanFile(_)) => {
342                Some(coven_protocol::write::WriteBlock::InvalidPackage {
343                    reason: self.to_string(),
344                })
345            }
346            Self::Preparation(StorePreparationError::AssetUpload(_))
347            | Self::Preparation(StorePreparationError::Storage { .. }) => None,
348        };
349        block.map(|block| (attempted_write.clone(), block))
350    }
351}
352
353impl From<crate::sync::store::CirclePackageReadError> for StoreError {
354    fn from(error: crate::sync::store::CirclePackageReadError) -> Self {
355        Self::CirclePackage(Box::new(error))
356    }
357}
358
359impl From<coven_protocol::prepared_commit::PreparedCommitError> for StoreError {
360    fn from(error: coven_protocol::prepared_commit::PreparedCommitError) -> Self {
361        StoreError::PreparedCommit(error)
362    }
363}
364
365impl From<coven_protocol::membership_mutation::MembershipPreparationError> for StoreError {
366    fn from(error: coven_protocol::membership_mutation::MembershipPreparationError) -> Self {
367        StoreError::MembershipPreparation(error)
368    }
369}
370
371impl From<crate::sync::store::MembershipMutationError> for StoreError {
372    fn from(error: crate::sync::store::MembershipMutationError) -> Self {
373        Self::Keyring(Box::new(error))
374    }
375}
376
377impl From<crate::sync::store::StoreWriterAuthorizationError> for StoreError {
378    fn from(error: crate::sync::store::StoreWriterAuthorizationError) -> Self {
379        Self::WriterAuthorization(Box::new(error))
380    }
381}
382
383impl From<crate::sync::cycle::SyncCycleFailure> for StoreError {
384    fn from(error: crate::sync::cycle::SyncCycleFailure) -> Self {
385        Self::SyncCycle(Box::new(error))
386    }
387}
388
389impl From<crate::sync::store::AnchoredChainError> for StoreError {
390    fn from(error: crate::sync::store::AnchoredChainError) -> Self {
391        Self::AnchoredChain(Box::new(error))
392    }
393}
394
395impl From<crate::sync::store::protocol_root::StoreProtocolRootError> for StoreError {
396    fn from(error: crate::sync::store::protocol_root::StoreProtocolRootError) -> Self {
397        Self::ProtocolRoot(Box::new(error))
398    }
399}
400
401#[cfg(test)]
402mod tests {
403    use super::*;
404    use coven_protocol::write::{
405        AffectedRow, WriteBlock, WriteId, WriteRebaseConflict, WriteRebaseConflictReason,
406    };
407
408    #[test]
409    fn write_rebase_conflicts_remain_typed_at_both_publication_boundaries() {
410        let conflict = WriteRebaseConflict {
411            write_id: WriteId::from_generated("pending-write".into()),
412            affected_rows: vec![AffectedRow {
413                table: "notes".into(),
414                primary_key: "private-note".into(),
415            }],
416            reason: WriteRebaseConflictReason::PrivateShared,
417        };
418        let wrapped = || {
419            coven_database::DbError::context(
420                "rebase pending writes",
421                coven_database::DbError::from(conflict.clone()),
422            )
423        };
424        let attempted = WriteId::from_generated("attempted-write".into());
425        for error in [
426            StoreError::Database(wrapped()),
427            StoreError::Preparation(StorePreparationError::Database(wrapped())),
428        ] {
429            assert_eq!(
430                error.write_block(&attempted),
431                Some((
432                    conflict.write_id.clone(),
433                    WriteBlock::RebaseConflict(conflict.clone())
434                ))
435            );
436        }
437    }
438
439    #[test]
440    fn write_rebase_conflicts_remain_typed_through_pull_and_cycle_causes() {
441        let conflict = WriteRebaseConflict {
442            write_id: WriteId::from_generated("dependent-write".into()),
443            affected_rows: vec![AffectedRow {
444                table: "notes".into(),
445                primary_key: "conflicting-note".into(),
446            }],
447            reason: WriteRebaseConflictReason::Constraint {
448                message: "CHECK constraint failed".into(),
449            },
450        };
451        let pull = || {
452            crate::sync::store::StorePullError::context(
453                "install accepted snapshot",
454                coven_database::DbError::context(
455                    "rebase pending suffix",
456                    coven_database::DbError::from(conflict.clone()),
457                ),
458            )
459        };
460        let attempted = WriteId::from_generated("attempted-write".into());
461        for error in [
462            StoreError::Pull(pull()),
463            StoreError::from(crate::sync::cycle::SyncCycleFailure::operation(
464                "install publication winner",
465                pull(),
466            )),
467            StoreError::PublicationSettlement {
468                publication: coven_protocol::objects::StorageError::Storage(
469                    "publication response unavailable".into(),
470                ),
471                verification: Box::new(StoreError::Pull(pull())),
472            },
473        ] {
474            assert_eq!(
475                error.write_block(&attempted),
476                Some((
477                    conflict.write_id.clone(),
478                    WriteBlock::RebaseConflict(conflict.clone())
479                )),
480                "{error}"
481            );
482        }
483    }
484
485    #[test]
486    fn transient_pull_and_cycle_causes_do_not_block_a_write() {
487        let attempted = WriteId::from_generated("attempted-write".into());
488        let pull = || {
489            crate::sync::store::StorePullError::context(
490                "load publication winner",
491                crate::sync::store::StorePullError::Storage(
492                    coven_protocol::objects::StorageError::Storage("provider unavailable".into()),
493                ),
494            )
495        };
496        for error in [
497            StoreError::Pull(pull()),
498            StoreError::from(crate::sync::cycle::SyncCycleFailure::operation(
499                "install publication winner",
500                pull(),
501            )),
502            StoreError::PublicationSettlement {
503                publication: coven_protocol::objects::StorageError::Storage(
504                    "publication response unavailable".into(),
505                ),
506                verification: Box::new(StoreError::Pull(pull())),
507            },
508        ] {
509            assert_eq!(error.write_block(&attempted), None, "{error}");
510        }
511    }
512
513    #[test]
514    fn unattributed_write_fault_blocks_the_attempted_write() {
515        let attempted = WriteId::from_generated("attempted-write".into());
516        let error = StoreError::MissingBlob {
517            namespace: "documents".into(),
518            id: "missing-content".into(),
519        };
520        assert_eq!(
521            error.write_block(&attempted),
522            Some((
523                attempted,
524                WriteBlock::MissingBlob {
525                    namespace: "documents".into(),
526                    id: "missing-content".into(),
527                }
528            ))
529        );
530    }
531}
coven_replication/sync/store/error.rs

coven_replication/sync/store/
error.rs
