ssp2/
decode.rs

1//! SSP2 envelope decoding (SPEC.md §1.2), request/response frame grammar
2//! (§1.5, §1.6), and the closed decode-error boundary of §1.7: exactly the
3//! listed violations are rejected here, nothing more.
4
5use crate::error::{DecodeError, Result};
6use crate::model::frame_type as ft;
7use crate::model::{
8    Change, Frame, MediaType, Message, MsgKind, Op, OpResult, Operation, PushStatus, SubStatus,
9};
10use crate::primitives::Reader;
11use crate::segment::decode_rows_segment;
12
13pub const SSP2_MAGIC: &[u8; 4] = b"SSP2";
14pub const WIRE_VERSION: u16 = 1;
15
16/// Decode a complete SSP2 message, enforcing envelope rules and the per-kind
17/// frame grammar. Unknown frame types are preserved byte-for-byte (§1.2
18/// rule 2).
19pub fn decode_message(bytes: &[u8]) -> Result<Message> {
20    let mut r = Reader::new(bytes);
21    let magic = r.take(4, "envelope magic")?;
22    if magic != SSP2_MAGIC {
23        return Err(DecodeError::invalid("bad envelope magic (expected SSP2)"));
24    }
25    let wire_version = r.u16("wireVersion")?;
26    if wire_version != WIRE_VERSION {
27        return Err(DecodeError::invalid(format!(
28            "unsupported wireVersion {wire_version}"
29        )));
30    }
31    let msg_kind = match r.u8("msgKind")? {
32        0x01 => MsgKind::Request,
33        0x02 => MsgKind::Response,
34        other => {
35            return Err(DecodeError::invalid(format!(
36                "unknown msgKind 0x{other:02x}"
37            )))
38        }
39    };
40    let flags = r.u8("flags")?;
41    if flags != 0 {
42        return Err(DecodeError::invalid(format!(
43            "envelope flags must be 0x00, got 0x{flags:02x}"
44        )));
45    }
46
47    let frames = match msg_kind {
48        MsgKind::Request => decode_request_frames(&mut r)?,
49        MsgKind::Response => decode_response_frames(&mut r)?,
50    };
51    if !r.is_empty() {
52        return Err(DecodeError::invalid(
53            "trailing bytes after the END frame (canonical encoding)",
54        ));
55    }
56    Ok(Message { msg_kind, frames })
57}
58
59/// Read one frame header + payload. Returns `None` for END.
60fn next_frame<'a>(r: &mut Reader<'a>) -> Result<Option<(u8, &'a [u8])>> {
61    if r.is_empty() {
62        return Err(DecodeError::invalid(
63            "truncated message: body ends without an END frame",
64        ));
65    }
66    let ty = r.u8("frameType")?;
67    let len = r.u32("frameLength")? as usize;
68    if len > r.remaining() {
69        return Err(DecodeError::invalid(format!(
70            "frame 0x{ty:02x} length {len} exceeds the {} remaining bytes",
71            r.remaining()
72        )));
73    }
74    let payload = r.take(len, "frame payload")?;
75    if ty == ft::END {
76        if len != 0 {
77            return Err(DecodeError::invalid("END frame must have frameLength 0"));
78        }
79        return Ok(None);
80    }
81    Ok(Some((ty, payload)))
82}
83
84/// "Unknown" per §1.2 rule 2: a frameType with no layout in this wire version
85/// for *either* message kind. A type registered for the other kind is a
86/// decode error, not unknown.
87fn classify(ty: u8, kind: MsgKind) -> Result<bool> {
88    let (own, other) = match kind {
89        MsgKind::Request => (ft::REQUEST_TYPES, ft::RESPONSE_TYPES),
90        MsgKind::Response => (ft::RESPONSE_TYPES, ft::REQUEST_TYPES),
91    };
92    if own.contains(&ty) {
93        Ok(true)
94    } else if other.contains(&ty) {
95        Err(DecodeError::invalid(format!(
96            "frame type 0x{ty:02x} is registered for the other message kind"
97        )))
98    } else {
99        Ok(false)
100    }
101}
102
103/// §1.5 grammar: REQ_HEADER, PUSH_COMMIT × N, PULL_HEADER 0|1,
104/// SUBSCRIPTION × M (only if PULL_HEADER present), END. A request with
105/// neither push nor pull frames is invalid.
106fn decode_request_frames(r: &mut Reader<'_>) -> Result<Vec<Frame>> {
107    let mut frames = Vec::new();
108    let mut seen_header = false;
109    let mut seen_pull = false;
110    let mut seen_push = false;
111
112    while let Some((ty, payload)) = next_frame(r)? {
113        let known = classify(ty, MsgKind::Request)?;
114        if !known {
115            if !seen_header {
116                return Err(DecodeError::invalid(
117                    "unknown frame before REQ_HEADER (unknown frames are only legal between the header frame and END)",
118                ));
119            }
120            frames.push(Frame::Unknown {
121                frame_type: ty,
122                payload: payload.to_vec(),
123            });
124            continue;
125        }
126        if !seen_header {
127            if ty != ft::REQ_HEADER {
128                return Err(DecodeError::invalid(
129                    "first frame of a request must be REQ_HEADER",
130                ));
131            }
132            frames.push(decode_req_header(payload)?);
133            seen_header = true;
134            continue;
135        }
136        match ty {
137            ft::REQ_HEADER => {
138                return Err(DecodeError::invalid("duplicate REQ_HEADER frame"));
139            }
140            ft::PUSH_COMMIT => {
141                if seen_pull {
142                    return Err(DecodeError::invalid(
143                        "PUSH_COMMIT after PULL_HEADER (out of grammar order)",
144                    ));
145                }
146                frames.push(decode_push_commit(payload)?);
147                seen_push = true;
148            }
149            ft::PULL_HEADER => {
150                if seen_pull {
151                    return Err(DecodeError::invalid("duplicate PULL_HEADER frame"));
152                }
153                frames.push(decode_pull_header(payload)?);
154                seen_pull = true;
155            }
156            ft::SUBSCRIPTION => {
157                if !seen_pull {
158                    return Err(DecodeError::invalid(
159                        "SUBSCRIPTION frame without a preceding PULL_HEADER",
160                    ));
161                }
162                frames.push(decode_subscription(payload)?);
163            }
164            _ => unreachable!("classified as known request type"),
165        }
166    }
167    if !seen_header {
168        return Err(DecodeError::invalid("request has no REQ_HEADER frame"));
169    }
170    if !seen_push && !seen_pull {
171        return Err(DecodeError::invalid(
172            "request has neither PUSH_COMMIT nor PULL_HEADER frames",
173        ));
174    }
175    Ok(frames)
176}
177
178/// §1.6 grammar: RESP_HEADER, PUSH_RESULT × N, then per subscription
179/// SUB_START (COMMIT × k | segment frames × m) SUB_END; ERROR 0|1 anywhere
180/// after RESP_HEADER, and if present the next frame MUST be END (no frame of
181/// any type, unknown included, may follow it).
182fn decode_response_frames(r: &mut Reader<'_>) -> Result<Vec<Frame>> {
183    let mut frames = Vec::new();
184    let mut seen_header = false;
185    let mut seen_lease = false; // §7.3.2: at most one LEASE, before the body
186    let mut seen_body = false; // any non-RESP_HEADER/non-LEASE frame
187    let mut past_results = false; // a SUB_START has been seen
188    let mut sub_open = false;
189    let mut sub_seen_commit = false;
190    let mut sub_seen_segment = false;
191    let mut error_seen = false;
192
193    while let Some((ty, payload)) = next_frame(r)? {
194        if error_seen {
195            return Err(DecodeError::invalid(
196                "frame after ERROR: the next frame after ERROR must be END",
197            ));
198        }
199        let known = classify(ty, MsgKind::Response)?;
200        if !known {
201            if !seen_header {
202                return Err(DecodeError::invalid(
203                    "unknown frame before RESP_HEADER (unknown frames are only legal between the header frame and END)",
204                ));
205            }
206            frames.push(Frame::Unknown {
207                frame_type: ty,
208                payload: payload.to_vec(),
209            });
210            seen_body = true;
211            continue;
212        }
213        if !seen_header {
214            if ty != ft::RESP_HEADER {
215                return Err(DecodeError::invalid(
216                    "first frame of a response must be RESP_HEADER",
217                ));
218            }
219            frames.push(decode_resp_header(payload)?);
220            seen_header = true;
221            continue;
222        }
223        match ty {
224            ft::RESP_HEADER => {
225                return Err(DecodeError::invalid("duplicate RESP_HEADER frame"));
226            }
227            ft::LEASE => {
228                // §7.3.2: at most one LEASE, immediately after RESP_HEADER.
229                if seen_lease {
230                    return Err(DecodeError::invalid("duplicate LEASE frame"));
231                }
232                if seen_body {
233                    return Err(DecodeError::invalid(
234                        "LEASE frame must immediately follow RESP_HEADER",
235                    ));
236                }
237                frames.push(decode_lease(payload)?);
238                seen_lease = true;
239            }
240            ft::PUSH_RESULT => {
241                if past_results {
242                    return Err(DecodeError::invalid(
243                        "PUSH_RESULT after a SUB_START (out of grammar order)",
244                    ));
245                }
246                frames.push(decode_push_result(payload)?);
247                seen_body = true;
248            }
249            ft::SUB_START => {
250                if sub_open {
251                    return Err(DecodeError::invalid(
252                        "SUB_START while a subscription context is already open",
253                    ));
254                }
255                frames.push(decode_sub_start(payload)?);
256                past_results = true;
257                seen_body = true;
258                sub_open = true;
259                sub_seen_commit = false;
260                sub_seen_segment = false;
261            }
262            ft::COMMIT => {
263                if !sub_open {
264                    return Err(DecodeError::invalid(
265                        "COMMIT frame outside an open subscription context",
266                    ));
267                }
268                if sub_seen_segment {
269                    return Err(DecodeError::invalid(
270                        "COMMIT and segment frames must not both appear for one subscription",
271                    ));
272                }
273                frames.push(decode_commit(payload)?);
274                sub_seen_commit = true;
275                seen_body = true;
276            }
277            ft::SEGMENT_REF | ft::SEGMENT_INLINE => {
278                if !sub_open {
279                    return Err(DecodeError::invalid(
280                        "segment frame outside an open subscription context",
281                    ));
282                }
283                if sub_seen_commit {
284                    return Err(DecodeError::invalid(
285                        "COMMIT and segment frames must not both appear for one subscription",
286                    ));
287                }
288                if ty == ft::SEGMENT_REF {
289                    frames.push(decode_segment_ref(payload)?);
290                } else {
291                    // §5.7: the payload must be a structurally valid rows
292                    // segment; the raw bytes are preserved for re-encoding.
293                    decode_rows_segment(payload)?;
294                    frames.push(Frame::SegmentInline {
295                        payload: payload.to_vec(),
296                    });
297                }
298                sub_seen_segment = true;
299                seen_body = true;
300            }
301            ft::SUB_END => {
302                if !sub_open {
303                    return Err(DecodeError::invalid(
304                        "SUB_END without an open subscription context",
305                    ));
306                }
307                frames.push(decode_sub_end(payload)?);
308                sub_open = false;
309                seen_body = true;
310            }
311            ft::ERROR => {
312                frames.push(decode_error_frame(payload)?);
313                error_seen = true;
314                seen_body = true;
315            }
316            _ => unreachable!("classified as known response type"),
317        }
318    }
319    if !seen_header {
320        return Err(DecodeError::invalid("response has no RESP_HEADER frame"));
321    }
322    if sub_open && !error_seen {
323        return Err(DecodeError::invalid(
324            "message ended with an open subscription context (missing SUB_END)",
325        ));
326    }
327    Ok(frames)
328}
329
330/// Run `parse` over a frame payload; trailing bytes inside a known frame are
331/// a decode error (§1.2 rule 3).
332fn frame_payload<T>(
333    payload: &[u8],
334    name: &str,
335    parse: impl FnOnce(&mut Reader<'_>) -> Result<T>,
336) -> Result<T> {
337    let mut r = Reader::new(payload);
338    let value = parse(&mut r)?;
339    if !r.is_empty() {
340        return Err(DecodeError::invalid(format!(
341            "trailing bytes inside {name} frame ({} bytes unconsumed)",
342            r.remaining()
343        )));
344    }
345    Ok(value)
346}
347
348fn decode_req_header(payload: &[u8]) -> Result<Frame> {
349    frame_payload(payload, "REQ_HEADER", |r| {
350        let client_id = r.str("clientId")?;
351        if client_id.is_empty() {
352            return Err(DecodeError::invalid("clientId must be non-empty"));
353        }
354        let schema_version = r.i32("schemaVersion")?;
355        if schema_version < 1 {
356            return Err(DecodeError::invalid(format!(
357                "schemaVersion must be ≥ 1, got {schema_version}"
358            )));
359        }
360        Ok(Frame::ReqHeader {
361            client_id,
362            schema_version,
363        })
364    })
365}
366
367fn decode_op(r: &mut Reader<'_>) -> Result<Op> {
368    match r.u8("op")? {
369        1 => Ok(Op::Upsert),
370        2 => Ok(Op::Delete),
371        other => Err(DecodeError::invalid(format!("unknown op byte {other}"))),
372    }
373}
374
375fn decode_push_commit(payload: &[u8]) -> Result<Frame> {
376    frame_payload(payload, "PUSH_COMMIT", |r| {
377        let client_commit_id = r.str("clientCommitId")?;
378        if client_commit_id.is_empty() {
379            return Err(DecodeError::invalid("clientCommitId must be non-empty"));
380        }
381        let count = r.u32("operations count")? as usize;
382        if count == 0 {
383            return Err(DecodeError::empty_commit(
384                "PUSH_COMMIT with zero operations",
385            ));
386        }
387        let mut operations = Vec::with_capacity(count.min(4096));
388        for _ in 0..count {
389            let table = r.str("operation table")?;
390            let row_id = r.str("operation rowId")?;
391            let op = decode_op(r)?;
392            let base_version = if r.presence("baseVersion")? {
393                Some(r.i64("baseVersion")?)
394            } else {
395                None
396            };
397            let payload_bytes = if r.presence("payload")? {
398                Some(r.bytes("payload")?)
399            } else {
400                None
401            };
402            // §6.1 presence tie: payload present iff upsert.
403            match op {
404                Op::Upsert if payload_bytes.is_none() => {
405                    return Err(DecodeError::invalid(
406                        "upsert operation without a payload (presence invariant)",
407                    ));
408                }
409                Op::Delete if payload_bytes.is_some() => {
410                    return Err(DecodeError::invalid(
411                        "delete operation with a payload (presence invariant)",
412                    ));
413                }
414                _ => {}
415            }
416            operations.push(Operation {
417                table,
418                row_id,
419                op,
420                base_version,
421                payload: payload_bytes,
422            });
423        }
424        Ok(Frame::PushCommit {
425            client_commit_id,
426            operations,
427        })
428    })
429}
430
431fn decode_pull_header(payload: &[u8]) -> Result<Frame> {
432    frame_payload(payload, "PULL_HEADER", |r| {
433        let limit_commits = r.i32("limitCommits")?;
434        let limit_snapshot_rows = r.i32("limitSnapshotRows")?;
435        let max_snapshot_pages = r.i32("maxSnapshotPages")?;
436        let accept = r.u8("accept")?;
437        if accept & 0xF0 != 0 {
438            return Err(DecodeError::invalid(format!(
439                "accept bits 4–7 must be zero, got 0b{accept:08b}"
440            )));
441        }
442        Ok(Frame::PullHeader {
443            limit_commits,
444            limit_snapshot_rows,
445            max_snapshot_pages,
446            accept,
447        })
448    })
449}
450
451fn decode_subscription(payload: &[u8]) -> Result<Frame> {
452    frame_payload(payload, "SUBSCRIPTION", |r| {
453        let id = r.str("id")?;
454        let table = r.str("table")?;
455        let scopes = r.scope_map("scopes")?;
456        let params = if r.presence("params")? {
457            Some(r.json("params")?)
458        } else {
459            None
460        };
461        let cursor = r.i64("cursor")?;
462        let bootstrap_state = if r.presence("bootstrapState")? {
463            Some(r.json("bootstrapState")?)
464        } else {
465            None
466        };
467        Ok(Frame::Subscription {
468            id,
469            table,
470            scopes,
471            params,
472            cursor,
473            bootstrap_state,
474        })
475    })
476}
477
478fn decode_resp_header(payload: &[u8]) -> Result<Frame> {
479    frame_payload(payload, "RESP_HEADER", |r| {
480        let required_schema_version = if r.presence("requiredSchemaVersion")? {
481            Some(r.i32("requiredSchemaVersion")?)
482        } else {
483            None
484        };
485        let latest_schema_version = if r.presence("latestSchemaVersion")? {
486            Some(r.i32("latestSchemaVersion")?)
487        } else {
488            None
489        };
490        Ok(Frame::RespHeader {
491            required_schema_version,
492            latest_schema_version,
493        })
494    })
495}
496
497fn decode_lease(payload: &[u8]) -> Result<Frame> {
498    frame_payload(payload, "LEASE", |r| {
499        let lease_id = r.str("leaseId")?;
500        if lease_id.is_empty() {
501            return Err(DecodeError::invalid("LEASE.leaseId must be non-empty"));
502        }
503        let expires_at_ms = r.i64("expiresAtMs")?;
504        Ok(Frame::Lease {
505            lease_id,
506            expires_at_ms,
507        })
508    })
509}
510
511fn decode_push_result(payload: &[u8]) -> Result<Frame> {
512    frame_payload(payload, "PUSH_RESULT", |r| {
513        let client_commit_id = r.str("clientCommitId")?;
514        let status = match r.u8("status")? {
515            1 => PushStatus::Applied,
516            2 => PushStatus::Cached,
517            3 => PushStatus::Rejected,
518            other => {
519                return Err(DecodeError::invalid(format!(
520                    "unknown PUSH_RESULT status byte {other}"
521                )))
522            }
523        };
524        let commit_seq = if r.presence("commitSeq")? {
525            Some(r.i64("commitSeq")?)
526        } else {
527            None
528        };
529        // §6.3 presence tie: commitSeq present for applied/cached, absent for
530        // rejected.
531        match status {
532            PushStatus::Rejected if commit_seq.is_some() => {
533                return Err(DecodeError::invalid(
534                    "commitSeq present on a rejected PUSH_RESULT (presence invariant)",
535                ));
536            }
537            PushStatus::Applied | PushStatus::Cached if commit_seq.is_none() => {
538                return Err(DecodeError::invalid(
539                    "commitSeq absent on an applied/cached PUSH_RESULT (presence invariant)",
540                ));
541            }
542            _ => {}
543        }
544        let count = r.u32("results count")? as usize;
545        let mut results = Vec::with_capacity(count.min(4096));
546        for _ in 0..count {
547            let op_index = r.i32("opIndex")?;
548            results.push(match r.u8("result status")? {
549                1 => OpResult::Applied { op_index },
550                2 => OpResult::Conflict {
551                    op_index,
552                    code: r.str("conflict code")?,
553                    message: r.str("conflict message")?,
554                    server_version: r.i64("serverVersion")?,
555                    server_row: r.bytes("serverRow")?,
556                },
557                3 => OpResult::Error {
558                    op_index,
559                    code: r.str("error code")?,
560                    message: r.str("error message")?,
561                    retryable: r.bool("retryable")?,
562                },
563                other => {
564                    return Err(DecodeError::invalid(format!(
565                        "unknown result record status byte {other}"
566                    )))
567                }
568            });
569        }
570        Ok(Frame::PushResult {
571            client_commit_id,
572            status,
573            commit_seq,
574            results,
575        })
576    })
577}
578
579fn decode_sub_start(payload: &[u8]) -> Result<Frame> {
580    frame_payload(payload, "SUB_START", |r| {
581        let id = r.str("id")?;
582        let status = match r.u8("status")? {
583            1 => SubStatus::Active,
584            2 => SubStatus::Revoked,
585            3 => SubStatus::Reset,
586            other => {
587                return Err(DecodeError::invalid(format!(
588                    "unknown SUB_START status byte {other}"
589                )))
590            }
591        };
592        let reason_code = r.str("reasonCode")?;
593        let effective_scopes = r.scope_map("effectiveScopes")?;
594        let bootstrap = r.bool("bootstrap")?;
595        Ok(Frame::SubStart {
596            id,
597            status,
598            reason_code,
599            effective_scopes,
600            bootstrap,
601        })
602    })
603}
604
605fn decode_commit(payload: &[u8]) -> Result<Frame> {
606    frame_payload(payload, "COMMIT", |r| {
607        let commit_seq = r.i64("commitSeq")?;
608        let created_at_ms = r.i64("createdAtMs")?;
609        let actor_id = r.str("actorId")?;
610        let table_count = r.u32("tables count")? as usize;
611        let mut tables = Vec::with_capacity(table_count.min(4096));
612        for _ in 0..table_count {
613            tables.push(r.str("table")?);
614        }
615        let change_count = r.u32("changes count")? as usize;
616        let mut changes = Vec::with_capacity(change_count.min(4096));
617        for _ in 0..change_count {
618            let table_index = r.u16("tableIndex")?;
619            if table_index as usize >= tables.len() {
620                return Err(DecodeError::invalid(format!(
621                    "change tableIndex {table_index} out of range (tables: {})",
622                    tables.len()
623                )));
624            }
625            let row_id = r.str("rowId")?;
626            let op = decode_op(r)?;
627            let row_version = if r.presence("rowVersion")? {
628                Some(r.i64("rowVersion")?)
629            } else {
630                None
631            };
632            let scopes = r.str_map("change scopes")?;
633            let row = if r.presence("row")? {
634                Some(r.bytes("row")?)
635            } else {
636                None
637            };
638            // §4.5 presence ties: rowVersion and row present iff upsert.
639            let ok = match op {
640                Op::Upsert => row_version.is_some() && row.is_some(),
641                Op::Delete => row_version.is_none() && row.is_none(),
642            };
643            if !ok {
644                return Err(DecodeError::invalid(
645                    "change rowVersion/row presence does not match op (presence invariant)",
646                ));
647            }
648            changes.push(Change {
649                table_index,
650                row_id,
651                op,
652                row_version,
653                scopes,
654                row,
655            });
656        }
657        Ok(Frame::Commit {
658            commit_seq,
659            created_at_ms,
660            actor_id,
661            tables,
662            changes,
663        })
664    })
665}
666
667fn decode_segment_ref(payload: &[u8]) -> Result<Frame> {
668    frame_payload(payload, "SEGMENT_REF", |r| {
669        let segment_id = r.str("segmentId")?;
670        let media_type = match r.u8("mediaType")? {
671            1 => MediaType::Rows,
672            2 => MediaType::Sqlite,
673            other => {
674                return Err(DecodeError::invalid(format!(
675                    "unknown mediaType byte {other}"
676                )))
677            }
678        };
679        let table = r.str("table")?;
680        let byte_length = r.i64("byteLength")?;
681        let row_count = r.i64("rowCount")?;
682        let as_of_commit_seq = r.i64("asOfCommitSeq")?;
683        let scope_digest = r.str("scopeDigest")?;
684        let row_cursor = if r.presence("rowCursor")? {
685            Some(r.str("rowCursor")?)
686        } else {
687            None
688        };
689        let next_row_cursor = if r.presence("nextRowCursor")? {
690            Some(r.str("nextRowCursor")?)
691        } else {
692            None
693        };
694        let url = if r.presence("url")? {
695            Some(r.str("url")?)
696        } else {
697            None
698        };
699        let url_expires_at_ms = if r.presence("urlExpiresAtMs")? {
700            Some(r.i64("urlExpiresAtMs")?)
701        } else {
702            None
703        };
704        // §5.4 presence tie: urlExpiresAtMs present iff url is.
705        if url.is_some() != url_expires_at_ms.is_some() {
706            return Err(DecodeError::invalid(
707                "url and urlExpiresAtMs must be present together (presence invariant)",
708            ));
709        }
710        Ok(Frame::SegmentRef {
711            segment_id,
712            media_type,
713            table,
714            byte_length,
715            row_count,
716            as_of_commit_seq,
717            scope_digest,
718            row_cursor,
719            next_row_cursor,
720            url,
721            url_expires_at_ms,
722        })
723    })
724}
725
726fn decode_sub_end(payload: &[u8]) -> Result<Frame> {
727    frame_payload(payload, "SUB_END", |r| {
728        let next_cursor = r.i64("nextCursor")?;
729        let bootstrap_state = if r.presence("bootstrapState")? {
730            Some(r.json("bootstrapState")?)
731        } else {
732            None
733        };
734        Ok(Frame::SubEnd {
735            next_cursor,
736            bootstrap_state,
737        })
738    })
739}
740
741fn decode_error_frame(payload: &[u8]) -> Result<Frame> {
742    frame_payload(payload, "ERROR", |r| {
743        let code = r.str("code")?;
744        let message = r.str("message")?;
745        let category = r.str("category")?;
746        let retryable = r.bool("retryable")?;
747        let recommended_action = r.str("recommendedAction")?;
748        let details = if r.presence("details")? {
749            Some(r.json("details")?)
750        } else {
751            None
752        };
753        Ok(Frame::Error {
754            code,
755            message,
756            category,
757            retryable,
758            recommended_action,
759            details,
760        })
761    })
762}
ssp2/decode.rs

ssp2/
decode.rs
