tawasuyu/llimphi

1//! llimphi-wasm-net — transporte P2P para distribución de apps WASM Tier 3.
2//!
3//! Un protocolo de stream mínimo sobre `BrahmanNet` (card-net, la malla libp2p
4//! TCP+Noise+Yamux+Kad compartida):
5//!
6//! - [`serve_blobs`] levanta un responder que, por cada stream entrante en
7//!   `/llimphi-wasm/blob/1.0.0`, atiende un `Get{hash}` sirviendo el bytecode
8//!   de un [`DiskStore`] (o `NotFound`).
9//! - [`fetch_blob`] abre un stream a un provider y pide un blob por hash.
10//! - [`fetch_blob_dht`] descubre providers por la DHT y prueba con cada uno.
11//! - [`RemoteSource`] envuelve lo anterior como un [`BlobSource`] síncrono, para
12//!   que la cadena `resolve→verificar` de `llimphi-wasm-dist` corra sobre la red
13//!   sin enterarse del transporte.
14//!
15//! El framing copia el de minga (`[u32 LE len][postcard]`). La integridad NO se
16//! confía a la red: `llimphi-wasm-dist::resolve` rehashea el blob recibido
17//! contra el hash pedido, así un provider malicioso no puede colar otro wasm.
18
19use std::sync::Arc;
20use std::time::{Duration, Instant};
21
22use card_net::{BrahmanNet, PeerId, Stream, StreamProtocol};
23use futures::{AsyncReadExt, AsyncWriteExt, StreamExt};
24use serde::{Deserialize, Serialize};
25
26pub use llimphi_wasm_core::{BlobSource, DiskStore, Hash};
27
28/// El protocolo de stream para pedir bytecodes por hash.
29pub const BLOB_PROTOCOL: StreamProtocol = StreamProtocol::new("/llimphi-wasm/blob/1.0.0");
30
31/// Tope de un blob en la red (64 MiB). Un wasm de app real está muy por debajo.
32const MAX_BLOB: u32 = 64 * 1024 * 1024;
33
34/// Mensajes del protocolo. Un request, una respuesta, stream cerrado.
35#[derive(Debug, Serialize, Deserialize)]
36enum BlobMsg {
37    Get { hash: Hash },
38    Data { hash: Hash, data: Vec<u8> },
39    NotFound { hash: Hash },
40}
41
42/// Errores del transporte.
43#[derive(Debug)]
44pub enum NetError {
45    /// No se pudo abrir el stream al provider.
46    OpenStream(String),
47    /// El provider no tiene el blob.
48    NotFound,
49    /// Respuesta fuera de protocolo.
50    Protocol,
51    /// Ningún provider entregó el blob.
52    SinProviders,
53    /// Error de I/O sobre el stream.
54    Io(std::io::Error),
55}
56
57impl std::fmt::Display for NetError {
58    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59        match self {
60            NetError::OpenStream(e) => write!(f, "abrir stream: {e}"),
61            NetError::NotFound => write!(f, "el provider no tiene el blob"),
62            NetError::Protocol => write!(f, "respuesta fuera de protocolo"),
63            NetError::SinProviders => write!(f, "ningún provider entregó el blob"),
64            NetError::Io(e) => write!(f, "io: {e}"),
65        }
66    }
67}
68
69impl std::error::Error for NetError {}
70
71// =====================================================================
72// Framing — [u32 LE len][postcard], igual que minga
73// =====================================================================
74
75async fn send_msg<S: futures::AsyncWrite + Unpin>(s: &mut S, m: &BlobMsg) -> std::io::Result<()> {
76    let bytes = postcard::to_allocvec(m).expect("postcard encode");
77    let len = bytes.len() as u32;
78    s.write_all(&len.to_le_bytes()).await?;
79    s.write_all(&bytes).await?;
80    s.flush().await?;
81    Ok(())
82}
83
84async fn recv_msg<S: futures::AsyncRead + Unpin>(s: &mut S) -> std::io::Result<BlobMsg> {
85    let mut len_buf = [0u8; 4];
86    s.read_exact(&mut len_buf).await?;
87    let len = u32::from_le_bytes(len_buf);
88    if len > MAX_BLOB {
89        return Err(std::io::Error::new(
90            std::io::ErrorKind::InvalidData,
91            "blob excede el tope",
92        ));
93    }
94    let mut buf = vec![0u8; len as usize];
95    s.read_exact(&mut buf).await?;
96    postcard::from_bytes(&buf)
97        .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "postcard inválido"))
98}
99
100// =====================================================================
101// Responder
102// =====================================================================
103
104/// Levanta el responder de blobs sobre `net`: por cada stream entrante en
105/// [`BLOB_PROTOCOL`], sirve el bytecode pedido desde `store`. Devuelve el
106/// `JoinHandle` del loop de aceptación (vive mientras se sirva).
107///
108/// Llamar UNA vez por `net` (una sola `accept` por protocolo).
109pub fn serve_blobs(net: &BrahmanNet, store: Arc<DiskStore>) -> tokio::task::JoinHandle<()> {
110    let mut control = net.control.clone();
111    tokio::spawn(async move {
112        let mut incoming = match control.accept(BLOB_PROTOCOL) {
113            Ok(i) => i,
114            Err(_) => return, // ya hay otro handle aceptando este protocolo
115        };
116        while let Some((_peer, stream)) = incoming.next().await {
117            let store = Arc::clone(&store);
118            tokio::spawn(serve_one(stream, store));
119        }
120    })
121}
122
123async fn serve_one(mut stream: Stream, store: Arc<DiskStore>) {
124    if let Ok(BlobMsg::Get { hash }) = recv_msg(&mut stream).await {
125        let reply = match store.get(&hash) {
126            Some(data) => BlobMsg::Data { hash, data },
127            None => BlobMsg::NotFound { hash },
128        };
129        let _ = send_msg(&mut stream, &reply).await;
130        let _ = stream.close().await;
131    }
132}
133
134/// Anuncia en la DHT que este nodo provee un bytecode (para [`fetch_blob_dht`]).
135pub fn announce(net: &BrahmanNet, hash: &Hash) {
136    net.start_providing(hash);
137}
138
139// =====================================================================
140// Cliente
141// =====================================================================
142
143/// Pide un blob por hash a un provider concreto. Reintenta abrir el stream
144/// hasta 5 s (la conexión puede no estar lista al instante tras un `dial`).
145pub async fn fetch_blob(net: &BrahmanNet, peer: PeerId, hash: &Hash) -> Result<Vec<u8>, NetError> {
146    let mut control = net.control.clone();
147    let deadline = Instant::now() + Duration::from_secs(5);
148    let mut stream = loop {
149        match control.open_stream(peer, BLOB_PROTOCOL).await {
150            Ok(s) => break s,
151            Err(_) if Instant::now() < deadline => {
152                tokio::time::sleep(Duration::from_millis(50)).await;
153            }
154            Err(e) => return Err(NetError::OpenStream(e.to_string())),
155        }
156    };
157    send_msg(&mut stream, &BlobMsg::Get { hash: *hash })
158        .await
159        .map_err(NetError::Io)?;
160    match recv_msg(&mut stream).await.map_err(NetError::Io)? {
161        BlobMsg::Data { data, .. } => Ok(data),
162        BlobMsg::NotFound { .. } => Err(NetError::NotFound),
163        _ => Err(NetError::Protocol),
164    }
165}
166
167/// Descubre providers del `hash` por la DHT y pide el blob a cada uno hasta que
168/// alguno lo entregue.
169pub async fn fetch_blob_dht(net: &BrahmanNet, hash: &Hash) -> Result<Vec<u8>, NetError> {
170    let providers = net.find_providers(hash).await;
171    for peer in providers {
172        if let Ok(data) = fetch_blob(net, peer, hash).await {
173            return Ok(data);
174        }
175    }
176    Err(NetError::SinProviders)
177}
178
179// =====================================================================
180// Puente síncrono: BlobSource sobre la red
181// =====================================================================
182
183/// Adapta el fetch P2P (async) al trait síncrono [`BlobSource`], para componerlo
184/// con `llimphi_wasm_dist::{resolve, LayeredSource}`. Usa la DHT para descubrir
185/// providers.
186///
187/// `fetch` hace `block_on` sobre el handle del runtime, así que **no debe
188/// llamarse desde dentro de ese runtime** (sí desde el loop síncrono del runner
189/// Llimphi, que no es async). Componer con un [`DiskStore`] vía `LayeredSource`
190/// para servir desde caché local primero y caer a la red sólo si falta.
191pub struct RemoteSource {
192    net: Arc<BrahmanNet>,
193    rt: tokio::runtime::Handle,
194}
195
196impl RemoteSource {
197    pub fn new(net: Arc<BrahmanNet>, rt: tokio::runtime::Handle) -> Self {
198        Self { net, rt }
199    }
200}
201
202impl BlobSource for RemoteSource {
203    fn fetch(&self, hash: &Hash) -> Option<Vec<u8>> {
204        self.rt.block_on(fetch_blob_dht(&self.net, hash)).ok()
205    }
206}