tawasuyu/brahman

1//! Driver de sincronización sobre I/O asíncrona.
2//!
3//! Bridge entre la `SyncSession` puramente lógica y cualquier
4//! transporte que implemente `AsyncRead + AsyncWrite`. Encuadre
5//! length-prefixed: cada `Message` se serializa con postcard y se
6//! envía precedido de un `u32 LE` con su longitud en bytes.
7//!
8//! La estructura del bucle es:
9//! 1. Drenar todos los `Message`s pendientes a la salida.
10//! 2. Si la sesión declara `is_done`, salir.
11//! 3. Bloquear esperando un `Message` entrante; alimentarlo a la
12//!    sesión y volver al paso 1.
13//!
14//! Esto funciona porque cada paso del state machine emite los
15//! mensajes que necesita inmediatamente — nunca quedan colgados
16//! mensajes por un `Message` futuro. La única espera real ocurre en
17//! el paso 3, cuando estamos esperando que el peer responda.
18
19use std::collections::VecDeque;
20use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
21
22use crate::message::Message;
23use crate::session::SyncSession;
24
25/// Cota dura sobre el tamaño de un frame, para evitar que un peer
26/// malicioso (o un bug) cause asignaciones desbocadas. 16 MB es de
27/// sobra para mensajes de sync — un `AttestPush` de cien mil
28/// atestaciones cabe en ~13 MB.
29const MAX_FRAME_SIZE: u32 = 16 * 1024 * 1024;
30
31#[derive(Debug, thiserror::Error)]
32pub enum AsyncSyncError {
33    #[error("io: {0}")]
34    Io(#[from] std::io::Error),
35
36    #[error("decode postcard: {0}")]
37    Decode(#[from] postcard::Error),
38
39    #[error("frame demasiado grande: {0} bytes")]
40    FrameTooLarge(u32),
41
42    #[error("la sesión cerró sin alcanzar `is_done`")]
43    UnexpectedClose,
44}
45
46/// Ejecuta una sesión de sincronización completa sobre una stream
47/// duplex. Devuelve la `SyncSession` resultante (con el `Mst`,
48/// `MemStore` y `AttestationStore` ya mergeados con el peer).
49pub async fn run_sync_async<S>(
50    mut session: SyncSession,
51    mut stream: S,
52) -> Result<SyncSession, AsyncSyncError>
53where
54    S: AsyncRead + AsyncWrite + Unpin,
55{
56    let mut outbound: VecDeque<Message> = session.start().into();
57
58    loop {
59        while let Some(msg) = outbound.pop_front() {
60            send_frame(&mut stream, &msg).await?;
61        }
62
63        if session.is_done() {
64            return Ok(session);
65        }
66
67        let msg = recv_frame(&mut stream).await?;
68        outbound.extend(session.handle(msg));
69    }
70}
71
72async fn send_frame<S>(stream: &mut S, msg: &Message) -> Result<(), AsyncSyncError>
73where
74    S: AsyncWrite + Unpin,
75{
76    let bytes = msg.encode();
77    let len = bytes.len() as u32;
78    if len > MAX_FRAME_SIZE {
79        return Err(AsyncSyncError::FrameTooLarge(len));
80    }
81    stream.write_all(&len.to_le_bytes()).await?;
82    stream.write_all(&bytes).await?;
83    stream.flush().await?;
84    Ok(())
85}
86
87async fn recv_frame<S>(stream: &mut S) -> Result<Message, AsyncSyncError>
88where
89    S: AsyncRead + Unpin,
90{
91    let mut len_buf = [0u8; 4];
92    stream.read_exact(&mut len_buf).await?;
93    let len = u32::from_le_bytes(len_buf);
94    if len > MAX_FRAME_SIZE {
95        return Err(AsyncSyncError::FrameTooLarge(len));
96    }
97    let mut buf = vec![0u8; len as usize];
98    stream.read_exact(&mut buf).await?;
99    Ok(Message::decode(&buf)?)
100}