structured_asic/zip/
stream.rs

1//! Complete member streams reuse the common sticky CMS source protocol.
2
3use super::{Archive, Member, MemberId};
4use crate::error::{Error, Limit, Unsupported, resource};
5use structured_asn1::range::RangeSource;
6use structured_asn1::{
7    Span,
8    inflate::{Envelope, Error as InflateError, Inflater, Progress},
9    memory::{MemoryError, OwnedVec, Work},
10};
11use structured_cms::stream::{AsyncSource, Receiver, Refused};
12
13/// Original member-acquisition result, including source identity on refusal.
14pub type Acquisition<T, S> = Result<
15    T,
16    ReadFailure<
17        <S as RangeSource>::Error,
18        <S as RangeSource>::Document,
19        <S as RangeSource>::Generation,
20    >,
21>;
22
23/// The complete original member could not be delivered.
24#[derive(Debug)]
25pub enum ReadFailure<E, D, G> {
26    /// Original range/work/storage refusal.
27    Processing {
28        /// Exact original occurrence.
29        member: MemberId<D, G>,
30        /// Original typed processing cause.
31        cause: Error<E>,
32    },
33    /// An original envelope or payload integrity check failed.
34    Integrity {
35        /// Exact original occurrence.
36        member: MemberId<D, G>,
37        /// Physical original member enclosure.
38        range: Span,
39        /// Independent integrity cause.
40        kind: Integrity,
41    },
42    /// The common stream driver retains the receiver's original refusal.
43    Receiver(Refused),
44}
45
46/// Payload integrity is distinct from signature mathematics and trust.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum Integrity {
49    /// No unique consistent local record belongs to this occurrence.
50    Framing,
51    /// Invalid or truncated raw DEFLATE syntax.
52    Compression,
53    /// Declared compressed/decoded extent disagrees with complete decoding.
54    Length,
55    /// CRC-32 disagrees with the original central/local metadata.
56    Crc,
57}
58
59/// One original entry selected before digest/provider state creation.
60pub struct Source<'v, 's, S: RangeSource> {
61    archive: &'v Archive<'s, S>,
62    ordinal: usize,
63}
64
65impl<'v, 's, S: RangeSource> Member<'v, 's, S> {
66    /// Checks source, method and unambiguous framing before constructing a
67    /// payload source. CRC and final extent remain complete-EOF obligations.
68    pub fn source<W: Work>(&self, work: &W) -> Acquisition<Source<'v, 's, S>, S> {
69        check(self.archive, self.ordinal, work)?;
70        Ok(Source {
71            archive: self.archive,
72            ordinal: self.ordinal,
73        })
74    }
75
76    /// Retains a complete carried CMS/XML member under actual-capacity funding.
77    /// Large data objects should be streamed; no successful prefix is returned
78    /// after source, CRC, decoded-length or receiver refusal.
79    ///
80    /// # Panics
81    /// An internal framing invariant fails if a preflighted readable member
82    /// lacks its already checked original payload range.
83    pub async fn read_bytes<W: Work>(
84        &self,
85        work: &W,
86    ) -> Result<OwnedVec<u8>, ReadFailure<S::Error, S::Document, S::Generation>> {
87        let mut source = self.source(work)?;
88        let execution = self.archive.session.query(work);
89        let range = self.entry().data.expect("checked original payload");
90        let mut bytes = execution.vector();
91        let count = usize::try_from(self.entry().decoded)
92            .map_err(|_| failure(self, resource(range, MemoryError::LimitExceeded)))?;
93        bytes
94            .try_reserve_exact(count)
95            .map_err(|cause| failure(self, resource(range, cause)))?;
96        let mut receiver = Collect {
97            bytes: &mut bytes,
98            first: None,
99        };
100        let result = source.stream(&mut receiver, &execution).await;
101        if let Some(cause) = receiver.first {
102            return Err(failure(self, resource(range, cause)));
103        }
104        result?;
105        Ok(bytes.into_owned())
106    }
107}
108
109struct Collect<'a, W: Work> {
110    bytes: &'a mut OwnedVec<u8, W>,
111    first: Option<MemoryError>,
112}
113impl<W: Work> Receiver for Collect<'_, W> {
114    fn write(&mut self, bytes: &[u8]) -> Result<(), Refused> {
115        if self.first.is_some() {
116            return Err(Refused);
117        }
118        if let Err(cause) = self.bytes.try_extend_from_slice(bytes) {
119            self.first = Some(cause);
120            return Err(Refused);
121        }
122        Ok(())
123    }
124}
125
126fn failure<S: RangeSource>(
127    member: &Member<'_, '_, S>,
128    cause: Error<S::Error>,
129) -> ReadFailure<S::Error, S::Document, S::Generation> {
130    ReadFailure::Processing {
131        member: member.id(),
132        cause,
133    }
134}
135
136fn check<S: RangeSource, W: Work>(
137    archive: &Archive<'_, S>,
138    ordinal: usize,
139    work: &W,
140) -> Acquisition<(), S> {
141    let member = Member { archive, ordinal };
142    let entry = member.entry();
143    archive
144        .admit(work, entry.central)
145        .map_err(|cause| failure(&member, cause))?;
146    if entry.flags & 0x2041 != 0 {
147        return Err(failure(
148            &member,
149            Error::Unsupported {
150                range: entry.central,
151                kind: Unsupported::Encryption,
152            },
153        ));
154    }
155    if entry.method != 0 && entry.method != 8 {
156        return Err(failure(
157            &member,
158            Error::Unsupported {
159                range: entry.central,
160                kind: Unsupported::Compression(entry.method),
161            },
162        ));
163    }
164    if !entry.readable || entry.data.is_none() {
165        return Err(ReadFailure::Integrity {
166            member: member.id(),
167            range: entry.central,
168            kind: Integrity::Framing,
169        });
170    }
171    if entry.decoded > archive.session.remaining() {
172        return Err(failure(
173            &member,
174            Error::Limit {
175                range: entry.central,
176                kind: Limit::Decoded,
177            },
178        ));
179    }
180    Ok(())
181}
182
183impl<S: RangeSource> AsyncSource for Source<'_, '_, S> {
184    type Error = ReadFailure<S::Error, S::Document, S::Generation>;
185    fn stream<'q, W: Work + 'q, R: Receiver + 'q>(
186        &'q mut self,
187        receiver: &'q mut R,
188        work: &'q W,
189    ) -> impl core::future::Future<Output = Result<(), Self::Error>> + 'q {
190        acquire(self.archive, self.ordinal, receiver, work)
191    }
192}
193
194#[expect(
195    clippy::too_many_lines,
196    reason = "one streaming transaction keeps inflater progress, cumulative output admission, CRC and exact EOF validation in a single state machine"
197)]
198async fn acquire<S: RangeSource, W: Work>(
199    archive: &Archive<'_, S>,
200    ordinal: usize,
201    receiver: &mut impl Receiver,
202    work: &W,
203) -> Result<(), ReadFailure<S::Error, S::Document, S::Generation>> {
204    check(archive, ordinal, work)?;
205    let member = Member { archive, ordinal };
206    let entry = member.entry();
207    let range = entry.data.expect("checked original payload");
208    let processing = |cause| failure(&member, cause);
209    let integrity = |kind| ReadFailure::Integrity {
210        member: member.id(),
211        range,
212        kind,
213    };
214    let execution = archive.session.query(work);
215    let mut window = archive.window(work, range).map_err(processing)?;
216    let mut inflater = if entry.method == 8 {
217        Some(
218            Inflater::new(Envelope::Raw, &execution).map_err(|cause| match cause {
219                InflateError::Resource(cause) => processing(resource(range, cause)),
220                _ => integrity(Integrity::Compression),
221            })?,
222        )
223    } else {
224        None
225    };
226    let mut crc = crc32fast::Hasher::new();
227    let mut at = range.start;
228    let mut produced = 0_u64;
229    let mut quantum = 0;
230    loop {
231        let input = window
232            .chunk_at(at, range.end())
233            .await
234            .map_err(|cause| processing(cause.into()))?;
235        let output = if let Some(inflater) = &mut inflater {
236            let remaining = entry.decoded - produced;
237            if quantum == 0 {
238                quantum = input.len().clamp(1, 4096);
239            }
240            let maximum = usize::try_from(
241                remaining
242                    .min(archive.session.remaining())
243                    .min(quantum as u64),
244            )
245            .expect("bounded decoder allowance");
246            let result = inflater.step(input, at + (input.len() as u64) < range.end(), maximum);
247            let step = match result {
248                Ok(step) => step,
249                Err(InflateError::Resource(cause)) => {
250                    return Err(processing(resource(range, cause)));
251                }
252                Err(InflateError::OutputLimit) => {
253                    if archive.session.remaining() < remaining {
254                        return Err(processing(Error::Limit {
255                            range,
256                            kind: Limit::Decoded,
257                        }));
258                    }
259                    return Err(integrity(Integrity::Length));
260                }
261                Err(InflateError::Malformed { produced }) => {
262                    archive
263                        .session
264                        .decode(produced as u64, range)
265                        .map_err(processing)?;
266                    return Err(integrity(Integrity::Compression));
267                }
268                Err(InflateError::State) => return Err(integrity(Integrity::Compression)),
269            };
270            at += step.consumed as u64;
271            let done = step.progress == Progress::Done;
272            let output = step.output;
273            // Short acquisition must not prepay a 4KiB output on every
274            // seven-byte input. Expand only when actual output fills the
275            // allowance; full-window/high-expansion paths still reach 4KiB.
276            quantum = if step.progress == Progress::Output {
277                (maximum * 2).clamp(1, 4096)
278            } else {
279                output.len().max(input.len()).clamp(1, 4096)
280            };
281            archive
282                .session
283                .decode(output.len() as u64, range)
284                .map_err(processing)?;
285            produced += output.len() as u64;
286            execution
287                .charge_work(output.len() as u64 + 1)
288                .map_err(|cause| processing(resource(range, cause)))?;
289            crc.update(output);
290            if !output.is_empty() {
291                receiver.write(output).map_err(ReadFailure::Receiver)?;
292            }
293            if done {
294                if at != range.end() || produced != entry.decoded {
295                    return Err(integrity(Integrity::Length));
296                }
297                break;
298            }
299            continue;
300        } else {
301            if at == range.end() {
302                break;
303            }
304            archive
305                .session
306                .decode(input.len() as u64, range)
307                .map_err(processing)?;
308            at += input.len() as u64;
309            produced += input.len() as u64;
310            input
311        };
312        execution
313            .charge_work(output.len() as u64 + 1)
314            .map_err(|cause| processing(resource(range, cause)))?;
315        crc.update(output);
316        receiver.write(output).map_err(ReadFailure::Receiver)?;
317    }
318    archive.admit(work, range).map_err(processing)?;
319    if produced != entry.decoded {
320        return Err(integrity(Integrity::Length));
321    }
322    if crc.finalize() != entry.crc {
323        return Err(integrity(Integrity::Crc));
324    }
325    Ok(())
326}
structured_asic/zip/stream.rs

structured_asic/zip/
stream.rs
