subc_daemon/
router.rs

1use std::{
2    collections::HashMap,
3    error::Error,
4    fmt,
5    sync::{
6        atomic::{AtomicU64, Ordering},
7        Arc,
8    },
9    time::Instant,
10};
11
12use subc_protocol::{ErrorBody, Flags, FrameType, Priority};
13use tokio::sync::mpsc;
14use tracing::debug;
15
16use crate::{
17    control::ControlHandler,
18    forwarding::{
19        CloseReason, ConnectionCloseReceiver, DataRoute, DataRouteState, ForwardingError,
20        ForwardingTable, RouteBinding, RouteRelease,
21    },
22    registry::ConnectionId,
23    DaemonCounters, Frame, FrameBuildError,
24};
25
26/// One queued outbound frame plus the instant it entered the writer queue.
27///
28/// The stamp exists for the reply-path half of slow-control diagnosis: a
29/// handler can finish in microseconds while the reply sits in this queue
30/// waiting for the writer task to be scheduled, and without a per-item stamp
31/// that wait is invisible to every other timing point (the client's round
32/// trip is the only witness, and it cannot say which side ate the time).
33/// Constructed exclusively inside [`FrameSink`] so no caller can forget it.
34#[derive(Debug)]
35pub struct OutboundFrame {
36    pub frame: Frame,
37    pub enqueued_at: std::time::Instant,
38    pub(crate) flushed: Option<tokio::sync::oneshot::Sender<()>>,
39}
40
41impl OutboundFrame {
42    fn now(frame: Frame) -> Self {
43        Self {
44            frame,
45            enqueued_at: std::time::Instant::now(),
46            flushed: None,
47        }
48    }
49}
50
51/// An `OutboundFrame` is a stamped `Frame`; deref keeps every existing frame
52/// read (headers, bodies, assertions) working on queued items unchanged.
53impl std::ops::Deref for OutboundFrame {
54    type Target = Frame;
55
56    fn deref(&self) -> &Frame {
57        &self.frame
58    }
59}
60
61/// Shared tracing-capture helpers for timing-observability tests across
62/// modules (router dispatch, server reply path). Test-only.
63#[cfg(test)]
64pub(crate) mod test_log {
65    use std::{
66        io::Write,
67        sync::{Arc, Mutex},
68    };
69
70    #[derive(Clone)]
71    struct TestLogWriter(Arc<Mutex<Vec<u8>>>);
72
73    impl Write for TestLogWriter {
74        fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
75            self.0
76                .lock()
77                .expect("test log capture is not poisoned")
78                .extend(buffer);
79            Ok(buffer.len())
80        }
81
82        fn flush(&mut self) -> std::io::Result<()> {
83            Ok(())
84        }
85    }
86
87    pub(crate) fn log_capture(
88        level: tracing::Level,
89    ) -> (Arc<Mutex<Vec<u8>>>, tracing::dispatcher::DefaultGuard) {
90        let output = Arc::new(Mutex::new(Vec::new()));
91        let writer = Arc::clone(&output);
92        let subscriber = tracing_subscriber::fmt()
93            .with_max_level(level)
94            .with_ansi(false)
95            .without_time()
96            .with_target(false)
97            .with_writer(move || TestLogWriter(Arc::clone(&writer)))
98            .finish();
99        let guard = tracing::subscriber::set_default(subscriber);
100        (output, guard)
101    }
102
103    pub(crate) fn captured_logs(output: &Arc<Mutex<Vec<u8>>>) -> String {
104        String::from_utf8(
105            output
106                .lock()
107                .expect("test log capture is not poisoned")
108                .clone(),
109        )
110        .expect("tracing output is UTF-8")
111    }
112}
113
114/// Cheaply cloneable handle to one connection's bounded outbound frame queue.
115///
116/// Backends emit responses, streaming frames, and future PUSH frames through this
117/// single path. The bounded `mpsc` sender is the connection-level backpressure
118/// substrate; the socket layer owns the sole receiver/writer.
119#[derive(Debug, Clone)]
120pub struct FrameSink {
121    tx: mpsc::Sender<OutboundFrame>,
122}
123
124impl FrameSink {
125    pub fn new(tx: mpsc::Sender<OutboundFrame>) -> Self {
126        Self { tx }
127    }
128
129    pub async fn send(&self, frame: Frame) -> Result<(), RouterError> {
130        let channel = frame.header.channel;
131        let epoch = frame.header.epoch;
132        let corr = frame.header.corr;
133        self.tx.send(OutboundFrame::now(frame)).await.map_err(|_| {
134            RouterError::backend_with_epoch(channel, epoch, corr, "connection writer closed")
135        })
136    }
137
138    /// Shutdown notices must leave the socket writer before an idle daemon exits.
139    /// Queue admission alone does not prove this; the writer acknowledges flush.
140    #[cfg(unix)]
141    pub(crate) async fn send_flushed(&self, frame: Frame) -> Result<(), RouterError> {
142        let (tx, rx) = tokio::sync::oneshot::channel();
143        let mut outbound = OutboundFrame::now(frame);
144        outbound.flushed = Some(tx);
145        self.tx
146            .send(outbound)
147            .await
148            .map_err(|_| RouterError::backend(0, 0, "connection writer closed"))?;
149        rx.await
150            .map_err(|_| RouterError::backend(0, 0, "connection flush failed"))
151    }
152
153    pub(crate) async fn reserve_owned(
154        &self,
155    ) -> Result<mpsc::OwnedPermit<OutboundFrame>, RouterError> {
156        self.tx
157            .clone()
158            .reserve_owned()
159            .await
160            .map_err(|_| RouterError::backend(0, 0, "connection writer closed"))
161    }
162
163    #[cfg(test)]
164    pub(crate) fn try_reserve_owned(
165        &self,
166    ) -> Result<mpsc::OwnedPermit<OutboundFrame>, RouterError> {
167        self.tx
168            .clone()
169            .try_reserve_owned()
170            .map_err(|err| RouterError::backend(0, 0, err.to_string()))
171    }
172
173    pub(crate) fn is_closed(&self) -> bool {
174        self.tx.is_closed()
175    }
176
177    pub(crate) fn try_send(&self, frame: Frame) -> Result<(), RouterError> {
178        let channel = frame.header.channel;
179        let epoch = frame.header.epoch;
180        let corr = frame.header.corr;
181        self.tx.try_send(OutboundFrame::now(frame)).map_err(|err| {
182            RouterError::backend_with_epoch(
183                channel,
184                epoch,
185                corr,
186                format!("connection writer unavailable: {err}"),
187            )
188        })
189    }
190}
191
192/// Per-route context shared with backends besides the frame itself.
193#[derive(Debug, Clone)]
194pub struct RouteCtx {
195    pub connection_id: ConnectionId,
196    pub egress: FrameSink,
197}
198
199/// Closed set of data-plane backends for non-zero channels.
200///
201/// Channel 0 is structurally special and remains [`Router::control`], not an
202/// enum variant. Static/test backends live in [`Router::backends`]; the forwarding
203/// backend is selected dynamically only when a per-connection forwarding binding exists.
204#[derive(Debug, Clone)]
205pub enum Backend {
206    Echo(EchoBackend),
207    Forward(ForwardBackend),
208}
209
210impl From<EchoBackend> for Backend {
211    fn from(backend: EchoBackend) -> Self {
212        Self::Echo(backend)
213    }
214}
215
216impl From<ForwardBackend> for Backend {
217    fn from(backend: ForwardBackend) -> Self {
218        Self::Forward(backend)
219    }
220}
221
222impl Backend {
223    pub async fn handle(&self, ctx: RouteCtx, frame: Frame) -> Result<(), RouterError> {
224        match self {
225            Self::Echo(backend) => backend.handle(ctx, frame).await,
226            Self::Forward(backend) => backend.handle(ctx, frame).await,
227        }
228    }
229}
230
231/// I/O-agnostic splice router keyed by envelope `channel`.
232///
233/// Channel 0 is reserved for subc itself and is always dispatched to the
234/// dedicated control handler. Other channels must be explicitly registered.
235/// Unknown client-originated non-zero channels are translated to canonical JSON `ERROR` frames on
236/// the connection sink so the peer can continue using the same socket. Module-originated frames for
237/// released route channels are logged and dropped as the channel-gone race backstop.
238///
239/// DATA-PLANE BODIES ARE NEVER DECODED HERE, and the consequence is worth stating
240/// because it looks like a guarantee and is not. Additive fields in a request or
241/// response body reach the far side untouched — not because anything permits them,
242/// but because routing reads only the 21-byte header and treats the body as opaque
243/// bytes. That is a performance property, so **nothing prevents it from changing**:
244/// a future reason to inspect a body would convert a wire-transparent path into a
245/// filtering one, and nobody would think of it as a contract change.
246///
247/// So when a peer asks whether subc sees an additive field, the answer is per-path
248/// and this path's zero means WE NEVER LOOK rather than WE LOOK AT EVERYTHING. The
249/// control plane (typed enums at the frame boundary) and the MCP gateway (envelope
250/// unwrap plus a named struct) both narrow; only this one does not.
251pub struct Router {
252    backends: HashMap<u16, Backend>,
253    control: Arc<ControlHandler>,
254    forwarding: Arc<ForwardingTable>,
255    forward_backend: ForwardBackend,
256    counters: DaemonCounters,
257    next_connection_id: AtomicU64,
258}
259
260impl Router {
261    pub fn with_control_handler(control: Arc<ControlHandler>) -> Self {
262        // The handler the router serves is the one a swap must tell when it
263        // promotes a candidate; this is where it first sits behind an `Arc`.
264        control.install_swap_promotion_observer();
265        let forwarding = control.forwarding();
266        let counters = control.counters();
267        Self {
268            backends: HashMap::new(),
269            control,
270            forwarding: Arc::clone(&forwarding),
271            forward_backend: ForwardBackend::new(forwarding),
272            counters,
273            // ConnectionId::LOCAL is 0; real socket ids start at 1 and never collide.
274            next_connection_id: AtomicU64::new(1),
275        }
276    }
277
278    pub fn with_default_self_handler() -> Self {
279        Self::with_control_handler(Arc::new(ControlHandler::default()))
280    }
281
282    pub fn forwarding(&self) -> Arc<ForwardingTable> {
283        Arc::clone(&self.forwarding)
284    }
285
286    pub fn register_backend(
287        &mut self,
288        channel: u16,
289        backend: impl Into<Backend>,
290    ) -> Result<(), RouterError> {
291        self.register_backend_arc(channel, Arc::new(backend.into()))
292    }
293
294    pub(crate) fn register_backend_arc(
295        &mut self,
296        channel: u16,
297        backend: Arc<Backend>,
298    ) -> Result<(), RouterError> {
299        if channel == 0 {
300            return Err(RouterError::ReservedChannelZero);
301        }
302        if self.backends.contains_key(&channel) {
303            return Err(RouterError::DuplicateChannel { channel });
304        }
305        self.backends.insert(channel, backend.as_ref().clone());
306        Ok(())
307    }
308
309    /// Start a connection-scoped routing context. Dropping the guard releases
310    /// any control-plane registrations owned by the connection.
311    fn record_module_frame_drop(&self, connection_id: ConnectionId) -> Result<(), RouterError> {
312        let module_id = self
313            .forwarding
314            .module_id_for_connection(connection_id)
315            .map_err(RouterError::Forwarding)?;
316        self.counters
317            .increment_module_frames_dropped_no_route(module_id.as_deref());
318        Ok(())
319    }
320
321    pub fn begin_connection(&self) -> RouterConnection {
322        let raw = self.next_connection_id.fetch_add(1, Ordering::Relaxed);
323        let id = ConnectionId::new(raw);
324        let close_receiver = self.forwarding.register_connection_close(id);
325        RouterConnection {
326            id,
327            control_handler: Arc::clone(&self.control),
328            forwarding: Arc::clone(&self.forwarding),
329            close_receiver: Some(close_receiver),
330        }
331    }
332
333    pub(crate) fn route_open_target(&self, frame: &Frame) -> Option<String> {
334        self.control.route_open_target(frame)
335    }
336
337    pub(crate) fn route_open_capacity_refusal(
338        &self,
339        ctx: &RouteCtx,
340        frame: &Frame,
341        target_module_id: &str,
342        limit: usize,
343    ) -> Result<Frame, RouterError> {
344        self.control
345            .route_open_capacity_refusal(ctx, frame, target_module_id, limit)
346    }
347
348    pub async fn route_for_connection(
349        &self,
350        ctx: &RouteCtx,
351        frame: Frame,
352    ) -> Result<(), RouterError> {
353        self.route_for_connection_started(ctx, frame, None).await
354    }
355
356    pub(crate) async fn route_for_connection_started(
357        &self,
358        ctx: &RouteCtx,
359        frame: Frame,
360        dispatch_started_at: Option<Instant>,
361    ) -> Result<(), RouterError> {
362        let channel = frame.header.channel;
363        let epoch = frame.header.epoch;
364        let corr = frame.header.corr;
365        if channel == 0 {
366            debug!(
367                connection_id = ctx.connection_id.get(),
368                corr,
369                frame_type = ?frame.header.ty,
370                "routing control frame"
371            );
372            // The connection loop dispatches every control operation directly
373            // except `route.open`. Its task passes a timestamp captured before
374            // spawn, so slow-dispatch timing includes scheduler delay but no
375            // wait in an application-owned queue.
376            let dispatch_started_at = (frame.header.ty == FrameType::Request)
377                .then(|| dispatch_started_at.unwrap_or_else(Instant::now));
378            let responses = self
379                .control
380                .handle_control_frame_timed(ctx, frame, dispatch_started_at)
381                .await?;
382            for response in responses {
383                ctx.egress.send(response).await?;
384            }
385            return Ok(());
386        }
387
388        let data_route = self
389            .forwarding
390            .lookup_data_route(ctx.connection_id, channel, epoch)
391            .map_err(RouterError::Forwarding)?;
392
393        match data_route {
394            DataRoute::Module(DataRouteState::EpochMismatch) => {
395                if frame.header.ty == FrameType::Request {
396                    self.counters
397                        .increment_module_requests_dropped_stale_route();
398                    let err = RouterError::StaleRouteEpoch {
399                        channel,
400                        epoch,
401                        corr,
402                    };
403                    if let Some(error_frame) = err.to_error_frame() {
404                        ctx.egress.send(error_frame).await?;
405                    }
406                } else {
407                    self.record_module_frame_drop(ctx.connection_id)?;
408                }
409                debug!(
410                    connection_id = ctx.connection_id.get(),
411                    channel, epoch, corr, "dropping module frame for stale route epoch"
412                );
413                return Ok(());
414            }
415            DataRoute::Module(DataRouteState::Reserved) => {
416                if frame.header.ty == FrameType::Request {
417                    self.counters
418                        .increment_module_requests_dropped_stale_route();
419                    let err = RouterError::UnknownChannel {
420                        channel,
421                        epoch,
422                        corr,
423                    };
424                    if let Some(error_frame) = err.to_error_frame() {
425                        ctx.egress.send(error_frame).await?;
426                    }
427                } else {
428                    self.record_module_frame_drop(ctx.connection_id)?;
429                }
430                debug!(
431                    connection_id = ctx.connection_id.get(),
432                    channel, epoch, corr, "dropping module frame for reserved route handle"
433                );
434                return Ok(());
435            }
436            DataRoute::Module(DataRouteState::Absent) => {
437                if frame.header.ty == FrameType::Request {
438                    self.counters
439                        .increment_module_requests_dropped_stale_route();
440                    let err = RouterError::UnknownChannel {
441                        channel,
442                        epoch,
443                        corr,
444                    };
445                    if let Some(error_frame) = err.to_error_frame() {
446                        ctx.egress.send(error_frame).await?;
447                    }
448                } else {
449                    self.record_module_frame_drop(ctx.connection_id)?;
450                }
451                debug!(
452                    connection_id = ctx.connection_id.get(),
453                    channel, epoch, corr, "dropping module frame for absent route handle"
454                );
455                return Ok(());
456            }
457            DataRoute::Module(DataRouteState::Bound(route)) => {
458                if frame.header.ty == FrameType::Goodbye {
459                    if let RouteRelease::Removed(target) = self
460                        .forwarding
461                        .release_module_route(ctx.connection_id, channel, epoch)
462                        .map_err(RouterError::Forwarding)?
463                    {
464                        let mut goodbye = frame;
465                        goodbye.header.channel = target.channel;
466                        goodbye.header.epoch = target.epoch;
467                        if let Err(err) = target.sink.try_send(goodbye) {
468                            if target.close_on_delivery_failure()
469                                && self
470                                    .forwarding
471                                    .escalate_client_delivery_failure(
472                                        target.connection_id,
473                                        target.channel,
474                                        target.epoch,
475                                        CloseReason::new(
476                                            "route_goodbye_delivery_failed",
477                                            format!(
478                                                "failed to enqueue route GOODBYE for client channel {}: {err}",
479                                                target.channel
480                                            ),
481                                        ),
482                                    )
483                                    .map_err(RouterError::Forwarding)?
484                            {
485                                self.counters.increment_goodbye_relay_client_failed();
486                            }
487                        }
488                    }
489                    return Ok(());
490                }
491
492                let releases_credit = is_terminal_frame(frame.header.ty);
493                let mut frame = frame;
494                frame.header.channel = route.client_channel;
495                frame.header.epoch = route.client_epoch;
496                if let Err(err) = route.client_sink.try_send(frame) {
497                    if self
498                        .forwarding
499                        .escalate_client_delivery_failure(
500                            route.client_connection_id,
501                            route.client_channel,
502                            route.client_epoch,
503                            CloseReason::new(
504                                "module_to_client_delivery_failed",
505                                format!(
506                                    "failed to enqueue module frame for client channel {} corr {corr}: {err}",
507                                    route.client_channel
508                                ),
509                            ),
510                        )
511                        .map_err(RouterError::Forwarding)?
512                    {
513                        self.counters
514                            .increment_client_egress_close_delivery_failed();
515                    }
516                    return Ok(());
517                }
518                if releases_credit {
519                    route.flow.release_corr(corr);
520                }
521                return Ok(());
522            }
523            DataRoute::Client(DataRouteState::EpochMismatch) => {
524                if frame.header.ty == FrameType::Request {
525                    self.counters.increment_client_frames_dropped_stale_route();
526                    // Dropped before forwarding; a re-bind retry cannot double-execute this request.
527                    let err = RouterError::StaleRouteEpoch {
528                        channel,
529                        epoch,
530                        corr,
531                    };
532                    if let Some(error_frame) = err.to_error_frame() {
533                        ctx.egress.send(error_frame).await?;
534                    }
535                }
536                debug!(
537                    connection_id = ctx.connection_id.get(),
538                    channel, epoch, corr, "dropping client frame for stale route epoch"
539                );
540                return Ok(());
541            }
542            DataRoute::Client(DataRouteState::Reserved) => {
543                if frame.header.ty == FrameType::Request {
544                    let err = RouterError::UnknownChannel {
545                        channel,
546                        epoch,
547                        corr,
548                    };
549                    if let Some(error_frame) = err.to_error_frame() {
550                        ctx.egress.send(error_frame).await?;
551                    }
552                }
553                return Ok(());
554            }
555            DataRoute::Client(DataRouteState::Bound(route)) => {
556                if frame.header.ty == FrameType::Goodbye {
557                    let _ = self
558                        .control
559                        .handle_route_goodbye(ctx.connection_id, channel, epoch)?;
560                    return Ok(());
561                }
562                return self.forward_backend.handle_bound(frame, route).await;
563            }
564            DataRoute::Client(DataRouteState::Absent) => {}
565        }
566
567        if let Some(backend) = self.backends.get(&channel) {
568            return backend.handle(ctx.clone(), frame).await;
569        }
570        if frame.header.ty == FrameType::Request {
571            let err = RouterError::UnknownChannel {
572                channel,
573                epoch,
574                corr,
575            };
576            if let Some(error_frame) = err.to_error_frame() {
577                ctx.egress.send(error_frame).await?;
578            }
579        }
580        Ok(())
581    }
582}
583
584impl Default for Router {
585    fn default() -> Self {
586        Self::with_default_self_handler()
587    }
588}
589
590/// Connection-scoped cleanup guard returned by [`Router::begin_connection`].
591#[must_use]
592pub struct RouterConnection {
593    id: ConnectionId,
594    control_handler: Arc<ControlHandler>,
595    forwarding: Arc<ForwardingTable>,
596    close_receiver: Option<ConnectionCloseReceiver>,
597}
598
599impl RouterConnection {
600    pub fn id(&self) -> ConnectionId {
601        self.id
602    }
603
604    pub(crate) fn take_close_receiver(&mut self) -> ConnectionCloseReceiver {
605        self.close_receiver
606            .take()
607            .expect("connection close receiver can only be taken once")
608    }
609}
610
611impl Drop for RouterConnection {
612    fn drop(&mut self) {
613        self.forwarding.unregister_connection_close(self.id);
614        // GOODBYE (explicit) and connection-drop cleanup both call the same
615        // idempotent deregistration path.
616        let _ = self.control_handler.cleanup_connection(self.id);
617    }
618}
619
620/// Minimal in-memory backend used by tests and early wiring: it emits a
621/// `RESPONSE` on the same channel/correlation id with the exact same body bytes.
622#[derive(Debug, Default, Clone, Copy)]
623pub struct EchoBackend;
624
625impl EchoBackend {
626    pub async fn handle(&self, ctx: RouteCtx, frame: Frame) -> Result<(), RouterError> {
627        let response = Frame::build_with_version(
628            frame.header.ver,
629            FrameType::Response,
630            frame.header.flags,
631            frame.header.channel,
632            frame.header.epoch,
633            frame.header.corr,
634            frame.body,
635        )
636        .map_err(RouterError::FrameBuild)?;
637        ctx.egress.send(response).await
638    }
639}
640
641/// Data-plane backend that splices client frames to the module connection bound at attach time.
642#[derive(Debug, Clone)]
643pub struct ForwardBackend {
644    forwarding: Arc<ForwardingTable>,
645}
646
647impl ForwardBackend {
648    pub fn new(forwarding: Arc<ForwardingTable>) -> Self {
649        Self { forwarding }
650    }
651
652    pub async fn handle(&self, ctx: RouteCtx, frame: Frame) -> Result<(), RouterError> {
653        let channel = frame.header.channel;
654        let corr = frame.header.corr;
655        let route = match self
656            .forwarding
657            .lookup_data_route(ctx.connection_id, channel, frame.header.epoch)
658            .map_err(RouterError::Forwarding)?
659        {
660            DataRoute::Client(DataRouteState::Bound(route)) => route,
661            DataRoute::Client(_) | DataRoute::Module(_) => {
662                return Err(RouterError::UnknownChannel {
663                    channel,
664                    epoch: frame.header.epoch,
665                    corr,
666                });
667            }
668        };
669        self.handle_bound(frame, route).await
670    }
671
672    pub(crate) async fn handle_bound(
673        &self,
674        frame: Frame,
675        route: Arc<RouteBinding>,
676    ) -> Result<(), RouterError> {
677        let channel = frame.header.channel;
678        let corr = frame.header.corr;
679        let frame_type = frame.header.ty;
680
681        // CANCEL and other non-REQUEST frames bypass the request-credit window;
682        // the original request's credit is freed only by the module's terminal frame.
683        let acquired_credit = frame_type == FrameType::Request;
684        if acquired_credit {
685            if let Err(err) = route
686                .flow
687                .acquire_tagged(corr, frame.header.flags.is_subscription())
688                .await
689            {
690                // `module_reloading` here is answered BEFORE the frame is
691                // forwarded, so the module never sees the request and callers
692                // may re-dispatch it after reopening the route. That is a wire
693                // guarantee documented on `error_codes::MODULE_RELOADING`; do
694                // not emit this code for a request that already reached
695                // `module_sink.send` below.
696                if self
697                    .forwarding
698                    .endpoint_is_draining(route.module_endpoint)
699                    .map_err(RouterError::Forwarding)?
700                {
701                    return Err(RouterError::route_error_with_epoch(
702                        channel,
703                        frame.header.epoch,
704                        corr,
705                        "module_reloading",
706                        format!("module endpoint for route channel {channel} is reloading"),
707                    ));
708                }
709                return Err(RouterError::backend_with_epoch(
710                    channel,
711                    frame.header.epoch,
712                    corr,
713                    format!("{err} for route channel {channel}"),
714                ));
715            }
716        }
717
718        let mut frame = frame;
719        frame.header.channel = route.module_channel;
720        frame.header.epoch = route.module_epoch;
721        let result = route.module_sink.send(frame).await.map_err(|err| {
722            RouterError::backend_with_epoch(channel, route.client_epoch, corr, err.to_string())
723        });
724        if acquired_credit && result.is_err() {
725            route.flow.release_corr(corr);
726        }
727        result
728    }
729}
730
731fn is_terminal_frame(frame_type: FrameType) -> bool {
732    matches!(
733        frame_type,
734        FrameType::Response | FrameType::Error | FrameType::StreamEnd
735    )
736}
737
738/// Typed router errors. Routable failures can be translated to canonical JSON
739/// `ERROR` frames with [`RouterError::to_error_frame`].
740#[derive(Debug, Clone, PartialEq, Eq)]
741pub enum RouterError {
742    ReservedChannelZero,
743    DuplicateChannel {
744        channel: u16,
745    },
746    UnknownChannel {
747        channel: u16,
748        epoch: u32,
749        corr: u64,
750    },
751    StaleRouteEpoch {
752        channel: u16,
753        epoch: u32,
754        corr: u64,
755    },
756    Backend {
757        channel: u16,
758        epoch: u32,
759        corr: u64,
760        message: String,
761    },
762    RouteError {
763        channel: u16,
764        epoch: u32,
765        corr: u64,
766        code: String,
767        message: String,
768    },
769    FrameBuild(FrameBuildError),
770    Forwarding(ForwardingError),
771}
772
773impl RouterError {
774    pub fn backend(channel: u16, corr: u64, message: impl Into<String>) -> Self {
775        Self::backend_with_epoch(channel, 0, corr, message)
776    }
777
778    pub fn backend_with_epoch(
779        channel: u16,
780        epoch: u32,
781        corr: u64,
782        message: impl Into<String>,
783    ) -> Self {
784        Self::Backend {
785            channel,
786            epoch,
787            corr,
788            message: message.into(),
789        }
790    }
791
792    pub fn route_error(
793        channel: u16,
794        corr: u64,
795        code: impl Into<String>,
796        message: impl Into<String>,
797    ) -> Self {
798        Self::route_error_with_epoch(channel, 0, corr, code, message)
799    }
800
801    pub fn route_error_with_epoch(
802        channel: u16,
803        epoch: u32,
804        corr: u64,
805        code: impl Into<String>,
806        message: impl Into<String>,
807    ) -> Self {
808        Self::RouteError {
809            channel,
810            epoch,
811            corr,
812            code: code.into(),
813            message: message.into(),
814        }
815    }
816
817    /// Translate route failures that belong on the wire into an `ERROR` frame.
818    pub fn to_error_frame(&self) -> Option<Frame> {
819        match self {
820            Self::UnknownChannel {
821                channel,
822                epoch,
823                corr,
824            } => error_frame(
825                *channel,
826                *epoch,
827                *corr,
828                "unknown_channel",
829                format!("unknown channel {channel}"),
830            ),
831            Self::StaleRouteEpoch {
832                channel,
833                epoch,
834                corr,
835            } => error_frame(
836                *channel,
837                *epoch,
838                *corr,
839                "stale_route_epoch",
840                format!("stale route epoch for channel {channel}"),
841            ),
842            Self::Backend {
843                channel,
844                epoch,
845                corr,
846                message,
847            } => error_frame(*channel, *epoch, *corr, "backend_error", message.clone()),
848            Self::RouteError {
849                channel,
850                epoch,
851                corr,
852                code,
853                message,
854            } => error_frame(*channel, *epoch, *corr, code, message.clone()),
855            Self::ReservedChannelZero
856            | Self::DuplicateChannel { .. }
857            | Self::FrameBuild(_)
858            | Self::Forwarding(_) => None,
859        }
860    }
861}
862
863fn error_frame(channel: u16, epoch: u32, corr: u64, code: &str, message: String) -> Option<Frame> {
864    let body = serde_json::to_vec(&ErrorBody {
865        code: code.to_string(),
866        message,
867        detail: None,
868    })
869    .ok()?;
870
871    Frame::build(
872        FrameType::Error,
873        Flags::new(false, Priority::Passive, false),
874        channel,
875        epoch,
876        corr,
877        body,
878    )
879    .ok()
880}
881
882impl fmt::Display for RouterError {
883    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
884        match self {
885            Self::ReservedChannelZero => write!(f, "channel 0 is reserved for subc"),
886            Self::DuplicateChannel { channel } => {
887                write!(f, "backend already registered for channel {channel}")
888            }
889            Self::UnknownChannel { channel, corr, .. } => {
890                write!(f, "unknown channel {channel} for corr {corr}")
891            }
892            Self::StaleRouteEpoch { channel, corr, .. } => {
893                write!(f, "stale route epoch for channel {channel} corr {corr}")
894            }
895            Self::Backend {
896                channel,
897                corr,
898                message,
899                ..
900            } => write!(
901                f,
902                "backend error on channel {channel} corr {corr}: {message}"
903            ),
904            Self::RouteError {
905                channel,
906                corr,
907                code,
908                message,
909                ..
910            } => write!(
911                f,
912                "route error {code} on channel {channel} corr {corr}: {message}"
913            ),
914            Self::FrameBuild(err) => write!(f, "failed to build routed frame: {err}"),
915            Self::Forwarding(err) => write!(f, "forwarding error: {err}"),
916        }
917    }
918}
919
920impl Error for RouterError {
921    fn source(&self) -> Option<&(dyn Error + 'static)> {
922        match self {
923            Self::FrameBuild(err) => Some(err),
924            Self::Forwarding(err) => Some(err),
925            Self::ReservedChannelZero
926            | Self::DuplicateChannel { .. }
927            | Self::UnknownChannel { .. }
928            | Self::StaleRouteEpoch { .. }
929            | Self::Backend { .. }
930            | Self::RouteError { .. } => None,
931        }
932    }
933}
934
935#[cfg(test)]
936mod tests {
937    use super::*;
938    use crate::{
939        forwarding::RouteBindRelayOutcome,
940        supervise::{ModuleSpec, RestartPolicy, Supervisor, SupervisorHandle},
941        ControlHandler, Registry,
942    };
943    use std::{
944        sync::{mpsc as std_mpsc, Arc},
945        time::Duration,
946    };
947    use subc_control::ModuleProtocol;
948    use subc_protocol::{manifest::Concurrency, ErrorBody, Flags, FrameType, Priority};
949    use tokio::sync::mpsc;
950
951    pub(crate) use crate::router::test_log::{captured_logs, log_capture};
952
953    fn logged_millis(logs: &str, field: &str) -> u64 {
954        logs.split_whitespace()
955            .find_map(|part| part.strip_prefix(field))
956            .and_then(|value| value.parse().ok())
957            .unwrap_or_else(|| panic!("missing numeric {field} in logs: {logs}"))
958    }
959
960    fn request(channel: u16, corr: u64, body: &[u8]) -> Frame {
961        Frame::build(
962            FrameType::Request,
963            Flags::new(true, Priority::Interactive, false),
964            channel,
965            0,
966            corr,
967            body.to_vec(),
968        )
969        .unwrap()
970    }
971
972    fn ping(corr: u64) -> Frame {
973        Frame::build(
974            FrameType::Ping,
975            Flags::new(false, Priority::Passive, false),
976            0,
977            0,
978            corr,
979            Vec::new(),
980        )
981        .unwrap()
982    }
983
984    fn route_ctx() -> (RouteCtx, mpsc::Receiver<crate::router::OutboundFrame>) {
985        let (tx, rx) = mpsc::channel(8);
986        (
987            RouteCtx {
988                connection_id: ConnectionId::LOCAL,
989                egress: FrameSink::new(tx),
990            },
991            rx,
992        )
993    }
994
995    #[tokio::test]
996    async fn echo_backend_returns_response_with_byte_identical_body() {
997        let mut router = Router::with_default_self_handler();
998        router.register_backend(7, EchoBackend).unwrap();
999        let (ctx, mut rx) = route_ctx();
1000        let body = b"{not parsed}\0\xff";
1001
1002        router
1003            .route_for_connection(&ctx, request(7, 123, body))
1004            .await
1005            .unwrap();
1006        let response = rx.recv().await.unwrap();
1007
1008        assert_eq!(response.header.ty, FrameType::Response);
1009        assert_eq!(response.header.channel, 7);
1010        assert_eq!(response.header.corr, 123);
1011        assert_eq!(response.body, body);
1012        assert!(rx.try_recv().is_err());
1013    }
1014
1015    #[tokio::test]
1016    async fn unknown_channel_emits_canonical_error_frame() {
1017        let router = Router::with_default_self_handler();
1018        let (ctx, mut rx) = route_ctx();
1019
1020        router
1021            .route_for_connection(&ctx, request(99, 5, b"payload"))
1022            .await
1023            .unwrap();
1024        let error_frame = rx.recv().await.unwrap();
1025
1026        assert_eq!(error_frame.header.ty, FrameType::Error);
1027        assert_eq!(error_frame.header.channel, 99);
1028        assert_eq!(error_frame.header.corr, 5);
1029        let body: ErrorBody = serde_json::from_slice(&error_frame.body).unwrap();
1030        assert_eq!(body.code, "unknown_channel");
1031        assert_eq!(body.message, "unknown channel 99");
1032    }
1033
1034    #[tokio::test]
1035    async fn channel_zero_uses_control_handler_not_backend_registry() {
1036        let mut router = Router::with_default_self_handler();
1037        router.register_backend(1, EchoBackend).unwrap();
1038        let (ctx, mut rx) = route_ctx();
1039
1040        router.route_for_connection(&ctx, ping(77)).await.unwrap();
1041        let response = rx.recv().await.unwrap();
1042
1043        assert_eq!(response.header.ty, FrameType::Pong);
1044        assert_eq!(response.header.channel, 0);
1045        assert_eq!(response.header.corr, 77);
1046        assert!(response.body.is_empty());
1047    }
1048
1049    #[tokio::test]
1050    async fn slow_control_dispatch_logs_decoded_op_and_elapsed_time() {
1051        let control = Arc::new(
1052            ControlHandler::new(Arc::new(Registry::default()))
1053                .with_control_dispatch_delay(Duration::from_millis(1050)),
1054        );
1055        let router = Router::with_control_handler(control);
1056        let (ctx, mut rx) = route_ctx();
1057        let (output, guard) = log_capture(tracing::Level::WARN);
1058
1059        router
1060            .route_for_connection(&ctx, request(0, 41, br#"{"op":"server.describe"}"#))
1061            .await
1062            .expect("slow request routes");
1063        assert!(rx.recv().await.is_some(), "request receives a response");
1064        drop(guard);
1065
1066        let logs = captured_logs(&output);
1067        assert!(logs.contains("slow control dispatch"));
1068        assert!(logs.contains("op=server.describe"));
1069        assert!(logs.contains("connection_id=0"));
1070        assert!(logs.contains("corr=41"));
1071        assert!(
1072            logged_millis(&logs, "elapsed_ms=") >= 1050,
1073            "elapsed must include the injected handler delay: {logs}"
1074        );
1075    }
1076
1077    #[tokio::test]
1078    async fn fast_control_dispatch_emits_arrival_without_slow_warning() {
1079        let router = Router::with_default_self_handler();
1080        let (ctx, mut rx) = route_ctx();
1081        let (output, guard) = log_capture(tracing::Level::DEBUG);
1082
1083        router
1084            .route_for_connection(&ctx, request(0, 42, br#"{"op":"server.describe"}"#))
1085            .await
1086            .expect("fast request routes");
1087        assert!(rx.recv().await.is_some(), "request receives a response");
1088        drop(guard);
1089
1090        let logs = captured_logs(&output);
1091        assert!(logs.contains("control dispatch op=server.describe connection_id=0 corr=42"));
1092        assert!(!logs.contains("slow control dispatch"));
1093    }
1094
1095    #[tokio::test]
1096    async fn control_dispatch_arrival_is_hidden_at_info() {
1097        let router = Router::with_default_self_handler();
1098        let (ctx, mut rx) = route_ctx();
1099        let (output, guard) = log_capture(tracing::Level::INFO);
1100
1101        router
1102            .route_for_connection(&ctx, request(0, 43, br#"{"op":"server.describe"}"#))
1103            .await
1104            .expect("fast request routes");
1105        assert!(rx.recv().await.is_some(), "request receives a response");
1106        drop(guard);
1107
1108        assert!(
1109            !captured_logs(&output).contains("control dispatch"),
1110            "arrival logging must stay hidden at INFO"
1111        );
1112    }
1113
1114    #[tokio::test]
1115    async fn supervisor_list_logs_contended_snapshot_lock_only() {
1116        let registry = Arc::new(Registry::default());
1117        let handle = SupervisorHandle::new();
1118        let supervisor = Supervisor::new(Arc::clone(&registry), RestartPolicy::default())
1119            .with_handle(handle.clone());
1120        let module = supervisor
1121            .supervise_configured(
1122                ModuleSpec {
1123                    module_id: "held-module".to_string(),
1124                    program: "test-module".into(),
1125                    args: Vec::new(),
1126                    env: Vec::new(),
1127                    reserved: false,
1128                    reserved_prefixes: Vec::new(),
1129                    protocol: ModuleProtocol::Subc,
1130                    overlap: Default::default(),
1131                },
1132                false,
1133            )
1134            .expect("disabled test module is supervised");
1135        let router = Router::with_control_handler(Arc::new(
1136            ControlHandler::new(Arc::clone(&registry)).with_supervisor(handle),
1137        ));
1138        let (ctx, mut rx) = route_ctx();
1139        let (acquired, ready) = std_mpsc::channel();
1140        let holder = module.hold_snapshot_for_test(acquired, Duration::from_millis(400));
1141        ready.recv().expect("holder acquired snapshot lock");
1142        let (output, guard) = log_capture(tracing::Level::WARN);
1143
1144        router
1145            .route_for_connection(&ctx, request(0, 44, br#"{"op":"supervisor.list"}"#))
1146            .await
1147            .expect("list request routes after the lock releases");
1148        assert!(
1149            rx.recv().await.is_some(),
1150            "list request receives a response"
1151        );
1152        holder.join().expect("snapshot holder exits cleanly");
1153        drop(guard);
1154
1155        let logs = captured_logs(&output);
1156        assert!(logs.contains("slow snapshot lock"));
1157        assert!(logs.contains("module_id=held-module"));
1158        assert!(logs.contains("caller=list"));
1159        assert!(
1160            logged_millis(&logs, "waited_ms=") >= 250,
1161            "wait must exceed the slow-lock threshold: {logs}"
1162        );
1163
1164        let (output, guard) = log_capture(tracing::Level::WARN);
1165        router
1166            .route_for_connection(&ctx, request(0, 45, br#"{"op":"supervisor.list"}"#))
1167            .await
1168            .expect("uncontended list request routes");
1169        assert!(
1170            rx.recv().await.is_some(),
1171            "uncontended list receives a response"
1172        );
1173        drop(guard);
1174        assert!(
1175            !captured_logs(&output).contains("slow snapshot lock"),
1176            "uncontended list acquisition must not warn"
1177        );
1178    }
1179
1180    #[tokio::test]
1181    async fn full_module_to_client_sink_requests_client_close_without_erroring_module() {
1182        let forwarding = Arc::new(ForwardingTable::default());
1183        let control = Arc::new(ControlHandler::with_forwarding(
1184            Arc::new(crate::Registry::default()),
1185            Arc::clone(&forwarding),
1186        ));
1187        let router = Router::with_control_handler(control);
1188        let module_connection = ConnectionId::new(10);
1189        let client_connection = ConnectionId::new(20);
1190        let mut close_receiver = forwarding.register_connection_close(client_connection);
1191        let (module_tx, _module_rx) = mpsc::channel(1);
1192        forwarding
1193            .register_module_connection(
1194                module_connection,
1195                "full-sink-provider".to_string(),
1196                1,
1197                Concurrency::ModuleManaged,
1198                FrameSink::new(module_tx),
1199            )
1200            .unwrap();
1201        let (client_tx, mut client_rx) = mpsc::channel(1);
1202        let pending = forwarding
1203            .begin_route_bind_relay_for_test(
1204                client_connection,
1205                FrameSink::new(client_tx),
1206                700,
1207                "full-sink-provider",
1208            )
1209            .unwrap();
1210        forwarding
1211            .complete_pending_relay(
1212                module_connection,
1213                pending.corr,
1214                RouteBindRelayOutcome::Accepted,
1215            )
1216            .unwrap();
1217
1218        let (module_egress_tx, _module_egress_rx) = mpsc::channel(1);
1219        let module_ctx = RouteCtx {
1220            connection_id: module_connection,
1221            egress: FrameSink::new(module_egress_tx),
1222        };
1223        let terminal = Frame::build(
1224            FrameType::Response,
1225            Flags::new(false, Priority::Interactive, true),
1226            pending.module_channel,
1227            pending.module_epoch,
1228            701,
1229            b"terminal".to_vec(),
1230        )
1231        .unwrap();
1232
1233        router
1234            .route_for_connection(&module_ctx, terminal)
1235            .await
1236            .unwrap();
1237        let reason = tokio::time::timeout(Duration::from_secs(1), &mut close_receiver)
1238            .await
1239            .expect("close request should be sent for the full client sink")
1240            .expect("close sender should include a reason");
1241        assert!(
1242            reason
1243                .to_string()
1244                .contains("module_to_client_delivery_failed"),
1245            "unexpected close reason: {reason}"
1246        );
1247        assert_eq!(client_rx.try_recv().unwrap().header.corr, 700);
1248        assert!(client_rx.try_recv().is_err());
1249        assert_eq!(
1250            router.counters.snapshot()["client_egress_close_delivery_failed"],
1251            1
1252        );
1253    }
1254
1255    #[tokio::test]
1256    async fn full_route_goodbye_sink_requests_target_close_without_erroring_module() {
1257        let forwarding = Arc::new(ForwardingTable::default());
1258        let control = Arc::new(ControlHandler::with_forwarding(
1259            Arc::new(crate::Registry::default()),
1260            Arc::clone(&forwarding),
1261        ));
1262        let router = Router::with_control_handler(control);
1263        let module_connection = ConnectionId::new(30);
1264        let client_connection = ConnectionId::new(40);
1265        let mut close_receiver = forwarding.register_connection_close(client_connection);
1266        let (module_tx, _module_rx) = mpsc::channel(1);
1267        forwarding
1268            .register_module_connection(
1269                module_connection,
1270                "goodbye-full-provider".to_string(),
1271                1,
1272                Concurrency::ModuleManaged,
1273                FrameSink::new(module_tx),
1274            )
1275            .unwrap();
1276        let (client_tx, mut client_rx) = mpsc::channel(1);
1277        let pending = forwarding
1278            .begin_route_bind_relay_for_test(
1279                client_connection,
1280                FrameSink::new(client_tx),
1281                800,
1282                "goodbye-full-provider",
1283            )
1284            .unwrap();
1285        forwarding
1286            .complete_pending_relay(
1287                module_connection,
1288                pending.corr,
1289                RouteBindRelayOutcome::Accepted,
1290            )
1291            .unwrap();
1292
1293        let (module_egress_tx, _module_egress_rx) = mpsc::channel(1);
1294        let module_ctx = RouteCtx {
1295            connection_id: module_connection,
1296            egress: FrameSink::new(module_egress_tx),
1297        };
1298        let goodbye = Frame::build(
1299            FrameType::Goodbye,
1300            Flags::new(false, Priority::Passive, true),
1301            pending.module_channel,
1302            pending.module_epoch,
1303            801,
1304            Vec::new(),
1305        )
1306        .unwrap();
1307
1308        router
1309            .route_for_connection(&module_ctx, goodbye)
1310            .await
1311            .unwrap();
1312        let reason = tokio::time::timeout(Duration::from_secs(1), &mut close_receiver)
1313            .await
1314            .expect("close request should be sent for the full GOODBYE sink")
1315            .expect("close sender should include a reason");
1316        assert!(
1317            reason.to_string().contains("route_goodbye_delivery_failed"),
1318            "unexpected close reason: {reason}"
1319        );
1320        assert_eq!(client_rx.try_recv().unwrap().header.corr, 800);
1321        assert!(client_rx.try_recv().is_err());
1322        assert_eq!(router.counters.snapshot()["goodbye_relay_client_failed"], 1);
1323        assert_eq!(router.counters.snapshot()["route_released_epoch_fenced"], 1);
1324    }
1325
1326    fn route_frame(ty: FrameType, channel: u16, epoch: u32, corr: u64) -> Frame {
1327        Frame::build(
1328            ty,
1329            Flags::new(false, Priority::Interactive, false),
1330            channel,
1331            epoch,
1332            corr,
1333            if ty == FrameType::Request || ty == FrameType::Response {
1334                b"route-body".to_vec()
1335            } else {
1336                Vec::new()
1337            },
1338        )
1339        .unwrap()
1340    }
1341
1342    type DynamicRouteFixture = (
1343        Router,
1344        Arc<ForwardingTable>,
1345        RouteCtx,
1346        mpsc::Receiver<crate::router::OutboundFrame>,
1347        RouteCtx,
1348        mpsc::Receiver<crate::router::OutboundFrame>,
1349        mpsc::Receiver<crate::router::OutboundFrame>,
1350        crate::forwarding::PendingRouteBindRelay,
1351    );
1352
1353    fn dynamic_route_fixture(commit: bool) -> DynamicRouteFixture {
1354        let forwarding = Arc::new(ForwardingTable::default());
1355        let control = Arc::new(crate::ControlHandler::with_forwarding(
1356            Arc::new(crate::Registry::default()),
1357            Arc::clone(&forwarding),
1358        ));
1359        let router = Router::with_control_handler(control);
1360        let module_connection = ConnectionId::new(500);
1361        let client_connection = ConnectionId::new(501);
1362        let (module_tx, module_rx) = mpsc::channel(8);
1363        forwarding
1364            .register_module_connection(
1365                module_connection,
1366                "epoch-router".into(),
1367                2,
1368                Concurrency::ModuleManaged,
1369                FrameSink::new(module_tx),
1370            )
1371            .unwrap();
1372        let (client_tx, client_rx) = mpsc::channel(8);
1373        let client_sink = FrameSink::new(client_tx);
1374        let pending = forwarding
1375            .begin_route_bind_relay_for_test(
1376                client_connection,
1377                client_sink.clone(),
1378                700,
1379                "epoch-router",
1380            )
1381            .unwrap();
1382        if commit {
1383            forwarding
1384                .complete_pending_relay(
1385                    module_connection,
1386                    pending.corr,
1387                    RouteBindRelayOutcome::Accepted,
1388                )
1389                .unwrap();
1390        }
1391        let (module_egress_tx, module_egress_rx) = mpsc::channel(8);
1392        (
1393            router,
1394            forwarding,
1395            RouteCtx {
1396                connection_id: client_connection,
1397                egress: client_sink,
1398            },
1399            client_rx,
1400            RouteCtx {
1401                connection_id: module_connection,
1402                egress: FrameSink::new(module_egress_tx),
1403            },
1404            module_egress_rx,
1405            module_rx,
1406            pending,
1407        )
1408    }
1409
1410    #[tokio::test]
1411    async fn route_epochs_validate_both_directions_and_rewrite_to_peer_handle() {
1412        let (
1413            router,
1414            _forwarding,
1415            client_ctx,
1416            mut client_rx,
1417            module_ctx,
1418            _module_egress_rx,
1419            mut module_rx,
1420            pending,
1421        ) = dynamic_route_fixture(true);
1422        let route_open = client_rx.recv().await.unwrap();
1423        assert_eq!(route_open.header.corr, 700);
1424
1425        router
1426            .route_for_connection(
1427                &client_ctx,
1428                route_frame(
1429                    FrameType::Request,
1430                    pending.client_channel,
1431                    pending.client_epoch,
1432                    701,
1433                ),
1434            )
1435            .await
1436            .unwrap();
1437        let forwarded = module_rx.recv().await.unwrap();
1438        assert_eq!(forwarded.header.channel, pending.module_channel);
1439        assert_eq!(forwarded.header.epoch, pending.module_epoch);
1440
1441        router
1442            .route_for_connection(
1443                &module_ctx,
1444                route_frame(
1445                    FrameType::Response,
1446                    pending.module_channel,
1447                    pending.module_epoch,
1448                    701,
1449                ),
1450            )
1451            .await
1452            .unwrap();
1453        let delivered = client_rx.recv().await.unwrap();
1454        assert_eq!(delivered.header.channel, pending.client_channel);
1455        assert_eq!(delivered.header.epoch, pending.client_epoch);
1456
1457        router
1458            .route_for_connection(
1459                &client_ctx,
1460                route_frame(
1461                    FrameType::Request,
1462                    pending.client_channel,
1463                    pending.client_epoch + 1,
1464                    702,
1465                ),
1466            )
1467            .await
1468            .unwrap();
1469        router
1470            .route_for_connection(
1471                &module_ctx,
1472                route_frame(
1473                    FrameType::Response,
1474                    pending.module_channel,
1475                    pending.module_epoch + 1,
1476                    703,
1477                ),
1478            )
1479            .await
1480            .unwrap();
1481        let stale_error = client_rx.recv().await.unwrap();
1482        assert_eq!(stale_error.header.ty, FrameType::Error);
1483        assert_eq!(stale_error.header.channel, pending.client_channel);
1484        assert_eq!(stale_error.header.epoch, pending.client_epoch + 1);
1485        assert_eq!(stale_error.header.corr, 702);
1486        let body: ErrorBody = serde_json::from_slice(&stale_error.body).unwrap();
1487        assert_eq!(body.code, "stale_route_epoch");
1488        assert!(module_rx.try_recv().is_err());
1489        assert!(client_rx.try_recv().is_err());
1490        let counters = router.counters.snapshot();
1491        assert_eq!(counters["client_frames_dropped_stale_route"], 1);
1492        assert_eq!(counters["module_frames_dropped_no_route"], 1);
1493    }
1494
1495    #[tokio::test]
1496    async fn accepted_route_publishes_route_open_before_immediate_reverse_request() {
1497        let (
1498            router,
1499            _,
1500            _client_ctx,
1501            mut client_rx,
1502            module_ctx,
1503            _module_egress_rx,
1504            _module_rx,
1505            pending,
1506        ) = dynamic_route_fixture(true);
1507        router
1508            .route_for_connection(
1509                &module_ctx,
1510                route_frame(
1511                    FrameType::Request,
1512                    pending.module_channel,
1513                    pending.module_epoch,
1514                    800,
1515                ),
1516            )
1517            .await
1518            .unwrap();
1519
1520        let first = client_rx.recv().await.unwrap();
1521        let second = client_rx.recv().await.unwrap();
1522        assert_eq!(first.header.channel, 0);
1523        assert_eq!(first.header.corr, 700);
1524        assert_eq!(second.header.channel, pending.client_channel);
1525        assert_eq!(second.header.epoch, pending.client_epoch);
1526        assert_eq!(second.header.corr, 800);
1527    }
1528
1529    #[tokio::test]
1530    async fn reserved_slot_ingress_errors_only_matching_client_requests() {
1531        let (
1532            router,
1533            _forwarding,
1534            client_ctx,
1535            mut client_rx,
1536            _module_ctx,
1537            _module_egress_rx,
1538            mut module_rx,
1539            pending,
1540        ) = dynamic_route_fixture(false);
1541        router
1542            .route_for_connection(
1543                &client_ctx,
1544                route_frame(
1545                    FrameType::Request,
1546                    pending.client_channel,
1547                    pending.client_epoch,
1548                    900,
1549                ),
1550            )
1551            .await
1552            .unwrap();
1553        let error = client_rx.recv().await.unwrap();
1554        assert_eq!(error.header.ty, FrameType::Error);
1555        assert_eq!(error.header.channel, pending.client_channel);
1556        assert_eq!(error.header.epoch, pending.client_epoch);
1557        assert_eq!(error.header.corr, 900);
1558
1559        router
1560            .route_for_connection(
1561                &client_ctx,
1562                route_frame(
1563                    FrameType::Response,
1564                    pending.client_channel,
1565                    pending.client_epoch,
1566                    901,
1567                ),
1568            )
1569            .await
1570            .unwrap();
1571        router
1572            .route_for_connection(
1573                &client_ctx,
1574                route_frame(
1575                    FrameType::Request,
1576                    pending.client_channel,
1577                    pending.client_epoch + 1,
1578                    902,
1579                ),
1580            )
1581            .await
1582            .unwrap();
1583        let stale_error = client_rx.recv().await.unwrap();
1584        assert_eq!(stale_error.header.ty, FrameType::Error);
1585        assert_eq!(stale_error.header.channel, pending.client_channel);
1586        assert_eq!(stale_error.header.epoch, pending.client_epoch + 1);
1587        assert_eq!(stale_error.header.corr, 902);
1588        let body: ErrorBody = serde_json::from_slice(&stale_error.body).unwrap();
1589        assert_eq!(body.code, "stale_route_epoch");
1590        assert!(module_rx.try_recv().is_err());
1591        let counters = router.counters.snapshot();
1592        assert_eq!(counters["client_frames_dropped_stale_route"], 1);
1593        assert_eq!(counters["module_frames_dropped_no_route"], 0);
1594    }
1595
1596    #[tokio::test]
1597    async fn dropped_module_route_goodbye_increments_counter() {
1598        let (
1599            router,
1600            _forwarding,
1601            client_ctx,
1602            mut client_rx,
1603            _module_ctx,
1604            _module_egress_rx,
1605            mut module_rx,
1606            pending,
1607        ) = dynamic_route_fixture(true);
1608        let _ = client_rx.recv().await;
1609        module_rx.close();
1610
1611        router
1612            .route_for_connection(
1613                &client_ctx,
1614                route_frame(
1615                    FrameType::Goodbye,
1616                    pending.client_channel,
1617                    pending.client_epoch,
1618                    999,
1619                ),
1620            )
1621            .await
1622            .unwrap();
1623
1624        let counters = router.counters.snapshot();
1625        assert_eq!(counters["goodbye_relay_module_dropped"], 1);
1626        assert_eq!(
1627            counters["goodbye_relay_module_dropped_by_module"],
1628            serde_json::json!({ "epoch-router": 1 })
1629        );
1630        assert_eq!(counters["route_released_epoch_fenced"], 1);
1631    }
1632
1633    #[tokio::test]
1634    async fn module_request_on_stale_epoch_receives_stale_route_epoch() {
1635        let (
1636            router,
1637            _forwarding,
1638            _client_ctx,
1639            _client_rx,
1640            module_ctx,
1641            mut module_egress_rx,
1642            mut module_rx,
1643            pending,
1644        ) = dynamic_route_fixture(true);
1645
1646        router
1647            .route_for_connection(
1648                &module_ctx,
1649                route_frame(
1650                    FrameType::Request,
1651                    pending.module_channel,
1652                    pending.module_epoch + 1,
1653                    1_000,
1654                ),
1655            )
1656            .await
1657            .unwrap();
1658
1659        let error = module_egress_rx.try_recv().unwrap();
1660        assert_eq!(error.header.ty, FrameType::Error);
1661        assert_eq!(error.header.channel, pending.module_channel);
1662        assert_eq!(error.header.epoch, pending.module_epoch + 1);
1663        assert_eq!(error.header.corr, 1_000);
1664        let body: ErrorBody = serde_json::from_slice(&error.body).unwrap();
1665        assert_eq!(body.code, "stale_route_epoch");
1666        assert!(module_rx.try_recv().is_err());
1667        let counters = router.counters.snapshot();
1668        assert_eq!(counters["module_requests_dropped_stale_route"], 1);
1669        assert_eq!(counters["module_frames_dropped_no_route"], 0);
1670    }
1671
1672    #[tokio::test]
1673    async fn module_request_on_reserved_or_absent_route_receives_unknown_channel() {
1674        let (
1675            reserved_router,
1676            _forwarding,
1677            _client_ctx,
1678            _client_rx,
1679            reserved_module_ctx,
1680            mut reserved_module_egress_rx,
1681            _module_rx,
1682            reserved,
1683        ) = dynamic_route_fixture(false);
1684        reserved_router
1685            .route_for_connection(
1686                &reserved_module_ctx,
1687                route_frame(
1688                    FrameType::Request,
1689                    reserved.module_channel,
1690                    reserved.module_epoch,
1691                    1_001,
1692                ),
1693            )
1694            .await
1695            .unwrap();
1696        let reserved_error = reserved_module_egress_rx.try_recv().unwrap();
1697        let reserved_body: ErrorBody = serde_json::from_slice(&reserved_error.body).unwrap();
1698        assert_eq!(reserved_error.header.ty, FrameType::Error);
1699        assert_eq!(reserved_error.header.channel, reserved.module_channel);
1700        assert_eq!(reserved_error.header.epoch, reserved.module_epoch);
1701        assert_eq!(reserved_error.header.corr, 1_001);
1702        assert_eq!(reserved_body.code, "unknown_channel");
1703        assert_eq!(
1704            reserved_router.counters.snapshot()["module_requests_dropped_stale_route"],
1705            1
1706        );
1707
1708        let (
1709            absent_router,
1710            _forwarding,
1711            _client_ctx,
1712            _client_rx,
1713            absent_module_ctx,
1714            mut absent_module_egress_rx,
1715            _module_rx,
1716            absent,
1717        ) = dynamic_route_fixture(false);
1718        absent_router
1719            .route_for_connection(
1720                &absent_module_ctx,
1721                route_frame(
1722                    FrameType::Request,
1723                    absent.module_channel + 1,
1724                    absent.module_epoch,
1725                    1_002,
1726                ),
1727            )
1728            .await
1729            .unwrap();
1730        let absent_error = absent_module_egress_rx.try_recv().unwrap();
1731        let absent_body: ErrorBody = serde_json::from_slice(&absent_error.body).unwrap();
1732        assert_eq!(absent_error.header.ty, FrameType::Error);
1733        assert_eq!(absent_error.header.channel, absent.module_channel + 1);
1734        assert_eq!(absent_error.header.epoch, absent.module_epoch);
1735        assert_eq!(absent_error.header.corr, 1_002);
1736        assert_eq!(absent_body.code, "unknown_channel");
1737        assert_eq!(
1738            absent_router.counters.snapshot()["module_requests_dropped_stale_route"],
1739            1
1740        );
1741    }
1742
1743    #[tokio::test]
1744    async fn non_request_module_frame_on_dead_route_is_counted_without_error() {
1745        let (
1746            router,
1747            forwarding,
1748            client_ctx,
1749            mut client_rx,
1750            module_ctx,
1751            mut module_egress_rx,
1752            mut module_rx,
1753            pending,
1754        ) = dynamic_route_fixture(true);
1755        let (other_module_tx, _other_module_rx) = mpsc::channel(8);
1756        forwarding
1757            .register_module_connection(
1758                ConnectionId::new(502),
1759                "other-module".into(),
1760                2,
1761                Concurrency::ModuleManaged,
1762                FrameSink::new(other_module_tx),
1763            )
1764            .unwrap();
1765        let _ = client_rx.recv().await.unwrap();
1766
1767        router
1768            .route_for_connection(
1769                &client_ctx,
1770                route_frame(
1771                    FrameType::Goodbye,
1772                    pending.client_channel,
1773                    pending.client_epoch,
1774                    1_003,
1775                ),
1776            )
1777            .await
1778            .unwrap();
1779        let _ = module_rx.recv().await.unwrap();
1780
1781        router
1782            .route_for_connection(
1783                &module_ctx,
1784                route_frame(
1785                    FrameType::StreamData,
1786                    pending.module_channel,
1787                    pending.module_epoch,
1788                    1_004,
1789                ),
1790            )
1791            .await
1792            .unwrap();
1793
1794        assert!(module_egress_rx.try_recv().is_err());
1795        let counters = router.counters.snapshot();
1796        assert_eq!(counters["module_frames_dropped_no_route"], 1);
1797        assert_eq!(
1798            counters["module_frames_dropped_no_route_by_module"],
1799            serde_json::json!({ "epoch-router": 1 })
1800        );
1801        assert_eq!(counters["module_requests_dropped_stale_route"], 0);
1802    }
1803
1804    #[test]
1805    fn channel_zero_cannot_be_registered_as_backend() {
1806        let mut router = Router::with_default_self_handler();
1807
1808        let err = router.register_backend(0, EchoBackend).unwrap_err();
1809
1810        assert_eq!(err, RouterError::ReservedChannelZero);
1811    }
1812}
subc_daemon/router.rs

subc_daemon/
router.rs
