feat(renaser): Fase 20 — Akasha Over Ether (grafo distribuido)
1// =============================================================================
2// renaser :: kernel/src/akasha — Fase 20 :: Akasha Over Ether
3// -----------------------------------------------------------------------------
4// El servicio del kernel que habla AoE: el respondedor de Akasha Over Ether.
5// Tres oficios:
6//
7// 1. Drena la cola RX del dispositivo de red y DEMULTIPLEXA cada frame:
8// - Si su EtherType es `0x88B5` y el payload deserializa como un
9// `MensajeAkasha`, lo procesa en el kernel.
10// - Cualquier otro frame se encola hacia el userspace: las apps lo
11// recibiran via `sys_net_recibir` como hasta ahora.
12// 2. Atiende los mensajes AoE:
13// - `SolicitarObjeto(id)` → si tenemos `id` en el grafo,
14// respondemos con `ProveedorObjeto`.
15// - `ProveedorObjeto(id, p)` → si la integridad cuadra, lo absorbemos
16// al grafo local.
17// - `AnunciarRaiz(id)` → se contabiliza y, si no tenemos el
18// nodo, se le solicita al emisor.
19// 3. Difunde periodicamente nuestra raiz del grafo (`AnunciarRaiz` con el
20// hash del manifiesto) — el faro que delata nuestra presencia.
21//
22// Con esto, el grafo de objetos —direccionado por contenido, ya BLAKE3— deja
23// de ser una propiedad local del disco y empieza a ser una propiedad
24// distribuida del cable. Tres frames bastan, sin TCP, sin IP, sin DNS.
25// =============================================================================
26
27use alloc::collections::VecDeque;
28use alloc::vec::Vec;
29use core::fmt::Write;
30use core::sync::atomic::{AtomicU64, Ordering};
31
32use spin::{Mutex, Once};
33
34use crate::async_system::reloj;
35
36use akasha::{
37 analizar_frame, componer_frame, ErrorAkasha, Mac, MensajeAkasha, MAC_BROADCAST,
38};
39use formato::Hash;
40
41use crate::almacen;
42use crate::baliza;
43use crate::drivers::red;
44
45// =============================================================================
46// Cola del userspace — frames NO-AoE en espera de `sys_net_recibir`
47// =============================================================================
48
49/// Cuantos frames como mucho retenemos en la cola del userspace. Cota dura:
50/// si el userspace se duerme, los frames mas antiguos se pierden antes que
51/// la cola se desborde.
52const PROFUNDIDAD_COLA_USUARIO: usize = 64;
53
54/// La cola FIFO de frames que NO son AoE y aguardan a `sys_net_recibir`. Cada
55/// frame se copia tal cual lo entrega el driver; el userspace ve la cabecera
56/// Ethernet completa, como antes de la Fase 20.
57static COLA_USUARIO: Mutex<VecDeque<Vec<u8>>> = Mutex::new(VecDeque::new());
58
59// =============================================================================
60// Contadores — la voz que la barra y el diagnostico leen
61// =============================================================================
62
63/// Frames AoE procesados en RX, por variante.
64static RX_SOLICITUDES: AtomicU64 = AtomicU64::new(0);
65static RX_PROVEEDORES: AtomicU64 = AtomicU64::new(0);
66static RX_ANUNCIOS: AtomicU64 = AtomicU64::new(0);
67
68/// Frames AoE emitidos en TX, por variante.
69static TX_SOLICITUDES: AtomicU64 = AtomicU64::new(0);
70static TX_PROVEEDORES: AtomicU64 = AtomicU64::new(0);
71static TX_ANUNCIOS: AtomicU64 = AtomicU64::new(0);
72
73/// Frames descartados en RX por ser basura (payload AoE invalido).
74static RX_DESCARTADOS: AtomicU64 = AtomicU64::new(0);
75
76/// Frames no-AoE encolados hacia el userspace.
77static USUARIO_ENCOLADOS: AtomicU64 = AtomicU64::new(0);
78/// Frames no-AoE descartados por desbordamiento de la cola del userspace.
79static USUARIO_DESBORDADOS: AtomicU64 = AtomicU64::new(0);
80
81/// La MAC con la que firmamos los frames AoE. La cachea `montar`.
82static NUESTRA_MAC: Once<Mac> = Once::new();
83
84// =============================================================================
85// Montaje
86// =============================================================================
87
88/// Anota la MAC del dispositivo de red. La llama el orquestador cuando el
89/// driver entrega su `Mac` tras `red::montar`.
90pub fn montar(mac: Mac) {
91 NUESTRA_MAC.call_once(|| mac);
92}
93
94// =============================================================================
95// Demultiplexor — el oficio numero 1 de la Fase 20
96// =============================================================================
97
98/// Drena la cola RX del dispositivo y demultiplexa cada frame:
99/// - Frames AoE (EtherType `0x88B5` con payload valido) → `procesar`.
100/// - Cualquier otro frame → cola del userspace (lo recogera `sys_net_recibir`).
101///
102/// Llamada en CADA fotograma desde la tarea cooperativa de red. El cerrojo
103/// del driver virtio-net se libera entre frame y frame; el del userspace solo
104/// se toma para empujar, sin solapar con el del driver.
105pub fn drenar_y_demultiplexar() {
106 let mac = match NUESTRA_MAC.get().copied() {
107 Some(m) => m,
108 None => return,
109 };
110 red::drenar_rx(|frame| {
111 // Intentar analizar como AoE. Si el EtherType cuadra Y el payload
112 // deserializa, procesamos en el kernel; el frame NO viaja al
113 // userspace —el protocolo es asunto del nucleo—.
114 match analizar_frame(frame) {
115 Ok((origen, mensaje)) => procesar(mensaje, origen, mac),
116 // EtherType ajeno o frame truncado: a la cola del userspace.
117 Err(ErrorAkasha::EtherTypeAjeno) | Err(ErrorAkasha::FrameDemasiadoCorto) => {
118 encolar_para_usuario(frame)
119 }
120 // EtherType nuestro pero payload no-postcard: el ejemplo canonico
121 // es el saludo en texto plano de `pregon`. NO es Akasha legitimo,
122 // pero tampoco basura del cable — es un protocolo userspace que
123 // comparte EtherType. Lo contamos y lo dejamos pasar al userspace
124 // para que el app destinatario (si lo hay) lo recoja.
125 Err(ErrorAkasha::PayloadInvalido) | Err(ErrorAkasha::PayloadDemasiadoLargo) => {
126 RX_DESCARTADOS.fetch_add(1, Ordering::Relaxed);
127 encolar_para_usuario(frame);
128 }
129 }
130 });
131}
132
133/// Encola un frame entero hacia el userspace. Si la cola esta llena, descarta
134/// el mas antiguo —preferimos perder el pasado a quedar sordos del futuro—.
135fn encolar_para_usuario(frame: &[u8]) {
136 let mut cola = COLA_USUARIO.lock();
137 if cola.len() >= PROFUNDIDAD_COLA_USUARIO {
138 cola.pop_front();
139 USUARIO_DESBORDADOS.fetch_add(1, Ordering::Relaxed);
140 }
141 cola.push_back(frame.to_vec());
142 USUARIO_ENCOLADOS.fetch_add(1, Ordering::Relaxed);
143}
144
145// =============================================================================
146// Interfaz que ve `sys_net_recibir`
147// =============================================================================
148
149/// Saca UN frame de la cola del userspace hacia `buf`. Devuelve los bytes
150/// copiados (acotados por `buf.len()`), o `0` si no hay frame pendiente. El
151/// reemplazo natural de `red::recibir_en` para el lado WASM —ahora el
152/// userspace ya no compite con el kernel por la cola RX del dispositivo—.
153pub fn pop_usuario(buf: &mut [u8]) -> usize {
154 let frame = match COLA_USUARIO.lock().pop_front() {
155 Some(f) => f,
156 None => return 0,
157 };
158 let n = frame.len().min(buf.len());
159 buf[..n].copy_from_slice(&frame[..n]);
160 n
161}
162
163// =============================================================================
164// Atencion de mensajes — el oficio numero 2
165// =============================================================================
166
167/// Procesa UN mensaje AoE entrante. Contabiliza, traza y —si toca— responde.
168fn procesar(mensaje: MensajeAkasha, origen: Mac, nuestra: Mac) {
169 match mensaje {
170 MensajeAkasha::SolicitarObjeto(id) => {
171 RX_SOLICITUDES.fetch_add(1, Ordering::Relaxed);
172 atender_solicitud(id, origen, nuestra);
173 }
174 MensajeAkasha::ProveedorObjeto(id, payload) => {
175 RX_PROVEEDORES.fetch_add(1, Ordering::Relaxed);
176 absorber_proveedor(id, &payload, origen);
177 }
178 MensajeAkasha::AnunciarRaiz(id) => {
179 RX_ANUNCIOS.fetch_add(1, Ordering::Relaxed);
180 atender_anuncio(id, origen, nuestra);
181 }
182 }
183}
184
185/// Si tenemos el objeto que nos piden, le respondemos al solicitante (unicast)
186/// con un `ProveedorObjeto` que carga la forma serializada del nodo. La
187/// integridad se preserva por construccion: `payload` es exactamente la
188/// secuencia que rehashea al `id` pedido. Si no lo tenemos, no decimos nada
189/// —AoE no tiene «not found»; un par puede preguntarle a otro—.
190fn atender_solicitud(id: Hash, origen: Mac, nuestra: Mac) {
191 let objeto = match almacen::recuperar(&id) {
192 Ok(Some(o)) => o,
193 Ok(None) => {
194 let _ = writeln!(
195 baliza::Serie,
196 "akasha :: solicitud rechazada (objeto ausente) :: {}",
197 FormatoHash(&id)
198 );
199 return;
200 }
201 Err(motivo) => {
202 let _ = writeln!(baliza::Serie, "akasha :: solicitud fallida :: {motivo}");
203 return;
204 }
205 };
206 let payload = match objeto.serializar() {
207 Ok(p) => p,
208 Err(_) => return,
209 };
210 // Defensa en profundidad: el rehash DEBE coincidir. postcard es canonico
211 // pero verificamos antes de poner algo en el cable que dice ser `id`.
212 if formato::hash(&payload) != id {
213 let _ = writeln!(
214 baliza::Serie,
215 "akasha :: rehash no coincide al servir :: descartado"
216 );
217 return;
218 }
219 let mensaje = MensajeAkasha::ProveedorObjeto(id, payload);
220 if enviar(&mensaje, nuestra, origen).is_ok() {
221 TX_PROVEEDORES.fetch_add(1, Ordering::Relaxed);
222 let _ = writeln!(
223 baliza::Serie,
224 "akasha :: PROVEEDOR enviado :: {} -> {}",
225 FormatoHash(&id),
226 FormatoMac(&origen)
227 );
228 }
229}
230
231/// Acepta un objeto venido del cable. Verifica que su forma serializada
232/// rehashea al `id` que el remitente afirma; si cuadra, lo deposita en el
233/// grafo local; si no, lo descarta con una traza. La unica entrada de
234/// integridad esta aqui: el grafo local no admite mentiras.
235fn absorber_proveedor(id: Hash, payload: &[u8], origen: Mac) {
236 if formato::hash(payload) != id {
237 let _ = writeln!(
238 baliza::Serie,
239 "akasha :: proveedor rechazado (rehash no coincide) :: src={}",
240 FormatoMac(&origen)
241 );
242 return;
243 }
244 let objeto = match formato::Objeto::deserializar(payload) {
245 Ok(o) => o,
246 Err(motivo) => {
247 let _ = writeln!(
248 baliza::Serie,
249 "akasha :: proveedor no deserializable :: {motivo}"
250 );
251 return;
252 }
253 };
254 match almacen::almacenar(objeto.datos, objeto.hijos) {
255 Ok(hash) => {
256 let _ = writeln!(
257 baliza::Serie,
258 "akasha :: PROVEEDOR absorbido :: {} <- {}",
259 FormatoHash(&hash),
260 FormatoMac(&origen)
261 );
262 }
263 Err(motivo) => {
264 let _ = writeln!(
265 baliza::Serie,
266 "akasha :: no se pudo absorber proveedor :: {motivo}"
267 );
268 }
269 }
270}
271
272/// Un par nos anuncio su raiz. Si la conocemos, no hay nada que hacer; si no,
273/// le pedimos el nodo: el `AnunciarRaiz` es el faro que basta para iniciar la
274/// replicacion.
275fn atender_anuncio(id: Hash, origen: Mac, nuestra: Mac) {
276 // ¿Lo tenemos ya? Entonces nada que pedir.
277 if let Ok(Some(_)) = almacen::recuperar(&id) {
278 let _ = writeln!(
279 baliza::Serie,
280 "akasha :: anuncio reconocido (ya lo tenemos) :: {} de {}",
281 FormatoHash(&id),
282 FormatoMac(&origen)
283 );
284 return;
285 }
286 // No lo tenemos — pedirlo al emisor en unicast.
287 let mensaje = MensajeAkasha::SolicitarObjeto(id);
288 if enviar(&mensaje, nuestra, origen).is_ok() {
289 TX_SOLICITUDES.fetch_add(1, Ordering::Relaxed);
290 let _ = writeln!(
291 baliza::Serie,
292 "akasha :: SOLICITUD enviada :: {} -> {}",
293 FormatoHash(&id),
294 FormatoMac(&origen)
295 );
296 }
297}
298
299// =============================================================================
300// Difusion periodica de nuestra raiz — el oficio numero 3
301// =============================================================================
302
303/// Intervalo entre faros AoE consecutivos, en milisegundos. 5 s es un
304/// compromiso conservador: lo bastante frecuente para descubrir vecinos
305/// nuevos sin saturar la red de capa-2 con anuncios.
306const INTERVALO_FARO_MS: u64 = 5_000;
307
308/// Marca del reloj monotono (`reloj::milisegundos`) en la que se difundira el
309/// proximo faro. La inicializamos a `0` — la primera difusion ocurre cuanto
310/// antes pase `tic_compositor` con el manifiesto montado—.
311static PROXIMO_FARO_MS: AtomicU64 = AtomicU64::new(0);
312
313/// Punto de entrada que el tic del compositor llama una vez por fotograma.
314/// Junta los dos oficios AoE que viven en linea con el latido del escritorio:
315/// - drenar la cola RX del dispositivo y demultiplexar Akasha vs userspace,
316/// - difundir nuestra raiz cada `INTERVALO_FARO_MS` segun reloj monotono.
317///
318/// El compositor late de forma fiable (avanza el reloj de la barra cada
319/// segundo, prueba indirecta de que su tarea se atiende sin atascos); por
320/// eso elegimos su tic como portador. La cadencia del faro se mide contra
321/// el reloj monotono y NO contra los awaits, asi el ritmo del faro es
322/// independiente de cuanto trabajo del reactor consume cada vuelta.
323pub fn tic_compositor() {
324 drenar_y_demultiplexar();
325 let ahora = reloj::milisegundos();
326 let proximo = PROXIMO_FARO_MS.load(Ordering::Relaxed);
327 if ahora >= proximo {
328 difundir_raiz();
329 PROXIMO_FARO_MS.store(ahora + INTERVALO_FARO_MS, Ordering::Relaxed);
330 }
331}
332
333/// Anuncia, por broadcast, el hash del manifiesto actual. Es el faro de
334/// renaser: quien escuche en la red de capa-2 sabra de nuestra existencia y
335/// del nodo raiz del grafo. Si aun no hay manifiesto anclado, no se difunde
336/// nada — el silencio es preferible a un faro vacio.
337pub fn difundir_raiz() {
338 let nuestra = match NUESTRA_MAC.get().copied() {
339 Some(m) => m,
340 None => return,
341 };
342 let id = match almacen::manifiesto() {
343 Some(h) => h,
344 None => return,
345 };
346 let mensaje = MensajeAkasha::AnunciarRaiz(id);
347 if enviar(&mensaje, nuestra, MAC_BROADCAST).is_ok() {
348 TX_ANUNCIOS.fetch_add(1, Ordering::Relaxed);
349 let _ = writeln!(
350 baliza::Serie,
351 "akasha :: ANUNCIO emitido :: raiz={}",
352 FormatoHash(&id)
353 );
354 }
355}
356
357// =============================================================================
358// Helpers internos
359// =============================================================================
360
361/// Compone un frame AoE y lo entrega al driver. Punto de salida unico de TX.
362fn enviar(mensaje: &MensajeAkasha, src: Mac, dst: Mac) -> Result<(), ()> {
363 let frame = componer_frame(src, dst, mensaje).map_err(|_| ())?;
364 red::enviar(&frame).map_err(|motivo| {
365 let _ = writeln!(baliza::Serie, "akasha :: envio fallido :: {motivo}");
366 })
367}
368
369// =============================================================================
370// Lectores de estado — los expone la barra de tareas (futuro indicador) y el
371// diagnostico de COM1
372// =============================================================================
373
374/// Resumen de actividad AoE, en una sola lectura coherente. Los enteros se
375/// leen `Relaxed`: el resumen es informativo, no transaccional.
376#[allow(dead_code)]
377#[derive(Clone, Copy, Debug)]
378pub struct ResumenAkasha {
379 pub rx_solicitudes: u64,
380 pub rx_proveedores: u64,
381 pub rx_anuncios: u64,
382 pub tx_solicitudes: u64,
383 pub tx_proveedores: u64,
384 pub tx_anuncios: u64,
385 pub rx_descartados: u64,
386 pub usuario_encolados: u64,
387 pub usuario_desbordados: u64,
388}
389
390#[allow(dead_code)]
391pub fn resumen() -> ResumenAkasha {
392 ResumenAkasha {
393 rx_solicitudes: RX_SOLICITUDES.load(Ordering::Relaxed),
394 rx_proveedores: RX_PROVEEDORES.load(Ordering::Relaxed),
395 rx_anuncios: RX_ANUNCIOS.load(Ordering::Relaxed),
396 tx_solicitudes: TX_SOLICITUDES.load(Ordering::Relaxed),
397 tx_proveedores: TX_PROVEEDORES.load(Ordering::Relaxed),
398 tx_anuncios: TX_ANUNCIOS.load(Ordering::Relaxed),
399 rx_descartados: RX_DESCARTADOS.load(Ordering::Relaxed),
400 usuario_encolados: USUARIO_ENCOLADOS.load(Ordering::Relaxed),
401 usuario_desbordados: USUARIO_DESBORDADOS.load(Ordering::Relaxed),
402 }
403}
404
405// =============================================================================
406// Formateadores cortos para la traza de COM1 — sin allocations innecesarias
407// =============================================================================
408
409struct FormatoHash<'h>(&'h Hash);
410impl core::fmt::Display for FormatoHash<'_> {
411 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
412 // Primeros 8 bytes en hex — suficiente para distinguir en una traza.
413 for b in &self.0[..8] {
414 write!(f, "{b:02x}")?;
415 }
416 write!(f, "..")
417 }
418}
419
420struct FormatoMac<'m>(&'m Mac);
421impl core::fmt::Display for FormatoMac<'_> {
422 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
423 for (i, b) in self.0.iter().enumerate() {
424 if i > 0 {
425 write!(f, ":")?;
426 }
427 write!(f, "{b:02x}")?;
428 }
429 Ok(())
430 }
431}