harmos_sdk/operations/serving/
codec.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod payload {
13    use serde::{Deserialize, Deserializer, Serializer};
14
15    pub fn serialize<S: Serializer>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
16        serializer.serialize_bytes(bytes)
17    }
18
19    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<&'de [u8], D::Error> {
20        <&[u8]>::deserialize(deserializer)
21    }
22}
23
24/* === Re-Exports === */
25
26/* === Imports === */
27
28// Standard Library Imports
29
30// Third Party Imports
31use serde::{Deserialize, Serialize};
32// Local Imports
33use super::{WireAnswer, WireChild, WireChildAction, WireChildCall, WireChildControl, WireCommand};
34
35/* === Constants === */
36
37const VERSION: u8 = 6;
38pub(super) const LIMIT: usize = 16 * 1024 * 1024;
39
40/* === Macros === */
41
42/* === Type Aliases === */
43
44/* === Structs === */
45
46#[derive(Serialize, Deserialize)]
47struct Value<'a> {
48    key: &'a str,
49
50    #[serde(borrow, with = "payload")]
51    payload: &'a [u8],
52}
53
54/* === Enums === */
55
56#[derive(Debug, thiserror::Error)]
57pub enum WireCodecError {
58    #[error("operation envelope exceeds the 16 MiB limit")]
59    Oversized,
60
61    #[error("unsupported operation envelope version")]
62    Version,
63
64    #[error("unexpected operation envelope kind")]
65    Kind,
66
67    #[error("trailing operation envelope bytes")]
68    Trailing,
69
70    #[error("invalid operation envelope")]
71    Encoding(#[from] postcard::Error),
72}
73
74// Variant Order Is Part Of Version 6. Borrowed Fields Validate Before Allocation.
75// As In The Transport Experiment, Payloads Use Postcard Byte Strings, Not Sequences.
76#[derive(Serialize, Deserialize)]
77enum Envelope<'a> {
78    Prepare {
79        invocation: u64,
80        operation: &'a str,
81
82        #[serde(borrow, with = "payload")]
83        payload: &'a [u8],
84    },
85    Initialize {
86        invocation: u64,
87        context: [u8; 32],
88    },
89    Attempt {
90        invocation: u64,
91        context: [u8; 32],
92
93        #[serde(borrow)]
94        values: Vec<Value<'a>>,
95    },
96    Finalize {
97        invocation: u64,
98    },
99    Ready {
100        #[serde(borrow, with = "payload")]
101        payload: &'a [u8],
102    },
103    Failed {
104        failure: crate::Failure,
105    },
106    Child {
107        context: [u8; 32],
108        generation_context: Option<[u8; 32]>,
109
110        #[serde(borrow)]
111        values: Vec<Value<'a>>,
112
113        operation: &'a str,
114
115        #[serde(borrow, with = "payload")]
116        payload: &'a [u8],
117
118        submit: bool,
119    },
120    PrepareGrpc {
121        invocation: u64,
122        operation: &'a str,
123
124        #[serde(borrow, with = "payload")]
125        payload: &'a [u8],
126    },
127    StreamAttempt {
128        invocation: u64,
129        epoch: u64,
130        context: [u8; 32],
131
132        #[serde(borrow)]
133        values: Vec<Value<'a>>,
134    },
135    StreamSend {
136        invocation: u64,
137        epoch: u64,
138
139        #[serde(borrow, with = "payload")]
140        payload: &'a [u8],
141    },
142    StreamFinish {
143        invocation: u64,
144        epoch: u64,
145    },
146    StreamReceive {
147        invocation: u64,
148        epoch: u64,
149    },
150    StreamItem {
151        epoch: u64,
152        transferred: bool,
153
154        #[serde(borrow, with = "payload")]
155        payload: &'a [u8],
156    },
157    StreamEnd {
158        epoch: u64,
159        transferred: bool,
160    },
161    ChildControl {
162        context: [u8; 32],
163        instance: u64,
164        action: WireChildAction,
165    },
166}
167
168/* === Traits === */
169
170/* === Implementations === */
171
172impl WireCommand {
173    pub fn encode(&self) -> Result<Vec<u8>, WireCodecError> {
174        encode(self.envelope())
175    }
176
177    pub fn decode(bytes: &[u8]) -> Result<Self, WireCodecError> {
178        Ok(match decode(bytes)? {
179            Envelope::Prepare { invocation, operation, payload } => {
180                Self::Prepare { invocation, operation: operation.to_owned(), payload: payload.to_vec() }
181            }
182            Envelope::PrepareGrpc { invocation, operation, payload } => {
183                Self::PrepareGrpc { invocation, operation: operation.to_owned(), payload: payload.to_vec() }
184            }
185            Envelope::Initialize { invocation, context } => Self::Initialize { invocation, context },
186            Envelope::Attempt { invocation, context, values } => Self::Attempt { invocation, context, values: owned(values)? },
187            Envelope::Finalize { invocation } => Self::Finalize { invocation },
188            Envelope::StreamAttempt { invocation, epoch, context, values } => {
189                Self::StreamAttempt { invocation, epoch, context, values: owned(values)? }
190            }
191            Envelope::StreamSend { invocation, epoch, payload } => Self::StreamSend { invocation, epoch, payload: payload.to_vec() },
192            Envelope::StreamFinish { invocation, epoch } => Self::StreamFinish { invocation, epoch },
193            Envelope::StreamReceive { invocation, epoch } => Self::StreamReceive { invocation, epoch },
194            _ => return Err(WireCodecError::Kind),
195        })
196    }
197
198    fn envelope(&self) -> Envelope<'_> {
199        match self {
200            Self::Prepare { invocation, operation, payload } => Envelope::Prepare { invocation: *invocation, operation, payload },
201            Self::PrepareGrpc { invocation, operation, payload } => Envelope::PrepareGrpc { invocation: *invocation, operation, payload },
202            Self::Initialize { invocation, context } => Envelope::Initialize { invocation: *invocation, context: *context },
203            Self::Attempt { invocation, context, values } => {
204                Envelope::Attempt { invocation: *invocation, context: *context, values: borrowed(values) }
205            }
206            Self::StreamAttempt { invocation, epoch, context, values } => {
207                Envelope::StreamAttempt { invocation: *invocation, epoch: *epoch, context: *context, values: borrowed(values) }
208            }
209            Self::StreamSend { invocation, epoch, payload } => Envelope::StreamSend { invocation: *invocation, epoch: *epoch, payload },
210            Self::StreamFinish { invocation, epoch } => Envelope::StreamFinish { invocation: *invocation, epoch: *epoch },
211            Self::StreamReceive { invocation, epoch } => Envelope::StreamReceive { invocation: *invocation, epoch: *epoch },
212            Self::Finalize { invocation } => Envelope::Finalize { invocation: *invocation },
213        }
214    }
215}
216
217impl WireAnswer {
218    pub fn encode(&self) -> Result<Vec<u8>, WireCodecError> {
219        let envelope = match self {
220            Self::Ready { payload } => Envelope::Ready { payload },
221            Self::Stream { epoch, transferred, payload: Some(payload) } => {
222                Envelope::StreamItem { epoch: *epoch, transferred: *transferred, payload }
223            }
224            Self::Stream { epoch, transferred, payload: None } => Envelope::StreamEnd { epoch: *epoch, transferred: *transferred },
225            Self::Failed { failure } => Envelope::Failed { failure: failure.clone() },
226        };
227        encode(envelope)
228    }
229
230    pub fn decode(bytes: &[u8]) -> Result<Self, WireCodecError> {
231        Ok(match decode(bytes)? {
232            Envelope::StreamItem { epoch, transferred, payload } => Self::Stream { epoch, transferred, payload: Some(payload.to_vec()) },
233            Envelope::StreamEnd { epoch, transferred } => Self::Stream { epoch, transferred, payload: None },
234            Envelope::Ready { payload } => Self::Ready { payload: payload.to_vec() },
235            Envelope::Failed { failure } => Self::Failed { failure },
236            _ => return Err(WireCodecError::Kind),
237        })
238    }
239}
240
241impl WireChildCall {
242    pub fn encode(&self) -> Result<Vec<u8>, WireCodecError> {
243        check_values(&self.values)?;
244        encode(Envelope::Child {
245            context: self.context,
246            generation_context: self.generation_context,
247            values: borrowed(&self.values),
248            operation: &self.operation,
249            payload: &self.payload,
250            submit: self.submit,
251        })
252    }
253
254    pub fn decode(bytes: &[u8]) -> Result<Self, WireCodecError> {
255        match WireChild::decode(bytes)? {
256            WireChild::Call(call) => Ok(call),
257            WireChild::Control(_) => Err(WireCodecError::Kind),
258        }
259    }
260}
261
262impl WireChildControl {
263    pub fn encode(&self) -> Result<Vec<u8>, WireCodecError> {
264        encode(Envelope::ChildControl { context: self.context, instance: self.instance, action: self.action })
265    }
266}
267
268impl WireChild {
269    pub fn decode(bytes: &[u8]) -> Result<Self, WireCodecError> {
270        let child = match decode(bytes)? {
271            Envelope::Child { context, generation_context, values, operation, payload, submit } => Self::Call(WireChildCall {
272                context,
273                generation_context,
274                values: owned(values)?,
275                operation: operation.to_owned(),
276                payload: payload.to_vec(),
277                submit,
278            }),
279            Envelope::ChildControl { context, instance, action } => Self::Control(WireChildControl { context, instance, action }),
280            _ => return Err(WireCodecError::Kind),
281        };
282        Ok(child)
283    }
284
285    pub fn context(&self) -> &[u8; 32] {
286        match self {
287            Self::Call(call) => &call.context,
288            Self::Control(control) => &control.context,
289        }
290    }
291}
292
293/* === Public Functions === */
294
295/* === Private Functions === */
296
297fn check_values(values: &[super::super::CallValue]) -> Result<(), WireCodecError> {
298    let bytes = values.iter().map(|value| value.key.len().saturating_add(value.payload.len())).fold(0usize, usize::saturating_add);
299    if values.len() > super::super::PORTABLE_VALUES || bytes > super::super::PORTABLE_BYTES {
300        return Err(WireCodecError::Oversized);
301    }
302    for (index, value) in values.iter().enumerate() {
303        if values[..index].iter().any(|previous| previous.key == value.key) {
304            return Err(WireCodecError::Kind);
305        }
306    }
307    Ok(())
308}
309
310fn borrowed(values: &[super::super::CallValue]) -> Vec<Value<'_>> {
311    values.iter().map(|value| Value { key: &value.key, payload: &value.payload }).collect()
312}
313
314fn owned(values: Vec<Value<'_>>) -> Result<Vec<super::super::CallValue>, WireCodecError> {
315    let bytes = values.iter().map(|value| value.key.len() + value.payload.len()).sum::<usize>();
316    if values.len() > super::super::PORTABLE_VALUES || bytes > super::super::PORTABLE_BYTES {
317        return Err(WireCodecError::Oversized);
318    }
319    Ok(values.into_iter().map(|value| super::super::CallValue { key: value.key.to_owned(), payload: value.payload.to_vec() }).collect())
320}
321
322fn encode(envelope: Envelope<'_>) -> Result<Vec<u8>, WireCodecError> {
323    let message = (VERSION, envelope);
324    let size = postcard::experimental::serialized_size(&message)?;
325    if size > LIMIT {
326        return Err(WireCodecError::Oversized);
327    }
328    Ok(postcard::to_allocvec(&message)?)
329}
330
331fn decode(bytes: &[u8]) -> Result<Envelope<'_>, WireCodecError> {
332    if bytes.len() > LIMIT {
333        return Err(WireCodecError::Oversized);
334    }
335    let (version, rest) = postcard::take_from_bytes::<u8>(bytes)?;
336    if version != VERSION {
337        return Err(WireCodecError::Version);
338    }
339    let (envelope, rest) = postcard::take_from_bytes::<Envelope<'_>>(rest)?;
340    if !rest.is_empty() {
341        return Err(WireCodecError::Trailing);
342    }
343    Ok(envelope)
344}
345
346/* === Tests === */
harmos_sdk/operations/serving/codec.rs

harmos_sdk/operations/serving/
codec.rs
