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    #[serde(with = "crate::encoding::bin_bytes")]
50    pub payload: Vec<u8>,
51}
52
53impl MutationCommandEnvelope {
54    pub fn validate(&self) -> Result<(), &'static str> {
55        if self.v == 0 {
56            return Err("mutation command version must not be zero");
57        }
58        if self.operation_id == 0 {
59            return Err("mutation command operation id must not be zero");
60        }
61        Ok(())
62    }
63}
64
65#[cfg(all(test, feature = "cbor"))]
66mod tests {
67    use super::*;
68    use crate::codes::{AGDX_KV_SET_CODE, KV_OP_VERSION};
69    use crate::framing::{decode_named, encode_named};
70
71    #[test]
72    fn given_a_command_envelope_when_round_tripped_then_should_preserve_request_bytes() {
73        let envelope = MutationCommandEnvelope {
74            v: KV_OP_VERSION,
75            operation_id: 42,
76            timestamp_micros: 1_700_000_000_000_000,
77            command_code: AGDX_KV_SET_CODE,
78            payload: vec![0, 1, 2, 255],
79        };
80        let bytes = encode_named(&envelope).expect("encodes");
81        let back: MutationCommandEnvelope = decode_named(&bytes).expect("decodes");
82        back.validate().expect("the envelope is valid");
83        assert_eq!(back.v, KV_OP_VERSION);
84        assert_eq!(back.operation_id, 42);
85        assert_eq!(back.timestamp_micros, 1_700_000_000_000_000);
86        assert_eq!(back.command_code, AGDX_KV_SET_CODE);
87        assert_eq!(back.payload, vec![0, 1, 2, 255]);
88    }
89
90    #[test]
91    fn given_a_managed_request_when_round_tripped_then_should_preserve_operation_identity() {
92        let request = ManagedRequestEnvelope {
93            v: MANAGED_REQUEST_VERSION,
94            operation_id: u128::MAX,
95            payload: vec![4, 3, 2, 1],
96        };
97
98        let bytes = encode_named(&request).expect("managed request encodes");
99        let back: ManagedRequestEnvelope = decode_named(&bytes).expect("managed request decodes");
100
101        assert_eq!(back, request);
102        back.validate().expect("the request is valid");
103    }
104
105    #[test]
106    fn given_a_zero_operation_identity_when_validated_then_should_reject_it() {
107        let request = ManagedRequestEnvelope {
108            v: MANAGED_REQUEST_VERSION,
109            operation_id: 0,
110            payload: Vec::new(),
111        };
112        let command = MutationCommandEnvelope {
113            v: KV_OP_VERSION,
114            operation_id: 0,
115            timestamp_micros: 1,
116            command_code: AGDX_KV_SET_CODE,
117            payload: Vec::new(),
118        };
119
120        assert!(request.validate().is_err());
121        assert!(command.validate().is_err());
122    }
123}
laser_wire/mutation.rs

laser_wire/
mutation.rs
