chore: slim the futures facade dependency

crates/app/src/sse/mod.rs

@@ -472,7 +472,7 @@ async fn stream_once(
             return StreamOutcome::Error { productive: false };
         }
     };
-    futures::pin_mut!(stream);
+    let mut stream = std::pin::pin!(stream);
 
     let mut productive = false;
     loop {

crates/cluster/Cargo.toml

@@ -25,7 +25,7 @@ pluto-k1util.workspace = true
 pluto-ssz.workspace = true
 k256.workspace = true
 tokio.workspace = true
-futures.workspace = true
+tokio-stream.workspace = true
 tracing.workspace = true
 reqwest = { workspace = true, features = ["json", "stream"] }
 # Workaround to use test code from different crate.

crates/cluster/src/helpers.rs

@@ -84,7 +84,7 @@ async fn read_body_capped(
     response: reqwest::Response,
     max: usize,
 ) -> std::result::Result<Vec<u8>, FetchError> {
-    use futures::StreamExt;
+    use tokio_stream::StreamExt;
 
     // Reject early if the server advertised an oversized body.
     if let Some(len) = response.content_length()

crates/consensus/src/qbft/p2p.rs

@@ -761,10 +761,10 @@ mod tests {
         collections::{BTreeMap, HashSet},
         error::Error as StdError,
         sync::OnceLock,
-        task::{Context, Poll},
+        task::{Context, Poll, Waker},
     };
 
-    use futures::{StreamExt as _, io::Cursor, task::noop_waker};
+    use futures::{StreamExt as _, io::Cursor};
     use k256::SecretKey;
     use libp2p::{
         Multiaddr, PeerId,
@@ -1506,8 +1506,7 @@ mod tests {
     fn drain_behaviour_events(
         behaviour: &mut Behaviour,
     ) -> Vec<ToSwarm<Event, THandlerInEvent<Behaviour>>> {
-        let waker = noop_waker();
-        let mut cx = Context::from_waker(&waker);
+        let mut cx = Context::from_waker(Waker::noop());
         let mut events = Vec::new();
 
         while let Poll::Ready(event) = NetworkBehaviour::poll(behaviour, &mut cx) {

crates/core/Cargo.toml

@@ -17,7 +17,6 @@ chrono.workspace = true
 crossbeam.workspace = true
 dyn-clone.workspace = true
 dyn-eq.workspace = true
-futures.workspace = true
 hex.workspace = true
 vise.workspace = true
 pluto-crypto.workspace = true

crates/core/src/bcast/recast.rs

@@ -1,10 +1,10 @@
 use std::{
     collections::{HashMap, HashSet},
     future::Future,
+    pin::Pin,
     sync::{Arc, Mutex},
 };
 
-use futures::future::BoxFuture;
 use pluto_eth2api::BeaconNodeClient;
 
 use crate::{
@@ -15,7 +15,7 @@ use crate::{
     types::{Duty, DutyType, PubKey, SignedData, SignedDataSet, Slot},
 };
 
-type RecastFuture = BoxFuture<'static, Result<()>>;
+type RecastFuture = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
 type RecastSubscriber = Arc<dyn Fn(Duty, SignedDataSet) -> RecastFuture + Send + Sync>;
 
 #[derive(Clone)]

crates/dkg/src/sync/behaviour.rs

@@ -247,9 +247,8 @@ impl NetworkBehaviour for Behaviour {
 
 #[cfg(test)]
 mod tests {
-    use std::task::Context;
+    use std::task::{Context, Waker};
 
-    use futures::task::noop_waker_ref;
     use libp2p::{
         core::{ConnectedPoint, Endpoint, transport::PortUse},
         swarm::{
@@ -281,8 +280,7 @@ mod tests {
     }
 
     fn assert_next_dial(behaviour: &mut Behaviour, peer_id: PeerId, message: &str) {
-        let waker = noop_waker_ref();
-        let mut cx = Context::from_waker(waker);
+        let mut cx = Context::from_waker(Waker::noop());
         let poll = NetworkBehaviour::poll(behaviour, &mut cx);
 
         let Poll::Ready(ToSwarm::Dial { opts }) = poll else {
@@ -292,8 +290,7 @@ mod tests {
     }
 
     fn assert_pending(behaviour: &mut Behaviour, message: &str) {
-        let waker = noop_waker_ref();
-        let mut cx = Context::from_waker(waker);
+        let mut cx = Context::from_waker(Waker::noop());
         assert!(
             NetworkBehaviour::poll(behaviour, &mut cx).is_pending(),
             "{message}"

crates/dkg/src/sync/handler.rs

@@ -525,9 +525,8 @@ fn is_relay_io_error(error: &io::Error) -> bool {
 
 #[cfg(test)]
 mod tests {
-    use std::task::{Context, Poll};
+    use std::task::{Context, Poll, Waker};
 
-    use futures::task::noop_waker_ref;
     use libp2p::swarm::{ConnectionHandler, ConnectionHandlerEvent};
     use pluto_core::version::SemVer;
     use tokio::{sync::mpsc, time::Duration};
@@ -571,8 +570,7 @@ mod tests {
         handler.schedule_retry();
 
         tokio::time::sleep(Duration::from_millis(2)).await;
-        let waker = noop_waker_ref();
-        let mut cx = Context::from_waker(waker);
+        let mut cx = Context::from_waker(Waker::noop());
 
         let poll = ConnectionHandler::poll(&mut handler, &mut cx);
         assert!(matches!(poll, Poll::Pending));

crates/eth2api/Cargo.toml

@@ -16,7 +16,6 @@ anyhow.workspace = true
 async-trait.workspace = true
 bon.workspace = true
 eventsource-stream.workspace = true
-futures.workspace = true
 http.workspace = true
 oas3-gen-support.workspace = true
 regex.workspace = true
@@ -35,6 +34,7 @@ tree_hash_derive.workspace = true
 alloy.workspace = true
 pluto-ssz.workspace = true
 tokio.workspace = true
+tokio-stream.workspace = true
 vise.workspace = true
 
 [dev-dependencies]

crates/eth2api/src/extensions.rs

@@ -8,14 +8,14 @@ use crate::{
 };
 use chrono::{DateTime, Utc};
 use eventsource_stream::Eventsource;
-use futures::{Stream, StreamExt};
 use reqwest::Url;
 use std::{
     collections::{HashMap, HashSet},
     sync::{Arc, LazyLock, Mutex},
     time,
 };
 use tokio::sync::OnceCell;
+use tokio_stream::{Stream, StreamExt};
 use tree_hash::TreeHash;
 
 /// Error that can occur when using the
@@ -1011,7 +1011,7 @@ mod tests {
     #[tokio::test]
     async fn event_stream_preserves_topic_and_raw_data() {
         use crate::EventstreamRequestQueryTopic;
-        use futures::StreamExt;
+        use tokio_stream::StreamExt;
         use wiremock::{
             Mock, MockServer, ResponseTemplate,
             matchers::{method, path},
@@ -1036,7 +1036,7 @@ mod tests {
             ])
             .await
             .expect("open stream");
-        futures::pin_mut!(stream);
+        let mut stream = std::pin::pin!(stream);
 
         let first = stream.next().await.expect("first event").expect("ok event");
         assert_eq!(first.topic, "head");

crates/p2p/src/bandwidth.rs

@@ -244,6 +244,8 @@ impl<S: AsyncWrite> AsyncWrite for PeerInstrumentedStream<S> {
 #[cfg(test)]
 #[allow(clippy::arithmetic_side_effects)]
 mod tests {
+    use std::task::Waker;
+
     use super::*;
 
     struct MockStream {
@@ -315,7 +317,7 @@ mod tests {
         let initial = received.get();
 
         let mut buf = [0u8; 3];
-        let mut cx = Context::from_waker(futures::task::noop_waker_ref());
+        let mut cx = Context::from_waker(Waker::noop());
         let _ = Pin::new(&mut stream).poll_read(&mut cx, &mut buf);
 
         assert_eq!(received.get(), initial + 3);
@@ -327,7 +329,7 @@ mod tests {
         let initial = sent.get();
 
         let data = b"hello";
-        let mut cx = Context::from_waker(futures::task::noop_waker_ref());
+        let mut cx = Context::from_waker(Waker::noop());
         let _ = Pin::new(&mut stream).poll_write(&mut cx, data);
 
         assert_eq!(sent.get(), initial + 5);
@@ -345,7 +347,7 @@ mod tests {
         let initial_recv = received.get();
         let initial_sent = sent.get();
 
-        let mut cx = Context::from_waker(futures::task::noop_waker_ref());
+        let mut cx = Context::from_waker(Waker::noop());
 
         let mut buf = [0u8; 3];
         let _ = Pin::new(&mut stream2).poll_read(&mut cx, &mut buf);

crates/p2p/src/gater.rs

@@ -214,6 +214,8 @@ impl std::error::Error for PeerNotAllowed {}
 
 #[cfg(test)]
 mod tests {
+    use std::task::Waker;
+
     use libp2p::core::{Endpoint, transport::PortUse};
 
     use super::*;
@@ -258,8 +260,7 @@ mod tests {
     /// Drains a single event from `poll`, mirroring how the swarm would
     /// consume generated events.
     fn poll_event(gater: &mut ConnGater) -> Option<Event> {
-        let waker = futures::task::noop_waker_ref();
-        let mut cx = Context::from_waker(waker);
+        let mut cx = Context::from_waker(Waker::noop());
         match gater.poll(&mut cx) {
             Poll::Ready(ToSwarm::GenerateEvent(event)) => Some(event),
             _ => None,

crates/p2p/src/relay/manager/tests.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashSet, str::FromStr};
+use std::{collections::HashSet, str::FromStr, task::Waker};
 
 use super::*;
 use crate::relay::dial::RelayDialState;
@@ -843,8 +843,7 @@ async fn poll_fires_swept_peer_dial_within_the_same_watchdog_pass() {
     // waits a full extra watchdog tick.
     let target = PeerId::random();
     let mut mgr = manager_with_reserved_relay(vec![target]);
-    let waker = futures::task::noop_waker();
-    let mut cx = Context::from_waker(&waker);
+    let mut cx = Context::from_waker(Waker::noop());
 
     // Drain until Pending: initialises the watchdog.
     while mgr.poll(&mut cx).is_ready() {}

crates/peerinfo/src/handler.rs

@@ -294,8 +294,7 @@ async fn send_peer_info(
     request: PeerInfo,
     timeout: Duration,
 ) -> Result<(Stream, PeerInfo), Failure> {
-    let send = protocol.send_peer_info(stream, &request);
-    futures::pin_mut!(send);
+    let send = std::pin::pin!(protocol.send_peer_info(stream, &request));
 
     match future::select(send, Delay::new(timeout)).await {
         future::Either::Left((Ok((stream, response)), _)) => Ok((stream, response)),