proxima_primitives/pipe/
capabilities.rs

1use core::future::Future;
2use core::time::Duration;
3
4#[cfg(feature = "alloc")]
5use alloc::borrow::Cow;
6
7use bytes::Bytes;
8use proxima_core::ProximaError;
9
10// ── payload capability traits ─────────────────────────────────────────────────
11//
12// These traits express what a payload must provide to generic middleware
13// skeletons. The skeletons (Retry, etc.) are generic over the inner pipe;
14// these traits are the injected seams. HTTP implements them on
15// Request/Response in proxima-pipe (orphan rule). Non-HTTP payloads implement
16// them in their own crate.
17//
18// A pass/reject decision is NOT one of these seams: `In -> Result<In, Err>`
19// is a pipe (`Ok` = admit, `Err` = reject), composed via `and_then` — not a
20// `decide(&In) -> bool` capability with companions to carry the item and the
21// rejection back (see `pipe/fan_in.rs`'s `FanInStrategy` doc for the pipe-vs-
22// strategy line this used to violate).
23
24/// An input that can be re-fed to the inner pipe for each retry attempt. HTTP
25/// forks the body stream through a `Replay` (bounded by `replay_cap_bytes`); a
26/// cloneable payload (an event) clones. The skeleton owns neither mechanism.
27pub trait Replayable: Sized {
28    type Source: Send;
29    fn fork(self, replay_cap_bytes: usize) -> (Self, Self::Source);
30    fn replay(source: &Self::Source) -> Result<Self, ProximaError>;
31}
32
33/// Whether an input is safe to retry (HTTP: idempotent method).
34pub trait Idempotent {
35    fn is_idempotent(&self) -> bool;
36}
37
38/// Outcome classification for retry (HTTP: status codes). A statusless outcome
39/// (a plain event) returns `None` and is governed only by error retry.
40pub trait Retryable {
41    fn retry_status(&self) -> Option<u16>;
42    fn is_success(&self) -> bool;
43}
44
45/// The monotonic-clock seam the timer-driven executors (`Retry`, and later
46/// `Timeout`/`Hedge`) are generic over. `Delay` is an associated future, so the
47/// executor holds it inline in its state machine — no boxing, no alloc, no_std.
48///
49/// Production impls wrap `proxima-time` (`crate::pipe::clock::TimeClock`,
50/// whose `Sleep` is a concrete `Delay`); deterministic tests reach for
51/// `crate::pipe::clock::testing::{MockClock, RecordingClock}` — the canonical
52/// doubles, not a bespoke fake per call site.
53///
54/// Two other `Clock` traits exist in the workspace, and neither collapses
55/// into this one: `prime::core::timer::Clock` is a lower-level,
56/// resolution-agnostic tick source (NOT nanosecond-pinned — `TimerWheel`
57/// needs no `Delay` and a coarser unit) for the wheel data structure this
58/// trait's production `Sleep` is ultimately built over; `proxima_telemetry
59/// ::clock::Clock` is nanosecond-pinned like this one but deliberately
60/// object-safe (no associated type), because the per-core `Recorder` erases
61/// its clock behind `Arc<dyn Clock + Send + Sync>` — a shape this trait's
62/// `Delay` associated type cannot support without boxing.
63pub trait Clock {
64    /// A concrete, `Sized` future that resolves once `dur` has elapsed.
65    type Delay: Future<Output = ()>;
66    /// Monotonic nanoseconds; feeds `RetryController`'s deadline arithmetic.
67    fn now_nanos(&self) -> u64;
68    fn delay(&self, dur: Duration) -> Self::Delay;
69}
70
71/// A value that can apply a sequence of typed ops to itself.
72///
73/// Implement this on any input or output type to make Transform work over it.
74/// HTTP: `Request<Bytes>: ApplyOps<RequestOp>`, `Response<Bytes>: ApplyOps<ResponseOp>`.
75pub trait ApplyOps<Op>: Sized {
76    fn apply(self, ops: &[Op]) -> Self;
77}
78
79/// A value that exposes a mutable byte buffer for fuzzing.
80///
81/// Implement this on any payload to make `Mutation` / `MutateOp` work over it.
82/// HTTP impls it on `Request` and `Response` (their `body`); a non-HTTP payload
83/// implements it over its own bytes.
84pub trait BytePayload: Sized {
85    /// Replace the payload's bytes with `bytes`.
86    fn set_bytes(&mut self, bytes: Bytes);
87    /// Borrow the payload's current bytes.
88    fn bytes(&self) -> &[u8];
89}
90
91/// The action taken when a rate-limit bucket is exhausted.
92#[derive(Debug, Clone, Copy)]
93pub enum ExceededAction {
94    Reject429 { retry_after_ms: u64 },
95}
96
97/// Extracts a rate-limit bucket key from an input and constructs the rejection
98/// output when the bucket is exhausted.
99///
100/// Implement this on any input type to make `RateLimit` work over it.
101/// HTTP: `Request<Bytes>: KeyOf<KeyExtractor>` with `type Rejection = Response<Bytes>`.
102///
103/// Additive at the alloc tier: the key is borrowed-or-owned (`Cow<[u8]>`), so
104/// this seam needs `alloc`. The rest of the vocabulary is core-tier.
105#[cfg(feature = "alloc")]
106pub trait KeyOf<Extractor>: Sized {
107    /// The output value produced when this input is rate-limited.
108    type Rejection;
109
110    /// The bucket key for this input given the extractor config.
111    fn rate_key<'a>(&'a self, extractor: &'a Extractor) -> Cow<'a, [u8]>;
112
113    /// Constructs the rejection output. Called only when the bucket is empty.
114    fn build_rejection(action: &ExceededAction) -> Self::Rejection;
115}
116
117/// Outcome of a check: either admitted (pass through to inner) or rejected
118/// (short-circuit with a pre-built output).
119pub enum CheckOutcome<In, Out> {
120    Pass(In),
121    Reject(Out),
122}
123
124/// A value that can check itself against an op and decide pass/reject.
125///
126/// Implement this on any input type to make Validate work over it.
127/// HTTP: `Request<Bytes>: Checkable<ValidateOp>` with `Out = Response<Bytes>`.
128pub trait Checkable<Op>: Sized {
129    type Out;
130
131    fn check(
132        self,
133        op: &Op,
134    ) -> impl Future<Output = Result<CheckOutcome<Self, Self::Out>, ProximaError>> + Send;
135}
proxima_primitives/pipe/capabilities.rs

proxima_primitives/pipe/
capabilities.rs
