laser_wire/
mutation.rs

1use serde::{Deserialize, Serialize};
2
3pub const MANAGED_REQUEST_VERSION: u32 = 1;
4
5/// Position of one applied mutation on the KV mutation topic: the fold
6/// coordinate a barriered read waits for. A lease grant or renewal returns the
7/// position at which the answering fold applied it, and a
8/// [`crate::kv::KvGet`] carrying it as `min_position` must not be answered by
9/// a fold that has not applied that position yet. Positions compare only
10/// within one `topic_generation`: a generation mismatch is fail-closed
11/// ([`crate::kv::KvError::Stale`]), never an implicit reset.
12#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
13pub struct MutationPosition {
14    pub topic_generation: u64,
15    pub partition: u32,
16    pub offset: u64,
17}
18
19#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
20pub struct ManagedRequestEnvelope {
21    pub v: u32,
22    pub operation_id: u128,
23    #[serde(with = "crate::encoding::bin_bytes")]
24    pub payload: Vec<u8>,
25}
26
27impl ManagedRequestEnvelope {
28    pub fn validate(&self) -> Result<(), &'static str> {
29        if self.v != MANAGED_REQUEST_VERSION {
30            return Err("unsupported managed request version");
31        }
32        if self.operation_id == 0 {
33            return Err("managed request operation id must not be zero");
34        }
35        Ok(())
36    }
37}
38
39/// A managed surface with its own mutation log and fold.
40#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
41#[serde(rename_all = "snake_case")]
42pub enum PositionSurface {
43    Kv,
44    Fork,
45    Graph,
46}
47
48impl PositionSurface {
49    /// The surface a managed read or mutation acts on, or `None` for a
50    /// command outside the three mutation logs.
51    pub const fn of_code(code: u32) -> Option<Self> {
52        use crate::codes::{
53            AGDX_FORK_CREATE_CODE, AGDX_FORK_DELETE_CODE, AGDX_FORK_LIST_CODE,
54            AGDX_FORK_PROMOTE_CODE, AGDX_FORK_PUT_CODE, AGDX_GRAPH_NEIGHBORS_CODE,
55            AGDX_GRAPH_QUERY_CODE, AGDX_GRAPH_UPSERT_CODE, AGDX_KV_CAS_CODE,
56            AGDX_KV_CAS_FENCED_CODE, AGDX_KV_COPY_CODE, AGDX_KV_DELETE_CODE,
57            AGDX_KV_DELETE_MANY_CODE, AGDX_KV_EXISTS_CODE, AGDX_KV_EXPIRE_CODE, AGDX_KV_GET_CODE,
58            AGDX_KV_LEASE_CODE, AGDX_KV_LEASE_RENEW_CODE, AGDX_KV_MOVE_CODE,
59            AGDX_KV_NAMESPACES_CODE, AGDX_KV_PATCH_CODE, AGDX_KV_RELEASE_CODE, AGDX_KV_SCAN_CODE,
60            AGDX_KV_SET_CODE,
61        };
62        match code {
63            AGDX_KV_GET_CODE
64            | AGDX_KV_SCAN_CODE
65            | AGDX_KV_NAMESPACES_CODE
66            | AGDX_KV_EXISTS_CODE
67            | AGDX_KV_SET_CODE
68            | AGDX_KV_CAS_CODE
69            | AGDX_KV_DELETE_CODE
70            | AGDX_KV_DELETE_MANY_CODE
71            | AGDX_KV_EXPIRE_CODE
72            | AGDX_KV_PATCH_CODE
73            | AGDX_KV_COPY_CODE
74            | AGDX_KV_MOVE_CODE
75            | AGDX_KV_CAS_FENCED_CODE
76            | AGDX_KV_LEASE_CODE
77            | AGDX_KV_LEASE_RENEW_CODE
78            | AGDX_KV_RELEASE_CODE => Some(Self::Kv),
79            AGDX_FORK_LIST_CODE
80            | AGDX_FORK_CREATE_CODE
81            | AGDX_FORK_DELETE_CODE
82            | AGDX_FORK_PROMOTE_CODE
83            | AGDX_FORK_PUT_CODE => Some(Self::Fork),
84            AGDX_GRAPH_QUERY_CODE | AGDX_GRAPH_NEIGHBORS_CODE | AGDX_GRAPH_UPSERT_CODE => {
85                Some(Self::Graph)
86            }
87            _ => None,
88        }
89    }
90}
91
92/// A managed read sent with a consistency requirement. Unknown fields are
93/// refused, so no plain read request is ever taken for this frame.
94///
95/// With `min_position`, the answering node serves the read only once its fold
96/// for the read's surface has applied that position. A client sends the
97/// highest position it has seen for the surface, so a read never misses its
98/// own earlier writes or goes back behind an earlier read, whichever node
99/// answers. With `log_head`, the streaming server forwards the read with a
100/// read barrier and the node first waits until its fold has applied every
101/// record the surface's log held when the read arrived. No Laser SDK sets it:
102/// the clients carry `min_position` instead, which covers their own earlier
103/// writes without waiting for the whole log. A node that does not get there
104/// within its wait bound answers `Stale`, never older state.
105#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
106#[serde(deny_unknown_fields)]
107pub struct ManagedReadEnvelope {
108    pub v: u32,
109    /// The read request the client would otherwise send on its own.
110    #[serde(with = "crate::encoding::bin_bytes")]
111    pub read: Vec<u8>,
112    #[serde(default, skip_serializing_if = "Option::is_none")]
113    pub min_position: Option<MutationPosition>,
114    #[serde(default, skip_serializing_if = "is_false")]
115    pub log_head: bool,
116    /// For a session read: the session fold must have applied each of these
117    /// positions that still names a live, registered source partition of the
118    /// read's stream. Positions of other streams are ignored.
119    #[serde(default, skip_serializing_if = "Vec::is_empty")]
120    pub min_sessions: Vec<SessionPosition>,
121}
122
123/// How far a node's session fold has applied one partition of a session
124/// source. Positions compare only within one stream and topic generation.
125#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
126#[serde(deny_unknown_fields)]
127pub struct SessionPosition {
128    pub stream_id: u32,
129    /// Creation time of the stream incarnation, in microseconds.
130    pub stream_generation: u64,
131    pub topic: String,
132    /// Creation time of the topic incarnation, in microseconds.
133    pub topic_generation: u64,
134    pub partition_id: u32,
135    /// The last offset the fold applied.
136    pub offset: u64,
137}
138
139/// The positions a node's folds had applied when it answered, per surface.
140/// `None` for a surface whose fold has applied nothing yet. `sessions` covers
141/// the stream of a session read and is empty for every other reply.
142#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
143#[serde(deny_unknown_fields)]
144pub struct AppliedFrontier {
145    #[serde(default, skip_serializing_if = "Option::is_none")]
146    pub kv: Option<MutationPosition>,
147    #[serde(default, skip_serializing_if = "Option::is_none")]
148    pub fork: Option<MutationPosition>,
149    #[serde(default, skip_serializing_if = "Option::is_none")]
150    pub graph: Option<MutationPosition>,
151    #[serde(default, skip_serializing_if = "Vec::is_empty")]
152    pub sessions: Vec<SessionPosition>,
153}
154
155/// The reply to a managed request sent by an SDK client: the surface reply,
156/// plus the frontier the answering node had applied once it computed that
157/// reply. After a mutation the frontier covers the mutation itself, and after
158/// a read it covers every write the read could see. The client keeps the
159/// highest position per surface and sends it back as
160/// [`ManagedReadEnvelope::min_position`]. Unknown fields are refused, so no
161/// surface reply is ever taken for this frame.
162#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
163#[serde(deny_unknown_fields)]
164pub struct ManagedReplyEnvelope {
165    pub v: u32,
166    #[serde(default)]
167    pub frontier: AppliedFrontier,
168    #[serde(with = "crate::encoding::bin_bytes")]
169    pub reply: Vec<u8>,
170}
171
172#[expect(clippy::trivially_copy_pass_by_ref, reason = "serde skip predicate")]
173fn is_false(value: &bool) -> bool {
174    !*value
175}
176
177/// One managed command stored on a mutation topic. `payload` is the canonical
178/// typed request encoded with the normal wire framing for `command_code`, so
179/// append and fold share one request schema instead of maintaining a second
180/// mutation representation.
181#[derive(Clone, Debug, Serialize, Deserialize)]
182pub struct MutationCommandEnvelope {
183    pub v: u32,
184    pub operation_id: u128,
185    pub deployment_id: u64,
186    pub timestamp_micros: u64,
187    pub command_code: u32,
188    /// Trusted stream identity carried through durable replay.
189    #[serde(default, skip_serializing_if = "Option::is_none")]
190    pub scope: Option<crate::forward::ForwardedScope>,
191    #[serde(with = "crate::encoding::bin_bytes")]
192    pub payload: Vec<u8>,
193    /// The client that asked for the command, stamped before the command
194    /// enters the log and the only actor it names. Absent on a command the
195    /// backend issues on its own.
196    #[serde(default, skip_serializing_if = "Option::is_none")]
197    pub caller: Option<crate::forward::Caller>,
198}
199
200impl MutationCommandEnvelope {
201    pub fn validate(&self) -> Result<(), &'static str> {
202        if self.v == 0 {
203            return Err("mutation command version must not be zero");
204        }
205        if self.operation_id == 0 {
206            return Err("mutation command operation id must not be zero");
207        }
208        Ok(())
209    }
210}
211
212#[cfg(all(test, feature = "cbor"))]
213mod tests {
214    use super::*;
215    use crate::codes::{AGDX_KV_SET_CODE, KV_OP_VERSION};
216    use crate::framing::{decode_named, encode_named};
217
218    #[test]
219    fn given_a_read_envelope_when_round_tripped_then_should_keep_the_inner_read() {
220        let get = crate::kv::KvGet {
221            v: KV_OP_VERSION,
222            namespace: "orders".to_owned(),
223            key: b"k".to_vec(),
224            if_none_match: None,
225            min_position: None,
226        };
227        let read = encode_named(&get).expect("encodes");
228        let envelope = ManagedReadEnvelope {
229            v: MANAGED_REQUEST_VERSION,
230            read: read.clone(),
231            min_position: Some(MutationPosition {
232                topic_generation: 3,
233                partition: 0,
234                offset: 41,
235            }),
236            log_head: true,
237            min_sessions: vec![SessionPosition {
238                stream_id: 2,
239                stream_generation: 10,
240                topic: "agent.sessions".to_owned(),
241                topic_generation: 11,
242                partition_id: 1,
243                offset: 7,
244            }],
245        };
246        let decoded: ManagedReadEnvelope =
247            decode_named(&encode_named(&envelope).expect("encodes")).expect("decodes");
248        assert_eq!(decoded, envelope);
249        // A plain read never decodes as the barrier frame, so an unwrapped
250        // request is never mistaken for a barriered one.
251        assert!(decode_named::<ManagedReadEnvelope>(&read).is_err());
252        let managed = encode_named(&ManagedRequestEnvelope {
253            v: MANAGED_REQUEST_VERSION,
254            operation_id: 9,
255            payload: read,
256        })
257        .expect("encodes");
258        assert!(decode_named::<ManagedReadEnvelope>(&managed).is_err());
259    }
260
261    #[test]
262    fn given_a_reply_envelope_when_round_tripped_then_should_keep_the_reply_and_refuse_plain_replies()
263     {
264        let reply =
265            encode_named(&crate::kv::KvReply::Ok(crate::kv::KvOutcome::Written)).expect("encodes");
266        let envelope = ManagedReplyEnvelope {
267            v: MANAGED_REQUEST_VERSION,
268            frontier: AppliedFrontier {
269                kv: Some(MutationPosition {
270                    topic_generation: 1,
271                    partition: 0,
272                    offset: 7,
273                }),
274                fork: None,
275                graph: None,
276                sessions: Vec::new(),
277            },
278            reply: reply.clone(),
279        };
280        let decoded: ManagedReplyEnvelope =
281            decode_named(&encode_named(&envelope).expect("encodes")).expect("decodes");
282        assert_eq!(decoded, envelope);
283        assert!(decode_named::<ManagedReplyEnvelope>(&reply).is_err());
284        let refusal = encode_named(&crate::result::CommandError::new(
285            crate::result::ResultCode::Stale,
286            "behind",
287        ))
288        .expect("encodes");
289        assert!(decode_named::<ManagedReplyEnvelope>(&refusal).is_err());
290    }
291
292    #[test]
293    fn given_command_codes_when_classified_then_should_name_their_surface() {
294        use crate::codes::{
295            AGDX_FORK_LIST_CODE, AGDX_GRAPH_UPSERT_CODE, AGDX_KV_LEASE_CODE, AGDX_QUERY_CODE,
296            AGDX_SESSION_GET_CODE,
297        };
298        assert_eq!(
299            PositionSurface::of_code(AGDX_KV_LEASE_CODE),
300            Some(PositionSurface::Kv)
301        );
302        assert_eq!(
303            PositionSurface::of_code(AGDX_FORK_LIST_CODE),
304            Some(PositionSurface::Fork)
305        );
306        assert_eq!(
307            PositionSurface::of_code(AGDX_GRAPH_UPSERT_CODE),
308            Some(PositionSurface::Graph)
309        );
310        assert_eq!(PositionSurface::of_code(AGDX_SESSION_GET_CODE), None);
311        assert_eq!(PositionSurface::of_code(AGDX_QUERY_CODE), None);
312    }
313
314    #[test]
315    fn given_a_command_envelope_when_round_tripped_then_should_preserve_request_bytes() {
316        let envelope = MutationCommandEnvelope {
317            v: KV_OP_VERSION,
318            operation_id: 42,
319            deployment_id: 3,
320            timestamp_micros: 1_700_000_000_000_000,
321            command_code: AGDX_KV_SET_CODE,
322            scope: None,
323            payload: vec![0, 1, 2, 255],
324            caller: Some(crate::forward::Caller {
325                user_id: 7,
326                agent: None,
327                on_behalf_of: None,
328            }),
329        };
330        let bytes = encode_named(&envelope).expect("encodes");
331        let back: MutationCommandEnvelope = decode_named(&bytes).expect("decodes");
332        back.validate().expect("the envelope is valid");
333        assert_eq!(back.v, KV_OP_VERSION);
334        assert_eq!(back.operation_id, 42);
335        assert_eq!(back.caller.map(|caller| caller.user_id), Some(7));
336        assert_eq!(back.deployment_id, 3);
337        assert_eq!(back.timestamp_micros, 1_700_000_000_000_000);
338        assert_eq!(back.command_code, AGDX_KV_SET_CODE);
339        assert_eq!(back.payload, vec![0, 1, 2, 255]);
340    }
341
342    #[test]
343    fn given_a_managed_request_when_round_tripped_then_should_preserve_operation_identity() {
344        let request = ManagedRequestEnvelope {
345            v: MANAGED_REQUEST_VERSION,
346            operation_id: u128::MAX,
347            payload: vec![4, 3, 2, 1],
348        };
349
350        let bytes = encode_named(&request).expect("managed request encodes");
351        let back: ManagedRequestEnvelope = decode_named(&bytes).expect("managed request decodes");
352
353        assert_eq!(back, request);
354        back.validate().expect("the request is valid");
355    }
356
357    #[test]
358    fn given_a_zero_operation_identity_when_validated_then_should_reject_it() {
359        let request = ManagedRequestEnvelope {
360            v: MANAGED_REQUEST_VERSION,
361            operation_id: 0,
362            payload: Vec::new(),
363        };
364        let command = MutationCommandEnvelope {
365            v: KV_OP_VERSION,
366            operation_id: 0,
367            timestamp_micros: 1,
368            deployment_id: 3,
369            command_code: AGDX_KV_SET_CODE,
370            scope: None,
371            payload: Vec::new(),
372            caller: None,
373        };
374
375        assert!(request.validate().is_err());
376        assert!(command.validate().is_err());
377    }
378}
laser_wire/mutation.rs

laser_wire/
mutation.rs
