arcly_http_core/web/
boundary.rs

1//! Boundary adapter: converts an axum `Request` into the opaque
2//! `RequestContext` and dispatches to the macro-generated thunk.
3//!
4//! This is the *only* place in the framework where axum extraction primitives
5//! are touched outside of the launch path. `assemble_context` is the single
6//! request→context pipeline — plugin routes (`web::plugin_routes`) reuse it,
7//! so body limits, trace propagation, and credential extraction can never
8//! drift between entry points.
9
10use axum::body::Body;
11use axum::extract::{RawPathParams, Request, State};
12use axum::http::request::Parts;
13use axum::response::Response;
14use axum::routing::{on, MethodFilter, MethodRouter};
15use smallvec::SmallVec;
16use smol_str::SmolStr;
17
18use crate::core::engine::{FrozenDiContainer, RouteDescriptor, RouteSpec};
19use crate::observability::lean_telemetry::on_request_start;
20use crate::web::context::RequestContext;
21
22/// Maximum request body size — protects against memory exhaustion.
23/// Default 8 MiB; configured via `LaunchConfig::max_body_bytes` and stored
24/// in the frozen DI container, so the per-request cost is one lock-free
25/// frozen-map probe and concurrent apps (tests!) can't clobber each other.
26#[doc(hidden)]
27pub struct BodyLimit(pub usize);
28
29const DEFAULT_MAX_BODY: usize = 8 * 1024 * 1024;
30
31/// Pre-body request filter, registered by plugins via
32/// `ArclyPluginContext::register_boundary_filter`.
33///
34/// Runs on every request *before the body is buffered* — the cheap
35/// early-reject point. `Break(resp)` short-circuits the request without
36/// paying for body read, auth extraction, or context assembly. Filters must
37/// be synchronous and cheap (header checks, atomic counters); anything
38/// needing async or the `RequestContext` belongs in an `Interceptor`.
39pub trait BoundaryFilter: Send + Sync + 'static {
40    /// `parts` is [`crate::http::RequestParts`] — implementors never need to
41    /// name `axum`.
42    fn before_body(
43        &'static self,
44        parts: &crate::http::RequestParts,
45    ) -> std::ops::ControlFlow<Response>;
46}
47
48/// Run registered boundary filters in order; first `Break` wins.
49#[inline]
50pub(crate) fn run_boundary_filters(
51    filters: &'static [&'static dyn BoundaryFilter],
52    parts: &Parts,
53) -> Option<Response> {
54    for f in filters {
55        if let std::ops::ControlFlow::Break(resp) = f.before_body(parts) {
56            return Some(resp);
57        }
58    }
59    None
60}
61
62/// RAII guard that tracks the in-flight request count in the Prometheus gauge.
63/// Pairs with `lean_telemetry::RequestGuard` which tracks the same count in the
64/// raw atomic used by health endpoints.
65pub(crate) struct InFlightGuard;
66impl InFlightGuard {
67    #[inline]
68    pub(crate) fn new() -> Self {
69        metrics::gauge!("http_requests_in_flight").increment(1.0);
70        Self
71    }
72}
73impl Drop for InFlightGuard {
74    #[inline]
75    fn drop(&mut self) {
76        metrics::gauge!("http_requests_in_flight").decrement(1.0);
77    }
78}
79
80/// The single request→`RequestContext` pipeline:
81/// body read (capped) → W3C trace propagation → credential extraction
82/// (Bearer / cookie / session) → context construction.
83///
84/// Every HTTP entry point — macro routes here, plugin routes in
85/// `web::plugin_routes` — goes through this function.
86///
87/// Returns `Err(413)` when the body exceeds the cap. Truncating silently
88/// (the previous behaviour) handed handlers an *empty* body for oversized
89/// requests — a correctness hazard, not a defence.
90#[doc(hidden)]
91// The `Err` is an axum `Response` (the early-reject path); boxing it would just
92// re-allocate on the boundary hot path for no benefit. Newer clippy flags its
93// size under `-D warnings`; the allocation trade-off is intentional here.
94#[allow(clippy::result_large_err)]
95pub async fn assemble_context(
96    parts: Parts,
97    body: Body,
98    params: SmallVec<[(SmolStr, SmolStr); 4]>,
99    container: std::sync::Arc<FrozenDiContainer>,
100    route_pattern: &'static str,
101    route_spec: Option<&'static RouteSpec>,
102) -> Result<RequestContext, Response> {
103    let cap = container
104        .try_get::<BodyLimit>()
105        .map(|b| b.0)
106        .unwrap_or(DEFAULT_MAX_BODY);
107    let bytes = match axum::body::to_bytes(body, cap).await {
108        Ok(b) => b,
109        Err(_) => {
110            metrics::counter!("http_requests_body_too_large_total").increment(1);
111            return Err(Response::builder()
112                .status(413)
113                .body(Body::from("request body exceeds the configured limit"))
114                .expect("static 413 response"));
115        }
116    };
117
118    // One shared extraction for trace + tenant + credentials — identical to
119    // what WebSocket handshakes and the consumer mesh run (`pipeline`).
120    let provenance = crate::pipeline::Provenance::from_headers(&parts.headers, &container).await;
121
122    // Peer IP: prefer a trusted `x-forwarded-for` (first hop) for proxied
123    // deployments, else the connection's peer address (present when the server
124    // was started with connection info via `into_make_service_with_connect_info`).
125    let client_ip = parts
126        .headers
127        .get("x-forwarded-for")
128        .and_then(|v| v.to_str().ok())
129        .and_then(|s| s.split(',').next())
130        .and_then(|s| s.trim().parse::<std::net::IpAddr>().ok())
131        .or_else(|| {
132            parts
133                .extensions
134                .get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
135                .map(|ci| ci.0.ip())
136        });
137
138    Ok(RequestContext::__new(
139        parts.method,
140        SmolStr::new(parts.uri.path()),
141        SmolStr::new(parts.uri.query().unwrap_or("")),
142        params,
143        parts.headers,
144        bytes,
145        provenance.trace.trace_id,
146        provenance.trace.span_id,
147        provenance.trace.parent_span_id,
148        container,
149        route_pattern,
150        route_spec,
151    )
152    .__with_claims(provenance.claims)
153    .__with_session(provenance.session)
154    .__with_tenant(provenance.tenant)
155    .__with_client_ip(client_ip))
156}
157
158/// The ONE entry sequence every HTTP mount point runs — macro routes,
159/// plugin routes, and dynamic `/_plugins` routes all call this instead of
160/// keeping their own copy (three copies previously meant fixes like the
161/// 413 body-cap landed three times):
162///
163/// boundary filters → context assembly (body cap → provenance) →
164/// telemetry guards → interceptor chain.
165#[allow(clippy::too_many_arguments)] // the full entry contract, one call site per mount kind
166pub(crate) async fn run_entry(
167    parts: Parts,
168    body: Body,
169    params: SmallVec<[(SmolStr, SmolStr); 4]>,
170    container: std::sync::Arc<FrozenDiContainer>,
171    route_pattern: &'static str,
172    route_spec: Option<&'static RouteSpec>,
173    filters: &'static [&'static dyn BoundaryFilter],
174    chain: &std::sync::Arc<
175        dyn Fn(RequestContext) -> futures::future::BoxFuture<'static, Response> + Send + Sync,
176    >,
177) -> Response {
178    if let Some(reject) = run_boundary_filters(filters, &parts) {
179        return reject;
180    }
181    let ctx =
182        match assemble_context(parts, body, params, container, route_pattern, route_spec).await {
183            Ok(ctx) => ctx,
184            Err(reject) => return reject,
185        };
186
187    let _telemetry = on_request_start();
188    let _in_flight = InFlightGuard::new();
189    chain(ctx).await
190}
191
192/// Wrap a macro-generated route descriptor in an axum `MethodRouter`.
193/// The HTTP method filter is baked in so the caller can drop the result
194/// straight into `Router::route`.
195/// Plugin-registered global interceptors compose as the outermost layers
196/// around the handler (which already carries any `#[UseInterceptors]` chain).
197pub fn adapt(
198    rt: &'static RouteDescriptor,
199    globals: &'static [&'static dyn crate::web::interceptors::Interceptor],
200    filters: &'static [&'static dyn BoundaryFilter],
201) -> MethodRouter<std::sync::Arc<FrozenDiContainer>> {
202    let filter = MethodFilter::try_from(axum::http::Method::from(rt.method))
203        .expect("Arcly: unsupported HTTP method");
204
205    let chain = crate::web::interceptors::compose_chain(
206        globals,
207        std::sync::Arc::new(move |ctx| (rt.handler)(ctx)),
208    );
209
210    let handler = move |State(container): State<std::sync::Arc<FrozenDiContainer>>,
211                        raw_params: RawPathParams,
212                        req: Request| {
213        let chain = chain.clone();
214        async move {
215            let params: SmallVec<[(SmolStr, SmolStr); 4]> = raw_params
216                .iter()
217                .map(|(k, v)| (SmolStr::new(k), SmolStr::new(v)))
218                .collect();
219
220            let (parts, body) = req.into_parts();
221            let resp: Response = run_entry(
222                parts,
223                body,
224                params,
225                container,
226                rt.path,
227                Some(rt.spec),
228                filters,
229                &chain,
230            )
231            .await;
232
233            let (mut p, b) = resp.into_parts();
234            // RFC 8594: announce deprecation + sunset date on versioned routes
235            // marked #[Deprecated(sunset = "…")].
236            if !rt.spec.sunset.is_empty() {
237                p.headers
238                    .insert("deprecation", axum::http::HeaderValue::from_static("true"));
239                if let Ok(v) = axum::http::HeaderValue::from_str(rt.spec.sunset) {
240                    p.headers.insert("sunset", v);
241                }
242            }
243            Response::from_parts(p, Body::new(b))
244        }
245    };
246
247    on(filter, handler)
248}
arcly_http_core/web/boundary.rs

arcly_http_core/web/
boundary.rs
