sark_core/body/
mod.rs

1use core::{fmt, marker, num::NonZero};
2
3use o3::mem::Rejected;
4use sark_http::{
5    owner::handoff::{
6        self,
7        input::{Borrowed, Bound},
8    },
9    payload::owned,
10    request::body,
11};
12
13pub mod admit;
14pub mod buffered;
15pub mod credit;
16pub mod flow;
17pub mod ingress;
18pub mod input;
19pub mod stream;
20
21pub const DEFAULT_MAX_REQUEST_CHUNKS: usize = 4 * 1024;
22pub const DEFAULT_MAX_CONNECTION_CHUNKS: usize = 16 * 1024;
23pub const DEFAULT_BUILD_STEP_BUDGET: NonZero<usize> =
24    NonZero::<usize>::MIN.saturating_add(4 * 1024 - 1);
25
26/// Projection keeps the exact final owner or returns the original source on rejection.
27pub type Projection<T, O, Source> = Result<(T, O), Rejected<Source, body::Error>>;
28
29pub trait Project: Sized {
30    type View<'req, 'd>: Into<body::Ref<'req, 'd>> + Copy
31    where
32        'd: 'req;
33
34    const RETENTION: admit::Retention;
35
36    fn empty<'req, 'd: 'req>() -> Scoped<'req, 'd, Self>;
37
38    fn try_project<'a, 'd, S, V, T>(
39        body: &'a mut Body<S>,
40        value: V,
41        project: impl for<'view> FnOnce(Scoped<'view, 'd, Self, V>) -> T,
42    ) -> Result<T, Rejected<V, body::Error>>
43    where
44        S: body::Source<'d>,
45        'd: 'a,
46    {
47        let input = body.input();
48        Self::try_input(input, value, |scope| {
49            let (value, body) = scope.into_parts();
50            project(Scoped::new(body.value, value))
51        })
52        .map(|(output, ())| output)
53        .map_err(|failure| {
54            let ((_, value), error) = failure.into_parts();
55            Rejected::new(value, error)
56        })
57    }
58
59    /// Projects one affine source and returns its exact final owner.
60    fn try_input<'d, I, V, T>(
61        input: Input<I>,
62        value: V,
63        project: impl for<'view> FnOnce(
64            Scoped<'view, 'd, Self, V, handoff::Slot<'view, 'd, (), Bound<'view, 'd, I::Proof>>>,
65        ) -> T,
66    ) -> Projection<T, I::Owner, (Input<I>, V)>
67    where
68        I: handoff::Project<'d>;
69}
70
71pub enum Contiguous {}
72
73impl Project for Contiguous {
74    type View<'req, 'd>
75        = body::View<'req, 'd>
76    where
77        'd: 'req;
78
79    const RETENTION: admit::Retention = admit::Retention::Retained;
80
81    fn empty<'req, 'd: 'req>() -> Scoped<'req, 'd, Self> {
82        Scoped::<Self>::new(body::View::Slice(&[]), ())
83    }
84
85    fn try_input<'d, I, V, T>(
86        input: Input<I>,
87        value: V,
88        project: impl for<'view> FnOnce(
89            Scoped<'view, 'd, Self, V, handoff::Slot<'view, 'd, (), Bound<'view, 'd, I::Proof>>>,
90        ) -> T,
91    ) -> Projection<T, I::Owner, (Input<I>, V)>
92    where
93        I: handoff::Project<'d>,
94    {
95        let Input { source, len } = input;
96        source
97            .contiguous(value, |value, view, source| {
98                project(Scoped::from_source(view, value, source))
99            })
100            .map_err(|(source, value, error)| Rejected::new((Input { source, len }, value), error))
101    }
102}
103
104pub enum Discarded {}
105
106impl Project for Discarded {
107    type View<'req, 'd>
108        = Len
109    where
110        'd: 'req;
111
112    const RETENTION: admit::Retention = admit::Retention::Discarded;
113
114    fn empty<'req, 'd: 'req>() -> Scoped<'req, 'd, Self> {
115        Scoped::<Self>::new(Len(0), ())
116    }
117
118    fn try_input<'d, I, V, T>(
119        input: Input<I>,
120        value: V,
121        project: impl for<'view> FnOnce(
122            Scoped<'view, 'd, Self, V, handoff::Slot<'view, 'd, (), Bound<'view, 'd, I::Proof>>>,
123        ) -> T,
124    ) -> Projection<T, I::Owner, (Input<I>, V)>
125    where
126        I: handoff::Project<'d>,
127    {
128        let Input { source, len } = input;
129        Ok(source.discarded(value, |value, source| {
130            project(Scoped::from_source(Len(len), value, source))
131        }))
132    }
133}
134
135pub enum Segmented {}
136
137impl Project for Segmented {
138    type View<'req, 'd>
139        = body::Ref<'req, 'd>
140    where
141        'd: 'req;
142
143    const RETENTION: admit::Retention = admit::Retention::Retained;
144
145    fn empty<'req, 'd: 'req>() -> Scoped<'req, 'd, Self> {
146        Scoped::<Self>::new(body::Ref::from_slice(&[]), ())
147    }
148
149    fn try_input<'d, I, V, T>(
150        input: Input<I>,
151        value: V,
152        project: impl for<'view> FnOnce(
153            Scoped<'view, 'd, Self, V, handoff::Slot<'view, 'd, (), Bound<'view, 'd, I::Proof>>>,
154        ) -> T,
155    ) -> Projection<T, I::Owner, (Input<I>, V)>
156    where
157        I: handoff::Project<'d>,
158    {
159        Ok(input.source.segmented(value, |value, view, source| {
160            project(Scoped::from_source(view, value, source))
161        }))
162    }
163}
164
165/// Projection coupled to its storage lifetime and the owned call context.
166///
167#[repr(C)]
168pub struct Scoped<'req, 'd: 'req, P: Project, V: 'req = (), Q = ()> {
169    value: P::View<'req, 'd>,
170    context: V,
171    source: Q,
172    proof: marker::PhantomData<(&'req &'d (), &'req V)>,
173}
174
175impl<'req, 'd: 'req, P: Project, V: 'req> Scoped<'req, 'd, P, V> {
176    const fn new(value: P::View<'req, 'd>, context: V) -> Self {
177        Self::from_source(value, context, ())
178    }
179}
180
181impl<'req, 'd: 'req, P: Project, V: 'req, Q> Scoped<'req, 'd, P, V, Q> {
182    const fn from_source(value: P::View<'req, 'd>, context: V, source: Q) -> Self {
183        Self {
184            value,
185            context,
186            source,
187            proof: marker::PhantomData,
188        }
189    }
190
191    pub fn into_parts(self) -> (V, Scoped<'req, 'd, P>) {
192        (self.context, Scoped::new(self.value, ()))
193    }
194
195    pub fn into_source(self) -> (V, Scoped<'req, 'd, P>, Q) {
196        (self.context, Scoped::new(self.value, ()), self.source)
197    }
198}
199
200impl<'req, 'd: 'req, P: Project> Scoped<'req, 'd, P> {
201    pub fn into_inner(self) -> P::View<'req, 'd> {
202        self.value
203    }
204
205    pub fn body(&self) -> body::Ref<'req, 'd> {
206        self.value.into()
207    }
208}
209
210const _: () = {
211    use core::mem::{align_of, size_of};
212
213    assert!(size_of::<Scoped<'static, 'static, Discarded>>() == size_of::<Len>());
214    assert!(
215        size_of::<Scoped<'static, 'static, Contiguous>>()
216            == size_of::<body::View<'static, 'static>>()
217    );
218    assert!(
219        size_of::<Option<Scoped<'static, 'static, Contiguous>>>()
220            == size_of::<body::View<'static, 'static>>()
221    );
222    assert!(
223        size_of::<Scoped<'static, 'static, Segmented>>()
224            == size_of::<body::Ref<'static, 'static>>()
225    );
226    assert!(
227        align_of::<Scoped<'static, 'static, Segmented>>()
228            == align_of::<body::Ref<'static, 'static>>()
229    );
230};
231
232#[derive(Clone, Copy)]
233#[repr(transparent)]
234pub struct Len(usize);
235
236impl<'req, 'd> From<Len> for body::Ref<'req, 'd> {
237    fn from(length: Len) -> Self {
238        Self::discarded(length.0)
239    }
240}
241
242/// Completed logical payload coupled to its transport storage.
243#[derive(Debug)]
244pub struct Body<S> {
245    source: S,
246    len: usize,
247}
248
249impl<S> Body<S> {
250    pub fn try_new<'d>(
251        proof: admit::Track<admit::Finished>,
252        source: S,
253    ) -> Result<Self, Rejected<(admit::Track<admit::Finished>, S), Error>>
254    where
255        S: body::Source<'d>,
256    {
257        let len = proof.len();
258        let seen = source.body().into().len();
259        let valid = match proof.retention() {
260            admit::Retention::Discarded => seen == 0,
261            admit::Retention::Retained => seen == len,
262        };
263        if !valid {
264            return Err(Rejected::new((proof, source), Error::SegmentMismatch));
265        }
266        Ok(Self { source, len })
267    }
268
269    pub const fn len(&self) -> usize {
270        self.len
271    }
272
273    pub const fn is_empty(&self) -> bool {
274        self.len == 0
275    }
276
277    pub fn into_source(self) -> S {
278        self.source
279    }
280
281    pub fn input(&mut self) -> Input<Borrowed<'_, S>> {
282        Input {
283            source: Borrowed::new(&mut self.source),
284            len: self.len,
285        }
286    }
287}
288
289/// One consumed projection choice retaining the completed logical body length.
290pub struct Input<I> {
291    pub(crate) source: I,
292    pub(crate) len: usize,
293}
294
295impl<'d> Body<owned::Segments<'d>> {
296    pub fn into_body(self) -> owned::Segments<'d> {
297        self.source
298    }
299
300    #[doc(hidden)]
301    pub fn into_parts(self) -> (owned::Segments<'d>, usize) {
302        (self.source, self.len)
303    }
304}
305
306const _: () = assert!(core::mem::size_of::<Len>() == core::mem::size_of::<usize>());
307
308/// Linear successor of one cooperative construction turn.
309pub enum Step<B, O, E> {
310    NeedInput {
311        build: B,
312        released: credit::Amount,
313    },
314    Pending {
315        build: B,
316        released: credit::Amount,
317    },
318    Ready {
319        output: O,
320        released: credit::Amount,
321    },
322    Failed {
323        build: B,
324        error: E,
325        released: credit::Amount,
326    },
327}
328
329impl<B, O, E> Step<B, O, E> {
330    pub fn map<C, T, F>(
331        self,
332        build: impl FnOnce(B) -> C,
333        output: impl FnOnce(O) -> T,
334        error: impl FnOnce(E) -> F,
335    ) -> Step<C, T, F> {
336        match self {
337            Self::NeedInput {
338                build: next,
339                released,
340            } => Step::NeedInput {
341                build: build(next),
342                released,
343            },
344            Self::Pending {
345                build: next,
346                released,
347            } => Step::Pending {
348                build: build(next),
349                released,
350            },
351            Self::Ready {
352                output: value,
353                released,
354            } => Step::Ready {
355                output: output(value),
356                released,
357            },
358            Self::Failed {
359                build: next,
360                error: cause,
361                released,
362            } => Step::Failed {
363                build: build(next),
364                error: error(cause),
365                released,
366            },
367        }
368    }
369}
370
371/// Result of one scheduled cooperative build turn.
372pub enum BuildOutcome<K, B, E> {
373    Idle,
374    Pending,
375    Ready { key: K, build: B },
376    Failed { key: K, error: E },
377}
378
379/// Builder ownership after one retained segment transfer attempt.
380pub enum Transfer<B, E> {
381    Accepted(B),
382    Rejected { build: B, error: E },
383}
384
385impl<B, E> Transfer<B, E> {
386    /// Moves the successor builder while preserving the exact failure.
387    pub fn map<C>(self, map: impl FnOnce(B) -> C) -> Transfer<C, E> {
388        match self {
389            Self::Accepted(build) => Transfer::Accepted(map(build)),
390            Self::Rejected { build, error } => Transfer::Rejected {
391                build: map(build),
392                error,
393            },
394        }
395    }
396
397    pub fn map_err<F>(self, map: impl FnOnce(E) -> F) -> Transfer<B, F> {
398        match self {
399            Self::Accepted(build) => Transfer::Accepted(build),
400            Self::Rejected { build, error } => Transfer::Rejected {
401                build,
402                error: map(error),
403            },
404        }
405    }
406}
407
408/// Route-erased cooperative request builder.
409pub trait Build<'d>: Sized {
410    type Output;
411    type StepError: fmt::Debug;
412
413    fn finish_input(self, len: usize) -> Self;
414
415    fn step(self, budget: NonZero<usize>) -> Step<Self, Self::Output, Self::StepError>;
416}
417
418/// Capability to accept retained logical payload segments.
419pub trait Push<'d>: Build<'d> {
420    type Error;
421
422    /// Borrows the exact local failure without consuming retained input.
423    fn observe<C: Copy, S: Copy, O: crate::life::Observer<C, S>>(
424        error: &Self::Error,
425        observer: &mut O,
426    );
427
428    fn push(self, part: owned::Segment<'d>) -> Transfer<Self, Self::Error>;
429}
430
431#[derive(Clone, Copy, PartialEq, Eq, Debug)]
432pub enum Error {
433    TooLarge { actual: usize, limit: usize },
434    LengthMismatch { actual: usize, declared: usize },
435    Overflow { current: usize, added: usize },
436    InputIncomplete,
437    InputFinished,
438    SegmentMismatch,
439    Credit(credit::Error),
440}
441
442const _: () = {
443    assert!(core::mem::size_of::<Error>() == 3 * core::mem::size_of::<usize>());
444    assert!(core::mem::size_of::<Result<(), Error>>() == 3 * core::mem::size_of::<usize>());
445};
sark_core/body/mod.rs

sark_core/body/
mod.rs
