harmos_sidecar/
wire.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::ops::{Deref, Range};
18use std::sync::Arc;
19
20// Third Party Imports
21use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt};
22// Local Imports
23use harmos_domain::Message;
24use harmos_sdk::operations::{CHILD_METHOD, CallValue, OperationDeclaration, WIRE_METHOD, WireAnswer, WireChildCall, WireCommand};
25
26/* === Constants === */
27
28/* === Macros === */
29
30/* === Type Aliases === */
31
32/* === Structs === */
33
34#[derive(Clone, Default)]
35pub struct FramePayload {
36    bytes: Arc<Vec<u8>>,
37    range: Range<usize>,
38}
39
40#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
41#[serde(deny_unknown_fields)]
42pub struct PluginCatalog {
43    pub registrations: Vec<crate::Registration>,
44    pub observers: Vec<crate::inventory::observers::Declaration>,
45    pub operations: Vec<crate::operations::OperationDefinition>,
46}
47
48#[derive(Clone, Debug, PartialEq, Eq)]
49pub struct Publication {
50    pub stream: String,
51    pub schema: u32,
52    pub sequence: u64,
53    pub published_unix_micros: Option<i64>,
54    pub context: Option<String>,
55    pub payload: Vec<u8>,
56}
57
58/* === Enums === */
59
60#[derive(Clone, Debug, PartialEq, Eq)]
61pub enum Frame {
62    Call { id: u64, method: String, params: FramePayload },
63    Cancel { id: u64 },
64    Push { method: String, params: FramePayload },
65    Answer { id: u64, result: FramePayload },
66    Refusal { id: u64, refusal: harmos_domain::Refusal },
67}
68
69/* === Traits === */
70
71/* === Implementations === */
72
73impl FramePayload {
74    pub fn as_slice(&self) -> &[u8] {
75        &self.bytes[self.range.clone()]
76    }
77
78    /// Compacts a unique backing allocation into its visible bytes.
79    /// Shared payloads copy their visible bytes into an independent allocation.
80    pub fn into_vec(self) -> Vec<u8> {
81        let Self { bytes, range } = self;
82        let mut bytes = match Arc::try_unwrap(bytes) {
83            Ok(bytes) => bytes,
84            Err(bytes) => return bytes[range].to_vec(),
85        };
86        let length = range.len();
87        if range.start != 0 {
88            bytes.copy_within(range, 0);
89        }
90        bytes.truncate(length);
91        bytes
92    }
93
94    pub(crate) fn from_range(bytes: Vec<u8>, range: Range<usize>) -> Self {
95        assert!(range.start <= range.end && range.end <= bytes.len());
96        Self { bytes: Arc::new(bytes), range }
97    }
98}
99
100impl From<Vec<u8>> for FramePayload {
101    fn from(bytes: Vec<u8>) -> Self {
102        let range = 0..bytes.len();
103        Self::from_range(bytes, range)
104    }
105}
106
107impl Deref for FramePayload {
108    type Target = [u8];
109    fn deref(&self) -> &[u8] {
110        self.as_slice()
111    }
112}
113
114impl AsRef<[u8]> for FramePayload {
115    fn as_ref(&self) -> &[u8] {
116        self.as_slice()
117    }
118}
119
120impl<T: AsRef<[u8]> + ?Sized> PartialEq<T> for FramePayload {
121    fn eq(&self, other: &T) -> bool {
122        self.as_slice() == other.as_ref()
123    }
124}
125
126impl Eq for FramePayload {}
127
128impl harmos_domain::MemorySize for FramePayload {
129    fn memory_size(&self) -> usize {
130        std::mem::size_of::<Self>()
131            .saturating_add(std::mem::size_of::<Vec<u8>>())
132            .saturating_add(2 * std::mem::size_of::<usize>())
133            .saturating_add(self.bytes.capacity())
134    }
135}
136
137impl std::fmt::Debug for FramePayload {
138    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
139        self.as_slice().fmt(formatter)
140    }
141}
142
143impl Frame {
144    pub const REGISTER: &'static str = "harmos.register";
145    pub const EXITING: &'static str = "harmos.exiting";
146    pub const LIMIT: usize = 16 * 1024 * 1024;
147    pub const OVERSIZED: i64 = -32_603;
148
149    /// Bounds a routed Prepare or child call with no portable context values.
150    /// Uses maximal identifiers and includes a generation context for the child.
151    /// Frame lengths exclude the four-byte physical packet prefix.
152    pub fn request_encoded_len<O: OperationDeclaration>(request: &O::Request) -> usize {
153        Self::request_payload_encoded_len(O::ID, harmos_domain::Message::encoded_size(request))
154    }
155
156    /// Bounds a routed Prepare or empty-values child carrying this encoded payload.
157    /// Use child_payload_encoded_len when the child carries portable values.
158    pub fn request_payload_encoded_len(operation: &str, payload: usize) -> usize {
159        if operation.len() > Self::LIMIT {
160            return usize::MAX;
161        }
162        let prepare = WireCommand::prepare_encoded_len(u64::MAX, operation, [255; 32], u64::MAX, payload).unwrap_or(usize::MAX);
163        call_encoded_len(u64::MAX, WIRE_METHOD, prepare).max(Self::child_payload_encoded_len(operation, u64::MAX, payload, true, &[]))
164    }
165
166    /// Bounds a child call using its actual portable values and generation presence.
167    /// Only the unknown outer identifier is bounded by its maximal encoded width.
168    pub fn child_payload_encoded_len(
169        operation: &str, generation: u64, payload: usize, generation_context: bool, values: &[CallValue],
170    ) -> usize {
171        let child = WireChildCall::payload_encoded_len(operation, generation, payload, generation_context, values).unwrap_or(usize::MAX);
172        call_encoded_len(u64::MAX, CHILD_METHOD, child)
173    }
174
175    pub(crate) fn child_encoded_from_empty(empty_envelope: usize, payload: usize) -> usize {
176        let growth = binary_bytes_len(payload).saturating_sub(1);
177        call_encoded_len(u64::MAX, CHILD_METHOD, empty_envelope.saturating_add(growth))
178    }
179
180    /// Tests the routed Prepare or empty-values child bound; not an arbitrary child context.
181    pub fn request_fits<O: OperationDeclaration>(request: &O::Request) -> bool {
182        Self::request_encoded_len::<O>(request) <= Self::LIMIT
183    }
184
185    pub fn reply_encoded_len<O: OperationDeclaration>(reply: &O::Response) -> usize {
186        let payload = WireAnswer::ready_encoded_len(reply.encoded_size()).unwrap_or(usize::MAX);
187        answer_encoded_len(u64::MAX, payload)
188    }
189
190    pub fn reply_fits<O: OperationDeclaration>(reply: &O::Response) -> bool {
191        Self::reply_encoded_len::<O>(reply) <= Self::LIMIT
192    }
193
194    pub fn encoded_len(&self) -> usize {
195        match self {
196            Self::Call { id, method, params } => call_encoded_len(*id, method, params.len()),
197            Self::Answer { id, result } => answer_encoded_len(*id, result.len()),
198            Self::Refusal { id, refusal } => refusal_encoded_len(*id, refusal),
199            Self::Cancel { id } => 2 + varint_len(*id) + varint_len(u64::MAX),
200            Self::Push { method, params } => {
201                2usize.saturating_add(binary_bytes_len(method.len())).saturating_add(binary_bytes_len(params.len()))
202            }
203        }
204    }
205
206    pub fn fits(&self) -> bool {
207        self.encoded_len() <= Self::LIMIT
208    }
209
210    #[cfg(feature = "kit")]
211    pub(crate) fn exiting(reason: String) -> Self {
212        Self::Push {
213            method: Self::EXITING.to_owned(),
214            params: reason.encode().into(),
215        }
216    }
217
218    pub fn exiting_reason(params: &[u8]) -> Option<String> {
219        String::decode(params).ok()
220    }
221
222    pub async fn diagnostic<R>(input: &mut R) -> std::io::Result<Option<String>>
223    where
224        R: AsyncBufRead + Unpin + Send,
225    {
226        loop {
227            let mut bytes = Vec::new();
228            let read = input.take(Self::LIMIT as u64 + 1).read_until(b'\n', &mut bytes).await?;
229            if read == 0 {
230                return Ok(None);
231            }
232            if bytes.len() > Self::LIMIT {
233                let mut rest = Vec::new();
234                input.read_until(b'\n', &mut rest).await?;
235                continue;
236            }
237
238            let text = String::from_utf8_lossy(&bytes);
239            return Ok(Some(text.trim_end_matches(['\n', '\r']).to_owned()));
240        }
241    }
242}
243
244/* === Public Functions === */
245
246// Postcard Length Of A Publication Without Its Following Rows Field.
247#[cfg(feature = "kit")]
248pub(crate) fn publication_encoded_len(
249    stream: &str, schema: u32, sequence: u64, payload: usize, published_unix_micros: Option<i64>, context: Option<&str>,
250) -> usize {
251    binary_bytes_len(stream.len())
252        .saturating_add(varint_len(u64::from(schema)))
253        .saturating_add(1 + varint_len(sequence))
254        .saturating_add(1 + published_unix_micros.map_or(0, zigzag_len))
255        .saturating_add(1 + context.map_or(0, |value| binary_bytes_len(value.len())))
256        .saturating_add(binary_bytes_len(payload))
257}
258
259// Postcard Length Of One Following Row.
260#[cfg(feature = "kit")]
261pub(crate) fn following_encoded_len(payload: usize, published_delta_micros: Option<i64>) -> usize {
262    binary_bytes_len(payload).saturating_add(1 + published_delta_micros.map_or(0, zigzag_len))
263}
264
265pub(crate) fn varint_len(value: u64) -> usize {
266    ((64 - value.leading_zeros()).max(1) as usize).div_ceil(7)
267}
268
269/* === Private Functions === */
270
271fn call_encoded_len(id: u64, method: &str, payload_len: usize) -> usize {
272    2usize.saturating_add(varint_len(id)).saturating_add(binary_bytes_len(method.len())).saturating_add(binary_bytes_len(payload_len))
273}
274
275fn answer_encoded_len(id: u64, payload_len: usize) -> usize {
276    2usize.saturating_add(varint_len(id)).saturating_add(binary_bytes_len(payload_len))
277}
278
279fn refusal_encoded_len(id: u64, refusal: &harmos_domain::Refusal) -> usize {
280    let refusal = postcard::experimental::serialized_size(refusal).unwrap_or(usize::MAX);
281    2usize.saturating_add(varint_len(id)).saturating_add(refusal)
282}
283
284#[cfg(feature = "kit")]
285fn zigzag_len(value: i64) -> usize {
286    varint_len(((value as u64) << 1) ^ ((value >> 63) as u64))
287}
288
289fn binary_bytes_len(length: usize) -> usize {
290    varint_len(length as u64).saturating_add(length)
291}
292
293/* === Tests === */
294
295#[cfg(test)]
296mod tests {
297    use super::*;
298    use harmos_domain::MemorySize;
299
300    #[test]
301    fn unique_payload_compacts() {
302        let mut bytes = Vec::with_capacity(128);
303        bytes.extend(0..32_u8);
304        let pointer = bytes.as_ptr();
305        let capacity = bytes.capacity();
306
307        let payload = FramePayload::from_range(bytes, 8..24);
308        let expected = std::mem::size_of::<FramePayload>() + std::mem::size_of::<Vec<u8>>() + 2 * std::mem::size_of::<usize>() + capacity;
309        assert_eq!(payload.memory_size(), expected);
310
311        let owned = payload.into_vec();
312        assert_eq!(owned.as_ptr(), pointer);
313        assert_eq!(owned.capacity(), capacity);
314        assert_eq!(owned, (8..24).collect::<Vec<u8>>());
315    }
316
317    #[test]
318    fn shared_payload_independent() {
319        let mut bytes = Vec::with_capacity(128);
320        bytes.extend(0..32_u8);
321        let payload = FramePayload::from_range(bytes, 8..24);
322        let alias = payload.clone();
323        let pointer = alias.as_ptr();
324        let memory = alias.memory_size();
325
326        let mut owned = payload.into_vec();
327        assert_ne!(owned.as_ptr(), pointer);
328        owned[0] = 99;
329        assert_eq!(alias.as_ptr(), pointer);
330        assert_eq!(alias.as_slice(), (8..24).collect::<Vec<u8>>());
331        assert_eq!(alias.memory_size(), memory);
332    }
333
334    #[test]
335    fn request_sizes_match() {
336        for size in [0, 127, 128, 16383, 16384, 2097151, 2097152] {
337            let operation = "operation.測定";
338            let prepare = WireCommand::Prepare {
339                invocation: u64::MAX,
340                operation: operation.into(),
341                fingerprint: [255; 32],
342                generation: u64::MAX,
343                payload: vec![255; size],
344            };
345            let prepare = Frame::Call {
346                id: u64::MAX,
347                method: WIRE_METHOD.into(),
348                params: prepare.encode().unwrap().into(),
349            };
350            let child = WireChildCall {
351                context: [255; 32],
352                generation_context: Some([255; 32]),
353                values: Vec::new(),
354                operation: operation.into(),
355                fingerprint: [1; 32],
356                generation: u64::MAX,
357                payload: vec![255; size],
358                submit: true,
359            };
360            let child = Frame::Call {
361                id: u64::MAX,
362                method: CHILD_METHOD.into(),
363                params: child.encode().unwrap().into(),
364            };
365            assert_eq!(Frame::request_payload_encoded_len(operation, size), prepare.encoded_len().max(child.encoded_len()));
366        }
367        assert_eq!(Frame::request_payload_encoded_len("op", usize::MAX), usize::MAX);
368    }
369
370    #[test]
371    fn typed_sizes_match() {
372        use harmos_domain::model::storage::files::{ByteRange, FileCompletion, FileSource};
373        use harmos_operations::storage::ReadFile;
374        let request = ReadFile {
375            source: FileSource::Local("data/測定.bin".into()),
376            range: ByteRange { offset: 127, limit: Some(128) },
377        };
378        assert_eq!(
379            Frame::request_encoded_len::<ReadFile>(&request),
380            Frame::request_payload_encoded_len(ReadFile::ID, request.encode().len())
381        );
382        let reply = FileCompletion { bytes: u64::MAX };
383        let envelope = WireAnswer::Ready { payload: reply.encode() }.encode().unwrap();
384        let frame = Frame::Answer { id: u64::MAX, result: envelope.into() };
385        assert_eq!(Frame::reply_encoded_len::<ReadFile>(&reply), frame.encoded_len());
386        assert!(Frame::request_fits::<ReadFile>(&request));
387        assert!(Frame::reply_fits::<ReadFile>(&reply));
388    }
389
390    #[test]
391    fn child_growth_matches() {
392        let values = vec![CallValue {
393            key: "context.測定".into(),
394            payload: vec![255; 16384],
395        }];
396        for generation in [false, true] {
397            let empty = WireChildCall::payload_encoded_len("op", u64::MAX, 0, generation, &values).unwrap();
398            for size in [0, 127, 128, 16383, 16384] {
399                let call = WireChildCall {
400                    context: [255; 32],
401                    generation_context: generation.then_some([255; 32]),
402                    values: values.clone(),
403                    operation: "op".into(),
404                    fingerprint: [1; 32],
405                    generation: u64::MAX,
406                    payload: vec![0; size],
407                    submit: false,
408                };
409                let frame = Frame::Call {
410                    id: u64::MAX,
411                    method: CHILD_METHOD.into(),
412                    params: call.encode().unwrap().into(),
413                };
414                assert_eq!(Frame::child_payload_encoded_len("op", u64::MAX, size, generation, &values), frame.encoded_len());
415                assert_eq!(Frame::child_encoded_from_empty(empty, size), frame.encoded_len());
416            }
417        }
418    }
419
420    #[test]
421    fn frame_prefix_limits() {
422        let mut low = 0;
423        let mut high = Frame::LIMIT + 1;
424        while low + 1 < high {
425            let middle = low + (high - low) / 2;
426            if Frame::child_payload_encoded_len("op", u64::MAX, middle, true, &[]) <= Frame::LIMIT {
427                low = middle;
428            } else {
429                high = middle;
430            }
431        }
432        let mut call = WireChildCall {
433            context: [0; 32],
434            generation_context: Some([0; 32]),
435            values: Vec::new(),
436            operation: "op".into(),
437            fingerprint: [1; 32],
438            generation: u64::MAX,
439            payload: vec![0; low],
440            submit: true,
441        };
442        let frame = Frame::Call {
443            id: u64::MAX,
444            method: CHILD_METHOD.into(),
445            params: call.encode().unwrap().into(),
446        };
447        assert_eq!(frame.encoded_len(), Frame::LIMIT);
448        assert!(frame.fits());
449        let packet = crate::transport::Packet::Frame(frame);
450        assert_eq!(packet.encoded_len().unwrap(), Frame::LIMIT + 4);
451        assert_eq!(packet.encode().unwrap().len(), Frame::LIMIT + 4);
452        drop(packet);
453        call.payload.push(0);
454        let frame = Frame::Call {
455            id: u64::MAX,
456            method: CHILD_METHOD.into(),
457            params: call.encode().unwrap().into(),
458        };
459        assert_eq!(frame.encoded_len(), Frame::LIMIT + 1);
460        assert!(!frame.fits());
461        assert!(crate::transport::Packet::Frame(frame).encoded_len().is_err());
462    }
463}
harmos_sidecar/wire.rs

harmos_sidecar/
wire.rs
