questdb/egress/wire/header.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//! 12-byte QWP frame header. All multi-byte fields little-endian.
26//!
27//! ```text
28//! Offset Size Field Description
29//! 0 4 magic "QWP1" = 0x31_50_57_51 LE
30//! 4 1 version Negotiated QWP version
31//! 5 1 flags Per-message flag bits
32//! 6 2 table_count 1 for RESULT_BATCH; 0 otherwise
33//! 8 4 payload_length Payload size in bytes
34//! ```
35
36use crate::error::{Result, fmt};
37
38/// `"QWP1"` interpreted as a little-endian `u32`.
39pub const MAGIC: u32 = u32::from_le_bytes(*b"QWP1");
40
41/// The single QWP protocol version. Every frame's `version` byte is written
42/// as this value and validated to equal it on parse.
43pub const PROTOCOL_VERSION: u8 = 1;
44
45/// Length of the wire frame header in bytes.
46pub const HEADER_LEN: usize = 12;
47
48/// Per-frame flag bits (`flags` byte).
49pub mod flags {
50 /// Timestamp/date columns may use delta-of-delta (Gorilla) encoding.
51 pub const GORILLA: u8 = 0x04;
52 /// `RESULT_BATCH` carries a delta symbol-dict section.
53 pub const DELTA_SYMBOL_DICT: u8 = 0x08;
54 /// Payload (after `msg_kind/request_id/batch_seq`) is zstd-compressed.
55 pub const ZSTD: u8 = 0x10;
56}
57
58/// Parsed wire frame header.
59#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60pub struct FrameHeader {
61 pub version: u8,
62 pub flags: u8,
63 pub table_count: u16,
64 pub payload_length: u32,
65}
66
67impl FrameHeader {
68 /// Parse a header from exactly [`HEADER_LEN`] bytes.
69 pub fn parse(bytes: &[u8]) -> Result<Self> {
70 if bytes.len() < HEADER_LEN {
71 return Err(fmt!(
72 ProtocolError,
73 "frame header truncated: got {} bytes, need {}",
74 bytes.len(),
75 HEADER_LEN
76 ));
77 }
78 let magic = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
79 if magic != MAGIC {
80 return Err(fmt!(
81 ProtocolError,
82 "bad frame magic: 0x{:08X} (expected 0x{:08X})",
83 magic,
84 MAGIC
85 ));
86 }
87 // QWP runs at a single version. Each frame fails fast on its own here,
88 // independently of the negotiated version checked in `transport.rs`.
89 let version = bytes[4];
90 if version != PROTOCOL_VERSION {
91 return Err(fmt!(
92 ProtocolError,
93 "unsupported QWP frame version {} (expected {})",
94 version,
95 PROTOCOL_VERSION
96 ));
97 }
98 Ok(FrameHeader {
99 version,
100 flags: bytes[5],
101 table_count: u16::from_le_bytes([bytes[6], bytes[7]]),
102 payload_length: u32::from_le_bytes([bytes[8], bytes[9], bytes[10], bytes[11]]),
103 })
104 }
105
106 /// Serialize this header into the first [`HEADER_LEN`] bytes of `out`.
107 ///
108 /// The `version` byte is always written as [`PROTOCOL_VERSION`] — the only
109 /// value [`FrameHeader::parse`] accepts — and `self.version` is
110 /// debug-asserted to match, so a header built with a stale version can't
111 /// serialize bytes this module would then refuse to parse.
112 pub fn write(self, out: &mut [u8; HEADER_LEN]) {
113 debug_assert_eq!(
114 self.version, PROTOCOL_VERSION,
115 "FrameHeader::write must only serialize the pinned protocol version"
116 );
117 out[0..4].copy_from_slice(&MAGIC.to_le_bytes());
118 out[4] = PROTOCOL_VERSION;
119 out[5] = self.flags;
120 out[6..8].copy_from_slice(&self.table_count.to_le_bytes());
121 out[8..12].copy_from_slice(&self.payload_length.to_le_bytes());
122 }
123
124 /// Convenience: write into a fresh `[u8; 12]`.
125 pub fn to_bytes(self) -> [u8; HEADER_LEN] {
126 let mut out = [0u8; HEADER_LEN];
127 self.write(&mut out);
128 out
129 }
130}
131
132#[cfg(test)]
133mod tests {
134 use super::*;
135 use crate::error::ErrorCode;
136
137 #[test]
138 fn magic_is_qwp1_le() {
139 assert_eq!(&MAGIC.to_le_bytes(), b"QWP1");
140 }
141
142 #[test]
143 fn roundtrip() {
144 let h = FrameHeader {
145 version: 1,
146 flags: flags::GORILLA | flags::DELTA_SYMBOL_DICT,
147 table_count: 1,
148 payload_length: 0xDEAD_BEEF,
149 };
150 let bytes = h.to_bytes();
151 let parsed = FrameHeader::parse(&bytes).unwrap();
152 assert_eq!(parsed, h);
153 }
154
155 #[test]
156 fn truncated_rejected() {
157 let bytes = [0u8; HEADER_LEN - 1];
158 assert_eq!(
159 FrameHeader::parse(&bytes).unwrap_err().code(),
160 ErrorCode::ProtocolError
161 );
162 }
163
164 #[test]
165 fn bad_magic_rejected() {
166 let mut bytes = [0u8; HEADER_LEN];
167 bytes[0..4].copy_from_slice(b"NOPE");
168 assert_eq!(
169 FrameHeader::parse(&bytes).unwrap_err().code(),
170 ErrorCode::ProtocolError
171 );
172 }
173
174 #[test]
175 fn wrong_version_rejected() {
176 // Any version byte other than the pinned one — below it (0) or
177 // above it (2, 0xFF) — must be rejected per-frame, independently
178 // of the handshake-level check.
179 let valid = FrameHeader {
180 version: PROTOCOL_VERSION,
181 flags: 0,
182 table_count: 0,
183 payload_length: 0,
184 }
185 .to_bytes();
186 for wrong in [0u8, 2, 0xFF] {
187 let mut bytes = valid;
188 bytes[4] = wrong;
189 let err = FrameHeader::parse(&bytes).unwrap_err();
190 assert_eq!(err.code(), ErrorCode::ProtocolError);
191 assert!(
192 err.msg().contains(&format!("version {wrong}")),
193 "message should name the rejected version {wrong}: {}",
194 err.msg()
195 );
196 }
197 }
198
199 #[test]
200 fn extra_bytes_ignored() {
201 let h = FrameHeader {
202 version: 1,
203 flags: 0,
204 table_count: 0,
205 payload_length: 0,
206 };
207 let mut buf = vec![0u8; HEADER_LEN + 8];
208 let mut hdr_buf = [0u8; HEADER_LEN];
209 h.write(&mut hdr_buf);
210 buf[..HEADER_LEN].copy_from_slice(&hdr_buf);
211 let parsed = FrameHeader::parse(&buf).unwrap();
212 assert_eq!(parsed, h);
213 }
214}