Move message sending into meta

Author: rklaehn

src/store/fs.rs

@@ -246,20 +246,8 @@ impl HashContext {
     }
 
     /// Update the entry state in the database, and wait for completion.
-    pub async fn update(&self, hash: Hash, state: EntryState<Bytes>) -> io::Result<()> {
-        let (tx, rx) = oneshot::channel();
-        self.db()
-            .send(
-                meta::Update {
-                    hash,
-                    state,
-                    tx: Some(tx),
-                    span: tracing::Span::current(),
-                }
-                .into(),
-            )
-            .await?;
-        rx.await.map_err(|_e| io::Error::other(""))??;
+    pub async fn update_await(&self, hash: Hash, state: EntryState<Bytes>) -> io::Result<()> {
+        self.db().update_await(hash, state).await?;
         Ok(())
     }
 
@@ -269,40 +257,13 @@ impl HashContext {
                 data_location: DataLocation::Inline(Bytes::new()),
                 outboard_location: OutboardLocation::NotNeeded,
             }));
-        }
-        let (tx, rx) = oneshot::channel();
-        self.db()
-            .send(
-                meta::Get {
-                    hash,
-                    tx,
-                    span: tracing::Span::current(),
-                }
-                .into(),
-            )
-            .await
-            .ok();
-        let res = rx.await.map_err(io::Error::other)?;
-        Ok(res.state?)
+        };
+        self.db().get(hash).await
     }
 
     /// Update the entry state in the database, and wait for completion.
     pub async fn set(&self, hash: Hash, state: EntryState<Bytes>) -> io::Result<()> {
-        let (tx, rx) = oneshot::channel();
-        self.db()
-            .send(
-                meta::Set {
-                    hash,
-                    state,
-                    tx,
-                    span: tracing::Span::current(),
-                }
-                .into(),
-            )
-            .await
-            .map_err(io::Error::other)?;
-        rx.await.map_err(|_e| io::Error::other(""))??;
-        Ok(())
+        self.db().set(hash, state).await
     }
 
     pub async fn get_maybe_create(&self, hash: Hash, create: bool) -> api::Result<BaoFileHandle> {
@@ -892,7 +853,7 @@ async fn finish_import_impl(import_data: ImportEntry, ctx: HashContext) -> io::R
         data_location,
         outboard_location,
     };
-    ctx.update(hash, state).await?;
+    ctx.update_await(hash, state).await?;
     Ok(())
 }
 

src/store/fs/meta.rs

@@ -96,6 +96,25 @@ impl Db {
         Self { sender }
     }
 
+    pub async fn update_await(&self, hash: Hash, state: EntryState<Bytes>) -> io::Result<()> {
+        let (tx, rx) = oneshot::channel();
+        self.sender
+            .send(
+                Update {
+                    hash,
+                    state,
+                    tx: Some(tx),
+                    span: tracing::Span::current(),
+                }
+                .into(),
+            )
+            .await
+            .map_err(|_| io::Error::other("send update"))?;
+        rx.await
+            .map_err(|_e| io::Error::other("receive update"))??;
+        Ok(())
+    }
+
     /// Update the entry state for a hash, without awaiting completion.
     pub async fn update(&self, hash: Hash, state: EntryState<Bytes>) -> io::Result<()> {
         self.sender
@@ -132,7 +151,7 @@ impl Db {
     }
 
     /// Get the entry state for a hash, if any.
-    pub async fn get(&self, hash: Hash) -> anyhow::Result<Option<EntryState<Bytes>>> {
+    pub async fn get(&self, hash: Hash) -> io::Result<Option<EntryState<Bytes>>> {
         let (tx, rx) = oneshot::channel();
         self.sender
             .send(
@@ -143,8 +162,9 @@ impl Db {
                 }
                 .into(),
             )
-            .await?;
-        let res = rx.await?;
+            .await
+            .map_err(|_| io::Error::other("send get"))?;
+        let res = rx.await.map_err(|_| io::Error::other("receive get"))?;
         Ok(res.state?)
     }