tawasuyu/brahman

1//! Test de integración real con libp2p.
2//!
3//! Dos `LibP2pNode`s independientes en localhost:
4//! - cada uno con su propia identidad libp2p,
5//! - conectados por TCP (con cifrado Noise + multiplexado Yamux),
6//! - intercambiando una sesión completa de sync vía bidirectional
7//!   streams sobre el protocolo `/minga/sync/1.0.0`.
8//!
9//! Lo único que el wire añade respecto al harness in-memory es el
10//! transporte. La lógica del protocolo y el state machine son los
11//! mismos — eso es exactamente lo que queríamos demostrar.
12
13use std::time::Duration;
14
15use futures::StreamExt;
16use minga_core::{parse, ContentHash, Keypair, MemStore, Mst, NodeStore};
17use minga_p2p::{run_sync_async, LibP2pNode, SyncSession, SYNC_PROTOCOL};
18use tokio_util::compat::FuturesAsyncReadCompatExt;
19
20fn kp(seed: u8) -> Keypair {
21    Keypair::from_seed(&[seed; 32])
22}
23
24fn build_repo(sources: &[&str]) -> (Mst, MemStore, Vec<ContentHash>) {
25    let mut mst = Mst::new();
26    let mut store = MemStore::new();
27    let mut roots = Vec::new();
28    for src in sources {
29        let n = parse::rust(src).unwrap();
30        let h = store.put(&n);
31        mst.insert(h);
32        roots.push(h);
33    }
34    (mst, store, roots)
35}
36
37#[tokio::test]
38async fn libp2p_sync_two_peers_over_tcp() {
39    let node_a = LibP2pNode::new().unwrap();
40    let node_b = LibP2pNode::new().unwrap();
41    let peer_b = node_b.peer_id;
42
43    // Solo B necesita escuchar; A inicia el dial.
44    let addr_b = node_b
45        .listen("/ip4/127.0.0.1/tcp/0".parse().unwrap())
46        .await;
47
48    // B acepta streams del protocolo Minga en una tarea.
49    let only_b_sources = &["fn from_b(x: i32) -> i32 { x + 1 }"];
50    let (mst_b, store_b, _) = build_repo(only_b_sources);
51    let session_b = SyncSession::without_attestations(mst_b, store_b, kp(2));
52
53    let mut control_b = node_b.control.clone();
54    let task_b = tokio::spawn(async move {
55        let mut incoming = control_b.accept(SYNC_PROTOCOL).unwrap();
56        let (_peer, stream) = incoming.next().await.expect("incoming stream");
57        run_sync_async(session_b, stream.compat()).await
58    });
59
60    // A dializa B y abre stream. Reintenta hasta que la conexión esté
61    // arriba (puede tardar unos ms el handshake Noise+Yamux).
62    node_a.dial(addr_b);
63    let mut control_a = node_a.control.clone();
64    let stream_a = {
65        let deadline = std::time::Instant::now() + Duration::from_secs(5);
66        loop {
67            match control_a.open_stream(peer_b, SYNC_PROTOCOL).await {
68                Ok(s) => break s,
69                Err(_) if std::time::Instant::now() < deadline => {
70                    tokio::time::sleep(Duration::from_millis(50)).await;
71                }
72                Err(e) => panic!("no se pudo abrir stream tras 5s: {e:?}"),
73            }
74        }
75    };
76
77    let only_a_sources = &["fn from_a() -> i32 { 0 }"];
78    let (mst_a, store_a, _) = build_repo(only_a_sources);
79    let session_a = SyncSession::without_attestations(mst_a, store_a, kp(1));
80
81    let task_a = tokio::spawn(async move { run_sync_async(session_a, stream_a.compat()).await });
82
83    let result_a = task_a.await.expect("task A").expect("sync A");
84    let result_b = task_b.await.expect("task B").expect("sync B");
85
86    // Convergencia tras viajar sobre TCP real.
87    assert_eq!(result_a.mst().root_hash(), result_b.mst().root_hash());
88    assert_eq!(result_a.mst().len(), 2);
89    assert_eq!(result_b.mst().len(), 2);
90
91    // Cada peer terminó con la identidad libp2p del otro autenticada.
92    // (Las identidades libp2p no son las mismas que los DIDs Minga —
93    // las primeras autentican el canal, los segundos firman contenido.)
94    assert!(result_a.peer_did().is_some());
95    assert!(result_b.peer_did().is_some());
96}
97
98#[tokio::test]
99async fn libp2p_sync_with_attestations() {
100    use minga_core::{Attestation, AttestationStore};
101
102    let node_a = LibP2pNode::new().unwrap();
103    let node_b = LibP2pNode::new().unwrap();
104    let peer_b = node_b.peer_id;
105
106    let addr_b = node_b
107        .listen("/ip4/127.0.0.1/tcp/0".parse().unwrap())
108        .await;
109
110    let kp_a = kp(10);
111    let kp_b = kp(20);
112
113    let (mst_a, store_a, roots_a) = build_repo(&["fn signed_by_a() -> i32 { 1 }"]);
114    let (mst_b, store_b, roots_b) = build_repo(&["fn signed_by_b() -> i32 { 2 }"]);
115
116    let mut atts_a = AttestationStore::new();
117    atts_a.add(Attestation::create(&kp_a, roots_a[0])).unwrap();
118
119    let mut atts_b = AttestationStore::new();
120    atts_b.add(Attestation::create(&kp_b, roots_b[0])).unwrap();
121
122    let session_a = SyncSession::new(mst_a, store_a, atts_a, kp_a.clone());
123    let session_b = SyncSession::new(mst_b, store_b, atts_b, kp_b.clone());
124
125    let mut control_b = node_b.control.clone();
126    let task_b = tokio::spawn(async move {
127        let mut incoming = control_b.accept(SYNC_PROTOCOL).unwrap();
128        let (_peer, stream) = incoming.next().await.expect("incoming stream");
129        run_sync_async(session_b, stream.compat()).await
130    });
131
132    node_a.dial(addr_b);
133    let mut control_a = node_a.control.clone();
134    let stream_a = {
135        let deadline = std::time::Instant::now() + Duration::from_secs(5);
136        loop {
137            match control_a.open_stream(peer_b, SYNC_PROTOCOL).await {
138                Ok(s) => break s,
139                Err(_) if std::time::Instant::now() < deadline => {
140                    tokio::time::sleep(Duration::from_millis(50)).await;
141                }
142                Err(e) => panic!("no se pudo abrir stream: {e:?}"),
143            }
144        }
145    };
146
147    let task_a = tokio::spawn(async move { run_sync_async(session_a, stream_a.compat()).await });
148
149    let result_a = task_a.await.unwrap().unwrap();
150    let result_b = task_b.await.unwrap().unwrap();
151
152    // Atestaciones cruzaron criptográficamente verificadas.
153    assert_eq!(
154        result_a.attestations().authors_of(&roots_b[0]),
155        vec![kp_b.did()]
156    );
157    assert_eq!(
158        result_b.attestations().authors_of(&roots_a[0]),
159        vec![kp_a.did()]
160    );
161}