refactor(iroh-blobs): use oneshot channel from oneshot crate (#2624)

## Description

In several places we need a oneshot channel to communicate between sync
and async code. In particular, there is one place where we intentionally
block inside async code to be able to implement a sync callback (in the
docs code, will go away at some point hopefully).

So far we have used flume and now async_channel channels with capacity 1
for this, which is quite wasteful.

This replaces all oneshot use in the fs store with the oneshot channel
from the oneshot crate, which is pretty minimal and does not panic when
calling blocking_recv from a tokio context.

## Breaking Changes

None

## Notes & open questions

Note: this was part of an attempt to get rid of async_channel in
iroh-blobs. We still need it for sync interactions with the blob store
from inside tokio tasks, for the current docs impl. But hopefully we
should be able to replace the channel for the redb actor with a mpsc
channel once that is sorted out.

## Change checklist

- [x] Self-review.
- [x] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [x] Tests if relevant.
- [x] All breaking changes documented.

iroh-blobs/Cargo.toml

@@ -32,6 +32,7 @@ iroh-io = { version = "0.6.0", features = ["stats"] }
 iroh-metrics = { version = "0.22.0", path = "../iroh-metrics", default-features = false }
 iroh-net = { version = "0.22.0", path = "../iroh-net" }
 num_cpus = "1.15.0"
+oneshot = "0.1.8"
 parking_lot = { version = "0.12.1", optional = true }
 pin-project = "1.1.5"
 postcard = { version = "1", default-features = false, features = ["alloc", "use-std", "experimental-derive"] }

iroh-blobs/src/store/fs.rs

@@ -83,7 +83,7 @@ use iroh_io::AsyncSliceReader;
 use redb::{AccessGuard, DatabaseError, ReadableTable, StorageError};
 use serde::{Deserialize, Serialize};
 use smallvec::SmallVec;
-use tokio::{io::AsyncWriteExt, sync::oneshot};
+use tokio::io::AsyncWriteExt;
 use tracing::trace_span;
 
 mod import_flat_store;
@@ -534,25 +534,25 @@ pub(crate) enum ActorMessage {
     /// Query method: get the rough entry status for a hash. Just complete, partial or not found.
     EntryStatus {
         hash: Hash,
-        tx: async_channel::Sender<ActorResult<EntryStatus>>,
+        tx: oneshot::Sender<ActorResult<EntryStatus>>,
     },
     #[cfg(test)]
     /// Query method: get the full entry state for a hash, both in memory and in redb.
     /// This is everything we got about the entry, including the actual inline outboard and data.
     EntryState {
         hash: Hash,
-        tx: async_channel::Sender<ActorResult<test_support::EntryStateResponse>>,
+        tx: oneshot::Sender<ActorResult<test_support::EntryStateResponse>>,
     },
     /// Query method: get the full entry state for a hash.
     GetFullEntryState {
         hash: Hash,
-        tx: async_channel::Sender<ActorResult<Option<EntryData>>>,
+        tx: oneshot::Sender<ActorResult<Option<EntryData>>>,
     },
     /// Modification method: set the full entry state for a hash.
     SetFullEntryState {
         hash: Hash,
         entry: Option<EntryData>,
-        tx: async_channel::Sender<ActorResult<()>>,
+        tx: oneshot::Sender<ActorResult<()>>,
     },
     /// Modification method: get or create a file handle for a hash.
     ///
@@ -575,7 +575,7 @@ pub(crate) enum ActorMessage {
     /// At this point the size, hash and outboard must already be known.
     Import {
         cmd: Import,
-        tx: async_channel::Sender<ActorResult<(TempTag, u64)>>,
+        tx: oneshot::Sender<ActorResult<(TempTag, u64)>>,
     },
     /// Modification method: export data from a redb store
     ///
@@ -921,18 +921,18 @@ impl StoreInner {
     }
 
     async fn entry_status(&self, hash: &Hash) -> OuterResult<EntryStatus> {
-        let (tx, rx) = async_channel::bounded(1);
+        let (tx, rx) = oneshot::channel();
         self.tx
             .send(ActorMessage::EntryStatus { hash: *hash, tx })
             .await?;
-        Ok(rx.recv().await??)
+        Ok(rx.await??)
     }
 
     fn entry_status_sync(&self, hash: &Hash) -> OuterResult<EntryStatus> {
-        let (tx, rx) = async_channel::bounded(1);
+        let (tx, rx) = oneshot::channel();
         self.tx
             .send_blocking(ActorMessage::EntryStatus { hash: *hash, tx })?;
-        Ok(rx.recv_blocking()??)
+        Ok(rx.recv()??)
     }
 
     async fn complete(&self, entry: Entry) -> OuterResult<()> {
@@ -1128,7 +1128,7 @@ impl StoreInner {
         let tag = self.temp.temp_tag(HashAndFormat { hash, format });
         let hash = *tag.hash();
         // blocking send for the import
-        let (tx, rx) = async_channel::bounded(1);
+        let (tx, rx) = oneshot::channel();
         self.tx.send_blocking(ActorMessage::Import {
             cmd: Import {
                 content_id: HashAndFormat { hash, format },
@@ -1138,7 +1138,7 @@ impl StoreInner {
             },
             tx,
         })?;
-        Ok(rx.recv_blocking()??)
+        Ok(rx.recv()??)
     }
 
     fn temp_file_name(&self) -> PathBuf {
@@ -1231,18 +1231,24 @@ pub(crate) type ActorResult<T> = std::result::Result<T, ActorError>;
 pub(crate) enum OuterError {
     #[error("inner error: {0}")]
     Inner(#[from] ActorError),
-    #[error("send error: {0}")]
-    Send(#[from] async_channel::SendError<ActorMessage>),
+    #[error("send error")]
+    Send,
     #[error("progress send error: {0}")]
     ProgressSend(#[from] ProgressSendError),
     #[error("recv error: {0}")]
-    Recv(#[from] oneshot::error::RecvError),
+    Recv(#[from] oneshot::RecvError),
     #[error("recv error: {0}")]
     AsyncChannelRecv(#[from] async_channel::RecvError),
     #[error("join error: {0}")]
     JoinTask(#[from] tokio::task::JoinError),
 }
 
+impl From<async_channel::SendError<ActorMessage>> for OuterError {
+    fn from(_e: async_channel::SendError<ActorMessage>) -> Self {
+        OuterError::Send
+    }
+}
+
 /// Result type for calling the redb actor from the store.
 ///
 /// See [`OuterError`] for what can go wrong.
@@ -2236,7 +2242,7 @@ impl ActorState {
             }
             ActorMessage::EntryStatus { hash, tx } => {
                 let res = self.entry_status(tables, hash);
-                tx.send_blocking(res).ok();
+                tx.send(res).ok();
             }
             ActorMessage::Blobs { filter, tx } => {
                 let res = self.blobs(tables, filter);
@@ -2256,11 +2262,11 @@ impl ActorState {
             }
             #[cfg(test)]
             ActorMessage::EntryState { hash, tx } => {
-                tx.send_blocking(self.entry_state(tables, hash)).ok();
+                tx.send(self.entry_state(tables, hash)).ok();
             }
             ActorMessage::GetFullEntryState { hash, tx } => {
                 let res = self.get_full_entry_state(tables, hash);
-                tx.send_blocking(res).ok();
+                tx.send(res).ok();
             }
             x => return Ok(Err(x)),
         }
@@ -2275,7 +2281,7 @@ impl ActorState {
         match msg {
             ActorMessage::Import { cmd, tx } => {
                 let res = self.import(tables, cmd);
-                tx.send_blocking(res).ok();
+                tx.send(res).ok();
             }
             ActorMessage::SetTag { tag, value, tx } => {
                 let res = self.set_tag(tables, tag, value);
@@ -2306,7 +2312,7 @@ impl ActorState {
             }
             ActorMessage::SetFullEntryState { hash, entry, tx } => {
                 let res = self.set_full_entry_state(tables, hash, entry);
-                tx.send_blocking(res).ok();
+                tx.send(res).ok();
             }
             msg => {
                 // try to handle it as readonly

iroh-blobs/src/store/fs/test_support.rs

@@ -7,8 +7,6 @@ use std::{
     path::{Path, PathBuf},
 };
 
-use tokio::sync::oneshot;
-
 use super::{
     tables::{ReadableTables, Tables},
     ActorError, ActorMessage, ActorResult, ActorState, DataLocation, EntryState, FilterPredicate,
@@ -106,25 +104,25 @@ impl Store {
 impl StoreInner {
     #[cfg(test)]
     async fn entry_state(&self, hash: Hash) -> OuterResult<EntryStateResponse> {
-        let (tx, rx) = async_channel::bounded(1);
+        let (tx, rx) = oneshot::channel();
         self.tx.send(ActorMessage::EntryState { hash, tx }).await?;
-        Ok(rx.recv().await??)
+        Ok(rx.await??)
     }
 
     async fn set_full_entry_state(&self, hash: Hash, entry: Option<EntryData>) -> OuterResult<()> {
-        let (tx, rx) = async_channel::bounded(1);
+        let (tx, rx) = oneshot::channel();
         self.tx
             .send(ActorMessage::SetFullEntryState { hash, entry, tx })
             .await?;
-        Ok(rx.recv().await??)
+        Ok(rx.await??)
     }
 
     async fn get_full_entry_state(&self, hash: Hash) -> OuterResult<Option<EntryData>> {
-        let (tx, rx) = async_channel::bounded(1);
+        let (tx, rx) = oneshot::channel();
         self.tx
             .send(ActorMessage::GetFullEntryState { hash, tx })
             .await?;
-        Ok(rx.recv().await??)
+        Ok(rx.await??)
     }
 
     async fn all_blobs(&self) -> OuterResult<Vec<io::Result<Hash>>> {