kevy_rt/
client_ops.rs

1//! CLIENT-face machinery: the per-conn row renderer shared by
2//! `CLIENT LIST` / `CLIENT INFO`, the `CLIENT KILL` selector, and the
3//! per-shard fan-out handlers. Everything here runs on the owning
4//! shard's reactor thread, where the conn table is plain data
5//! (thread-per-core, no locks).
6
7use crate::Commands;
8use crate::conn::Conn;
9use crate::message::Part;
10use crate::shard::Shard;
11use kevy_resp::ArgvView;
12
13/// Parsed `CLIENT KILL` selector. `Addr` matches the peer `ip:port`
14/// exactly; `Id` matches the instance-unique conn id.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub enum ClientKillFilter {
17    /// Peer address (`ip:port`) equality.
18    Addr(Vec<u8>),
19    /// Instance-unique conn id equality.
20    Id(u64),
21}
22
23impl ClientKillFilter {
24    /// Parse the argv of `CLIENT KILL …`. Returns the selector plus
25    /// `true` for the legacy positional form (`CLIENT KILL addr:port`),
26    /// whose reply is `+OK` / `-ERR` instead of the filtered form's
27    /// killed-count integer. `None` = a shape this server doesn't
28    /// support (the caller answers with a syntax error).
29    pub fn parse<A: ArgvView + ?Sized>(args: &A) -> Option<(Self, bool)> {
30        match args.len() {
31            3 => {
32                let a = args.get(2)?;
33                a.contains(&b':').then(|| (Self::Addr(a.to_vec()), true))
34            }
35            4 => {
36                let kind = args.get(2)?.to_ascii_uppercase();
37                let val = args.get(3)?;
38                match kind.as_slice() {
39                    b"ID" => std::str::from_utf8(val)
40                        .ok()?
41                        .parse()
42                        .ok()
43                        .map(|id| (Self::Id(id), false)),
44                    b"ADDR" => Some((Self::Addr(val.to_vec()), false)),
45                    _ => None,
46                }
47            }
48            _ => None,
49        }
50    }
51}
52
53/// Render one `CLIENT LIST` / `CLIENT INFO` row into `out`. The field
54/// set mirrors the Redis 7.x shape; fields kevy keeps no per-conn
55/// state for are reported at their idle defaults (`cmd=NULL` — the
56/// last-command name is not tracked).
57pub(crate) fn client_row(id: u64, conn: &Conn, out: &mut Vec<u8>) {
58    use std::fmt::Write as _;
59    let mut s = String::with_capacity(224);
60    let _ = writeln!(
61        s,
62        "id={id} addr={}:{} laddr=0.0.0.0:0 fd={} name={} age={} idle=0 \
63         flags=N db=0 sub={} psub={} ssub=0 multi={} watch={} qbuf={} \
64         qbuf-free=0 argv-mem=0 multi-mem=0 tot-mem=0 rbs=0 rbp=0 obl={} \
65         oll=0 omem=0 events=r cmd=NULL user=default redir=-1 resp={} \
66         lib-name= lib-ver=",
67        conn.peer.0,
68        conn.peer.1,
69        conn.sock.raw(),
70        String::from_utf8_lossy(&conn.client_name),
71        conn.created.elapsed().as_secs(),
72        conn.sub.len(),
73        conn.psub.len(),
74        conn.multi.as_ref().map_or(-1, |q| q.len() as i64),
75        conn.watched.len(),
76        conn.input.len(),
77        conn.output.len().saturating_sub(conn.write_pos),
78        match conn.proto {
79            kevy_resp::RespVersion::V2 => 2,
80            kevy_resp::RespVersion::V3 => 3,
81        },
82    );
83    out.extend_from_slice(s.as_bytes());
84}
85
86impl<C: Commands> Shard<C> {
87    /// `Op::ClientList` — render every real client conn on this shard
88    /// (cluster-bus links excluded: infra, not clients).
89    pub(crate) fn exec_client_list(&mut self) -> Part {
90        let mut text = Vec::with_capacity(self.conns.len() * 192);
91        for (id, conn) in &self.conns {
92            if conn.cluster {
93                continue;
94            }
95            client_row(*id, conn, &mut text);
96        }
97        Part::ExtensionChunk(text)
98    }
99
100    /// `Op::ClientKill` — mark every matching conn closing and hand it
101    /// to the reactor's sweep (epoll: the dirty-flush close path;
102    /// io_uring: the periodic closing-set reap). Teardown waits for
103    /// the conn's output to drain, so a self-kill still delivers its
104    /// own reply first. Returns the matched count.
105    pub(crate) fn exec_client_kill(&mut self, filter: &ClientKillFilter) -> Part {
106        let mut victims: Vec<u64> = Vec::new();
107        for (id, conn) in &self.conns {
108            if conn.cluster || conn.closing {
109                continue;
110            }
111            let hit = match filter {
112                ClientKillFilter::Id(want) => *id == *want,
113                ClientKillFilter::Addr(addr) => {
114                    format!("{}:{}", conn.peer.0, conn.peer.1).as_bytes() == addr.as_slice()
115                }
116            };
117            if hit {
118                victims.push(*id);
119            }
120        }
121        for id in &victims {
122            if let Some(conn) = self.conns.get_mut(id) {
123                conn.closing = true;
124            }
125            self.dirty.push(*id);
126            self.closing_uring_conns.push(*id);
127            // Eagerly cancel the victim's block waiters (parked
128            // BLPOP/XREAD + cross-shard arbiter registrations), same
129            // as the QUIT/EOF path. The io_uring reap runs on a 1/16
130            // iteration throttle — without this a killed-but-unreaped
131            // conn's waiter stayed live and could consume a push
132            // (e.g. an LPUSH element) meant for a live client.
133            self.blocked.drop_for_conn(*id);
134            self.cancel_xshard_on_close(*id);
135        }
136        Part::Int(victims.len() as i64)
137    }
138}
139
140#[cfg(test)]
141mod tests {
142    use super::ClientKillFilter;
143    use kevy_resp::Argv;
144
145    fn argv(parts: &[&[u8]]) -> Argv {
146        let mut a = Argv::default();
147        for p in parts {
148            a.push(p);
149        }
150        a
151    }
152
153    #[test]
154    fn parse_legacy_addr_form() {
155        let a = argv(&[b"CLIENT", b"KILL", b"127.0.0.1:50123"]);
156        assert_eq!(
157            ClientKillFilter::parse(&a),
158            Some((ClientKillFilter::Addr(b"127.0.0.1:50123".to_vec()), true))
159        );
160    }
161
162    #[test]
163    fn parse_id_and_addr_filters() {
164        let a = argv(&[b"CLIENT", b"KILL", b"ID", b"42"]);
165        assert_eq!(ClientKillFilter::parse(&a), Some((ClientKillFilter::Id(42), false)));
166        let a = argv(&[b"CLIENT", b"KILL", b"addr", b"10.0.0.1:1"]);
167        assert_eq!(
168            ClientKillFilter::parse(&a),
169            Some((ClientKillFilter::Addr(b"10.0.0.1:1".to_vec()), false))
170        );
171    }
172
173    #[test]
174    fn parse_rejects_unsupported_shapes() {
175        assert_eq!(ClientKillFilter::parse(&argv(&[b"CLIENT", b"KILL"])), None);
176        assert_eq!(ClientKillFilter::parse(&argv(&[b"CLIENT", b"KILL", b"noport"])), None);
177        assert_eq!(
178            ClientKillFilter::parse(&argv(&[b"CLIENT", b"KILL", b"LADDR", b"1.2.3.4:5"])),
179            None
180        );
181        assert_eq!(
182            ClientKillFilter::parse(&argv(&[b"CLIENT", b"KILL", b"ID", b"notanum"])),
183            None
184        );
185    }
186}
kevy_rt/client_ops.rs

kevy_rt/
client_ops.rs
