lace_sans_io/iltp/
codec.rs

1//! Incremental ILTP codec of one connected carrier.
2
3use super::control::{RecordLabel, item_len};
4use super::encoded_stream::{Incoming, Outgoing, install};
5use super::framing::{
6    Framed, ITEM_MAX_BYTES, IltpError, IltpItem, IltpTruncation, ItemFramer, PrefaceRule, malformed,
7};
8use super::output::{Output, OutputBytes, RecordRead};
9use crate::records::ValidatedHeader;
10
11/// Bytes per transmitted or received chunk unless a runtime configures more.
12pub const DEFAULT_CHUNK_BYTES: usize = 64 * 1024;
13/// The smallest chunk a codec accepts.
14pub const MIN_CHUNK_BYTES: usize = 256;
15
16/// Encodes accepted output batches and decodes peer bytes for one connected
17/// carrier. Prefaces pass raw; the two prefaces of each logical session select
18/// and install both directions' 065 stream encodings.
19///
20/// Output is one accepted batch at a time, produced progressively into chunks
21/// of at most the codec's chunk size: the writer takes
22/// [`output_bytes`](Self::output_bytes), reports what it wrote with
23/// [`advance_output`](Self::advance_output), and asks again until the batch is
24/// complete. Each batch's bytes end with the encoding's flush.
25pub struct IltpCodec {
26    outgoing: Outgoing,
27    incoming: Incoming,
28    output: Output,
29    frame: ItemFramer,
30    finished: bool,
31    item_max_bytes: usize,
32    chunk_bytes: usize,
33}
34
35impl Default for IltpCodec {
36    fn default() -> Self {
37        Self::new(DEFAULT_CHUNK_BYTES)
38    }
39}
40
41impl std::fmt::Debug for IltpCodec {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        f.debug_struct("IltpCodec").finish_non_exhaustive()
44    }
45}
46
47impl IltpCodec {
48    /// A codec whose output chunks and retained decoded input are at most
49    /// `chunk_bytes`.
50    ///
51    /// # Panics
52    ///
53    /// If `chunk_bytes` is below [`MIN_CHUNK_BYTES`].
54    pub fn new(chunk_bytes: usize) -> Self {
55        assert!(
56            chunk_bytes >= MIN_CHUNK_BYTES,
57            "ILTP chunk size is below {MIN_CHUNK_BYTES} bytes"
58        );
59        Self {
60            outgoing: Outgoing::Raw,
61            incoming: Incoming::Raw,
62            output: Output::new(chunk_bytes),
63            frame: ItemFramer::default(),
64            finished: false,
65            item_max_bytes: ITEM_MAX_BYTES,
66            chunk_bytes,
67        }
68    }
69
70    /// Accept one machine output batch, validated completely before any of
71    /// its bytes are produced. A preface first ends the previous logical
72    /// session's encoded stream with its end mark.
73    pub fn accept_output(&mut self, items: Vec<IltpItem>) -> Result<(), IltpError> {
74        self.check_idle()?;
75        let mut plaintext = 0usize;
76        for item in &items {
77            plaintext = plaintext.saturating_add(canonical_item_size(item)?);
78        }
79        self.output.accept(items, plaintext);
80        Ok(())
81    }
82
83    /// Accept the end of the outgoing encoded stream: a compressed encoding's
84    /// end mark. Only a fresh preface may be accepted afterwards.
85    pub fn accept_end(&mut self) -> Result<(), IltpError> {
86        self.check_idle()?;
87        self.output.accept_end();
88        Ok(())
89    }
90
91    fn check_idle(&self) -> Result<(), IltpError> {
92        if self.output.is_pending() {
93            return Err(IltpError::Local("an ILTP output is already accepted"));
94        }
95        Ok(())
96    }
97
98    /// Whether accepted output has bytes the writer has not taken.
99    pub fn output_pending(&self) -> bool {
100        self.output.is_pending()
101    }
102
103    /// The next bytes of the accepted output, produced when the previous ones
104    /// are taken, or the record read their production needs.
105    pub fn output_bytes(&mut self) -> Result<OutputBytes<'_>, IltpError> {
106        self.output.bytes(&mut self.outgoing, &mut self.incoming)
107    }
108
109    /// Return a [`RecordRead`] that [`output_bytes`](Self::output_bytes)
110    /// handed out, after running it.
111    pub fn complete_read(&mut self, read: RecordRead) -> Result<(), IltpError> {
112        self.output.complete_read(read)
113    }
114
115    /// Accept `written` bytes of the current [`output_bytes`](Self::output_bytes).
116    pub fn advance_output(&mut self, written: usize) {
117        self.output.advance(written);
118    }
119
120    /// Drop accepted output the peer can no longer receive. The outgoing
121    /// encoding is unusable afterwards; only terminal retirement follows.
122    pub fn discard_output(&mut self) {
123        self.output.discard();
124    }
125
126    /// Append every remaining byte of the accepted output to `out`,
127    /// performing record reads on the calling thread.
128    pub fn drain_output(&mut self, out: &mut Vec<u8>) -> Result<(), IltpError> {
129        loop {
130            match self.output_bytes()? {
131                OutputBytes::Write(bytes) => {
132                    out.extend_from_slice(bytes);
133                    let written = bytes.len();
134                    self.advance_output(written);
135                }
136                OutputBytes::Read(read) => {
137                    let read = read
138                        .run()
139                        .map_err(|_| IltpError::Local("record bytes could not be read"))?;
140                    self.complete_read(read)?;
141                }
142                OutputBytes::Complete => return Ok(()),
143            }
144        }
145    }
146
147    /// Decode a prefix of `input` through at most one [`Decoded`] event and
148    /// return the consumed length. Raw input after the event stays with the
149    /// caller; decoded bytes after it stay in the codec, so call again, with
150    /// empty input if none remains, until no event is returned.
151    ///
152    /// A record item is not retained: its owner writes the prefix of
153    /// [`Decoded::RecordStart`] and every [`Decoded::RecordData`] to its
154    /// record sink before decoding continues, and only
155    /// [`Decoded::RecordEnd`] proves the bytes valid.
156    pub fn decode<'a>(
157        &'a mut self,
158        input: &'a [u8],
159    ) -> Result<(usize, Option<Decoded<'a>>), IltpError> {
160        if self.finished {
161            if input.is_empty() {
162                return Ok((0, None));
163            }
164            return Err(malformed("ILTP decoder is finalized"));
165        }
166        let mut consumed = 0;
167        let step = loop {
168            let (framed, source) = match &mut self.incoming {
169                Incoming::Pending(_) => break None,
170                Incoming::Raw | Incoming::Identity if consumed == input.len() => break None,
171                Incoming::Raw | Incoming::Identity => {
172                    let rule = if matches!(self.incoming, Incoming::Raw) {
173                        PrefaceRule::Only
174                    } else {
175                        PrefaceRule::Allowed
176                    };
177                    let (count, framed) =
178                        self.frame
179                            .consume(&input[consumed..], self.item_max_bytes, rule)?;
180                    consumed += count;
181                    (framed, Source::Input(consumed - count..consumed))
182                }
183                Incoming::Compressed(decoding) => {
184                    if !decoding.buffered().is_empty() {
185                        let (count, framed) = self.frame.consume(
186                            decoding.buffered(),
187                            self.item_max_bytes,
188                            PrefaceRule::Forbidden,
189                        )?;
190                        (framed, Source::Decoded(decoding.advance(count)))
191                    } else if decoding.is_ended() {
192                        if !self.frame.is_empty() {
193                            return Err(malformed("compressed ILTP stream ended inside an item"));
194                        }
195                        self.incoming = Incoming::Raw;
196                        continue;
197                    } else {
198                        let progress = decoding.fill(&input[consumed..])?;
199                        consumed += progress.consumed;
200                        if progress.consumed == 0 && progress.produced == 0 {
201                            break None;
202                        }
203                        continue;
204                    }
205                }
206            };
207            match framed {
208                None => {}
209                Some(Framed::Item(item)) => break Some(Step::Item(item)),
210                Some(Framed::RecordStart) => break Some(Step::Start),
211                Some(Framed::RecordData) => break Some(Step::Data(source)),
212                Some(Framed::RecordEnd(label, header)) => {
213                    break Some(Step::End(label, header));
214                }
215            }
216        };
217        let event = match step {
218            None => None,
219            Some(Step::Item(item)) => {
220                self.accept(&item)?;
221                Some(Decoded::Item(item))
222            }
223            Some(Step::Start) => {
224                let (label, prefix, canonical_len) = self
225                    .frame
226                    .record_start()
227                    .ok_or(IltpError::Local("record start without a complete prefix"))?;
228                Some(Decoded::RecordStart {
229                    label,
230                    prefix,
231                    canonical_len,
232                })
233            }
234            Some(Step::Data(Source::Input(range))) => Some(Decoded::RecordData(&input[range])),
235            Some(Step::Data(Source::Decoded(range))) => match &self.incoming {
236                Incoming::Compressed(decoding) => Some(Decoded::RecordData(decoding.framed(range))),
237                _ => return Err(IltpError::Local("decoded record data without its decoder")),
238            },
239            Some(Step::End(label, header)) => Some(Decoded::RecordEnd(label, header)),
240        };
241        Ok((consumed, event))
242    }
243
244    /// Whether decoding waits for the local preface: the peer preface is
245    /// decoded and the session's encodings install once the local one is
246    /// written.
247    pub fn awaits_local_preface(&self) -> bool {
248        matches!(self.incoming, Incoming::Pending(_))
249    }
250
251    /// Finalize decoding at EOF: boundary EOF needs an item boundary and the
252    /// direction's encoding at its end. Every event must have been decoded.
253    pub fn finish(&mut self) -> Result<(), IltpError> {
254        if self.finished {
255            return Err(malformed("ILTP decoder is already finalized"));
256        }
257        if self.decode(&[])?.1.is_some() {
258            return Err(malformed(
259                "finish requires complete items to be consumed first",
260            ));
261        }
262        self.finished = true;
263        if !self.frame.is_empty() {
264            return Err(IltpError::Truncated(self.frame.truncation()));
265        }
266        if !self.incoming.at_end() {
267            return Err(IltpError::Truncated(IltpTruncation::StreamEncoding));
268        }
269        Ok(())
270    }
271
272    /// A decoded preface completes the session's pair when the local preface
273    /// has been written; otherwise decoding waits for it.
274    fn accept(&mut self, item: &IltpItem) -> Result<(), IltpError> {
275        if let IltpItem::SessionPreface(preface) = item {
276            if let Outgoing::Pending(local) = &self.outgoing {
277                let local = local.clone();
278                install(
279                    &mut self.outgoing,
280                    &mut self.incoming,
281                    &local,
282                    &preface.offers,
283                    self.chunk_bytes,
284                )?;
285            } else {
286                self.incoming = Incoming::Pending(preface.offers.clone());
287            }
288        }
289        Ok(())
290    }
291}
292
293/// One decoded event of peer input; see [`IltpCodec::decode`].
294#[derive(Debug)]
295pub enum Decoded<'a> {
296    /// A complete preface, line item or list item.
297    Item(IltpItem),
298    /// A record item's prefix is complete and well formed and its
299    /// canonical length is known. Neither is evidence that the record is valid.
300    RecordStart {
301        label: RecordLabel,
302        prefix: &'a [u8],
303        canonical_len: usize,
304    },
305    /// The record's next data bytes.
306    RecordData(&'a [u8]),
307    /// The record's bytes are complete and verified and the LF after it
308    /// consumed. It is reported before any byte of the next item is decoded.
309    RecordEnd(RecordLabel, ValidatedHeader),
310}
311
312/// Where a framed data range lies.
313enum Source {
314    Input(std::ops::Range<usize>),
315    Decoded(std::ops::Range<usize>),
316}
317
318/// A framing result before it borrows its bytes.
319enum Step {
320    Item(IltpItem),
321    Start,
322    Data(Source),
323    End(RecordLabel, ValidatedHeader),
324}
325
326/// Canonical encoded size of one item, without rendering or copying it.
327/// A control item beyond the item cap is `ItemTooLarge`.
328pub fn canonical_item_size(item: &IltpItem) -> Result<usize, IltpError> {
329    let len = item_len(item);
330    if len > ITEM_MAX_BYTES {
331        return Err(IltpError::ItemTooLarge);
332    }
333    Ok(len)
334}
335
336#[cfg(test)]
337mod linear_tests;
338#[cfg(test)]
339pub(crate) mod test_support;
340#[cfg(test)]
341mod tests;
lace_sans_io/iltp/codec.rs

lace_sans_io/iltp/
codec.rs
