Krolyxon
Commit 55c1225

Branches

Tags

Refs →

ferreniumsrcservermod.rs

297 lines · 10.3 KB · Rust

Raw Blame History Download
1//! HTTP server: router, middleware, listeners, signals and shutdown.
2
3pub mod assets;
4pub mod dispatch;
5pub mod headers;
6
7use std::future::{Future, IntoFuture};
8use std::sync::Arc;
9use std::time::Duration;
10
11use anyhow::Context;
12use axum::Router;
13use axum::extract::{Request, State};
14use axum::http::{HeaderValue, StatusCode, Uri, header};
15use axum::middleware::{self, Next};
16use axum::response::{IntoResponse, Redirect, Response};
17use axum::routing::get;
18use sd_notify::NotifyState;
19use tokio::signal::unix::{SignalKind, signal};
20use tokio::sync::watch;
21use tower::{Layer, ServiceBuilder};
22use tower_http::CompressionLevel;
23use tower_http::catch_panic::CatchPanicLayer;
24use tower_http::compression::predicate::{NotForContentType, Predicate};
25use tower_http::compression::{CompressionLayer, DefaultPredicate};
26use tower_http::timeout::TimeoutLayer;
27use tower_http::trace::{DefaultOnResponse, TraceLayer};
28use tracing::Level;
29
30use crate::config::Listen;
31use crate::state::AppState;
32use crate::{handlers, index};
33
34/// How long in-flight requests may run after SIGTERM before we exit anyway.
35const DRAIN_DEADLINE: Duration = Duration::from_secs(10);
36
37/// Build the full application (routes + middleware) for the current config.
38pub fn app(state: Arc<AppState>) -> Router {
39 let config = state.config.load();
40
41 let routes = Router::new()
42 .route("/", get(handlers::index::index))
43 .route("/-/health", get(health))
44 .route("/-/site.css", get(assets::site_css))
45 .route("/-/static/{hash}/{*path}", get(assets::serve))
46 .route("/{*path}", get(dispatch::dispatch))
47 .fallback(handlers::error::not_found)
48 .with_state(state.clone());
49
50 // The prefix must be stripped *before* routing, so wrap the whole router
51 // (Router::layer would run after route matching).
52 let routes = match config.server.base_path.as_str() {
53 "/" => routes,
54 base => Router::new().fallback_service(
55 middleware::from_fn_with_state(
56 Arc::<str>::from(base.trim_end_matches('/')),
57 strip_base_path,
58 )
59 .layer(routes),
60 ),
61 };
62
63 let trusted_proxy = config.server.trusted_proxy;
64 let compression = CompressionLayer::new()
65 .no_deflate()
66 .no_zstd()
67 .quality(CompressionLevel::Precise(4));
68 // Always installed (so the body type is fixed); disabling turns off every
69 // encoding instead.
70 let compression = if config.server.compression {
71 compression
72 } else {
73 compression.no_gzip().no_br()
74 }
75 // woff2 is already compressed.
76 .compress_when(DefaultPredicate::new().and(NotForContentType::const_new("font/")));
77
78 let [csp, nosniff, referrer, frame] = headers::layers();
79 routes.layer(
80 ServiceBuilder::new()
81 .layer(
82 TraceLayer::new_for_http()
83 .make_span_with(move |req: &Request| {
84 let client = trusted_proxy
85 .then(|| forwarded_for(req))
86 .flatten()
87 .unwrap_or("-");
88 tracing::info_span!(
89 "request",
90 method = %req.method(),
91 uri = %req.uri(),
92 client = %client,
93 )
94 })
95 .on_request(())
96 .on_response(DefaultOnResponse::new().level(Level::INFO)),
97 )
98 .layer(CatchPanicLayer::new())
99 .layer(TimeoutLayer::with_status_code(
100 StatusCode::SERVICE_UNAVAILABLE,
101 Duration::from_secs(config.limits.request_timeout_secs),
102 ))
103 .layer(compression)
104 .layer(csp)
105 .layer(nosniff)
106 .layer(referrer)
107 .layer(frame),
108 )
109}
110
111/// First address in `X-Forwarded-For` (the original client, as seen by the
112/// trusted proxy).
113fn forwarded_for(req: &Request) -> Option<&str> {
114 let value = req.headers().get("x-forwarded-for")?.to_str().ok()?;
115 value
116 .split(',')
117 .next()
118 .map(str::trim)
119 .filter(|s| !s.is_empty())
120}
121
122/// Serve under `server.base_path`: strip the prefix before routing, redirect
123/// the bare prefix to its slash form, and 404 everything outside it.
124async fn strip_base_path(State(base): State<Arc<str>>, mut req: Request, next: Next) -> Response {
125 let path = req.uri().path();
126 if path == &*base {
127 return Redirect::permanent(&format!("{base}/")).into_response();
128 }
129 let Some(rest) = path.strip_prefix(&*base).filter(|r| r.starts_with('/')) else {
130 return StatusCode::NOT_FOUND.into_response();
131 };
132 let new_uri = match req.uri().query() {
133 Some(q) => format!("{rest}?{q}"),
134 None => rest.to_string(),
135 };
136 match new_uri.parse::<Uri>() {
137 Ok(uri) => {
138 *req.uri_mut() = uri;
139 next.run(req).await
140 }
141 Err(_) => StatusCode::BAD_REQUEST.into_response(),
142 }
143}
144
145async fn health() -> impl IntoResponse {
146 (
147 [(header::CACHE_CONTROL, HeaderValue::from_static("no-store"))],
148 "ok\n",
149 )
150}
151
152/// Run the server until SIGTERM/SIGINT. SIGHUP reloads the config.
153pub async fn run(state: Arc<AppState>) -> anyhow::Result<()> {
154 let listen = state.config.load().server.listen.clone();
155 let (shutdown_tx, shutdown_rx) = watch::channel(false);
156 tokio::spawn(handle_signals(state.clone(), shutdown_tx)?);
157
158 // The index is complete before we report READY.
159 let st = state.clone();
160 let stats = tokio::task::spawn_blocking(move || {
161 let config = st.config.load_full();
162 st.index.full_scan(&config)
163 })
164 .await
165 .context("initial scan failed")?;
166 let mut seen = Default::default();
167 index::watch::log_scan(&stats, &state.index.load(), &mut seen, true);
168 tokio::spawn(index::watch::run(state.clone(), seen));
169 // Parse the embedded grammars now rather than on the first blob view.
170 tokio::task::spawn_blocking(|| {
171 let start = std::time::Instant::now();
172 crate::render::highlight::syntaxes();
173 tracing::debug!("syntaxes loaded in {:?}", start.elapsed());
174 });
175
176 let app = app(state.clone());
177
178 match &listen {
179 Listen::Tcp(addr) => {
180 let listener = tokio::net::TcpListener::bind(addr)
181 .await
182 .with_context(|| format!("cannot listen on {addr}"))?;
183 serve(listener, app, &listen, shutdown_rx).await
184 }
185 Listen::Unix(path) => {
186 let listener = bind_unix(path)?;
187 let result = serve(listener, app, &listen, shutdown_rx).await;
188 let _ = std::fs::remove_file(path);
189 result
190 }
191 }
192}
193
194async fn serve<L>(
195 listener: L,
196 app: Router,
197 listen: &Listen,
198 mut shutdown_rx: watch::Receiver<bool>,
199) -> anyhow::Result<()>
200where
201 L: axum::serve::Listener,
202 L::Addr: std::fmt::Debug,
203{
204 tracing::info!("listening on {listen}");
205 notify(&[NotifyState::Ready, NotifyState::Status("serving")]);
206
207 let mut drain_rx = shutdown_rx.clone();
208 let server = axum::serve(listener, app)
209 .with_graceful_shutdown(async move {
210 let _ = shutdown_rx.wait_for(|stop| *stop).await;
211 })
212 .into_future();
213
214 tokio::select! {
215 result = server => result.context("server error")?,
216 _ = async {
217 let _ = drain_rx.wait_for(|stop| *stop).await;
218 tokio::time::sleep(DRAIN_DEADLINE).await;
219 } => {
220 tracing::warn!(
221 "connections still open after {}s; exiting anyway",
222 DRAIN_DEADLINE.as_secs()
223 );
224 }
225 }
226 tracing::info!("stopped");
227 Ok(())
228}
229
230/// Bind a unix socket, replacing a stale socket file from a previous run.
231fn bind_unix(path: &std::path::Path) -> anyhow::Result<tokio::net::UnixListener> {
232 use std::os::unix::fs::{FileTypeExt, PermissionsExt};
233
234 if let Ok(meta) = std::fs::symlink_metadata(path) {
235 anyhow::ensure!(
236 meta.file_type().is_socket(),
237 "{} exists and is not a socket",
238 path.display()
239 );
240 std::fs::remove_file(path)
241 .with_context(|| format!("cannot remove stale socket {}", path.display()))?;
242 }
243 let listener = tokio::net::UnixListener::bind(path)
244 .with_context(|| format!("cannot listen on unix:{}", path.display()))?;
245 // Same reachability as a loopback TCP port: any local user (i.e. the
246 // reverse proxy) may connect. All content served is public anyway.
247 std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))
248 .with_context(|| format!("cannot set permissions on {}", path.display()))?;
249 Ok(listener)
250}
251
252/// Install signal handlers (synchronously, so a failure aborts startup) and
253/// return the task that services them.
254fn handle_signals(
255 state: Arc<AppState>,
256 shutdown_tx: watch::Sender<bool>,
257) -> anyhow::Result<impl Future<Output = ()> + Send + 'static> {
258 let mut term = signal(SignalKind::terminate()).context("cannot install SIGTERM handler")?;
259 let mut int = signal(SignalKind::interrupt()).context("cannot install SIGINT handler")?;
260 let mut hup = signal(SignalKind::hangup()).context("cannot install SIGHUP handler")?;
261
262 Ok(async move {
263 loop {
264 tokio::select! {
265 _ = term.recv() => { tracing::info!("SIGTERM received; draining"); break; }
266 _ = int.recv() => { tracing::info!("SIGINT received; draining"); break; }
267 _ = hup.recv() => reload(&state),
268 }
269 }
270 notify(&[NotifyState::Stopping]);
271 let _ = shutdown_tx.send(true);
272 })
273}
274
275fn reload(state: &AppState) {
276 tracing::info!("SIGHUP received; reloading configuration");
277 let mut msg = vec![NotifyState::Reloading];
278 if let Ok(now) = NotifyState::monotonic_usec_now() {
279 msg.push(now);
280 }
281 notify(&msg);
282 match state.reload() {
283 Ok(config) => {
284 crate::logging::apply(&config);
285 tracing::info!("configuration reloaded");
286 state.rescan.notify_one();
287 }
288 Err(e) => tracing::error!("reload failed, keeping current configuration: {e}"),
289 }
290 notify(&[NotifyState::Ready]);
291}
292
293fn notify(states: &[NotifyState]) {
294 if let Err(e) = sd_notify::notify(states) {
295 tracing::debug!("sd_notify failed: {e}");
296 }
297}