Remove futures crate

Squashed commit of the following:

commit eb8cabbf7aa3aa09cffdf8d5d5e0d1e56994cacd
Author: Lukas Herman <[email protected]>
Date:   Fri Dec 12 15:41:19 2025 -0500

    compilable

commit 8207e71
Author: Lukas Herman <[email protected]>
Date:   Fri Dec 12 14:39:21 2025 -0500

    init futures removal

Author: lherman-cs

Cargo.lock

@@ -32,7 +32,7 @@ serde = { version = "1", features = ["derive"]}
 thiserror = "2.0"
 tokio = { version = "1", features = ["full"] }
 tracing = { version = "0.1", features = ["release_max_level_debug"] }
-futures = "0.3.31"
+futures-lite = "2.6.1"
 rand = "0.9"
 bytes = "1.10.1"
 str0m = { git = "https://github.com/PulseBeamDev/str0m.git", branch = "patch/main", default-features = false, features = ["openssl", "vendored", "sha1"] }

Cargo.toml

@@ -19,7 +19,7 @@ ffmpeg-static = ["ffmpeg-next/build"]
 ffmpeg-dynamic = []
 
 [dependencies]
-futures = { workspace = true }
+futures-lite = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
 str0m = { workspace = true }

pulsebeam-agent/Cargo.toml

@@ -25,7 +25,7 @@ tokio = { workspace = true }
 tracing = { workspace = true }
 rand = { workspace = true }
 bytes = { workspace = true }
-futures.workspace = true
+futures-lite = { workspace = true }
 turmoil = "0.6.6"
 quinn-udp = "0.5.14"
 socket2 = "0.6.0"

pulsebeam-runtime/Cargo.toml

@@ -1,11 +1,11 @@
 use criterion::{Criterion, criterion_group, criterion_main};
-use futures::{StreamExt, future::join_all};
 use futures_concurrency::stream::Merge;
+use futures_lite::StreamExt;
 use rand::seq::index;
 use std::time::{Duration, Instant};
-use tokio::runtime::Runtime;
 use tokio::sync::mpsc;
 use tokio::task;
+use tokio::{runtime::Runtime, task::JoinSet};
 
 // Use the specified import path for the SPMC channel implementation.
 use pulsebeam_runtime::sync::spmc::{RecvError, Sender, channel};
@@ -111,7 +111,7 @@ async fn run_interactive_room_mesh_mpsc_fanout_test() {
     }
 
     let simulation_start = Instant::now();
-    join_all(publisher_tasks).await;
+    JoinSet::from_iter(publisher_tasks).join_all().await;
     let total_send_duration = simulation_start.elapsed();
 
     let all_latencies = aggregate_latencies(subscriber_tasks).await;
@@ -210,7 +210,7 @@ async fn run_interactive_room_mesh_spawn_test() {
     }
 
     let simulation_start = Instant::now();
-    join_all(publisher_tasks).await;
+    JoinSet::from_iter(publisher_tasks).join_all().await;
     let total_send_duration = simulation_start.elapsed();
 
     let all_latencies = aggregate_latencies(subscriber_tasks).await;
@@ -298,7 +298,7 @@ async fn run_interactive_room_mesh_futures_unordered_test() {
     }
 
     let simulation_start = Instant::now();
-    join_all(publisher_tasks).await;
+    JoinSet::from_iter(publisher_tasks).join_all().await;
     let total_send_duration = simulation_start.elapsed();
 
     let all_latencies = aggregate_latencies(subscriber_tasks).await;
@@ -416,12 +416,11 @@ async fn create_publisher_load_mpsc(tx: mpsc::Sender<(usize, Instant)>, num_pack
 }
 
 async fn aggregate_latencies(handles: Vec<task::JoinHandle<Vec<Duration>>>) -> Vec<Duration> {
-    let results = join_all(handles).await;
+    let results = JoinSet::from_iter(handles).join_all().await;
+    let results = results.iter().flatten();
     let mut all_latencies = Vec::new();
     for result in results {
-        if let Ok(subscriber_latencies) = result {
-            all_latencies.extend(subscriber_latencies);
-        }
+        all_latencies.extend(result);
     }
     all_latencies
 }

