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                // A terminal frame ends the request at the module whether or not
493                // the client can still take it, so its credit is released before
494                // delivery is attempted. Releasing only after a successful
495                // enqueue would leave a drain counting a finished request until
496                // the client connection's cleanup removes the route.
497                let releases_credit = is_terminal_frame(frame.header.ty);
498                if releases_credit {
499                    route.flow.release_corr(corr);
500                }
501                let mut frame = frame;
502                frame.header.channel = route.client_channel;
503                frame.header.epoch = route.client_epoch;
504                if let Err(err) = route.client_sink.try_send(frame) {
505                    if self
506                        .forwarding
507                        .escalate_client_delivery_failure(
508                            route.client_connection_id,
509                            route.client_channel,
510                            route.client_epoch,
511                            CloseReason::new(
512                                "module_to_client_delivery_failed",
513                                format!(
514                                    "failed to enqueue module frame for client channel {} corr {corr}: {err}",
515                                    route.client_channel
516                                ),
517                            ),
518                        )
519                        .map_err(RouterError::Forwarding)?
520                    {
521                        self.counters
522                            .increment_client_egress_close_delivery_failed();
523                    }
524                    return Ok(());
525                }
526                return Ok(());
527            }
528            DataRoute::Client(DataRouteState::EpochMismatch) => {
529                if frame.header.ty == FrameType::Request {
530                    self.counters.increment_client_frames_dropped_stale_route();
531                    // Dropped before forwarding; a re-bind retry cannot double-execute this request.
532                    let err = RouterError::StaleRouteEpoch {
533                        channel,
534                        epoch,
535                        corr,
536                    };
537                    if let Some(error_frame) = err.to_error_frame() {
538                        ctx.egress.send(error_frame).await?;
539                    }
540                }
541                debug!(
542                    connection_id = ctx.connection_id.get(),
543                    channel, epoch, corr, "dropping client frame for stale route epoch"
544                );
545                return Ok(());
546            }
547            DataRoute::Client(DataRouteState::Reserved) => {
548                if frame.header.ty == FrameType::Request {
549                    let err = RouterError::UnknownChannel {
550                        channel,
551                        epoch,
552                        corr,
553                    };
554                    if let Some(error_frame) = err.to_error_frame() {
555                        ctx.egress.send(error_frame).await?;
556                    }
557                }
558                return Ok(());
559            }
560            DataRoute::Client(DataRouteState::Bound(route)) => {
561                if frame.header.ty == FrameType::Goodbye {
562                    let _ = self
563                        .control
564                        .handle_route_goodbye(ctx.connection_id, channel, epoch)?;
565                    return Ok(());
566                }
567                return self.forward_backend.handle_bound(frame, route).await;
568            }
569            DataRoute::Client(DataRouteState::Absent) => {}
570        }
571
572        if let Some(backend) = self.backends.get(&channel) {
573            return backend.handle(ctx.clone(), frame).await;
574        }
575        if frame.header.ty == FrameType::Request {
576            let err = RouterError::UnknownChannel {
577                channel,
578                epoch,
579                corr,
580            };
581            if let Some(error_frame) = err.to_error_frame() {
582                ctx.egress.send(error_frame).await?;
583            }
584        }
585        Ok(())
586    }
587}
588
589impl Default for Router {
590    fn default() -> Self {
591        Self::with_default_self_handler()
592    }
593}
594
595/// Connection-scoped cleanup guard returned by [`Router::begin_connection`].
596#[must_use]
597pub struct RouterConnection {
598    id: ConnectionId,
599    control_handler: Arc<ControlHandler>,
600    forwarding: Arc<ForwardingTable>,
601    close_receiver: Option<ConnectionCloseReceiver>,
602}
603
604impl RouterConnection {
605    pub fn id(&self) -> ConnectionId {
606        self.id
607    }
608
609    pub(crate) fn take_close_receiver(&mut self) -> ConnectionCloseReceiver {
610        self.close_receiver
611            .take()
612            .expect("connection close receiver can only be taken once")
613    }
614}
615
616impl Drop for RouterConnection {
617    fn drop(&mut self) {
618        self.forwarding.unregister_connection_close(self.id);
619        // GOODBYE (explicit) and connection-drop cleanup both call the same
620        // idempotent deregistration path.
621        let _ = self.control_handler.cleanup_connection(self.id);
622    }
623}
624
625/// Minimal in-memory backend used by tests and early wiring: it emits a
626/// `RESPONSE` on the same channel/correlation id with the exact same body bytes.
627#[derive(Debug, Default, Clone, Copy)]
628pub struct EchoBackend;
629
630impl EchoBackend {
631    pub async fn handle(&self, ctx: RouteCtx, frame: Frame) -> Result<(), RouterError> {
632        let response = Frame::build_with_version(
633            frame.header.ver,
634            FrameType::Response,
635            frame.header.flags,
636            frame.header.channel,
637            frame.header.epoch,
638            frame.header.corr,
639            frame.body,
640        )
641        .map_err(RouterError::FrameBuild)?;
642        ctx.egress.send(response).await
643    }
644}
645
646/// Data-plane backend that splices client frames to the module connection bound at attach time.
647#[derive(Debug, Clone)]
648pub struct ForwardBackend {
649    forwarding: Arc<ForwardingTable>,
650}
651
652impl ForwardBackend {
653    pub fn new(forwarding: Arc<ForwardingTable>) -> Self {
654        Self { forwarding }
655    }
656
657    pub async fn handle(&self, ctx: RouteCtx, frame: Frame) -> Result<(), RouterError> {
658        let channel = frame.header.channel;
659        let corr = frame.header.corr;
660        let route = match self
661            .forwarding
662            .lookup_data_route(ctx.connection_id, channel, frame.header.epoch)
663            .map_err(RouterError::Forwarding)?
664        {
665            DataRoute::Client(DataRouteState::Bound(route)) => route,
666            DataRoute::Client(_) | DataRoute::Module(_) => {
667                return Err(RouterError::UnknownChannel {
668                    channel,
669                    epoch: frame.header.epoch,
670                    corr,
671                });
672            }
673        };
674        self.handle_bound(frame, route).await
675    }
676
677    pub(crate) async fn handle_bound(
678        &self,
679        frame: Frame,
680        route: Arc<RouteBinding>,
681    ) -> Result<(), RouterError> {
682        let channel = frame.header.channel;
683        let corr = frame.header.corr;
684        let frame_type = frame.header.ty;
685
686        // CANCEL and other non-REQUEST frames bypass the request-credit window;
687        // the original request's credit is freed only by the module's terminal frame.
688        let acquired_credit = frame_type == FrameType::Request;
689        if acquired_credit {
690            if let Err(err) = route
691                .flow
692                .acquire_tagged(corr, frame.header.flags.is_subscription())
693                .await
694            {
695                // `module_reloading` here is answered BEFORE the frame is
696                // forwarded, so the module never sees the request and callers
697                // may re-dispatch it after reopening the route. That is a wire
698                // guarantee documented on `error_codes::MODULE_RELOADING`; do
699                // not emit this code for a request that already reached
700                // `module_sink.send` below.
701                if self
702                    .forwarding
703                    .endpoint_is_draining(route.module_endpoint)
704                    .map_err(RouterError::Forwarding)?
705                {
706                    return Err(RouterError::route_error_with_epoch(
707                        channel,
708                        frame.header.epoch,
709                        corr,
710                        "module_reloading",
711                        format!("module endpoint for route channel {channel} is reloading"),
712                    ));
713                }
714                return Err(RouterError::backend_with_epoch(
715                    channel,
716                    frame.header.epoch,
717                    corr,
718                    format!("{err} for route channel {channel}"),
719                ));
720            }
721        }
722
723        let mut frame = frame;
724        frame.header.channel = route.module_channel;
725        frame.header.epoch = route.module_epoch;
726        let result = route.module_sink.send(frame).await.map_err(|err| {
727            RouterError::backend_with_epoch(channel, route.client_epoch, corr, err.to_string())
728        });
729        if acquired_credit && result.is_err() {
730            route.flow.release_corr(corr);
731        }
732        result
733    }
734}
735
736fn is_terminal_frame(frame_type: FrameType) -> bool {
737    matches!(
738        frame_type,
739        FrameType::Response | FrameType::Error | FrameType::StreamEnd
740    )
741}
742
743/// Typed router errors. Routable failures can be translated to canonical JSON
744/// `ERROR` frames with [`RouterError::to_error_frame`].
745#[derive(Debug, Clone, PartialEq, Eq)]
746pub enum RouterError {
747    ReservedChannelZero,
748    DuplicateChannel {
749        channel: u16,
750    },
751    UnknownChannel {
752        channel: u16,
753        epoch: u32,
754        corr: u64,
755    },
756    StaleRouteEpoch {
757        channel: u16,
758        epoch: u32,
759        corr: u64,
760    },
761    Backend {
762        channel: u16,
763        epoch: u32,
764        corr: u64,
765        message: String,
766    },
767    RouteError {
768        channel: u16,
769        epoch: u32,
770        corr: u64,
771        code: String,
772        message: String,
773    },
774    FrameBuild(FrameBuildError),
775    Forwarding(ForwardingError),
776}
777
778impl RouterError {
779    pub fn backend(channel: u16, corr: u64, message: impl Into<String>) -> Self {
780        Self::backend_with_epoch(channel, 0, corr, message)
781    }
782
783    pub fn backend_with_epoch(
784        channel: u16,
785        epoch: u32,
786        corr: u64,
787        message: impl Into<String>,
788    ) -> Self {
789        Self::Backend {
790            channel,
791            epoch,
792            corr,
793            message: message.into(),
794        }
795    }
796
797    pub fn route_error(
798        channel: u16,
799        corr: u64,
800        code: impl Into<String>,
801        message: impl Into<String>,
802    ) -> Self {
803        Self::route_error_with_epoch(channel, 0, corr, code, message)
804    }
805
806    pub fn route_error_with_epoch(
807        channel: u16,
808        epoch: u32,
809        corr: u64,
810        code: impl Into<String>,
811        message: impl Into<String>,
812    ) -> Self {
813        Self::RouteError {
814            channel,
815            epoch,
816            corr,
817            code: code.into(),
818            message: message.into(),
819        }
820    }
821
822    /// Translate route failures that belong on the wire into an `ERROR` frame.
823    pub fn to_error_frame(&self) -> Option<Frame> {
824        match self {
825            Self::UnknownChannel {
826                channel,
827                epoch,
828                corr,
829            } => error_frame(
830                *channel,
831                *epoch,
832                *corr,
833                "unknown_channel",
834                format!("unknown channel {channel}"),
835            ),
836            Self::StaleRouteEpoch {
837                channel,
838                epoch,
839                corr,
840            } => error_frame(
841                *channel,
842                *epoch,
843                *corr,
844                "stale_route_epoch",
845                format!("stale route epoch for channel {channel}"),
846            ),
847            Self::Backend {
848                channel,
849                epoch,
850                corr,
851                message,
852            } => error_frame(*channel, *epoch, *corr, "backend_error", message.clone()),
853            Self::RouteError {
854                channel,
855                epoch,
856                corr,
857                code,
858                message,
859            } => error_frame(*channel, *epoch, *corr, code, message.clone()),
860            Self::ReservedChannelZero
861            | Self::DuplicateChannel { .. }
862            | Self::FrameBuild(_)
863            | Self::Forwarding(_) => None,
864        }
865    }
866}
867
868fn error_frame(channel: u16, epoch: u32, corr: u64, code: &str, message: String) -> Option<Frame> {
869    let body = serde_json::to_vec(&ErrorBody {
870        code: code.to_string(),
871        message,
872        detail: None,
873    })
874    .ok()?;
875
876    Frame::build(
877        FrameType::Error,
878        Flags::new(false, Priority::Passive, false),
879        channel,
880        epoch,
881        corr,
882        body,
883    )
884    .ok()
885}
886
887impl fmt::Display for RouterError {
888    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
889        match self {
890            Self::ReservedChannelZero => write!(f, "channel 0 is reserved for subc"),
891            Self::DuplicateChannel { channel } => {
892                write!(f, "backend already registered for channel {channel}")
893            }
894            Self::UnknownChannel { channel, corr, .. } => {
895                write!(f, "unknown channel {channel} for corr {corr}")
896            }
897            Self::StaleRouteEpoch { channel, corr, .. } => {
898                write!(f, "stale route epoch for channel {channel} corr {corr}")
899            }
900            Self::Backend {
901                channel,
902                corr,
903                message,
904                ..
905            } => write!(
906                f,
907                "backend error on channel {channel} corr {corr}: {message}"
908            ),
909            Self::RouteError {
910                channel,
911                corr,
912                code,
913                message,
914                ..
915            } => write!(
916                f,
917                "route error {code} on channel {channel} corr {corr}: {message}"
918            ),
919            Self::FrameBuild(err) => write!(f, "failed to build routed frame: {err}"),
920            Self::Forwarding(err) => write!(f, "forwarding error: {err}"),
921        }
922    }
923}
924
925impl Error for RouterError {
926    fn source(&self) -> Option<&(dyn Error + 'static)> {
927        match self {
928            Self::FrameBuild(err) => Some(err),
929            Self::Forwarding(err) => Some(err),
930            Self::ReservedChannelZero
931            | Self::DuplicateChannel { .. }
932            | Self::UnknownChannel { .. }
933            | Self::StaleRouteEpoch { .. }
934            | Self::Backend { .. }
935            | Self::RouteError { .. } => None,
936        }
937    }
938}
939
940#[cfg(test)]
941mod tests {
942    use super::*;
943    use crate::{
944        forwarding::RouteBindRelayOutcome,
945        supervise::{ModuleSpec, RestartPolicy, Supervisor, SupervisorHandle},
946        ControlHandler, Registry,
947    };
948    use std::{
949        sync::{mpsc as std_mpsc, Arc},
950        time::Duration,
951    };
952    use subc_control::ModuleProtocol;
953    use subc_protocol::{manifest::Concurrency, ErrorBody, Flags, FrameType, Priority};
954    use tokio::sync::mpsc;
955
956    pub(crate) use crate::router::test_log::{captured_logs, log_capture};
957
958    fn logged_millis(logs: &str, field: &str) -> u64 {
959        logs.split_whitespace()
960            .find_map(|part| part.strip_prefix(field))
961            .and_then(|value| value.parse().ok())
962            .unwrap_or_else(|| panic!("missing numeric {field} in logs: {logs}"))
963    }
964
965    fn request(channel: u16, corr: u64, body: &[u8]) -> Frame {
966        Frame::build(
967            FrameType::Request,
968            Flags::new(true, Priority::Interactive, false),
969            channel,
970            0,
971            corr,
972            body.to_vec(),
973        )
974        .unwrap()
975    }
976
977    fn ping(corr: u64) -> Frame {
978        Frame::build(
979            FrameType::Ping,
980            Flags::new(false, Priority::Passive, false),
981            0,
982            0,
983            corr,
984            Vec::new(),
985        )
986        .unwrap()
987    }
988
989    fn route_ctx() -> (RouteCtx, mpsc::Receiver<crate::router::OutboundFrame>) {
990        let (tx, rx) = mpsc::channel(8);
991        (
992            RouteCtx {
993                connection_id: ConnectionId::LOCAL,
994                egress: FrameSink::new(tx),
995            },
996            rx,
997        )
998    }
999
1000    #[tokio::test]
1001    async fn echo_backend_returns_response_with_byte_identical_body() {
1002        let mut router = Router::with_default_self_handler();
1003        router.register_backend(7, EchoBackend).unwrap();
1004        let (ctx, mut rx) = route_ctx();
1005        let body = b"{not parsed}\0\xff";
1006
1007        router
1008            .route_for_connection(&ctx, request(7, 123, body))
1009            .await
1010            .unwrap();
1011        let response = rx.recv().await.unwrap();
1012
1013        assert_eq!(response.header.ty, FrameType::Response);
1014        assert_eq!(response.header.channel, 7);
1015        assert_eq!(response.header.corr, 123);
1016        assert_eq!(response.body, body);
1017        assert!(rx.try_recv().is_err());
1018    }
1019
1020    #[tokio::test]
1021    async fn unknown_channel_emits_canonical_error_frame() {
1022        let router = Router::with_default_self_handler();
1023        let (ctx, mut rx) = route_ctx();
1024
1025        router
1026            .route_for_connection(&ctx, request(99, 5, b"payload"))
1027            .await
1028            .unwrap();
1029        let error_frame = rx.recv().await.unwrap();
1030
1031        assert_eq!(error_frame.header.ty, FrameType::Error);
1032        assert_eq!(error_frame.header.channel, 99);
1033        assert_eq!(error_frame.header.corr, 5);
1034        let body: ErrorBody = serde_json::from_slice(&error_frame.body).unwrap();
1035        assert_eq!(body.code, "unknown_channel");
1036        assert_eq!(body.message, "unknown channel 99");
1037    }
1038
1039    #[tokio::test]
1040    async fn channel_zero_uses_control_handler_not_backend_registry() {
1041        let mut router = Router::with_default_self_handler();
1042        router.register_backend(1, EchoBackend).unwrap();
1043        let (ctx, mut rx) = route_ctx();
1044
1045        router.route_for_connection(&ctx, ping(77)).await.unwrap();
1046        let response = rx.recv().await.unwrap();
1047
1048        assert_eq!(response.header.ty, FrameType::Pong);
1049        assert_eq!(response.header.channel, 0);
1050        assert_eq!(response.header.corr, 77);
1051        assert!(response.body.is_empty());
1052    }
1053
1054    #[tokio::test]
1055    async fn slow_control_dispatch_logs_decoded_op_and_elapsed_time() {
1056        let control = Arc::new(
1057            ControlHandler::new(Arc::new(Registry::default()))
1058                .with_control_dispatch_delay(Duration::from_millis(1050)),
1059        );
1060        let router = Router::with_control_handler(control);
1061        let (ctx, mut rx) = route_ctx();
1062        let (output, guard) = log_capture(tracing::Level::WARN);
1063
1064        router
1065            .route_for_connection(&ctx, request(0, 41, br#"{"op":"server.describe"}"#))
1066            .await
1067            .expect("slow request routes");
1068        assert!(rx.recv().await.is_some(), "request receives a response");
1069        drop(guard);
1070
1071        let logs = captured_logs(&output);
1072        assert!(logs.contains("slow control dispatch"));
1073        assert!(logs.contains("op=server.describe"));
1074        assert!(logs.contains("connection_id=0"));
1075        assert!(logs.contains("corr=41"));
1076        assert!(
1077            logged_millis(&logs, "elapsed_ms=") >= 1050,
1078            "elapsed must include the injected handler delay: {logs}"
1079        );
1080    }
1081
1082    #[tokio::test]
1083    async fn fast_control_dispatch_emits_arrival_without_slow_warning() {
1084        let router = Router::with_default_self_handler();
1085        let (ctx, mut rx) = route_ctx();
1086        let (output, guard) = log_capture(tracing::Level::DEBUG);
1087
1088        router
1089            .route_for_connection(&ctx, request(0, 42, br#"{"op":"server.describe"}"#))
1090            .await
1091            .expect("fast request routes");
1092        assert!(rx.recv().await.is_some(), "request receives a response");
1093        drop(guard);
1094
1095        let logs = captured_logs(&output);
1096        assert!(logs.contains("control dispatch op=server.describe connection_id=0 corr=42"));
1097        assert!(!logs.contains("slow control dispatch"));
1098    }
1099
1100    #[tokio::test]
1101    async fn control_dispatch_arrival_is_hidden_at_info() {
1102        let router = Router::with_default_self_handler();
1103        let (ctx, mut rx) = route_ctx();
1104        let (output, guard) = log_capture(tracing::Level::INFO);
1105
1106        router
1107            .route_for_connection(&ctx, request(0, 43, br#"{"op":"server.describe"}"#))
1108            .await
1109            .expect("fast request routes");
1110        assert!(rx.recv().await.is_some(), "request receives a response");
1111        drop(guard);
1112
1113        assert!(
1114            !captured_logs(&output).contains("control dispatch"),
1115            "arrival logging must stay hidden at INFO"
1116        );
1117    }
1118
1119    #[tokio::test]
1120    async fn supervisor_list_logs_contended_snapshot_lock_only() {
1121        let registry = Arc::new(Registry::default());
1122        let handle = SupervisorHandle::new();
1123        let supervisor = Supervisor::new(Arc::clone(&registry), RestartPolicy::default())
1124            .with_handle(handle.clone());
1125        let module = supervisor
1126            .supervise_configured(
1127                ModuleSpec {
1128                    module_id: "held-module".to_string(),
1129                    program: "test-module".into(),
1130                    args: Vec::new(),
1131                    env: Vec::new(),
1132                    reserved: false,
1133                    reserved_prefixes: Vec::new(),
1134                    protocol: ModuleProtocol::Subc,
1135                    overlap: Default::default(),
1136                },
1137                false,
1138            )
1139            .expect("disabled test module is supervised");
1140        let router = Router::with_control_handler(Arc::new(
1141            ControlHandler::new(Arc::clone(&registry)).with_supervisor(handle),
1142        ));
1143        let (ctx, mut rx) = route_ctx();
1144        let (acquired, ready) = std_mpsc::channel();
1145        let holder = module.hold_snapshot_for_test(acquired, Duration::from_millis(400));
1146        ready.recv().expect("holder acquired snapshot lock");
1147        let (output, guard) = log_capture(tracing::Level::WARN);
1148
1149        router
1150            .route_for_connection(&ctx, request(0, 44, br#"{"op":"supervisor.list"}"#))
1151            .await
1152            .expect("list request routes after the lock releases");
1153        assert!(
1154            rx.recv().await.is_some(),
1155            "list request receives a response"
1156        );
1157        holder.join().expect("snapshot holder exits cleanly");
1158        drop(guard);
1159
1160        let logs = captured_logs(&output);
1161        assert!(logs.contains("slow snapshot lock"));
1162        assert!(logs.contains("module_id=held-module"));
1163        assert!(logs.contains("caller=list"));
1164        assert!(
1165            logged_millis(&logs, "waited_ms=") >= 250,
1166            "wait must exceed the slow-lock threshold: {logs}"
1167        );
1168
1169        let (output, guard) = log_capture(tracing::Level::WARN);
1170        router
1171            .route_for_connection(&ctx, request(0, 45, br#"{"op":"supervisor.list"}"#))
1172            .await
1173            .expect("uncontended list request routes");
1174        assert!(
1175            rx.recv().await.is_some(),
1176            "uncontended list receives a response"
1177        );
1178        drop(guard);
1179        assert!(
1180            !captured_logs(&output).contains("slow snapshot lock"),
1181            "uncontended list acquisition must not warn"
1182        );
1183    }
1184
1185    #[tokio::test]
1186    async fn full_module_to_client_sink_requests_client_close_without_erroring_module() {
1187        let forwarding = Arc::new(ForwardingTable::default());
1188        let control = Arc::new(ControlHandler::with_forwarding(
1189            Arc::new(crate::Registry::default()),
1190            Arc::clone(&forwarding),
1191        ));
1192        let router = Router::with_control_handler(control);
1193        let module_connection = ConnectionId::new(10);
1194        let client_connection = ConnectionId::new(20);
1195        let mut close_receiver = forwarding.register_connection_close(client_connection);
1196        let (module_tx, _module_rx) = mpsc::channel(1);
1197        forwarding
1198            .register_module_connection(
1199                module_connection,
1200                "full-sink-provider".to_string(),
1201                1,
1202                Concurrency::ModuleManaged,
1203                FrameSink::new(module_tx),
1204            )
1205            .unwrap();
1206        let (client_tx, mut client_rx) = mpsc::channel(1);
1207        let pending = forwarding
1208            .begin_route_bind_relay_for_test(
1209                client_connection,
1210                FrameSink::new(client_tx),
1211                700,
1212                "full-sink-provider",
1213            )
1214            .unwrap();
1215        forwarding
1216            .complete_pending_relay(
1217                module_connection,
1218                pending.corr,
1219                RouteBindRelayOutcome::Accepted,
1220            )
1221            .unwrap();
1222
1223        let (module_egress_tx, _module_egress_rx) = mpsc::channel(1);
1224        let module_ctx = RouteCtx {
1225            connection_id: module_connection,
1226            egress: FrameSink::new(module_egress_tx),
1227        };
1228        let terminal = Frame::build(
1229            FrameType::Response,
1230            Flags::new(false, Priority::Interactive, true),
1231            pending.module_channel,
1232            pending.module_epoch,
1233            701,
1234            b"terminal".to_vec(),
1235        )
1236        .unwrap();
1237
1238        router
1239            .route_for_connection(&module_ctx, terminal)
1240            .await
1241            .unwrap();
1242        let reason = tokio::time::timeout(Duration::from_secs(1), &mut close_receiver)
1243            .await
1244            .expect("close request should be sent for the full client sink")
1245            .expect("close sender should include a reason");
1246        assert!(
1247            reason
1248                .to_string()
1249                .contains("module_to_client_delivery_failed"),
1250            "unexpected close reason: {reason}"
1251        );
1252        assert_eq!(client_rx.try_recv().unwrap().header.corr, 700);
1253        assert!(client_rx.try_recv().is_err());
1254        assert_eq!(
1255            router.counters.snapshot()["client_egress_close_delivery_failed"],
1256            1
1257        );
1258    }
1259
1260    /// A terminal frame ends the request at the module even when the client
1261    /// cannot take it, so the drain must stop counting it at once rather than
1262    /// when the client connection's cleanup later removes the route.
1263    #[tokio::test]
1264    async fn terminal_frame_releases_its_credit_even_when_client_delivery_fails() {
1265        let forwarding = Arc::new(ForwardingTable::default());
1266        let control = Arc::new(ControlHandler::with_forwarding(
1267            Arc::new(crate::Registry::default()),
1268            Arc::clone(&forwarding),
1269        ));
1270        let router = Router::with_control_handler(control);
1271        let module_connection = ConnectionId::new(11);
1272        let client_connection = ConnectionId::new(21);
1273        let _close_receiver = forwarding.register_connection_close(client_connection);
1274        let (module_tx, _module_rx) = mpsc::channel(1);
1275        forwarding
1276            .register_module_connection(
1277                module_connection,
1278                "credit-provider".to_string(),
1279                1,
1280                Concurrency::ModuleManaged,
1281                FrameSink::new(module_tx),
1282            )
1283            .unwrap();
1284        // Capacity one, filled by the route.open response, so the terminal
1285        // frame below cannot be enqueued for the client.
1286        let (client_tx, _client_rx) = mpsc::channel(1);
1287        let pending = forwarding
1288            .begin_route_bind_relay_for_test(
1289                client_connection,
1290                FrameSink::new(client_tx),
1291                800,
1292                "credit-provider",
1293            )
1294            .unwrap();
1295        forwarding
1296            .complete_pending_relay(
1297                module_connection,
1298                pending.corr,
1299                RouteBindRelayOutcome::Accepted,
1300            )
1301            .unwrap();
1302        let DataRoute::Client(DataRouteState::Bound(route)) = forwarding
1303            .lookup_data_route(
1304                client_connection,
1305                pending.client_channel,
1306                pending.client_epoch,
1307            )
1308            .unwrap()
1309        else {
1310            panic!("expected a bound client route");
1311        };
1312        route.flow.acquire_tagged(801, false).await.unwrap();
1313        assert_eq!(route.flow.drain_in_flight(), 1);
1314
1315        let (module_egress_tx, _module_egress_rx) = mpsc::channel(1);
1316        let module_ctx = RouteCtx {
1317            connection_id: module_connection,
1318            egress: FrameSink::new(module_egress_tx),
1319        };
1320        let terminal = Frame::build(
1321            FrameType::Response,
1322            Flags::new(false, Priority::Interactive, true),
1323            pending.module_channel,
1324            pending.module_epoch,
1325            801,
1326            b"terminal".to_vec(),
1327        )
1328        .unwrap();
1329        router
1330            .route_for_connection(&module_ctx, terminal)
1331            .await
1332            .unwrap();
1333
1334        assert_eq!(
1335            router.counters.snapshot()["client_egress_close_delivery_failed"],
1336            1,
1337            "the client delivery must have failed for this test to mean anything"
1338        );
1339        assert_eq!(
1340            route.flow.drain_in_flight(),
1341            0,
1342            "the module's terminal frame must release its credit even though the client could not take it"
1343        );
1344    }
1345
1346    #[tokio::test]
1347    async fn full_route_goodbye_sink_requests_target_close_without_erroring_module() {
1348        let forwarding = Arc::new(ForwardingTable::default());
1349        let control = Arc::new(ControlHandler::with_forwarding(
1350            Arc::new(crate::Registry::default()),
1351            Arc::clone(&forwarding),
1352        ));
1353        let router = Router::with_control_handler(control);
1354        let module_connection = ConnectionId::new(30);
1355        let client_connection = ConnectionId::new(40);
1356        let mut close_receiver = forwarding.register_connection_close(client_connection);
1357        let (module_tx, _module_rx) = mpsc::channel(1);
1358        forwarding
1359            .register_module_connection(
1360                module_connection,
1361                "goodbye-full-provider".to_string(),
1362                1,
1363                Concurrency::ModuleManaged,
1364                FrameSink::new(module_tx),
1365            )
1366            .unwrap();
1367        let (client_tx, mut client_rx) = mpsc::channel(1);
1368        let pending = forwarding
1369            .begin_route_bind_relay_for_test(
1370                client_connection,
1371                FrameSink::new(client_tx),
1372                800,
1373                "goodbye-full-provider",
1374            )
1375            .unwrap();
1376        forwarding
1377            .complete_pending_relay(
1378                module_connection,
1379                pending.corr,
1380                RouteBindRelayOutcome::Accepted,
1381            )
1382            .unwrap();
1383
1384        let (module_egress_tx, _module_egress_rx) = mpsc::channel(1);
1385        let module_ctx = RouteCtx {
1386            connection_id: module_connection,
1387            egress: FrameSink::new(module_egress_tx),
1388        };
1389        let goodbye = Frame::build(
1390            FrameType::Goodbye,
1391            Flags::new(false, Priority::Passive, true),
1392            pending.module_channel,
1393            pending.module_epoch,
1394            801,
1395            Vec::new(),
1396        )
1397        .unwrap();
1398
1399        router
1400            .route_for_connection(&module_ctx, goodbye)
1401            .await
1402            .unwrap();
1403        let reason = tokio::time::timeout(Duration::from_secs(1), &mut close_receiver)
1404            .await
1405            .expect("close request should be sent for the full GOODBYE sink")
1406            .expect("close sender should include a reason");
1407        assert!(
1408            reason.to_string().contains("route_goodbye_delivery_failed"),
1409            "unexpected close reason: {reason}"
1410        );
1411        assert_eq!(client_rx.try_recv().unwrap().header.corr, 800);
1412        assert!(client_rx.try_recv().is_err());
1413        assert_eq!(router.counters.snapshot()["goodbye_relay_client_failed"], 1);
1414        assert_eq!(router.counters.snapshot()["route_released_epoch_fenced"], 1);
1415    }
1416
1417    fn route_frame(ty: FrameType, channel: u16, epoch: u32, corr: u64) -> Frame {
1418        Frame::build(
1419            ty,
1420            Flags::new(false, Priority::Interactive, false),
1421            channel,
1422            epoch,
1423            corr,
1424            if ty == FrameType::Request || ty == FrameType::Response {
1425                b"route-body".to_vec()
1426            } else {
1427                Vec::new()
1428            },
1429        )
1430        .unwrap()
1431    }
1432
1433    type DynamicRouteFixture = (
1434        Router,
1435        Arc<ForwardingTable>,
1436        RouteCtx,
1437        mpsc::Receiver<crate::router::OutboundFrame>,
1438        RouteCtx,
1439        mpsc::Receiver<crate::router::OutboundFrame>,
1440        mpsc::Receiver<crate::router::OutboundFrame>,
1441        crate::forwarding::PendingRouteBindRelay,
1442    );
1443
1444    fn dynamic_route_fixture(commit: bool) -> DynamicRouteFixture {
1445        let forwarding = Arc::new(ForwardingTable::default());
1446        let control = Arc::new(crate::ControlHandler::with_forwarding(
1447            Arc::new(crate::Registry::default()),
1448            Arc::clone(&forwarding),
1449        ));
1450        let router = Router::with_control_handler(control);
1451        let module_connection = ConnectionId::new(500);
1452        let client_connection = ConnectionId::new(501);
1453        let (module_tx, module_rx) = mpsc::channel(8);
1454        forwarding
1455            .register_module_connection(
1456                module_connection,
1457                "epoch-router".into(),
1458                2,
1459                Concurrency::ModuleManaged,
1460                FrameSink::new(module_tx),
1461            )
1462            .unwrap();
1463        let (client_tx, client_rx) = mpsc::channel(8);
1464        let client_sink = FrameSink::new(client_tx);
1465        let pending = forwarding
1466            .begin_route_bind_relay_for_test(
1467                client_connection,
1468                client_sink.clone(),
1469                700,
1470                "epoch-router",
1471            )
1472            .unwrap();
1473        if commit {
1474            forwarding
1475                .complete_pending_relay(
1476                    module_connection,
1477                    pending.corr,
1478                    RouteBindRelayOutcome::Accepted,
1479                )
1480                .unwrap();
1481        }
1482        let (module_egress_tx, module_egress_rx) = mpsc::channel(8);
1483        (
1484            router,
1485            forwarding,
1486            RouteCtx {
1487                connection_id: client_connection,
1488                egress: client_sink,
1489            },
1490            client_rx,
1491            RouteCtx {
1492                connection_id: module_connection,
1493                egress: FrameSink::new(module_egress_tx),
1494            },
1495            module_egress_rx,
1496            module_rx,
1497            pending,
1498        )
1499    }
1500
1501    #[tokio::test]
1502    async fn route_epochs_validate_both_directions_and_rewrite_to_peer_handle() {
1503        let (
1504            router,
1505            _forwarding,
1506            client_ctx,
1507            mut client_rx,
1508            module_ctx,
1509            _module_egress_rx,
1510            mut module_rx,
1511            pending,
1512        ) = dynamic_route_fixture(true);
1513        let route_open = client_rx.recv().await.unwrap();
1514        assert_eq!(route_open.header.corr, 700);
1515
1516        router
1517            .route_for_connection(
1518                &client_ctx,
1519                route_frame(
1520                    FrameType::Request,
1521                    pending.client_channel,
1522                    pending.client_epoch,
1523                    701,
1524                ),
1525            )
1526            .await
1527            .unwrap();
1528        let forwarded = module_rx.recv().await.unwrap();
1529        assert_eq!(forwarded.header.channel, pending.module_channel);
1530        assert_eq!(forwarded.header.epoch, pending.module_epoch);
1531
1532        router
1533            .route_for_connection(
1534                &module_ctx,
1535                route_frame(
1536                    FrameType::Response,
1537                    pending.module_channel,
1538                    pending.module_epoch,
1539                    701,
1540                ),
1541            )
1542            .await
1543            .unwrap();
1544        let delivered = client_rx.recv().await.unwrap();
1545        assert_eq!(delivered.header.channel, pending.client_channel);
1546        assert_eq!(delivered.header.epoch, pending.client_epoch);
1547
1548        router
1549            .route_for_connection(
1550                &client_ctx,
1551                route_frame(
1552                    FrameType::Request,
1553                    pending.client_channel,
1554                    pending.client_epoch + 1,
1555                    702,
1556                ),
1557            )
1558            .await
1559            .unwrap();
1560        router
1561            .route_for_connection(
1562                &module_ctx,
1563                route_frame(
1564                    FrameType::Response,
1565                    pending.module_channel,
1566                    pending.module_epoch + 1,
1567                    703,
1568                ),
1569            )
1570            .await
1571            .unwrap();
1572        let stale_error = client_rx.recv().await.unwrap();
1573        assert_eq!(stale_error.header.ty, FrameType::Error);
1574        assert_eq!(stale_error.header.channel, pending.client_channel);
1575        assert_eq!(stale_error.header.epoch, pending.client_epoch + 1);
1576        assert_eq!(stale_error.header.corr, 702);
1577        let body: ErrorBody = serde_json::from_slice(&stale_error.body).unwrap();
1578        assert_eq!(body.code, "stale_route_epoch");
1579        assert!(module_rx.try_recv().is_err());
1580        assert!(client_rx.try_recv().is_err());
1581        let counters = router.counters.snapshot();
1582        assert_eq!(counters["client_frames_dropped_stale_route"], 1);
1583        assert_eq!(counters["module_frames_dropped_no_route"], 1);
1584    }
1585
1586    #[tokio::test]
1587    async fn accepted_route_publishes_route_open_before_immediate_reverse_request() {
1588        let (
1589            router,
1590            _,
1591            _client_ctx,
1592            mut client_rx,
1593            module_ctx,
1594            _module_egress_rx,
1595            _module_rx,
1596            pending,
1597        ) = dynamic_route_fixture(true);
1598        router
1599            .route_for_connection(
1600                &module_ctx,
1601                route_frame(
1602                    FrameType::Request,
1603                    pending.module_channel,
1604                    pending.module_epoch,
1605                    800,
1606                ),
1607            )
1608            .await
1609            .unwrap();
1610
1611        let first = client_rx.recv().await.unwrap();
1612        let second = client_rx.recv().await.unwrap();
1613        assert_eq!(first.header.channel, 0);
1614        assert_eq!(first.header.corr, 700);
1615        assert_eq!(second.header.channel, pending.client_channel);
1616        assert_eq!(second.header.epoch, pending.client_epoch);
1617        assert_eq!(second.header.corr, 800);
1618    }
1619
1620    #[tokio::test]
1621    async fn reserved_slot_ingress_errors_only_matching_client_requests() {
1622        let (
1623            router,
1624            _forwarding,
1625            client_ctx,
1626            mut client_rx,
1627            _module_ctx,
1628            _module_egress_rx,
1629            mut module_rx,
1630            pending,
1631        ) = dynamic_route_fixture(false);
1632        router
1633            .route_for_connection(
1634                &client_ctx,
1635                route_frame(
1636                    FrameType::Request,
1637                    pending.client_channel,
1638                    pending.client_epoch,
1639                    900,
1640                ),
1641            )
1642            .await
1643            .unwrap();
1644        let error = client_rx.recv().await.unwrap();
1645        assert_eq!(error.header.ty, FrameType::Error);
1646        assert_eq!(error.header.channel, pending.client_channel);
1647        assert_eq!(error.header.epoch, pending.client_epoch);
1648        assert_eq!(error.header.corr, 900);
1649
1650        router
1651            .route_for_connection(
1652                &client_ctx,
1653                route_frame(
1654                    FrameType::Response,
1655                    pending.client_channel,
1656                    pending.client_epoch,
1657                    901,
1658                ),
1659            )
1660            .await
1661            .unwrap();
1662        router
1663            .route_for_connection(
1664                &client_ctx,
1665                route_frame(
1666                    FrameType::Request,
1667                    pending.client_channel,
1668                    pending.client_epoch + 1,
1669                    902,
1670                ),
1671            )
1672            .await
1673            .unwrap();
1674        let stale_error = client_rx.recv().await.unwrap();
1675        assert_eq!(stale_error.header.ty, FrameType::Error);
1676        assert_eq!(stale_error.header.channel, pending.client_channel);
1677        assert_eq!(stale_error.header.epoch, pending.client_epoch + 1);
1678        assert_eq!(stale_error.header.corr, 902);
1679        let body: ErrorBody = serde_json::from_slice(&stale_error.body).unwrap();
1680        assert_eq!(body.code, "stale_route_epoch");
1681        assert!(module_rx.try_recv().is_err());
1682        let counters = router.counters.snapshot();
1683        assert_eq!(counters["client_frames_dropped_stale_route"], 1);
1684        assert_eq!(counters["module_frames_dropped_no_route"], 0);
1685    }
1686
1687    #[tokio::test]
1688    async fn dropped_module_route_goodbye_increments_counter() {
1689        let (
1690            router,
1691            _forwarding,
1692            client_ctx,
1693            mut client_rx,
1694            _module_ctx,
1695            _module_egress_rx,
1696            mut module_rx,
1697            pending,
1698        ) = dynamic_route_fixture(true);
1699        let _ = client_rx.recv().await;
1700        module_rx.close();
1701
1702        router
1703            .route_for_connection(
1704                &client_ctx,
1705                route_frame(
1706                    FrameType::Goodbye,
1707                    pending.client_channel,
1708                    pending.client_epoch,
1709                    999,
1710                ),
1711            )
1712            .await
1713            .unwrap();
1714
1715        let counters = router.counters.snapshot();
1716        assert_eq!(counters["goodbye_relay_module_dropped"], 1);
1717        assert_eq!(
1718            counters["goodbye_relay_module_dropped_by_module"],
1719            serde_json::json!({ "epoch-router": 1 })
1720        );
1721        assert_eq!(counters["route_released_epoch_fenced"], 1);
1722    }
1723
1724    #[tokio::test]
1725    async fn module_request_on_stale_epoch_receives_stale_route_epoch() {
1726        let (
1727            router,
1728            _forwarding,
1729            _client_ctx,
1730            _client_rx,
1731            module_ctx,
1732            mut module_egress_rx,
1733            mut module_rx,
1734            pending,
1735        ) = dynamic_route_fixture(true);
1736
1737        router
1738            .route_for_connection(
1739                &module_ctx,
1740                route_frame(
1741                    FrameType::Request,
1742                    pending.module_channel,
1743                    pending.module_epoch + 1,
1744                    1_000,
1745                ),
1746            )
1747            .await
1748            .unwrap();
1749
1750        let error = module_egress_rx.try_recv().unwrap();
1751        assert_eq!(error.header.ty, FrameType::Error);
1752        assert_eq!(error.header.channel, pending.module_channel);
1753        assert_eq!(error.header.epoch, pending.module_epoch + 1);
1754        assert_eq!(error.header.corr, 1_000);
1755        let body: ErrorBody = serde_json::from_slice(&error.body).unwrap();
1756        assert_eq!(body.code, "stale_route_epoch");
1757        assert!(module_rx.try_recv().is_err());
1758        let counters = router.counters.snapshot();
1759        assert_eq!(counters["module_requests_dropped_stale_route"], 1);
1760        assert_eq!(counters["module_frames_dropped_no_route"], 0);
1761    }
1762
1763    #[tokio::test]
1764    async fn module_request_on_reserved_or_absent_route_receives_unknown_channel() {
1765        let (
1766            reserved_router,
1767            _forwarding,
1768            _client_ctx,
1769            _client_rx,
1770            reserved_module_ctx,
1771            mut reserved_module_egress_rx,
1772            _module_rx,
1773            reserved,
1774        ) = dynamic_route_fixture(false);
1775        reserved_router
1776            .route_for_connection(
1777                &reserved_module_ctx,
1778                route_frame(
1779                    FrameType::Request,
1780                    reserved.module_channel,
1781                    reserved.module_epoch,
1782                    1_001,
1783                ),
1784            )
1785            .await
1786            .unwrap();
1787        let reserved_error = reserved_module_egress_rx.try_recv().unwrap();
1788        let reserved_body: ErrorBody = serde_json::from_slice(&reserved_error.body).unwrap();
1789        assert_eq!(reserved_error.header.ty, FrameType::Error);
1790        assert_eq!(reserved_error.header.channel, reserved.module_channel);
1791        assert_eq!(reserved_error.header.epoch, reserved.module_epoch);
1792        assert_eq!(reserved_error.header.corr, 1_001);
1793        assert_eq!(reserved_body.code, "unknown_channel");
1794        assert_eq!(
1795            reserved_router.counters.snapshot()["module_requests_dropped_stale_route"],
1796            1
1797        );
1798
1799        let (
1800            absent_router,
1801            _forwarding,
1802            _client_ctx,
1803            _client_rx,
1804            absent_module_ctx,
1805            mut absent_module_egress_rx,
1806            _module_rx,
1807            absent,
1808        ) = dynamic_route_fixture(false);
1809        absent_router
1810            .route_for_connection(
1811                &absent_module_ctx,
1812                route_frame(
1813                    FrameType::Request,
1814                    absent.module_channel + 1,
1815                    absent.module_epoch,
1816                    1_002,
1817                ),
1818            )
1819            .await
1820            .unwrap();
1821        let absent_error = absent_module_egress_rx.try_recv().unwrap();
1822        let absent_body: ErrorBody = serde_json::from_slice(&absent_error.body).unwrap();
1823        assert_eq!(absent_error.header.ty, FrameType::Error);
1824        assert_eq!(absent_error.header.channel, absent.module_channel + 1);
1825        assert_eq!(absent_error.header.epoch, absent.module_epoch);
1826        assert_eq!(absent_error.header.corr, 1_002);
1827        assert_eq!(absent_body.code, "unknown_channel");
1828        assert_eq!(
1829            absent_router.counters.snapshot()["module_requests_dropped_stale_route"],
1830            1
1831        );
1832    }
1833
1834    #[tokio::test]
1835    async fn non_request_module_frame_on_dead_route_is_counted_without_error() {
1836        let (
1837            router,
1838            forwarding,
1839            client_ctx,
1840            mut client_rx,
1841            module_ctx,
1842            mut module_egress_rx,
1843            mut module_rx,
1844            pending,
1845        ) = dynamic_route_fixture(true);
1846        let (other_module_tx, _other_module_rx) = mpsc::channel(8);
1847        forwarding
1848            .register_module_connection(
1849                ConnectionId::new(502),
1850                "other-module".into(),
1851                2,
1852                Concurrency::ModuleManaged,
1853                FrameSink::new(other_module_tx),
1854            )
1855            .unwrap();
1856        let _ = client_rx.recv().await.unwrap();
1857
1858        router
1859            .route_for_connection(
1860                &client_ctx,
1861                route_frame(
1862                    FrameType::Goodbye,
1863                    pending.client_channel,
1864                    pending.client_epoch,
1865                    1_003,
1866                ),
1867            )
1868            .await
1869            .unwrap();
1870        let _ = module_rx.recv().await.unwrap();
1871
1872        router
1873            .route_for_connection(
1874                &module_ctx,
1875                route_frame(
1876                    FrameType::StreamData,
1877                    pending.module_channel,
1878                    pending.module_epoch,
1879                    1_004,
1880                ),
1881            )
1882            .await
1883            .unwrap();
1884
1885        assert!(module_egress_rx.try_recv().is_err());
1886        let counters = router.counters.snapshot();
1887        assert_eq!(counters["module_frames_dropped_no_route"], 1);
1888        assert_eq!(
1889            counters["module_frames_dropped_no_route_by_module"],
1890            serde_json::json!({ "epoch-router": 1 })
1891        );
1892        assert_eq!(counters["module_requests_dropped_stale_route"], 0);
1893    }
1894
1895    #[test]
1896    fn channel_zero_cannot_be_registered_as_backend() {
1897        let mut router = Router::with_default_self_handler();
1898
1899        let err = router.register_backend(0, EchoBackend).unwrap_err();
1900
1901        assert_eq!(err, RouterError::ReservedChannelZero);
1902    }
1903}
subc_daemon/router.rs

subc_daemon/
router.rs
