fix orphan storage compatibility

This commit is contained in:
Quentin 2023-11-17 15:02:43 +01:00
parent 4a33ac2265
commit e92dc35564
Signed by: quentin
GPG key ID: E9602264D639FF68
5 changed files with 8 additions and 7 deletions

View file

@ -77,7 +77,7 @@ async fn incoming_mail_watch_process_internal(
tokio::select! { tokio::select! {
inc_k = wait_new_mail => Some(inc_k), inc_k = wait_new_mail => Some(inc_k),
_ = tokio::time::sleep(MAIL_CHECK_INTERVAL) => Some(k2v.from_orphan(incoming_key.to_orphan())), _ = tokio::time::sleep(MAIL_CHECK_INTERVAL) => Some(k2v.from_orphan(incoming_key.to_orphan()).expect("Incompatible source & target storage")),
_ = lock_held.changed() => None, _ = lock_held.changed() => None,
_ = rx_inbox_id.changed() => None, _ = rx_inbox_id.changed() => None,
} }
@ -358,7 +358,7 @@ async fn k2v_lock_loop_internal(
)); ));
lock[8..].copy_from_slice(&our_pid.0); lock[8..].copy_from_slice(&our_pid.0);
let row = match ct { let row = match ct {
Some(orphan) => k2v.from_orphan(orphan), Some(orphan) => k2v.from_orphan(orphan).expect("Source & target must be storage compatible"),
None => k2v.row(pk, sk), None => k2v.row(pk, sk),
}; };
if let Err(e) = row.set_value(lock).push().await { if let Err(e) = row.set_value(lock).push().await {
@ -387,7 +387,7 @@ async fn k2v_lock_loop_internal(
_ => None, _ => None,
}; };
if let Some(ct) = release { if let Some(ct) = release {
let row = k2v.from_orphan(ct); let row = k2v.from_orphan(ct).expect("Incompatible source & target storage");
let _ = row.rm().await; let _ = row.rm().await;
} }
} }

View file

@ -306,7 +306,7 @@ impl User {
) -> Result<()> { ) -> Result<()> {
let list_blob = seal_serialize(list, &self.creds.keys.master)?; let list_blob = seal_serialize(list, &self.creds.keys.master)?;
let rref = match ct { let rref = match ct {
Some(x) => self.k2v.from_orphan(x), Some(x) => self.k2v.from_orphan(x).expect("Source & target must be same storage"),
None => self.k2v.row(MAILBOX_LIST_PK, MAILBOX_LIST_SK), None => self.k2v.row(MAILBOX_LIST_PK, MAILBOX_LIST_SK),
}; };
rref.set_value(list_blob).push().await?; rref.set_value(list_blob).push().await?;

View file

@ -36,7 +36,7 @@ impl IRowStore for GrgStore {
unimplemented!(); unimplemented!();
} }
fn from_orphan(&self, orphan: OrphanRowRef) -> RowRef { fn from_orphan(&self, orphan: OrphanRowRef) -> Result<RowRef, StorageError> {
unimplemented!(); unimplemented!();
} }
} }

View file

@ -37,7 +37,7 @@ impl IRowStore for MemStore {
unimplemented!(); unimplemented!();
} }
fn from_orphan(&self, orphan: OrphanRowRef) -> RowRef { fn from_orphan(&self, orphan: OrphanRowRef) -> Result<RowRef, StorageError> {
unimplemented!(); unimplemented!();
} }
} }

View file

@ -36,6 +36,7 @@ pub enum Selector<'a> {
pub enum StorageError { pub enum StorageError {
NotFound, NotFound,
Internal, Internal,
IncompatibleOrphan,
} }
impl std::fmt::Display for StorageError { impl std::fmt::Display for StorageError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@ -87,7 +88,7 @@ pub trait IRowStore
fn row(&self, partition: &str, sort: &str) -> RowRef; fn row(&self, partition: &str, sort: &str) -> RowRef;
fn select(&self, selector: Selector) -> AsyncResult<Vec<RowValue>>; fn select(&self, selector: Selector) -> AsyncResult<Vec<RowValue>>;
fn rm(&self, selector: Selector) -> AsyncResult<()>; fn rm(&self, selector: Selector) -> AsyncResult<()>;
fn from_orphan(&self, orphan: OrphanRowRef) -> RowRef; fn from_orphan(&self, orphan: OrphanRowRef) -> Result<RowRef, StorageError>;
} }
pub type RowStore = Box<dyn IRowStore + Sync + Send>; pub type RowStore = Box<dyn IRowStore + Sync + Send>;