pulsebeam-runtime/benches/sync.rs

@@ -1,9 +1,10 @@
 use crate::actor_loop;
-use futures::FutureExt;
+use futures_lite::FutureExt;
 use std::any::Any;
 use std::cell::RefCell;
 use std::collections::HashMap;
 use std::fmt::{Debug, Display, Formatter};
+use std::future::Future;
 use std::hash::Hash;
 use std::panic::AssertUnwindSafe;
 use std::pin::Pin;
@@ -25,7 +26,7 @@ tokio::task_local! {
 }
 
 pub struct JoinHandle<M: MessageSet> {
-    inner: Pin<Box<dyn futures::Future<Output = (M::Meta, ActorStatus)> + Send>>,
+    inner: Pin<Box<dyn Future<Output = (M::Meta, ActorStatus)> + Send>>,
     abort_handle: tokio::task::AbortHandle,
 }
 
@@ -41,7 +42,7 @@ impl<M: MessageSet> Drop for JoinHandle<M> {
     }
 }
 
-impl<M: MessageSet> futures::Future for JoinHandle<M> {
+impl<M: MessageSet> Future for JoinHandle<M> {
     type Output = (M::Meta, ActorStatus);
 
     fn poll(
@@ -117,7 +118,7 @@ pub trait Actor<M: MessageSet>: Sized + Send + 'static {
     fn run(
         &mut self,
         ctx: &mut ActorContext<M>,
-    ) -> impl futures::Future<Output = Result<(), ActorError>> + Send {
+    ) -> impl Future<Output = Result<(), ActorError>> + Send {
         async {
             actor_loop!(self, ctx);
             Ok(())
@@ -128,15 +129,15 @@ pub trait Actor<M: MessageSet>: Sized + Send + 'static {
         &mut self,
         _ctx: &mut ActorContext<M>,
         _msg: SystemMsg<M::ObservableState>,
-    ) -> impl futures::Future<Output = ()> + Send {
+    ) -> impl Future<Output = ()> + Send {
         async move {}
     }
 
     fn on_msg(
         &mut self,
         _ctx: &mut ActorContext<M>,
         _msg: M::Msg,
-    ) -> impl futures::Future<Output = ()> + Send {
+    ) -> impl Future<Output = ()> + Send {
         async move {}
     }
 }
@@ -254,40 +255,12 @@ where
     let runnable = tracing::Instrument::instrument(runnable, span);
 
     let actor_id = handle.meta.clone();
-    let fut = runnable.map(move |status| (actor_id, status));
-
-    (handle, fut)
-}
-
-pub fn spawn<A, M>(a: A, config: RunnerConfig) -> (ActorHandle<M>, JoinHandle<M>)
-where
-    M: MessageSet,
-    A: Actor<M>,
-{
-    let (handle, runnable) = prepare(a, config);
-    let join = tokio::spawn(runnable);
-    let abort_handle = join.abort_handle();
-
-    let actor_id = handle.meta.clone();
-    let join = join
-        .map(|res| res.unwrap_or((actor_id, ActorStatus::ShutDown)))
-        .boxed();
-
-    (
-        handle,
-        JoinHandle {
-            inner: join,
-            abort_handle,
-        },
-    )
-}
+    let runnable = async move {
+        let status = runnable.await;
+        (actor_id, status)
+    };
 
-pub fn spawn_default<A, M>(a: A) -> (ActorHandle<M>, JoinHandle<M>)
-where
-    M: MessageSet,
-    A: Actor<M>,
-{
-    spawn(a, RunnerConfig::default())
+    (handle, runnable)
 }
 
 async fn run<A, M>(mut a: A, mut ctx: ActorContext<M>) -> ActorStatus

pulsebeam-runtime/src/actor.rs

@@ -1,5 +1,5 @@
 pub use crate::actor::{Actor, ActorHandle};
 pub use crate::rand::{RngCore, SeedableRng};
-pub use futures::{FutureExt, StreamExt};
+pub use futures_lite::{FutureExt, StreamExt};
 pub use once_cell::sync::Lazy;
 pub use tokio_metrics::TaskMonitor;

pulsebeam-runtime/src/prelude.rs

@@ -1,18 +1,24 @@
 use crossbeam_utils::CachePadded;
 use event_listener::{Event, EventListener};
+use futures_lite::Stream;
 use parking_lot::RwLock;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::task::{Context, Poll};
+use std::task::{Context, Poll, ready};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum RecvError {
     Lagged(u64),
     Closed,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum StreamRecvError {
+    Lagged(u64),
+}
+
 #[derive(Debug)]
 struct Slot<T> {
     seq: u64,
@@ -171,6 +177,21 @@ impl<T: Clone> Receiver<T> {
     }
 }
 
+impl<T: Clone> Stream for Receiver<T> {
+    type Item = Result<T, StreamRecvError>;
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.get_mut();
+
+        let res = match ready!(this.poll_recv(cx)) {
+            Ok(item) => Some(Ok(item)),
+            Err(RecvError::Lagged(n)) => Some(Err(StreamRecvError::Lagged(n))),
+            Err(RecvError::Closed) => None,
+        };
+
+        Poll::Ready(res)
+    }
+}
+
 impl<T: Clone> Clone for Receiver<T> {
     fn clone(&self) -> Self {
         Self {

pulsebeam-runtime/src/sync/spmc.rs

@@ -11,7 +11,7 @@ license.workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
-futures = { workspace = true }
+futures-lite = { workspace = true }
 serde = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }

pulsebeam/Cargo.toml

@@ -4,11 +4,10 @@ use crate::{
     entity::{ParticipantId, RoomId},
     node, room,
 };
-use futures::stream::FuturesUnordered;
-use pulsebeam_runtime::actor::{self, ActorKind};
+use pulsebeam_runtime::actor::{self, ActorKind, ActorStatus};
 use pulsebeam_runtime::prelude::*;
 use str0m::{Candidate, RtcConfig, RtcError, change::SdpOffer, error::SdpError};
-use tokio::sync::oneshot;
+use tokio::{sync::oneshot, task::JoinSet};
 
 #[derive(thiserror::Error, Debug)]
 pub enum ControllerError {
@@ -54,7 +53,7 @@ pub struct ControllerActor {
     local_addrs: Vec<SocketAddr>,
 
     rooms: HashMap<Arc<RoomId>, room::RoomHandle>,
-    room_tasks: FuturesUnordered<actor::JoinHandle<room::RoomMessageSet>>,
+    room_tasks: JoinSet<(Arc<RoomId>, ActorStatus)>,
 }
 
 impl actor::Actor<ControllerMessageSet> for ControllerActor {
@@ -79,7 +78,7 @@ impl actor::Actor<ControllerMessageSet> for ControllerActor {
     ) -> Result<(), actor::ActorError> {
         pulsebeam_runtime::actor_loop!(self, ctx, pre_select:{} ,
             select: {
-                Some((room_id, _)) = self.room_tasks.next() => {
+                Some(Ok((room_id, _))) = self.room_tasks.join_next() => {
                     self.rooms.remove(&room_id);
                 }
             }
@@ -203,11 +202,11 @@ impl ControllerActor {
         } else {
             tracing::info!("create_room: {}", room_id);
             let room_actor = room::RoomActor::new(self.node_ctx.clone(), room_id.clone());
-            let (room_handle, room_join) = actor::spawn(
+            let (room_handle, room_task) = actor::prepare(
                 room_actor,
                 actor::RunnerConfig::default().with_mailbox_cap(1024),
             );
-            self.room_tasks.push(room_join);
+            self.room_tasks.spawn(room_task);
 
             self.rooms.insert(room_id.clone(), room_handle.clone());
 
@@ -227,7 +226,7 @@ impl ControllerActor {
             node_ctx: system_ctx,
             local_addrs,
             rooms: HashMap::new(),
-            room_tasks: FuturesUnordered::new(),
+            room_tasks: JoinSet::new(),
         }
     }
 }