swadhin / justapi public

p4: Unix domain socket listener support (justapi serve --unix <path>) (ADR-062)

swadhinbiswas committed 3 months ago

19a6fb8

1 parent 7289418

commit 19a6fb8

5 files changed

+739 -251

Files

crates

justapi-cli

src

main.rs +64 -15 workers.rs +41 -6

justapi-core

src

server

mod.rs +576 -230 BENCHMARKS.md +17 DECISIONS.md +41

Showing 5 changed files+739-251

BENCHMARKS.md DECISIONS.md crates/justapi-cli/src/main.rs crates/justapi-cli/src/workers.rs crates/justapi-core/src/server/mod.rs

▼BENCHMARKS.md

+17-0

@@ -1501,3 +1501,20 @@ Unit tests: `make_spawn_argv` (fd+args carried), `bind_listener` (actually binds
1501	1501	ephemeral port). Gates green. Throughput now scales with core count instead of
1502	1502	one accept loop; no p99 regression on the request path. Auto-scaling workers and
1503	1503	HTTP/3/Unix-socket remain follow-ups.
	1504	+
	1505	+### ADR-062 — Unix domain socket listener (`--unix <path>`)
	1506	+
	1507	+`justapi serve` now accepts `--unix <path>` (precedes `--addr`), serving HTTP/1.1
	1508	+over an AF_UNIX socket with the same middleware/handler pipeline as TCP. Works
	1509	+with `--workers N` prefork via fd inheritance (parent binds once, workers recover
	1510	+via `listener_from_unix_fd`).
	1511	+
	1512	+\| Property \| Result \|
	1513	+\|---\|---\|
	1514	+\| `--unix /tmp/j.sock` single-process \| "Listening on /tmp/j.sock (plain HTTP/1.1, unix)"; curl `--unix-socket` served \|
	1515	+\| `--unix /tmp/j.sock --workers 2` \| parent + 2 workers share the UDS fd; curl served \|
	1516	+\| UDS + SIGTERM (parent) \| both workers drain + exit 0, tree reaped \|
	1517	+\| TLS over UDS \| unsupported (clear error) — documented limitation \|
	1518	+
	1519	+No request-path perf regression vs TCP (shared `serve_connection`). Unit test
	1520	+`test_unix_socket_serves_http` added. Gates green.
1504	1521

▼DECISIONS.md

+41-0

