tawasuyu/brahman

1//! Tests de descubrimiento vía Kademlia DHT.
2
3use std::time::Duration;
4
5use minga_core::{parse, AttestationStore, Keypair, MemStore, Mst, NodeStore};
6use minga_p2p::{LibP2pNode, MingaPeer};
7
8#[tokio::test]
9async fn identify_auto_populates_kad_routing_table() {
10    // Sin `add_dht_peer` manual: solo dial. Identify intercambia
11    // direcciones automáticamente y poblamos Kad con ellas. Tras
12    // unos cientos de ms, A puede consultar B vía DHT.
13    let a = LibP2pNode::new().unwrap();
14    let b = LibP2pNode::new().unwrap();
15
16    let addr_b = b.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
17
18    a.dial(addr_b);
19
20    // Margen para handshake Noise + Yamux + Identify.
21    tokio::time::sleep(Duration::from_millis(500)).await;
22
23    let result = a.find_closest_peers(b.peer_id).await;
24    assert!(
25        result.iter().any(|p| p.peer_id == b.peer_id),
26        "tras Identify, B debe estar en el routing de A. Obtuvo: {:?}",
27        result.iter().map(|p| p.peer_id).collect::<Vec<_>>()
28    );
29}
30
31#[tokio::test]
32async fn kad_two_node_basic_discovery() {
33    // A escucha. B dializa, añade A al routing table de Kad.
34    // Tras el handshake Kad, B puede consultar el DHT y encontrar A.
35    let a = LibP2pNode::new().unwrap();
36    let b = LibP2pNode::new().unwrap();
37
38    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
39
40    b.add_dht_peer(a.peer_id, addr_a.clone());
41    b.dial(addr_a.clone());
42
43    // Damos margen para handshake Noise+Yamux+Kad.
44    tokio::time::sleep(Duration::from_millis(300)).await;
45
46    let result = b.find_closest_peers(a.peer_id).await;
47    assert!(
48        result.iter().any(|p| p.peer_id == a.peer_id),
49        "B debe encontrar A vía DHT, obtuvo {:?}",
50        result
51    );
52}
53
54#[tokio::test]
55async fn kad_three_node_discovery_via_rendezvous() {
56    // Test canónico de descubrimiento DHT:
57    // - A es un peer "rendezvous" que pre-conoce a B y C (en una red
58    //   real, A los aprendería de los handshakes Kad cuando B y C se
59    //   conectan; aquí lo seedeamos explícitamente para no depender
60    //   de timing de propagación).
61    // - B solo conoce a A.
62    // - B pregunta al DHT por C: la query va a A, A responde con C,
63    //   B aprende la dirección de C sin haberle hablado nunca.
64    //
65    // Este es exactamente el patrón de IPFS, libp2p bootstrap nodes
66    // y cualquier P2P descentralizado real.
67
68    let a = LibP2pNode::new().unwrap(); // rendezvous
69    let b = LibP2pNode::new().unwrap();
70    let c = LibP2pNode::new().unwrap();
71
72    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
73    let addr_b = b.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
74    let addr_c = c.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
75
76    // A (el rendezvous) tiene a B y C en su routing table.
77    a.add_dht_peer(b.peer_id, addr_b);
78    a.add_dht_peer(c.peer_id, addr_c);
79
80    // B solo conoce a A.
81    b.add_dht_peer(a.peer_id, addr_a.clone());
82    b.dial(addr_a.clone());
83
84    // Margen para que la conexión Kad B↔A se establezca.
85    tokio::time::sleep(Duration::from_millis(300)).await;
86
87    // B pregunta al DHT por C. Su routing table solo tiene A; la
88    // query va a A; A responde con C de su table. B descubre.
89    let result = b.find_closest_peers(c.peer_id).await;
90    assert!(
91        result.iter().any(|p| p.peer_id == c.peer_id),
92        "B debe descubrir C vía A; obtuvo: {:?}",
93        result.iter().map(|p| p.peer_id).collect::<Vec<_>>()
94    );
95
96    // Y la dirección de C debe haber viajado en el resultado, así
97    // que B podría dialarlo directamente sin pasar por A.
98    let c_entry = result.iter().find(|p| p.peer_id == c.peer_id).unwrap();
99    assert!(!c_entry.addrs.is_empty(), "C debe venir con address resoluble");
100}
101
102#[tokio::test]
103async fn kad_discovery_then_sync() {
104    // Cierre del bucle: B descubre C vía DHT a través de A, y luego
105    // sincroniza directamente con C. Discovery + transport + sync
106    // protocolar autenticado, todo end-to-end sobre red real.
107
108    fn singleton(seed: u8, src: &str) -> MingaPeer {
109        let mut mst = Mst::new();
110        let mut store = MemStore::new();
111        let h = store.put(&parse::rust(src).unwrap());
112        mst.insert(h);
113        MingaPeer::new(
114            Keypair::from_seed(&[seed; 32]),
115            mst,
116            store,
117            AttestationStore::new(),
118        )
119        .unwrap()
120    }
121
122    // A: rendezvous puro, solo Kad (no MingaPeer, no necesita estado).
123    let a = LibP2pNode::new().unwrap();
124    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
125
126    // C: tiene una función que B querrá. Pasivo para aceptar el sync.
127    let c = singleton(3, "fn from_c(x: i32) -> i32 { x + 100 }");
128    let addr_c = c.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
129    let _accept_c = c.run_passive_accept();
130
131    // A pre-conoce a C en su routing table (rendezvous comportándose
132    // como tal).
133    a.add_dht_peer(c.peer_id(), addr_c);
134
135    // B: tiene su propia función. Solo conoce A.
136    let b = singleton(2, "fn from_b() -> i32 { 0 }");
137    b.add_dht_peer(a.peer_id, addr_a.clone());
138    b.dial(addr_a.clone());
139
140    tokio::time::sleep(Duration::from_millis(300)).await;
141
142    // B descubre a C vía DHT.
143    let discovered = b.find_closest_peers(c.peer_id()).await;
144    let c_entry = discovered
145        .iter()
146        .find(|p| p.peer_id == c.peer_id())
147        .unwrap_or_else(|| {
148            panic!(
149                "B no descubrió C; encontró: {:?}",
150                discovered.iter().map(|p| p.peer_id).collect::<Vec<_>>()
151            )
152        });
153
154    // B usa la dirección descubierta para dial directo y sync.
155    let addr_c_via_dht = c_entry.addrs[0].clone();
156    b.dial(addr_c_via_dht);
157
158    // Reintentamos sync hasta que la conexión esté arriba.
159    let deadline = std::time::Instant::now() + Duration::from_secs(5);
160    loop {
161        if b.sync_with(c.peer_id()).await.is_ok() {
162            break;
163        }
164        if std::time::Instant::now() >= deadline {
165            panic!("sync no completó en 5s");
166        }
167        tokio::time::sleep(Duration::from_millis(50)).await;
168    }
169
170    // Tras el sync, B y C tienen el mismo MST (unión). El merge de
171    // C sucede en su task de accept (paralela a B); esperamos a que
172    // ese merge se vea reflejado en su state.
173    let deadline = std::time::Instant::now() + Duration::from_secs(2);
174    loop {
175        let (mst_b, _, _) = b.snapshot().await;
176        let (mst_c, _, _) = c.snapshot().await;
177        if mst_b.root_hash() == mst_c.root_hash() && mst_b.len() == 2 {
178            break;
179        }
180        if std::time::Instant::now() >= deadline {
181            panic!(
182                "no convergencia tras 2s: |B|={}, |C|={}",
183                mst_b.len(),
184                mst_c.len()
185            );
186        }
187        tokio::time::sleep(Duration::from_millis(20)).await;
188    }
189}