fix(actor): drain Actor::tasks JoinSet in run_async (#96)

## Description

`Actor::tasks` is a `JoinSet<()>` that the three streaming-reply actions
(`Action::ListAuthors`, `Action::ListReplicas`,
`ReplicaAction::GetMany`) push tasks into via `spawn_local`, but
`run_async`'s `tokio::select!` had no `join_next` arm. Completed task
headers accumulate for the lifetime of the actor and only get released
when `abort_all()` runs at shutdown.

The fix adds the missing arm:

```rust
Some(res) = self.tasks.join_next(), if !self.tasks.is_empty() => {
    if let Err(err) = res {
        if !err.is_cancelled() {
            warn!(?err, "actor reply-streamer task panicked");
        }
    }
    continue;
}
```

This matches the canonical drain pattern used everywhere else in the
crate:

- `LiveActor::run_inner` drains three JoinSets in the same `select!`:
`running_sync_connect`, `running_sync_accept`, and `download_tasks`
(`src/engine/live.rs`).
- `GossipActor::progress` drains `active_tasks` with the same
`is_cancelled()` skip on `JoinError` (`src/engine/gossip.rs`).

`Actor::tasks` was the only `JoinSet` in the crate not following the
pattern. There is also a pre-existing TODO in `src/engine/live.rs` that
calls this out:

```
// TODO: abort_all and join_next all JoinSets to catch panics
```

The three `spawn_local` sites are kept as-is. `iter_to_irpc` is a
sync-iter → async-mpsc bridge that can stall on a slow consumer, so
inlining it would block the action queue (notably the `MAX_COMMIT_DELAY`
timeout flush). The fix lives entirely in the `select!`.

A regression test is added next to the existing `open_close` test in
`src/actor.rs::tests`. It fires 1000 calls of each streaming-reply shape
(3000 streamer tasks total), drains every reply stream to completion,
then asserts that `Actor::tasks` shrinks back to a small bound. Without
the new arm the residual is ~3000; with it the count drops to near zero
within tens of milliseconds. A `#[cfg(test)]`-gated
`Action::DebugTasksLen` and `SyncHandle::debug_tasks_len` provide the
introspection.

## Breaking Changes

None.

## Notes & open questions

The test introspection is `#[cfg(test)]`-only. Happy to instead expose a
public `wait_idle()`-style helper if you'd prefer that shape.

## 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: cbenhagen

src/actor.rs

@@ -89,6 +89,12 @@ enum Action {
         #[debug("reply")]
         reply: Option<oneshot::Sender<Store>>,
     },
+    #[cfg(test)]
+    #[display("DebugTasksLen")]
+    DebugTasksLen {
+        #[debug("reply")]
+        reply: oneshot::Sender<usize>,
+    },
 }
 
 #[derive(derive_more::Debug, strum::Display)]
@@ -514,6 +520,13 @@ impl SyncHandle {
         Ok(store)
     }
 
+    #[cfg(test)]
+    async fn debug_tasks_len(&self) -> Result<usize> {
+        let (reply, rx) = oneshot::channel();
+        self.send(Action::DebugTasksLen { reply }).await?;
+        Ok(rx.await?)
+    }
+
     pub async fn list_authors(
         &self,
         reply: mpsc::Sender<RpcResult<AuthorListResponse>>,
@@ -661,6 +674,14 @@ impl Actor {
                     }
                     continue;
                 }
+                Some(res) = self.tasks.join_next(), if !self.tasks.is_empty() => {
+                    if let Err(err) = res {
+                        if !err.is_cancelled() {
+                            warn!(?err, "actor reply-streamer task panicked");
+                        }
+                    }
+                    continue;
+                }
                 action = self.action_rx.recv() => {
                     match action {
                         Ok(action) => action,
@@ -702,6 +723,8 @@ impl Actor {
             Action::Shutdown { .. } => {
                 unreachable!("Shutdown is handled in run()")
             }
+            #[cfg(test)]
+            Action::DebugTasksLen { reply } => send_reply(reply, self.tasks.len()),
             Action::ImportAuthor { author, reply } => {
                 let id = author.id();
                 send_reply(reply, self.store.import_author(author).map(|_| id))
@@ -1112,4 +1135,60 @@ mod tests {
         assert!(rx.recv().await.is_err());
         Ok(())
     }
+
+    /// Tests that streamer tasks spawned into `Actor.tasks` are reaped
+    /// once they complete.
+    ///
+    /// The three streaming actions (`ListAuthors`, `ListReplicas`, and
+    /// `ReplicaAction::GetMany`) each `spawn_local` a task into
+    /// `Actor.tasks` to drive their reply channel. The actor must
+    /// `join_next` those tasks once they finish, otherwise the
+    /// `JoinSet` grows without bound for the lifetime of the actor.
+    #[tokio::test]
+    async fn actor_tasks_joinset_drain() -> anyhow::Result<()> {
+        let store = store::Store::memory();
+        let sync = SyncHandle::spawn(store, None, "drain".into());
+
+        let namespace = NamespaceSecret::new(&mut rand::rng());
+        let id = namespace.id();
+        sync.import_namespace(namespace.into()).await?;
+        sync.open(id, Default::default()).await?;
+
+        const ITERATIONS: usize = 1000;
+
+        for _ in 0..ITERATIONS {
+            let (tx, mut rx) = mpsc::channel(64);
+            sync.list_authors(tx).await?;
+            while rx.recv().await?.is_some() {}
+        }
+
+        for _ in 0..ITERATIONS {
+            let (tx, mut rx) = mpsc::channel(64);
+            sync.list_replicas(tx).await?;
+            while rx.recv().await?.is_some() {}
+        }
+
+        for _ in 0..ITERATIONS {
+            let (tx, mut rx) = mpsc::channel(64);
+            sync.get_many(id, store::Query::all().into(), tx).await?;
+            while rx.recv().await?.is_some() {}
+        }
+
+        let mut last = sync.debug_tasks_len().await?;
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        while last > 16 && std::time::Instant::now() < deadline {
+            tokio::time::sleep(Duration::from_millis(50)).await;
+            last = sync.debug_tasks_len().await?;
+        }
+
+        assert!(
+            last <= 16,
+            "residual Actor.tasks JoinSet len = {last}, expected <= 16 \
+             (was the join_next arm in run_async lost? streamer tasks \
+             for ListAuthors / ListReplicas / GetMany are not being reaped)"
+        );
+
+        sync.close(id).await?;
+        Ok(())
+    }
 }