1use crate::query_mapped_rows;
2use crate::*;
3#[cfg(any(test, feature = "test-utils"))]
4use coven_protocol::store_commit::StoreDeviceRegistration;
5use coven_protocol::store_commit::{
6 ActivatedStoreDeviceRegistration, CommitFrontier, ReferencedStoreDeviceRegistration,
7 ResolvedStoreDeviceState, StoreBatchCommitRef, StoreDeviceProposalAck,
8 StoreDeviceRegistrationRef, StoreDeviceStateRef, StoreHistoryCut,
9};
10use rusqlite::{Connection, OptionalExtension};
11use std::collections::BTreeMap;
12
13use super::*;
14
15impl StoreSession<'_> {
16 fn materialized_frontier(&mut self) -> Result<BTreeMap<String, StoreBatchCommitRef>, DbError> {
17 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
18 .materialized_frontier()
19 }
20
21 fn retained_merge_replay_inputs(
22 &mut self,
23 root: coven_protocol::store_commit::StoreRootRef,
24 ) -> Result<Vec<OwnedVerifiedMergeMaterialization>, DbError> {
25 self.verified_store_authority.retained_replay_inputs_on(
26 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
27 &root,
28 )
29 }
30
31 fn retained_merge_materialization_refs(&mut self) -> Result<Vec<StoreBatchCommitRef>, DbError> {
32 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
33 .retained_merge_materialization_refs()
34 }
35
36 fn retained_merge_materialization(
37 &mut self,
38 root: coven_protocol::store_commit::StoreRootRef,
39 reference: StoreBatchCommitRef,
40 ) -> Result<OwnedVerifiedMergeMaterialization, DbError> {
41 self.verified_store_authority
42 .retained_replay_inputs_on(
43 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
44 &root,
45 )?
46 .into_iter()
47 .find(|materialization| materialization.commit_ref() == &reference)
48 .ok_or_else(|| {
49 DbError::Message(
50 "retained Merge materialization is absent at its exact coordinate".to_string(),
51 )
52 })
53 }
54
55 fn retained_merge_history_frontier(
56 &mut self,
57 root: coven_protocol::store_commit::StoreRootRef,
58 references: Vec<StoreBatchCommitRef>,
59 ) -> Result<Vec<RetainedMergeHistoryCheckpoint>, DbError> {
60 let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
61 let authority = &mut *self.verified_store_authority;
62 let retained = authority.retained_replay_inputs_on(records, &root)?;
63 let baseline = authority.retained_replay_baseline_on(records)?.clone();
66 let by_reference = retained
67 .iter()
68 .map(|materialization| (materialization.commit_ref().clone(), materialization))
69 .collect::<BTreeMap<_, _>>();
70 let mut pending = references;
71 let mut visited = std::collections::BTreeSet::new();
72 let mut checkpoints = Vec::new();
73 while let Some(reference) = pending.pop() {
74 if !visited.insert(reference.clone()) {
75 continue;
76 }
77 match by_reference.get(&reference) {
78 Some(materialization) => {
79 pending.extend(
80 materialization
81 .commit()
82 .order
83 .predecessor_cut()
84 .map_err(DbError::from)?
85 .0
86 .into_values(),
87 );
88 checkpoints
89 .push(authority.retained_history_checkpoint_on(records, &reference)?);
90 }
91 None => checkpoints.push(StoreDatabase::load_retained_merge_history_checkpoint_on(
92 records, &root, authority, &baseline, &reference,
93 )?),
94 }
95 }
96 Ok(checkpoints)
97 }
98
99 fn exact_materialized_ref(
100 &mut self,
101 stream_id: String,
102 sequence: u64,
103 ) -> Result<Option<StoreBatchCommitRef>, DbError> {
104 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
105 .materialized_commit_ref(&stream_id, sequence)
106 }
107
108 fn snapshot_coverage_frontier(&mut self) -> Result<CommitFrontier, DbError> {
109 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
110 .snapshot_coverage_frontier()
111 }
112
113 fn installed_replay_baseline(&mut self) -> Result<crate::InstalledReplayBaseline, DbError> {
114 let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
115 let coverage = self.snapshot_coverage_frontier()?;
116 let covered_states =
117 crate::store::store_device_state::load_covered_store_device_snapshots_on(
118 self.conn, &coverage,
119 )?;
120 let (summary, snapshot) =
123 match crate::store::retained_replay::load_replay_baseline_metadata_on(records)? {
124 Some(baseline) => match &baseline.authority {
125 crate::RetainedReplayAuthority::InstalledSnapshot(authority) => {
126 let snapshot = authority.snapshot.clone();
127 (
128 Some(
129 crate::StoreDatabase::open_installed_baseline_history_summary(
130 records, &baseline,
131 )?,
132 ),
133 Some(snapshot),
134 )
135 }
136 crate::RetainedReplayAuthority::Genesis(_) => (None, None),
137 },
138 None => (None, None),
139 };
140 Ok(crate::InstalledReplayBaseline::new(
141 coverage,
142 covered_states,
143 summary,
144 snapshot,
145 ))
146 }
147
148 fn snapshot_announcement_frontier(
158 &mut self,
159 ) -> Result<
160 BTreeMap<
161 coven_protocol::causal_grants::AuthorStreamId,
162 coven_protocol::store_commit::RetainedAcceptedStoreAnnouncement,
163 >,
164 DbError,
165 > {
166 let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
167 let baseline = self
168 .verified_store_authority
169 .retained_replay_baseline_on(records)?;
170 let crate::RetainedReplayAuthority::InstalledSnapshot(authority) = &baseline.authority
171 else {
172 return Ok(BTreeMap::new());
173 };
174 let summary = &authority.metadata.history_summary;
175 let coverage = &authority.metadata.coverage.0;
176 let frontier = summary.announcement_frontier.clone();
177 for (stream_id, announcement) in &frontier {
178 if coverage.get(stream_id) != Some(&announcement.value.commit) {
183 return Err(DbError::Message(
184 "snapshot announcement frontier differs from its own coverage".to_string(),
185 ));
186 }
187 }
188 Ok(frontier)
189 }
190
191 fn store_device_state_for_history_cut(
192 &mut self,
193 cut: StoreHistoryCut,
194 ) -> Result<(StoreDeviceStateRef, ResolvedStoreDeviceState), DbError> {
195 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
196 .store_device_state_for_history_cut(&cut)
197 }
198
199 fn resolved_store_device_state(
200 &mut self,
201 reference: StoreDeviceStateRef,
202 ) -> Result<ResolvedStoreDeviceState, DbError> {
203 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
204 .declared_store_device_state(&reference)
205 }
206
207 fn store_device_exclusion_freezes(&mut self) -> Result<Vec<StoreDeviceProposalAck>, DbError> {
208 let root = self
209 .root_authority()?
210 .map(|(reference, _)| reference)
211 .ok_or_else(|| {
212 DbError::Message("Store root is absent while loading exclusion freezes".to_string())
213 })?;
214 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
215 .store_device_exclusion_freezes(&root)
216 }
217
218 fn activated_store_device_registration_records(
219 &mut self,
220 ) -> Result<Vec<ReferencedStoreDeviceRegistration>, DbError> {
221 let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
222 let root = self
223 .verified_store_authority
224 .root_authority_on(records)?
225 .map(|(reference, _)| reference)
226 .ok_or_else(|| {
227 DbError::Message("Store root is absent while loading activated devices".to_string())
228 })?;
229 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
230 .activated_registration_references()?
231 .into_iter()
232 .map(|reference| {
233 let device_id = reference.device_id;
234 let registration = self
235 .verified_store_authority
236 .activated_registration_on(records, &root, &reference)?;
237 ReferencedStoreDeviceRegistration::verified(reference, registration).map_err(
238 |error| {
239 DbError::context(
240 format!(
241 "activated Store device registration {device_id} exact reference"
242 ),
243 error,
244 )
245 },
246 )
247 })
248 .collect::<Result<Vec<_>, DbError>>()
249 }
250
251 fn activated_store_device_registration(
252 &mut self,
253 reference: StoreDeviceRegistrationRef,
254 ) -> Result<ReferencedStoreDeviceRegistration, DbError> {
255 let root = self
256 .root_authority()?
257 .map(|(reference, _)| reference)
258 .ok_or_else(|| {
259 DbError::Message(
260 "Store root is absent while loading an activated device".to_string(),
261 )
262 })?;
263 let registration = self.verified_store_authority.activated_registration_on(
264 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
265 &root,
266 &reference,
267 )?;
268 ReferencedStoreDeviceRegistration::verified(reference, registration).map_err(DbError::from)
269 }
270
271 fn local_activated_registration_ref(
272 &mut self,
273 ) -> Result<Option<StoreDeviceRegistrationRef>, DbError> {
274 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
275 .local_activated_registration_ref()
276 }
277
278 fn activated_store_device_registration_with_authority(
279 &mut self,
280 root: coven_protocol::store_commit::StoreRootRef,
281 reference: StoreDeviceRegistrationRef,
282 ) -> Result<ActivatedStoreDeviceRegistration, DbError> {
283 let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
284 let registration = self
285 .verified_store_authority
286 .activated_registration_on(records, &root, &reference)?;
287 let authority = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
288 .activated_registration_authority(&reference)?;
289 let authority = serde_json::from_str(&authority)
290 .map_err(|error| DbError::context("activated Store registration authority", error))?;
291 let registration = ReferencedStoreDeviceRegistration::verified(reference, registration)
292 .map_err(DbError::from)?;
293 ActivatedStoreDeviceRegistration::verified(registration, authority).map_err(DbError::from)
294 }
295
296 fn activated_store_device_registration_for_device(
297 &mut self,
298 device_id: coven_protocol::store_commit::StoreDeviceId,
299 ) -> Result<Option<ActivatedStoreDeviceRegistration>, DbError> {
300 let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
301 let root = self
302 .verified_store_authority
303 .root_authority_on(records)?
304 .map(|(reference, _)| reference)
305 .ok_or_else(|| {
306 DbError::Message(
307 "Store root is absent while loading an activated device".to_string(),
308 )
309 })?;
310 let stored = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
311 .activated_registration_row_for_device(device_id)?;
312 let Some((reference, authority)) = stored else {
313 return Ok(None);
314 };
315 let reference: StoreDeviceRegistrationRef = serde_json::from_str(&reference)
316 .map_err(|error| DbError::context("activated Store registration ref", error))?;
317 if reference.device_id != device_id {
318 return Err(DbError::Message(
319 "activated Store registration row names another device".to_string(),
320 ));
321 }
322 let registration = self
323 .verified_store_authority
324 .activated_registration_on(records, &root, &reference)?;
325 let authority = serde_json::from_str(&authority)
326 .map_err(|error| DbError::context("activated Store registration authority", error))?;
327 let registration = ReferencedStoreDeviceRegistration::verified(reference, registration)
328 .map_err(DbError::from)?;
329 ActivatedStoreDeviceRegistration::verified(registration, authority)
330 .map(Some)
331 .map_err(DbError::from)
332 }
333}
334
335impl StoreDatabase {
336 pub async fn materialized_frontier(
337 &self,
338 ) -> Result<BTreeMap<String, StoreBatchCommitRef>, DbError> {
339 self.call_store(|session| session.materialized_frontier())
340 .await
341 }
342
343 pub async fn retained_merge_replay_inputs(
344 &self,
345 root: coven_protocol::store_commit::StoreRootRef,
346 ) -> Result<Vec<OwnedVerifiedMergeMaterialization>, DbError> {
347 self.call_store(move |session| session.retained_merge_replay_inputs(root))
348 .await
349 }
350
351 pub async fn retained_merge_materialization_refs(
352 &self,
353 ) -> Result<Vec<StoreBatchCommitRef>, DbError> {
354 self.call_store(|session| session.retained_merge_materialization_refs())
355 .await
356 }
357
358 pub async fn retained_merge_materialization(
359 &self,
360 root: coven_protocol::store_commit::StoreRootRef,
361 reference: StoreBatchCommitRef,
362 ) -> Result<OwnedVerifiedMergeMaterialization, DbError> {
363 self.call_store(move |session| session.retained_merge_materialization(root, reference))
364 .await
365 }
366
367 pub async fn retained_merge_history_frontier(
368 &self,
369 root: coven_protocol::store_commit::StoreRootRef,
370 references: Vec<StoreBatchCommitRef>,
371 ) -> Result<Vec<RetainedMergeHistoryCheckpoint>, DbError> {
372 self.call_store(move |session| session.retained_merge_history_frontier(root, references))
373 .await
374 }
375
376 pub async fn exact_materialized_ref(
377 &self,
378 stream_id: &str,
379 sequence: u64,
380 ) -> Result<Option<StoreBatchCommitRef>, DbError> {
381 let stream_id = stream_id.to_string();
382 self.call_store(move |session| session.exact_materialized_ref(stream_id, sequence))
383 .await
384 }
385
386 pub async fn snapshot_coverage_frontier(&self) -> Result<CommitFrontier, DbError> {
387 self.call_store(|session| session.snapshot_coverage_frontier())
388 .await
389 }
390
391 pub async fn installed_replay_baseline(
394 &self,
395 ) -> Result<crate::InstalledReplayBaseline, DbError> {
396 self.call_store(|session| session.installed_replay_baseline())
397 .await
398 }
399
400 pub async fn snapshot_announcement_frontier(
401 &self,
402 ) -> Result<
403 BTreeMap<
404 coven_protocol::causal_grants::AuthorStreamId,
405 coven_protocol::store_commit::RetainedAcceptedStoreAnnouncement,
406 >,
407 DbError,
408 > {
409 self.call_store(|session| session.snapshot_announcement_frontier())
410 .await
411 }
412
413 pub async fn store_device_state_for_order(
414 &self,
415 order: &coven_protocol::store_commit::StoreCommitOrder,
416 ) -> Result<(StoreDeviceStateRef, ResolvedStoreDeviceState), DbError> {
417 let cut = order.predecessor_cut().map_err(DbError::from)?;
418 self.call_store(move |session| session.store_device_state_for_history_cut(cut))
419 .await
420 }
421
422 pub async fn store_device_state_for_history_cut(
423 &self,
424 cut: &StoreHistoryCut,
425 ) -> Result<(StoreDeviceStateRef, ResolvedStoreDeviceState), DbError> {
426 let cut = cut.clone();
427 self.call_store(move |session| session.store_device_state_for_history_cut(cut))
428 .await
429 }
430
431 pub async fn resolved_store_device_state(
432 &self,
433 reference: &StoreDeviceStateRef,
434 ) -> Result<ResolvedStoreDeviceState, DbError> {
435 let reference = reference.clone();
436 self.call_store(move |session| session.resolved_store_device_state(reference))
437 .await
438 }
439
440 pub async fn store_device_exclusion_freezes(
441 &self,
442 ) -> Result<Vec<StoreDeviceProposalAck>, DbError> {
443 self.call_store(|session| session.store_device_exclusion_freezes())
444 .await
445 }
446
447 pub async fn activated_store_device_registration_records(
448 &self,
449 ) -> Result<Vec<ReferencedStoreDeviceRegistration>, DbError> {
450 self.call_store(|session| session.activated_store_device_registration_records())
451 .await
452 }
453
454 pub async fn activated_store_device_registration(
455 &self,
456 reference: StoreDeviceRegistrationRef,
457 ) -> Result<ReferencedStoreDeviceRegistration, DbError> {
458 self.call_store(move |session| session.activated_store_device_registration(reference))
459 .await
460 }
461
462 pub async fn local_activated_registration_ref(
466 &self,
467 ) -> Result<Option<StoreDeviceRegistrationRef>, DbError> {
468 self.call_store(|session| session.local_activated_registration_ref())
469 .await
470 }
471
472 pub async fn local_blob_write_authority(
473 &self,
474 ) -> Result<ReferencedStoreDeviceRegistration, DbError> {
475 self.call_store(|session| session.local_store_authority())
476 .await
477 }
478
479 pub async fn activated_store_device_registration_with_authority(
480 &self,
481 root: &coven_protocol::store_commit::StoreRootRef,
482 reference: StoreDeviceRegistrationRef,
483 ) -> Result<ActivatedStoreDeviceRegistration, DbError> {
484 let root = root.clone();
485 self.call_store(move |session| {
486 session.activated_store_device_registration_with_authority(root, reference)
487 })
488 .await
489 }
490
491 pub async fn activated_store_device_registration_for_device(
492 &self,
493 device_id: coven_protocol::store_commit::StoreDeviceId,
494 ) -> Result<Option<ActivatedStoreDeviceRegistration>, DbError> {
495 self.call_store(move |session| {
496 session.activated_store_device_registration_for_device(device_id)
497 })
498 .await
499 }
500
501 #[cfg(any(test, feature = "test-utils"))]
502 pub async fn activated_store_device_registrations(
503 &self,
504 ) -> Result<Vec<StoreDeviceRegistration>, DbError> {
505 Ok(self
506 .activated_store_device_registration_records()
507 .await?
508 .into_iter()
509 .map(|registration| registration.value().clone())
510 .collect())
511 }
512}
513
514pub(crate) fn materialized_frontier_on(
515 conn: &Connection,
516 exclude_device: Option<&str>,
517) -> Result<BTreeMap<String, StoreBatchCommitRef>, DbError> {
518 let mut frontier = BTreeMap::new();
519 let rows = query_mapped_rows(
520 conn,
521 "SELECT m.device_id, m.seq, m.commit_ref,
522 m.retained_commit_ref, m.retained_input_hash \
523 FROM materialized_commits m \
524 JOIN (SELECT device_id, MAX(seq) AS seq FROM materialized_commits \
525 GROUP BY device_id) latest \
526 ON latest.device_id = m.device_id AND latest.seq = m.seq",
527 [],
528 |row| {
529 Ok((
530 row.get::<_, String>(0)?,
531 row.get::<_, i64>(1)?,
532 row.get::<_, String>(2)?,
533 row.get::<_, Option<String>>(3)?,
534 row.get::<_, Option<String>>(4)?,
535 ))
536 },
537 )?;
538 for row in rows {
539 let (device_id, seq, reference, retained_commit_ref, retained_input_hash) = row;
540 if exclude_device == Some(device_id.as_str()) {
541 continue;
542 }
543 let seq = Database::sequence_from_sqlite(&device_id, seq)?;
544 frontier.insert(
545 device_id.clone(),
546 parse_materialized_commit_row_on(
547 &device_id,
548 seq,
549 &reference,
550 retained_commit_ref.as_deref(),
551 retained_input_hash.as_deref(),
552 )?,
553 );
554 }
555
556 for (device_id, reference) in snapshot_coverage_on(conn)? {
557 if exclude_device == Some(device_id.as_str()) {
558 continue;
559 }
560 if frontier
561 .get(&device_id)
562 .is_none_or(|current| current.coord.sequence() < reference.coord.sequence())
563 {
564 frontier.insert(device_id, reference);
565 }
566 }
567 Ok(frontier)
568}
569
570pub(crate) fn snapshot_coverage_on(
574 conn: &Connection,
575) -> Result<BTreeMap<String, StoreBatchCommitRef>, DbError> {
576 let rows = query_mapped_rows(
577 conn,
578 "SELECT device_id, seq, commit_ref FROM snapshot_coverage",
579 [],
580 |row| {
581 Ok((
582 row.get::<_, String>(0)?,
583 row.get::<_, i64>(1)?,
584 row.get::<_, String>(2)?,
585 ))
586 },
587 )?;
588 let mut coverage = BTreeMap::new();
589 for (device_id, seq, reference) in rows {
590 let seq = Database::sequence_from_sqlite(&device_id, seq)?;
591 let reference = parse_stored_commit_ref(&device_id, seq, &reference)?;
592 coverage.insert(device_id, reference);
593 }
594 Ok(coverage)
595}
596
597pub(crate) fn parse_stored_commit_ref(
598 stream_id: &str,
599 sequence: u64,
600 encoded: &str,
601) -> Result<StoreBatchCommitRef, DbError> {
602 let reference: StoreBatchCommitRef = serde_json::from_str(encoded)
603 .map_err(|error| DbError::context("stored exact Store commit ref", error))?;
604 let coordinate_matches =
605 reference.coord.stream_id.to_string() == stream_id && reference.coord.sequence == sequence;
606 if !coordinate_matches {
607 return Err(DbError::Message(format!(
608 "stored exact Store commit ref differs from {stream_id}/{sequence}"
609 )));
610 }
611 Ok(reference)
612}
613
614fn parse_materialized_commit_row_on(
615 stream_id: &str,
616 sequence: u64,
617 encoded: &str,
618 retained_commit_ref: Option<&str>,
619 retained_input_hash: Option<&str>,
620) -> Result<StoreBatchCommitRef, DbError> {
621 let reference = parse_stored_commit_ref(stream_id, sequence, encoded)?;
622 if retained_commit_ref != Some(encoded) {
623 return Err(DbError::Message(format!(
624 "materialized coordinate {stream_id}/{sequence} does not bind its exact retained commit"
625 )));
626 }
627 let input_hash = retained_input_hash.ok_or_else(|| {
628 DbError::Message(format!(
629 "materialized coordinate {stream_id}/{sequence} has no retained input hash"
630 ))
631 })?;
632 input_hash
633 .parse::<coven_protocol::store_commit::ObjectHash>()
634 .map_err(|error| {
635 DbError::context(
636 format!(
637 "materialized coordinate {stream_id}/{sequence} retained input hash is invalid"
638 ),
639 error,
640 )
641 })?;
642 Ok(reference)
643}
644
645pub(crate) fn materialized_commit_ref_on(
646 conn: &Connection,
647 stream_id: &str,
648 sequence: u64,
649) -> Result<Option<StoreBatchCommitRef>, DbError> {
650 let seq = Database::sequence_to_sqlite(stream_id, sequence)?;
651 conn.query_row(
652 "SELECT commit_ref, retained_commit_ref, retained_input_hash
653 FROM materialized_commits WHERE device_id = ?1 AND seq = ?2",
654 (stream_id, seq),
655 |row| {
656 Ok((
657 row.get::<_, String>(0)?,
658 row.get::<_, Option<String>>(1)?,
659 row.get::<_, Option<String>>(2)?,
660 ))
661 },
662 )
663 .optional()
664 .map_err(DbError::from)?
665 .map(|(encoded, retained_commit_ref, retained_input_hash)| {
666 parse_materialized_commit_row_on(
667 stream_id,
668 sequence,
669 &encoded,
670 retained_commit_ref.as_deref(),
671 retained_input_hash.as_deref(),
672 )
673 })
674 .transpose()
675}
676
677pub(crate) fn latest_position_for_device_on(
678 conn: &Connection,
679 device_id: &str,
680) -> Result<Option<StoreBatchCommitRef>, DbError> {
681 let materialized = conn
682 .query_row(
683 "SELECT seq, commit_ref, retained_commit_ref, retained_input_hash
684 FROM materialized_commits
685 WHERE device_id = ?1 ORDER BY seq DESC LIMIT 1",
686 [device_id],
687 |row| {
688 Ok((
689 row.get::<_, i64>(0)?,
690 row.get::<_, String>(1)?,
691 row.get::<_, Option<String>>(2)?,
692 row.get::<_, Option<String>>(3)?,
693 ))
694 },
695 )
696 .optional()
697 .map_err(DbError::from)?;
698 let coverage = conn
699 .query_row(
700 "SELECT seq, commit_ref FROM snapshot_coverage WHERE device_id = ?1",
701 [device_id],
702 |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
703 )
704 .optional()
705 .map_err(DbError::from)?;
706 let mut references = Vec::new();
707 if let Some((seq, reference, retained_commit_ref, retained_input_hash)) = materialized {
708 let seq = Database::sequence_from_sqlite(device_id, seq)?;
709 references.push(parse_materialized_commit_row_on(
710 device_id,
711 seq,
712 &reference,
713 retained_commit_ref.as_deref(),
714 retained_input_hash.as_deref(),
715 )?);
716 }
717 if let Some((seq, reference)) = coverage {
718 let seq = Database::sequence_from_sqlite(device_id, seq)?;
719 references.push(parse_stored_commit_ref(device_id, seq, &reference)?);
720 }
721 if references.len() == 2
722 && references[0].coord.sequence() == references[1].coord.sequence()
723 && references[0] != references[1]
724 {
725 return Err(DbError::Message(format!(
726 "materialized ledger and snapshot coverage fork {device_id:?} at sequence {}",
727 references[0].coord.sequence()
728 )));
729 }
730 Ok(references
731 .into_iter()
732 .max_by_key(|reference| reference.coord.sequence()))
733}