⎇ 5edc912ed85acafe93abe9bca04113cc836a8f90
crates/core/brahman-handshake/src/server.rs
Historia
Quién tocó qué
crudo
feat: Phase D-3 + D-4 — service_socket en Card, providers coexisten
1//! Servidor de handshake. Listener Unix socket → sesiones por conexión.
2
3use std::collections::HashMap;
4use std::path::{Path, PathBuf};
5use std::sync::Arc;
6use std::time::{SystemTime, UNIX_EPOCH};
7
8use brahman_broker::{Broker, Endpoint};
9use brahman_card::{Card, ResolvedCard, WitInterface, CARD_SCHEMA_VERSION};
10use tokio::net::{UnixListener, UnixStream};
11use tokio::sync::{mpsc, Mutex};
12use tracing::{debug, warn};
13use ulid::Ulid;
14
15use crate::codec::{read_frame, write_frame};
16use crate::messages::{
17 Farewell, Frame, HandshakeError, Hello, HelloAck, MatchEvent, MatchEventKind, Ping, Pong,
18 SessionId,
19};
20
21/// Tabla de sesiones vivas indexada por `SessionId`.
22pub type SessionRegistry = Arc<Mutex<HashMap<SessionId, ResolvedCard>>>;
23
24/// Broker compartido (opcional) que el servidor mantiene en sincronía con
25/// el ciclo de vida de las sesiones.
26pub type SharedBroker = Arc<Mutex<Broker>>;
27
28/// Tabla de canales push por sesión: el server inyecta frames hacia el
29/// cliente (p. ej. `MatchEvent`) sin requerir que el cliente haga request.
30type SessionTxTable = Arc<Mutex<HashMap<SessionId, mpsc::Sender<Frame>>>>;
31
32/// Por sesión, último match conocido por nombre de input. Se usa para
33/// emitir diffs (Available/Lost) en lugar del estado completo.
34type LastMatches = Arc<Mutex<HashMap<SessionId, HashMap<String, Endpoint>>>>;
35
36/// Capacidad del canal push por sesión. Si se llena (cliente lento), los
37/// eventos extra se descartan — el cliente puede re-consultar el estado.
38const PUSH_CHANNEL_CAPACITY: usize = 32;
39
40/// Configuración del servidor.
41#[derive(Debug, Clone, Default)]
42pub struct ServerConfig {
43 /// `true` si el Init está atado al servidor (se reporta en `HelloAck`).
44 pub init_attached: bool,
45 /// Broker compartido. Si está presente, el servidor llama
46 /// `register` tras un Hello aceptado y `unregister` al cerrar la
47 /// sesión (Farewell o EOF). Si es `None`, el broker no se usa.
48 pub broker: Option<SharedBroker>,
49}
50
51/// Servidor de handshake escuchando en un Unix socket.
52pub struct Server {
53 listener: UnixListener,
54 socket_path: PathBuf,
55 sessions: SessionRegistry,
56 push_table: SessionTxTable,
57 last_matches: LastMatches,
58 config: ServerConfig,
59}
60
61impl Server {
62 /// Crea el listener en `path`. Si el archivo existe, lo elimina (asume
63 /// que es un socket stale de una sesión previa).
64 pub fn bind(path: impl Into<PathBuf>, config: ServerConfig) -> std::io::Result<Self> {
65 let socket_path = path.into();
66 if socket_path.exists() {
67 std::fs::remove_file(&socket_path)?;
68 }
69 if let Some(parent) = socket_path.parent() {
70 if !parent.as_os_str().is_empty() {
71 std::fs::create_dir_all(parent)?;
72 }
73 }
74 let listener = UnixListener::bind(&socket_path)?;
75 Ok(Self {
76 listener,
77 socket_path,
78 sessions: Arc::new(Mutex::new(HashMap::new())),
79 push_table: Arc::new(Mutex::new(HashMap::new())),
80 last_matches: Arc::new(Mutex::new(HashMap::new())),
81 config,
82 })
83 }
84
85 /// Devuelve la ruta del socket (útil para clientes en el mismo proceso).
86 pub fn socket_path(&self) -> &Path {
87 &self.socket_path
88 }
89
90 /// Vista compartida del registro de sesiones — útil para el Init/Admin
91 /// para inspeccionar quién está conectado.
92 pub fn sessions(&self) -> SessionRegistry {
93 self.sessions.clone()
94 }
95
96 /// Acepta UNA conexión, devuelve la `Session` lista para `handle()`.
97 /// No corre el handler — eso es responsabilidad del llamante.
98 pub async fn accept_one(&self) -> std::io::Result<Session> {
99 let (stream, _addr) = self.listener.accept().await?;
100 Ok(Session {
101 stream,
102 sessions: self.sessions.clone(),
103 push_table: self.push_table.clone(),
104 last_matches: self.last_matches.clone(),
105 config: self.config.clone(),
106 })
107 }
108
109 /// Loop de aceptación: cada conexión se despacha en una task separada.
110 /// Vive hasta que el listener falle o el caller drop el future.
111 pub async fn run(self) -> std::io::Result<()> {
112 loop {
113 let session = self.accept_one().await?;
114 tokio::spawn(async move {
115 if let Err(e) = session.handle().await {
116 warn!(error = %e, "session terminó con error");
117 }
118 });
119 }
120 }
121}
122
123impl Drop for Server {
124 fn drop(&mut self) {
125 // Limpieza best-effort del socket. Si falla, log y seguir.
126 if let Err(e) = std::fs::remove_file(&self.socket_path) {
127 if e.kind() != std::io::ErrorKind::NotFound {
128 warn!(path = %self.socket_path.display(), error = %e, "no se pudo limpiar socket");
129 }
130 }
131 }
132}
133
134/// Conexión individual aceptada por el servidor.
135pub struct Session {
136 stream: UnixStream,
137 sessions: SessionRegistry,
138 push_table: SessionTxTable,
139 last_matches: LastMatches,
140 config: ServerConfig,
141}
142
143impl Session {
144 /// Procesa la conexión hasta `Farewell` o EOF: handshake + loop de pings.
145 /// Garantiza cleanup (sessions + broker) sin importar la rama de salida.
146 pub async fn handle(mut self) -> std::io::Result<()> {
147 let session_id = match self.do_handshake().await? {
148 Some(id) => id,
149 None => return Ok(()), // Hello rechazado, no se registró nada
150 };
151
152 let result = self.run_post_handshake(session_id).await;
153 self.cleanup(session_id).await;
154 result
155 }
156
157 async fn run_post_handshake(&mut self, session_id: SessionId) -> std::io::Result<()> {
158 // Canal por donde el server inyecta frames push (MatchEvent, etc.).
159 let (tx, mut rx) = mpsc::channel::<Frame>(PUSH_CHANNEL_CAPACITY);
160 self.push_table.lock().await.insert(session_id, tx);
161
162 // Tras registrar el canal, recomputar matches y emitir diffs a
163 // todas las sesiones afectadas (incluida ésta, si tiene inputs).
164 self.broadcast_match_diffs().await;
165
166 loop {
167 tokio::select! {
168 // Frame entrante del cliente.
169 res = read_frame(&mut self.stream) => {
170 match res {
171 Ok(frame) => {
172 if !self.handle_inbound_frame(session_id, frame).await? {
173 return Ok(());
174 }
175 }
176 Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
177 debug!(session = %session_id, "cliente cerró sin Farewell");
178 return Ok(());
179 }
180 Err(e) => return Err(e),
181 }
182 }
183 // Frame push desde el server (MatchEvent).
184 Some(frame) = rx.recv() => {
185 write_frame(&mut self.stream, &frame).await?;
186 }
187 }
188 }
189 }
190
191 /// Maneja un frame entrante. Devuelve `Ok(false)` si la sesión debe
192 /// cerrarse limpiamente (Farewell con session-id correcto).
193 async fn handle_inbound_frame(
194 &mut self,
195 session_id: SessionId,
196 frame: Frame,
197 ) -> std::io::Result<bool> {
198 match frame {
199 Frame::Ping(Ping { session }) if session == session_id => {
200 let pong = Pong {
201 timestamp_ms: now_ms(),
202 };
203 write_frame(&mut self.stream, &Frame::Pong(pong)).await?;
204 Ok(true)
205 }
206 Frame::Ping(_) => {
207 write_frame(
208 &mut self.stream,
209 &Frame::Error(HandshakeError::Unauthorized(
210 "session-id no coincide".into(),
211 )),
212 )
213 .await?;
214 Ok(true)
215 }
216 Frame::Farewell(Farewell { session }) if session == session_id => Ok(false),
217 Frame::Farewell(_) => {
218 write_frame(
219 &mut self.stream,
220 &Frame::Error(HandshakeError::Unauthorized(
221 "session-id no coincide".into(),
222 )),
223 )
224 .await?;
225 Ok(true)
226 }
227 _ => {
228 write_frame(
229 &mut self.stream,
230 &Frame::Error(HandshakeError::Rejected(
231 "frame inesperado tras handshake".into(),
232 )),
233 )
234 .await?;
235 Ok(true)
236 }
237 }
238 }
239
240 /// Limpieza atómica de TRES vistas: registro de sesiones + broker +
241 /// canal push. Se ejecuta tanto si la sesión cierra por Farewell, EOF,
242 /// o error. Tras desregistrar, emite diffs a las sesiones que perdieron
243 /// el match contra ésta.
244 async fn cleanup(&self, session_id: SessionId) {
245 self.sessions.lock().await.remove(&session_id);
246 self.push_table.lock().await.remove(&session_id);
247 self.last_matches.lock().await.remove(&session_id);
248 if let Some(broker) = &self.config.broker {
249 broker.lock().await.unregister(session_id);
250 }
251 self.broadcast_match_diffs().await;
252 }
253
254 /// Recomputa los matches para todas las sesiones registradas y empuja
255 /// `MatchEvent::Available` / `MatchEvent::Lost` por las que cambiaron
256 /// respecto al último estado conocido.
257 ///
258 /// Se llama tras cada `register_session` y `cleanup`. Las inserciones
259 /// al canal usan `try_send` (non-blocking); si el cliente está lento
260 /// y se llena el buffer, los eventos extra se pierden — es ephemeral
261 /// y el cliente puede re-consultar el estado vía `brahman-status`.
262 async fn broadcast_match_diffs(&self) {
263 let broker = match &self.config.broker {
264 Some(b) => b,
265 None => return,
266 };
267
268 let b = broker.lock().await;
269 let push_table = self.push_table.lock().await;
270 let mut last = self.last_matches.lock().await;
271
272 debug!(
273 target: "brahman_handshake::broadcast",
274 cards = b.len(),
275 push_subscribers = push_table.len(),
276 "broadcast_match_diffs"
277 );
278
279 // Snapshot de cards para no tener que sostener el lock del broker.
280 let cards: Vec<_> = b.cards().cloned().collect();
281
282 for cons in &cards {
283 let cons_session = cons.session;
284 let tx = match push_table.get(&cons_session) {
285 Some(tx) => tx,
286 None => continue, // todavía no tiene canal push
287 };
288 let cons_last = last.entry(cons_session).or_default();
289
290 for input in &cons.inputs {
291 let new_match = b.find_producer_for(cons_session, &input.name);
292 let new_endpoint = new_match.as_ref().map(|m| m.producer.clone());
293 let old_endpoint = cons_last.get(&input.name).cloned();
294
295 if old_endpoint == new_endpoint {
296 continue;
297 }
298
299 if let Some(m) = &new_match {
300 // Resolvemos el service_socket del productor desde
301 // la BrokeredCard; pasarlo en el evento permite al
302 // consumer conectar directo sin discovery extra.
303 let producer_service_socket = b
304 .cards()
305 .find(|c| c.session == m.producer.session)
306 .and_then(|c| c.service_socket.clone());
307 let event = MatchEvent {
308 kind: MatchEventKind::Available,
309 consumer_flow: input.name.clone(),
310 producer_session: m.producer.session,
311 producer_label: m.producer_label.clone(),
312 producer_flow: m.producer.flow_name.clone(),
313 ty: m.ty.clone(),
314 via: m.via,
315 pinned: m.pinned,
316 producer_service_socket,
317 };
318 let send_res = tx.try_send(Frame::MatchEvent(event));
319 debug!(
320 target: "brahman_handshake::broadcast",
321 consumer = %cons_session,
322 flow = %input.name,
323 producer = %m.producer_label,
324 result = ?send_res.as_ref().map(|_| "ok").unwrap_or_else(|e| match e {
325 tokio::sync::mpsc::error::TrySendError::Full(_) => "full",
326 tokio::sync::mpsc::error::TrySendError::Closed(_) => "closed",
327 }),
328 "Available pushed"
329 );
330 } else {
331 // Tenía match, ahora no.
332 let event = MatchEvent {
333 kind: MatchEventKind::Lost,
334 consumer_flow: input.name.clone(),
335 producer_session: Ulid::nil(),
336 producer_label: String::new(),
337 producer_flow: String::new(),
338 ty: input.ty.clone(),
339 via: brahman_broker::MatchStrategy::Exact,
340 pinned: false,
341 producer_service_socket: None,
342 };
343 let _ = tx.try_send(Frame::MatchEvent(event));
344 }
345
346 if let Some(ep) = new_endpoint {
347 cons_last.insert(input.name.clone(), ep);
348 } else {
349 cons_last.remove(&input.name);
350 }
351 }
352 }
353 }
354
355 /// Lee el Hello, valida, registra la sesión y emite HelloAck.
356 /// Devuelve `Some(session_id)` si el handshake fue exitoso.
357 async fn do_handshake(&mut self) -> std::io::Result<Option<SessionId>> {
358 let frame = read_frame(&mut self.stream).await?;
359 let hello = match frame {
360 Frame::Hello(h) => h,
361 _ => {
362 write_frame(
363 &mut self.stream,
364 &Frame::Error(HandshakeError::Rejected(
365 "primer frame debe ser Hello".into(),
366 )),
367 )
368 .await?;
369 return Ok(None);
370 }
371 };
372
373 if let Some(err) = self.validate_hello(&hello) {
374 write_frame(&mut self.stream, &Frame::Error(err)).await?;
375 return Ok(None);
376 }
377
378 let session_id = Ulid::new();
379 // WireCard → Card: extensiones quedan vacías post-wire (es el contrato).
380 let card: Card = hello.card.into();
381 self.register_session(session_id, card, hello.wit).await;
382
383 let ack = HelloAck {
384 server_version: crate::HANDSHAKE_VERSION.to_string(),
385 protocol_version: brahman_card::PROTOCOL_VERSION.to_string(),
386 session: session_id,
387 init_attached: self.config.init_attached,
388 };
389 write_frame(&mut self.stream, &Frame::HelloAck(ack)).await?;
390 debug!(session = %session_id, "handshake completado");
391 Ok(Some(session_id))
392 }
393
394 /// Indexa la sesión: ResolvedCard en sessions + Card en broker (si hay).
395 /// Si `wit` está presente, el módulo se registra como "consciente".
396 async fn register_session(
397 &self,
398 session_id: SessionId,
399 card: Card,
400 wit: Option<WitInterface>,
401 ) {
402 if let Some(broker) = &self.config.broker {
403 broker
404 .lock()
405 .await
406 .register(session_id, &card, wit.clone());
407 }
408 let resolved = match wit {
409 Some(w) => ResolvedCard::from_conscious(card, w),
410 None => ResolvedCard::from_agnostic(card),
411 };
412 self.sessions.lock().await.insert(session_id, resolved);
413 }
414
415 /// Validaciones que el servidor aplica al Hello del cliente.
416 fn validate_hello(&self, hello: &Hello) -> Option<HandshakeError> {
417 if hello.schema_version != CARD_SCHEMA_VERSION {
418 return Some(HandshakeError::SchemaMismatch {
419 client: hello.schema_version,
420 server: CARD_SCHEMA_VERSION,
421 });
422 }
423 if hello.protocol_version != brahman_card::PROTOCOL_VERSION {
424 return Some(HandshakeError::ProtocolMismatch(format!(
425 "cliente={}, servidor={}",
426 hello.protocol_version,
427 brahman_card::PROTOCOL_VERSION
428 )));
429 }
430 // Validamos contra Card (la rica) — convertir es barato y centraliza
431 // la lógica de validación en un solo lugar.
432 let as_card: Card = Card::from(hello.card.clone());
433 if let Err(e) = as_card.validate() {
434 return Some(HandshakeError::InvalidCard(e.to_string()));
435 }
436 None
437 }
438}
439
440fn now_ms() -> u64 {
441 SystemTime::now()
442 .duration_since(UNIX_EPOCH)
443 .map(|d| d.as_millis() as u64)
444 .unwrap_or(0)
445}