coven_replication/blob/
mod.rs

1//! Blob domain workflows: locality transitions, tombstone lifecycle, retry
2//! policy, and local cleanup. The blob value model — references, locators,
3//! scopes, transfer limits, and the transition observer port — lives in
4//! [`coven_protocol::blob`]; upload execution outcomes live here with the
5//! database and filesystem errors they preserve.
6
7pub(crate) mod delete;
8pub(crate) mod progress;
9pub(crate) mod retry;
10pub mod transition;
11
12pub use delete::BlobTombstoneJson;
13pub use transition::{MakeLocalError, MakeRemoteError, MakeRemoteRoot};
14
15#[derive(Debug)]
16pub enum DrainOutcome {
17    Drained {
18        uploaded: usize,
19        yielded_for_publish: bool,
20        failures: UploadFailures,
21    },
22    QueueEmpty,
23    AllInBackoff,
24    Paused,
25}
26
27#[cfg(any(test, feature = "test-utils"))]
28impl DrainOutcome {
29    #[track_caller]
30    fn drained(&self) -> (usize, bool, &UploadFailures) {
31        match self {
32            Self::Drained {
33                uploaded,
34                yielded_for_publish,
35                failures,
36            } => (*uploaded, *yielded_for_publish, failures),
37            other => panic!("expected a drain that attempted queued entries, got {other:?}"),
38        }
39    }
40
41    #[track_caller]
42    pub fn uploaded(&self) -> usize {
43        self.drained().0
44    }
45
46    #[track_caller]
47    pub fn yielded_for_publish(&self) -> bool {
48        self.drained().1
49    }
50
51    #[track_caller]
52    pub fn failures(&self) -> &UploadFailures {
53        self.drained().2
54    }
55
56    #[track_caller]
57    pub fn into_failures(self) -> UploadFailures {
58        match self {
59            Self::Drained { failures, .. } => failures,
60            other => panic!("expected a drain that attempted queued entries, got {other:?}"),
61        }
62    }
63}
64
65#[derive(Debug, thiserror::Error)]
66pub enum UploadFailureCause {
67    #[error("local upload state: {0}")]
68    Database(#[from] coven_database::DbError),
69    #[error("local upload locator: {0}")]
70    Locator(#[from] coven_protocol::blob::locator::BlobLocatorError),
71    #[error("local upload file: {0}")]
72    File(#[from] coven_foundation::atomic_file::FileError),
73    #[error("local upload pin: {0}")]
74    Pin(#[from] coven_foundation::store_dir::StoreBlobFileError),
75    #[error("cancelled cache copy: {0}")]
76    CachedRemoval(#[from] coven_foundation::store_dir::CachedLocatorRemovalError),
77    #[error("blob storage: {0}")]
78    Storage(#[from] coven_protocol::objects::StorageError),
79    #[error("local upload state: {0}")]
80    InvalidState(String),
81}
82
83#[derive(Debug)]
84pub struct UploadFailure {
85    pub entry_id: i64,
86    pub object_key: String,
87    pub cause: UploadFailureCause,
88}
89
90#[derive(Debug)]
91pub struct UploadFailures(Vec<UploadFailure>);
92
93impl UploadFailures {
94    pub fn new(failures: Vec<UploadFailure>) -> Self {
95        Self(failures)
96    }
97
98    pub fn failures(&self) -> &[UploadFailure] {
99        &self.0
100    }
101
102    pub fn has_transport_failure(&self) -> bool {
103        self.0.iter().any(|failure| {
104            matches!(&failure.cause, UploadFailureCause::Storage(error) if error.is_transport())
105        })
106    }
107}
108
109impl std::fmt::Display for UploadFailures {
110    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111        write!(formatter, "{} blob upload(s) failed", self.0.len())?;
112        for failure in &self.0 {
113            write!(
114                formatter,
115                "; entry {} {}: {}",
116                failure.entry_id, failure.object_key, failure.cause
117            )?;
118        }
119        Ok(())
120    }
121}
122
123impl std::error::Error for UploadFailures {
124    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
125        self.0.iter().find_map(|failure| match &failure.cause {
126            UploadFailureCause::Storage(error) if error.is_transport() => {
127                Some(error as &(dyn std::error::Error + 'static))
128            }
129            _ => None,
130        })
131    }
132}
133
134#[cfg(test)]
135mod upload_tests;
136
137#[cfg(test)]
138mod transition_tests;
139
140#[cfg(test)]
141mod local_store_tests;
142
143#[cfg(test)]
144mod delete_tests;
coven_replication/blob/mod.rs

coven_replication/blob/
mod.rs
