tawasuyu/brahman

1//! Tests del `MingaPeer` con backing persistente.
2//!
3//! Verifica que:
4//! - Abrir un path nuevo crea un repo vacío.
5//! - Datos ingresados a un peer abierto se persisten a disco.
6//! - Tras cerrar y reabrir el mismo path, el estado completo se
7//!   recupera (MST con mismo `root_hash`, store con todos los nodos
8//!   reconstruibles, atestaciones intactas y verificables).
9//! - El sync sobre red poblando un peer persistente sobrevive
10//!   reinicio.
11
12use std::time::Duration;
13
14use minga_core::{parse, Attestation, AttestationStore, Keypair, MemStore, Mst, NodeStore};
15use minga_p2p::{MingaPeer, SyncSession};
16use tempfile::TempDir;
17
18fn kp(seed: u8) -> Keypair {
19    Keypair::from_seed(&[seed; 32])
20}
21
22#[tokio::test]
23async fn open_creates_empty_repo_at_new_path() {
24    let dir = TempDir::new().unwrap();
25    let peer = MingaPeer::open(kp(1), dir.path()).unwrap();
26    let (mst, store, atts) = peer.snapshot().await;
27    assert!(mst.is_empty());
28    assert!(store.is_empty());
29    assert!(atts.is_empty());
30}
31
32#[tokio::test]
33async fn ingest_persists_across_restart() {
34    let dir = TempDir::new().unwrap();
35    let kp_a = kp(1);
36
37    let n = parse::rust("fn add(x: i32, y: i32) -> i32 { x + y }").unwrap();
38    let h_expected = minga_core::hash_node(&n);
39
40    // Sesión 1: abrir, ingerir, flush, drop.
41    {
42        let peer = MingaPeer::open(kp_a.clone(), dir.path()).unwrap();
43        let h = peer.ingest(&n).await;
44        assert_eq!(h, h_expected);
45        peer.flush().await.unwrap();
46    }
47
48    // Sesión 2: reabrir, verificar que todo está intacto.
49    {
50        let peer = MingaPeer::open(kp_a, dir.path()).unwrap();
51        let (mst, store, _) = peer.snapshot().await;
52        assert_eq!(mst.len(), 1);
53        assert!(mst.contains(&h_expected));
54        assert!(store.contains(&h_expected));
55
56        // Reconstrucción exacta del árbol original.
57        let reconstructed = store.reconstruct(&h_expected).unwrap();
58        assert_eq!(reconstructed, n);
59    }
60}
61
62#[tokio::test]
63async fn ingest_attestation_persists_across_restart() {
64    let dir = TempDir::new().unwrap();
65    let kp_owner = kp(1);
66    let kp_signer = kp(2);
67
68    let n = parse::rust("fn signed_function() -> i32 { 42 }").unwrap();
69    let h = minga_core::hash_node(&n);
70
71    {
72        let peer = MingaPeer::open(kp_owner.clone(), dir.path()).unwrap();
73        peer.ingest(&n).await;
74        let att = Attestation::create(&kp_signer, h);
75        peer.ingest_attestation(att).await.unwrap();
76        peer.flush().await.unwrap();
77    }
78
79    {
80        let peer = MingaPeer::open(kp_owner, dir.path()).unwrap();
81        let (_, _, atts) = peer.snapshot().await;
82        let authors = atts.authors_of(&h);
83        assert_eq!(authors, vec![kp_signer.did()]);
84
85        // La firma sigue verificando tras viajar disco→memoria.
86        let stored_atts = atts.get(&h);
87        assert_eq!(stored_atts.len(), 1);
88        assert!(stored_atts[0].verify());
89    }
90}
91
92#[tokio::test]
93async fn ingest_multiple_authors_for_same_content_persist() {
94    let dir = TempDir::new().unwrap();
95    let kp_owner = kp(1);
96    let alice = kp(10);
97    let bob = kp(20);
98    let carol = kp(30);
99
100    let n = parse::rust("fn shared() -> i32 { 0 }").unwrap();
101    let h = minga_core::hash_node(&n);
102
103    {
104        let peer = MingaPeer::open(kp_owner.clone(), dir.path()).unwrap();
105        peer.ingest(&n).await;
106        peer.ingest_attestation(Attestation::create(&alice, h))
107            .await
108            .unwrap();
109        peer.ingest_attestation(Attestation::create(&bob, h))
110            .await
111            .unwrap();
112        peer.ingest_attestation(Attestation::create(&carol, h))
113            .await
114            .unwrap();
115        peer.flush().await.unwrap();
116    }
117
118    {
119        let peer = MingaPeer::open(kp_owner, dir.path()).unwrap();
120        let (_, _, atts) = peer.snapshot().await;
121        let mut authors = atts.authors_of(&h);
122        authors.sort_by_key(|d| d.0);
123        assert_eq!(authors.len(), 3);
124        let mut expected = vec![alice.did(), bob.did(), carol.did()];
125        expected.sort_by_key(|d| d.0);
126        assert_eq!(authors, expected);
127    }
128}
129
130#[tokio::test]
131async fn root_hash_stable_across_restart() {
132    // El `root_hash` del MST es función pura del set de claves. Tras
133    // reabrir desde disco, debe ser idéntico.
134    let dir = TempDir::new().unwrap();
135    let kp_a = kp(1);
136
137    let target_root_hash;
138    {
139        let peer = MingaPeer::open(kp_a.clone(), dir.path()).unwrap();
140        for src in &[
141            "fn one() -> i32 { 1 }",
142            "fn two() -> i32 { 2 }",
143            "fn three(x: i32) -> i32 { x * x }",
144        ] {
145            peer.ingest(&parse::rust(src).unwrap()).await;
146        }
147        target_root_hash = peer.snapshot().await.0.root_hash();
148        peer.flush().await.unwrap();
149    }
150
151    {
152        let peer = MingaPeer::open(kp_a, dir.path()).unwrap();
153        let (mst, _, _) = peer.snapshot().await;
154        assert_eq!(mst.root_hash(), target_root_hash);
155        assert_eq!(mst.len(), 3);
156    }
157}
158
159#[tokio::test]
160async fn sync_into_persistent_peer_survives_restart() {
161    // Caso end-to-end: peer A pasivo y persistente. B sincroniza con
162    // A. A persiste lo que recibió. Cerramos A. Reabrimos. El estado
163    // sincronizado sigue ahí.
164    let dir = TempDir::new().unwrap();
165    let kp_a = kp(1);
166
167    let n = parse::rust("fn from_b(z: i32) -> i32 { z + 7 }").unwrap();
168    let h_b = minga_core::hash_node(&n);
169
170    // ── Sesión 1: A persistente acepta sync de B ─────────────────
171    {
172        let a = MingaPeer::open(kp_a.clone(), dir.path()).unwrap();
173        let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
174        let accept = a.run_passive_accept();
175
176        // B en memoria, le sincroniza su contenido.
177        let mut store_b = MemStore::new();
178        let mut mst_b = Mst::new();
179        let h = store_b.put(&n);
180        mst_b.insert(h);
181        let b = MingaPeer::new(kp(2), mst_b, store_b, AttestationStore::new()).unwrap();
182        b.dial(addr_a);
183
184        // Reintentar sync hasta éxito.
185        let deadline = std::time::Instant::now() + Duration::from_secs(5);
186        loop {
187            if b.sync_with(a.peer_id()).await.is_ok() {
188                break;
189            }
190            if std::time::Instant::now() >= deadline {
191                panic!("sync no completó en 5s");
192            }
193            tokio::time::sleep(Duration::from_millis(50)).await;
194        }
195
196        // Esperar a que A's accept handler haya mergeado.
197        let deadline = std::time::Instant::now() + Duration::from_secs(2);
198        loop {
199            let (mst_a, _, _) = a.snapshot().await;
200            if mst_a.contains(&h_b) {
201                break;
202            }
203            if std::time::Instant::now() >= deadline {
204                panic!("merge en A no se vio en 2s");
205            }
206            tokio::time::sleep(Duration::from_millis(20)).await;
207        }
208
209        a.flush().await.unwrap();
210
211        // Cleanup explícito: abort la accept task y espera a que
212        // termine para liberar el lock de sled.
213        accept.abort();
214        let _ = accept.await;
215    }
216
217    // Pequeño margen para que tasks spawneadas terminen y los Arc
218    // se liberen.
219    tokio::time::sleep(Duration::from_millis(200)).await;
220
221    // ── Sesión 2: reabrir A, verificar contenido sincronizado ────
222    {
223        let a = MingaPeer::open(kp_a, dir.path()).unwrap();
224        let (mst_a, store_a, _) = a.snapshot().await;
225        assert!(
226            mst_a.contains(&h_b),
227            "el contenido de B no sobrevivió al reinicio"
228        );
229        assert!(store_a.contains(&h_b));
230
231        // Reconstruimos: lo que B firmó sigue ahí íntegro.
232        let reconstructed = store_a.reconstruct(&h_b).unwrap();
233        assert_eq!(reconstructed, n);
234    }
235}
236
237// Helper: silencia un warning si SyncSession se importa pero no se usa.
238#[allow(dead_code)]
239fn _session_marker(_: SyncSession) {}