dynomite/net/
dispatcher.rs

1//! Cluster-side dispatch hook.
2//!
3//! Routing decisions (whether to send a request to the local
4//! datastore, fan it out across racks, or relay it to a remote DC)
5//! live in the [`crate::cluster`] module. The per-connection FSMs
6//! own the wire path and expose a seam, [`Dispatcher`], that the
7//! cluster layer plugs into.
8//!
9//! [`Dispatcher`] is the seam between the two layers. The
10//! client / dnode-client FSMs hand each fully parsed [`Msg`] to a
11//! `Dispatcher` and inspect [`DispatchOutcome`] to decide whether
12//! the response can be returned synchronously or whether they
13//! should wait for a downstream response.
14//! [`crate::cluster::dispatch::ClusterDispatcher`] is the
15//! cluster-aware implementation; [`NoopDispatcher`] is the test
16//! double.
17//!
18//! [`Msg`]: crate::msg::Msg
19
20use std::sync::Arc;
21
22use tokio::sync::mpsc;
23
24use crate::msg::Msg;
25
26/// Outcome of dispatching a parsed message.
27#[derive(Debug)]
28pub enum DispatchOutcome {
29    /// The dispatcher took ownership of the request and will deliver
30    /// the response asynchronously (over the connection's response
31    /// channel installed by the FSM).
32    Pending,
33    /// The dispatcher wants the FSM to reply with the supplied
34    /// message immediately. Used for control plane / synthetic
35    /// responses (e.g. swallowed `QUIT` commands).
36    Inline(Msg),
37    /// The dispatcher rejected the request with an error response
38    /// the FSM should return to the client immediately.
39    Error(Msg),
40    /// The request must be dropped; no response will be sent. Used
41    /// for swallowed / quit messages.
42    Drop,
43}
44
45/// Cluster-side dispatch hook implemented by the cluster layer and
46/// by tests.
47///
48/// The dispatcher is invoked from a tokio task; implementations may
49/// do async work but should avoid blocking. The trait uses
50/// `&self` so the dispatcher can be shared across many connections.
51pub trait Dispatcher: Send + Sync {
52    /// Hand a parsed request to the dispatcher.
53    ///
54    /// `responder` is a per-connection channel the dispatcher uses
55    /// to deliver responses (or errors) back to the FSM that owns
56    /// the originating client connection.
57    fn dispatch(&self, req: Msg, responder: ServerSink) -> DispatchOutcome;
58}
59
60/// Channel the dispatcher uses to send responses back to a client
61/// FSM. The FSM owns the receiving half.
62pub type ServerSink = mpsc::Sender<OutboundEnvelope>;
63
64/// Envelope wrapping a dispatcher response and the request id it
65/// corresponds to.
66///
67/// `span` carries the originating request span back to the
68/// client-side FSM so the response writeback nests under the
69/// originating client span. The default is
70/// [`tracing::Span::none`].
71///
72/// `source_peer_idx` identifies the peer this response came from
73/// when the dispatcher fanned the request to multiple replicas.
74/// `None` is used for synthetic / inline / single-target paths
75/// where the source is unambiguous.
76#[derive(Debug)]
77pub struct OutboundEnvelope {
78    /// Request id the response is for.
79    pub req_id: crate::core::types::MsgId,
80    /// The response message.
81    pub rsp: Msg,
82    /// Originating request span for cross-task propagation.
83    pub span: tracing::Span,
84    /// Index of the peer this reply was produced by, when known.
85    /// Set by the per-target server / dnode-server drivers and
86    /// consumed by the per-request reply coalescer.
87    pub source_peer_idx: Option<u32>,
88}
89
90/// Dispatcher that drops every request and emits no response.
91///
92/// Useful as a placeholder in tests that only exercise framing.
93#[derive(Debug, Default, Clone)]
94pub struct NoopDispatcher;
95
96impl Dispatcher for NoopDispatcher {
97    fn dispatch(&self, _req: Msg, _responder: ServerSink) -> DispatchOutcome {
98        DispatchOutcome::Drop
99    }
100}
101
102impl<T: Dispatcher + ?Sized> Dispatcher for Arc<T> {
103    fn dispatch(&self, req: Msg, responder: ServerSink) -> DispatchOutcome {
104        (**self).dispatch(req, responder)
105    }
106}
107
108#[cfg(test)]
109mod tests {
110    use super::*;
111    use crate::msg::MsgType;
112
113    #[test]
114    fn noop_returns_drop() {
115        let (tx, _rx) = mpsc::channel(1);
116        let outcome = NoopDispatcher.dispatch(Msg::new(1, MsgType::ReqRedisGet, true), tx);
117        matches!(outcome, DispatchOutcome::Drop);
118    }
119}
dynomite/net/dispatcher.rs

dynomite/net/
dispatcher.rs
