Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
29 kB · 872 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873uniffi::setup_scaffolding!();
pub mod c2pa;pub mod error;pub mod node_addr;pub mod public_key;pub mod streams;
use std::sync::{LazyLock, Once};
mod db;pub use db::*;#[cfg(test)]mod tests;
#[cfg(test)]mod test_stream;
/// Lazily initialized Tokio runtime for use in uniffi methods that need a runtime.static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| tokio::runtime::Runtime::new().unwrap());
/// Ensure logging is only initialized oncestatic LOGGING_INIT: Once = Once::new();
use std::{ collections::{BTreeMap, BTreeSet, HashSet}, str::FromStr, sync::Arc,};
use bytes::Bytes;use iroh::{NodeId, PublicKey, RelayMode, SecretKey, discovery::static_provider::StaticProvider};use iroh_base::ticket::NodeTicket;use iroh_gossip::{net::Gossip, proto::TopicId};use irpc::{WithChannels, rpc::RemoteService};use irpc_iroh::{IrohProtocol, IrohRemoteConnection};use n0_future::future::Boxed;
mod rpc { //! Protocol API use bytes::Bytes; use iroh::NodeId; use irpc::{channel::oneshot, rpc_requests}; use serde::{Deserialize, Serialize};
pub const ALPN: &[u8] = b"/iroh/streamplace/1";
/// Subscribe to the given `key` #[derive(Debug, Serialize, Deserialize)] pub struct Subscribe { pub key: String, // TODO: verify pub remote_id: NodeId, }
/// Unsubscribe from the given `key` #[derive(Debug, Serialize, Deserialize)] pub struct Unsubscribe { pub key: String, // TODO: verify pub remote_id: NodeId, }
// #[derive(Debug, Serialize, Deserialize)] // pub struct SendSegment { // pub key: String, // pub data: Bytes, // }
#[derive(Debug, Clone, Serialize, Deserialize)] pub struct RecvSegment { pub from: NodeId, pub key: String, pub data: Bytes, }
// Use the macro to generate both the Protocol and Message enums // plus implement Channels for each type #[rpc_requests(message = Message)] #[derive(Serialize, Deserialize, Debug)] pub enum Protocol { #[rpc(tx=oneshot::Sender<()>)] Subscribe(Subscribe), #[rpc(tx=oneshot::Sender<()>)] Unsubscribe(Unsubscribe), #[rpc(tx=oneshot::Sender<()>)] RecvSegment(RecvSegment), }}
mod api { //! Protocol API use bytes::Bytes; use iroh::{NodeAddr, NodeId}; use irpc::{channel::oneshot, rpc_requests}; use serde::{Deserialize, Serialize};
/// Subscribe to the given `key` #[derive(Debug, Serialize, Deserialize)] pub struct Subscribe { pub key: String, // TODO: verify pub remote_id: NodeId, }
/// Unsubscribe from the given `key` #[derive(Debug, Serialize, Deserialize)] pub struct Unsubscribe { pub key: String, // TODO: verify pub remote_id: NodeId, }
#[derive(Debug, Serialize, Deserialize)] pub struct SendSegment { pub key: String, pub data: Bytes, }
#[derive(Debug, Serialize, Deserialize)] pub struct JoinPeers { pub peers: Vec<NodeAddr>, }
#[derive(Debug, Serialize, Deserialize)] pub struct AddTickets { pub peers: Vec<NodeAddr>, }
#[derive(Debug, Serialize, Deserialize)] pub struct GetNodeAddr;
#[derive(Debug, Serialize, Deserialize)] pub struct Shutdown;
// Use the macro to generate both the Protocol and Message enums // plus implement Channels for each type #[rpc_requests(message = Message)] #[derive(Serialize, Deserialize, Debug)] pub enum Protocol { #[rpc(tx=oneshot::Sender<()>)] Subscribe(Subscribe), #[rpc(tx=oneshot::Sender<()>)] Unsubscribe(Unsubscribe), #[rpc(tx=oneshot::Sender<()>)] SendSegment(SendSegment), #[rpc(tx=oneshot::Sender<()>)] JoinPeers(JoinPeers), #[rpc(tx=oneshot::Sender<()>)] AddTickets(AddTickets), #[rpc(tx=oneshot::Sender<NodeAddr>)] GetNodeAddr(GetNodeAddr), #[rpc(tx=oneshot::Sender<()>)] Shutdown(Shutdown), }}use api::{Message as ApiMessage, Protocol as ApiProtocol};use n0_future::{FuturesUnordered, StreamExt};use rpc::{Message as RpcMessage, Protocol as RpcProtocol};use snafu::Snafu;use tracing::{Instrument, debug, error, trace, trace_span, warn};
use crate::rpc::RecvSegment;
/// Initialize logging with the default subscriber that respects RUST_LOG environment variable./// This function is safe to call multiple times - it will only initialize logging once.#[uniffi::export]pub fn init_logging() { LOGGING_INIT.call_once(|| { tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .init(); });}
/// Initialize logging with a custom log level./// This function is safe to call multiple times - it will only initialize logging once.////// # Arguments/// * `level` - Log level as a string (e.g., "trace", "debug", "info", "warn", "error")#[uniffi::export]pub fn init_logging_with_level(level: String) { LOGGING_INIT.call_once(|| { let filter = tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(level));
tracing_subscriber::fmt().with_env_filter(filter).init(); });}
pub(crate) enum HandlerMode { Sender, Forwarder, Receiver(Arc<dyn DataHandler>),}
impl HandlerMode { pub fn mode_str(&self) -> &'static str { match self { HandlerMode::Sender => "sender", HandlerMode::Forwarder => "forwarder", HandlerMode::Receiver(_) => "receiver", } }}
type Tasks = FuturesUnordered<Boxed<(NodeId, Result<(), RpcTaskError>)>>;
/// Actor that contains both a kv db for metadata and a handler for the rpc protocol.////// This can be used both for sender and receiver nodes. Sender nodes will just set the/// handler to None.struct Actor { /// Receiver for rpc messages from remote nodes rpc_rx: tokio::sync::mpsc::Receiver<RpcMessage>, /// Receiver for API messages from the user api_rx: tokio::sync::mpsc::Receiver<ApiMessage>, /// nodes I need to send to for each stream subscribers: BTreeMap<String, BTreeSet<NodeId>>, /// nodes I am subscribed to subscriptions: BTreeMap<String, NodeId>, /// lightweight typed connection pool connections: ConnectionPool, /// How to handle incoming data handler: HandlerMode, /// Static provider for node discovery sp: StaticProvider, /// Iroh protocol router, I need to keep it around to keep the protocol alive router: iroh::protocol::Router, /// Metadata db client: db::Db, /// Write scope for this node for the metadata db write: db::WriteScope, /// Ongoing tasks tasks: Tasks, /// Configuration, needed for timeouts etc. config: Arc<Config>,}
#[derive(Debug, Clone)]struct Connection { id: NodeId, rpc: irpc::Client<RpcProtocol>,}
#[derive(Debug, Snafu)]enum RpcTaskError { #[snafu(transparent)] Task { source: irpc::Error }, #[snafu(transparent)] Timeout { source: tokio::time::error::Elapsed },}
struct ConnectionPool { endpoint: iroh::Endpoint, connections: BTreeMap<NodeId, Connection>,}
impl ConnectionPool { fn new(endpoint: iroh::Endpoint) -> Self { Self { endpoint, connections: BTreeMap::new(), } }
/// Cheap conn pool hack fn get(&mut self, remote: &NodeId) -> Connection { if !self.connections.contains_key(remote) { trace!(remote = %remote.fmt_short(),"ConnectionPool.get is inserting a new remote connection"); let conn = IrohRemoteConnection::new( self.endpoint.clone(), (*remote).into(), rpc::ALPN.to_vec(), ); let conn = Connection { rpc: irpc::Client::boxed(conn), id: *remote, }; self.connections.insert(*remote, conn); } self.connections .get_mut(remote) .expect("just inserted") .clone() }
fn remove(&mut self, remote: &NodeId) { self.connections.remove(remote); }}
impl Actor { pub async fn spawn( endpoint: iroh::Endpoint, sp: StaticProvider, topic: iroh_gossip::proto::TopicId, config: Config, handler: HandlerMode, ) -> Result<(Node, impl Future<Output = ()>), iroh_gossip::api::ApiError> { let (rpc_tx, rpc_rx) = tokio::sync::mpsc::channel::<RpcMessage>(512); let (api_tx, api_rx) = tokio::sync::mpsc::channel::<ApiMessage>(512); let gossip = Gossip::builder().spawn(endpoint.clone()); let id = endpoint.node_id(); let router = iroh::protocol::Router::builder(endpoint.clone()) .accept(iroh_gossip::ALPN, gossip.clone()) .accept( rpc::ALPN, IrohProtocol::new(rpc::Protocol::remote_handler(rpc_tx.into())), ) .spawn(); let topic = gossip.subscribe(topic, vec![]).await?; let secret = router.endpoint().secret_key().clone(); let db_config = Default::default(); let client = iroh_smol_kv::Client::local(topic, db_config); let write = db::WriteScope::new(client.write(secret.clone())); let client = db::Db::new(client); let actor = Self { rpc_rx, api_rx, subscribers: BTreeMap::new(), subscriptions: BTreeMap::new(), connections: ConnectionPool::new(router.endpoint().clone()), handler, router, sp, write: write.clone(), client: client.clone(), tasks: FuturesUnordered::new(), config: Arc::new(config), }; let api = Node { client: Arc::new(client), write: Arc::new(write), api: irpc::Client::local(api_tx), }; Ok(( api, actor .run() .instrument(trace_span!("actor", id=%id.fmt_short())), )) }
async fn run(mut self) { loop { tokio::select! { msg = self.rpc_rx.recv() => { trace!("received local rpc message"); let Some(msg) = msg else { error!("rpc channel closed"); break; }; self.handle_rpc(msg).instrument(trace_span!("rpc")).await; } msg = self.api_rx.recv() => { trace!("received remote rpc message"); let Some(msg) = msg else { break; }; if let Some(shutdown) = self.handle_api(msg).instrument(trace_span!("api")).await { shutdown.send(()).await.ok(); break; } } res = self.tasks.next(), if !self.tasks.is_empty() => { trace!("processing task"); let Some((remote_id, res)) = res else { error!("task finished but no result"); break; }; match res { Ok(()) => {} Err(RpcTaskError::Timeout { source }) => { warn!("call to {remote_id} timed out: {source}"); } Err(RpcTaskError::Task { source }) => { warn!("call to {remote_id} failed: {source}"); } } self.connections.remove(&remote_id); } } } warn!("RPC Actor loop has closed"); }
async fn update_subscriber_meta(&mut self, key: &str) { let n = self .subscribers .get(key) .map(|s| s.len()) .unwrap_or_default(); let v = n.to_string().into_bytes(); self.write .put_impl(Some(key.as_bytes().to_vec()), b"subscribers", v.into()) .await .ok(); }
/// Requests from remote nodes async fn handle_rpc(&mut self, msg: RpcMessage) { trace!("RPC.handle_rpc"); match msg { RpcMessage::Subscribe(msg) => { trace!(inner = ?msg.inner, "RpcMessage::Subscribe"); let WithChannels { tx, inner: rpc::Subscribe { key, remote_id }, .. } = msg; self.subscribers .entry(key.clone()) .or_default() .insert(remote_id); self.update_subscriber_meta(&key).await; tx.send(()).await.ok(); } RpcMessage::Unsubscribe(msg) => { debug!(inner = ?msg.inner, "RpcMessage::Unsubscribe"); let WithChannels { tx, inner: rpc::Unsubscribe { key, remote_id }, .. } = msg; if let Some(e) = self.subscribers.get_mut(&key) && !e.remove(&remote_id) { warn!( "unsubscribe: no subscription for {} from {}", key, remote_id ); } if let Some(subscriptions) = self.subscribers.get(&key) && subscriptions.is_empty() { self.subscribers.remove(&key); } self.update_subscriber_meta(&key).await; tx.send(()).await.ok(); } RpcMessage::RecvSegment(msg) => { trace!(inner = ?msg.inner, "RpcMessage::RecvSegment"); let WithChannels { tx, inner: rpc::RecvSegment { key, from, data }, .. } = msg; match &self.handler { HandlerMode::Sender => { warn!("received segment but in sender mode"); } HandlerMode::Forwarder => { trace!("forwarding segment"); if let Some(remotes) = self.subscribers.get(&key) { Self::handle_send( &mut self.tasks, &mut self.connections, &self.config, key, data, remotes, ); } else { trace!("no subscribers for stream {}", key); } } HandlerMode::Receiver(handler) => { if self.subscriptions.contains_key(&key) { let from = Arc::new(from.into()); handler.handle_data(from, key, data.to_vec()).await; } else { warn!("received segment for unsubscribed key: {}", key); } } }; tx.send(()).await.ok(); } } }
async fn handle_api(&mut self, msg: ApiMessage) -> Option<irpc::channel::oneshot::Sender<()>> { trace!("RPC.handle_api"); match msg { ApiMessage::SendSegment(msg) => { trace!(inner = ?msg.inner, "ApiMessage::SendSegment"); let WithChannels { tx, inner: api::SendSegment { key, data }, .. } = msg; if let Some(remotes) = self.subscribers.get(&key) { Self::handle_send( &mut self.tasks, &mut self.connections, &self.config, key, data, remotes, ); } else { trace!("no subscribers for stream {}", key); } tx.send(()).await.ok(); } ApiMessage::Subscribe(msg) => { trace!(inner = ?msg.inner, "ApiMessage::Subscribe"); let WithChannels { tx, inner: api::Subscribe { key, remote_id }, .. } = msg; let conn = self.connections.get(&remote_id); tx.send(()).await.ok(); trace!(remote = %remote_id.fmt_short(), key = %key, "send rpc::Subscribe message"); conn.rpc .rpc(rpc::Subscribe { key: key.clone(), remote_id: self.node_id(), }) .await .ok(); trace!(remote = %remote_id.fmt_short(), key = %key, "inserting subscription"); self.subscriptions.insert(key, remote_id); trace!("finished inserting subscription"); } ApiMessage::Unsubscribe(msg) => { trace!(inner = ?msg.inner, "ApiMessage::Unsubscribe"); let WithChannels { tx, inner: api::Unsubscribe { key, remote_id }, .. } = msg; let conn = self.connections.get(&remote_id); tx.send(()).await.ok(); conn.rpc .rpc(rpc::Unsubscribe { key: key.clone(), remote_id: self.node_id(), }) .await .ok(); self.subscriptions.remove(&key); } ApiMessage::AddTickets(msg) => { trace!(inner = ?msg.inner, "ApiMessage::AddTickets"); let WithChannels { tx, inner: api::AddTickets { peers }, .. } = msg; for addr in &peers { self.sp.add_node_info(addr.clone()); } // self.client.inner().join_peers(ids).await.ok(); tx.send(()).await.ok(); } ApiMessage::JoinPeers(msg) => { trace!(inner = ?msg.inner, "ApiMessage::JoinPeers"); let WithChannels { tx, inner: api::JoinPeers { peers }, .. } = msg; let ids = peers .iter() .map(|a| a.node_id) .filter(|id| *id != self.node_id()) .collect::<HashSet<_>>(); for addr in &peers { self.sp.add_node_info(addr.clone()); } self.client.inner().join_peers(ids).await.ok(); tx.send(()).await.ok(); } ApiMessage::GetNodeAddr(msg) => { trace!(inner = ?msg.inner, "ApiMessage::GetNodeAddr"); let WithChannels { tx, .. } = msg; if !self.config.disable_relay { // don't await home relay if we have disabled relays, this will hang forever self.router.endpoint().online().await; } let addr = self.router.endpoint().node_addr(); tx.send(addr).await.ok(); } ApiMessage::Shutdown(msg) => { trace!(inner = ?msg.inner, "ApiMessage::Shutdown"); return Some(msg.tx); } } None }
fn handle_send( tasks: &mut Tasks, connections: &mut ConnectionPool, config: &Arc<Config>, key: String, data: Bytes, remotes: &BTreeSet<NodeId>, ) { let me = connections.endpoint.node_id(); let msg = rpc::RecvSegment { key, data, from: me, }; for remote in remotes { trace!(remote = %remote.fmt_short(), key = %msg.key, "handle_send to remote"); let conn = connections.get(remote); tasks.push(Box::pin(Self::forward_task( config.clone(), conn, msg.clone(), ))); } }
async fn forward_task( config: Arc<Config>, conn: Connection, msg: RecvSegment, ) -> (NodeId, Result<(), RpcTaskError>) { let id = conn.id; let res = async move { tokio::time::timeout(config.max_send_duration, conn.rpc.rpc(msg)).await??; Ok(()) } .await; (id, res) }
fn node_id(&self) -> PublicKey { self.router.endpoint().node_id() }}
/// Iroh-streamplace node that can send, forward or receive stream segments.#[derive(Clone, uniffi::Object)]pub struct Node { client: Arc<db::Db>, write: Arc<db::WriteScope>, api: irpc::Client<ApiProtocol>,}
impl Node { pub(crate) async fn new_in_runtime( config: Config, handler: HandlerMode, ) -> Result<Arc<Self>, CreateError> { let mode_str = Bytes::from(handler.mode_str()); let secret_key = SecretKey::from_bytes(&<[u8; 32]>::try_from(config.key.clone()).map_err(|e| { CreateError::PrivateKey { size: e.len() as u64, } })?); let topic = TopicId::from_bytes(<[u8; 32]>::try_from(config.topic.clone()).map_err(|e| { CreateError::Topic { size: e.len() as u64, } })?); let relay_mode = if config.disable_relay { RelayMode::Disabled } else { RelayMode::Default }; let sp = StaticProvider::new(); let endpoint = iroh::Endpoint::builder() .secret_key(secret_key) .relay_mode(relay_mode) .discovery(sp.clone()) .bind() .await .map_err(|e| CreateError::Bind { message: e.to_string(), })?; let (api, actor) = Actor::spawn(endpoint, sp, topic, config, handler) .await .map_err(|e| CreateError::Subscribe { message: e.to_string(), })?; api.node_scope() .put_impl(Option::<Vec<u8>>::None, b"mode", mode_str) .await .ok(); tokio::spawn(actor); Ok(Arc::new(api)) }}
/// DataHandler trait that is exported to go for receiving data callbacks.#[uniffi::export(with_foreign)]#[async_trait::async_trait]pub trait DataHandler: Send + Sync { async fn handle_data( &self, from: Arc<crate::public_key::PublicKey>, topic: String, data: Vec<u8>, );}
#[uniffi::export]impl Node { /// Create a new streamplace client node. #[uniffi::constructor] pub async fn sender(config: Config) -> Result<Arc<Self>, CreateError> { RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Sender)) }
#[uniffi::constructor] pub async fn forwarder(config: Config) -> Result<Arc<Self>, CreateError> { RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Forwarder)) }
#[uniffi::constructor] pub async fn receiver( config: Config, handler: Arc<dyn DataHandler>, ) -> Result<Arc<Self>, CreateError> { RUNTIME.block_on(Self::new_in_runtime(config, HandlerMode::Receiver(handler))) }
/// Get a handle to the db to watch for changes locally or globally. pub fn db(&self) -> Arc<db::Db> { self.client.clone() }
/// Get a handle to the write scope for this node. /// /// This is equivalent to calling `db.write(...)` with the secret key used to create the node. pub fn node_scope(&self) -> Arc<db::WriteScope> { self.write.clone() }
/// Subscribe to updates for a given stream from a remote node. pub async fn subscribe( &self, key: String, remote_id: Arc<crate::public_key::PublicKey>, ) -> Result<(), PutError> { self.api .rpc(api::Subscribe { key, remote_id: remote_id.as_ref().into(), }) .await .map_err(|e| PutError::Irpc { message: e.to_string(), }) }
/// Unsubscribe from updates for a given stream from a remote node. pub async fn unsubscribe( &self, key: String, remote_id: Arc<crate::public_key::PublicKey>, ) -> Result<(), PutError> { self.api .rpc(api::Unsubscribe { key, remote_id: remote_id.as_ref().into(), }) .await .map_err(|e| PutError::Irpc { message: e.to_string(), }) }
/// Send a segment to all subscribers of the given stream. pub async fn send_segment(&self, key: String, data: Vec<u8>) -> Result<(), PutError> { debug!(key = &key, data_len = data.len(), "Node.send_segment"); self.api .rpc(api::SendSegment { key, data: data.into(), }) .await .map_err(|e| PutError::Irpc { message: e.to_string(), }) }
/// Join peers by their node tickets. pub async fn join_peers(&self, peers: Vec<String>) -> Result<(), JoinPeersError> { let peers = peers .iter() .map(|p| NodeTicket::from_str(p)) .collect::<Result<Vec<_>, _>>() .map_err(|e| JoinPeersError::Ticket { message: e.to_string(), })?; let addrs = peers .iter() .map(|t| t.node_addr().clone()) .collect::<Vec<_>>(); self.api .rpc(api::JoinPeers { peers: addrs }) .await .map_err(|e| JoinPeersError::Irpc { message: e.to_string(), }) }
/// Add tickets for remote peers pub async fn add_tickets(&self, peers: Vec<String>) -> Result<(), JoinPeersError> { let peers = peers .iter() .map(|p| NodeTicket::from_str(p)) .collect::<Result<Vec<_>, _>>() .map_err(|e| JoinPeersError::Ticket { message: e.to_string(), })?; let addrs = peers .iter() .map(|t| t.node_addr().clone()) .collect::<Vec<_>>(); self.api .rpc(api::AddTickets { peers: addrs }) .await .map_err(|e| JoinPeersError::Irpc { message: e.to_string(), }) }
/// Get this node's ticket. pub async fn ticket(&self) -> Result<String, PutError> { let addr = self .api .rpc(api::GetNodeAddr) .await .map_err(|e| PutError::Irpc { message: e.to_string(), })?; Ok(NodeTicket::from(addr).to_string()) }
/// Get this node's node ID. pub async fn node_id(&self) -> Result<Arc<crate::public_key::PublicKey>, PutError> { let addr = self .api .rpc(api::GetNodeAddr) .await .map_err(|e| PutError::Irpc { message: e.to_string(), })?; Ok(Arc::new(addr.node_id.into())) }
/// Shutdown the node, including the streaming system and the metadata db. pub async fn shutdown(&self) -> Result<(), ShutdownError> { // shut down both the streams and the db concurrently, even if one fails let (res1, res2) = tokio::join!(self.shutdown_streams(), self.client.shutdown()); res1?; res2?; Ok(()) }}
impl Node { async fn shutdown_streams(&self) -> std::result::Result<(), ShutdownError> { self.api .rpc(api::Shutdown) .await .map_err(|e| ShutdownError::Irpc { message: e.to_string(), }) }}