iggy_binary_protocol/
dispatch.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Command dispatch table mapping codes and operations to metadata.
19//!
20//! This is the protocol's identity registry. Every command has an entry
21//! with its numeric code, human-readable name, and optional VSR operation.
22//!
23//! Two lookup paths:
24//! - `lookup_command(code)`: current framing reads `[length][code][payload]`, looks up by code
25//! - `lookup_by_operation(op)`: future VSR framing reads 256-byte header, looks up by operation
26
27#[allow(clippy::wildcard_imports)]
28use crate::codes::*;
29use crate::consensus::Operation;
30
31/// Metadata for a single protocol command.
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub struct CommandMeta {
34    pub code: u32,
35    pub name: &'static str,
36    /// VSR operation for replicated commands. `None` for non-replicated
37    /// commands that bypass consensus.
38    pub operation: Option<Operation>,
39}
40
41impl CommandMeta {
42    /// Returns `true` if this command is replicated through VSR consensus.
43    #[must_use]
44    pub const fn is_replicated(&self) -> bool {
45        self.operation.is_some()
46    }
47
48    const fn new(code: u32, name: &'static str, operation: Option<Operation>) -> Self {
49        Self {
50            code,
51            name,
52            operation,
53        }
54    }
55
56    const fn non_replicated(code: u32, name: &'static str) -> Self {
57        Self::new(code, name, None)
58    }
59
60    const fn replicated(code: u32, name: &'static str, op: Operation) -> Self {
61        Self::new(code, name, Some(op))
62    }
63}
64
65/// All known command metadata entries.
66pub const COMMAND_TABLE: &[CommandMeta] = &[
67    // System
68    CommandMeta::non_replicated(PING_CODE, "ping"),
69    CommandMeta::non_replicated(GET_STATS_CODE, "stats"),
70    CommandMeta::non_replicated(GET_SNAPSHOT_FILE_CODE, "snapshot"),
71    CommandMeta::non_replicated(GET_CLUSTER_METADATA_CODE, "cluster.metadata"),
72    CommandMeta::non_replicated(GET_ME_CODE, "me"),
73    CommandMeta::non_replicated(GET_CLIENT_CODE, "client.get"),
74    CommandMeta::non_replicated(GET_CLIENTS_CODE, "client.list"),
75    // Users
76    CommandMeta::non_replicated(GET_USER_CODE, "user.get"),
77    CommandMeta::non_replicated(GET_USERS_CODE, "user.list"),
78    CommandMeta::replicated(CREATE_USER_CODE, "user.create", Operation::CreateUser),
79    CommandMeta::replicated(DELETE_USER_CODE, "user.delete", Operation::DeleteUser),
80    CommandMeta::replicated(UPDATE_USER_CODE, "user.update", Operation::UpdateUser),
81    CommandMeta::replicated(
82        UPDATE_PERMISSIONS_CODE,
83        "user.permissions",
84        Operation::UpdatePermissions,
85    ),
86    CommandMeta::replicated(
87        CHANGE_PASSWORD_CODE,
88        "user.password",
89        Operation::ChangePassword,
90    ),
91    CommandMeta::non_replicated(LOGIN_USER_CODE, "user.login"),
92    CommandMeta::non_replicated(LOGOUT_USER_CODE, "user.logout"),
93    CommandMeta::non_replicated(LOGIN_REGISTER_CODE, "user.login_register"),
94    // Personal Access Tokens
95    CommandMeta::non_replicated(
96        GET_PERSONAL_ACCESS_TOKENS_CODE,
97        "personal_access_token.list",
98    ),
99    CommandMeta::replicated(
100        CREATE_PERSONAL_ACCESS_TOKEN_CODE,
101        "personal_access_token.create",
102        Operation::CreatePersonalAccessToken,
103    ),
104    CommandMeta::replicated(
105        DELETE_PERSONAL_ACCESS_TOKEN_CODE,
106        "personal_access_token.delete",
107        Operation::DeletePersonalAccessToken,
108    ),
109    CommandMeta::non_replicated(
110        LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE,
111        "personal_access_token.login",
112    ),
113    // Messages
114    CommandMeta::non_replicated(POLL_MESSAGES_CODE, "message.poll"),
115    CommandMeta::replicated(SEND_MESSAGES_CODE, "message.send", Operation::SendMessages),
116    CommandMeta::non_replicated(FLUSH_UNSAVED_BUFFER_CODE, "message.flush_unsaved_buffer"),
117    // Consumer Offsets
118    CommandMeta::non_replicated(GET_CONSUMER_OFFSET_CODE, "consumer_offset.get"),
119    CommandMeta::replicated(
120        STORE_CONSUMER_OFFSET_CODE,
121        "consumer_offset.store",
122        Operation::StoreConsumerOffset,
123    ),
124    CommandMeta::replicated(
125        DELETE_CONSUMER_OFFSET_CODE,
126        "consumer_offset.delete",
127        Operation::DeleteConsumerOffset,
128    ),
129    // Streams
130    CommandMeta::non_replicated(GET_STREAM_CODE, "stream.get"),
131    CommandMeta::non_replicated(GET_STREAMS_CODE, "stream.list"),
132    CommandMeta::replicated(CREATE_STREAM_CODE, "stream.create", Operation::CreateStream),
133    CommandMeta::replicated(DELETE_STREAM_CODE, "stream.delete", Operation::DeleteStream),
134    CommandMeta::replicated(UPDATE_STREAM_CODE, "stream.update", Operation::UpdateStream),
135    CommandMeta::replicated(PURGE_STREAM_CODE, "stream.purge", Operation::PurgeStream),
136    // Topics
137    CommandMeta::non_replicated(GET_TOPIC_CODE, "topic.get"),
138    CommandMeta::non_replicated(GET_TOPICS_CODE, "topic.list"),
139    CommandMeta::replicated(CREATE_TOPIC_CODE, "topic.create", Operation::CreateTopic),
140    CommandMeta::replicated(DELETE_TOPIC_CODE, "topic.delete", Operation::DeleteTopic),
141    CommandMeta::replicated(UPDATE_TOPIC_CODE, "topic.update", Operation::UpdateTopic),
142    CommandMeta::replicated(PURGE_TOPIC_CODE, "topic.purge", Operation::PurgeTopic),
143    // Partitions
144    CommandMeta::replicated(
145        CREATE_PARTITIONS_CODE,
146        "partition.create",
147        Operation::CreatePartitions,
148    ),
149    CommandMeta::replicated(
150        DELETE_PARTITIONS_CODE,
151        "partition.delete",
152        Operation::DeletePartitions,
153    ),
154    // Segments
155    CommandMeta::replicated(
156        DELETE_SEGMENTS_CODE,
157        "segment.delete",
158        Operation::DeleteSegments,
159    ),
160    // Consumer Groups
161    CommandMeta::non_replicated(GET_CONSUMER_GROUP_CODE, "consumer_group.get"),
162    CommandMeta::non_replicated(GET_CONSUMER_GROUPS_CODE, "consumer_group.list"),
163    CommandMeta::replicated(
164        CREATE_CONSUMER_GROUP_CODE,
165        "consumer_group.create",
166        Operation::CreateConsumerGroup,
167    ),
168    CommandMeta::replicated(
169        DELETE_CONSUMER_GROUP_CODE,
170        "consumer_group.delete",
171        Operation::DeleteConsumerGroup,
172    ),
173    CommandMeta::replicated(
174        JOIN_CONSUMER_GROUP_CODE,
175        "consumer_group.join",
176        Operation::JoinConsumerGroup,
177    ),
178    CommandMeta::replicated(
179        LEAVE_CONSUMER_GROUP_CODE,
180        "consumer_group.leave",
181        Operation::LeaveConsumerGroup,
182    ),
183    CommandMeta::non_replicated(SYNC_CONSUMER_GROUP_CODE, "consumer_group.sync"),
184    // Login + Register (PAT - Personal Access Token variant)
185    CommandMeta::non_replicated(LOGIN_REGISTER_WITH_PAT_CODE, "user.login_register_with_pat"),
186    // Options catalog discovery
187    CommandMeta::non_replicated(DESCRIBE_OPTIONS_CODE, "options.describe"),
188    CommandMeta::non_replicated(ATTACH_CONSUMER_SESSION_CODE, "consumer_session.attach"),
189    CommandMeta::non_replicated(GET_POLL_ROUTING_CODE, "message.poll_routing"),
190    CommandMeta::non_replicated(POLL_MESSAGES_ON_PRIMARY_CODE, "message.poll_primary"),
191    CommandMeta::non_replicated(GET_CONSUMER_OFFSET_ROUTING_CODE, "consumer_offset.routing"),
192];
193
194/// Lookup command metadata by command code.
195///
196/// Uses a `match` (compiled to a jump table / binary search) for O(1) lookup
197/// instead of linear scan. The match maps code -> table index, keeping
198/// `COMMAND_TABLE` as the single source of truth for all metadata.
199#[must_use]
200pub const fn lookup_command(code: u32) -> Option<&'static CommandMeta> {
201    // Indices must match the order of entries in COMMAND_TABLE above.
202    let idx = match code {
203        PING_CODE => 0,
204        GET_STATS_CODE => 1,
205        GET_SNAPSHOT_FILE_CODE => 2,
206        GET_CLUSTER_METADATA_CODE => 3,
207        GET_ME_CODE => 4,
208        GET_CLIENT_CODE => 5,
209        GET_CLIENTS_CODE => 6,
210        GET_USER_CODE => 7,
211        GET_USERS_CODE => 8,
212        CREATE_USER_CODE => 9,
213        DELETE_USER_CODE => 10,
214        UPDATE_USER_CODE => 11,
215        UPDATE_PERMISSIONS_CODE => 12,
216        CHANGE_PASSWORD_CODE => 13,
217        LOGIN_USER_CODE => 14,
218        LOGOUT_USER_CODE => 15,
219        LOGIN_REGISTER_CODE => 16,
220        GET_PERSONAL_ACCESS_TOKENS_CODE => 17,
221        CREATE_PERSONAL_ACCESS_TOKEN_CODE => 18,
222        DELETE_PERSONAL_ACCESS_TOKEN_CODE => 19,
223        LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE => 20,
224        POLL_MESSAGES_CODE => 21,
225        SEND_MESSAGES_CODE => 22,
226        FLUSH_UNSAVED_BUFFER_CODE => 23,
227        GET_CONSUMER_OFFSET_CODE => 24,
228        STORE_CONSUMER_OFFSET_CODE => 25,
229        DELETE_CONSUMER_OFFSET_CODE => 26,
230        GET_STREAM_CODE => 27,
231        GET_STREAMS_CODE => 28,
232        CREATE_STREAM_CODE => 29,
233        DELETE_STREAM_CODE => 30,
234        UPDATE_STREAM_CODE => 31,
235        PURGE_STREAM_CODE => 32,
236        GET_TOPIC_CODE => 33,
237        GET_TOPICS_CODE => 34,
238        CREATE_TOPIC_CODE => 35,
239        DELETE_TOPIC_CODE => 36,
240        UPDATE_TOPIC_CODE => 37,
241        PURGE_TOPIC_CODE => 38,
242        CREATE_PARTITIONS_CODE => 39,
243        DELETE_PARTITIONS_CODE => 40,
244        DELETE_SEGMENTS_CODE => 41,
245        GET_CONSUMER_GROUP_CODE => 42,
246        GET_CONSUMER_GROUPS_CODE => 43,
247        CREATE_CONSUMER_GROUP_CODE => 44,
248        DELETE_CONSUMER_GROUP_CODE => 45,
249        JOIN_CONSUMER_GROUP_CODE => 46,
250        LEAVE_CONSUMER_GROUP_CODE => 47,
251        SYNC_CONSUMER_GROUP_CODE => 48,
252        LOGIN_REGISTER_WITH_PAT_CODE => 49,
253        DESCRIBE_OPTIONS_CODE => 50,
254        ATTACH_CONSUMER_SESSION_CODE => 51,
255        GET_POLL_ROUTING_CODE => 52,
256        POLL_MESSAGES_ON_PRIMARY_CODE => 53,
257        GET_CONSUMER_OFFSET_ROUTING_CODE => 54,
258        _ => return None,
259    };
260    Some(&COMMAND_TABLE[idx])
261}
262
263/// Lookup command metadata by VSR operation.
264///
265/// Returns `None` for `Operation::Reserved` and for non-replicated commands
266/// that have no operation mapping.
267#[must_use]
268pub const fn lookup_by_operation(op: Operation) -> Option<&'static CommandMeta> {
269    // Indices must match the order of entries in COMMAND_TABLE above.
270    let idx = match op {
271        Operation::CreateStream => 29,
272        Operation::UpdateStream => 31,
273        Operation::DeleteStream => 30,
274        Operation::PurgeStream => 32,
275        Operation::CreateTopic => 35,
276        Operation::UpdateTopic => 37,
277        Operation::DeleteTopic => 36,
278        Operation::PurgeTopic => 38,
279        Operation::CreatePartitions => 39,
280        Operation::DeletePartitions => 40,
281        Operation::DeleteSegments => 41,
282        Operation::CreateConsumerGroup => 44,
283        Operation::DeleteConsumerGroup => 45,
284        Operation::JoinConsumerGroup => 46,
285        Operation::LeaveConsumerGroup => 47,
286        Operation::CreateUser => 9,
287        Operation::UpdateUser => 11,
288        Operation::DeleteUser => 10,
289        Operation::ChangePassword => 13,
290        Operation::UpdatePermissions => 12,
291        Operation::CreatePersonalAccessToken => 18,
292        Operation::DeletePersonalAccessToken => 19,
293        Operation::SendMessages => 22,
294        Operation::StoreConsumerOffset => 25,
295        Operation::DeleteConsumerOffset => 26,
296        Operation::CreateTopicWithAssignments
297        | Operation::CreatePartitionsWithAssignments
298        | Operation::RemoveConsumerGroupMember
299        | Operation::CompleteConsumerGroupRevocation
300        | Operation::TruncatePartition
301        | Operation::Reserved
302        | Operation::Register
303        | Operation::Logout
304        | Operation::NonReplicated => return None,
305    };
306    Some(&COMMAND_TABLE[idx])
307}
308
309#[cfg(test)]
310mod tests {
311    use super::*;
312
313    #[test]
314    fn every_code_has_entry() {
315        let all_codes = [
316            PING_CODE,
317            GET_STATS_CODE,
318            GET_SNAPSHOT_FILE_CODE,
319            GET_CLUSTER_METADATA_CODE,
320            GET_ME_CODE,
321            GET_CLIENT_CODE,
322            GET_CLIENTS_CODE,
323            GET_USER_CODE,
324            GET_USERS_CODE,
325            CREATE_USER_CODE,
326            DELETE_USER_CODE,
327            UPDATE_USER_CODE,
328            UPDATE_PERMISSIONS_CODE,
329            CHANGE_PASSWORD_CODE,
330            LOGIN_USER_CODE,
331            LOGOUT_USER_CODE,
332            LOGIN_REGISTER_CODE,
333            LOGIN_REGISTER_WITH_PAT_CODE,
334            GET_PERSONAL_ACCESS_TOKENS_CODE,
335            CREATE_PERSONAL_ACCESS_TOKEN_CODE,
336            DELETE_PERSONAL_ACCESS_TOKEN_CODE,
337            LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE,
338            POLL_MESSAGES_CODE,
339            SEND_MESSAGES_CODE,
340            FLUSH_UNSAVED_BUFFER_CODE,
341            GET_CONSUMER_OFFSET_CODE,
342            STORE_CONSUMER_OFFSET_CODE,
343            DELETE_CONSUMER_OFFSET_CODE,
344            GET_STREAM_CODE,
345            GET_STREAMS_CODE,
346            CREATE_STREAM_CODE,
347            DELETE_STREAM_CODE,
348            UPDATE_STREAM_CODE,
349            PURGE_STREAM_CODE,
350            GET_TOPIC_CODE,
351            GET_TOPICS_CODE,
352            CREATE_TOPIC_CODE,
353            DELETE_TOPIC_CODE,
354            UPDATE_TOPIC_CODE,
355            PURGE_TOPIC_CODE,
356            CREATE_PARTITIONS_CODE,
357            DELETE_PARTITIONS_CODE,
358            DELETE_SEGMENTS_CODE,
359            GET_CONSUMER_GROUP_CODE,
360            GET_CONSUMER_GROUPS_CODE,
361            CREATE_CONSUMER_GROUP_CODE,
362            DELETE_CONSUMER_GROUP_CODE,
363            JOIN_CONSUMER_GROUP_CODE,
364            LEAVE_CONSUMER_GROUP_CODE,
365            DESCRIBE_OPTIONS_CODE,
366        ];
367        for code in all_codes {
368            assert!(
369                lookup_command(code).is_some(),
370                "missing dispatch entry for code {code}"
371            );
372        }
373    }
374
375    #[test]
376    fn no_duplicate_codes_in_table() {
377        let mut seen = std::collections::HashSet::new();
378        for entry in COMMAND_TABLE {
379            assert!(
380                seen.insert(entry.code),
381                "duplicate code {} ({}) in COMMAND_TABLE",
382                entry.code,
383                entry.name
384            );
385        }
386    }
387
388    #[test]
389    fn unknown_code_returns_none() {
390        assert!(lookup_command(9999).is_none());
391    }
392
393    #[test]
394    fn names_are_non_empty() {
395        for entry in COMMAND_TABLE {
396            assert!(!entry.name.is_empty(), "empty name for code {}", entry.code);
397        }
398    }
399
400    #[test]
401    fn lookup_by_operation_roundtrips_with_lookup_command() {
402        let replicated_ops = [
403            Operation::CreateStream,
404            Operation::UpdateStream,
405            Operation::DeleteStream,
406            Operation::PurgeStream,
407            Operation::CreateTopic,
408            Operation::UpdateTopic,
409            Operation::DeleteTopic,
410            Operation::PurgeTopic,
411            Operation::CreatePartitions,
412            Operation::DeletePartitions,
413            Operation::DeleteSegments,
414            Operation::CreateConsumerGroup,
415            Operation::DeleteConsumerGroup,
416            Operation::CreateUser,
417            Operation::UpdateUser,
418            Operation::DeleteUser,
419            Operation::ChangePassword,
420            Operation::UpdatePermissions,
421            Operation::CreatePersonalAccessToken,
422            Operation::DeletePersonalAccessToken,
423            Operation::JoinConsumerGroup,
424            Operation::LeaveConsumerGroup,
425            Operation::SendMessages,
426            Operation::StoreConsumerOffset,
427            Operation::DeleteConsumerOffset,
428        ];
429        for op in replicated_ops {
430            let meta = lookup_by_operation(op)
431                .unwrap_or_else(|| panic!("no dispatch entry for operation {op:?}"));
432
433            let by_code = lookup_command(meta.code)
434                .unwrap_or_else(|| panic!("no dispatch entry for code {}", meta.code));
435
436            assert_eq!(
437                meta.code, by_code.code,
438                "lookup_by_operation and lookup disagree for {op:?}"
439            );
440        }
441    }
442
443    #[test]
444    fn reserved_operation_returns_none() {
445        assert!(lookup_by_operation(Operation::Reserved).is_none());
446    }
447
448    #[test]
449    fn no_duplicate_operations_in_table() {
450        let mut seen = std::collections::HashSet::new();
451        for entry in COMMAND_TABLE {
452            if let Some(op) = entry.operation {
453                assert!(
454                    seen.insert(op as u8),
455                    "duplicate operation {:?} ({}) in COMMAND_TABLE",
456                    op,
457                    entry.name
458                );
459            }
460        }
461    }
462
463    /// Verify that the match indices in `lookup_command` point to entries
464    /// whose `.code` field actually matches the looked-up code.
465    /// Catches table reordering that would silently break the match.
466    #[test]
467    fn lookup_command_indices_match_table_codes() {
468        for (i, entry) in COMMAND_TABLE.iter().enumerate() {
469            let looked_up = lookup_command(entry.code)
470                .unwrap_or_else(|| panic!("lookup_command({}) returned None", entry.code));
471            assert_eq!(
472                looked_up.code, entry.code,
473                "lookup_command({}) returned entry at wrong index \
474                 (expected table[{i}].code={}, got code={})",
475                entry.code, entry.code, looked_up.code
476            );
477        }
478    }
479
480    /// Verify that `lookup_by_operation` indices point to entries whose
481    /// `.operation` field matches the looked-up operation.
482    #[test]
483    fn lookup_by_operation_indices_match_table_ops() {
484        for entry in COMMAND_TABLE {
485            if let Some(op) = entry.operation {
486                let looked_up = lookup_by_operation(op)
487                    .unwrap_or_else(|| panic!("lookup_by_operation({op:?}) returned None"));
488                assert_eq!(
489                    looked_up.operation,
490                    Some(op),
491                    "lookup_by_operation({op:?}) returned entry with wrong operation: {:?}",
492                    looked_up.operation
493                );
494            }
495        }
496    }
497}
iggy_binary_protocol/dispatch.rs

iggy_binary_protocol/
dispatch.rs
