⎇ 5091106c2177cd4d337fb5f4e9b93eeb15daa4ae
crates/core/brahman-broker/src/lib.rs
Ñawpa willay
Pi imata llamkarqan
ch'uya
feat(core): brahman-broker — matching híbrido productor↔consumidor
1//! `brahman-broker` — empareja productores y consumidores por tipo de flujo.
2//!
3//! El broker indexa [`brahman_card::Card`]s registradas por `SessionId` y,
4//! para cada `flow.input` de un consumidor, busca el `flow.output`
5//! compatible de mejor calidad entre los demás. Tres ejes:
6//!
7//! 1. **Estrategia de matching** ([`MatchStrategy`]):
8//! - `Exact`: igualdad estricta de [`brahman_card::TypeRef`].
9//! - `Structural`: misma forma (mismo `package` + `name` para Wit;
10//! ignora `interface`).
11//! - `ExactThenStructural`: prefiere exact; cae en structural si no hay.
12//!
13//! 2. **Override `pin_to`**: si el consumidor declara `pin_to = "label"`,
14//! el broker prefiere productores cuya Card tenga ese `label` (siempre
15//! que el tipo siga matcheando). Si la pista no resuelve, cae en
16//! matching por tipo normal.
17//!
18//! 3. **Prioridad**: empate de tipo se resuelve por
19//! [`brahman_card::Priority`] del productor (mayor gana). Empate de
20//! prioridad se resuelve lexicográficamente por `label` (estable y
21//! determinista).
22//!
23//! El broker es **stateless w.r.t. routes**: cada `find_producer_for` o
24//! `all_matches` se calcula bajo demanda. La única persistencia es el
25//! índice de Cards registradas. Esto permite re-evaluar matches cuando
26//! cambia el set sin invalidar caches.
27
28#![forbid(unsafe_code)]
29#![warn(rust_2018_idioms)]
30
31use std::collections::BTreeMap;
32
33use brahman_card::{Card, Flow, Priority, TypeRef};
34use serde::{Deserialize, Serialize};
35use ulid::Ulid;
36
37/// Identificador de sesión emitido por el handshake. Idéntico al usado por
38/// `brahman-handshake` (no es un re-export para evitar la dependencia).
39pub type SessionId = Ulid;
40
41/// Estrategia de matching de tipos.
42#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
43#[serde(rename_all = "kebab-case")]
44pub enum MatchStrategy {
45 /// Igualdad estricta de `TypeRef`.
46 Exact,
47 /// Misma forma: para `Wit`, mismo `package` + `name`; para
48 /// `Primitive`, mismo `name`.
49 Structural,
50 /// Híbrido: intenta `Exact` primero; si no matchea, `Structural`.
51 /// Reporta cuál estrategia ganó en [`Match::via`].
52 #[default]
53 ExactThenStructural,
54}
55
56/// Configuración del broker.
57#[derive(Debug, Clone, Default)]
58pub struct BrokerConfig {
59 pub strategy: MatchStrategy,
60}
61
62/// Vista mínima de una Card que el broker necesita.
63#[derive(Debug, Clone)]
64pub struct BrokeredCard {
65 pub session: SessionId,
66 pub label: String,
67 pub priority: Priority,
68 pub inputs: Vec<Flow>,
69 pub outputs: Vec<Flow>,
70}
71
72impl BrokeredCard {
73 fn from_card(session: SessionId, card: &Card) -> Self {
74 Self {
75 session,
76 label: card.label.clone(),
77 priority: card.priority,
78 inputs: card.flow.input.clone(),
79 outputs: card.flow.output.clone(),
80 }
81 }
82}
83
84/// Punto extremo de un flujo: qué sesión + nombre del flow dentro de su Card.
85#[derive(Debug, Clone, PartialEq, Eq, Hash)]
86pub struct Endpoint {
87 pub session: SessionId,
88 pub flow_name: String,
89}
90
91/// Match concreto entre un consumidor y un productor.
92#[derive(Debug, Clone)]
93pub struct Match {
94 pub consumer: Endpoint,
95 pub consumer_label: String,
96 pub producer: Endpoint,
97 pub producer_label: String,
98 /// Tipo del flow (lado consumidor — lado productor coincide en
99 /// estrategia Exact, puede diferir en `interface` en Structural).
100 pub ty: TypeRef,
101 /// Estrategia que efectivamente matcheó.
102 pub via: MatchStrategy,
103 /// `true` si el match fue resuelto por `pin_to` y no por type-search.
104 pub pinned: bool,
105}
106
107// =====================================================================
108// Broker
109// =====================================================================
110
111/// El broker. Registra Cards por SessionId, computa matches bajo demanda.
112#[derive(Debug, Clone, Default)]
113pub struct Broker {
114 cards: BTreeMap<SessionId, BrokeredCard>,
115 config: BrokerConfig,
116}
117
118impl Broker {
119 pub fn new(config: BrokerConfig) -> Self {
120 Self {
121 cards: BTreeMap::new(),
122 config,
123 }
124 }
125
126 /// Registra una Card. Devuelve `Some(prev)` si reemplazó una existente.
127 pub fn register(&mut self, session: SessionId, card: &Card) -> Option<BrokeredCard> {
128 self.cards.insert(session, BrokeredCard::from_card(session, card))
129 }
130
131 /// Quita una Card por sesión.
132 pub fn unregister(&mut self, session: SessionId) -> Option<BrokeredCard> {
133 self.cards.remove(&session)
134 }
135
136 /// Cardinalidad del registro.
137 pub fn len(&self) -> usize {
138 self.cards.len()
139 }
140
141 pub fn is_empty(&self) -> bool {
142 self.cards.is_empty()
143 }
144
145 /// Iterador sobre las sesiones registradas.
146 pub fn sessions(&self) -> impl Iterator<Item = SessionId> + '_ {
147 self.cards.keys().copied()
148 }
149
150 /// Busca el mejor productor para un input específico de un consumidor.
151 ///
152 /// Algoritmo:
153 /// 1. Resuelve el flow input en el consumidor.
154 /// 2. Si tiene `pin_to`, prefiere productores con ese `label` que
155 /// matcheen el tipo (cualquier estrategia configurada).
156 /// 3. Si no hay pin_to o la pista falló, escanea todos los outputs
157 /// de las otras Cards. Filtra por compatibilidad de tipo.
158 /// 4. Ordena por (priority desc, label asc) y devuelve el primero.
159 pub fn find_producer_for(&self, consumer: SessionId, input_name: &str) -> Option<Match> {
160 let cons = self.cards.get(&consumer)?;
161 let input = cons.inputs.iter().find(|f| f.name == input_name)?;
162
163 // pin_to: short-circuit. Si la pista resuelve, gana.
164 if let Some(pin) = &input.pin_to {
165 for prod in self.cards.values() {
166 if prod.session == consumer || &prod.label != pin {
167 continue;
168 }
169 for out in &prod.outputs {
170 if let Some(via) = self.types_match(&input.ty, &out.ty) {
171 return Some(self.make_match(cons, prod, input, out, via, true));
172 }
173 }
174 }
175 // Fall through: pin no resuelto, type-search general.
176 }
177
178 let mut candidates: Vec<(&BrokeredCard, &Flow, MatchStrategy)> = Vec::new();
179 for prod in self.cards.values() {
180 if prod.session == consumer {
181 continue;
182 }
183 for out in &prod.outputs {
184 if let Some(via) = self.types_match(&input.ty, &out.ty) {
185 candidates.push((prod, out, via));
186 }
187 }
188 }
189
190 candidates.sort_by(|(a, _, _), (b, _, _)| {
191 b.priority
192 .cmp(&a.priority)
193 .then_with(|| a.label.cmp(&b.label))
194 });
195
196 let (prod, out, via) = candidates.into_iter().next()?;
197 Some(self.make_match(cons, prod, input, out, via, false))
198 }
199
200 /// Calcula todos los matches consumer→producer en el set actual.
201 /// Útil para introspección o para que el Admin emita rutas en lote.
202 pub fn all_matches(&self) -> Vec<Match> {
203 let mut out = Vec::new();
204 for cons in self.cards.values() {
205 for input in &cons.inputs {
206 if let Some(m) = self.find_producer_for(cons.session, &input.name) {
207 out.push(m);
208 }
209 }
210 }
211 out
212 }
213
214 fn types_match(&self, consumer_ty: &TypeRef, producer_ty: &TypeRef) -> Option<MatchStrategy> {
215 match self.config.strategy {
216 MatchStrategy::Exact => exact_match(consumer_ty, producer_ty).then_some(MatchStrategy::Exact),
217 MatchStrategy::Structural => {
218 structural_match(consumer_ty, producer_ty).then_some(MatchStrategy::Structural)
219 }
220 MatchStrategy::ExactThenStructural => {
221 if exact_match(consumer_ty, producer_ty) {
222 Some(MatchStrategy::Exact)
223 } else if structural_match(consumer_ty, producer_ty) {
224 Some(MatchStrategy::Structural)
225 } else {
226 None
227 }
228 }
229 }
230 }
231
232 fn make_match(
233 &self,
234 cons: &BrokeredCard,
235 prod: &BrokeredCard,
236 input: &Flow,
237 output: &Flow,
238 via: MatchStrategy,
239 pinned: bool,
240 ) -> Match {
241 Match {
242 consumer: Endpoint {
243 session: cons.session,
244 flow_name: input.name.clone(),
245 },
246 consumer_label: cons.label.clone(),
247 producer: Endpoint {
248 session: prod.session,
249 flow_name: output.name.clone(),
250 },
251 producer_label: prod.label.clone(),
252 ty: input.ty.clone(),
253 via,
254 pinned,
255 }
256 }
257}
258
259// =====================================================================
260// Predicados de matching (libres, testeables aislados)
261// =====================================================================
262
263fn exact_match(a: &TypeRef, b: &TypeRef) -> bool {
264 a == b
265}
266
267fn structural_match(a: &TypeRef, b: &TypeRef) -> bool {
268 match (a, b) {
269 (TypeRef::Primitive { name: na }, TypeRef::Primitive { name: nb }) => na == nb,
270 (
271 TypeRef::Wit {
272 package: pa, name: na, ..
273 },
274 TypeRef::Wit {
275 package: pb, name: nb, ..
276 },
277 ) => pa == pb && na == nb,
278 _ => false,
279 }
280}
281
282// =====================================================================
283// Tests
284// =====================================================================
285
286#[cfg(test)]
287mod tests {
288 use super::*;
289 use brahman_card::{Card, Flows, Payload, Supervision, CARD_SCHEMA_VERSION};
290
291 fn card(label: &str, priority: Priority, flows: Flows) -> Card {
292 Card {
293 schema_version: CARD_SCHEMA_VERSION,
294 id: Ulid::new(),
295 label: label.into(),
296 payload: Payload::Virtual,
297 supervision: Supervision::OneShot,
298 priority,
299 flow: flows,
300 ..Default::default()
301 }
302 }
303
304 fn prim(name: &str) -> TypeRef {
305 TypeRef::Primitive { name: name.into() }
306 }
307
308 fn wit(pkg: &str, iface: Option<&str>, name: &str) -> TypeRef {
309 TypeRef::Wit {
310 package: pkg.into(),
311 interface: iface.map(|s| s.into()),
312 name: name.into(),
313 }
314 }
315
316 fn flow(name: &str, ty: TypeRef, pin: Option<&str>) -> Flow {
317 Flow {
318 name: name.into(),
319 ty,
320 pin_to: pin.map(|s| s.into()),
321 }
322 }
323
324 #[test]
325 fn exact_match_same_typeref() {
326 let mut b = Broker::new(BrokerConfig {
327 strategy: MatchStrategy::Exact,
328 });
329 let producer = card(
330 "dht",
331 Priority::Normal,
332 Flows {
333 input: vec![],
334 output: vec![flow("results", prim("string"), None)],
335 },
336 );
337 let consumer = card(
338 "ui",
339 Priority::Normal,
340 Flows {
341 input: vec![flow("query", prim("string"), None)],
342 output: vec![],
343 },
344 );
345 let s_prod = Ulid::new();
346 let s_cons = Ulid::new();
347 b.register(s_prod, &producer);
348 b.register(s_cons, &consumer);
349
350 let m = b.find_producer_for(s_cons, "query").expect("match");
351 assert_eq!(m.producer_label, "dht");
352 assert_eq!(m.via, MatchStrategy::Exact);
353 assert!(!m.pinned);
354 }
355
356 #[test]
357 fn structural_ignores_interface() {
358 let mut b = Broker::new(BrokerConfig {
359 strategy: MatchStrategy::Structural,
360 });
361 let producer = card(
362 "dht",
363 Priority::Normal,
364 Flows {
365 input: vec![],
366 output: vec![flow(
367 "out",
368 wit("brahman:dht", Some("v1"), "entity-result"),
369 None,
370 )],
371 },
372 );
373 let consumer = card(
374 "ui",
375 Priority::Normal,
376 Flows {
377 input: vec![flow(
378 "in",
379 wit("brahman:dht", Some("v2"), "entity-result"),
380 None,
381 )],
382 output: vec![],
383 },
384 );
385 let s_prod = Ulid::new();
386 let s_cons = Ulid::new();
387 b.register(s_prod, &producer);
388 b.register(s_cons, &consumer);
389
390 let m = b.find_producer_for(s_cons, "in").expect("match");
391 assert_eq!(m.via, MatchStrategy::Structural);
392 }
393
394 #[test]
395 fn exact_strategy_rejects_interface_mismatch() {
396 let mut b = Broker::new(BrokerConfig {
397 strategy: MatchStrategy::Exact,
398 });
399 let producer = card(
400 "dht",
401 Priority::Normal,
402 Flows {
403 input: vec![],
404 output: vec![flow(
405 "out",
406 wit("brahman:dht", Some("v1"), "entity-result"),
407 None,
408 )],
409 },
410 );
411 let consumer = card(
412 "ui",
413 Priority::Normal,
414 Flows {
415 input: vec![flow(
416 "in",
417 wit("brahman:dht", Some("v2"), "entity-result"),
418 None,
419 )],
420 output: vec![],
421 },
422 );
423 b.register(Ulid::new(), &producer);
424 let s_cons = Ulid::new();
425 b.register(s_cons, &consumer);
426
427 assert!(b.find_producer_for(s_cons, "in").is_none());
428 }
429
430 #[test]
431 fn exact_then_structural_prefers_exact() {
432 let mut b = Broker::new(BrokerConfig {
433 strategy: MatchStrategy::ExactThenStructural,
434 });
435 // Productor 1: match estructural (interface diferente)
436 let p_struct = card(
437 "dht-cache",
438 Priority::Normal,
439 Flows {
440 input: vec![],
441 output: vec![flow(
442 "out",
443 wit("brahman:dht", Some("v2"), "entity-result"),
444 None,
445 )],
446 },
447 );
448 // Productor 2: match exact (interface igual)
449 let p_exact = card(
450 "dht",
451 Priority::Normal,
452 Flows {
453 input: vec![],
454 output: vec![flow(
455 "out",
456 wit("brahman:dht", Some("v1"), "entity-result"),
457 None,
458 )],
459 },
460 );
461 let consumer = card(
462 "ui",
463 Priority::Normal,
464 Flows {
465 input: vec![flow(
466 "in",
467 wit("brahman:dht", Some("v1"), "entity-result"),
468 None,
469 )],
470 output: vec![],
471 },
472 );
473 b.register(Ulid::new(), &p_struct);
474 b.register(Ulid::new(), &p_exact);
475 let s_cons = Ulid::new();
476 b.register(s_cons, &consumer);
477
478 let m = b.find_producer_for(s_cons, "in").expect("match");
479 // El exact gana incluso si tiene priority igual: por estrategia.
480 assert_eq!(m.producer_label, "dht");
481 assert_eq!(m.via, MatchStrategy::Exact);
482 }
483
484 #[test]
485 fn pin_to_overrides_type_search() {
486 let mut b = Broker::new(BrokerConfig::default());
487 // Dos productores que producen el mismo tipo.
488 let p1 = card(
489 "dht-prod",
490 Priority::Normal,
491 Flows {
492 input: vec![],
493 output: vec![flow("out", prim("string"), None)],
494 },
495 );
496 let p2 = card(
497 "dht-test",
498 Priority::Normal,
499 Flows {
500 input: vec![],
501 output: vec![flow("out", prim("string"), None)],
502 },
503 );
504 let consumer = card(
505 "ui",
506 Priority::Normal,
507 Flows {
508 input: vec![flow("in", prim("string"), Some("dht-test"))],
509 output: vec![],
510 },
511 );
512 b.register(Ulid::new(), &p1);
513 b.register(Ulid::new(), &p2);
514 let s_cons = Ulid::new();
515 b.register(s_cons, &consumer);
516
517 let m = b.find_producer_for(s_cons, "in").expect("match");
518 assert_eq!(m.producer_label, "dht-test");
519 assert!(m.pinned);
520 }
521
522 #[test]
523 fn pin_to_unresolvable_falls_back_to_type_match() {
524 let mut b = Broker::new(BrokerConfig::default());
525 let p = card(
526 "real-dht",
527 Priority::Normal,
528 Flows {
529 input: vec![],
530 output: vec![flow("out", prim("string"), None)],
531 },
532 );
533 let consumer = card(
534 "ui",
535 Priority::Normal,
536 Flows {
537 input: vec![flow("in", prim("string"), Some("nonexistent"))],
538 output: vec![],
539 },
540 );
541 b.register(Ulid::new(), &p);
542 let s_cons = Ulid::new();
543 b.register(s_cons, &consumer);
544
545 let m = b.find_producer_for(s_cons, "in").expect("match");
546 assert_eq!(m.producer_label, "real-dht");
547 assert!(!m.pinned);
548 }
549
550 #[test]
551 fn priority_breaks_ties() {
552 let mut b = Broker::new(BrokerConfig::default());
553 let p_low = card(
554 "z-dht",
555 Priority::Low,
556 Flows {
557 input: vec![],
558 output: vec![flow("out", prim("string"), None)],
559 },
560 );
561 let p_high = card(
562 "a-dht",
563 Priority::High,
564 Flows {
565 input: vec![],
566 output: vec![flow("out", prim("string"), None)],
567 },
568 );
569 let consumer = card(
570 "ui",
571 Priority::Normal,
572 Flows {
573 input: vec![flow("in", prim("string"), None)],
574 output: vec![],
575 },
576 );
577 b.register(Ulid::new(), &p_low);
578 b.register(Ulid::new(), &p_high);
579 let s_cons = Ulid::new();
580 b.register(s_cons, &consumer);
581
582 let m = b.find_producer_for(s_cons, "in").expect("match");
583 assert_eq!(m.producer_label, "a-dht"); // priority High > Low
584 }
585
586 #[test]
587 fn label_alpha_breaks_priority_ties() {
588 let mut b = Broker::new(BrokerConfig::default());
589 let p1 = card(
590 "z-dht",
591 Priority::Normal,
592 Flows {
593 input: vec![],
594 output: vec![flow("out", prim("string"), None)],
595 },
596 );
597 let p2 = card(
598 "a-dht",
599 Priority::Normal,
600 Flows {
601 input: vec![],
602 output: vec![flow("out", prim("string"), None)],
603 },
604 );
605 let consumer = card(
606 "ui",
607 Priority::Normal,
608 Flows {
609 input: vec![flow("in", prim("string"), None)],
610 output: vec![],
611 },
612 );
613 b.register(Ulid::new(), &p1);
614 b.register(Ulid::new(), &p2);
615 let s_cons = Ulid::new();
616 b.register(s_cons, &consumer);
617
618 let m = b.find_producer_for(s_cons, "in").expect("match");
619 assert_eq!(m.producer_label, "a-dht"); // alfabético gana
620 }
621
622 #[test]
623 fn unregister_removes_producer() {
624 let mut b = Broker::new(BrokerConfig::default());
625 let p = card(
626 "dht",
627 Priority::Normal,
628 Flows {
629 input: vec![],
630 output: vec![flow("out", prim("string"), None)],
631 },
632 );
633 let consumer = card(
634 "ui",
635 Priority::Normal,
636 Flows {
637 input: vec![flow("in", prim("string"), None)],
638 output: vec![],
639 },
640 );
641 let s_p = Ulid::new();
642 b.register(s_p, &p);
643 let s_c = Ulid::new();
644 b.register(s_c, &consumer);
645
646 assert!(b.find_producer_for(s_c, "in").is_some());
647 b.unregister(s_p);
648 assert!(b.find_producer_for(s_c, "in").is_none());
649 }
650
651 #[test]
652 fn no_self_loops() {
653 let mut b = Broker::new(BrokerConfig::default());
654 let same = card(
655 "echo",
656 Priority::Normal,
657 Flows {
658 input: vec![flow("in", prim("string"), None)],
659 output: vec![flow("out", prim("string"), None)],
660 },
661 );
662 let s = Ulid::new();
663 b.register(s, &same);
664
665 // Solo una Card registrada — no hay otra que produzca string.
666 assert!(b.find_producer_for(s, "in").is_none());
667 }
668
669 #[test]
670 fn all_matches_lists_pairs() {
671 let mut b = Broker::new(BrokerConfig::default());
672 let dht = card(
673 "dht",
674 Priority::Normal,
675 Flows {
676 input: vec![flow("query", prim("string"), None)],
677 output: vec![flow("results", prim("bytes"), None)],
678 },
679 );
680 let ui = card(
681 "ui",
682 Priority::Normal,
683 Flows {
684 input: vec![flow("data", prim("bytes"), None)],
685 output: vec![flow("user-input", prim("string"), None)],
686 },
687 );
688 b.register(Ulid::new(), &dht);
689 b.register(Ulid::new(), &ui);
690
691 let matches = b.all_matches();
692 assert_eq!(matches.len(), 2);
693 // dht.query ← ui.user-input y ui.data ← dht.results
694 let pairs: Vec<_> = matches
695 .iter()
696 .map(|m| (m.consumer_label.as_str(), m.producer_label.as_str()))
697 .collect();
698 assert!(pairs.contains(&("dht", "ui")));
699 assert!(pairs.contains(&("ui", "dht")));
700 }
701}