coven_core/sync/store/pull/
model.rs

1use super::*;
2
3#[derive(Debug, Clone, PartialEq, Eq)]
4pub enum HeldStorePositionReason {
5    MissingCommit,
6    MissingPackage,
7    MissingDeviceRegistration {
8        device_id: String,
9        revision: u64,
10        registration_hash: ObjectHash,
11    },
12    MissingPredecessor(StoreBatchCommitRef),
13    MissingDependency {
14        device_id: String,
15        commit: StoreBatchCommitRef,
16    },
17    NewerSchema {
18        local: u32,
19        required: u32,
20    },
21    Unauthorized,
22    DeviceExclusionFreeze {
23        proposal: super::store_commit::StoreDeviceExclusionProposalRef,
24        target_cut: StoreHistoryCut,
25    },
26    InactiveDevice {
27        terminals: Vec<super::store_commit::StoreDeviceExclusionRef>,
28        accepted_cut: StoreHistoryCut,
29    },
30    InvalidChangeset(String),
31    InvalidRowIdentity {
32        table: String,
33        reason: String,
34    },
35    BlobDownloadFailed,
36    ForeignKeyDependency,
37    ConstraintConflict(Vec<String>),
38    HashMismatch {
39        referenced_device_id: String,
40        referenced_commit: StoreBatchCommitRef,
41        materialized_hash: ObjectHash,
42    },
43    InvalidSignature,
44    WrongSlot(String),
45    ObjectCollision(String),
46    ObjectUnreadable {
47        key: String,
48        detail: String,
49    },
50    InvalidObject(String),
51}
52
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub enum HeldStoreCoordinate {
55    Head {
56        device_id: String,
57        seq: u64,
58        head_hash: ObjectHash,
59    },
60    Commit {
61        device_id: String,
62        commit: StoreBatchCommitRef,
63    },
64    Package {
65        device_id: String,
66        seq: u64,
67        package_hash: ObjectHash,
68    },
69    Dependency {
70        dependent_device_id: String,
71        dependent_commit: StoreBatchCommitRef,
72        required_device_id: String,
73        required_commit: StoreBatchCommitRef,
74    },
75}
76
77impl HeldStoreCoordinate {
78    pub fn device_id(&self) -> &str {
79        match self {
80            Self::Head { device_id, .. }
81            | Self::Commit { device_id, .. }
82            | Self::Package { device_id, .. } => device_id,
83            Self::Dependency {
84                dependent_device_id,
85                ..
86            } => dependent_device_id,
87        }
88    }
89
90    pub fn seq(&self) -> u64 {
91        match self {
92            Self::Head { seq, .. } | Self::Package { seq, .. } => *seq,
93            Self::Commit { commit, .. } => commit.coord.sequence(),
94            Self::Dependency {
95                dependent_commit, ..
96            } => dependent_commit.coord.sequence(),
97        }
98    }
99}
100
101#[derive(Debug, Clone, PartialEq, Eq)]
102pub struct HeldStorePosition {
103    pub coordinate: HeldStoreCoordinate,
104    pub reason: HeldStorePositionReason,
105}
106
107#[derive(Debug)]
108pub struct StorePullResult {
109    pub changesets_applied: u64,
110    pub devices_pulled: u64,
111    pub held_positions: Vec<HeldStorePosition>,
112    pub visible_heads: Vec<VerifiedStoreDeviceHead>,
113    pub row_changes: Vec<RowChange>,
114    pub asset_downloads_failed: bool,
115    pub local_blob_cleanup_pending: bool,
116    pub frontier: BTreeMap<String, StoreBatchCommitRef>,
117}
118
119#[derive(Debug, Clone)]
120pub struct VerifiedStoreDeviceHead {
121    pub head: StoreDeviceHead,
122    pub author: StoreDeviceRegistration,
123}
124
125#[derive(Clone, Copy, Debug, PartialEq, Eq)]
126pub(crate) enum LocalStoreMembership {
127    Current,
128    NotYetMember,
129    Removed,
130    IdentityNotSupplied,
131}
132
133impl LocalStoreMembership {
134    pub(crate) fn from_membership(
135        membership: &MembershipChain,
136        identity: Option<&crate::keys::UserKeypair>,
137    ) -> Result<Self, crate::sync::membership::MembershipError> {
138        membership.ensure_resolved()?;
139        let Some(identity) = identity else {
140            return Ok(Self::IdentityNotSupplied);
141        };
142        let identity = crate::keys::public_key_hex(identity);
143        if membership
144            .current_members()
145            .iter()
146            .any(|(member, _)| member == &identity)
147        {
148            Ok(Self::Current)
149        } else if membership.contains_member_history(&identity) {
150            Ok(Self::Removed)
151        } else {
152            Ok(Self::NotYetMember)
153        }
154    }
155
156    pub(crate) fn allows_circle_access(self) -> bool {
157        matches!(self, Self::Current)
158    }
159
160    pub(crate) fn retains_circle_rows(self) -> bool {
161        !matches!(self, Self::Removed)
162    }
163}
164
165#[derive(Debug, thiserror::Error)]
166pub enum StorePullError {
167    #[error("{0}")]
168    Object(#[from] StoreObjectError),
169    #[error("database: {0}")]
170    Database(String),
171    #[error("active Store device {device_id} for member {member:?} has no activated acknowledgement for the selected snapshot")]
172    SnapshotNotStable { member: String, device_id: String },
173    #[error("Store snapshot author is inactive in its exact covered device state")]
174    SnapshotAuthorInactive,
175    #[error("Store snapshot author is not an Owner in its exact membership state")]
176    SnapshotAuthorNotOwner,
177    #[error("membership: {0}")]
178    Membership(#[source] StorePullMembershipError),
179    #[error("{0}")]
180    BlobDownloads(#[source] crate::sync::store::pull::BlobDownloadFailures),
181    #[error("storage: {0}")]
182    Storage(#[from] StorageError),
183}
184
185#[derive(Debug, thiserror::Error)]
186pub enum StorePullMembershipError {
187    #[error("{0}")]
188    Object(#[source] StoreObjectError),
189    #[error("{0}")]
190    Chain(#[source] super::membership_ops::AnchoredChainError),
191    #[error("{0}")]
192    State(#[source] crate::sync::membership::MembershipError),
193    #[error("{0}")]
194    Message(String),
195}
196
197pub(crate) type StorePullFuture<'a, T> =
198    Pin<Box<dyn Future<Output = Result<T, StorePullError>> + Send + 'a>>;
199
200impl From<DbError> for StorePullError {
201    fn from(error: DbError) -> Self {
202        Self::Database(error.into_message())
203    }
204}
205
206#[derive(Clone)]
207pub(crate) struct Candidate {
208    pub(crate) commit_ref: StoreBatchCommitRef,
209    pub(crate) commit: StoreBatchCommit,
210    pub(crate) author: StoreDeviceRegistration,
211    pub(crate) package: Option<Vec<u8>>,
212    pub(crate) registrations: Vec<(StoreDeviceRegistration, StoreDeviceRegistrationActivation)>,
213}
214
215#[derive(Clone)]
216pub(crate) struct LoadedCirclePackage {
217    pub(crate) reference: CirclePackageRef,
218    pub(crate) bytes: Vec<u8>,
219    pub(crate) blob_protection: BlobSpoolProtection,
220}
221
222pub(crate) fn parse_candidate_store_package(
223    candidate: &Candidate,
224    bytes: &[u8],
225) -> Result<AudiencePackage, String> {
226    let package = AudiencePackage::parse(bytes)
227        .map_err(|error| format!("invalid Store audience package: {error}"))?;
228    if !matches!(package.audience(), PackageAudience::Store)
229        || package.store_root_hash() != candidate.commit.store_root_hash
230        || package.write_id() != &candidate.commit.write_id
231        || package.commit_coord() != &candidate.commit_ref.coord
232        || package.candidate_family() != candidate.commit.candidate_family()
233        || candidate
234            .commit
235            .store_package()
236            .as_ref()
237            .is_none_or(|reference| package.schema_version() != reference.schema_version)
238    {
239        return Err("Store audience package differs from its exact commit".to_string());
240    }
241    Ok(package)
242}
243
244pub(crate) fn parse_candidate_circle_package(
245    candidate: &Candidate,
246    loaded: &LoadedCirclePackage,
247) -> Result<AudiencePackage, String> {
248    let package = AudiencePackage::parse(&loaded.bytes)
249        .map_err(|error| format!("invalid Circle audience package: {error}"))?;
250    let expected = &loaded.reference;
251    if !matches!(
252        package.audience(),
253        PackageAudience::Circle {
254            circle_id,
255            control,
256            key_fingerprint,
257        } if *circle_id == expected.circle_id
258            && control == &expected.control
259            && *key_fingerprint == expected.key_fingerprint
260    ) || package.store_root_hash() != candidate.commit.store_root_hash
261        || package.write_id() != &candidate.commit.write_id
262        || package.commit_coord() != &candidate.commit_ref.coord
263        || package.candidate_family() != candidate.commit.candidate_family()
264        || package.schema_version() != expected.package.schema_version
265    {
266        return Err("Circle audience package differs from its exact commit".to_string());
267    }
268    package
269        .validate_blob_uploader(&candidate.commit.author_registration)
270        .map_err(|error| format!("invalid Circle blob authority: {error}"))?;
271    Ok(package)
272}
273
274pub(crate) fn held_commit(
275    reference: &StoreBatchCommitRef,
276    reason: HeldStorePositionReason,
277) -> HeldStorePosition {
278    HeldStorePosition {
279        coordinate: HeldStoreCoordinate::Commit {
280            device_id: commit_stream_id(&reference.coord),
281            commit: reference.clone(),
282        },
283        reason,
284    }
285}
286
287pub(crate) fn held_package(
288    reference: &StoreBatchCommitRef,
289    commit: &StoreBatchCommit,
290    reason: HeldStorePositionReason,
291) -> HeldStorePosition {
292    let package = commit
293        .store_package()
294        .expect("held Store package is named by the commit");
295    HeldStorePosition {
296        coordinate: HeldStoreCoordinate::Package {
297            device_id: commit_stream_id(&reference.coord),
298            seq: commit.seq(),
299            package_hash: package.content_hash,
300        },
301        reason,
302    }
303}
304
305pub(crate) fn held_dependency(
306    dependent: &StoreBatchCommitRef,
307    required_device_id: &str,
308    required: &StoreBatchCommitRef,
309    reason: HeldStorePositionReason,
310) -> HeldStorePosition {
311    HeldStorePosition {
312        coordinate: HeldStoreCoordinate::Dependency {
313            dependent_device_id: commit_stream_id(&dependent.coord),
314            dependent_commit: dependent.clone(),
315            required_device_id: required_device_id.to_string(),
316            required_commit: required.clone(),
317        },
318        reason,
319    }
320}
321
322pub(crate) fn commit_stream_id(coord: &StoreCommitCoord) -> String {
323    coord.stream_id.to_string()
324}
coven_core/sync/store/pull/model.rs

coven_core/sync/store/pull/
model.rs
