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

kevy_rt/
client_ops.rs
