nodedb_types/protocol/
frames.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Native protocol request and response frame types.
4
5use serde::{Deserialize, Serialize};
6
7use crate::value::Value;
8
9use super::auth::AuthResponse;
10use super::opcodes::ResponseStatus;
11use super::request_fields::RequestFields;
12
13// ─── Request Frame ──────────────────────────────────────────────────
14
15/// A request sent from client to server over the native protocol.
16///
17/// Serialized as MessagePack. The `op` field selects the handler,
18/// `seq` correlates request to response.
19#[derive(Debug, Clone, Serialize, Deserialize)]
20pub struct NativeRequest {
21    /// Operation code.
22    pub op: super::opcodes::OpCode,
23    /// Client-assigned sequence number for request/response correlation.
24    pub seq: u64,
25    /// Operation-specific fields (flattened into the same map).
26    #[serde(flatten)]
27    pub fields: RequestFields,
28}
29
30impl zerompk::ToMessagePack for NativeRequest {
31    fn write<W: zerompk::Write>(&self, writer: &mut W) -> zerompk::Result<()> {
32        writer.write_array_len(3)?;
33        self.op.write(writer)?;
34        writer.write_u64(self.seq)?;
35        self.fields.write(writer)
36    }
37}
38
39impl<'a> zerompk::FromMessagePack<'a> for NativeRequest {
40    fn read<R: zerompk::Read<'a>>(reader: &mut R) -> zerompk::Result<Self> {
41        let len = reader.read_array_len()?;
42        if len != 3 {
43            return Err(zerompk::Error::ArrayLengthMismatch {
44                expected: 3,
45                actual: len,
46            });
47        }
48        let op = super::opcodes::OpCode::read(reader)?;
49        let seq = reader.read_u64()?;
50        let fields = RequestFields::read(reader)?;
51        Ok(Self { op, seq, fields })
52    }
53}
54
55// ─── Response Frame ─────────────────────────────────────────────────
56
57/// A response sent from server to client over the native protocol.
58#[derive(
59    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
60)]
61#[msgpack(map)]
62pub struct NativeResponse {
63    /// Echoed from the request for correlation.
64    pub seq: u64,
65    /// Execution outcome.
66    pub status: ResponseStatus,
67    /// Column names (for query results).
68    #[serde(skip_serializing_if = "Option::is_none")]
69    pub columns: Option<Vec<String>>,
70    /// Row data (for query results). Each row is a Vec of Values.
71    #[serde(skip_serializing_if = "Option::is_none")]
72    pub rows: Option<Vec<Vec<Value>>>,
73    /// Number of rows affected (for writes).
74    #[serde(skip_serializing_if = "Option::is_none")]
75    pub rows_affected: Option<u64>,
76    /// The statement's command verb (`INSERT`, `UPDATE`, `DELETE`, `UPSERT`,
77    /// `MERGE`, `TRUNCATE`, ...). `None` when the statement produced no DML
78    /// outcome.
79    #[serde(default, skip_serializing_if = "Option::is_none")]
80    #[msgpack(default)]
81    pub command: Option<String>,
82    /// WAL LSN watermark at time of computation.
83    pub watermark_lsn: u64,
84    /// Error details (if status == Error).
85    #[serde(skip_serializing_if = "Option::is_none")]
86    pub error: Option<ErrorPayload>,
87    /// Auth response (if op == Auth and status == Ok).
88    #[serde(skip_serializing_if = "Option::is_none")]
89    pub auth: Option<AuthResponse>,
90    /// Advisory warnings (e.g. password expiry grace period, must_change_password).
91    /// Empty in the common case; `#[serde(default)]` and `#[msgpack(default)]`
92    /// keep this additive and backward-compatible with older clients.
93    #[serde(default, skip_serializing_if = "Vec::is_empty")]
94    #[msgpack(default)]
95    pub warnings: Vec<String>,
96}
97
98/// Error details in a response.
99///
100/// Encoded as a MessagePack map rather than an array so that fields can be
101/// added without invalidating the frames older peers already understand —
102/// the same reason [`NativeResponse`] is a map. An array representation has
103/// no field names, so a decoder cannot tell a missing optional field from a
104/// truncated frame and every addition is a breaking change.
105#[derive(
106    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
107)]
108#[msgpack(map)]
109pub struct ErrorPayload {
110    /// SQLSTATE-style error code (e.g., "42P01" for undefined table).
111    pub code: String,
112    /// Human-readable error message.
113    pub message: String,
114    /// Stable numeric NodeDB error code (`nodedb_types::ErrorCode`'s inner
115    /// value, e.g. 1000 for a constraint violation).
116    ///
117    /// SQLSTATE is many-to-one — a unique violation and a duplicate
118    /// idempotency key both render as `23505`, and everything unclassified
119    /// renders as `XX000` — so it cannot carry the classification a client
120    /// needs to reconstruct a typed error. This field is the authoritative
121    /// one for that. `0` means the sending server predates this field, and a
122    /// client must fall back to a generic error rather than guess.
123    /// `#[serde(default)]` + `#[msgpack(default)]` keep it additive and
124    /// backward-compatible with peers that never send it.
125    #[serde(default, skip_serializing_if = "is_zero")]
126    #[msgpack(default)]
127    pub ndb_code: u16,
128    /// The structured details the server held: the collection, gate, or
129    /// document the error names. `None` when the server had none to send.
130    #[serde(default, skip_serializing_if = "Option::is_none")]
131    #[msgpack(default)]
132    pub details: Option<crate::error::ErrorDetails>,
133    /// The typed error that caused this one, when the server held one: the
134    /// Data-Plane refusal behind a phase failure, say. `None` when there is
135    /// no cause to report.
136    #[serde(default, skip_serializing_if = "Option::is_none")]
137    #[msgpack(default)]
138    pub cause: Option<super::error_cause::ErrorCausePayload>,
139}
140
141/// `skip_serializing_if` predicate for [`ErrorPayload::ndb_code`]: zero is the
142/// "not carried" sentinel, so it is omitted rather than sent as a real code.
143fn is_zero(code: &u16) -> bool {
144    *code == 0
145}
146
147impl NativeResponse {
148    /// Create a successful response with no data.
149    pub fn ok(seq: u64) -> Self {
150        Self {
151            seq,
152            status: ResponseStatus::Ok,
153            columns: None,
154            rows: None,
155            rows_affected: None,
156            command: None,
157            watermark_lsn: 0,
158            error: None,
159            auth: None,
160            warnings: Vec::new(),
161        }
162    }
163
164    /// Create a successful response from a `QueryResult`.
165    pub fn from_query_result(seq: u64, qr: crate::result::QueryResult, lsn: u64) -> Self {
166        Self {
167            seq,
168            status: ResponseStatus::Ok,
169            columns: Some(qr.columns),
170            rows: Some(qr.rows),
171            rows_affected: Some(qr.rows_affected),
172            command: qr.command,
173            watermark_lsn: lsn,
174            error: None,
175            auth: None,
176            warnings: Vec::new(),
177        }
178    }
179
180    /// Create an error response that carries only a SQLSTATE.
181    ///
182    /// Use [`Self::error_with_code`] wherever the numeric NodeDB code is
183    /// known: a frame built here reaches the client with `ndb_code == 0` and
184    /// collapses to a generic internal error on the far side.
185    pub fn error(seq: u64, code: impl Into<String>, message: impl Into<String>) -> Self {
186        Self::error_with_code(seq, code, message, 0)
187    }
188
189    /// Create an error response carrying both the SQLSTATE and the stable
190    /// numeric NodeDB code, so the client can rebuild the typed error rather
191    /// than inferring one from a many-to-one SQLSTATE.
192    pub fn error_with_code(
193        seq: u64,
194        code: impl Into<String>,
195        message: impl Into<String>,
196        ndb_code: u16,
197    ) -> Self {
198        Self {
199            seq,
200            status: ResponseStatus::Error,
201            columns: None,
202            rows: None,
203            rows_affected: None,
204            command: None,
205            watermark_lsn: 0,
206            error: Some(ErrorPayload {
207                code: code.into(),
208                message: message.into(),
209                ndb_code,
210                details: None,
211                cause: None,
212            }),
213            auth: None,
214            warnings: Vec::new(),
215        }
216    }
217
218    /// Attach the structured error details to an error response. A response
219    /// with no error payload is returned unchanged.
220    pub fn with_error_details(mut self, details: crate::error::ErrorDetails) -> Self {
221        if let Some(payload) = self.error.as_mut() {
222            payload.details = Some(details);
223        }
224        self
225    }
226
227    /// Attach the typed cause to an error response. A response with no error
228    /// payload is returned unchanged.
229    pub fn with_error_cause(mut self, cause: super::error_cause::ErrorCausePayload) -> Self {
230        if let Some(payload) = self.error.as_mut() {
231            payload.cause = Some(cause);
232        }
233        self
234    }
235
236    /// Create an auth success response.
237    pub fn auth_ok(seq: u64, username: String, tenant_id: u64) -> Self {
238        Self {
239            seq,
240            status: ResponseStatus::Ok,
241            columns: None,
242            rows: None,
243            rows_affected: None,
244            command: None,
245            watermark_lsn: 0,
246            error: None,
247            auth: Some(AuthResponse {
248                username,
249                tenant_id,
250            }),
251            warnings: Vec::new(),
252        }
253    }
254
255    /// Create a response with a single "status" column and one row.
256    pub fn status_row(seq: u64, message: impl Into<String>) -> Self {
257        Self {
258            seq,
259            status: ResponseStatus::Ok,
260            columns: Some(vec!["status".into()]),
261            rows: Some(vec![vec![Value::String(message.into())]]),
262            rows_affected: Some(1),
263            command: None,
264            watermark_lsn: 0,
265            error: None,
266            auth: None,
267            warnings: Vec::new(),
268        }
269    }
270}
271
272#[cfg(test)]
273mod tests {
274    use super::super::opcodes::OpCode;
275    use super::super::text_fields::TextFields;
276    use super::*;
277
278    #[test]
279    fn native_response_ok() {
280        let r = NativeResponse::ok(42);
281        assert_eq!(r.seq, 42);
282        assert_eq!(r.status, ResponseStatus::Ok);
283        assert!(r.error.is_none());
284    }
285
286    #[test]
287    fn native_response_error() {
288        let r = NativeResponse::error(1, "42P01", "collection not found");
289        assert_eq!(r.status, ResponseStatus::Error);
290        let e = r.error.unwrap();
291        assert_eq!(e.code, "42P01");
292        assert_eq!(e.message, "collection not found");
293    }
294
295    #[test]
296    fn native_response_from_query_result() {
297        let qr = crate::result::QueryResult {
298            columns: vec!["id".into(), "name".into()],
299            rows: vec![vec![
300                Value::String("u1".into()),
301                Value::String("Alice".into()),
302            ]],
303            rows_affected: 0,
304            command: None,
305        };
306        let r = NativeResponse::from_query_result(5, qr, 100);
307        assert_eq!(r.seq, 5);
308        assert_eq!(r.watermark_lsn, 100);
309        assert_eq!(r.columns.as_ref().unwrap().len(), 2);
310        assert_eq!(r.rows.as_ref().unwrap().len(), 1);
311    }
312
313    #[test]
314    fn native_response_status_row() {
315        let r = NativeResponse::status_row(3, "OK");
316        assert_eq!(r.columns.as_ref().unwrap(), &["status"]);
317        assert_eq!(r.rows.as_ref().unwrap()[0][0].as_str(), Some("OK"));
318    }
319
320    #[test]
321    fn msgpack_roundtrip_request() {
322        let req = NativeRequest {
323            op: OpCode::Sql,
324            seq: 1,
325            fields: RequestFields::Text(TextFields {
326                sql: Some("SELECT 1".into()),
327                ..Default::default()
328            }),
329        };
330        let bytes = zerompk::to_msgpack_vec(&req).unwrap();
331        let decoded: NativeRequest = zerompk::from_msgpack(&bytes).unwrap();
332        assert_eq!(decoded.op, OpCode::Sql);
333        assert_eq!(decoded.seq, 1);
334    }
335
336    #[test]
337    fn msgpack_roundtrip_response() {
338        let resp = NativeResponse::from_query_result(
339            7,
340            crate::result::QueryResult {
341                columns: vec!["x".into()],
342                rows: vec![vec![Value::Integer(42)]],
343                rows_affected: 0,
344                command: None,
345            },
346            99,
347        );
348        let bytes = zerompk::to_msgpack_vec(&resp).unwrap();
349        let decoded: NativeResponse = zerompk::from_msgpack(&bytes).unwrap();
350        assert_eq!(decoded.seq, 7);
351        assert_eq!(decoded.watermark_lsn, 99);
352        assert_eq!(decoded.rows.unwrap()[0][0].as_i64(), Some(42));
353    }
354
355    #[test]
356    fn error_payload_round_trips_the_numeric_code() {
357        let frame = NativeResponse::error_with_code(7, "23505", "duplicate key", 1000);
358        let bytes = zerompk::to_msgpack_vec(&frame).expect("encode");
359        let decoded: NativeResponse = zerompk::from_msgpack(&bytes).expect("decode");
360        let payload = decoded.error.expect("error payload survives the wire");
361        assert_eq!(payload.code, "23505");
362        assert_eq!(payload.ndb_code, 1000);
363    }
364
365    #[test]
366    fn error_payload_round_trips_the_details() {
367        let details = crate::error::ErrorDetails::Overflow {
368            collection: "counters".into(),
369        };
370        let frame = NativeResponse::error_with_code(7, "22003", "overflow on counters", 1021)
371            .with_error_details(details.clone());
372        let bytes = zerompk::to_msgpack_vec(&frame).expect("encode");
373        let decoded: NativeResponse = zerompk::from_msgpack(&bytes).expect("decode");
374        let payload = decoded.error.expect("error payload survives the wire");
375        assert_eq!(payload.details, Some(details));
376    }
377
378    #[test]
379    fn error_payload_round_trips_the_cause() {
380        let cause = super::super::error_cause::ErrorCausePayload::from(
381            &crate::error::NodeDbError::division_by_zero(),
382        );
383        let frame = NativeResponse::error_with_code(7, "XX000", "snapshot failed", 1602)
384            .with_error_cause(cause.clone());
385        let bytes = zerompk::to_msgpack_vec(&frame).expect("encode");
386        let decoded: NativeResponse = zerompk::from_msgpack(&bytes).expect("decode");
387        let payload = decoded.error.expect("error payload survives the wire");
388        assert_eq!(payload.ndb_code, 1602);
389        assert_eq!(payload.cause, Some(cause));
390    }
391
392    #[test]
393    fn error_payload_without_numeric_code_still_decodes() {
394        // Hand-rolled 2-key map: exactly what a peer built before the numeric
395        // code was added sends. It must decode with `ndb_code == 0` rather
396        // than fail, which is what makes the field additive.
397        let mut legacy = vec![0x82u8];
398        for (key, value) in [("code", "42P01"), ("message", "boom")] {
399            legacy.push(0xA0 | key.len() as u8);
400            legacy.extend_from_slice(key.as_bytes());
401            legacy.push(0xA0 | value.len() as u8);
402            legacy.extend_from_slice(value.as_bytes());
403        }
404        let decoded: ErrorPayload = zerompk::from_msgpack(&legacy).expect("legacy frame decodes");
405        assert_eq!(decoded.code, "42P01");
406        assert_eq!(decoded.message, "boom");
407        assert_eq!(
408            decoded.ndb_code, 0,
409            "an absent numeric code must read as the 'not carried' sentinel"
410        );
411    }
412}
nodedb_types/protocol/frames.rs

nodedb_types/protocol/
frames.rs
