reddb_wire/redwire/
frame.rs

1//! RedWire frame layout — 16-byte header + payload, little-endian.
2//!
3//! ```text
4//! ┌──────────────────────────────────────────────────────────┐
5//! │ Header (16 bytes)                                         │
6//! │   u32   length         total frame size, incl. header     │
7//! │   u8    kind           MessageKind                         │
8//! │   u8    flags          COMPRESSED | MORE_FRAMES | …        │
9//! │   u16   stream_id      0 = unsolicited; otherwise multiplex│
10//! │   u64   correlation_id request↔response pairing           │
11//! ├──────────────────────────────────────────────────────────┤
12//! │ Payload (length - 16 bytes)                               │
13//! └──────────────────────────────────────────────────────────┘
14//! ```
15//!
16//! Data-plane kinds live at 0x01..0x0F; handshake / lifecycle at
17//! 0x10..0x1F; control plane at 0x20..0x3F.
18
19pub const FRAME_HEADER_SIZE: usize = 16;
20pub const MAX_FRAME_SIZE: u32 = 16 * 1024 * 1024;
21
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct Frame {
24    pub kind: MessageKind,
25    pub flags: Flags,
26    pub stream_id: u16,
27    pub correlation_id: u64,
28    pub payload: Vec<u8>,
29}
30
31impl Frame {
32    pub fn new(kind: MessageKind, correlation_id: u64, payload: Vec<u8>) -> Self {
33        Self {
34            kind,
35            flags: Flags::empty(),
36            stream_id: 0,
37            correlation_id,
38            payload,
39        }
40    }
41
42    pub fn with_stream(mut self, stream_id: u16) -> Self {
43        self.stream_id = stream_id;
44        self
45    }
46
47    pub fn with_flags(mut self, flags: Flags) -> Self {
48        self.flags = flags;
49        self
50    }
51
52    pub fn encoded_len(&self) -> u32 {
53        (FRAME_HEADER_SIZE + self.payload.len()) as u32
54    }
55}
56
57/// Single-byte message-kind discriminator. Numeric values are
58/// part of the wire spec — never repurpose a value once shipped.
59#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60#[repr(u8)]
61pub enum MessageKind {
62    // Data-plane codes.
63    Query = 0x01,
64    Result = 0x02,
65    Error = 0x03,
66    BulkInsert = 0x04,
67    BulkOk = 0x05,
68    BulkInsertBinary = 0x06,
69    QueryBinary = 0x07,
70    BulkInsertPrevalidated = 0x08,
71    BulkStreamStart = 0x09,
72    BulkStreamRows = 0x0A,
73    BulkStreamCommit = 0x0B,
74    BulkStreamAck = 0x0C,
75    Prepare = 0x0D,
76    PreparedOk = 0x0E,
77    ExecutePrepared = 0x0F,
78
79    // Handshake / lifecycle.
80    Hello = 0x10,
81    HelloAck = 0x11,
82    AuthRequest = 0x12,
83    AuthResponse = 0x13,
84    AuthOk = 0x14,
85    AuthFail = 0x15,
86    Bye = 0x16,
87    Ping = 0x17,
88    Pong = 0x18,
89    Get = 0x19,
90    Delete = 0x1A,
91    DeleteOk = 0x1B,
92
93    // Control plane.
94    Cancel = 0x20,
95    Compress = 0x21,
96    SetSession = 0x22,
97    Notice = 0x23,
98
99    // Streamed responses.
100    RowDescription = 0x24,
101    StreamEnd = 0x25,
102
103    // RedDB-native data plane.
104    VectorSearch = 0x26,
105    GraphTraverse = 0x27,
106    QueryWithParams = 0x28,
107
108    // Output stream lifecycle (issue #762 / PRD #759 S3). Streamed
109    // class — these envelopes describe an in-flight multiplexed
110    // stream over the existing `stream_id` field.
111    OpenStream = 0x29,
112    OpenAck = 0x2A,
113    StreamChunk = 0x2B,
114    StreamError = 0x2C,
115    StreamCancel = 0x2D,
116
117    // Live queue wait (issue #917 / PRD #915). A `QueueWaitOpen`
118    // registers an async waiter on a queue; the server pushes a
119    // `QueueEventPush` the instant a message becomes deliverable.
120    // Distinct from the OpenStream/StreamChunk output-stream family
121    // (which stays query-result pull) — these carry queue delivery,
122    // multiplexed over the frame's `stream_id` like the other
123    // streamed envelopes.
124    QueueWaitOpen = 0x2E,
125    QueueEventPush = 0x2F,
126    // Live queue-wait timeout (issue #919 / PRD #915). Pushed when a
127    // wait elapses with no deliverable message — a distinct kind so the
128    // outcome is unambiguous on the wire: not a `QueueEventPush`
129    // (delivery), not a `StreamError` (cancellation / failure).
130    QueueWaitTimeout = 0x30,
131    // Stale-ownership redirect. Carries a MOVED payload naming the current
132    // range owner, ownership epoch, and catalog version.
133    MovedRedirect = 0x31,
134    /// Authenticated topology refresh. Reply: Result with canonical topology bytes.
135    Topology = 0x32,
136}
137
138/// Coarse routing class for a `MessageKind`.
139///
140/// The numeric ranges in the wire spec (0x01..0x0F data plane,
141/// 0x10..0x1F handshake/lifecycle, 0x20..0x3F control plane) are
142/// turned into a typed catalog so dispatch sites can interrogate
143/// a kind's role without re-implementing the comment-grouped match
144/// arms. `Streamed` is split out from `DataPlane` for kinds that
145/// describe an in-flight stream envelope rather than a request/reply.
146#[derive(Debug, Clone, Copy, PartialEq, Eq)]
147pub enum MessageClass {
148    DataPlane,
149    Handshake,
150    ControlPlane,
151    Streamed,
152}
153
154/// Who is allowed to put this kind on the wire.
155///
156/// The handshake and lifecycle frames split cleanly between the two
157/// peers (Hello is client→server, HelloAck is server→client, etc.);
158/// the data-plane request/reply pairs follow the same split. `Both`
159/// is reserved for symmetric frames such as `Bye` (either side may
160/// initiate the disconnect).
161#[derive(Debug, Clone, Copy, PartialEq, Eq)]
162pub enum MessageDirection {
163    ClientToServer,
164    ServerToClient,
165    Both,
166}
167
168impl MessageKind {
169    /// Routing class derived from the comment-grouped wire ranges.
170    pub fn class(&self) -> MessageClass {
171        match self {
172            // 0x01..0x0F — data plane request/reply pairs. The
173            // BulkStream* family is in this range for backward
174            // compatibility but is reclassified as `Streamed` so
175            // dispatch can treat it as a long-running envelope.
176            Self::Query
177            | Self::Result
178            | Self::Error
179            | Self::BulkInsert
180            | Self::BulkOk
181            | Self::BulkInsertBinary
182            | Self::QueryBinary
183            | Self::BulkInsertPrevalidated
184            | Self::Prepare
185            | Self::PreparedOk
186            | Self::ExecutePrepared
187            | Self::Get
188            | Self::Delete
189            | Self::DeleteOk
190            | Self::VectorSearch
191            | Self::GraphTraverse
192            | Self::QueryWithParams
193            | Self::MovedRedirect => MessageClass::DataPlane,
194
195            // BulkStream* + RowDescription/StreamEnd describe an
196            // in-flight stream rather than a single round trip.
197            // OpenStream / OpenAck / StreamChunk / StreamError /
198            // StreamCancel (issue #762) also describe an in-flight
199            // multiplexed stream and share the same class.
200            Self::BulkStreamStart
201            | Self::BulkStreamRows
202            | Self::BulkStreamCommit
203            | Self::BulkStreamAck
204            | Self::RowDescription
205            | Self::StreamEnd
206            | Self::OpenStream
207            | Self::OpenAck
208            | Self::StreamChunk
209            | Self::StreamError
210            | Self::StreamCancel
211            // Live queue-wait envelopes (issue #917) describe an
212            // in-flight subscription multiplexed by `stream_id`.
213            | Self::QueueWaitOpen
214            | Self::QueueEventPush
215            // Live queue-wait timeout push (issue #919) — same streamed
216            // class, multiplexed by `stream_id`.
217            | Self::QueueWaitTimeout => MessageClass::Streamed,
218
219            // 0x10..0x1F — handshake / lifecycle.
220            Self::Hello
221            | Self::HelloAck
222            | Self::AuthRequest
223            | Self::AuthResponse
224            | Self::AuthOk
225            | Self::AuthFail
226            | Self::Bye
227            | Self::Ping
228            | Self::Pong => MessageClass::Handshake,
229
230            // 0x20..0x3F — control plane.
231            Self::Cancel | Self::Compress | Self::SetSession | Self::Notice | Self::Topology => {
232                MessageClass::ControlPlane
233            }
234        }
235    }
236
237    /// Bitset of `Flags` values this kind may legitimately carry.
238    ///
239    /// Pinned conservatively: `MORE_FRAMES` is universal (any frame
240    /// may be split), but `COMPRESSED` is whitelisted only on kinds
241    /// whose payloads are big enough to benefit from compression.
242    /// Handshake/lifecycle payloads (Hello, AuthRequest, Ping, …)
243    /// are tiny and stay uncompressed today; future contributors
244    /// who want to flip that decision must update both the matrix
245    /// and the unit tests that pin it.
246    pub fn allowed_flags(&self) -> Flags {
247        match self {
248            // Handshake / lifecycle — tiny payloads, never
249            // compressed today.
250            Self::Hello
251            | Self::HelloAck
252            | Self::AuthRequest
253            | Self::AuthResponse
254            | Self::AuthOk
255            | Self::AuthFail
256            | Self::Bye
257            | Self::Ping
258            | Self::Pong => Flags::MORE_FRAMES,
259
260            // Everything else may carry both documented flags.
261            _ => Flags::COMPRESSED.insert(Flags::MORE_FRAMES),
262        }
263    }
264
265    /// `true` when this kind belongs to the handshake/lifecycle group
266    /// (Hello, AuthRequest, AuthOk, …, Bye, Ping, Pong). Equivalent to
267    /// `class() == MessageClass::Handshake` and exists so dispatch sites
268    /// can read the predicate without importing `MessageClass`.
269    pub fn is_handshake(&self) -> bool {
270        matches!(self.class(), MessageClass::Handshake)
271    }
272
273    /// `true` when every flag bit in `flags` is in `allowed_flags()`.
274    /// The catalog is the single source of truth for which flag bits a
275    /// kind may carry; both the codec (decode side) and the builder
276    /// (encode side) consult this so a misframed frame fails at the
277    /// boundary rather than reaching the dispatch arms.
278    pub fn permits_flags(&self, flags: Flags) -> bool {
279        let allowed = self.allowed_flags().bits();
280        (flags.bits() & !allowed) == 0
281    }
282
283    /// Which peer is allowed to originate this kind.
284    pub fn direction(&self) -> MessageDirection {
285        match self {
286            // Client-originated requests.
287            Self::Hello
288            | Self::AuthResponse
289            | Self::Query
290            | Self::QueryBinary
291            | Self::BulkInsert
292            | Self::BulkInsertBinary
293            | Self::BulkInsertPrevalidated
294            | Self::BulkStreamStart
295            | Self::BulkStreamRows
296            | Self::BulkStreamCommit
297            | Self::Prepare
298            | Self::ExecutePrepared
299            | Self::Get
300            | Self::Delete
301            | Self::Cancel
302            | Self::Compress
303            | Self::SetSession
304            | Self::VectorSearch
305            | Self::GraphTraverse
306            | Self::QueryWithParams
307            | Self::Topology
308            | Self::OpenStream
309            | Self::StreamCancel
310            // Client opens a live queue wait (issue #917).
311            | Self::QueueWaitOpen => MessageDirection::ClientToServer,
312
313            // `StreamChunk` is symmetric (issue #764 / PRD #759 S5):
314            // the server emits chunks on an *output* stream, and the
315            // client emits chunks of rows on an *input* stream. Both
316            // are routed by the frame's `stream_id`, so the kind has
317            // to be legal in either direction.
318            Self::StreamChunk => MessageDirection::Both,
319
320            // Server-originated replies / push frames.
321            Self::HelloAck
322            | Self::AuthRequest
323            | Self::AuthOk
324            | Self::AuthFail
325            | Self::Result
326            | Self::Error
327            | Self::BulkOk
328            | Self::BulkStreamAck
329            | Self::PreparedOk
330            | Self::DeleteOk
331            | Self::Notice
332            | Self::RowDescription
333            | Self::StreamEnd
334            | Self::OpenAck
335            | Self::StreamError
336            // Server pushes the delivered queue message (issue #917)
337            // and the distinct wait-timeout outcome (issue #919).
338            | Self::QueueEventPush
339            | Self::QueueWaitTimeout
340            | Self::MovedRedirect => MessageDirection::ServerToClient,
341
342            // Symmetric — either peer may initiate. (`StreamChunk` is
343            // also symmetric but has its own arm above — see issue
344            // #764.)
345            Self::Bye | Self::Ping | Self::Pong => MessageDirection::Both,
346        }
347    }
348
349    pub fn from_u8(byte: u8) -> Option<Self> {
350        match byte {
351            0x01 => Some(Self::Query),
352            0x02 => Some(Self::Result),
353            0x03 => Some(Self::Error),
354            0x04 => Some(Self::BulkInsert),
355            0x05 => Some(Self::BulkOk),
356            0x06 => Some(Self::BulkInsertBinary),
357            0x07 => Some(Self::QueryBinary),
358            0x08 => Some(Self::BulkInsertPrevalidated),
359            0x09 => Some(Self::BulkStreamStart),
360            0x0A => Some(Self::BulkStreamRows),
361            0x0B => Some(Self::BulkStreamCommit),
362            0x0C => Some(Self::BulkStreamAck),
363            0x0D => Some(Self::Prepare),
364            0x0E => Some(Self::PreparedOk),
365            0x0F => Some(Self::ExecutePrepared),
366            0x10 => Some(Self::Hello),
367            0x11 => Some(Self::HelloAck),
368            0x12 => Some(Self::AuthRequest),
369            0x13 => Some(Self::AuthResponse),
370            0x14 => Some(Self::AuthOk),
371            0x15 => Some(Self::AuthFail),
372            0x16 => Some(Self::Bye),
373            0x17 => Some(Self::Ping),
374            0x18 => Some(Self::Pong),
375            0x19 => Some(Self::Get),
376            0x1A => Some(Self::Delete),
377            0x1B => Some(Self::DeleteOk),
378            0x20 => Some(Self::Cancel),
379            0x21 => Some(Self::Compress),
380            0x22 => Some(Self::SetSession),
381            0x23 => Some(Self::Notice),
382            0x24 => Some(Self::RowDescription),
383            0x25 => Some(Self::StreamEnd),
384            0x26 => Some(Self::VectorSearch),
385            0x27 => Some(Self::GraphTraverse),
386            0x28 => Some(Self::QueryWithParams),
387            0x29 => Some(Self::OpenStream),
388            0x2A => Some(Self::OpenAck),
389            0x2B => Some(Self::StreamChunk),
390            0x2C => Some(Self::StreamError),
391            0x2D => Some(Self::StreamCancel),
392            0x2E => Some(Self::QueueWaitOpen),
393            0x2F => Some(Self::QueueEventPush),
394            0x30 => Some(Self::QueueWaitTimeout),
395            0x31 => Some(Self::MovedRedirect),
396            0x32 => Some(Self::Topology),
397            _ => None,
398        }
399    }
400}
401
402#[derive(Debug, Clone, Copy, PartialEq, Eq)]
403pub struct Flags(u8);
404
405impl Flags {
406    pub const COMPRESSED: Self = Self(0b0000_0001);
407    pub const MORE_FRAMES: Self = Self(0b0000_0010);
408
409    pub const fn empty() -> Self {
410        Self(0)
411    }
412
413    pub const fn bits(self) -> u8 {
414        self.0
415    }
416
417    pub const fn from_bits(bits: u8) -> Self {
418        Self(bits)
419    }
420
421    pub const fn contains(self, other: Self) -> bool {
422        (self.0 & other.0) == other.0
423    }
424
425    pub const fn insert(self, other: Self) -> Self {
426        Self(self.0 | other.0)
427    }
428}
429
430impl std::ops::BitOr for Flags {
431    type Output = Self;
432    fn bitor(self, rhs: Self) -> Self {
433        self.insert(rhs)
434    }
435}
436
437#[cfg(test)]
438mod catalog_tests {
439    use super::*;
440
441    /// Every kind known to the wire spec — kept in sync with the
442    /// `from_u8` table. New entries must be added here so the
443    /// matrix tests below cover them.
444    const ALL_KINDS: &[MessageKind] = &[
445        MessageKind::Query,
446        MessageKind::Result,
447        MessageKind::Error,
448        MessageKind::BulkInsert,
449        MessageKind::BulkOk,
450        MessageKind::BulkInsertBinary,
451        MessageKind::QueryBinary,
452        MessageKind::BulkInsertPrevalidated,
453        MessageKind::BulkStreamStart,
454        MessageKind::BulkStreamRows,
455        MessageKind::BulkStreamCommit,
456        MessageKind::BulkStreamAck,
457        MessageKind::Prepare,
458        MessageKind::PreparedOk,
459        MessageKind::ExecutePrepared,
460        MessageKind::Hello,
461        MessageKind::HelloAck,
462        MessageKind::AuthRequest,
463        MessageKind::AuthResponse,
464        MessageKind::AuthOk,
465        MessageKind::AuthFail,
466        MessageKind::Bye,
467        MessageKind::Ping,
468        MessageKind::Pong,
469        MessageKind::Get,
470        MessageKind::Delete,
471        MessageKind::DeleteOk,
472        MessageKind::Cancel,
473        MessageKind::Compress,
474        MessageKind::SetSession,
475        MessageKind::Notice,
476        MessageKind::RowDescription,
477        MessageKind::StreamEnd,
478        MessageKind::VectorSearch,
479        MessageKind::GraphTraverse,
480        MessageKind::QueryWithParams,
481        MessageKind::OpenStream,
482        MessageKind::OpenAck,
483        MessageKind::StreamChunk,
484        MessageKind::StreamError,
485        MessageKind::StreamCancel,
486        MessageKind::QueueWaitOpen,
487        MessageKind::QueueEventPush,
488        MessageKind::QueueWaitTimeout,
489        MessageKind::MovedRedirect,
490        MessageKind::Topology,
491    ];
492
493    #[test]
494    fn class_matrix_is_pinned() {
495        // Handshake / lifecycle (0x10..0x1F minus Get/Delete/DeleteOk
496        // which are data plane despite the historic numbering).
497        assert_eq!(MessageKind::Hello.class(), MessageClass::Handshake);
498        assert_eq!(MessageKind::HelloAck.class(), MessageClass::Handshake);
499        assert_eq!(MessageKind::AuthRequest.class(), MessageClass::Handshake);
500        assert_eq!(MessageKind::AuthResponse.class(), MessageClass::Handshake);
501        assert_eq!(MessageKind::AuthOk.class(), MessageClass::Handshake);
502        assert_eq!(MessageKind::AuthFail.class(), MessageClass::Handshake);
503        assert_eq!(MessageKind::Bye.class(), MessageClass::Handshake);
504        assert_eq!(MessageKind::Ping.class(), MessageClass::Handshake);
505        assert_eq!(MessageKind::Pong.class(), MessageClass::Handshake);
506
507        // Data plane.
508        assert_eq!(MessageKind::Query.class(), MessageClass::DataPlane);
509        assert_eq!(MessageKind::Result.class(), MessageClass::DataPlane);
510        assert_eq!(MessageKind::BulkInsert.class(), MessageClass::DataPlane);
511        assert_eq!(MessageKind::Get.class(), MessageClass::DataPlane);
512        assert_eq!(MessageKind::Delete.class(), MessageClass::DataPlane);
513        assert_eq!(MessageKind::DeleteOk.class(), MessageClass::DataPlane);
514        assert_eq!(MessageKind::VectorSearch.class(), MessageClass::DataPlane);
515        assert_eq!(MessageKind::GraphTraverse.class(), MessageClass::DataPlane);
516        assert_eq!(
517            MessageKind::QueryWithParams.class(),
518            MessageClass::DataPlane
519        );
520        assert_eq!(MessageKind::MovedRedirect.class(), MessageClass::DataPlane);
521
522        // Streamed envelopes.
523        assert_eq!(MessageKind::BulkStreamStart.class(), MessageClass::Streamed);
524        assert_eq!(MessageKind::BulkStreamRows.class(), MessageClass::Streamed);
525        assert_eq!(
526            MessageKind::BulkStreamCommit.class(),
527            MessageClass::Streamed
528        );
529        assert_eq!(MessageKind::BulkStreamAck.class(), MessageClass::Streamed);
530        assert_eq!(MessageKind::RowDescription.class(), MessageClass::Streamed);
531        assert_eq!(MessageKind::StreamEnd.class(), MessageClass::Streamed);
532
533        // Output stream lifecycle envelopes (issue #762). All in
534        // the Streamed class — they describe in-flight multiplexed
535        // streams identified by the frame's `stream_id`.
536        assert_eq!(MessageKind::OpenStream.class(), MessageClass::Streamed);
537        assert_eq!(MessageKind::OpenAck.class(), MessageClass::Streamed);
538        assert_eq!(MessageKind::StreamChunk.class(), MessageClass::Streamed);
539        assert_eq!(MessageKind::StreamError.class(), MessageClass::Streamed);
540        assert_eq!(MessageKind::StreamCancel.class(), MessageClass::Streamed);
541
542        // Live queue-wait envelopes (issue #917) — Streamed class,
543        // multiplexed by `stream_id` like the output-stream family.
544        assert_eq!(MessageKind::QueueWaitOpen.class(), MessageClass::Streamed);
545        assert_eq!(MessageKind::QueueEventPush.class(), MessageClass::Streamed);
546        // Live queue-wait timeout push (issue #919).
547        assert_eq!(
548            MessageKind::QueueWaitTimeout.class(),
549            MessageClass::Streamed
550        );
551
552        // Control plane.
553        assert_eq!(MessageKind::Cancel.class(), MessageClass::ControlPlane);
554        assert_eq!(MessageKind::Compress.class(), MessageClass::ControlPlane);
555        assert_eq!(MessageKind::SetSession.class(), MessageClass::ControlPlane);
556        assert_eq!(MessageKind::Notice.class(), MessageClass::ControlPlane);
557
558        // Coverage check — every catalogued kind has a class.
559        for k in ALL_KINDS {
560            let _ = k.class();
561        }
562    }
563
564    #[test]
565    fn allowed_flags_matrix_is_pinned() {
566        // Handshake / lifecycle: MORE_FRAMES only — no COMPRESSED on
567        // tiny control-frame payloads. Flipping this requires updating
568        // the matrix below in lockstep.
569        let handshake = [
570            MessageKind::Hello,
571            MessageKind::HelloAck,
572            MessageKind::AuthRequest,
573            MessageKind::AuthResponse,
574            MessageKind::AuthOk,
575            MessageKind::AuthFail,
576            MessageKind::Bye,
577            MessageKind::Ping,
578            MessageKind::Pong,
579        ];
580        for k in handshake {
581            let f = k.allowed_flags();
582            assert!(
583                f.contains(Flags::MORE_FRAMES),
584                "{k:?} must allow MORE_FRAMES"
585            );
586            assert!(
587                !f.contains(Flags::COMPRESSED),
588                "{k:?} must NOT allow COMPRESSED today"
589            );
590        }
591
592        // Everything else: both documented flags allowed.
593        for k in ALL_KINDS {
594            if handshake.contains(k) {
595                continue;
596            }
597            let f = k.allowed_flags();
598            assert!(
599                f.contains(Flags::MORE_FRAMES),
600                "{k:?} must allow MORE_FRAMES"
601            );
602            assert!(f.contains(Flags::COMPRESSED), "{k:?} must allow COMPRESSED");
603        }
604    }
605
606    #[test]
607    fn every_kind_has_unique_byte_value() {
608        // The byte value is the wire spec — two kinds sharing a value
609        // would silently corrupt dispatch. The catalog must reject it.
610        let mut seen = std::collections::HashSet::new();
611        for k in ALL_KINDS {
612            let byte = *k as u8;
613            assert!(
614                seen.insert(byte),
615                "byte 0x{byte:02x} reused by {k:?}; catalog has a duplicate"
616            );
617        }
618    }
619
620    #[test]
621    fn from_u8_round_trips_for_every_kind() {
622        for k in ALL_KINDS {
623            let byte = *k as u8;
624            let decoded = MessageKind::from_u8(byte).unwrap_or_else(|| {
625                panic!("from_u8 returned None for catalog entry {k:?} (0x{byte:02x})")
626            });
627            assert_eq!(
628                decoded, *k,
629                "from_u8(0x{byte:02x}) must round-trip back to {k:?}"
630            );
631        }
632    }
633
634    #[test]
635    fn permits_flags_matches_allowed_flags() {
636        // Handshake kinds reject COMPRESSED, accept MORE_FRAMES.
637        assert!(MessageKind::Ping.permits_flags(Flags::MORE_FRAMES));
638        assert!(MessageKind::Ping.permits_flags(Flags::empty()));
639        assert!(!MessageKind::Ping.permits_flags(Flags::COMPRESSED));
640        assert!(!MessageKind::Ping.permits_flags(Flags::COMPRESSED | Flags::MORE_FRAMES));
641
642        // Streamed kinds accept both documented bits — the
643        // MORE_FRAMES invariant for in-flight stream envelopes is
644        // declared here through `allowed_flags`.
645        assert!(MessageKind::BulkStreamRows.permits_flags(Flags::MORE_FRAMES));
646        assert!(MessageKind::BulkStreamRows.permits_flags(Flags::COMPRESSED));
647        assert!(MessageKind::RowDescription.permits_flags(Flags::MORE_FRAMES));
648        assert!(MessageKind::StreamEnd.permits_flags(Flags::MORE_FRAMES));
649    }
650
651    #[test]
652    fn direction_matrix_is_pinned() {
653        // Client → Server.
654        for k in [
655            MessageKind::Hello,
656            MessageKind::AuthResponse,
657            MessageKind::Query,
658            MessageKind::QueryBinary,
659            MessageKind::BulkInsert,
660            MessageKind::BulkInsertBinary,
661            MessageKind::BulkInsertPrevalidated,
662            MessageKind::BulkStreamStart,
663            MessageKind::BulkStreamRows,
664            MessageKind::BulkStreamCommit,
665            MessageKind::Prepare,
666            MessageKind::ExecutePrepared,
667            MessageKind::Get,
668            MessageKind::Delete,
669            MessageKind::Cancel,
670            MessageKind::Compress,
671            MessageKind::SetSession,
672            MessageKind::VectorSearch,
673            MessageKind::GraphTraverse,
674            MessageKind::QueryWithParams,
675            MessageKind::OpenStream,
676            MessageKind::StreamCancel,
677            MessageKind::QueueWaitOpen,
678        ] {
679            assert_eq!(
680                k.direction(),
681                MessageDirection::ClientToServer,
682                "{k:?} should be client-originated"
683            );
684        }
685
686        // Server → Client.
687        for k in [
688            MessageKind::HelloAck,
689            MessageKind::AuthRequest,
690            MessageKind::AuthOk,
691            MessageKind::AuthFail,
692            MessageKind::Result,
693            MessageKind::Error,
694            MessageKind::BulkOk,
695            MessageKind::BulkStreamAck,
696            MessageKind::PreparedOk,
697            MessageKind::DeleteOk,
698            MessageKind::Notice,
699            MessageKind::RowDescription,
700            MessageKind::StreamEnd,
701            MessageKind::OpenAck,
702            MessageKind::StreamError,
703            MessageKind::QueueEventPush,
704            MessageKind::QueueWaitTimeout,
705        ] {
706            assert_eq!(
707                k.direction(),
708                MessageDirection::ServerToClient,
709                "{k:?} should be server-originated"
710            );
711        }
712
713        // Symmetric. `StreamChunk` (issue #764) is symmetric: the
714        // server emits it on output streams, the client emits it on
715        // input streams.
716        for k in [
717            MessageKind::Bye,
718            MessageKind::Ping,
719            MessageKind::Pong,
720            MessageKind::StreamChunk,
721        ] {
722            assert_eq!(
723                k.direction(),
724                MessageDirection::Both,
725                "{k:?} should be symmetric"
726            );
727        }
728    }
729}
reddb_wire/redwire/frame.rs

reddb_wire/redwire/
frame.rs
