laser_wire/forward.rs
1// Frames the managed backend consumes on behalf of a client. The SDK never
2// sends these directly. They live here so the encoding has one definition and
3// the golden corpus pins them like every other frame.
4
5use serde::{Deserialize, Serialize};
6
7/// Trusted stream identity stamped by the server before forwarding a session request.
8#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
9pub struct ForwardedScope {
10 pub stream_id: u32,
11 pub stream: String,
12 pub stream_created_at_micros: u64,
13}
14
15/// A forwarded managed query. Carries the authenticated identity the SDK cannot
16/// set itself, plus the opaque request the SDK sent. CBOR-encoded, named fields.
17#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
18pub struct ForwardedQuery {
19 /// Authenticated identity. Audit today, per-user stream scoping later.
20 /// Trusted: set by the server, not the client.
21 pub user_id: u32,
22 /// Originating client id, for logging and audit only.
23 pub client_id: u128,
24 /// `gen_ai.conversation.id` echoed for the audit log, never used to route.
25 pub correlation: Option<String>,
26 /// Opaque CBOR `QueryEnvelope` from the SDK, not decoded in transit.
27 #[serde(with = "crate::encoding::bin_bytes")]
28 pub query_envelope: Vec<u8>,
29 /// The caller's effective capability grants, stamped by the server. Empty
30 /// (the default, skipped on the wire) when authorization is not in effect.
31 #[serde(default, skip_serializing_if = "Vec::is_empty")]
32 pub grants: Vec<crate::authz::Grant>,
33 /// The tenancy of the server's replicated authorization state. The
34 /// backend refuses a frame whose tenancy differs from its own.
35 #[serde(default, skip_serializing_if = "crate::authz::Tenancy::is_deployment")]
36 pub tenancy: crate::authz::Tenancy,
37 /// The stream of the queried index, stamped once the caller's native read
38 /// authority over it was checked. The backend refuses a query of a
39 /// stream-scoped index that arrives without this scope or with another.
40 #[serde(default, skip_serializing_if = "Option::is_none")]
41 pub scope: Option<ForwardedScope>,
42}
43
44/// A forwarded keyed managed command (registry browse, key-value, forks). Unlike
45/// `ForwardedQuery`, several op kinds share one path, so the frame carries
46/// `command_code` and the backend dispatches on it. `payload` is the opaque
47/// CBOR request the SDK sent.
48#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
49pub struct ForwardedCommand {
50 /// Authenticated identity, set by the server. The key-value store scopes its
51 /// rows by this `user_id`, and the SDK cannot set it.
52 pub user_id: u32,
53 /// Originating client id, for logging and audit only.
54 pub client_id: u128,
55 /// `gen_ai.conversation.id` echoed for the audit log, never used to route.
56 pub correlation: Option<String>,
57 /// Stable identity minted by the originating SDK for a logical managed
58 /// mutation. Read-only commands leave it absent.
59 #[serde(default, skip_serializing_if = "Option::is_none")]
60 pub operation_id: Option<u128>,
61 /// Set by the server, like `user_id`: true when the caller holds a global
62 /// read permission. Widens the read-side ops (key-value scan/namespaces,
63 /// fork list) to every user's rows, while writes stay scoped to `user_id`
64 /// regardless.
65 #[serde(default)]
66 pub read_all: bool,
67 /// Server-stamped stream identity. A session command without it fails closed.
68 #[serde(default, skip_serializing_if = "Option::is_none")]
69 pub scope: Option<ForwardedScope>,
70 /// The managed command code (browse, key-value, or fork block). The backend
71 /// dispatches on it.
72 pub command_code: u32,
73 /// Opaque CBOR request the SDK sent, not decoded in transit.
74 #[serde(with = "crate::encoding::bin_bytes")]
75 pub payload: Vec<u8>,
76 /// The caller's effective capability grants, stamped by the server. Empty
77 /// (the default, skipped on the wire) when authorization is not in effect.
78 #[serde(default, skip_serializing_if = "Vec::is_empty")]
79 pub grants: Vec<crate::authz::Grant>,
80 /// Set by the server when the client wrapped this read in a
81 /// [`crate::mutation::ManagedReadEnvelope`]. The backend answers only once
82 /// its fold has reached the surface's mutation log head.
83 #[serde(default, skip_serializing_if = "is_false")]
84 pub read_barrier: bool,
85 /// Set by the server from [`crate::mutation::ManagedReadEnvelope`]: the
86 /// backend answers only once its fold for the read's surface has applied
87 /// this position.
88 #[serde(default, skip_serializing_if = "Option::is_none")]
89 pub min_position: Option<crate::mutation::MutationPosition>,
90 /// Set by the server for a managed request an SDK client sent over the
91 /// binary protocol. The backend answers with a
92 /// [`crate::mutation::ManagedReplyEnvelope`] carrying its applied
93 /// frontier.
94 #[serde(default, skip_serializing_if = "is_false")]
95 pub positioned_reply: bool,
96 /// Set by the server from [`crate::mutation::ManagedReadEnvelope`]: the
97 /// session positions a session read must find applied.
98 #[serde(default, skip_serializing_if = "Vec::is_empty")]
99 pub min_sessions: Vec<crate::mutation::SessionPosition>,
100 /// The tenancy of the server's replicated authorization state. The
101 /// backend refuses a frame whose tenancy differs from its own.
102 #[serde(default, skip_serializing_if = "crate::authz::Tenancy::is_deployment")]
103 pub tenancy: crate::authz::Tenancy,
104}
105
106#[expect(clippy::trivially_copy_pass_by_ref, reason = "serde skip predicate")]
107fn is_false(value: &bool) -> bool {
108 !*value
109}
110
111#[cfg(all(test, feature = "cbor"))]
112mod tests {
113 use super::*;
114 use crate::framing::{decode_named, encode_named};
115
116 // A contiguous byte-string run inside the encoded frame proves the opaque
117 // field rode as a CBOR byte string (major type 2), not an array of ints.
118 // The values are all >= 0x18, which an array would have to widen to two
119 // bytes each, so the byte string is also strictly smaller.
120 fn contains_byte_string(frame: &[u8], len: u8, fill: u8) -> bool {
121 let head = 0x40 | len; // byte string, length in the low 5 bits
122 let mut want = vec![head];
123 want.extend(std::iter::repeat_n(fill, len as usize));
124 frame.windows(want.len()).any(|w| w == want.as_slice())
125 }
126
127 #[test]
128 fn given_forwarded_query_when_encoded_then_envelope_rides_as_a_byte_string() {
129 let frame = encode_named(&ForwardedQuery {
130 user_id: 7,
131 client_id: 42,
132 correlation: Some("conv-1".to_owned()),
133 query_envelope: vec![0x90; 5],
134 grants: Vec::new(),
135 tenancy: Default::default(),
136 scope: None,
137 })
138 .expect("encodes");
139 assert!(
140 contains_byte_string(&frame, 5, 0x90),
141 "query_envelope must encode as a CBOR byte string, got {frame:02x?}"
142 );
143 }
144
145 #[test]
146 fn given_forwarded_command_when_encoded_then_payload_rides_as_a_byte_string() {
147 let frame = encode_named(&ForwardedCommand {
148 read_barrier: false,
149 min_position: None,
150 positioned_reply: false,
151 min_sessions: Vec::new(),
152 tenancy: Default::default(),
153 user_id: 7,
154 client_id: 42,
155 correlation: None,
156 operation_id: Some(17),
157 read_all: true,
158 scope: None,
159 command_code: 1_000_000,
160 payload: vec![0x90; 5],
161 grants: Vec::new(),
162 })
163 .expect("encodes");
164 assert!(
165 contains_byte_string(&frame, 5, 0x90),
166 "payload must encode as a CBOR byte string, got {frame:02x?}"
167 );
168 }
169
170 #[test]
171 fn given_forwarded_frames_when_round_tripped_then_should_preserve_fields() {
172 let query = ForwardedQuery {
173 user_id: 1,
174 client_id: u128::MAX,
175 correlation: Some("c".to_owned()),
176 query_envelope: vec![0xff, 0x00, 0x18, 0x7f],
177 grants: Vec::new(),
178 tenancy: Default::default(),
179 scope: Some(ForwardedScope {
180 stream_id: 3,
181 stream: "acme".to_owned(),
182 stream_created_at_micros: 9,
183 }),
184 };
185 let back: ForwardedQuery =
186 decode_named(&encode_named(&query).expect("encodes")).expect("decodes");
187 assert_eq!(back, query);
188
189 let command = ForwardedCommand {
190 read_barrier: false,
191 min_position: None,
192 positioned_reply: false,
193 min_sessions: Vec::new(),
194 tenancy: Default::default(),
195 user_id: 2,
196 client_id: 9,
197 correlation: None,
198 operation_id: Some(9),
199 read_all: false,
200 scope: None,
201 command_code: 42,
202 payload: vec![0xde, 0xad, 0xbe, 0xef],
203 grants: Vec::new(),
204 };
205 let back: ForwardedCommand =
206 decode_named(&encode_named(&command).expect("encodes")).expect("decodes");
207 assert_eq!(back, command);
208 }
209}