fix(iroh-blobs): use async_channel instead of flume for local_pool (#2533)

## Description

During soft shutdown of the local pool, a Finish message is sent to all
threads. On main, this occasionally hangs. Further investigation showed
that this is a message that is being sent but not received despite being
in the channel. Adding a simple timeout to the select! so the flume recv
call is executed again fixes it.

See discussion in
https://discord.com/channels/949724860232392765/950683937661935667/1265205285618847744

So it seems that there is a bug in flume that occasionally leads to
notifications being dropped.

This PR just does a 1:1 replacement of flume with async_channel.

Before this change, I can get the test_shutdown test to fail easily by
running it 1000 times:

```
for i in $(seq 1 1000); do cargo test --release -p iroh-blobs test_shutdown -- --nocapture >> log.txt; done
```

Result:
```
...
sending shutdown message
sending shutdown message
sending shutdown message
sending shutdown message
test util::local_pool::tests::test_shutdown has been running for over 60 seconds
```

After this change, I can not get test_finish (renamed because it tests
finish, not shutdown) to fail at all even after several 1000 tests.

## Breaking Changes

None

## Notes & open questions

Can somebody talk me out of this? I would prefer to keep flume, but the
evidence above seems conclusive...

Note: why not tokio::sync::mpsc::channel? I need a mpmc channel. The
handle can be cloned, and can send to any of the n worker threads.

## 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.~~

Author: rklaehn

Cargo.lock

@@ -17,6 +17,7 @@ workspace = true
 
 [dependencies]
 anyhow = { version = "1" }
+async-channel = "2.3.1"
 bao-tree = {  version = "0.13", features = ["tokio_fsm", "validate"], default-features = false }
 bytes = { version = "1.4", features = ["serde"] }
 chrono = "0.4.31"

iroh-blobs/Cargo.toml

@@ -61,7 +61,7 @@ impl Deref for LocalPool {
 #[derive(Debug, Clone)]
 pub struct LocalPoolHandle {
     /// The sender half of the channel used to send tasks to the pool
-    send: flume::Sender<Message>,
+    send: async_channel::Sender<Message>,
 }
 
 /// What to do when a panic occurs in a pool thread
@@ -124,7 +124,7 @@ impl LocalPool {
             panic_mode,
         } = config;
         let cancel_token = CancellationToken::new();
-        let (send, recv) = flume::unbounded::<Message>();
+        let (send, recv) = async_channel::unbounded::<Message>();
         let shutdown_sem = Arc::new(Semaphore::new(0));
         let handle = tokio::runtime::Handle::current();
         let handles = (0..threads)
@@ -159,7 +159,7 @@ impl LocalPool {
     /// Spawn a new pool thread.
     fn spawn_pool_thread(
         thread_name: String,
-        recv: flume::Receiver<Message>,
+        recv: async_channel::Receiver<Message>,
         cancel_token: CancellationToken,
         panic_mode: PanicMode,
         shutdown_sem: Arc<Semaphore>,
@@ -198,18 +198,18 @@ impl LocalPool {
                             // if the cancel token is cancelled, break the loop immediately
                             _ = cancel_token.cancelled() => break ShutdownMode::Stop,
                             // if we receive a message, execute it
-                            msg = recv.recv_async() => {
+                            msg = recv.recv() => {
                                 match msg {
-                                    // just push into the FuturesUnordered
+                                    // just push into the join set
                                     Ok(Message::Execute(f)) => {
                                         s.spawn_local((f)());
                                     }
                                     // break with optional semaphore
                                     Ok(Message::Finish) => break ShutdownMode::Finish,
                                     // if the sender is dropped, break the loop immediately
-                                    Err(flume::RecvError::Disconnected) => break ShutdownMode::Stop,
+                                    Err(async_channel::RecvError) => break ShutdownMode::Stop,
                                 }
-                            }
+                            },
                         }
                     }
                 }));
@@ -247,7 +247,7 @@ impl LocalPool {
 
     /// Immediately stop polling all tasks and wait for all threads to finish.
     ///
-    /// This is like droo, but waits for thread completion asynchronously.
+    /// This is like drop, but waits for thread completion asynchronously.
     ///
     /// If there was a panic on any of the threads, it will be re-thrown here.
     pub async fn shutdown(self) {
@@ -270,14 +270,13 @@ impl LocalPool {
         // we assume that there are exactly as many threads as there are handles.
         // also, we assume that the threads are still running.
         for _ in 0..self.threads_u32() {
-            println!("sending shutdown message");
             // send the shutdown message
             // sending will fail if all threads are already finished, but
             // in that case we don't need to do anything.
             //
             // Threads will add a permit in any case, so await_thread_completion
             // will then immediately return.
-            self.send.send(Message::Finish).ok();
+            self.send.send(Message::Finish).await.ok();
         }
         self.await_thread_completion().await;
     }
@@ -460,7 +459,7 @@ impl LocalPoolHandle {
     /// spawn a task in the pool.
     pub fn try_spawn_detached_boxed(&self, gen: SpawnFn) -> SpawnResult<()> {
         self.send
-            .send(Message::Execute(gen))
+            .send_blocking(Message::Execute(gen))
             .map_err(|_| SpawnError::Cancelled)
     }
 }
@@ -593,7 +592,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_shutdown() {
+    async fn test_finish() {
         let _ = tracing_subscriber::fmt::try_init();
         let pool = LocalPool::new(Config::default());
         let counter = Arc::new(AtomicU64::new(0));