⎇ 53dbdf0f1deea10d6fc567ea2b43e1e75d6d60cd
crates/modules/semantic_dht/minga-p2p/src/network.rs
History
Who touched what
raw
chore: monorepo inicial con arje + minga + yahweh absorbidos
1//! Integración libp2p con behaviour compuesto: streams Minga +
2//! Kademlia DHT.
3//!
4//! - **TCP + Noise + Yamux**: transporte autenticado y multiplexado.
5//! - **`stream::Behaviour`**: streams bidireccionales para el
6//! protocolo `/minga/sync/1.0.0`.
7//! - **`kad::Behaviour<MemoryStore>`**: tabla de routing distribuida
8//! para descubrimiento. Cada nodo arranca en modo `Server` y
9//! responde a queries del DHT.
10//!
11//! El swarm corre en una task tokio dedicada que procesa comandos
12//! externos (Dial, Listen, AddDhtPeer, FindClosestPeers) y eventos
13//! del swarm (NewListenAddr para señalar address resuelto, eventos
14//! Kad para completar queries). Los métodos públicos solo envían
15//! comandos por canal.
16
17use std::collections::HashMap;
18use std::time::Duration;
19
20use futures::StreamExt;
21use libp2p::{
22 identify, identity, kad, noise,
23 swarm::{NetworkBehaviour, SwarmEvent},
24 tcp, yamux, Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder,
25};
26use libp2p_stream as stream;
27use tokio::sync::{mpsc, oneshot, Mutex};
28
29pub const SYNC_PROTOCOL: StreamProtocol = StreamProtocol::new("/minga/sync/1.0.0");
30const IDENTIFY_PROTOCOL: &str = "/minga/0.1.0";
31
32#[derive(NetworkBehaviour)]
33struct MingaBehaviour {
34 stream: stream::Behaviour,
35 kad: kad::Behaviour<kad::store::MemoryStore>,
36 identify: identify::Behaviour,
37}
38
39#[derive(Debug, thiserror::Error)]
40pub enum NodeError {
41 #[error("transport build failed: {0}")]
42 Build(String),
43}
44
45#[derive(Debug)]
46enum Command {
47 Dial(Multiaddr),
48 Listen(Multiaddr),
49 AddDhtPeer(PeerId, Multiaddr),
50 FindClosestPeers(PeerId, oneshot::Sender<Vec<DiscoveredPeer>>),
51 StartProviding(Vec<u8>),
52 GetProviders(Vec<u8>, oneshot::Sender<Vec<PeerId>>),
53}
54
55/// Peer descubierto vía DHT: identidad + direcciones conocidas.
56#[derive(Debug, Clone)]
57pub struct DiscoveredPeer {
58 pub peer_id: PeerId,
59 pub addrs: Vec<Multiaddr>,
60}
61
62pub struct LibP2pNode {
63 pub peer_id: PeerId,
64 cmd_tx: mpsc::UnboundedSender<Command>,
65 listen_rx: Mutex<mpsc::UnboundedReceiver<Multiaddr>>,
66 /// Control para abrir/aceptar streams.
67 pub control: stream::Control,
68}
69
70impl LibP2pNode {
71 pub fn new() -> Result<Self, NodeError> {
72 let id = identity::Keypair::generate_ed25519();
73 let peer_id = id.public().to_peer_id();
74
75 let mut swarm: Swarm<MingaBehaviour> = SwarmBuilder::with_existing_identity(id)
76 .with_tokio()
77 .with_tcp(
78 tcp::Config::default(),
79 noise::Config::new,
80 yamux::Config::default,
81 )
82 .map_err(|e| NodeError::Build(format!("{e}")))?
83 .with_behaviour(|key| {
84 let local = key.public().to_peer_id();
85 let mut kad =
86 kad::Behaviour::new(local, kad::store::MemoryStore::new(local));
87 // Modo Server: respondemos a queries del DHT. Por
88 // defecto kad arranca en Auto, que requiere detectar
89 // reachability. Para tests en localhost forzamos Server.
90 kad.set_mode(Some(kad::Mode::Server));
91 let identify = identify::Behaviour::new(
92 identify::Config::new(IDENTIFY_PROTOCOL.to_string(), key.public())
93 .with_agent_version(format!("minga/{}", env!("CARGO_PKG_VERSION"))),
94 );
95 MingaBehaviour {
96 stream: stream::Behaviour::new(),
97 kad,
98 identify,
99 }
100 })
101 .map_err(|e| NodeError::Build(format!("{e}")))?
102 .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(60)))
103 .build();
104
105 let control = swarm.behaviour().stream.new_control();
106
107 let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<Command>();
108 let (listen_tx, listen_rx) = mpsc::unbounded_channel::<Multiaddr>();
109
110 tokio::spawn(async move {
111 let mut pending_finds: HashMap<
112 kad::QueryId,
113 oneshot::Sender<Vec<DiscoveredPeer>>,
114 > = HashMap::new();
115 let mut pending_providers: HashMap<
116 kad::QueryId,
117 (Vec<PeerId>, oneshot::Sender<Vec<PeerId>>),
118 > = HashMap::new();
119
120 loop {
121 tokio::select! {
122 Some(cmd) = cmd_rx.recv() => {
123 match cmd {
124 Command::Dial(addr) => {
125 let _ = swarm.dial(addr);
126 }
127 Command::Listen(addr) => {
128 let _ = swarm.listen_on(addr);
129 }
130 Command::AddDhtPeer(peer, addr) => {
131 swarm.behaviour_mut().kad.add_address(&peer, addr);
132 }
133 Command::FindClosestPeers(target, tx) => {
134 let qid = swarm.behaviour_mut().kad.get_closest_peers(target);
135 pending_finds.insert(qid, tx);
136 }
137 Command::StartProviding(key) => {
138 // Best-effort: si falla (sin peers cercanos para
139 // replicar), seguirá viviendo en el local store
140 // y se servirá vía get_providers de quien
141 // tenga conexión con nosotros.
142 let _ = swarm.behaviour_mut().kad.start_providing(key.into());
143 }
144 Command::GetProviders(key, tx) => {
145 let qid = swarm.behaviour_mut().kad.get_providers(key.into());
146 pending_providers.insert(qid, (Vec::new(), tx));
147 }
148 }
149 }
150 event = swarm.select_next_some() => {
151 match event {
152 SwarmEvent::NewListenAddr { address, .. } => {
153 let _ = listen_tx.send(address);
154 }
155 // Identify nos dice las listen-addrs reales del
156 // peer. Las inyectamos a Kad para poblar el
157 // routing table sin necesidad de add_dht_peer
158 // manual — la propagación pasa a ser automática.
159 SwarmEvent::Behaviour(MingaBehaviourEvent::Identify(
160 identify::Event::Received { peer_id, info, .. }
161 )) => {
162 for addr in info.listen_addrs {
163 swarm.behaviour_mut().kad.add_address(&peer_id, addr);
164 }
165 }
166 SwarmEvent::Behaviour(MingaBehaviourEvent::Kad(
167 kad::Event::OutboundQueryProgressed { id, result, step, .. }
168 )) => {
169 match result {
170 kad::QueryResult::GetClosestPeers(Ok(ok)) if step.last => {
171 if let Some(tx) = pending_finds.remove(&id) {
172 let infos = ok.peers.into_iter()
173 .map(|p| DiscoveredPeer {
174 peer_id: p.peer_id,
175 addrs: p.addrs,
176 })
177 .collect();
178 let _ = tx.send(infos);
179 }
180 }
181 kad::QueryResult::GetClosestPeers(Err(_)) if step.last => {
182 if let Some(tx) = pending_finds.remove(&id) {
183 let _ = tx.send(Vec::new());
184 }
185 }
186 kad::QueryResult::GetProviders(Ok(ok)) => {
187 if let Some((collected, _)) =
188 pending_providers.get_mut(&id)
189 {
190 if let kad::GetProvidersOk::FoundProviders {
191 providers, ..
192 } = ok
193 {
194 for p in providers {
195 if !collected.contains(&p) {
196 collected.push(p);
197 }
198 }
199 }
200 }
201 if step.last {
202 if let Some((providers, tx)) =
203 pending_providers.remove(&id)
204 {
205 let _ = tx.send(providers);
206 }
207 }
208 }
209 kad::QueryResult::GetProviders(Err(_)) if step.last => {
210 if let Some((providers, tx)) =
211 pending_providers.remove(&id)
212 {
213 let _ = tx.send(providers);
214 }
215 }
216 _ => {}
217 }
218 }
219 _ => {}
220 }
221 }
222 }
223 }
224 });
225
226 Ok(Self {
227 peer_id,
228 cmd_tx,
229 listen_rx: Mutex::new(listen_rx),
230 control,
231 })
232 }
233
234 pub async fn listen(&self, addr: Multiaddr) -> Multiaddr {
235 self.cmd_tx
236 .send(Command::Listen(addr))
237 .expect("swarm task alive");
238 let mut rx = self.listen_rx.lock().await;
239 rx.recv().await.expect("listen address arrives")
240 }
241
242 pub fn dial(&self, addr: Multiaddr) {
243 let _ = self.cmd_tx.send(Command::Dial(addr));
244 }
245
246 /// Añade un peer al routing table de Kademlia. Punto de entrada
247 /// para bootstrap: tras esto, el nodo puede dirigir queries DHT
248 /// a través de este peer.
249 pub fn add_dht_peer(&self, peer: PeerId, addr: Multiaddr) {
250 let _ = self.cmd_tx.send(Command::AddDhtPeer(peer, addr));
251 }
252
253 /// Consulta el DHT por los peers más cercanos al `target` PeerId.
254 /// Devuelve la lista resuelta (vacía si la query falla o si no
255 /// hay peers conocidos). Bloquea hasta que la query completa.
256 pub async fn find_closest_peers(&self, target: PeerId) -> Vec<DiscoveredPeer> {
257 let (tx, rx) = oneshot::channel();
258 let _ = self
259 .cmd_tx
260 .send(Command::FindClosestPeers(target, tx));
261 rx.await.unwrap_or_default()
262 }
263
264 /// Anuncia en el DHT que este peer tiene el contenido identificado
265 /// por `key`. Otros peers pueden luego descubrirlo vía
266 /// `find_providers(key)`. Best-effort: si la replicación falla
267 /// inicialmente, el record vive en el store local.
268 pub fn start_providing(&self, key: &[u8]) {
269 let _ = self.cmd_tx.send(Command::StartProviding(key.to_vec()));
270 }
271
272 /// Consulta el DHT por peers que han anunciado proveer `key`.
273 /// Devuelve la lista de `PeerId`s que se reportan como providers.
274 /// Lista vacía si nadie anuncia.
275 pub async fn find_providers(&self, key: &[u8]) -> Vec<PeerId> {
276 let (tx, rx) = oneshot::channel();
277 let _ = self
278 .cmd_tx
279 .send(Command::GetProviders(key.to_vec(), tx));
280 rx.await.unwrap_or_default()
281 }
282}