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