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

structured_cms/
digest.rs
