25568019997

Committed 08 May 2026 04:50PM UTC coverage: 57.28% (-1.3%) from 58.564%

Build # 25568019997

Build Type

push

github

Committed by

web-flow

Commit Message

Migrate from Trillium [part 7C]: admin routes and Trillium router removal (#2247)

* Migrate from Trillium [part 7C]: admin routes and Trillium router removal

- Move the admin queue routes (index, show, delete) to Axum handlers
gated by the new AdminPermissionsActor extractor, which centralizes
the admin-only check (returning 404 for non-admins to hide endpoint
existence). This is enforced by using a nested router, which has a
`route_layer` that layers `admin::require_admin` on top of everything
in that router, so we don't forget anything.

- Wire ReplaceMimeTypesLayer on the Axum /api sub-router now that all
API routes are served by Axum. Removed the Trillium routes() and
api_routes() functions, the api() handler chain, handler/misc.rs
(actor_required, admin_required), and all FromConn impls from route
files. Then I realized this is supposed to be in Part 8 and I stopped.

- Add #[serde(alias)]es to JobStatus so query-param
deserialization accepts lowercase values (matching the previous
QueryStrong behavior).

- Add #[cfg(all(assets)...] to a couple of telemetry things so we
don't commit compilation crimes. These'll get cleaned up when we
drop trillium_opentelemetry, I think.

Note that there is still cleanup to do in Part 8, even though the
migration is mostly done here. I wasn't super disciplined with what
I deleted and what I marked dead, for which I request grace. Just,
one thing at a time.

* Fix Docker dev compose

The Trillium `api()` handler chain had `state(User::for_integration_testing())`
gated on `#[cfg(feature = "integration-testing")]`, which injected an admin
user into every request. When 7C removed the Trillium router, this injection
was lost, causing the `pair_aggregator` container in compose.dev.yaml to get
403 Forbidden on `POST /api/aggregators` (n.b. the CLI uses `--token=""` and
relies on the integration-testing user for auth).

So, let's fix it. This is a new Axum middleware that unconditionally injects the
`integration-testin... (continued)

Coverage Stats

44 of 65 new or added lines in 5 files covered. (67.69%)

110 existing lines in 12 files now uncovered.

4351 of 7596 relevant lines covered (57.28%)

64.39 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.6

/src/handler/proxy.rs

//! Temporary reverse proxy handler that forwards unmatched Trillium requests to
//! the local Axum server. This exists only during the incremental migration and
//! will be removed once all routes have been moved to Axum.

use reqwest::{
    header::{HeaderName, CONNECTION, HOST, TE, TRAILER, TRANSFER_ENCODING},
    redirect,
};
use std::net::SocketAddr;
use trillium::{Conn, Handler, Status};

/// A Trillium [`Handler`] that proxies unhalted requests to a local Axum server.
#[derive(Debug)]
pub struct AxumProxy {
    upstream: String,
    client: reqwest::Client,
}

impl AxumProxy {
    pub fn new(addr: SocketAddr) -> Self {
        Self {
            upstream: format!("http://[::1]:{}", addr.port()),
            client: reqwest::Client::builder()
                .no_proxy()
                .redirect(redirect::Policy::none())
                .build()
                .expect("failed to build proxy HTTP client"),
        }
    }
}

/// Hop-by-hop headers that should not be forwarded through the proxy (RFC 7230 §6.1).
const UNPROXYABLE_HEADERS: [HeaderName; 5] = [HOST, TRANSFER_ENCODING, CONNECTION, TE, TRAILER];

#[trillium::async_trait]
impl Handler for AxumProxy {
    async fn run(&self, mut conn: Conn) -> Conn {
        // Only proxy requests that haven't been handled by earlier handlers.
        if conn.status().is_some() || conn.is_halted() {
            return conn;
        }

        let method = conn.method();
        let path = conn.path();
        let querystring = conn.querystring();

        let url = if querystring.is_empty() {
            format!("{}{}", self.upstream, path)
        } else {
            format!("{}{}?{}", self.upstream, path, querystring)
        };

        let reqwest_method = match reqwest::Method::from_bytes(method.as_ref().as_bytes()) {
            Ok(m) => m,
            Err(_) => return conn.with_status(Status::BadRequest).halt(),
        };

        let mut builder = self.client.request(reqwest_method, &url);

        // Forward request headers, filtering out hop-by-hop headers.
        for (name, values) in conn.request_headers() {
            let header_name = match HeaderName::from_bytes(name.as_ref().as_bytes()) {
                Ok(h) => h,
                Err(_) => continue,
            };
            if UNPROXYABLE_HEADERS.contains(&header_name) {
                continue;
            }
            for value in values.iter() {
                if let Some(s) = value.as_str() {
                    builder = builder.header(&header_name, s);
                }
            }
        }

        // Forward the request body. Note: no size limit is enforced here;
        // the Trillium API layer (trillium-api) enforces a 1 MiB limit before
        // requests reach this handler, so it's fine for the migration window.
        let body = conn.request_body().await.read_bytes().await;
        match body {
            Ok(bytes) if !bytes.is_empty() => {
                builder = builder.body(bytes);
            }
            Err(e) => {
                log::error!("axum proxy error reading request body: {e}");
                return conn.with_status(Status::BadRequest).halt();
            }
            _ => {}
        }

        let resp = match builder.send().await {
            Ok(resp) => resp,
            Err(e) => {
                log::error!("axum proxy error: {e}");
                return conn.with_status(Status::BadGateway).halt();
            }
        };

        let status = resp.status().as_u16();
        let resp_headers = resp.headers().clone();
        let body = match resp.bytes().await {
            Ok(b) => b,
            Err(e) => {
                log::error!("axum proxy error reading response: {e}");
                return conn.with_status(Status::BadGateway).halt();
            }
        };

        let mut conn = conn.with_status(status).halt();

        for (name, value) in resp_headers.iter() {
            if UNPROXYABLE_HEADERS.contains(name) {
                continue;
            }
            if let Ok(v) = value.to_str() {
                conn.response_headers_mut()
                    .append(name.as_str().to_owned(), v.to_owned());
            }
        }

        // This copies the response body; we could avoid it by streaming via
        // resp.into_body() + Body::new_streaming, but it's not worth the extra
        // plumbing for a temporary shim.
        conn.with_body(body.to_vec())
    }
}

1	//! Temporary reverse proxy handler that forwards unmatched Trillium requests to
2	//! the local Axum server. This exists only during the incremental migration and
3	//! will be removed once all routes have been moved to Axum.
4
5	use reqwest::{
6	header::{HeaderName, CONNECTION, HOST, TE, TRAILER, TRANSFER_ENCODING},
7	redirect,
8	};
9	use std::net::SocketAddr;
10	use trillium::{Conn, Handler, Status};
11
12	/// A Trillium [`Handler`] that proxies unhalted requests to a local Axum server.
13	#[derive(Debug)]
14	pub struct AxumProxy {
15	upstream: String,
16	client: reqwest::Client,
17	}
18
19	impl AxumProxy {
20	pub fn new(addr: SocketAddr) -> Self {	278✔
21	Self {	278✔
22	upstream: format!("http://[::1]:{}", addr.port()),	278✔
23	client: reqwest::Client::builder()	278✔
24	.no_proxy()	278✔
25	.redirect(redirect::Policy::none())	278✔
26	.build()	278✔
27	.expect("failed to build proxy HTTP client"),	278✔
28	}	278✔
29	}	278✔
30	}
31
32	/// Hop-by-hop headers that should not be forwarded through the proxy (RFC 7230 §6.1).
33	const UNPROXYABLE_HEADERS: [HeaderName; 5] = [HOST, TRANSFER_ENCODING, CONNECTION, TE, TRAILER];
34
35	#[trillium::async_trait]
36	impl Handler for AxumProxy {
37	async fn run(&self, mut conn: Conn) -> Conn {	546✔
38	// Only proxy requests that haven't been handled by earlier handlers.
39	if conn.status().is_some() \|\| conn.is_halted() {	273✔
UNCOV 40	return conn;	×
41	}	273✔
42
43	let method = conn.method();	273✔
44	let path = conn.path();	273✔
45	let querystring = conn.querystring();	273✔
46
47	let url = if querystring.is_empty() {	273✔
48	format!("{}{}", self.upstream, path)	268✔
49	} else {
50	format!("{}{}?{}", self.upstream, path, querystring)	5✔
51	};
52
53	let reqwest_method = match reqwest::Method::from_bytes(method.as_ref().as_bytes()) {	273✔
54	Ok(m) => m,	273✔
55	Err(_) => return conn.with_status(Status::BadRequest).halt(),	×
56	};
57
58	let mut builder = self.client.request(reqwest_method, &url);	273✔
59
60	// Forward request headers, filtering out hop-by-hop headers.
61	for (name, values) in conn.request_headers() {	1,219✔
62	let header_name = match HeaderName::from_bytes(name.as_ref().as_bytes()) {	1,219✔
63	Ok(h) => h,	1,219✔
64	Err(_) => continue,	×
65	};
66	if UNPROXYABLE_HEADERS.contains(&header_name) {	1,219✔
67	continue;	273✔
68	}	946✔
69	for value in values.iter() {	946✔
70	if let Some(s) = value.as_str() {	946✔
71	builder = builder.header(&header_name, s);	946✔
72	}	946✔
73	}
74	}
75
76	// Forward the request body. Note: no size limit is enforced here;
77	// the Trillium API layer (trillium-api) enforces a 1 MiB limit before
78	// requests reach this handler, so it's fine for the migration window.
79	let body = conn.request_body().await.read_bytes().await;	273✔
80	match body {	273✔
81	Ok(bytes) if !bytes.is_empty() => {	273✔
82	builder = builder.body(bytes);	110✔
83	}	110✔
84	Err(e) => {	×
85	log::error!("axum proxy error reading request body: {e}");	×
86	return conn.with_status(Status::BadRequest).halt();	×
87	}
88	_ => {}	163✔
89	}
90
91	let resp = match builder.send().await {	273✔
92	Ok(resp) => resp,	273✔
93	Err(e) => {	×
94	log::error!("axum proxy error: {e}");	×
95	return conn.with_status(Status::BadGateway).halt();	×
96	}
97	};
98
99	let status = resp.status().as_u16();	273✔
100	let resp_headers = resp.headers().clone();	273✔
101	let body = match resp.bytes().await {	273✔
102	Ok(b) => b,	273✔
103	Err(e) => {	×
104	log::error!("axum proxy error reading response: {e}");	×
105	return conn.with_status(Status::BadGateway).halt();	×
106	}
107	};
108
109	let mut conn = conn.with_status(status).halt();	273✔
110
111	for (name, value) in resp_headers.iter() {	2,041✔
112	if UNPROXYABLE_HEADERS.contains(name) {	2,041✔
113	continue;	×
114	}	2,041✔
115	if let Ok(v) = value.to_str() {	2,041✔
116	conn.response_headers_mut()	2,041✔
117	.append(name.as_str().to_owned(), v.to_owned());	2,041✔
118	}	2,041✔
119	}
120
121	// This copies the response body; we could avoid it by streaming via
122	// resp.into_body() + Body::new_streaming, but it's not worth the extra
123	// plumbing for a temporary shim.
124	conn.with_body(body.to_vec())	273✔
125	}	546✔
126	}

divviup / divviup-api / 25568019997

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous