1use super::*;
2
3#[derive(Debug, Clone)]
4pub enum HeldStorePositionReason {
5 MissingCommit,
6 MissingPredecessor(StoreBatchCommitRef),
7 MissingDependency {
8 device_id: String,
9 commit: StoreBatchCommitRef,
10 },
11 NewerSchema {
12 local: u32,
13 required: u32,
14 },
15 Unauthorized,
16 InvalidChangeset(String),
17 InvalidChangesetIdentity(std::sync::Arc<coven_database::ChangesetIdentityError>),
18 InvalidChangesetDatabase(std::sync::Arc<DbError>),
19 InvalidChangesetBlobDecl(std::sync::Arc<coven_database::BlobDeclError>),
20 InvalidStorePackage(std::sync::Arc<coven_protocol::audience_package::AudiencePackageError>),
21 StorePackageMismatch,
22 InvalidCirclePackage(std::sync::Arc<coven_protocol::audience_package::AudiencePackageError>),
23 CirclePackageMismatch,
24 InvalidCircleBlobAuthority(
25 std::sync::Arc<coven_protocol::audience_package::AudiencePackageError>,
26 ),
27 CirclePackageRead(std::sync::Arc<crate::sync::store::CirclePackageReadError>),
28 ChangesetUnreadable(std::sync::Arc<coven_database::ChangesetError>),
29 InvalidRowIdentity(std::sync::Arc<coven_protocol::synced_schema::RowIdentityError>),
30 ForeignKeyDependency,
31 ConstraintConflict(Vec<String>),
32 PrivateSharedConflict {
33 table: String,
34 row_id: String,
35 commit: StoreBatchCommitRef,
36 },
37 HashMismatch {
38 referenced_device_id: String,
39 referenced_commit: StoreBatchCommitRef,
40 materialized_hash: ObjectHash,
41 },
42 InvalidSignature,
43 WrongSlot(String),
44 WrongSlotProtocol(std::sync::Arc<StoreProtocolError>),
45 ObjectUnreadableStorage {
46 key: String,
47 source: std::sync::Arc<StorageError>,
48 },
49 ObjectUnreadableProtocol {
50 key: String,
51 source: std::sync::Arc<StoreProtocolError>,
52 },
53 ObjectUnreadablePull {
54 key: String,
55 source: std::sync::Arc<StorePullError>,
56 },
57 InvalidObject(String),
58 InvalidObjectJson(std::sync::Arc<serde_json::Error>),
59 InvalidObjectProtocol(std::sync::Arc<StoreProtocolError>),
60 InvalidObjectPull(std::sync::Arc<StorePullError>),
61}
62
63impl PartialEq for HeldStorePositionReason {
64 fn eq(&self, other: &Self) -> bool {
65 use HeldStorePositionReason as Reason;
66 match (self, other) {
67 (Reason::MissingCommit, Reason::MissingCommit)
68 | (Reason::Unauthorized, Reason::Unauthorized)
69 | (Reason::StorePackageMismatch, Reason::StorePackageMismatch)
70 | (Reason::CirclePackageMismatch, Reason::CirclePackageMismatch)
71 | (Reason::ForeignKeyDependency, Reason::ForeignKeyDependency)
72 | (Reason::InvalidSignature, Reason::InvalidSignature) => true,
73 (Reason::MissingPredecessor(left), Reason::MissingPredecessor(right)) => left == right,
74 (
75 Reason::MissingDependency {
76 device_id: ld,
77 commit: lc,
78 },
79 Reason::MissingDependency {
80 device_id: rd,
81 commit: rc,
82 },
83 ) => ld == rd && lc == rc,
84 (
85 Reason::NewerSchema {
86 local: ll,
87 required: lr,
88 },
89 Reason::NewerSchema {
90 local: rl,
91 required: rr,
92 },
93 ) => ll == rl && lr == rr,
94 (Reason::InvalidChangeset(left), Reason::InvalidChangeset(right))
95 | (Reason::WrongSlot(left), Reason::WrongSlot(right))
96 | (Reason::InvalidObject(left), Reason::InvalidObject(right)) => left == right,
97 (Reason::InvalidStorePackage(left), Reason::InvalidStorePackage(right))
98 | (Reason::InvalidCirclePackage(left), Reason::InvalidCirclePackage(right))
99 | (
100 Reason::InvalidCircleBlobAuthority(left),
101 Reason::InvalidCircleBlobAuthority(right),
102 ) => left.to_string() == right.to_string(),
103 (Reason::CirclePackageRead(left), Reason::CirclePackageRead(right)) => {
104 left.to_string() == right.to_string()
105 }
106 (Reason::InvalidRowIdentity(left), Reason::InvalidRowIdentity(right)) => left == right,
107 (Reason::ConstraintConflict(left), Reason::ConstraintConflict(right)) => left == right,
108 (
109 Reason::PrivateSharedConflict {
110 table: lt,
111 row_id: lr,
112 commit: lc,
113 },
114 Reason::PrivateSharedConflict {
115 table: rt,
116 row_id: rr,
117 commit: rc,
118 },
119 ) => lt == rt && lr == rr && lc == rc,
120 (
121 Reason::HashMismatch {
122 referenced_device_id: ld,
123 referenced_commit: lc,
124 materialized_hash: lh,
125 },
126 Reason::HashMismatch {
127 referenced_device_id: rd,
128 referenced_commit: rc,
129 materialized_hash: rh,
130 },
131 ) => ld == rd && lc == rc && lh == rh,
132 (
133 Reason::ObjectUnreadableStorage {
134 key: lk,
135 source: ls,
136 },
137 Reason::ObjectUnreadableStorage {
138 key: rk,
139 source: rs,
140 },
141 ) => lk == rk && ls.to_string() == rs.to_string(),
142 (
143 Reason::ObjectUnreadableProtocol {
144 key: lk,
145 source: ls,
146 },
147 Reason::ObjectUnreadableProtocol {
148 key: rk,
149 source: rs,
150 },
151 ) => lk == rk && ls.to_string() == rs.to_string(),
152 (
153 Reason::ObjectUnreadablePull {
154 key: lk,
155 source: ls,
156 },
157 Reason::ObjectUnreadablePull {
158 key: rk,
159 source: rs,
160 },
161 ) => lk == rk && ls.to_string() == rs.to_string(),
162 (Reason::InvalidChangesetIdentity(left), Reason::InvalidChangesetIdentity(right)) => {
163 left.to_string() == right.to_string()
164 }
165 (Reason::InvalidChangesetDatabase(left), Reason::InvalidChangesetDatabase(right)) => {
166 left.to_string() == right.to_string()
167 }
168 (Reason::InvalidChangesetBlobDecl(left), Reason::InvalidChangesetBlobDecl(right)) => {
169 left.to_string() == right.to_string()
170 }
171 (Reason::ChangesetUnreadable(left), Reason::ChangesetUnreadable(right)) => {
172 left.to_string() == right.to_string()
173 }
174 (Reason::InvalidObjectJson(left), Reason::InvalidObjectJson(right)) => {
175 left.to_string() == right.to_string()
176 }
177 (Reason::InvalidObjectPull(left), Reason::InvalidObjectPull(right)) => {
178 left.to_string() == right.to_string()
179 }
180 (Reason::WrongSlotProtocol(left), Reason::WrongSlotProtocol(right))
181 | (Reason::InvalidObjectProtocol(left), Reason::InvalidObjectProtocol(right)) => {
182 left.to_string() == right.to_string()
183 }
184 _ => false,
185 }
186 }
187}
188
189impl Eq for HeldStorePositionReason {}
190
191#[derive(Debug, Clone, PartialEq, Eq)]
192pub enum HeldStoreCoordinate {
193 Head {
194 device_id: String,
195 seq: u64,
196 head_hash: ObjectHash,
197 },
198 Commit {
199 device_id: String,
200 commit: StoreBatchCommitRef,
201 },
202 Package {
203 device_id: String,
204 seq: u64,
205 package_hash: ObjectHash,
206 },
207 Dependency {
208 dependent_device_id: String,
209 dependent_commit: StoreBatchCommitRef,
210 required_device_id: String,
211 required_commit: StoreBatchCommitRef,
212 },
213}
214
215impl HeldStoreCoordinate {
216 pub fn device_id(&self) -> &str {
217 match self {
218 Self::Head { device_id, .. }
219 | Self::Commit { device_id, .. }
220 | Self::Package { device_id, .. } => device_id,
221 Self::Dependency {
222 dependent_device_id,
223 ..
224 } => dependent_device_id,
225 }
226 }
227
228 pub fn seq(&self) -> u64 {
229 match self {
230 Self::Head { seq, .. } | Self::Package { seq, .. } => *seq,
231 Self::Commit { commit, .. } => commit.coord.sequence(),
232 Self::Dependency {
233 dependent_commit, ..
234 } => dependent_commit.coord.sequence(),
235 }
236 }
237}
238
239#[derive(Debug, Clone, PartialEq, Eq)]
240pub struct HeldStorePosition {
241 pub coordinate: HeldStoreCoordinate,
242 pub reason: HeldStorePositionReason,
243}
244
245impl HeldStorePosition {
246 pub(crate) fn commit(reference: &StoreBatchCommitRef, reason: HeldStorePositionReason) -> Self {
247 Self {
248 coordinate: HeldStoreCoordinate::Commit {
249 device_id: commit_stream_id(&reference.coord),
250 commit: reference.clone(),
251 },
252 reason,
253 }
254 }
255
256 pub(crate) fn package(
257 reference: &StoreBatchCommitRef,
258 commit: &StoreBatchCommit,
259 reason: HeldStorePositionReason,
260 ) -> Self {
261 let package = commit
262 .store_package()
263 .expect("held Store package is named by the commit");
264 Self {
265 coordinate: HeldStoreCoordinate::Package {
266 device_id: commit_stream_id(&reference.coord),
267 seq: commit.seq(),
268 package_hash: package.content_hash,
269 },
270 reason,
271 }
272 }
273
274 pub(crate) fn dependency(
275 dependent: &StoreBatchCommitRef,
276 required_device_id: &str,
277 required: &StoreBatchCommitRef,
278 reason: HeldStorePositionReason,
279 ) -> Self {
280 Self {
281 coordinate: HeldStoreCoordinate::Dependency {
282 dependent_device_id: commit_stream_id(&dependent.coord),
283 dependent_commit: dependent.clone(),
284 required_device_id: required_device_id.to_string(),
285 required_commit: required.clone(),
286 },
287 reason,
288 }
289 }
290}
291
292#[derive(Debug)]
293pub struct StorePullResult {
294 pub changesets_applied: u64,
295 pub held_positions: Vec<HeldStorePosition>,
296 pub visible_commits: Vec<VerifiedStoreBatchCommit>,
297 pub row_changes: Vec<RowChange>,
298 pub local_blob_cleanup_pending: bool,
299 #[cfg(any(test, feature = "test-utils"))]
300 pub frontier: BTreeMap<String, StoreBatchCommitRef>,
301}
302
303#[derive(Debug, thiserror::Error)]
304pub enum StorePullError {
305 #[error("{0}")]
306 Object(#[from] StoreObjectError),
307 #[error("database: {0}")]
308 Database(#[from] DbError),
309 #[error("Store protocol: {0}")]
310 Protocol(#[from] coven_protocol::store_commit::StoreProtocolError),
311 #[error("Store protocol root: {0}")]
312 ProtocolRoot(#[from] crate::sync::store::protocol_root::StoreProtocolRootError),
313 #[error("remote object record: {0}")]
314 RemoteObject(#[from] coven_protocol::remote_object::RemoteObjectRecordError),
315 #[error("membership chain: {0}")]
316 MembershipChain(#[from] crate::sync::store::membership::AnchoredChainError),
317 #[error("membership protocol: {0}")]
318 MembershipProtocol(#[from] coven_protocol::membership::MembershipError),
319 #[error("device join exchange: {0}")]
320 DeviceJoinExchange(
321 #[from] coven_protocol::store_commit::device_join_exchange::DeviceJoinExchangeError,
322 ),
323 #[error("Store operation: {0}")]
324 Store(#[source] Box<crate::sync::store::StoreError>),
325 #[error("serialization: {0}")]
326 Serialization(#[from] serde_json::Error),
327 #[error("row routing key: {0}")]
328 RowRoutingKey(#[from] coven_protocol::circle::RowRoutingKeyError),
329 #[error("Store pull state is invalid: {0}")]
334 InvalidState(String),
335 #[error("{context}: {source}")]
338 Context {
339 context: String,
340 source: Box<StorePullError>,
341 },
342 #[error("active Store device {device_id} for member {member:?} has no activated acknowledgement for the selected snapshot")]
343 SnapshotNotStable { member: String, device_id: String },
344 #[error("Store snapshot author is inactive in its exact covered device state")]
345 SnapshotAuthorInactive,
346 #[error("Store snapshot author is not an Owner in its exact membership state")]
347 SnapshotAuthorNotOwner,
348 #[error("Store snapshot is behind this device's installed replay baseline")]
355 SnapshotBehindReplayBaseline,
356 #[error("current membership is not named by accepted Store history")]
357 ReplayRetirementMembershipUnwitnessed,
358 #[error(
359 "replay retirement waits for Owner recovery device {device_id} for member {member} to activate"
360 )]
361 ReplayRetirementOwnerRecoveryPending { member: String, device_id: String },
362 #[error("membership: {0}")]
363 Membership(#[source] StorePullMembershipError),
364 #[error("storage: {0}")]
365 Storage(#[from] StorageError),
366 #[error("snapshot restoration: {0}")]
367 SnapshotRestoration(#[source] Box<crate::sync::store::snapshots::SnapshotError>),
368 #[error("snapshot preparation cleanup failed: {cleanup} (operation: {operation})")]
369 SnapshotPreparationCleanup {
370 #[source]
371 operation: Box<StorePullError>,
372 cleanup: DbError,
373 },
374 #[error("Circle package: {0}")]
375 CirclePackage(#[source] Box<crate::sync::store::CirclePackageReadError>),
376}
377
378impl From<crate::sync::store::CirclePackageReadError> for StorePullError {
379 fn from(error: crate::sync::store::CirclePackageReadError) -> Self {
380 Self::CirclePackage(Box::new(error))
381 }
382}
383
384impl StorePullError {
385 pub(crate) fn context(
387 context: impl Into<String>,
388 source: impl Into<StorePullError>,
389 ) -> StorePullError {
390 StorePullError::Context {
391 context: context.into(),
392 source: Box::new(source.into()),
393 }
394 }
395}
396
397#[derive(Debug, thiserror::Error)]
398pub enum StorePullMembershipError {
399 #[error("{0}")]
400 State(#[source] coven_protocol::membership::MembershipError),
401 #[error("{0}")]
402 Message(String),
403}
404
405#[derive(Clone)]
406pub(crate) struct Candidate {
407 pub(crate) verified: VerifiedStoreBatchCommit,
408 pub(crate) package: Option<Vec<u8>>,
409 pub(crate) registrations: Vec<ActivatedStoreDeviceRegistration>,
410}
411
412impl Candidate {
413 pub(crate) fn commit_ref(&self) -> &StoreBatchCommitRef {
414 self.verified.reference()
415 }
416
417 pub(crate) fn commit(&self) -> &StoreBatchCommit {
418 self.verified.value()
419 }
420
421 pub(crate) fn author(&self) -> &StoreDeviceRegistration {
422 self.verified.author()
423 }
424
425 pub(crate) fn parse_store_package(
426 &self,
427 bytes: &[u8],
428 ) -> Result<AudiencePackage, HeldStorePositionReason> {
429 let commit = self.commit();
430 let package = AudiencePackage::parse(bytes)
431 .map_err(|error| HeldStorePositionReason::InvalidStorePackage(error.into()))?;
432 if !matches!(package.audience(), PackageAudience::Store)
433 || package.store_root_hash() != commit.store_root_hash
434 || package.write_id() != &commit.write_id
435 || package.commit_coord() != &self.commit_ref().coord
436 || package.candidate_family() != commit.candidate_family()
437 || commit
438 .store_package()
439 .as_ref()
440 .is_none_or(|reference| package.schema_version() != reference.schema_version)
441 {
442 return Err(HeldStorePositionReason::StorePackageMismatch);
443 }
444 Ok(package)
445 }
446
447 pub(crate) fn parse_circle_package(
448 &self,
449 loaded: &LoadedCirclePackage,
450 ) -> Result<AudiencePackage, HeldStorePositionReason> {
451 let commit = self.commit();
452 let package = AudiencePackage::parse(&loaded.bytes)
453 .map_err(|error| HeldStorePositionReason::InvalidCirclePackage(error.into()))?;
454 let expected = &loaded.reference;
455 if !matches!(
456 package.audience(),
457 PackageAudience::Circle {
458 circle_id,
459 control,
460 key_fingerprint,
461 } if *circle_id == expected.circle_id
462 && control == &expected.control
463 && *key_fingerprint == expected.key_fingerprint
464 ) || package.store_root_hash() != commit.store_root_hash
465 || package.write_id() != &commit.write_id
466 || package.commit_coord() != &self.commit_ref().coord
467 || package.candidate_family() != commit.candidate_family()
468 || package.schema_version() != expected.package.schema_version
469 {
470 return Err(HeldStorePositionReason::CirclePackageMismatch);
471 }
472 package
473 .validate_blob_uploader(&commit.author_registration)
474 .map_err(|error| HeldStorePositionReason::InvalidCircleBlobAuthority(error.into()))?;
475 Ok(package)
476 }
477}
478
479#[derive(Clone)]
480pub struct LoadedCirclePackage {
481 pub(crate) reference: CirclePackageRef,
482 pub(crate) bytes: Vec<u8>,
483}
484
485pub(crate) fn commit_stream_id(coord: &StoreCommitCoord) -> String {
486 coord.stream_id.to_string()
487}