tawasuyu/brahman

1//! Test E2E de Fase 2: discovery remoto vía DHT.
2//!
3//! Pipeline:
4//! 1. **Provider node (A)**: arma server con `BrahmanNet` configurado;
5//!    listen TCP; un cliente local registra una Card con un output
6//!    flow. El server llama `announce_outputs` automáticamente, lo
7//!    que hace `start_providing` en el DHT bajo la key derivada del
8//!    flow.
9//! 2. **Consumer node (B)**: arma su propio `BrahmanNet`; dial-ea al
10//!    multiaddr del provider para que ambos se conozcan vía Identify
11//!    (esto popula sus respectivos routing tables de Kademlia).
12//! 3. **B llama `find_remote_providers(flow_name, type)`**: la query
13//!    DHT propaga vía Kad, y eventually el provider responde con su
14//!    `PeerId`.
15//! 4. **Verificación**: el `PeerId` que B descubre coincide con el
16//!    de A.
17//!
18//! Notas:
19//! - Kademlia replication factor por defecto es 20; con 2 nodos no
20//!   hay propagación material — A es el único provider, B llega a A
21//!   vía la conexión directa establecida en step 2 y obtiene el record
22//!   del store local de A.
23//! - El test usa flow `monad-list:json` por familiaridad (es el flow
24//!   real que `nouser daemon` declara). Sirve también como prueba de
25//!   que el sistema completo (daemon + DHT) funcionaría con cero
26//!   cambios en la Card.
27
28use std::collections::BTreeSet;
29use std::sync::Arc;
30use std::time::Duration;
31
32use brahman_broker::{Broker, BrokerConfig};
33use brahman_card::{
34    ulid::Ulid, Card, CardKind, Flow, Flows, Lifecycle, Payload, Priority, Supervision, TypeRef,
35    CARD_SCHEMA_VERSION,
36};
37use brahman_handshake::network::{find_remote_providers, run_libp2p_accept_loop};
38use brahman_handshake::server::{Server, ServerConfig};
39use brahman_net::{BrahmanNet, Multiaddr, Protocol};
40use tempfile::TempDir;
41use tokio::sync::Mutex;
42
43fn provider_card(label: &str, flow_name: &str, type_name: &str) -> Card {
44    Card {
45        schema_version: CARD_SCHEMA_VERSION,
46        id: Ulid::new(),
47        label: label.into(),
48        provides: BTreeSet::new(),
49        requires: BTreeSet::new(),
50        permissions: Default::default(),
51        soma: Default::default(),
52        payload: Payload::Virtual,
53        supervision: Supervision::Delegate,
54        lifecycle: Lifecycle::Daemon,
55        priority: Priority::Normal,
56        kind: CardKind::Ente,
57        flow: Flows {
58            input: vec![],
59            output: vec![Flow {
60                name: flow_name.into(),
61                ty: TypeRef::Primitive {
62                    name: type_name.into(),
63                },
64                pin_to: None,
65            }],
66        },
67        ..Default::default()
68    }
69}
70
71#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
72async fn dht_discovery_finds_remote_provider() {
73    // ---- Node A (provider): server + libp2p net + Card con output ----
74    let tmp = TempDir::new().unwrap();
75    let a_unix = tmp.path().join("a.sock");
76
77    let a_broker = Arc::new(Mutex::new(Broker::new(BrokerConfig::default())));
78    let a_net = Arc::new(BrahmanNet::new().unwrap());
79    let a_peer = a_net.peer_id;
80
81    let a_server = Arc::new(
82        Server::bind(
83            &a_unix,
84            ServerConfig {
85                init_attached: true,
86                broker: Some(a_broker.clone()),
87                net: Some(a_net.clone()), // ← clave Fase 2: anuncia al DHT
88                policy: None,
89            },
90        )
91        .unwrap(),
92    );
93
94    let listen_addr: Multiaddr = "/ip4/127.0.0.1/tcp/0".parse().unwrap();
95    let a_addr = a_net.listen(listen_addr).await;
96    let mut a_full_addr = a_addr.clone();
97    a_full_addr.push(Protocol::P2p(a_peer));
98
99    tokio::spawn(run_libp2p_accept_loop(a_server.clone(), a_net.clone()));
100
101    // Unix accept loop: necesario para que Client::connect al socket
102    // local no cuelgue (Server no se auto-accepta; el caller arma el
103    // loop). Cada session entrante corre en su propia task.
104    {
105        let s = a_server.clone();
106        tokio::spawn(async move {
107            loop {
108                match s.accept_one().await {
109                    Ok(session) => {
110                        tokio::spawn(async move {
111                            let _ = session.handle().await;
112                        });
113                    }
114                    Err(_) => break,
115                }
116            }
117        });
118    }
119
120    // Registrar la Card local en A con un flow output.
121    let card = provider_card("test.engine_remote", "monad-list", "json");
122    let mut local_client = brahman_handshake::client::Client::connect(&a_unix, card)
123        .await
124        .expect("registro local en A");
125
126    // ---- Node B (consumer): otro net que dial-a a A ----
127    let b_net = BrahmanNet::new().unwrap();
128    b_net.dial(a_full_addr.clone());
129
130    // Esperar a que la conexión se establezca y Identify popule el
131    // routing table de Kad. En localhost con 2 peers, ~250ms es de
132    // sobra; sumamos margen para CI.
133    tokio::time::sleep(Duration::from_millis(500)).await;
134
135    // ---- Discovery: B busca providers de "monad-list:json" ----
136    let providers = find_remote_providers(
137        &b_net,
138        "monad-list",
139        &TypeRef::Primitive {
140            name: "json".into(),
141        },
142    )
143    .await;
144
145    assert!(
146        providers.contains(&a_peer),
147        "B debería descubrir a A vía DHT. Encontrados: {:?}, esperado: {}",
148        providers,
149        a_peer
150    );
151
152    // Sanidad: el cliente local sigue vivo durante todo el test (lo
153    // que mantiene la Card registrada y por tanto el record DHT vivo).
154    local_client.farewell().await.ok();
155}
156
157#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
158async fn dht_discovery_negative_unknown_flow() {
159    // Mismo setup que el test happy-path, pero B busca un flow que A
160    // NO ofrece. Debe devolver lista vacía dentro del timeout
161    // razonable (no colgarse).
162    let tmp = TempDir::new().unwrap();
163    let a_unix = tmp.path().join("a.sock");
164    let a_broker = Arc::new(Mutex::new(Broker::new(BrokerConfig::default())));
165    let a_net = Arc::new(BrahmanNet::new().unwrap());
166    let a_peer = a_net.peer_id;
167
168    let a_server = Arc::new(
169        Server::bind(
170            &a_unix,
171            ServerConfig {
172                init_attached: true,
173                broker: Some(a_broker),
174                net: Some(a_net.clone()),
175                policy: None,
176            },
177        )
178        .unwrap(),
179    );
180
181    let a_addr = a_net.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
182    let mut a_full = a_addr.clone();
183    a_full.push(Protocol::P2p(a_peer));
184
185    tokio::spawn(run_libp2p_accept_loop(a_server.clone(), a_net.clone()));
186
187    // Unix accept loop: necesario para que Client::connect al socket
188    // local no cuelgue (Server no se auto-accepta; el caller arma el
189    // loop). Cada session entrante corre en su propia task.
190    {
191        let s = a_server.clone();
192        tokio::spawn(async move {
193            loop {
194                match s.accept_one().await {
195                    Ok(session) => {
196                        tokio::spawn(async move {
197                            let _ = session.handle().await;
198                        });
199                    }
200                    Err(_) => break,
201                }
202            }
203        });
204    }
205
206    let card = provider_card("test.engine_other", "monad-list", "json");
207    let mut local = brahman_handshake::client::Client::connect(&a_unix, card)
208        .await
209        .unwrap();
210
211    let b_net = BrahmanNet::new().unwrap();
212    b_net.dial(a_full);
213    tokio::time::sleep(Duration::from_millis(500)).await;
214
215    // Buscamos un flow que NADIE anunció.
216    let providers = find_remote_providers(
217        &b_net,
218        "flow-que-no-existe",
219        &TypeRef::Primitive {
220            name: "json".into(),
221        },
222    )
223    .await;
224
225    assert!(
226        providers.is_empty(),
227        "no debería haber providers para un flow inexistente, got: {:?}",
228        providers
229    );
230
231    local.farewell().await.ok();
232}
233
234/// stop_providing test: A registra Card con flow X, B descubre a A.
235/// El cliente local de A hace farewell → cleanup llama
236/// withdraw_outputs → A se quita del provider local store. Una nueva
237/// query desde B (que rutea por A, único peer en el DHT) ya no debe
238/// listarlo.
239#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
240async fn dht_discovery_withdraws_on_session_cleanup() {
241    let tmp = TempDir::new().unwrap();
242    let a_unix = tmp.path().join("a.sock");
243    let a_broker = Arc::new(Mutex::new(Broker::new(BrokerConfig::default())));
244    let a_net = Arc::new(BrahmanNet::new().unwrap());
245    let a_peer = a_net.peer_id;
246
247    let a_server = Arc::new(
248        Server::bind(
249            &a_unix,
250            ServerConfig {
251                init_attached: true,
252                broker: Some(a_broker),
253                net: Some(a_net.clone()),
254                policy: None,
255            },
256        )
257        .unwrap(),
258    );
259    let sessions = a_server.sessions();
260
261    let a_addr = a_net.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
262    let mut a_full = a_addr.clone();
263    a_full.push(Protocol::P2p(a_peer));
264
265    tokio::spawn(run_libp2p_accept_loop(a_server.clone(), a_net.clone()));
266    {
267        let s = a_server.clone();
268        tokio::spawn(async move {
269            loop {
270                match s.accept_one().await {
271                    Ok(session) => {
272                        tokio::spawn(async move {
273                            let _ = session.handle().await;
274                        });
275                    }
276                    Err(_) => break,
277                }
278            }
279        });
280    }
281
282    // Card con un flow output anunciable.
283    let card = provider_card("test.withdraws", "monad-list", "json");
284    let local = brahman_handshake::client::Client::connect(&a_unix, card)
285        .await
286        .expect("registro local en A");
287
288    let b_net = BrahmanNet::new().unwrap();
289    b_net.dial(a_full);
290    tokio::time::sleep(Duration::from_millis(500)).await;
291
292    // Confirmación previa: A es discoverable.
293    let before = find_remote_providers(
294        &b_net,
295        "monad-list",
296        &TypeRef::Primitive {
297            name: "json".into(),
298        },
299    )
300    .await;
301    assert!(
302        before.contains(&a_peer),
303        "antes del farewell A debería ser discoverable. got: {:?}",
304        before
305    );
306
307    // Farewell del cliente local → server.cleanup → withdraw_outputs.
308    local.farewell().await.ok();
309
310    // Esperamos a que la sesión salga del registro de A (señal de
311    // que cleanup completó).
312    let mut waited = 0;
313    while !sessions.lock().await.is_empty() && waited < 50 {
314        tokio::time::sleep(Duration::from_millis(20)).await;
315        waited += 1;
316    }
317    assert!(
318        sessions.lock().await.is_empty(),
319        "sesión debería estar removida tras farewell"
320    );
321
322    // Pequeño margen extra para que el Command::StopProviding lo
323    // procese el swarm task (no es await-able desde fuera).
324    tokio::time::sleep(Duration::from_millis(100)).await;
325
326    // Nueva query: A ya no debería listarse como provider.
327    let after = find_remote_providers(
328        &b_net,
329        "monad-list",
330        &TypeRef::Primitive {
331            name: "json".into(),
332        },
333    )
334    .await;
335    assert!(
336        !after.contains(&a_peer),
337        "tras farewell + withdraw_outputs, A NO debería ser discoverable. got: {:?}",
338        after
339    );
340}