churust_core/tower.rs
1//! Run a `tower::Service` as Churust [`Middleware`].
2//!
3//! Requires the `tower` feature.
4//!
5//! Churust has its own middleware model — [`Middleware`] + [`Next`], an onion
6//! where each layer awaits the next — and that stays the native way to write
7//! one. This adapter exists so an application does not have to reimplement
8//! something the ecosystem already ships as a `Layer`: response compression,
9//! structured tracing, request-id propagation, header manipulation, request
10//! validation, metrics.
11//!
12//! The adapter is deliberately **one-directional**. Churust middleware is not
13//! exposed as a tower `Service`; only the reverse.
14//!
15//! # What this does not do
16//!
17//! **Backpressure is not propagated.** tower's `poll_ready` contract — a
18//! service signalling that it cannot accept work yet — has no counterpart in
19//! `Middleware::handle`, which is handed a request that has already been
20//! accepted. The adapter drives `poll_ready` to readiness before calling, so a
21//! layer that uses it for load-shedding will *wait* rather than shed. Bound
22//! concurrency with `max_connections` instead.
23//!
24//! **The inner service must be called from the same task.** The request's
25//! continuation is held in a task-local, so a layer that hands the request to a
26//! worker task (`tower::buffer::Buffer`, `spawn_ready`, and anything built on
27//! them) cannot reach it and gets an error rather than a response.
28//!
29//! **The request method is not written back.** Routing has already happened by
30//! the time a layer runs, so a layer that rewrites the method would leave the
31//! call disagreeing with the handler it is about to reach. Header and URI
32//! rewrites *are* carried back.
33//!
34//! **A `Service` error becomes a `500`.** This keeps Churust's invariant that a
35//! response is always produced, without importing axum's tax for it: axum
36//! requires every composed service to have `Error = Infallible`, which is a
37//! genuine compile-time guarantee but forces a `HandleErrorLayer` around any
38//! fallible layer — adding a timeout to a route literally requires wrapping it.
39//! Here the error is absorbed through [`IntoError`](crate::IntoError), so the
40//! guarantee holds and the ergonomics stay Churust's.
41//!
42//! ```no_run
43//! use churust_core::{Churust, Call, tower::TowerMiddleware};
44//! # use std::sync::Arc;
45//! # fn some_layer() -> tower_layer::Identity { tower_layer::Identity::new() }
46//! let app = Churust::server()
47//! .install(TowerMiddleware::new(some_layer()))
48//! .routing(|r| { r.get("/", |_c: Call| async { "hi" }); })
49//! .build();
50//! ```
51
52use crate::app::{AppBuilder, Plugin};
53use crate::body::Body;
54use crate::call::Call;
55use crate::pipeline::{Middleware, Next, Phase};
56use crate::response::{IntoResponse, Response};
57use async_trait::async_trait;
58use std::sync::Arc;
59use tower_service::Service;
60
61tokio::task_local! {
62 /// The call and continuation for the request currently passing through the
63 /// adapter.
64 ///
65 /// A task-local rather than a request extension: `http::Extensions` requires
66 /// `Clone`, and [`Next`] is a one-shot continuation that must not be
67 /// cloneable. A task-local also lets the layered service be built **once**,
68 /// at install time, so a stateful layer keeps its state across requests
69 /// instead of being rebuilt per call.
70 static IN_FLIGHT: std::cell::RefCell<Option<(Call, Next)>>;
71}
72
73/// The innermost service: hands the request back to the Churust pipeline.
74///
75/// Everything a `Layer` wraps around this eventually calls it, which is the
76/// point at which the rest of the onion — including the route handler — runs.
77#[derive(Clone, Copy, Default)]
78pub struct NextService;
79
80impl Service<http::Request<Body>> for NextService {
81 type Response = http::Response<Body>;
82 type Error = crate::Error;
83 type Future = std::pin::Pin<
84 Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
85 >;
86
87 fn poll_ready(
88 &mut self,
89 _cx: &mut std::task::Context<'_>,
90 ) -> std::task::Poll<Result<(), Self::Error>> {
91 // Always ready: the request has already been accepted by the time any
92 // of this runs, so there is nothing left to refuse.
93 std::task::Poll::Ready(Ok(()))
94 }
95
96 fn call(&mut self, req: http::Request<Body>) -> Self::Future {
97 // Recover the pair stashed by the adapter. Taking it here means a layer
98 // that calls the inner service twice gets a clear error rather than a
99 // silently duplicated request.
100 // `try_with`, not `with`: `with` panics when the task-local is unset,
101 // which happens whenever a layer drives the inner service from a
102 // *different* task (tower::buffer::Buffer, spawn_ready, anything that
103 // hands the request to a worker). A panic there is caught upstream and
104 // rendered as a bare 500 with no hint of the cause; an error at least
105 // names the restriction.
106 let taken = match IN_FLIGHT.try_with(|slot| slot.borrow_mut().take()) {
107 Ok(taken) => taken,
108 Err(_) => {
109 return Box::pin(async {
110 Err(crate::Error::internal(
111 "this tower layer drove the inner service from another task; \
112 the Churust adapter cannot cross task boundaries",
113 ))
114 })
115 }
116 };
117 Box::pin(async move {
118 let Some((mut call, next)) = taken else {
119 return Err(crate::Error::internal(
120 "tower layer called the inner service more than once per request",
121 ));
122 };
123 // Header *and* URI changes made by the layers on the way in are the
124 // reason this adapter is worth having, so carry both onto the call.
125 // Copying only the headers meant a path-rewriting layer silently did
126 // nothing. The method is deliberately not written back: routing has
127 // already happened by this point, so changing it here would leave
128 // the call disagreeing with the handler it is about to reach.
129 *call.headers_mut() = req.headers().clone();
130 call.set_uri(req.uri().clone());
131 let res = next.run(call).await;
132 let mut out = http::Response::builder().status(res.status);
133 if let Some(h) = out.headers_mut() {
134 *h = res.headers;
135 }
136 Ok(out.body(res.body).expect("response build is infallible"))
137 })
138 }
139}
140
141/// Churust [`Middleware`] wrapping a tower `Layer` applied over [`NextService`].
142///
143/// Build one with [`TowerMiddleware::new`] and install it like any other
144/// plugin. The layered service is constructed once, so layer state persists
145/// across requests.
146pub struct TowerMiddleware<S> {
147 /// `Service::call` needs `&mut self`, but `Middleware::handle` takes
148 /// `&self` and runs concurrently. The mutex is held only long enough to
149 /// clone the service, which is what tower expects callers to do.
150 service: std::sync::Mutex<S>,
151 phase: Phase,
152}
153
154impl<S> TowerMiddleware<S> {
155 /// Apply `layer` over the Churust pipeline.
156 ///
157 /// Runs in [`Phase::Plugins`]; use [`TowerMiddleware::in_phase`] to place
158 /// it earlier or later in the onion.
159 pub fn new<L>(layer: L) -> Self
160 where
161 L: tower_layer::Layer<NextService, Service = S>,
162 {
163 TowerMiddleware {
164 service: std::sync::Mutex::new(layer.layer(NextService)),
165 phase: Phase::Plugins,
166 }
167 }
168
169 /// Run this layer in a specific [`Phase`] rather than [`Phase::Plugins`].
170 pub fn in_phase(mut self, phase: Phase) -> Self {
171 self.phase = phase;
172 self
173 }
174}
175
176#[async_trait]
177impl<S, ResBody> Middleware for TowerMiddleware<S>
178where
179 S: Service<http::Request<Body>, Response = http::Response<ResBody>> + Clone + Send + 'static,
180 S::Error: std::fmt::Display + Send,
181 S::Future: Send,
182 ResBody: http_body::Body<Data = bytes::Bytes> + Send + 'static,
183 ResBody::Error: std::fmt::Display,
184{
185 async fn handle(&self, call: Call, next: Next) -> Response {
186 // Clone per request: tower services are cheap to clone by design, and
187 // `call` needs `&mut`. State that matters lives behind the clone.
188 let mut service = match self.service.lock() {
189 Ok(s) => s.clone(),
190 // A poisoned mutex means some other request panicked while cloning.
191 // Nothing here is left inconsistent, so carry on rather than
192 // failing every subsequent request.
193 Err(poisoned) => poisoned.into_inner().clone(),
194 };
195
196 let req = {
197 let mut b = http::Request::builder()
198 .method(call.method().clone())
199 .uri(call.uri().clone());
200 if let Some(h) = b.headers_mut() {
201 *h = call.headers().clone();
202 }
203 // The body is not handed to the layer: it belongs to the call, and
204 // moving it here would consume it before the handler runs. Layers
205 // that rewrite request bodies are out of scope for this adapter.
206 b.body(Body::empty()).expect("request build is infallible")
207 };
208
209 let slot = std::cell::RefCell::new(Some((call, next)));
210 let outcome = IN_FLIGHT
211 .scope(slot, async move {
212 // Log the layer's own error, but do not render it: a
213 // `Service`'s `Display` is not written with a client audience
214 // in mind and routinely carries internal detail.
215 futures_util::future::poll_fn(|cx| service.poll_ready(cx))
216 .await
217 .map_err(|e| {
218 tracing::error!(error = %e, "tower service not ready");
219 })?;
220 service.call(req).await.map_err(|e| {
221 tracing::error!(error = %e, "tower service failed");
222 })
223 })
224 .await;
225
226 match outcome {
227 Ok(res) => {
228 let (parts, body) = res.into_parts();
229 let mut out = Response::new(parts.status);
230 out.headers = parts.headers;
231 // Read the length before rewrapping. Streaming everything is
232 // right for a compression layer, whose output size is unknown —
233 // but doing it unconditionally threw away the exact size of an
234 // already-buffered body, so every response through any layer
235 // (even `Identity`) lost its `Content-Length` and got chunked,
236 // and a synthesized `HEAD` could no longer report a length.
237 let exact = http_body::Body::size_hint(&body).exact();
238 out.body = Body::from_stream(http_body_util::BodyDataStream::new(body));
239 if let Some(len) = exact {
240 if !out.headers.contains_key(http::header::CONTENT_LENGTH) {
241 if let Ok(value) = http::HeaderValue::from_str(&len.to_string()) {
242 out.headers.insert(http::header::CONTENT_LENGTH, value);
243 }
244 }
245 }
246 out
247 }
248 // The invariant holds: a response is always produced — and it says
249 // no more than the catch_unwind path does.
250 Err(()) => crate::Error::internal("Internal Server Error").into_response(),
251 }
252 }
253}
254
255impl<S, ResBody> Plugin for TowerMiddleware<S>
256where
257 S: Service<http::Request<Body>, Response = http::Response<ResBody>> + Clone + Send + 'static,
258 S::Error: std::fmt::Display + Send,
259 S::Future: Send,
260 ResBody: http_body::Body<Data = bytes::Bytes> + Send + 'static,
261 ResBody::Error: std::fmt::Display,
262{
263 fn install(self: Box<Self>, app: &mut AppBuilder) {
264 let phase = self.phase;
265 app.add_middleware_in(phase, Arc::new(*self));
266 }
267}