cratefield_runtime_native/
server.rs

1//! The native server (issue #19): one axum router on a `TcpListener`,
2//! with client-IP sanitization outside everything else, the in-process
3//! cron scheduler, and graceful shutdown on SIGTERM/SIGINT (what
4//! `docker stop` sends).
5
6use std::net::SocketAddr;
7use std::sync::Arc;
8
9use axum::extract::{ConnectInfo, Request, State};
10use axum::middleware::{Next, from_fn_with_state};
11use axum::response::Response;
12use cratefield_core::{Config, Harness};
13use thiserror::Error;
14use tokio::net::TcpListener;
15
16use crate::config::EnvConfig;
17use crate::cron::{CronError, spawn_cron_scheduler};
18use crate::ports::{client_ip, normalize_headers, trusted_proxy_headers};
19use crate::runtime::{Native, clone_ports};
20use crate::tracing_setup::install_tracing;
21
22/// `LISTEN_ADDR` default: loopback only. A deployment open to the world
23/// must say so explicitly (`LISTEN_ADDR=0.0.0.0:8080`, as the compose
24/// example does) — the same fail-closed instinct as the empty
25/// trusted-proxy list.
26pub const DEFAULT_LISTEN_ADDR: &str = "127.0.0.1:8080";
27
28#[derive(Debug, Error)]
29pub enum ServeError {
30    /// `LISTEN_ADDR` is not a socket address.
31    #[error("LISTEN_ADDR {addr:?} is not a valid socket address")]
32    ListenAddr { addr: String },
33    /// The listener could not be bound.
34    #[error("failed to bind {addr}: {source}")]
35    Bind {
36        addr: String,
37        #[source]
38        source: std::io::Error,
39    },
40    /// A `CRONS` entry is not a valid cron expression; the process
41    /// refuses to start rather than silently skipping scheduled work
42    /// (wrangler rejects an invalid cron at deploy time for the same
43    /// reason).
44    #[error("invalid CRONS entry")]
45    Cron(#[from] CronError),
46    /// The serving loop failed.
47    #[error("server failed: {0}")]
48    Serve(#[source] std::io::Error),
49}
50
51/// Serves the harness: builds the router once, wraps it in the client-IP
52/// middleware, starts the cron scheduler for the `CRONS` config, and
53/// serves until SIGTERM or SIGINT.
54///
55/// # Errors
56///
57/// [`ServeError`] when `LISTEN_ADDR` cannot be parsed or bound, a cron
58/// expression is invalid, or the accept loop fails.
59///
60/// # Errors: shutdown semantics
61///
62/// Graceful shutdown waits for in-flight requests only. Deferred work
63/// (`Defer`/`tokio::spawn`) is not awaited on shutdown — a task killed
64/// mid-flight simply ends, like a Worker that stops getting slices
65/// after its 30 s `wait_until` budget. Handlers that must finish commit
66/// before responding.
67pub async fn serve(harness: Arc<Harness>, runtime: Native) -> Result<(), ServeError> {
68    install_tracing();
69    let config = Arc::new(EnvConfig);
70    let addr = listen_addr(&*config)?;
71    let listener = TcpListener::bind(addr)
72        .await
73        .map_err(|source| ServeError::Bind {
74            addr: addr.to_string(),
75            source,
76        })?;
77    serve_on(harness, runtime, listener).await
78}
79
80/// Serves the harness on an already-bound listener: builds the router
81/// once, wraps it in the client-IP middleware (trust list and `CRONS`
82/// from `std::env`), and serves until SIGTERM or SIGINT. `serve` is this
83/// plus `LISTEN_ADDR` resolution and binding; tests and ventures with
84/// their own socket management (fd passing, port 0) call this directly.
85///
86/// # Errors
87///
88/// [`ServeError`] when a `CRONS` entry is invalid or the accept loop
89/// fails. Shutdown semantics as on [`serve`].
90pub async fn serve_on(
91    harness: Arc<Harness>,
92    runtime: Native,
93    listener: TcpListener,
94) -> Result<(), ServeError> {
95    install_tracing();
96
97    let config = Arc::new(EnvConfig);
98    let ports = runtime.ports();
99
100    // Scheduled work: every expression gets its own task, fanned out to
101    // every module's `scheduled(ctx, cron)` — the same fan-out
102    // `serve_scheduled` performs per Workers cron trigger.
103    let crons = cron_expressions(&*config);
104    if !crons.is_empty() {
105        spawn_cron_scheduler(&harness, &ports, &crons)?;
106    }
107
108    let app = harness
109        .router(clone_ports(&ports))
110        .layer(from_fn_with_state(
111            IpTrust {
112                trusted: trusted_proxy_headers(&*config),
113            },
114            client_ip_layer,
115        ));
116
117    let addr = listener
118        .local_addr()
119        .map(|addr| addr.to_string())
120        .unwrap_or_default();
121    tracing::info!(%addr, "cratefield native runtime listening");
122
123    axum::serve(
124        listener,
125        app.into_make_service_with_connect_info::<SocketAddr>(),
126    )
127    .with_graceful_shutdown(shutdown_signal())
128    .await
129    .map_err(ServeError::Serve)
130}
131
132/// `CRONS`: comma-separated cron expressions (UTC, wrangler's `[triggers]
133/// crons` moved into config). Default: no scheduled work.
134fn cron_expressions(config: &dyn Config) -> Vec<String> {
135    config
136        .get("CRONS")
137        .map(|raw| {
138            raw.split(',')
139                .map(str::trim)
140                .filter(|entry| !entry.is_empty())
141                .map(str::to_owned)
142                .collect::<Vec<_>>()
143        })
144        .unwrap_or_default()
145}
146
147fn listen_addr(config: &dyn Config) -> Result<SocketAddr, ServeError> {
148    let raw = config
149        .get("LISTEN_ADDR")
150        .unwrap_or_else(|| DEFAULT_LISTEN_ADDR.to_owned());
151    raw.parse::<SocketAddr>()
152        .map_err(|_| ServeError::ListenAddr { addr: raw })
153}
154
155/// The trusted-proxy header list, resolved once per process and shared
156/// by every request (deployment config, not request state).
157#[derive(Clone)]
158struct IpTrust {
159    trusted: Vec<String>,
160}
161
162/// Client-IP middleware: resolve the address (trusted headers first,
163/// else the TCP peer), strip every forwarding header, and set the
164/// verdict into the header core's `client_ip` reads first. Runs outside
165/// the harness's own layers, so the request scope's `ip_hash` and every
166/// module's rate-limit key see the sanitized view.
167async fn client_ip_layer(
168    State(trust): State<IpTrust>,
169    mut request: Request,
170    next: Next,
171) -> Response {
172    let peer = request
173        .extensions()
174        .get::<ConnectInfo<SocketAddr>>()
175        .map(|ConnectInfo(addr)| addr.ip());
176    let resolved = client_ip(request.headers(), &trust.trusted, peer);
177    normalize_headers(request.headers_mut(), resolved);
178    next.run(request).await
179}
180
181/// Resolves when the process should stop: SIGINT (Ctrl-C) or, on Unix,
182/// SIGTERM (what `docker stop` and systemd send). Public so a venture
183/// that assembles its own serving loop gets the same semantics.
184///
185/// # Panics
186///
187/// Only if the signal handlers cannot be installed (the process
188/// environment disallows it); serving cannot continue meaningfully
189/// without shutdown signals.
190pub async fn shutdown_signal() {
191    let ctrl_c = tokio::signal::ctrl_c();
192    #[cfg(unix)]
193    {
194        let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
195            .expect("install SIGTERM handler");
196        tokio::select! {
197            _ = ctrl_c => {},
198            _ = sigterm.recv() => {},
199        }
200    }
201    #[cfg(not(unix))]
202    {
203        if ctrl_c.await.is_err() {
204            tracing::error!("ctrl_c handler failed; exiting");
205        }
206    }
207    tracing::info!("shutdown signal received; draining in-flight requests");
208}
209
210#[cfg(test)]
211mod tests {
212    use super::*;
213    use cratefield_core::MapConfig;
214
215    #[test]
216    fn crons_default_to_empty_and_split_on_commas() {
217        assert!(cron_expressions(&MapConfig::default()).is_empty());
218        let cfg = MapConfig::from_pairs([("CRONS", "*/5 * * * * , 0 0 * * *,,")]);
219        assert_eq!(cron_expressions(&cfg), ["*/5 * * * *", "0 0 * * *"]);
220    }
221
222    #[test]
223    fn listen_addr_defaults_to_loopback() {
224        let addr = listen_addr(&MapConfig::default()).expect("default parses");
225        assert_eq!(addr.to_string(), "127.0.0.1:8080");
226
227        let cfg = MapConfig::from_pairs([("LISTEN_ADDR", "0.0.0.0:9999")]);
228        let addr = listen_addr(&cfg).expect("parses");
229        assert_eq!(addr.to_string(), "0.0.0.0:9999");
230    }
231
232    #[test]
233    fn bad_listen_addr_is_an_error() {
234        let cfg = MapConfig::from_pairs([("LISTEN_ADDR", "not-an-addr")]);
235        assert!(matches!(
236            listen_addr(&cfg),
237            Err(ServeError::ListenAddr { .. })
238        ));
239    }
240}
cratefield_runtime_native/server.rs

cratefield_runtime_native/
server.rs
