questdb/egress/
server_event.rs

1/*******************************************************************************
2 *     ___                  _   ____  ____
3 *    / _ \ _   _  ___  ___| |_|  _ \| __ )
4 *   | | | | | | |/ _ \/ __| __| | | |  _ \
5 *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
6 *    \__\_\\__,_|\___||___/\__|____/|____/
7 *
8 *  Copyright (c) 2014-2019 Appsicle
9 *  Copyright (c) 2019-2025 QuestDB
10 *
11 *  Licensed under the Apache License, Version 2.0 (the "License");
12 *  you may not use this file except in compliance with the License.
13 *  You may obtain a copy of the License at
14 *
15 *  http://www.apache.org/licenses/LICENSE-2.0
16 *
17 *  Unless required by applicable law or agreed to in writing, software
18 *  distributed under the License is distributed on an "AS IS" BASIS,
19 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 *  See the License for the specific language governing permissions and
21 *  limitations under the License.
22 *
23 ******************************************************************************/
24
25//! Server → client message decoders and the top-level [`decode_frame`]
26//! dispatcher. RESULT_BATCH (`0x11`) decoding lives in
27//! [`crate::egress::decoder`]; everything else is here.
28
29use crate::egress::decoder::{DecodedBatch, ZstdScratch, decode_result_batch};
30use crate::egress::schema::Schema;
31use crate::egress::symbol_dict::SymbolDict;
32use crate::egress::wire::ByteReader;
33use crate::egress::wire::cache_reset::resets_dict;
34use crate::egress::wire::capabilities::has_zone;
35use crate::egress::wire::header::FrameHeader;
36use crate::egress::wire::msg_kind::{MsgKind, StatusCode};
37use crate::egress::wire::roles;
38use crate::error::{Result, fmt};
39use bytes::Bytes;
40
41// ---------------------------------------------------------------------------
42// Public types
43// ---------------------------------------------------------------------------
44
45/// QuestDB cluster role advertised by `SERVER_INFO`.
46///
47/// `#[non_exhaustive]` because new role bytes may be added in future
48/// protocol revisions; a future revision might also promote a known
49/// `Other(_)` byte to a named variant. Both should be additive.
50#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
51#[non_exhaustive]
52pub enum ServerRole {
53    Standalone,
54    Primary,
55    Replica,
56    PrimaryCatchup,
57    /// Forward-compat: a future role byte we don't recognise.
58    Other(u8),
59}
60
61impl ServerRole {
62    pub fn from_u8(byte: u8) -> Self {
63        match byte {
64            roles::STANDALONE => ServerRole::Standalone,
65            roles::PRIMARY => ServerRole::Primary,
66            roles::REPLICA => ServerRole::Replica,
67            roles::PRIMARY_CATCHUP => ServerRole::PrimaryCatchup,
68            other => ServerRole::Other(other),
69        }
70    }
71
72    /// ASCII token used on the wire (matches `X-QuestDB-Role` header
73    /// values and the spec's role enum names). Forward-compat
74    /// `Other(_)` is rendered as `UNKNOWN(<byte>)` so the byte is still
75    /// recoverable from a log line.
76    pub fn as_str(self) -> String {
77        match self {
78            ServerRole::Standalone => roles::NAME_STANDALONE.to_string(),
79            ServerRole::Primary => roles::NAME_PRIMARY.to_string(),
80            ServerRole::Replica => roles::NAME_REPLICA.to_string(),
81            ServerRole::PrimaryCatchup => roles::NAME_PRIMARY_CATCHUP.to_string(),
82            ServerRole::Other(b) => format!("UNKNOWN({})", b),
83        }
84    }
85
86    /// Raw wire byte. `from_u8(self.as_u8()) == self` for every variant
87    /// (round-trip-safe, including `Other(_)`).
88    pub fn as_u8(self) -> u8 {
89        match self {
90            ServerRole::Standalone => roles::STANDALONE,
91            ServerRole::Primary => roles::PRIMARY,
92            ServerRole::Replica => roles::REPLICA,
93            ServerRole::PrimaryCatchup => roles::PRIMARY_CATCHUP,
94            ServerRole::Other(b) => b,
95        }
96    }
97}
98
99/// Body of a `SERVER_INFO` frame.
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct ServerInfo {
102    pub role: ServerRole,
103    pub epoch: u64,
104    pub capabilities: u32,
105    pub server_wall_ns: i64,
106    pub cluster_id: String,
107    pub node_id: String,
108    /// Optional zone identifier, present iff the server set `CAP_ZONE` in
109    /// `capabilities`. Free-form, opaque, case-insensitively compared to
110    /// the client's `zone=` connect-string knob (failover.md §2). `None`
111    /// on every server that does not advertise `CAP_ZONE` — including
112    /// servers whose `capabilities` is hard-zero.
113    pub zone_id: Option<String>,
114}
115
116/// Upgrade-time topology rejection carried alongside an [`crate::Error`].
117///
118/// Populated when the server rejects the WebSocket upgrade with HTTP `421`
119/// plus an `X-QuestDB-Role` header (per failover.md §5), or when a
120/// `SERVER_INFO` frame advertises a role that does not match the configured
121/// `target=` filter. The host-health tracker (when present) reads this to
122/// decide whether the host is in `TransientReject` (`PRIMARY_CATCHUP`) or
123/// `TopologyReject` (every other role byte) and to update zone tier from
124/// the optional `X-QuestDB-Zone` / `SERVER_INFO.zone_id`.
125///
126/// `role_byte` is the raw `SERVER_INFO.role` byte from wire-egress.md §11.8
127/// (`0x00`=STANDALONE, `0x01`=PRIMARY, `0x02`=REPLICA, `0x03`=PRIMARY_CATCHUP);
128/// unrecognised values are carried through verbatim so a future role
129/// addition is observable to operators even on an older client build.
130/// `role_name` is the ASCII token actually seen on the wire (uppercased for
131/// the four named roles; the literal header value when the byte is unknown);
132/// it is kept so diagnostics surface what the server *said*, not what the
133/// client decided to call it. `zone` is `Some` only when the server
134/// advertised one (via `SERVER_INFO.zone_id` gated on `CAP_ZONE`, or the
135/// `X-QuestDB-Zone` upgrade header).
136///
137/// `#[non_exhaustive]` so future fields (a structured replay-hint, a
138/// retry-after value, a cluster-ID tag — anything the failover.md spec
139/// might extend `421` reject headers with) can be added without
140/// breaking downstream struct-literal construction or exhaustive
141/// destructuring. Use [`UpgradeReject::new`] to construct from
142/// external code.
143#[derive(Debug, Clone, PartialEq, Eq)]
144#[non_exhaustive]
145pub struct UpgradeReject {
146    pub role_byte: u8,
147    pub role_name: String,
148    pub zone: Option<String>,
149}
150
151impl UpgradeReject {
152    pub fn new(role_byte: u8, role_name: impl Into<String>, zone: Option<String>) -> Self {
153        Self {
154            role_byte,
155            role_name: role_name.into(),
156            zone,
157        }
158    }
159
160    /// True when the server-advertised role is `PRIMARY_CATCHUP` —
161    /// a transient state (promotion in flight) that the tracker should
162    /// classify as recoverable. Every other role byte is topological
163    /// (won't recover without operator intervention or topology change).
164    /// Per failover.md §6: any non-empty `X-QuestDB-Role` value other
165    /// than `PRIMARY_CATCHUP` is conservatively treated as topological,
166    /// including unrecognised tokens.
167    pub fn is_transient(&self) -> bool {
168        self.role_byte == roles::PRIMARY_CATCHUP
169            || self
170                .role_name
171                .eq_ignore_ascii_case(roles::NAME_PRIMARY_CATCHUP)
172    }
173}
174
175/// Single decoded server message.
176///
177/// One frame in, one event out. The dispatcher applies state mutations
178/// (symbol dict deltas, batch-0 inline-schema capture, cache resets)
179/// before returning so callers can treat each event idempotently.
180#[derive(Debug, Clone)]
181pub enum ServerEvent {
182    /// `RESULT_BATCH` (`0x11`).
183    Batch(DecodedBatch),
184    /// `RESULT_END` (`0x12`).
185    End {
186        request_id: i64,
187        final_seq: u64,
188        total_rows: u64,
189    },
190    /// `QUERY_ERROR` (`0x13`).
191    Error {
192        request_id: i64,
193        status: StatusCode,
194        message: String,
195    },
196    /// `EXEC_DONE` (`0x16`).
197    ExecDone {
198        request_id: i64,
199        op_type: u8,
200        rows_affected: u64,
201    },
202    /// `CACHE_RESET` (`0x17`). Mask bits already applied to the symbol dict.
203    CacheReset {
204        // `mask` is matched literally by tests (pattern `mask: 0x01`)
205        // but never read by the consumers — `decode_frame` performs
206        // the resets in place before returning the event. Marked
207        // `allow(dead_code)` so the wire-level visibility stays
208        // honest without tripping `-D dead_code`.
209        #[allow(dead_code)]
210        mask: u8,
211    },
212    /// `SERVER_INFO` (`0x18`).
213    ServerInfo(ServerInfo),
214}
215
216// ---------------------------------------------------------------------------
217// Top-level dispatcher
218// ---------------------------------------------------------------------------
219
220/// Decode one full frame (already split into header + payload).
221///
222/// `dict` and `query_schema` are mutated in place where the message demands
223/// it (delta dict, batch-0 inline schema). The returned event is what the
224/// caller's cursor / state machine should react to.
225pub fn decode_frame(
226    header: FrameHeader,
227    payload: &Bytes,
228    dict: &mut SymbolDict,
229    query_schema: &mut Option<Schema>,
230    zstd_scratch: &mut ZstdScratch,
231) -> Result<ServerEvent> {
232    if payload.is_empty() {
233        return Err(fmt!(ProtocolError, "frame payload is empty"));
234    }
235    let kind_byte = payload[0];
236    let kind = MsgKind::from_u8(kind_byte)?;
237    // Per `wire/header.rs`, `table_count` is `1` for `RESULT_BATCH` (the
238    // only frame that carries an actual table block) and `0` everywhere
239    // else. Catch frame-vs-kind drift up front rather than letting it
240    // surface as a confusing per-message decode failure downstream.
241    let expected_tc = if matches!(kind, MsgKind::ResultBatch) {
242        1
243    } else {
244        0
245    };
246    if header.table_count != expected_tc {
247        return Err(fmt!(
248            ProtocolError,
249            "frame for msg_kind 0x{:02X} has table_count {} (expected {})",
250            kind_byte,
251            header.table_count,
252            expected_tc
253        ));
254    }
255    match kind {
256        MsgKind::ResultBatch => Ok(ServerEvent::Batch(decode_result_batch(
257            payload,
258            header.flags,
259            dict,
260            query_schema,
261            zstd_scratch,
262        )?)),
263        MsgKind::ResultEnd => decode_result_end(payload),
264        MsgKind::QueryError => decode_query_error(payload),
265        MsgKind::ExecDone => decode_exec_done(payload),
266        MsgKind::CacheReset => decode_cache_reset(payload, dict),
267        MsgKind::ServerInfo => decode_server_info(payload),
268        // Server should never send these to us.
269        MsgKind::QueryRequest | MsgKind::Cancel | MsgKind::Credit => Err(fmt!(
270            ProtocolError,
271            "server sent client-only message kind 0x{:02X}",
272            kind_byte
273        )),
274    }
275}
276
277// ---------------------------------------------------------------------------
278// Per-message decoders
279// ---------------------------------------------------------------------------
280
281fn decode_result_end(payload: &[u8]) -> Result<ServerEvent> {
282    let mut r = ByteReader::new(payload);
283    expect_kind(&mut r, MsgKind::ResultEnd)?;
284    let request_id = r.read_i64_le()?;
285    let final_seq = r.read_varint_u64()?;
286    let total_rows = r.read_varint_u64()?;
287    expect_eof(&r, "RESULT_END")?;
288    Ok(ServerEvent::End {
289        request_id,
290        final_seq,
291        total_rows,
292    })
293}
294
295fn decode_query_error(payload: &[u8]) -> Result<ServerEvent> {
296    let mut r = ByteReader::new(payload);
297    expect_kind(&mut r, MsgKind::QueryError)?;
298    let request_id = r.read_i64_le()?;
299    let status = StatusCode::from_u8(r.read_u8()?)?;
300    let msg_len = r.read_u16_le()? as usize;
301    let bytes = r.read_bytes(msg_len)?;
302    let message = std::str::from_utf8(bytes)
303        .map_err(|e| fmt!(InvalidUtf8, "QUERY_ERROR message not valid UTF-8: {}", e))?
304        .to_string();
305    expect_eof(&r, "QUERY_ERROR")?;
306    Ok(ServerEvent::Error {
307        request_id,
308        status,
309        message,
310    })
311}
312
313fn decode_exec_done(payload: &[u8]) -> Result<ServerEvent> {
314    let mut r = ByteReader::new(payload);
315    expect_kind(&mut r, MsgKind::ExecDone)?;
316    let request_id = r.read_i64_le()?;
317    let op_type = r.read_u8()?;
318    let rows_affected = r.read_varint_u64()?;
319    expect_eof(&r, "EXEC_DONE")?;
320    Ok(ServerEvent::ExecDone {
321        request_id,
322        op_type,
323        rows_affected,
324    })
325}
326
327fn decode_cache_reset(payload: &[u8], dict: &mut SymbolDict) -> Result<ServerEvent> {
328    let mut r = ByteReader::new(payload);
329    expect_kind(&mut r, MsgKind::CacheReset)?;
330    let mask = r.read_u8()?;
331    expect_eof(&r, "CACHE_RESET")?;
332    // Per spec §11.7: "Reserved bits MUST be zero on transmit; recipients
333    // MUST ignore any reserved bits that are set." Apply the bits we know;
334    // ignore everything else so a future spec revision adding e.g.
335    // `RESET_MASK_PREPARED` doesn't make older clients reject every
336    // CACHE_RESET that carries the new bit alongside the known ones.
337    if resets_dict(mask) {
338        dict.reset();
339    }
340    Ok(ServerEvent::CacheReset { mask })
341}
342
343fn decode_server_info(payload: &[u8]) -> Result<ServerEvent> {
344    let mut r = ByteReader::new(payload);
345    expect_kind(&mut r, MsgKind::ServerInfo)?;
346    let role = ServerRole::from_u8(r.read_u8()?);
347    let epoch = r.read_u64_le()?;
348    let capabilities = r.read_u32_le()?;
349    let server_wall_ns = r.read_i64_le()?;
350    let cluster_id = read_u16_string(&mut r, "cluster_id")?;
351    let node_id = read_u16_string(&mut r, "node_id")?;
352    // `zone_id` is the only currently-defined trailing field, gated on
353    // CAP_ZONE (wire-egress.md §11.8). A server with capabilities=0
354    // never enters this branch and the byte layout matches the base
355    // spec. Future trailing fields will key off their own capability
356    // bits the same way; unknown bits are silently ignored so an older
357    // client reading a newer server tolerates new fields it doesn't know
358    // how to parse — those bytes get caught by `expect_eof` below until
359    // this client learns to consume them.
360    let zone_id = if has_zone(capabilities) {
361        Some(read_u16_string(&mut r, "zone_id")?)
362    } else {
363        None
364    };
365    expect_eof(&r, "SERVER_INFO")?;
366    Ok(ServerEvent::ServerInfo(ServerInfo {
367        role,
368        epoch,
369        capabilities,
370        server_wall_ns,
371        cluster_id,
372        node_id,
373        zone_id,
374    }))
375}
376
377// ---------------------------------------------------------------------------
378// Helpers
379// ---------------------------------------------------------------------------
380
381fn expect_kind(r: &mut ByteReader<'_>, expected: MsgKind) -> Result<()> {
382    let got = r.read_u8()?;
383    if got != expected.as_u8() {
384        return Err(fmt!(
385            ProtocolError,
386            "expected msg_kind 0x{:02X}, got 0x{:02X}",
387            expected.as_u8(),
388            got
389        ));
390    }
391    Ok(())
392}
393
394fn expect_eof(r: &ByteReader<'_>, msg_name: &str) -> Result<()> {
395    if !r.is_empty() {
396        return Err(fmt!(
397            ProtocolError,
398            "{} has {} trailing bytes",
399            msg_name,
400            r.remaining().len()
401        ));
402    }
403    Ok(())
404}
405
406fn read_u16_string(r: &mut ByteReader<'_>, field: &str) -> Result<String> {
407    let len = r.read_u16_le()? as usize;
408    let bytes = r.read_bytes(len)?;
409    std::str::from_utf8(bytes)
410        .map_err(|e| fmt!(InvalidUtf8, "{} not valid UTF-8: {}", field, e))
411        .map(|s| s.to_string())
412}
413
414// ---------------------------------------------------------------------------
415// Tests
416// ---------------------------------------------------------------------------
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421    use crate::egress::wire::header::{HEADER_LEN, PROTOCOL_VERSION};
422    use crate::egress::wire::varint::encode_u64;
423    use crate::error::ErrorCode;
424
425    fn header(payload_len: usize) -> FrameHeader {
426        FrameHeader {
427            version: PROTOCOL_VERSION,
428            flags: 0,
429            table_count: 0,
430            payload_length: payload_len as u32,
431        }
432    }
433
434    // --- RESULT_END ---------------------------------------------------------
435
436    fn build_result_end(rid: i64, final_seq: u64, total_rows: u64) -> Bytes {
437        let mut p = vec![MsgKind::ResultEnd.as_u8()];
438        p.extend_from_slice(&rid.to_le_bytes());
439        encode_u64(final_seq, &mut p);
440        encode_u64(total_rows, &mut p);
441        Bytes::from(p)
442    }
443
444    #[test]
445    fn decode_result_end_ok() {
446        let payload = build_result_end(42, 7, 1_000);
447        let mut dict = SymbolDict::new();
448        let mut schema: Option<Schema> = None;
449        let event = decode_frame(
450            header(payload.len()),
451            &payload,
452            &mut dict,
453            &mut schema,
454            &mut ZstdScratch::new(),
455        )
456        .unwrap();
457        match event {
458            ServerEvent::End {
459                request_id,
460                final_seq,
461                total_rows,
462            } => {
463                assert_eq!(request_id, 42);
464                assert_eq!(final_seq, 7);
465                assert_eq!(total_rows, 1000);
466            }
467            _ => panic!("wrong event"),
468        }
469    }
470
471    // --- QUERY_ERROR --------------------------------------------------------
472
473    fn build_query_error(rid: i64, status: StatusCode, msg: &str) -> Bytes {
474        let mut p = vec![MsgKind::QueryError.as_u8()];
475        p.extend_from_slice(&rid.to_le_bytes());
476        p.push(status.as_u8());
477        p.extend_from_slice(&(msg.len() as u16).to_le_bytes());
478        p.extend_from_slice(msg.as_bytes());
479        Bytes::from(p)
480    }
481
482    #[test]
483    fn decode_query_error_ok() {
484        let payload = build_query_error(9, StatusCode::ParseError, "bad SQL");
485        let mut dict = SymbolDict::new();
486        let mut schema: Option<Schema> = None;
487        let event = decode_frame(
488            header(payload.len()),
489            &payload,
490            &mut dict,
491            &mut schema,
492            &mut ZstdScratch::new(),
493        )
494        .unwrap();
495        match event {
496            ServerEvent::Error {
497                request_id,
498                status,
499                message,
500            } => {
501                assert_eq!(request_id, 9);
502                assert_eq!(status, StatusCode::ParseError);
503                assert_eq!(message, "bad SQL");
504            }
505            _ => panic!("wrong event"),
506        }
507    }
508
509    #[test]
510    fn query_error_truncated_message_rejected() {
511        let payload = build_query_error(1, StatusCode::InternalError, "details");
512        let truncated = payload.slice(..payload.len() - 3);
513        let mut dict = SymbolDict::new();
514        let mut schema: Option<Schema> = None;
515        let err = decode_frame(
516            header(truncated.len()),
517            &truncated,
518            &mut dict,
519            &mut schema,
520            &mut ZstdScratch::new(),
521        )
522        .unwrap_err();
523        assert_eq!(err.code(), ErrorCode::ProtocolError);
524    }
525
526    #[test]
527    fn query_error_invalid_utf8_rejected() {
528        let mut p = vec![MsgKind::QueryError.as_u8()];
529        p.extend_from_slice(&1i64.to_le_bytes());
530        p.push(StatusCode::InternalError.as_u8());
531        p.extend_from_slice(&2u16.to_le_bytes());
532        p.extend_from_slice(&[0xFF, 0xFE]);
533        let p = Bytes::from(p);
534        let mut dict = SymbolDict::new();
535        let mut schema: Option<Schema> = None;
536        let err = decode_frame(
537            header(p.len()),
538            &p,
539            &mut dict,
540            &mut schema,
541            &mut ZstdScratch::new(),
542        )
543        .unwrap_err();
544        assert_eq!(err.code(), ErrorCode::InvalidUtf8);
545    }
546
547    // --- EXEC_DONE ----------------------------------------------------------
548
549    #[test]
550    fn decode_exec_done_ok() {
551        let mut p = vec![MsgKind::ExecDone.as_u8()];
552        p.extend_from_slice(&5i64.to_le_bytes());
553        p.push(0xAB); // op_type
554        encode_u64(0, &mut p); // rows_affected for DDL
555        let p = Bytes::from(p);
556        let mut dict = SymbolDict::new();
557        let mut schema: Option<Schema> = None;
558        let event = decode_frame(
559            header(p.len()),
560            &p,
561            &mut dict,
562            &mut schema,
563            &mut ZstdScratch::new(),
564        )
565        .unwrap();
566        match event {
567            ServerEvent::ExecDone {
568                request_id,
569                op_type,
570                rows_affected,
571            } => {
572                assert_eq!(request_id, 5);
573                assert_eq!(op_type, 0xAB);
574                assert_eq!(rows_affected, 0);
575            }
576            _ => panic!("wrong event"),
577        }
578    }
579
580    // --- CACHE_RESET --------------------------------------------------------
581
582    fn build_cache_reset(mask: u8) -> Bytes {
583        Bytes::from(vec![MsgKind::CacheReset.as_u8(), mask])
584    }
585
586    #[test]
587    fn cache_reset_clears_dict() {
588        let mut dict = SymbolDict::new();
589        dict.apply_delta(0, [b"x".as_slice()]).unwrap();
590        let mut query_schema: Option<Schema> = None;
591
592        let payload = build_cache_reset(0x01);
593        let event = decode_frame(
594            header(payload.len()),
595            &payload,
596            &mut dict,
597            &mut query_schema,
598            &mut ZstdScratch::new(),
599        )
600        .unwrap();
601        assert!(matches!(event, ServerEvent::CacheReset { mask: 0x01 }));
602        assert_eq!(dict.len(), 0);
603    }
604
605    #[test]
606    fn cache_reset_ignores_reserved_bits() {
607        // Spec §11.7: "Reserved bits MUST be zero on transmit; recipients
608        // MUST ignore any reserved bits that are set." The schemas bit (0x02)
609        // is now reserved (the schema registry is gone); only the DICT bit is
610        // defined. A mask carrying DICT plus reserved bits must still apply
611        // DICT and not error.
612        let mut dict = SymbolDict::new();
613        dict.apply_delta(0, [b"x".as_slice()]).unwrap();
614        let mut query_schema: Option<Schema> = None;
615
616        // 0x83 = bit 0 (DICT) + bit 1 (reserved) + bit 7 (reserved future).
617        let payload = build_cache_reset(0x83);
618        let event = decode_frame(
619            header(payload.len()),
620            &payload,
621            &mut dict,
622            &mut query_schema,
623            &mut ZstdScratch::new(),
624        )
625        .unwrap();
626        assert!(matches!(event, ServerEvent::CacheReset { mask: 0x83 }));
627        assert_eq!(
628            dict.len(),
629            0,
630            "DICT bit must apply even with reserved bits set"
631        );
632    }
633
634    // --- SERVER_INFO --------------------------------------------------------
635
636    fn build_server_info(role: u8, cluster: &str, node: &str) -> Bytes {
637        build_server_info_with(role, 0, cluster, node, None)
638    }
639
640    /// Like `build_server_info` but parameterised over `capabilities` and
641    /// the optional trailing `zone_id`. Used to drive the CAP_ZONE path.
642    fn build_server_info_with(
643        role: u8,
644        capabilities: u32,
645        cluster: &str,
646        node: &str,
647        zone: Option<&str>,
648    ) -> Bytes {
649        let mut p = vec![MsgKind::ServerInfo.as_u8()];
650        p.push(role);
651        p.extend_from_slice(&7u64.to_le_bytes()); // epoch
652        p.extend_from_slice(&capabilities.to_le_bytes());
653        p.extend_from_slice(&123_456_789i64.to_le_bytes()); // server_wall_ns
654        p.extend_from_slice(&(cluster.len() as u16).to_le_bytes());
655        p.extend_from_slice(cluster.as_bytes());
656        p.extend_from_slice(&(node.len() as u16).to_le_bytes());
657        p.extend_from_slice(node.as_bytes());
658        if let Some(z) = zone {
659            p.extend_from_slice(&(z.len() as u16).to_le_bytes());
660            p.extend_from_slice(z.as_bytes());
661        }
662        Bytes::from(p)
663    }
664
665    #[test]
666    fn decode_server_info_primary() {
667        let payload = build_server_info(0x01, "cluster-A", "node-1");
668        let mut dict = SymbolDict::new();
669        let mut schema: Option<Schema> = None;
670        let event = decode_frame(
671            header(payload.len()),
672            &payload,
673            &mut dict,
674            &mut schema,
675            &mut ZstdScratch::new(),
676        )
677        .unwrap();
678        let ServerEvent::ServerInfo(info) = event else {
679            panic!()
680        };
681        assert_eq!(info.role, ServerRole::Primary);
682        assert_eq!(info.epoch, 7);
683        assert_eq!(info.capabilities, 0);
684        assert_eq!(info.server_wall_ns, 123_456_789);
685        assert_eq!(info.cluster_id, "cluster-A");
686        assert_eq!(info.node_id, "node-1");
687        assert_eq!(info.zone_id, None, "CAP_ZONE=0 leaves zone_id absent");
688    }
689
690    #[test]
691    fn unknown_role_byte_is_other_variant() {
692        let payload = build_server_info(0x55, "c", "n");
693        let mut dict = SymbolDict::new();
694        let mut schema: Option<Schema> = None;
695        let event = decode_frame(
696            header(payload.len()),
697            &payload,
698            &mut dict,
699            &mut schema,
700            &mut ZstdScratch::new(),
701        )
702        .unwrap();
703        let ServerEvent::ServerInfo(info) = event else {
704            panic!()
705        };
706        assert_eq!(info.role, ServerRole::Other(0x55));
707    }
708
709    #[test]
710    fn decode_server_info_with_cap_zone_reads_zone_id() {
711        // CAP_ZONE bit set → trailing zone_id is mandatory per §11.8.
712        let payload = build_server_info_with(
713            0x01,
714            crate::egress::wire::CAP_ZONE,
715            "cluster-A",
716            "node-1",
717            Some("eu-west-1a"),
718        );
719        let mut dict = SymbolDict::new();
720        let mut schema: Option<Schema> = None;
721        let event = decode_frame(
722            header(payload.len()),
723            &payload,
724            &mut dict,
725            &mut schema,
726            &mut ZstdScratch::new(),
727        )
728        .unwrap();
729        let ServerEvent::ServerInfo(info) = event else {
730            panic!()
731        };
732        assert_eq!(
733            info.capabilities & crate::egress::wire::CAP_ZONE,
734            crate::egress::wire::CAP_ZONE
735        );
736        assert_eq!(info.zone_id.as_deref(), Some("eu-west-1a"));
737    }
738
739    #[test]
740    fn cap_zone_set_but_zone_id_missing_is_protocol_error() {
741        // The server claims CAP_ZONE but omits the trailing field. Decode
742        // must fail rather than swallow the inconsistency — a server bug
743        // that ships uninitialised state should surface, not be silently
744        // tolerated.
745        let payload = build_server_info_with(
746            0x01,
747            crate::egress::wire::CAP_ZONE,
748            "c",
749            "n",
750            None, // no trailing zone_id despite CAP_ZONE
751        );
752        let mut dict = SymbolDict::new();
753        let mut schema: Option<Schema> = None;
754        let err = decode_frame(
755            header(payload.len()),
756            &payload,
757            &mut dict,
758            &mut schema,
759            &mut ZstdScratch::new(),
760        )
761        .unwrap_err();
762        assert_eq!(err.code(), ErrorCode::ProtocolError);
763    }
764
765    #[test]
766    fn unknown_capabilities_bit_with_trailing_bytes_is_protocol_error() {
767        // A future capability bit gates further trailing fields. A
768        // client that doesn't understand the bit MUST still reject
769        // unknown trailing bytes (caught by `expect_eof`) rather than
770        // silently ignoring them — the server is supposed to omit
771        // trailers behind bits the negotiated revision didn't define.
772        let mut payload = build_server_info(0x01, "c", "n").to_vec();
773        // Patch capabilities to a known-zero word — the trailing bytes
774        // below should then surface as `SERVER_INFO has N trailing bytes`.
775        payload.extend_from_slice(&[0xDE, 0xAD]);
776        let payload = Bytes::from(payload);
777        let mut dict = SymbolDict::new();
778        let mut schema: Option<Schema> = None;
779        let err = decode_frame(
780            header(payload.len()),
781            &payload,
782            &mut dict,
783            &mut schema,
784            &mut ZstdScratch::new(),
785        )
786        .unwrap_err();
787        assert_eq!(err.code(), ErrorCode::ProtocolError);
788    }
789
790    // --- Dispatcher edge cases ---------------------------------------------
791
792    #[test]
793    fn empty_payload_rejected() {
794        let mut dict = SymbolDict::new();
795        let mut schema: Option<Schema> = None;
796        let empty = Bytes::new();
797        let err = decode_frame(
798            header(0),
799            &empty,
800            &mut dict,
801            &mut schema,
802            &mut ZstdScratch::new(),
803        )
804        .unwrap_err();
805        assert_eq!(err.code(), ErrorCode::ProtocolError);
806    }
807
808    #[test]
809    fn unknown_msg_kind_rejected() {
810        let mut dict = SymbolDict::new();
811        let mut schema: Option<Schema> = None;
812        let p = Bytes::from(vec![0xAA]);
813        let err = decode_frame(
814            header(1),
815            &p,
816            &mut dict,
817            &mut schema,
818            &mut ZstdScratch::new(),
819        )
820        .unwrap_err();
821        assert_eq!(err.code(), ErrorCode::ProtocolError);
822    }
823
824    #[test]
825    fn client_only_kinds_rejected_from_server() {
826        for k in [
827            MsgKind::QueryRequest.as_u8(),
828            MsgKind::Cancel.as_u8(),
829            MsgKind::Credit.as_u8(),
830        ] {
831            let mut dict = SymbolDict::new();
832            let mut schema: Option<Schema> = None;
833            let p = Bytes::from(vec![k]);
834            let err = decode_frame(
835                header(1),
836                &p,
837                &mut dict,
838                &mut schema,
839                &mut ZstdScratch::new(),
840            )
841            .unwrap_err();
842            assert_eq!(err.code(), ErrorCode::ProtocolError);
843            assert!(err.msg().contains("client-only"));
844        }
845    }
846
847    #[test]
848    fn trailing_bytes_rejected_for_simple_messages() {
849        let payload = build_result_end(1, 0, 0);
850        let mut bytes_vec: Vec<u8> = payload.to_vec();
851        bytes_vec.push(0xFF);
852        let payload = Bytes::from(bytes_vec);
853        let mut dict = SymbolDict::new();
854        let mut schema: Option<Schema> = None;
855        let err = decode_frame(
856            header(payload.len()),
857            &payload,
858            &mut dict,
859            &mut schema,
860            &mut ZstdScratch::new(),
861        )
862        .unwrap_err();
863        assert_eq!(err.code(), ErrorCode::ProtocolError);
864    }
865
866    // Sanity: HEADER_LEN constant still wired up.
867    #[test]
868    fn header_len_is_12() {
869        assert_eq!(HEADER_LEN, 12);
870    }
871
872    #[test]
873    fn upgrade_reject_round_trips() {
874        let r = UpgradeReject::new(
875            roles::PRIMARY_CATCHUP,
876            roles::NAME_PRIMARY_CATCHUP,
877            Some("eu-west-1a".into()),
878        );
879        assert_eq!(r.role_byte, roles::PRIMARY_CATCHUP);
880        assert_eq!(r.zone.as_deref(), Some("eu-west-1a"));
881        assert!(r.is_transient());
882    }
883
884    #[test]
885    fn upgrade_reject_topological_for_non_catchup_roles() {
886        // STANDALONE / PRIMARY / REPLICA / unknown all classify as
887        // topological (won't recover without topology change).
888        for (byte, name) in [
889            (roles::STANDALONE, roles::NAME_STANDALONE),
890            (roles::PRIMARY, roles::NAME_PRIMARY),
891            (roles::REPLICA, roles::NAME_REPLICA),
892            (0x99, "FUTURE_ROLE"),
893        ] {
894            let r = UpgradeReject::new(byte, name, None);
895            assert!(!r.is_transient(), "role {} should be topological", name);
896        }
897    }
898
899    #[test]
900    fn upgrade_reject_is_transient_case_insensitive() {
901        let r = UpgradeReject::new(0x99, "primary_catchup", None);
902        assert!(r.is_transient(), "case-insensitive match per spec §5");
903    }
904}
questdb/egress/server_event.rs

questdb/egress/
server_event.rs
