refactor(workspace): standardise on async_trait, drop hand-rolled BoxFuture

crates/cli/Cargo.toml

@@ -11,6 +11,7 @@ name = "pluto"
 path = "src/main.rs"
 
 [dependencies]
+async-trait.workspace = true
 clap.workspace = true
 thiserror.workspace = true
 k256.workspace = true

crates/cli/src/commands/test/infra.rs

@@ -6,6 +6,7 @@ use std::{
     time::{Duration, Instant},
 };
 
+use async_trait::async_trait;
 use clap::Args;
 use serde::Deserialize;
 use tokio_util::sync::CancellationToken;
@@ -61,11 +62,7 @@ struct FioResultSingle {
     bw: f64,
 }
 
-// `expect` cannot be used here: the lint does not fire under the current
-// toolchain, so it would be flagged as an unfulfilled expectation. Kept as
-// `allow` for forward compatibility. Reason: internal trait not part of the
-// public API; no need for the `Send` bound the lint guards against.
-#[allow(async_fn_in_trait)]
+#[async_trait]
 trait DiskTestTool {
     async fn check_availability(&self) -> Result<()>;
     async fn write_speed(&self, path: &Path, block_size_kb: i32) -> Result<f64>;
@@ -76,6 +73,7 @@ trait DiskTestTool {
 
 struct FioTestTool;
 
+#[async_trait]
 impl DiskTestTool for FioTestTool {
     async fn check_availability(&self) -> Result<()> {
         let result = tokio::process::Command::new("fio")

crates/consensus/Cargo.toml

@@ -7,6 +7,7 @@ license.workspace = true
 publish.workspace = true
 
 [dependencies]
+async-trait.workspace = true
 axum.workspace = true
 cancellation.workspace = true
 chrono.workspace = true

crates/consensus/src/qbft/component.rs

@@ -6,6 +6,7 @@ use std::{
     sync::{Arc, Mutex, PoisonError},
 };
 
+use async_trait::async_trait;
 use futures::future::BoxFuture;
 use k256::{PublicKey, SecretKey};
 use prost::{Message, Name};
@@ -599,6 +600,7 @@ impl Consensus {
     }
 }
 
+#[async_trait]
 impl crate::wrapper::Consensus for Consensus {
     fn protocol_id(&self) -> String {
         self.protocol_id().to_string()
@@ -608,29 +610,21 @@ impl crate::wrapper::Consensus for Consensus {
         drop(Consensus::start(self, ct));
     }
 
-    fn participate(
-        &self,
-        ct: CancellationToken,
-        duty: Duty,
-    ) -> BoxFuture<'_, crate::wrapper::Result<()>> {
-        Box::pin(async move {
-            Consensus::participate(self, duty, &ct)
-                .await
-                .map_err(Into::into)
-        })
+    async fn participate(&self, ct: CancellationToken, duty: Duty) -> crate::wrapper::Result<()> {
+        Consensus::participate(self, duty, &ct)
+            .await
+            .map_err(Into::into)
     }
 
-    fn propose(
+    async fn propose(
         &self,
         ct: CancellationToken,
         duty: Duty,
         value: pbcore::UnsignedDataSet,
-    ) -> BoxFuture<'_, crate::wrapper::Result<()>> {
-        Box::pin(async move {
-            Consensus::propose(self, duty, value, &ct)
-                .await
-                .map_err(Into::into)
-        })
+    ) -> crate::wrapper::Result<()> {
+        Consensus::propose(self, duty, value, &ct)
+            .await
+            .map_err(Into::into)
     }
 
     fn subscribe(&self, subscriber: crate::wrapper::Subscriber) {

crates/consensus/src/wrapper.rs

@@ -5,7 +5,7 @@ use std::{
     sync::{Arc, PoisonError, RwLock},
 };
 
-use futures::future::BoxFuture;
+use async_trait::async_trait;
 use pluto_core::{corepb::v1::core as pbcore, types::Duty};
 use tokio_util::sync::CancellationToken;
 
@@ -30,6 +30,7 @@ pub type Subscriber =
     Box<dyn Fn(Duty, pbcore::UnsignedDataSet) -> SubscriberResult + Send + Sync + 'static>;
 
 /// Consensus implementation interface.
+#[async_trait]
 pub trait Consensus: Send + Sync {
     /// Returns the consensus protocol ID.
     fn protocol_id(&self) -> String;
@@ -38,15 +39,15 @@ pub trait Consensus: Send + Sync {
     fn start(&self, ct: CancellationToken);
 
     /// Starts participating in a consensus instance.
-    fn participate(&self, ct: CancellationToken, duty: Duty) -> BoxFuture<'_, Result<()>>;
+    async fn participate(&self, ct: CancellationToken, duty: Duty) -> Result<()>;
 
     /// Proposes unsigned duty data for a consensus instance.
-    fn propose(
+    async fn propose(
         &self,
         ct: CancellationToken,
         duty: Duty,
         value: pbcore::UnsignedDataSet,
-    ) -> BoxFuture<'_, Result<()>>;
+    ) -> Result<()>;
 
     /// Registers a callback for decided unsigned duty data.
     fn subscribe(&self, subscriber: Subscriber);
@@ -115,7 +116,6 @@ impl ConsensusWrapper {
 mod tests {
     use std::sync::Mutex;
 
-    use futures::FutureExt as _;
     use pluto_core::{
         corepb::v1::core as pbcore,
         types::{Duty, SlotNumber},
@@ -202,6 +202,7 @@ mod tests {
         }
     }
 
+    #[async_trait]
     impl Consensus for TestConsensus {
         fn protocol_id(&self) -> String {
             self.protocol_id.clone()
@@ -211,19 +212,19 @@ mod tests {
             self.record("start");
         }
 
-        fn participate(&self, _ct: CancellationToken, _duty: Duty) -> BoxFuture<'_, Result<()>> {
+        async fn participate(&self, _ct: CancellationToken, _duty: Duty) -> Result<()> {
             self.record("participate");
-            async { Ok(()) }.boxed()
+            Ok(())
         }
 
-        fn propose(
+        async fn propose(
             &self,
             _ct: CancellationToken,
             _duty: Duty,
             _value: pbcore::UnsignedDataSet,
-        ) -> BoxFuture<'_, Result<()>> {
+        ) -> Result<()> {
             self.record("propose");
-            async { Ok(()) }.boxed()
+            Ok(())
         }
 
         fn subscribe(&self, subscriber: Subscriber) {

crates/core/Cargo.toml

@@ -17,6 +17,7 @@ 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/aggsigdb/memory.rs

@@ -2,6 +2,7 @@ use crate::{
     aggsigdb::types::{AggSigDB, Error},
     deadline, types,
 };
+use async_trait::async_trait;
 use std::collections::{HashMap, hash_map::Entry};
 use tokio::sync;
 use tokio_util::sync::CancellationToken;
@@ -168,7 +169,7 @@ impl MemoryDBHandle {
     }
 }
 
-#[async_trait::async_trait]
+#[async_trait]
 impl AggSigDB for MemoryDBHandle {
     async fn store(&self, duty: types::Duty, set: types::SignedDataSet) -> Result<(), Error> {
         let (response_tx, response_rx) = sync::oneshot::channel();

crates/core/src/aggsigdb/types.rs

@@ -1,3 +1,5 @@
+use async_trait::async_trait;
+
 use crate::types;
 
 /// Errors for AggSigDB operations.
@@ -15,7 +17,7 @@ pub enum Error {
 }
 
 /// A persistent store for aggregated signed duty data.
-#[async_trait::async_trait]
+#[async_trait]
 pub trait AggSigDB {
     /// Stores aggregated signed duty data set.
     async fn store(&self, duty: types::Duty, data: types::SignedDataSet) -> Result<(), Error>;

crates/core/src/validatorapi/component.rs

@@ -4,10 +4,11 @@
 //! and public-share mappings needed to translate between distributed-validator
 //! root keys and this node's threshold-BLS share.
 
-use std::{any::Any, collections::HashMap, future::Future, pin::Pin, sync::Arc, time::Duration};
+use std::{any::Any, collections::HashMap, future::Future, sync::Arc, time::Duration};
 
 use async_trait::async_trait;
 use axum::http::StatusCode;
+use futures::future::BoxFuture;
 use pluto_eth2api::{
     EthBeaconNodeApiClient, GetAttesterDutiesRequest, GetAttesterDutiesResponse,
     GetProposerDutiesRequest, GetProposerDutiesResponse, GetStateValidatorsResponseResponse,
@@ -59,9 +60,6 @@ use crate::{
 /// Boxed error returned by registered callbacks.
 pub type CallbackError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
-/// Boxed async callback result.
-pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
-
 /// Subscriber callback for `Subscribe`. Receives the [`Duty`] and the
 /// [`ParSignedDataSet`] by reference; the registered wrapper clones the
 /// set exactly once before invoking the user closure so every subscriber

crates/dkg/src/bcast/component.rs

@@ -2,6 +2,7 @@
 
 use std::{collections::HashMap, sync::Arc};
 
+use async_trait::async_trait;
 use futures::future::BoxFuture;
 use libp2p::PeerId;
 use prost::{Message, Name};
@@ -37,20 +38,21 @@ pub(crate) struct BroadcastCommand {
 }
 
 /// Type-erased entry stored per registered message ID.
+#[async_trait]
 pub(crate) trait RegisteredMessage: Send + Sync {
     /// Validates the incoming wrapped protobuf message.
     fn check(&self, peer_id: PeerId, any: &Any) -> Result<()>;
 
     /// Dispatches the incoming wrapped protobuf message to the typed callback.
-    fn callback(&self, peer_id: PeerId, msg_id: String, any: Any)
-    -> BoxFuture<'static, Result<()>>;
+    async fn callback(&self, peer_id: PeerId, msg_id: String, any: Any) -> Result<()>;
 }
 
 struct TypedRegistration<M> {
     check: CheckFn<M>,
     callback: CallbackFn<M>,
 }
 
+#[async_trait]
 impl<M> RegisteredMessage for TypedRegistration<M>
 where
     M: Message + Name + Default + Clone + Send + Sync + 'static,
@@ -60,16 +62,9 @@ where
         (self.check)(peer_id, &message)
     }
 
-    fn callback(
-        &self,
-        peer_id: PeerId,
-        msg_id: String,
-        any: Any,
-    ) -> BoxFuture<'static, Result<()>> {
-        match any.to_msg::<M>() {
-            Ok(message) => (self.callback)(peer_id, msg_id, message),
-            Err(e) => Box::pin(async move { Err(e.into()) }),
-        }
+    async fn callback(&self, peer_id: PeerId, msg_id: String, any: Any) -> Result<()> {
+        let message = any.to_msg::<M>()?;
+        (self.callback)(peer_id, msg_id, message).await
     }
 }
 

crates/priority/src/component.rs

@@ -390,6 +390,7 @@ impl Component {
 
 #[cfg(test)]
 mod tests {
+    use async_trait::async_trait;
     use pluto_core::corepb::v1::{core::Duty, priority::PriorityScoredResult};
 
     use super::*;
@@ -601,7 +602,7 @@ mod tests {
     #[test]
     fn new_component_rejects_peer_absent_from_context() {
         struct NoopConsensus;
-        #[async_trait::async_trait]
+        #[async_trait]
         impl Consensus for NoopConsensus {
             async fn propose_priority(
                 &self,

crates/priority/src/p2p/mod.rs

@@ -66,26 +66,23 @@ impl Sender {
     /// Errors with [`Error::Shutdown`] if the behaviour has been dropped, and
     /// with [`Error::Transport`]/[`Error::Unsupported`] on dial or stream
     /// failure. The caller is responsible for applying an exchange timeout.
-    pub fn send_receive(
+    pub async fn send_receive(
         &self,
         peer: PeerId,
         request: PriorityMsg,
-    ) -> BoxFuture<'static, crate::Result<PriorityMsg>> {
-        let command_tx = self.command_tx.clone();
-        Box::pin(async move {
-            let (response_tx, response_rx) = oneshot::channel();
-            command_tx
-                .send(Command::SendReceive {
-                    peer,
-                    request: OutboundRequest {
-                        request,
-                        response: response_tx,
-                    },
-                })
-                .map_err(|_| Error::Shutdown)?;
-
-            response_rx.await.map_err(|_| Error::Shutdown)?
-        })
+    ) -> crate::Result<PriorityMsg> {
+        let (response_tx, response_rx) = oneshot::channel();
+        self.command_tx
+            .send(Command::SendReceive {
+                peer,
+                request: OutboundRequest {
+                    request,
+                    response: response_tx,
+                },
+            })
+            .map_err(|_| Error::Shutdown)?;
+
+        response_rx.await.map_err(|_| Error::Shutdown)?
     }
 }
 

crates/priority/src/prioritiser.rs

@@ -535,6 +535,7 @@ fn start_consensus(
 mod tests {
     use std::sync::Mutex as StdMutex;
 
+    use async_trait::async_trait;
     use chrono::{Duration as ChronoDuration, Utc};
     use pluto_core::{
         corepb::v1::priority::{PriorityResult, PriorityTopicProposal},
@@ -572,7 +573,7 @@ mod tests {
         proposed: Arc<StdMutex<Vec<(Duty, PriorityResult)>>>,
     }
 
-    #[async_trait::async_trait]
+    #[async_trait]
     impl Consensus for MockConsensus {
         async fn propose_priority(
             &self,