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

reddb_wire/redwire/
frame.rs
