Merge branch 'remove-db-public-send' into entity-manager

Author: rklaehn

src/api/blobs.rs

@@ -127,7 +127,7 @@ impl Blobs {
         .await
     }
 
-    pub fn add_slice(&self, data: impl AsRef<[u8]>) -> AddProgress {
+    pub fn add_slice(&self, data: impl AsRef<[u8]>) -> AddProgress<'_> {
         let options = ImportBytesRequest {
             data: Bytes::copy_from_slice(data.as_ref()),
             format: crate::BlobFormat::Raw,
@@ -136,7 +136,7 @@ impl Blobs {
         self.add_bytes_impl(options)
     }
 
-    pub fn add_bytes(&self, data: impl Into<bytes::Bytes>) -> AddProgress {
+    pub fn add_bytes(&self, data: impl Into<bytes::Bytes>) -> AddProgress<'_> {
         let options = ImportBytesRequest {
             data: data.into(),
             format: crate::BlobFormat::Raw,
@@ -145,7 +145,7 @@ impl Blobs {
         self.add_bytes_impl(options)
     }
 
-    pub fn add_bytes_with_opts(&self, options: impl Into<AddBytesOptions>) -> AddProgress {
+    pub fn add_bytes_with_opts(&self, options: impl Into<AddBytesOptions>) -> AddProgress<'_> {
         let options = options.into();
         let request = ImportBytesRequest {
             data: options.data,
@@ -155,7 +155,7 @@ impl Blobs {
         self.add_bytes_impl(request)
     }
 
-    fn add_bytes_impl(&self, options: ImportBytesRequest) -> AddProgress {
+    fn add_bytes_impl(&self, options: ImportBytesRequest) -> AddProgress<'_> {
         trace!("{options:?}");
         let this = self.clone();
         let stream = Gen::new(|co| async move {
@@ -180,7 +180,7 @@ impl Blobs {
         AddProgress::new(self, stream)
     }
 
-    pub fn add_path_with_opts(&self, options: impl Into<AddPathOptions>) -> AddProgress {
+    pub fn add_path_with_opts(&self, options: impl Into<AddPathOptions>) -> AddProgress<'_> {
         let options = options.into();
         self.add_path_with_opts_impl(ImportPathRequest {
             path: options.path,
@@ -190,7 +190,7 @@ impl Blobs {
         })
     }
 
-    fn add_path_with_opts_impl(&self, options: ImportPathRequest) -> AddProgress {
+    fn add_path_with_opts_impl(&self, options: ImportPathRequest) -> AddProgress<'_> {
         trace!("{:?}", options);
         let client = self.client.clone();
         let stream = Gen::new(|co| async move {
@@ -215,7 +215,7 @@ impl Blobs {
         AddProgress::new(self, stream)
     }
 
-    pub fn add_path(&self, path: impl AsRef<Path>) -> AddProgress {
+    pub fn add_path(&self, path: impl AsRef<Path>) -> AddProgress<'_> {
         self.add_path_with_opts(AddPathOptions {
             path: path.as_ref().to_owned(),
             mode: ImportMode::Copy,
@@ -226,7 +226,7 @@ impl Blobs {
     pub async fn add_stream(
         &self,
         data: impl Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
-    ) -> AddProgress {
+    ) -> AddProgress<'_> {
         let inner = ImportByteStreamRequest {
             format: crate::BlobFormat::Raw,
             scope: Scope::default(),
@@ -521,7 +521,7 @@ pub struct Batch<'a> {
 }
 
 impl<'a> Batch<'a> {
-    pub fn add_bytes(&self, data: impl Into<Bytes>) -> BatchAddProgress {
+    pub fn add_bytes(&self, data: impl Into<Bytes>) -> BatchAddProgress<'_> {
         let options = ImportBytesRequest {
             data: data.into(),
             format: crate::BlobFormat::Raw,
@@ -530,7 +530,7 @@ impl<'a> Batch<'a> {
         BatchAddProgress(self.blobs.add_bytes_impl(options))
     }
 
-    pub fn add_bytes_with_opts(&self, options: impl Into<AddBytesOptions>) -> BatchAddProgress {
+    pub fn add_bytes_with_opts(&self, options: impl Into<AddBytesOptions>) -> BatchAddProgress<'_> {
         let options = options.into();
         BatchAddProgress(self.blobs.add_bytes_impl(ImportBytesRequest {
             data: options.data,
@@ -539,7 +539,7 @@ impl<'a> Batch<'a> {
         }))
     }
 
-    pub fn add_slice(&self, data: impl AsRef<[u8]>) -> BatchAddProgress {
+    pub fn add_slice(&self, data: impl AsRef<[u8]>) -> BatchAddProgress<'_> {
         let options = ImportBytesRequest {
             data: Bytes::copy_from_slice(data.as_ref()),
             format: crate::BlobFormat::Raw,
@@ -548,7 +548,7 @@ impl<'a> Batch<'a> {
         BatchAddProgress(self.blobs.add_bytes_impl(options))
     }
 
-    pub fn add_path_with_opts(&self, options: impl Into<AddPathOptions>) -> BatchAddProgress {
+    pub fn add_path_with_opts(&self, options: impl Into<AddPathOptions>) -> BatchAddProgress<'_> {
         let options = options.into();
         BatchAddProgress(self.blobs.add_path_with_opts_impl(ImportPathRequest {
             path: options.path,

src/metrics.rs

@@ -4,6 +4,7 @@ use iroh_metrics::{Counter, MetricsGroup};
 
 /// Enum of metrics for the module
 #[allow(missing_docs)]
+#[allow(dead_code)]
 #[derive(Debug, Default, MetricsGroup)]
 #[metrics(name = "iroh-blobs")]
 pub struct Metrics {

src/store/fs.rs

@@ -247,20 +247,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(())
     }
 
@@ -270,40 +258,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> {
@@ -893,7 +854,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/bao_file.rs

@@ -21,21 +21,17 @@ use bytes::{Bytes, BytesMut};
 use derive_more::Debug;
 use irpc::channel::mpsc;
 use tokio::sync::watch;
-use tracing::{debug, error, info, trace, Span};
+use tracing::{debug, error, info, trace};
 
 use super::{
     entry_state::{DataLocation, EntryState, OutboardLocation},
-    meta::Update,
     options::{Options, PathOptions},
     BaoFilePart,
 };
 use crate::{
     api::blobs::Bitfield,
     store::{
-        fs::{
-            meta::{raw_outboard_size, Set},
-            TaskContext,
-        },
+        fs::{meta::raw_outboard_size, TaskContext},
         util::{
             read_checksummed_and_truncate, write_checksummed, FixedSize, MemOrFile,
             PartialMemStorage, SizeInfo, SparseMemFile, DD,
@@ -644,21 +640,7 @@ impl BaoFileHandle {
             let size = storage.bitfield.size;
             let (storage, entry_state) = storage.into_complete(size, &options)?;
             debug!("File was reconstructed as complete");
-            let (tx, rx) = crate::util::channel::oneshot::channel();
-            ctx.db
-                .sender
-                .send(
-                    Set {
-                        hash,
-                        state: entry_state,
-                        tx,
-                        span: Span::current(),
-                    }
-                    .into(),
-                )
-                .await
-                .map_err(|_| io::Error::other("send update"))?;
-            rx.await.map_err(|_| io::Error::other("receive update"))??;
+            ctx.db.set(hash, entry_state).await?;
             storage.into()
         } else {
             storage.into()
@@ -796,19 +778,7 @@ impl BaoFileHandle {
             true
         });
         if let Some(update) = res? {
-            ctx.db
-                .sender
-                .send(
-                    Update {
-                        hash: self.hash,
-                        state: update,
-                        tx: None,
-                        span: Span::current(),
-                    }
-                    .into(),
-                )
-                .await
-                .map_err(|_| io::Error::other("send update"))?;
+            ctx.db.update(self.hash, update).await?;
         }
         Ok(())
     }

src/store/fs/meta.rs

@@ -34,7 +34,7 @@ mod proto;
 pub use proto::*;
 pub(crate) mod tables;
 use tables::{ReadOnlyTables, ReadableTables, Tables};
-use tracing::{debug, error, info_span, trace};
+use tracing::{debug, error, info_span, trace, Span};
 
 use super::{
     delete_set::DeleteHandle,
@@ -88,16 +88,70 @@ pub type ActorResult<T> = Result<T, ActorError>;
 
 #[derive(Debug, Clone)]
 pub struct Db {
-    pub sender: tokio::sync::mpsc::Sender<Command>,
+    sender: tokio::sync::mpsc::Sender<Command>,
 }
 
 impl Db {
     pub fn new(sender: tokio::sync::mpsc::Sender<Command>) -> Self {
         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
+            .send(
+                Update {
+                    hash,
+                    state,
+                    tx: None,
+                    span: Span::current(),
+                }
+                .into(),
+            )
+            .await
+            .map_err(|_| io::Error::other("send update"))
+    }
+
+    /// Set the entry state and await completion.
+    pub async fn set(&self, hash: Hash, entry_state: EntryState<Bytes>) -> io::Result<()> {
+        let (tx, rx) = oneshot::channel();
+        self.sender
+            .send(
+                Set {
+                    hash,
+                    state: entry_state,
+                    tx,
+                    span: Span::current(),
+                }
+                .into(),
+            )
+            .await
+            .map_err(|_| io::Error::other("send update"))?;
+        rx.await.map_err(|_| io::Error::other("receive update"))??;
+        Ok(())
+    }
+
     /// 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(
@@ -108,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?)
     }