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/// One managed command stored on a mutation topic. `payload` is the canonical
40/// typed request encoded with the normal wire framing for `command_code`, so
41/// append and fold share one request schema instead of maintaining a second
42/// mutation representation.
43#[derive(Clone, Debug, Serialize, Deserialize)]
44pub struct MutationCommandEnvelope {
45    pub v: u32,
46    pub operation_id: u128,
47    pub timestamp_micros: u64,
48    pub command_code: u32,
49    /// Trusted stream identity carried through durable replay.
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pub scope: Option<crate::forward::ForwardedScope>,
52    #[serde(with = "crate::encoding::bin_bytes")]
53    pub payload: Vec<u8>,
54}
55
56impl MutationCommandEnvelope {
57    pub fn validate(&self) -> Result<(), &'static str> {
58        if self.v == 0 {
59            return Err("mutation command version must not be zero");
60        }
61        if self.operation_id == 0 {
62            return Err("mutation command operation id must not be zero");
63        }
64        Ok(())
65    }
66}
67
68#[cfg(all(test, feature = "cbor"))]
69mod tests {
70    use super::*;
71    use crate::codes::{AGDX_KV_SET_CODE, KV_OP_VERSION};
72    use crate::framing::{decode_named, encode_named};
73
74    #[test]
75    fn given_a_command_envelope_when_round_tripped_then_should_preserve_request_bytes() {
76        let envelope = MutationCommandEnvelope {
77            v: KV_OP_VERSION,
78            operation_id: 42,
79            timestamp_micros: 1_700_000_000_000_000,
80            command_code: AGDX_KV_SET_CODE,
81            scope: None,
82            payload: vec![0, 1, 2, 255],
83        };
84        let bytes = encode_named(&envelope).expect("encodes");
85        let back: MutationCommandEnvelope = decode_named(&bytes).expect("decodes");
86        back.validate().expect("the envelope is valid");
87        assert_eq!(back.v, KV_OP_VERSION);
88        assert_eq!(back.operation_id, 42);
89        assert_eq!(back.timestamp_micros, 1_700_000_000_000_000);
90        assert_eq!(back.command_code, AGDX_KV_SET_CODE);
91        assert_eq!(back.payload, vec![0, 1, 2, 255]);
92    }
93
94    #[test]
95    fn given_a_managed_request_when_round_tripped_then_should_preserve_operation_identity() {
96        let request = ManagedRequestEnvelope {
97            v: MANAGED_REQUEST_VERSION,
98            operation_id: u128::MAX,
99            payload: vec![4, 3, 2, 1],
100        };
101
102        let bytes = encode_named(&request).expect("managed request encodes");
103        let back: ManagedRequestEnvelope = decode_named(&bytes).expect("managed request decodes");
104
105        assert_eq!(back, request);
106        back.validate().expect("the request is valid");
107    }
108
109    #[test]
110    fn given_a_zero_operation_identity_when_validated_then_should_reject_it() {
111        let request = ManagedRequestEnvelope {
112            v: MANAGED_REQUEST_VERSION,
113            operation_id: 0,
114            payload: Vec::new(),
115        };
116        let command = MutationCommandEnvelope {
117            v: KV_OP_VERSION,
118            operation_id: 0,
119            timestamp_micros: 1,
120            command_code: AGDX_KV_SET_CODE,
121            scope: None,
122            payload: Vec::new(),
123        };
124
125        assert!(request.validate().is_err());
126        assert!(command.validate().is_err());
127    }
128}
laser_wire/mutation.rs

laser_wire/
mutation.rs