@@ -2309,3 +2309,44 @@ worker → respawned. Gates green: `cargo test --workspace --features db`, clipp
2309	2309	`-D warnings`, `cargo fmt --check`. (Windows/macOS: `--worker-fd` path is Unix-
2310	2310	gated; non-unix falls back to single-process with a clear error — auto-scaling
2311	2311	left for a follow-up.)
	2312	+
	2313	+## ADR-062 — 2026-07-17 — Unix domain socket listener support (`justapi serve --unix <path>`)
	2314	+
	2315	+Context: The feature list calls for "Unix domain socket support" for local
	2316	+deployments (no TCP port; socket-permission-gated access; easy systemd socket
	2317	+activation and nginx upstreams). `Server` only accepted a TCP `SocketAddr`.
	2318	+
	2319	+Decision / change:
	2320	+- `crates/justapi-core/src/server/mod.rs`: introduced a `ListenAddr` enum
	2321	+ (`Tcp(SocketAddr)` \| `Unix(PathBuf)`) with `From<SocketAddr>`, and
	2322	+ `Server::new` now takes `impl Into<ListenAddr>`. `Server::run()` branches on
	2323	+ TCP vs Unix; new `run_on_uds(UnixListener)` accepts an already-bound socket.
	2324	+ Per-connection logic (`serve_connection`) was extracted out of `serve_http`
	2325	+ and is shared by the TCP and Unix accept loops (`serve_unix` / `serve_http`).
	2326	+ `bind_unix_listener(path)` removes a stale socket file on bind and spawns a
	2327	+ `UnixSocketGuard` task that removes the file when the process's runtime
	2328	+ terminates. TLS over UDS is not supported (`serve_on_uds_tls` bails with a
	2329	+ clear error) — documented limitation.
	2330	+- `crates/justapi-cli/src/main.rs`: added a `--unix <path>` flag (takes
	2331	+ precedence over `--addr`). `build_server` now takes `&ListenAddr`; all four
	2332	+ paths (single-process, reload, worker `--worker-fd`, parent prefork) branch on
	2333	+ UDS vs TCP. In prefork, the parent binds the UDS once via
	2334	+ `workers::bind_unix_listener` (clears `FD_CLOEXEC`, keeps the listener alive
	2335	+ in the parent for the supervisor's lifetime) and hands the fd to each worker,
	2336	+ which recovers it with `workers::listener_from_unix_fd`.
	2337	+- `crates/justapi-cli/src/workers.rs`: added `bind_unix_listener` (sync,
	2338	+ non-CLOEXEC) mirroring `bind_listener`, and `listener_from_unix_fd` (recovers a
	2339	+ tokio `UnixListener` from an inherited fd, setting non-blocking first to avoid
	2340	+ the tokio "Registering a blocking socket" panic).
	2341	+
	2342	+Evidence: New `#[cfg(test)]` `test_unix_socket_serves_http` binds a UDS,
	2343	+serves a trivial handler, connects via `tokio::net::UnixStream`, and asserts a
	2344	+`200 {"uds":true}` HTTP response (exercises `serve_unix` → `serve_connection`,
	2345	+262 core tests in the workspace run). Manual: `justapi serve --unix /tmp/j.sock`
	2346	+→ "Listening on /tmp/j.sock (plain HTTP/1.1, unix)"; `curl --unix-socket /tmp/j.sock
	2347	+http://localhost/` → 404 (no route). Prefork: `justapi serve --unix /tmp/j.sock
	2348	+--workers 2` → parent + 2 workers, `curl --unix-socket` served; SIGTERM to parent
	2349	+→ both workers drain + exit 0 ("worker exited during shutdown", "all workers
	2350	+exited cleanly"), tree reaped. Gates green: `cargo test --workspace
	2351	+--features justapi-core/db`, clippy `-D warnings`, `cargo fmt --check`. (UDS is
	2352	+Unix-only; on non-unix `--unix` errors clearly.)
2312	2353

▼crates/justapi-cli/src/main.rs

+64-15

@@ -26,6 +26,11 @@ enum Commands {
26	26	#[arg(short, long, default_value = "127.0.0.1:8080")]
27	27	addr: SocketAddr,
28	28
	29	+ /// Bind to a Unix domain socket instead of TCP (e.g. `/run/justapi.sock`).
	30	+ /// Takes precedence over `--addr` when both are given. Unix only.
	31	+ #[arg(long)]
	32	+ unix: Option<String>,
	33	+
29	34	/// Serve static files from this directory
30	35	#[arg(long)]
31	36	static_dir: Option<String>,
@@ -531,7 +536,7 @@ fn build_engines(
531	536	/// wiring is applied uniformly for both the single-process and worker paths,
532	537	/// eliminating the previous duplication between the reload/non-reload branches.
533	538	fn build_server(
534		- addr: SocketAddr,
	539	+ listen_addr: &justapi_core::server::ListenAddr,
535	540	static_dir: &Option<String>,
536	541	#[cfg(feature = "compression")] compress: bool,
537	542	#[cfg(feature = "tls")] tls_cert: &Option<String>,
@@ -542,7 +547,7 @@ fn build_server(
542	547	>,
543	548	token: tokio_util::sync::CancellationToken,
544	549	) -> justapi_core::Server {
545		- let mut server = justapi_core::Server::new(addr).with_shutdown(token);
	550	+ let mut server = justapi_core::Server::new(listen_addr.clone()).with_shutdown(token);
546	551
547	552	if let Some(dir) = static_dir.clone() {
548	553	server = server.with_static_dir(dir);
@@ -609,6 +614,7 @@ async fn run() -> anyhow::Result<()> {
609	614	match cli.command {
610	615	Commands::Serve {
611	616	addr,
	617	+ unix,
612	618	static_dir,
613	619	#[cfg(feature = "compression")]
614	620	compress,
@@ -639,6 +645,15 @@ async fn run() -> anyhow::Result<()> {
639	645	} => {
640	646	let drain_duration = Duration::from_secs(drain_timeout);
641	647
	648	+ // Resolve the listen address: a Unix socket takes precedence over TCP.
	649	+ let is_unix = unix.is_some();
	650	+ let listen_addr = match &unix {
	651	+ Some(path) => {
	652	+ justapi_core::server::ListenAddr::Unix(std::path::PathBuf::from(path))
	653	+ }
	654	+ None => justapi_core::server::ListenAddr::Tcp(addr),
	655	+ };
	656	+
642	657	// Build the inference engines (no-op unless --model supplied).
643	658	#[cfg(feature = "inference")]
644	659	let (engine, scheduler_engine) =
@@ -648,11 +663,10 @@ async fn run() -> anyhow::Result<()> {
648	663	if let Some(fd) = worker_fd {
649	664	#[cfg(unix)]
650	665	{
651		- let listener = workers::listener_from_fd(fd)?;
652	666	let token = tokio_util::sync::CancellationToken::new();
653	667	wire_shutdown_signals(token.clone());
654	668	let server = build_server(
655		- addr,
	669	+ &listen_addr,
656	670	&static_dir,
657	671	#[cfg(feature = "compression")]
658	672	compress,
@@ -666,7 +680,11 @@ async fn run() -> anyhow::Result<()> {
666	680	&scheduler_engine,
667	681	token,
668	682	);
669		- let result = server.run_on(listener).await;
	683	+ let result = if is_unix {
	684	+ server.run_on_uds(workers::listener_from_unix_fd(fd)?).await
	685	+ } else {
	686	+ server.run_on(workers::listener_from_fd(fd)?).await
	687	+ };
670	688	justapi_core::tracing_setup::shutdown_tracing();
671	689	return result;
672	690	}
@@ -685,10 +703,27 @@ async fn run() -> anyhow::Result<()> {
685	703	}
686	704	anyhow::ensure!(workers <= 256, "worker count must be <= 256");
687	705
688		- let std_listener = workers::bind_listener(addr)?;
689		- let raw_fd = {
	706	+ // Bind the shared listener (TCP or Unix) once in the parent and
	707	+ // hand its fd to each worker child. The listener is kept alive in
	708	+ // the parent for the whole supervisor lifetime so its fd is not
	709	+ // closed before the children inherit it.
	710	+ #[cfg(unix)]
	711	+ let (raw_fd, _keep_alive) = {
690	712	use std::os::fd::AsRawFd;
691		- std_listener.as_raw_fd()
	713	+ if is_unix {
	714	+ let listener = workers::bind_unix_listener(unix.as_deref().unwrap())?;
	715	+ (listener.as_raw_fd(), Box::new(listener) as Box<dyn std::any::Any>)
	716	+ } else {
	717	+ let listener = workers::bind_listener(addr)?;
	718	+ (listener.as_raw_fd(), Box::new(listener) as Box<dyn std::any::Any>)
	719	+ }
	720	+ };
	721	+ #[cfg(not(unix))]
	722	+ let (raw_fd, _keep_alive) = {
	723	+ anyhow::ensure!(!is_unix, "Unix sockets require a Unix platform");
	724	+ use std::os::fd::AsRawFd;
	725	+ let listener = workers::bind_listener(addr)?;
	726	+ (listener.as_raw_fd(), Box::new(listener) as Box<dyn std::any::Any>)
692	727	};
693	728
694	729	// Original argv (minus program name) so each child reconstructs
@@ -699,7 +734,7 @@ async fn run() -> anyhow::Result<()> {
699	734	wire_shutdown_signals(shutdown.clone());
700	735
701	736	let spawn = workers::make_spawn_argv(&base_argv, raw_fd, drain_duration);
702		- tracing::info!(addr = %addr, workers, "starting prefork server");
	737	+ tracing::info!(addr = %listen_addr.display(), workers, "starting prefork server");
703	738	let result = workers::supervise(workers, spawn, shutdown).await;
704	739	justapi_core::tracing_setup::shutdown_tracing();
705	740	return result;
@@ -758,7 +793,7 @@ async fn run() -> anyhow::Result<()> {
758	793	});
759	794
760	795	let server = build_server(
761		- addr,
	796	+ &listen_addr,
762	797	&static_dir,
763	798	#[cfg(feature = "compression")]
764	799	compress,
@@ -773,9 +808,21 @@ async fn run() -> anyhow::Result<()> {
773	808	token.clone(),
774	809	);
775	810
776		- let std_listener = workers::bind_listener(addr)?;
777		- let listener = tokio::net::TcpListener::from_std(std_listener)?;
778		- let result = server.run_on(listener).await;
	811	+ let result = if is_unix {
	812	+ #[cfg(unix)]
	813	+ {
	814	+ let listener = workers::bind_unix_listener(unix.as_deref().unwrap())?;
	815	+ server.run_on_uds(tokio::net::UnixListener::from_std(listener)?).await
	816	+ }
	817	+ #[cfg(not(unix))]
	818	+ {
	819	+ anyhow::bail!("Unix sockets require a Unix platform")
	820	+ }
	821	+ } else {
	822	+ let std_listener = workers::bind_listener(addr)?;
	823	+ let listener = tokio::net::TcpListener::from_std(std_listener)?;
	824	+ server.run_on(listener).await
	825	+ };
779	826
780	827	// Recover the watcher receiver for the next iteration.
781	828	if let Ok(recovered) = return_rx.await {
@@ -800,7 +847,7 @@ async fn run() -> anyhow::Result<()> {
800	847
801	848	// Non-reload path: run server once and exit.
802	849	let server = build_server(
803		- addr,
	850	+ &listen_addr,
804	851	&static_dir,
805	852	#[cfg(feature = "compression")]
806	853	compress,
@@ -910,7 +957,9 @@ async fn run() -> anyhow::Result<()> {
910	957	}
911	958	}
912	959	} else {
913		- let router = justapi_core::Server::new("127.0.0.1:0".parse().unwrap());
	960	+ let router = justapi_core::Server::new(justapi_core::server::ListenAddr::Tcp(
	961	+ "127.0.0.1:0".parse().unwrap(),
	962	+ ));
914	963	let _ = router;
915	964	println!("Use --spec-file to load an OpenAPI spec for route listing.");
916	965	println!("For built-in routes, access /openapi.json on a running server.");
917	966

▼crates/justapi-cli/src/workers.rs

+41-6

@@ -76,6 +76,46 @@ pub fn listener_from_fd(fd: i32) -> anyhow::Result<tokio::net::TcpListener> {
76	76	.context("worker: failed to wrap inherited fd as tokio TcpListener")
77	77	}
78	78
	79	+/// Recover a tokio `UnixListener` from an inherited raw fd (Unix only).
	80	+#[cfg(unix)]
	81	+pub fn listener_from_unix_fd(fd: i32) -> anyhow::Result<tokio::net::UnixListener> {
	82	+ use std::os::fd::FromRawFd;
	83	+ // SAFETY: the fd was created by the parent via `bind_unix_listener` and is
	84	+ // not closed on exec, so it is valid and uniquely owned by this process
	85	+ // after fork/exec. We do not `from_raw_fd` it anywhere else.
	86	+ let std_listener = unsafe { <std::os::fd::OwnedFd as FromRawFd>::from_raw_fd(fd) };
	87	+ let std_listener = std::os::unix::net::UnixListener::from(std_listener);
	88	+ std_listener
	89	+ .set_nonblocking(true)
	90	+ .context("worker: failed to set inherited unix listener non-blocking")?;
	91	+ tokio::net::UnixListener::from_std(std_listener)
	92	+ .context("worker: failed to wrap inherited fd as tokio UnixListener")
	93	+}
	94	+
	95	+/// Bind a Unix domain socket once and return the std listener (non-CLOEXEC fd
	96	+/// so it survives the child re-exec on Unix). An existing socket file is
	97	+/// removed first to avoid EADDRINUSE on restart.
	98	+#[cfg(unix)]
	99	+pub fn bind_unix_listener(path: &str) -> anyhow::Result<std::os::unix::net::UnixListener> {
	100	+ use std::os::fd::AsRawFd;
	101	+ let p = std::path::Path::new(path);
	102	+ if p.exists() {
	103	+ let _ = std::fs::remove_file(p);
	104	+ }
	105	+ let listener = std::os::unix::net::UnixListener::bind(p)
	106	+ .with_context(\|\| format!("failed to bind unix socket {path}"))?;
	107	+ // Clear FD_CLOEXEC so the fd survives the child's exec (mirrors bind_listener).
	108	+ let fd = listener.as_raw_fd();
	109	+ // SAFETY: fd is a valid, owned listener socket; we only clear CLOEXEC.
	110	+ let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
	111	+ if flags >= 0 {
	112	+ unsafe {
	113	+ libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC);
	114	+ }
	115	+ }
	116	+ Ok(listener)
	117	+}
	118	+
79	119	/// Run the supervisor: spawn `count` workers and manage their lifecycle until
80	120	/// the shutdown token is cancelled (or a fatal spawn error occurs).
81	121	pub async fn supervise(
@@ -219,11 +259,7 @@ pub fn make_spawn_argv(
219	259	listener_fd: i32,
220	260	drain_timeout: Duration,
221	261	) -> WorkerSpawn {
222		- WorkerSpawn {
223		- argv: base_argv.to_vec(),
224		- listener_fd,
225		- drain_timeout,
226		- }
	262	+ WorkerSpawn { argv: base_argv.to_vec(), listener_fd, drain_timeout }
227	263	}
228	264
229	265	#[cfg(test)]
@@ -251,4 +287,3 @@ mod tests {
251	287	assert!(again.is_err(), "ephemeral port must be occupied by our listener");
252	288	}
253	289	}
254		-
255	290

▼crates/justapi-core/src/server/mod.rs

+576-230

@@ -297,8 +297,35 @@ pub struct TlsConfig {
297	297	// Server builder
298	298	// ---------------------------------------------------------------------------
299	299
	300	+/// Where the server listens. Either a TCP `SocketAddr` or a Unix domain socket
	301	+/// path. Both are bound by the parent and (in prefork mode) the fd is handed to
	302	+/// worker processes, so the same socket is shared across the worker fleet.
	303	+#[derive(Clone, Debug)]
	304	+pub enum ListenAddr {
	305	+ /// TCP listener, e.g. `127.0.0.1:8080`.
	306	+ Tcp(SocketAddr),
	307	+ /// Unix domain socket, e.g. `/run/justapi.sock` (Unix only).
	308	+ Unix(std::path::PathBuf),
	309	+}
	310	+
	311	+impl ListenAddr {
	312	+ /// Human-readable address for logs.
	313	+ pub fn display(&self) -> String {
	314	+ match self {
	315	+ ListenAddr::Tcp(a) => a.to_string(),
	316	+ ListenAddr::Unix(p) => p.display().to_string(),
	317	+ }
	318	+ }
	319	+}
	320	+
	321	+impl From<SocketAddr> for ListenAddr {
	322	+ fn from(addr: SocketAddr) -> Self {
	323	+ ListenAddr::Tcp(addr)
	324	+ }
	325	+}
	326	+
300	327	pub struct Server {
301		- addr: SocketAddr,
	328	+ addr: ListenAddr,
302	329	chain: MiddlewareChain,
303	330	static_dir: Option<StaticDir>,
304	331	static_mounts: Vec<StaticMount>,
@@ -328,7 +355,7 @@ pub struct Server {
328	355	}
329	356
330	357	impl Server {
331		- pub fn new(addr: SocketAddr) -> Self {
	358	+ pub fn new(addr: impl Into<ListenAddr>) -> Self {
332	359	let router = Router::new();
333	360	let metrics = Metrics::new();
334	361	let health_registry = Arc::new(HealthRegistry::new());
@@ -345,7 +372,7 @@ impl Server {
345	372	})
346	373	}));
347	374	Self {
348		- addr,
	375	+ addr: addr.into(),
349	376	chain,
350	377	static_dir: None,
351	378	static_mounts: Vec::new(),
@@ -807,8 +834,23 @@ impl Server {
807	834	self.chain.set_handler(handler);
808	835	}
809	836
810		- let listener = tokio::net::TcpListener::bind(self.addr).await?;
811		- self.run_on(listener).await
	837	+ match &self.addr {
	838	+ ListenAddr::Tcp(a) => {
	839	+ let listener = tokio::net::TcpListener::bind(*a).await?;
	840	+ self.run_on(listener).await
	841	+ }
	842	+ ListenAddr::Unix(path) => {
	843	+ #[cfg(unix)]
	844	+ {
	845	+ let listener = bind_unix_listener(path).await?;
	846	+ self.run_on_uds(listener).await
	847	+ }
	848	+ #[cfg(not(unix))]
	849	+ {
	850	+ anyhow::bail!("Unix domain sockets are only supported on Unix platforms")
	851	+ }
	852	+ }
	853	+ }
812	854	}
813	855
814	856	/// Run the server on an already-bound listener (e.g. chosen by the caller
@@ -827,7 +869,7 @@ impl Server {
827	869	self.chain.set_handler(handler);
828	870	}
829	871
830		- let local_addr = listener.local_addr()?;
	872	+ let local_addr = self.addr.display();
831	873
832	874	if let (Some(grpc_addr), Some(grpc_handler)) = (self.grpc_addr, self.grpc_handler.take()) {
833	875	tracing::info!("Starting gRPC server on {}", grpc_addr);
@@ -941,6 +983,111 @@ impl Server {
941	983	plugin_registry.on_shutdown_all().await?;
942	984	res
943	985	}
	986	+
	987	+ /// Run the server on an already-bound Unix domain socket listener (used by
	988	+ /// the prefork supervisor to hand a UDS fd to a worker child).
	989	+ pub async fn run_on_uds(mut self, listener: tokio::net::UnixListener) -> Result<()> {
	990	+ if let Some(router) = self.router.take() {
	991	+ let pool = Arc::new(BufferPool::new());
	992	+ let handler = make_handler(
	993	+ Arc::new(router),
	994	+ pool,
	995	+ self.metrics.clone(),
	996	+ self.health_registry.clone(),
	997	+ self.openapi_spec.clone(),
	998	+ self.max_body_size,
	999	+ );
	1000	+ self.chain.set_handler(handler);
	1001	+ }
	1002	+
	1003	+ let local_addr = self.addr.display();
	1004	+
	1005	+ if let (Some(grpc_addr), Some(grpc_handler)) = (self.grpc_addr, self.grpc_handler.take()) {
	1006	+ tracing::info!("Starting gRPC server on {}", grpc_addr);
	1007	+ let grpc_service = crate::grpc::DynamicGrpcService::new(grpc_handler);
	1008	+ let shutdown = self.shutdown.clone();
	1009	+ tokio::spawn(async move {
	1010	+ if let Ok(grpc_listener) = tokio::net::TcpListener::bind(grpc_addr).await {
	1011	+ loop {
	1012	+ if let Some(ref token) = shutdown {
	1013	+ tokio::select! {
	1014	+ result = grpc_listener.accept() => {
	1015	+ if let Ok((stream, _)) = result {
	1016	+ let io = hyper_util::rt::TokioIo::new(stream);
	1017	+ let svc = hyper_util::service::TowerToHyperService::new(grpc_service.clone());
	1018	+ tokio::spawn(async move {
	1019	+ let _ = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new())
	1020	+ .serve_connection_with_upgrades(io, svc)
	1021	+ .await;
	1022	+ });
	1023	+ }
	1024	+ }
	1025	+ _ = token.cancelled() => {
	1026	+ break;
	1027	+ }
	1028	+ }
	1029	+ } else {
	1030	+ if let Ok((stream, _)) = grpc_listener.accept().await {
	1031	+ let io = hyper_util::rt::TokioIo::new(stream);
	1032	+ let svc = hyper_util::service::TowerToHyperService::new(
	1033	+ grpc_service.clone(),
	1034	+ );
	1035	+ tokio::spawn(async move {
	1036	+ let _ = hyper_util::server::conn::auto::Builder::new(
	1037	+ hyper_util::rt::TokioExecutor::new(),
	1038	+ )
	1039	+ .serve_connection_with_upgrades(io, svc)
	1040	+ .await;
	1041	+ });
	1042	+ }
	1043	+ }
	1044	+ }
	1045	+ }
	1046	+ });
	1047	+ }
	1048	+
	1049	+ self.plugin_registry.on_startup_all().await?;
	1050	+
	1051	+ let chain = Arc::new(self.chain);
	1052	+ let static_dir = self.static_dir;
	1053	+ let metrics = self.metrics;
	1054	+ let plugin_registry = Arc::new(self.plugin_registry);
	1055	+ let wasm_middleware = self.wasm_middleware.clone();
	1056	+ #[cfg(feature = "ws")]
	1057	+ let ws_handler = self.ws_handler.clone();
	1058	+
	1059	+ #[cfg(feature = "tls")]
	1060	+ if self.tls.is_some() {
	1061	+ anyhow::bail!("TLS is not supported over Unix domain sockets");
	1062	+ }
	1063	+
	1064	+ tracing::info!("Listening on {} (plain HTTP/1.1, unix)", local_addr);
	1065	+ #[cfg(feature = "ws")]
	1066	+ let res = serve_unix(
	1067	+ listener,
	1068	+ chain,
	1069	+ static_dir,
	1070	+ self.static_mounts.clone(),
	1071	+ metrics,
	1072	+ self.shutdown,
	1073	+ wasm_middleware,
	1074	+ ws_handler,
	1075	+ )
	1076	+ .await;
	1077	+ #[cfg(not(feature = "ws"))]
	1078	+ let res = serve_unix(
	1079	+ listener,
	1080	+ chain,
	1081	+ static_dir,
	1082	+ self.static_mounts.clone(),
	1083	+ metrics,
	1084	+ self.shutdown,
	1085	+ wasm_middleware,
	1086	+ )
	1087	+ .await;
	1088	+ plugin_registry.on_shutdown_all().await?;
	1089	+ res
	1090	+ }
944	1091	}
945	1092
946	1093	// ---------------------------------------------------------------------------
@@ -1124,6 +1271,274 @@ async fn try_serve_static(
1124	1271	None
1125	1272	}
1126	1273
	1274	+/// Drive a single accepted connection to completion: acquire a connection
	1275	+/// permit, run it through hyper, the middleware chain, WASM/WS hooks, and the
	1276	+/// graceful-shutdown path. Shared by `serve_http` (TCP) and `serve_unix` (UDS)
	1277	+/// so the per-connection logic lives in exactly one place.
	1278	+///
	1279	+/// `peer_addr` is `None` for Unix sockets (no client IP); WebSocket `client`
	1280	+/// info and request extensions degrade gracefully when it is absent.
	1281	+async fn serve_connection(
	1282	+ stream: impl tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
	1283	+ peer_addr: Option<SocketAddr>,
	1284	+ chain: Arc<MiddlewareChain>,
	1285	+ static_dir: Option<StaticDir>,
	1286	+ static_mounts: Vec<StaticMount>,
	1287	+ conn_semaphore: std::sync::Arc<tokio::sync::Semaphore>,
	1288	+ metrics: Metrics,
	1289	+ shutdown: Option<CancellationToken>,
	1290	+ wasm_middleware: Option<Arc<crate::wasm::WasmEngine>>,
	1291	+ #[cfg(feature = "ws")] ws_handler: Option<WsHandler>,
	1292	+) {
	1293	+ // Bound concurrent connections: a permit is held for the life of
	1294	+ // the connection so a flood of accepts can't exhaust FDs/memory
	1295	+ // (connection-flood / slowloris resource exhaustion).
	1296	+ let _permit = conn_semaphore.acquire_owned().await.expect("connection semaphore closed");
	1297	+ let io = TokioIo::new(stream);
	1298	+ let arena = Arc::new(SharedArena::new());
	1299	+ let token_clone = shutdown.clone();
	1300	+ let spawn_metrics = metrics.clone();
	1301	+ let conn_metrics = metrics.clone();
	1302	+ conn_metrics.connection_opened();
	1303	+ let svc = service_fn(move \|mut req\| {
	1304	+ arena.reset();
	1305	+ let chain = chain.clone();
	1306	+ let arena = arena.clone();
	1307	+ let static_dir = static_dir.clone();
	1308	+ let static_mounts = static_mounts.clone();
	1309	+ let metrics = spawn_metrics.clone();
	1310	+ let wasm_middleware = wasm_middleware.clone();
	1311	+ #[cfg(feature = "ws")]
	1312	+ let ws_handler = ws_handler.clone();
	1313	+ async move {
	1314	+ let path = req.uri().path().to_string();
	1315	+ let method = req.method().clone();
	1316	+ let _arena_path = arena.alloc_str(&path);
	1317	+ let start = std::time::Instant::now();
	1318	+ req.extensions_mut().insert(peer_addr);
	1319	+
	1320	+ let span = tracing::info_span!(
	1321	+ "http.request",
	1322	+ http.method = method.as_str(),
	1323	+ http.path = %path,
	1324	+ http.status_code = tracing::field::Empty,
	1325	+ );
	1326	+
	1327	+ metrics.record_request();
	1328	+
	1329	+ #[cfg(feature = "ws")]
	1330	+ if let Some(ref handler) = ws_handler {
	1331	+ let is_ws = req
	1332	+ .headers()
	1333	+ .get(hyper::header::UPGRADE)
	1334	+ .and_then(\|v\| v.to_str().ok())
	1335	+ .map(\|s\| s.eq_ignore_ascii_case("websocket"))
	1336	+ .unwrap_or(false);
	1337	+
	1338	+ if is_ws {
	1339	+ if let Some(key) = req.headers().get("sec-websocket-key") {
	1340	+ let key_str = key.to_str().unwrap_or("");
	1341	+ let mut sha1 = sha1::Sha1::new();
	1342	+ use sha1::Digest;
	1343	+ sha1.update(key_str.as_bytes());
	1344	+ sha1.update(b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11");
	1345	+ use base64::Engine;
	1346	+ let accept_key =
	1347	+ base64::engine::general_purpose::STANDARD.encode(sha1.finalize());
	1348	+
	1349	+ let mut res = crate::json_response(StatusCode::SWITCHING_PROTOCOLS, "");
	1350	+ res.headers_mut().insert(
	1351	+ hyper::header::UPGRADE,
	1352	+ hyper::header::HeaderValue::from_static("websocket"),
	1353	+ );
	1354	+ res.headers_mut().insert(
	1355	+ hyper::header::CONNECTION,
	1356	+ hyper::header::HeaderValue::from_static("upgrade"),
	1357	+ );
	1358	+ res.headers_mut().insert(
	1359	+ "sec-websocket-accept",
	1360	+ hyper::header::HeaderValue::from_str(&accept_key).unwrap(),
	1361	+ );
	1362	+
	1363	+ let handler = handler.clone();
	1364	+ let ws_info = WsConnInfo {
	1365	+ path: path.clone(),
	1366	+ query_string: req.uri().query().unwrap_or("").as_bytes().to_vec(),
	1367	+ headers: req
	1368	+ .headers()
	1369	+ .iter()
	1370	+ .map(\|(k, v)\| {
	1371	+ (k.as_str().as_bytes().to_vec(), v.as_bytes().to_vec())
	1372	+ })
	1373	+ .collect(),
	1374	+ client: peer_addr.map(\|a\| (a.ip().to_string(), a.port())),
	1375	+ };
	1376	+
	1377	+ tokio::task::spawn(async move {
	1378	+ match hyper::upgrade::on(&mut req).await {
	1379	+ Ok(upgraded) => {
	1380	+ let upgraded_io = hyper_util::rt::TokioIo::new(upgraded);
	1381	+ let ws_stream =
	1382	+ tokio_tungstenite::WebSocketStream::from_raw_socket(
	1383	+ upgraded_io,
	1384	+ tokio_tungstenite::tungstenite::protocol::Role::Server,
	1385	+ None,
	1386	+ )
	1387	+ .await;
	1388	+ let (write, read) = ws_stream.split();
	1389	+ dispatch_ws(Box::new(read), Box::new(write), ws_info, &handler)
	1390	+ .await;
	1391	+ }
	1392	+ Err(e) => tracing::error!("WebSocket upgrade error: {}", e),
	1393	+ }
	1394	+ });
	1395	+ return Ok(res);
	1396	+ }
	1397	+ }
	1398	+ }
	1399	+
	1400	+ // WASM middleware (synchronous setup + one await)
	1401	+ if let Some(ref wasm) = wasm_middleware {
	1402	+ let wasm_span = tracing::debug_span!("wasm.middleware");
	1403	+ let mut hdrs = serde_json::Map::new();
	1404	+ for (k, v) in req.headers() {
	1405	+ if let Ok(v_str) = v.to_str() {
	1406	+ hdrs.insert(
	1407	+ k.as_str().to_string(),
	1408	+ serde_json::Value::String(v_str.to_string()),
	1409	+ );
	1410	+ }
	1411	+ }
	1412	+ let req_json = serde_json::json!({
	1413	+ "path": path,
	1414	+ "method": req.method().as_str(),
	1415	+ "headers": hdrs,
	1416	+ })
	1417	+ .to_string();
	1418	+
	1419	+ let wasm_result = {
	1420	+ let _ws = wasm_span.enter();
	1421	+ wasm.execute_middleware(&req_json).await
	1422	+ };
	1423	+
	1424	+ match wasm_result {
	1425	+ Ok(res_json) => {
	1426	+ if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&res_json) {
	1427	+ if let Some(status) = parsed.get("status") {
	1428	+ if let Some(status_code) = status.as_u64() {
	1429	+ if status_code != 200 {
	1430	+ let body = parsed
	1431	+ .get("body")
	1432	+ .and_then(\|v\| v.as_str())
	1433	+ .unwrap_or("")
	1434	+ .to_string();
	1435	+ span.record("http.status_code", status_code);
	1436	+ metrics.record_status(
	1437	+ StatusCode::from_u16(status_code as u16)
	1438	+ .unwrap_or(StatusCode::FORBIDDEN),
	1439	+ );
	1440	+ metrics
	1441	+ .record_latency(start.elapsed().as_secs_f64() * 1000.0);
	1442	+ return Ok(json_response(
	1443	+ StatusCode::from_u16(status_code as u16)
	1444	+ .unwrap_or(StatusCode::FORBIDDEN),
	1445	+ &body,
	1446	+ ));
	1447	+ }
	1448	+ }
	1449	+ }
	1450	+ }
	1451	+ }
	1452	+ Err(e) => {
	1453	+ tracing::error!("WASM execution error: {:?}", e);
	1454	+ }
	1455	+ }
	1456	+ }
	1457	+
	1458	+ // Run middleware chain (async, no entered span across await)
	1459	+ let resp = tokio::time::timeout(request_timeout(), chain.run(req)).await;
	1460	+
	1461	+ match resp {
	1462	+ Ok(r) => match r {
	1463	+ Ok(response) => {
	1464	+ let status = response.status();
	1465	+ span.record("http.status_code", status.as_u16());
	1466	+ metrics.record_status(status);
	1467	+ metrics.record_latency(start.elapsed().as_secs_f64() * 1000.0);
	1468	+
	1469	+ // If the response is 404, try serving static files.
	1470	+ if status == StatusCode::NOT_FOUND {
	1471	+ if let Some(resp) =
	1472	+ try_serve_static(&path, &static_dir, &static_mounts).await
	1473	+ {
	1474	+ return Ok::<_, anyhow::Error>(resp);
	1475	+ }
	1476	+ }
	1477	+ Ok(response)
	1478	+ }
	1479	+ Err(_) => {
	1480	+ span.record("http.status_code", 404u16);
	1481	+ metrics.record_status(StatusCode::NOT_FOUND);
	1482	+ metrics.record_latency(start.elapsed().as_secs_f64() * 1000.0);
	1483	+
	1484	+ // Middleware error — try static files
	1485	+ if let Some(resp) =
	1486	+ try_serve_static(&path, &static_dir, &static_mounts).await
	1487	+ {
	1488	+ return Ok::<_, anyhow::Error>(resp);
	1489	+ }
	1490	+ Ok(json_response(StatusCode::NOT_FOUND, r#"{"detail":"not found"}"#))
	1491	+ }
	1492	+ },
	1493	+ Err(_) => {
	1494	+ // Handler exceeded the request timeout.
	1495	+ span.record("http.status_code", 504u16);
	1496	+ metrics.record_status(StatusCode::GATEWAY_TIMEOUT);
	1497	+ metrics.record_latency(start.elapsed().as_secs_f64() * 1000.0);
	1498	+ tracing::warn!(
	1499	+ "Request to {} {} timed out after {:?}",
	1500	+ method,
	1501	+ path,
	1502	+ request_timeout()
	1503	+ );
	1504	+ Ok(json_response(
	1505	+ StatusCode::GATEWAY_TIMEOUT,
	1506	+ r#"{"detail":"request timeout"}"#,
	1507	+ ))
	1508	+ }
	1509	+ }
	1510	+ }
	1511	+ });
	1512	+ let mut builder =
	1513	+ hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
	1514	+ builder
	1515	+ .http1()
	1516	+ .timer(hyper_util::rt::TokioTimer::new())
	1517	+ .header_read_timeout(HEADER_READ_TIMEOUT);
	1518	+ let mut conn = std::pin::pin!(builder.serve_connection_with_upgrades(io, svc));
	1519	+
	1520	+ if let Some(token) = token_clone {
	1521	+ tokio::select! {
	1522	+ res = &mut conn => {
	1523	+ if let Err(e) = res {
	1524	+ tracing::warn!("Connection error: {}", e);
	1525	+ }
	1526	+ }
	1527	+ _ = token.cancelled() => {
	1528	+ conn.as_mut().graceful_shutdown();
	1529	+ if let Err(e) = conn.await {
	1530	+ tracing::warn!("Connection error during shutdown: {}", e);
	1531	+ }
	1532	+ }
	1533	+ }
	1534	+ } else {
	1535	+ if let Err(e) = conn.await {
	1536	+ tracing::warn!("Connection error: {}", e);
	1537	+ }
	1538	+ }
	1539	+ conn_metrics.connection_closed();
	1540	+}
	1541	+
1127	1542	async fn serve_http(
1128	1543	listener: TcpListener,
1129	1544	chain: Arc<MiddlewareChain>,
@@ -1157,9 +1572,6 @@ async fn serve_http(
1157	1572	let chain = chain.clone();
1158	1573	let static_dir = static_dir.clone();
1159	1574	let static_mounts = static_mounts.clone();
1160		- let conn_metrics = metrics.clone();
1161		- conn_metrics.connection_opened();
1162		- let spawn_metrics = conn_metrics.clone();
1163	1575	let wasm_middleware = wasm_middleware.clone();
1164	1576	let peer_addr = peer;
1165	1577
@@ -1168,235 +1580,126 @@ async fn serve_http(
1168	1580
1169	1581	let token_clone = shutdown.clone();
1170	1582	let conn_semaphore = conn_semaphore.clone();
1171		- connections.spawn(async move {
1172		- // Bound concurrent connections: a permit is held for the life of
1173		- // the connection so a flood of accepts can't exhaust FDs/memory
1174		- // (connection-flood / slowloris resource exhaustion).
1175		- let _permit = conn_semaphore
1176		- .acquire_owned()
1177		- .await
1178		- .expect("connection semaphore closed");
1179		- let io = TokioIo::new(stream);
1180		- let arena = Arc::new(SharedArena::new());
1181		- let svc = service_fn(move \|mut req\| {
1182		- arena.reset();
1183		- let chain = chain.clone();
1184		- let arena = arena.clone();
1185		- let static_dir = static_dir.clone();
1186		- let static_mounts = static_mounts.clone();
1187		- let metrics = spawn_metrics.clone();
1188		- let wasm_middleware = wasm_middleware.clone();
1189		- #[cfg(feature = "ws")]
1190		- let ws_handler = ws_handler.clone();
1191		- async move {
1192		- let path = req.uri().path().to_string();
1193		- let method = req.method().clone();
1194		- let _arena_path = arena.alloc_str(&path);
1195		- let start = std::time::Instant::now();
1196		- req.extensions_mut().insert(peer_addr);
1197		-
1198		- let span = tracing::info_span!(
1199		- "http.request",
1200		- http.method = method.as_str(),
1201		- http.path = %path,
1202		- http.status_code = tracing::field::Empty,
1203		- );
	1583	+ connections.spawn(serve_connection(
	1584	+ stream,
	1585	+ Some(peer_addr),
	1586	+ chain,
	1587	+ static_dir,
	1588	+ static_mounts,
	1589	+ conn_semaphore,
	1590	+ metrics.clone(),
	1591	+ token_clone,
	1592	+ wasm_middleware,
	1593	+ #[cfg(feature = "ws")]
	1594	+ ws_handler,
	1595	+ ));
	1596	+ }
1204	1597
1205		- metrics.record_request();
1206		-
1207		- #[cfg(feature = "ws")]
1208		- if let Some(ref handler) = ws_handler {
1209		- let is_ws = req.headers().get(hyper::header::UPGRADE)
1210		- .and_then(\|v\| v.to_str().ok())
1211		- .map(\|s\| s.eq_ignore_ascii_case("websocket"))
1212		- .unwrap_or(false);
1213		-
1214		- if is_ws {
1215		- if let Some(key) = req.headers().get("sec-websocket-key") {
1216		- let key_str = key.to_str().unwrap_or("");
1217		- let mut sha1 = sha1::Sha1::new();
1218		- use sha1::Digest;
1219		- sha1.update(key_str.as_bytes());
1220		- sha1.update(b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11");
1221		- use base64::Engine;
1222		- let accept_key = base64::engine::general_purpose::STANDARD.encode(sha1.finalize());
1223		-
1224		- let mut res = crate::json_response(StatusCode::SWITCHING_PROTOCOLS, "");
1225		- res.headers_mut().insert(hyper::header::UPGRADE, hyper::header::HeaderValue::from_static("websocket"));
1226		- res.headers_mut().insert(hyper::header::CONNECTION, hyper::header::HeaderValue::from_static("upgrade"));
1227		- res.headers_mut().insert("sec-websocket-accept", hyper::header::HeaderValue::from_str(&accept_key).unwrap());
1228		-
1229		- let handler = handler.clone();
1230		- let ws_info = WsConnInfo {
1231		- path: path.clone(),
1232		- query_string: req.uri().query().unwrap_or("").as_bytes().to_vec(),
1233		- headers: req
1234		- .headers()
1235		- .iter()
1236		- .map(\|(k, v)\| (k.as_str().as_bytes().to_vec(), v.as_bytes().to_vec()))
1237		- .collect(),
1238		- client: Some((peer_addr.ip().to_string(), peer_addr.port())),
1239		- };
1240		-
1241		- tokio::task::spawn(async move {
1242		- match hyper::upgrade::on(&mut req).await {
1243		- Ok(upgraded) => {
1244		- let upgraded_io = hyper_util::rt::TokioIo::new(upgraded);
1245		- let ws_stream = tokio_tungstenite::WebSocketStream::from_raw_socket(
1246		- upgraded_io,
1247		- tokio_tungstenite::tungstenite::protocol::Role::Server,
1248		- None
1249		- ).await;
1250		- let (write, read) = ws_stream.split();
1251		- dispatch_ws(Box::new(read), Box::new(write), ws_info, &handler).await;
1252		- }
1253		- Err(e) => tracing::error!("WebSocket upgrade error: {}", e),
1254		- }
1255		- });
1256		- return Ok(res);
1257		- }
1258		- }
1259		- }
	1598	+ if shutdown.is_some() {
	1599	+ tracing::info!("Waiting for {} active connections to drain...", connections.len());
	1600	+ tokio::select! {
	1601	+ _ = async { while connections.join_next().await.is_some() {} } => {
	1602	+ tracing::debug!("all connections closed gracefully");
	1603	+ }
	1604	+ _ = tokio::time::sleep(std::time::Duration::from_secs(30)) => {
	1605	+ tracing::warn!("Graceful shutdown timeout exceeded. Dropping remaining connections.");
	1606	+ }
	1607	+ }
	1608	+ } else {
	1609	+ while connections.join_next().await.is_some() {}
	1610	+ }
1260	1611
1261		- // WASM middleware (synchronous setup + one await)
1262		- if let Some(ref wasm) = wasm_middleware {
1263		- let wasm_span = tracing::debug_span!("wasm.middleware");
1264		- let mut hdrs = serde_json::Map::new();
1265		- for (k, v) in req.headers() {
1266		- if let Ok(v_str) = v.to_str() {
1267		- hdrs.insert(
1268		- k.as_str().to_string(),
1269		- serde_json::Value::String(v_str.to_string()),
1270		- );
1271		- }
1272		- }
1273		- let req_json = serde_json::json!({
1274		- "path": path,
1275		- "method": req.method().as_str(),
1276		- "headers": hdrs,
1277		- })
1278		- .to_string();
1279		-
1280		- let wasm_result = {
1281		- let _ws = wasm_span.enter();
1282		- wasm.execute_middleware(&req_json).await
1283		- };
	1612	+ Ok(())
	1613	+}
1284	1614
1285		- match wasm_result {
1286		- Ok(res_json) => {
1287		- if let Ok(parsed) =
1288		- serde_json::from_str::<serde_json::Value>(&res_json)
1289		- {
1290		- if let Some(status) = parsed.get("status") {
1291		- if let Some(status_code) = status.as_u64() {
1292		- if status_code != 200 {
1293		- let body = parsed
1294		- .get("body")
1295		- .and_then(\|v\| v.as_str())
1296		- .unwrap_or("")
1297		- .to_string();
1298		- span.record("http.status_code", status_code);
1299		- metrics.record_status(
1300		- StatusCode::from_u16(status_code as u16)
1301		- .unwrap_or(StatusCode::FORBIDDEN),
1302		- );
1303		- metrics.record_latency(
1304		- start.elapsed().as_secs_f64() * 1000.0,
1305		- );
1306		- return Ok(json_response(
1307		- StatusCode::from_u16(status_code as u16)
1308		- .unwrap_or(StatusCode::FORBIDDEN),
1309		- &body,
1310		- ));
1311		- }
1312		- }
1313		- }
1314		- }
1315		- }
1316		- Err(e) => {
1317		- tracing::error!("WASM execution error: {:?}", e);
1318		- }
1319		- }
1320		- }
	1615	+/// Bind a Unix domain socket, removing any stale socket file left from a
	1616	+/// previous run (a common footgun — `bind` fails with EADDRINUSE on an existing
	1617	+/// socket path). The socket file is unlinked when the listener is dropped.
	1618	+#[cfg(unix)]
	1619	+pub async fn bind_unix_listener(path: &std::path::Path) -> Result<tokio::net::UnixListener> {
	1620	+ // Remove a leftover socket file so re-start doesn't fail with EADDRINUSE.
	1621	+ if path.exists() {
	1622	+ let _ = std::fs::remove_file(path);
	1623	+ }
	1624	+ let listener = tokio::net::UnixListener::bind(path)
	1625	+ .map_err(\|e\| anyhow::anyhow!("failed to bind unix socket {}: {e}", path.display()))?;
	1626	+ // Best-effort: clean up the socket file on process exit.
	1627	+ let cleanup = path.to_path_buf();
	1628	+ tokio::spawn(async move {
	1629	+ // Hold a guard so the file is removed when the process terminates.
	1630	+ let _guard = UnixSocketGuard { path: cleanup };
	1631	+ // Never resolve: kept alive for the process lifetime.
	1632	+ std::future::pending::<()>().await;
	1633	+ });
	1634	+ Ok(listener)
	1635	+}
1321	1636
1322		- // Run middleware chain (async, no entered span across await)
1323		- let resp = tokio::time::timeout(request_timeout(), chain.run(req)).await;
	1637	+#[cfg(unix)]
	1638	+struct UnixSocketGuard {
	1639	+ path: std::path::PathBuf,
	1640	+}
1324	1641
1325		- match resp {
1326		- Ok(r) => match r {
1327		- Ok(response) => {
1328		- let status = response.status();
1329		- span.record("http.status_code", status.as_u16());
1330		- metrics.record_status(status);
1331		- metrics.record_latency(start.elapsed().as_secs_f64() * 1000.0);
	1642	+#[cfg(unix)]
	1643	+impl Drop for UnixSocketGuard {
	1644	+ fn drop(&mut self) {
	1645	+ let _ = std::fs::remove_file(&self.path);
	1646	+ }
	1647	+}
1332	1648
1333		- // If the response is 404, try serving static files.
1334		- if status == StatusCode::NOT_FOUND {
1335		- if let Some(resp) = try_serve_static(&path, &static_dir, &static_mounts).await {
1336		- return Ok::<_, anyhow::Error>(resp);
1337		- }
1338		- }
1339		- Ok(response)
1340		- }
1341		- Err(_) => {
1342		- span.record("http.status_code", 404u16);
1343		- metrics.record_status(StatusCode::NOT_FOUND);
1344		- metrics.record_latency(start.elapsed().as_secs_f64() * 1000.0);
1345		-
1346		- // Middleware error — try static files
1347		- if let Some(resp) = try_serve_static(&path, &static_dir, &static_mounts).await {
1348		- return Ok::<_, anyhow::Error>(resp);
1349		- }
1350		- Ok(json_response(
1351		- StatusCode::NOT_FOUND,
1352		- r#"{"detail":"not found"}"#
1353		- ))
1354		- }
1355		- },
1356		- Err(_) => {
1357		- // Handler exceeded the request timeout.
1358		- span.record("http.status_code", 504u16);
1359		- metrics.record_status(StatusCode::GATEWAY_TIMEOUT);
1360		- metrics.record_latency(start.elapsed().as_secs_f64() * 1000.0);
1361		- tracing::warn!(
1362		- "Request to {} {} timed out after {:?}",
1363		- method,
1364		- path,
1365		- request_timeout()
1366		- );
1367		- Ok(json_response(
1368		- StatusCode::GATEWAY_TIMEOUT,
1369		- r#"{"detail":"request timeout"}"#
1370		- ))
1371		- }
1372		- }
1373		- }
1374		- });
1375		- let mut builder = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
1376		- builder.http1().timer(hyper_util::rt::TokioTimer::new()).header_read_timeout(HEADER_READ_TIMEOUT);
1377		- let mut conn = std::pin::pin!(builder.serve_connection_with_upgrades(io, svc));
1378		-
1379		- if let Some(token) = token_clone {
1380		- tokio::select! {
1381		- res = &mut conn => {
1382		- if let Err(e) = res {
1383		- tracing::warn!("Connection error: {}", e);
1384		- }
1385		- }
1386		- _ = token.cancelled() => {
1387		- conn.as_mut().graceful_shutdown();
1388		- if let Err(e) = conn.await {
1389		- tracing::warn!("Connection error during shutdown: {}", e);
1390		- }
1391		- }
1392		- }
1393		- } else {
1394		- if let Err(e) = conn.await {
1395		- tracing::warn!("Connection error: {}", e);
	1649	+/// Accept loop for a Unix domain socket. Identical to `serve_http` except it
	1650	+/// serves over a `UnixListener` and passes `peer_addr: None` to the shared
	1651	+/// per-connection handler (Unix peers have no IP).
	1652	+#[cfg(unix)]
	1653	+async fn serve_unix(
	1654	+ listener: tokio::net::UnixListener,
	1655	+ chain: Arc<MiddlewareChain>,
	1656	+ static_dir: Option<StaticDir>,
	1657	+ static_mounts: Vec<StaticMount>,
	1658	+ metrics: Metrics,
	1659	+ shutdown: Option<CancellationToken>,
	1660	+ wasm_middleware: Option<Arc<crate::wasm::WasmEngine>>,
	1661	+ #[cfg(feature = "ws")] ws_handler: Option<WsHandler>,
	1662	+) -> Result<()> {
	1663	+ let mut connections = tokio::task::JoinSet::new();
	1664	+ let conn_semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(max_connections()));
	1665	+
	1666	+ loop {
	1667	+ let token = shutdown.as_ref().cloned();
	1668	+
	1669	+ let (stream, _peer) = if let Some(token) = &token {
	1670	+ tokio::select! {
	1671	+ result = listener.accept() => result?,
	1672	+ _ = token.cancelled() => {
	1673	+ tracing::info!("Shutdown signal received, stopping accept loop");
	1674	+ break;
1396	1675	}
1397	1676	}
1398		- conn_metrics.connection_closed();
1399		- });
	1677	+ } else {
	1678	+ listener.accept().await?
	1679	+ };
	1680	+
	1681	+ let chain = chain.clone();
	1682	+ let static_dir = static_dir.clone();
	1683	+ let static_mounts = static_mounts.clone();
	1684	+ let wasm_middleware = wasm_middleware.clone();
	1685	+
	1686	+ #[cfg(feature = "ws")]
	1687	+ let ws_handler = ws_handler.clone();
	1688	+
	1689	+ let conn_semaphore = conn_semaphore.clone();
	1690	+ connections.spawn(serve_connection(
	1691	+ stream,
	1692	+ None,
	1693	+ chain,
	1694	+ static_dir,
	1695	+ static_mounts,
	1696	+ conn_semaphore,
	1697	+ metrics.clone(),
	1698	+ shutdown.clone(),
	1699	+ wasm_middleware,
	1700	+ #[cfg(feature = "ws")]
	1701	+ ws_handler,
	1702	+ ));
1400	1703	}
1401	1704
1402	1705	if shutdown.is_some() {
@@ -1988,6 +2291,49 @@ mod tests {
1988	2291	let m = router.at(&Method::GET, "/metrics").unwrap();
1989	2292	assert!(matches!(m.handler, Handler::Prometheus));
1990	2293	}
	2294	+
	2295	+ // Bind a Unix socket, serve a trivial handler over it, and confirm a raw
	2296	+ // HTTP/1.1 request over the socket gets a 200 response. Exercises the
	2297	+ // shared `serve_connection` path through `serve_unix`.
	2298	+ #[cfg(unix)]
	2299	+ #[tokio::test]
	2300	+ async fn test_unix_socket_serves_http() {
	2301	+ use tokio::io::{AsyncReadExt, AsyncWriteExt};
	2302	+
	2303	+ let dir = std::env::temp_dir();
	2304	+ let path = dir.join(format!("justapi-test-uds-{}.sock", std::process::id()));
	2305	+ let _ = std::fs::remove_file(&path);
	2306	+
	2307	+ let listener = bind_unix_listener(&path).await.expect("bind unix listener");
	2308	+
	2309	+ let chain = std::sync::Arc::new(MiddlewareChain::new(std::sync::Arc::new(
	2310	+ \|_req: Request<Incoming>\| {
	2311	+ Box::pin(async move { Ok(crate::json_response(StatusCode::OK, r#"{"uds":true}"#)) })
	2312	+ },
	2313	+ )));
	2314	+
	2315	+ let server = tokio::spawn(async move {
	2316	+ #[cfg(feature = "ws")]
	2317	+ let _ = serve_unix(listener, chain, None, Vec::new(), Metrics::new(), None, None, None)
	2318	+ .await;
	2319	+ #[cfg(not(feature = "ws"))]
	2320	+ let _ = serve_unix(listener, chain, None, Vec::new(), Metrics::new(), None, None).await;
	2321	+ });
	2322	+
	2323	+ // Give the acceptor a moment to start.
	2324	+ tokio::time::sleep(std::time::Duration::from_millis(100)).await;
	2325	+
	2326	+ let mut stream = tokio::net::UnixStream::connect(&path).await.unwrap();
	2327	+ stream.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
	2328	+ let mut buf = Vec::new();
	2329	+ stream.read_to_end(&mut buf).await.unwrap();
	2330	+ let resp = String::from_utf8_lossy(&buf);
	2331	+ assert!(resp.contains("200"), "expected 200, got: {resp}");
	2332	+ assert!(resp.contains("\"uds\":true"), "expected uds body, got: {resp}");
	2333	+
	2334	+ server.abort();
	2335	+ let _ = std::fs::remove_file(&path);
	2336	+ }
1991	2337	}
1992	2338
1993	2339	// --- gRPC stub placeholder ---
1994	2340