1use std::sync::Arc;
6
7use futures::FutureExt;
8use tokio::{sync::mpsc, task::JoinSet};
9use tower::{buffer::Buffer, util::BoxCloneService, Service, ServiceExt};
10use tracing::{instrument, Instrument, Span};
11
12use cuprate_async_buffer::BufferStream;
13use cuprate_p2p_core::{
14 client::Connector,
15 services::{AddressBookRequest, AddressBookResponse},
16 CoreSyncSvc, NetworkZone, ProtocolRequestHandlerMaker,
17};
18
19pub mod block_downloader;
20mod broadcast;
21pub mod config;
22pub mod connection_maintainer;
23pub mod constants;
24mod inbound_server;
25mod peer_set;
26
27use block_downloader::{BlockBatch, BlockDownloaderConfig, ChainSvcRequest, ChainSvcResponse};
28pub use broadcast::{BroadcastRequest, BroadcastSvc};
29pub use config::{AddressBookConfig, P2PConfig};
30use connection_maintainer::MakeConnectionRequest;
31use peer_set::PeerSet;
32pub use peer_set::{ClientDropGuard, PeerSetRequest, PeerSetResponse};
33
34#[instrument(level = "debug", name = "net", skip_all, fields(zone = N::NAME))]
43pub async fn initialize_network<N, PR, CS>(
44 protocol_request_handler_maker: PR,
45 core_sync_svc: CS,
46 config: P2PConfig<N>,
47) -> Result<NetworkInterface<N>, tower::BoxError>
48where
49 N: NetworkZone,
50 N::Addr: borsh::BorshDeserialize + borsh::BorshSerialize,
51 PR: ProtocolRequestHandlerMaker<N> + Clone,
52 CS: CoreSyncSvc + Clone,
53{
54 let address_book =
55 cuprate_address_book::init_address_book(config.address_book_config.clone()).await?;
56 let address_book = Buffer::new(
57 address_book,
58 config
59 .max_inbound_connections
60 .checked_add(config.outbound_connections)
61 .unwrap(),
62 );
63
64 let (broadcast_svc, outbound_mkr, inbound_mkr) =
67 broadcast::init_broadcast_channels(broadcast::BroadcastConfig::default());
68
69 let mut basic_node_data = config.basic_node_data();
70
71 if !N::CHECK_NODE_ID {
72 basic_node_data.peer_id = 1;
73 }
74
75 let outbound_handshaker_builder =
76 cuprate_p2p_core::client::HandshakerBuilder::new(basic_node_data)
77 .with_address_book(address_book.clone())
78 .with_core_sync_svc(core_sync_svc)
79 .with_protocol_request_handler_maker(protocol_request_handler_maker)
80 .with_broadcast_stream_maker(outbound_mkr)
81 .with_connection_parent_span(Span::current());
82
83 let inbound_handshaker = outbound_handshaker_builder
84 .clone()
85 .with_broadcast_stream_maker(inbound_mkr)
86 .build();
87
88 let outbound_handshaker = outbound_handshaker_builder.build();
89
90 let (new_connection_tx, new_connection_rx) = mpsc::channel(
91 config
92 .outbound_connections
93 .checked_add(config.max_inbound_connections)
94 .unwrap(),
95 );
96 let (make_connection_tx, make_connection_rx) = mpsc::channel(3);
97
98 let outbound_connector = Connector::new(outbound_handshaker);
99 let outbound_connection_maintainer = connection_maintainer::OutboundConnectionKeeper::new(
100 config.clone(),
101 new_connection_tx.clone(),
102 make_connection_rx,
103 address_book.clone(),
104 outbound_connector,
105 );
106
107 let peer_set = PeerSet::new(new_connection_rx);
108
109 let mut background_tasks = JoinSet::new();
110
111 background_tasks.spawn(
112 outbound_connection_maintainer
113 .run()
114 .instrument(Span::current()),
115 );
116 background_tasks.spawn(
117 inbound_server::inbound_server(
118 new_connection_tx,
119 inbound_handshaker,
120 address_book.clone(),
121 config,
122 )
123 .map(|res| {
124 if let Err(e) = res {
125 tracing::error!("Error in inbound connection listener: {e}");
126 }
127
128 tracing::info!("Inbound connection listener shutdown");
129 })
130 .instrument(Span::current()),
131 );
132
133 Ok(NetworkInterface {
134 peer_set: Buffer::new(peer_set, 10).boxed_clone(),
135 broadcast_svc,
136 make_connection_tx,
137 address_book: address_book.boxed_clone(),
138 _background_tasks: Arc::new(background_tasks),
139 })
140}
141
142#[derive(Clone)]
144pub struct NetworkInterface<N: NetworkZone> {
145 peer_set: BoxCloneService<PeerSetRequest, PeerSetResponse<N>, tower::BoxError>,
147 broadcast_svc: BroadcastSvc<N>,
149 #[expect(dead_code, reason = "will be used eventually")]
151 make_connection_tx: mpsc::Sender<MakeConnectionRequest>,
152 address_book: BoxCloneService<AddressBookRequest<N>, AddressBookResponse<N>, tower::BoxError>,
154 _background_tasks: Arc<JoinSet<()>>,
156}
157
158impl<N: NetworkZone> NetworkInterface<N> {
159 pub fn broadcast_svc(&self) -> BroadcastSvc<N> {
161 self.broadcast_svc.clone()
162 }
163
164 pub fn block_downloader<C>(
166 &self,
167 our_chain_service: C,
168 config: BlockDownloaderConfig,
169 ) -> BufferStream<BlockBatch>
170 where
171 C: Service<ChainSvcRequest<N>, Response = ChainSvcResponse<N>, Error = tower::BoxError>
172 + Send
173 + 'static,
174 C::Future: Send + 'static,
175 {
176 block_downloader::download_blocks(self.peer_set.clone(), our_chain_service, config)
177 }
178
179 pub fn address_book(
181 &self,
182 ) -> BoxCloneService<AddressBookRequest<N>, AddressBookResponse<N>, tower::BoxError> {
183 self.address_book.clone()
184 }
185
186 pub fn peer_set(
188 &mut self,
189 ) -> &mut BoxCloneService<PeerSetRequest, PeerSetResponse<N>, tower::BoxError> {
190 &mut self.peer_set
191 }
192}