structured_cms/
digest.rs

1//! One-pass digest fan-out with one final output per resolved semantic identity.
2
3use crate::Error;
4use crate::provider::{Finalized, Finished, Prehash};
5use crate::stream::{self, Sink, Source, StreamError};
6use core::cmp::Ordering;
7use structured_asn1::digest::{Digest, Resolution};
8use structured_asn1::memory::{OwnedVec, SortError, Work};
9
10/// A complete pass preserves source, common-processing and provider causes.
11pub type PassError<SE, PE> = StreamError<SE, DigestError<PE>>;
12/// A finalized output with the selected provider's full identity and storage type.
13pub type DigestOutput<'a, P> =
14    Finalized<'a, P, <P as Digest<'a>>::Identity, <P as Digest<'a>>::Output>;
15/// Fallible digest processing, without reducing provider refusal to absence.
16pub type DigestResult<T, E> = Result<T, DigestError<E>>;
17/// A complete source pass, preserving acquisition and processing causes.
18pub type PassResult<T, SE, PE> = Result<T, PassError<SE, PE>>;
19
20/// Failure of common digest processing or the selected provider.
21#[derive(Debug)]
22pub enum DigestError<E> {
23    /// Common storage, dispatch, sorting or lookup refusal before work.
24    Processing(Error),
25    /// Original callback error, including any opaque backend resource refusal.
26    Callback(E),
27    /// A callback returned a width inconsistent with its resolved identity.
28    /// This is a provider contract failure, not malformed document data.
29    OutputWidth {
30        /// Fixed width promised by the selected resolved identity.
31        expected: usize,
32        /// Actual returned output width.
33        actual: usize,
34    },
35    /// The provider resolved this identity but cannot stream it, which CMS
36    /// content and attribute digests require. A capability answer of the
37    /// provider, not a document fault.
38    Unstreamable,
39}
40
41fn started<S, E>(start: Result<Resolution<S>, E>) -> Result<S, DigestError<E>> {
42    match start.map_err(DigestError::Callback)? {
43        Resolution::Supported(state) => Ok(state),
44        Resolution::Unsupported | Resolution::Malformed => Err(DigestError::Unstreamable),
45    }
46}
47
48struct Pending<I, S> {
49    identity: I,
50    state: S,
51}
52
53struct Request<I> {
54    ordinal: usize,
55    identity: I,
56}
57
58/// Prepared states from one selected provider; no state cloning or source replay.
59pub struct DigestBatch<'a, P: Digest<'a>> {
60    provider: &'a P,
61    pending: OwnedVec<Pending<P::Identity, P::State>>,
62    mapping: OwnedVec<usize>,
63    field: &'static str,
64    at: u64,
65}
66
67impl<'a, P: Digest<'a>> DigestBatch<'a, P> {
68    /// Moves resolved identities into the current operation and initializes each once.
69    ///
70    /// Identities must come from this selected provider's resolver. They are
71    /// sorted and deduplicated through its fallible semantic comparison, not
72    /// OID/width equality. Admission precedes every callback, move and allocation.
73    /// `field`/`at` identify the caller's enclosing original source structure.
74    pub fn prepare<W: Work>(
75        provider: &'a P,
76        identities: OwnedVec<P::Identity>,
77        execution: &W,
78        field: &'static str,
79        at: u64,
80    ) -> Result<Self, DigestError<P::Error>> {
81        crate::provider::require_instance(provider, field, at).map_err(DigestError::Processing)?;
82        let refusal = |reason| DigestError::Processing(Error::resource(reason, field, at));
83        execution.charge_work(1).map_err(refusal)?;
84        let count = identities.len();
85        let mut ordered = execution.vector();
86        ordered
87            .try_reserve_exact(identities.len())
88            .map_err(refusal)?;
89        execution
90            .charge_work(core::mem::size_of_val(&*identities) as u64)
91            .map_err(refusal)?;
92        for (ordinal, identity) in identities.into_iter().enumerate() {
93            ordered
94                .try_push(Request { ordinal, identity })
95                .map_err(refusal)?;
96        }
97        ordered
98            .try_sort_unstable_by(|left, right| {
99                provider.compare(&left.identity, &right.identity, execution)
100            })
101            .map_err(|error| match error {
102                SortError::Resource(reason) => refusal(reason),
103                SortError::Comparison(error) => DigestError::Callback(error),
104            })?;
105        let mut pending: OwnedVec<Pending<P::Identity, P::State>, _> = execution.vector();
106        let mut mapping = execution.vector();
107        mapping.try_reserve_exact(count).map_err(refusal)?;
108        // The sorted request records are a permutation of ordinals 0..count.
109        // Every slot is assigned exactly once before this mapping is exposed.
110        for _ in 0..count {
111            mapping.try_push(0).map_err(refusal)?;
112        }
113        execution
114            .charge_work(core::mem::size_of_val(&*ordered) as u64)
115            .map_err(refusal)?;
116        for Request { ordinal, identity } in ordered.into_owned() {
117            execution.charge_work(1).map_err(refusal)?;
118            if let Some(previous) = pending.last()
119                && provider
120                    .compare(&previous.identity, &identity, execution)
121                    .map_err(DigestError::Callback)?
122                    == Ordering::Equal
123            {
124                mapping[ordinal] = pending.len() - 1;
125                continue;
126            }
127            pending.try_reserve(1).map_err(refusal)?;
128            let state = started(provider.start(&identity, execution))?;
129            pending
130                .try_push(Pending { identity, state })
131                .map_err(refusal)?;
132            mapping[ordinal] = pending.len() - 1;
133        }
134        Ok(Self {
135            provider,
136            pending: pending.into_owned(),
137            mapping: mapping.into_owned(),
138            field,
139            at,
140        })
141    }
142
143    /// Number of unique resolved digest states, without rescanning identities.
144    #[must_use]
145    pub fn len(&self) -> usize {
146        self.pending.len()
147    }
148    /// Whether no digest was requested.
149    #[must_use]
150    pub fn is_empty(&self) -> bool {
151        self.pending.is_empty()
152    }
153
154    /// Feeds the source once; outputs are finalized only after complete successful EOF.
155    ///
156    /// A source or callback error drops every unfinished state and partial
157    /// output. No partial digest is returned, and a source ignoring a refused
158    /// write cannot make subsequent callbacks run or turn refusal into success.
159    pub fn run<S: Source, W: Work>(
160        mut self,
161        source: &mut S,
162        execution: &W,
163    ) -> Result<DigestMemo<'a, P>, PassError<S::Error, P::Error>> {
164        let field = self.field;
165        let at = self.at;
166        let mut outputs = execution.vector();
167        outputs
168            .try_reserve_exact(self.pending.len())
169            .map_err(|reason| StreamError::Processing(Error::resource(reason, field, at)))?;
170        stream::run(source, &mut BatchSink(&mut self), execution, field, at)?;
171        self.finish(outputs, execution)
172            .map_err(StreamError::Callback)
173    }
174
175    pub(crate) fn feed<W: Work>(
176        &mut self,
177        bytes: &[u8],
178        execution: &W,
179    ) -> Result<(), DigestError<P::Error>> {
180        for pending in &mut self.pending {
181            execution.charge_work(1).map_err(|reason| {
182                DigestError::Processing(Error::resource(reason, self.field, self.at))
183            })?;
184            self.provider
185                .update(&mut pending.state, bytes, execution)
186                .map_err(DigestError::Callback)?;
187        }
188        Ok(())
189    }
190
191    pub(crate) fn finish<W: Work>(
192        self,
193        mut outputs: OwnedVec<DigestOutput<'a, P>, W>,
194        execution: &W,
195    ) -> Result<DigestMemo<'a, P>, DigestError<P::Error>> {
196        let refusal =
197            |reason| DigestError::Processing(Error::resource(reason, self.field, self.at));
198        execution
199            .charge_work(core::mem::size_of_val(&*self.pending) as u64)
200            .map_err(refusal)?;
201        for pending in self.pending {
202            execution.charge_work(1).map_err(refusal)?;
203            let output = self
204                .provider
205                .finish(pending.state, execution)
206                .map_err(DigestError::Callback)?;
207            if let Some(expected) = self.provider.width(&pending.identity) {
208                let actual = output.as_ref().len();
209                if actual != expected {
210                    return Err(DigestError::OutputWidth { expected, actual });
211                }
212            }
213            outputs
214                .try_push(Finalized::new(self.provider, pending.identity, output))
215                .map_err(refusal)?;
216        }
217        Ok(DigestMemo {
218            provider: self.provider,
219            outputs: outputs.into_owned(),
220            mapping: self.mapping,
221            field: self.field,
222            at: self.at,
223        })
224    }
225
226    pub(crate) fn provider(&self) -> &'a P {
227        self.provider
228    }
229
230    pub(crate) fn location(&self) -> (&'static str, u64) {
231        (self.field, self.at)
232    }
233}
234
235struct BatchSink<'s, 'a, P: Digest<'a>>(&'s mut DigestBatch<'a, P>);
236
237impl<'a, P: Digest<'a>> Sink for BatchSink<'_, 'a, P> {
238    type Error = DigestError<P::Error>;
239    fn write<W: Work>(&mut self, bytes: &[u8], execution: &W) -> Result<(), Self::Error> {
240        self.0.feed(bytes, execution)
241    }
242}
243
244/// Immutable completed outputs, sorted by the selected provider's full identity.
245pub struct DigestMemo<'a, P: Digest<'a>> {
246    provider: &'a P,
247    outputs: OwnedVec<DigestOutput<'a, P>>,
248    mapping: OwnedVec<usize>,
249    field: &'static str,
250    at: u64,
251}
252
253/// A separately completed input under an existing memo's resolved identity.
254///
255/// Signed attributes need their own digest output, while sharing the payload's
256/// already resolved identity. This result borrows that identity and owns its
257/// output; it cannot outlive the memo or become a fabricated digest identity.
258pub struct RelatedDigest<'m, 'a, P: Digest<'a>> {
259    provider: &'a P,
260    identity: &'m P::Identity,
261    output: P::Output,
262}
263
264impl<'a, P: Digest<'a>> RelatedDigest<'_, 'a, P> {
265    /// Typed finalized input with short borrows, without identity/output clones.
266    #[must_use]
267    pub fn prehash(&self) -> Prehash<'_, '_, P, P::Identity> {
268        Prehash::new(self.provider, self.identity, self.output.as_ref())
269    }
270
271    /// Borrows the provider's own output for its comparison, with provenance.
272    #[must_use]
273    pub fn finished(&self) -> Finished<'_, P, P::Identity, P::Output> {
274        Finished::new(self.provider, self.identity, &self.output)
275    }
276}
277
278struct RelatedSink<'s, 'a, P: Digest<'a>> {
279    provider: &'a P,
280    state: &'s mut P::State,
281}
282
283impl<'a, P: Digest<'a>> Sink for RelatedSink<'_, 'a, P> {
284    type Error = DigestError<P::Error>;
285
286    fn write<W: Work>(&mut self, bytes: &[u8], execution: &W) -> Result<(), Self::Error> {
287        self.provider
288            .update(self.state, bytes, execution)
289            .map_err(DigestError::Callback)
290    }
291}
292
293impl<'a, P: Digest<'a>> DigestMemo<'a, P> {
294    /// Completes another input with the original request's resolved identity.
295    ///
296    /// For CMS signed attributes, feed their original retagged ranges here;
297    /// their output is distinct from the payload output used by message-digest.
298    /// No resolver, Identity clone, state reconstruction or payload replay is
299    /// performed. An absent ordinal does not call the source or provider.
300    /// `field`/`at` locate the caller's enclosing original input structure.
301    pub fn digest_related<'m, S: Source, W: Work>(
302        &'m self,
303        ordinal: usize,
304        source: &mut S,
305        execution: &W,
306        field: &'static str,
307        at: u64,
308    ) -> PassResult<Option<RelatedDigest<'m, 'a, P>>, S::Error, P::Error> {
309        let Some(existing) = self
310            .for_request(ordinal, execution)
311            .map_err(StreamError::Callback)?
312        else {
313            return Ok(None);
314        };
315        let identity = existing.identity();
316        let mut state =
317            started(self.provider.start(identity, execution)).map_err(StreamError::Callback)?;
318        stream::run(
319            source,
320            &mut RelatedSink {
321                provider: self.provider,
322                state: &mut state,
323            },
324            execution,
325            field,
326            at,
327        )?;
328        let output = self
329            .provider
330            .finish(state, execution)
331            .map_err(|error| StreamError::Callback(DigestError::Callback(error)))?;
332        if let Some(expected) = self.provider.width(identity) {
333            let actual = output.as_ref().len();
334            if actual != expected {
335                return Err(StreamError::Callback(DigestError::OutputWidth {
336                    expected,
337                    actual,
338                }));
339            }
340        }
341        Ok(Some(RelatedDigest {
342            provider: self.provider,
343            identity,
344            output,
345        }))
346    }
347
348    /// Number of unique finalized outputs.
349    #[must_use]
350    pub fn len(&self) -> usize {
351        self.outputs.len()
352    }
353    /// Whether no completed output is held.
354    #[must_use]
355    pub fn is_empty(&self) -> bool {
356        self.outputs.is_empty()
357    }
358
359    /// Number of original requests, including identities that shared a state.
360    #[must_use]
361    pub fn request_count(&self) -> usize {
362        self.mapping.len()
363    }
364
365    /// The already finalized output for an original request ordinal.
366    ///
367    /// This constant-work link avoids resolving an algorithm again or cloning
368    /// its identity/state merely to find a shared output. Out-of-range ordinals
369    /// mean absence only after current-operation admission.
370    pub fn for_request<W: Work>(
371        &self,
372        ordinal: usize,
373        execution: &W,
374    ) -> DigestResult<Option<&DigestOutput<'a, P>>, P::Error> {
375        execution.charge_work(1).map_err(|reason| {
376            DigestError::Processing(Error::resource(reason, self.field, self.at))
377        })?;
378        let Some(index) = self.mapping.get(ordinal) else {
379            return Ok(None);
380        };
381        // Preparation assigned every ordinal to a pending state; successful
382        // finalization produced exactly one output for every such state.
383        debug_assert!(*index < self.outputs.len());
384        Ok(Some(&self.outputs[*index]))
385    }
386
387    /// Looks up a full resolved identity in the current execution account.
388    ///
389    /// Storage remains on its original account. No state/output clone, new hash
390    /// or finalization is performed. A resource refusal cannot become absence.
391    pub fn find<W: Work>(
392        &self,
393        identity: &P::Identity,
394        execution: &W,
395    ) -> DigestResult<Option<&DigestOutput<'a, P>>, P::Error> {
396        let refusal =
397            |reason| DigestError::Processing(Error::resource(reason, self.field, self.at));
398        execution.charge_work(1).map_err(refusal)?;
399        let mut from = 0;
400        let mut to = self.outputs.len();
401        while from < to {
402            execution.charge_work(1).map_err(refusal)?;
403            let middle = from + (to - from) / 2;
404            let held = &self.outputs[middle];
405            match self
406                .provider
407                .compare(held.identity(), identity, execution)
408                .map_err(DigestError::Callback)?
409            {
410                Ordering::Equal => return Ok(Some(held)),
411                Ordering::Less => from = middle + 1,
412                Ordering::Greater => to = middle,
413            }
414        }
415        Ok(None)
416    }
417}
structured_cms/digest.rs

structured_cms/
digest.rs
