Commit 952cccf · swadhin/justapi

Showing 7 changed files+94-83

crates/justapi-core/src/compress.rs crates/justapi-core/src/health.rs crates/justapi-core/src/openai.rs crates/justapi-core/src/resilience.rs crates/justapi-core/src/router.rs crates/justapi-core/src/server/mod.rs crates/justapi-core/tests/edge_cases.rs

▼crates/justapi-core/src/compress.rs

+3-0

@@ -317,6 +317,9 @@ mod tests {
317	317
318	318	#[test]
319	319	fn test_encoding_from_wildcard() {
	320	+ #[cfg(feature = "brotli-compression")]
	321	+ assert_eq!(Encoding::from_accept_encoding("*"), Encoding::Brotli);
	322	+ #[cfg(not(feature = "brotli-compression"))]
320	323	assert_eq!(Encoding::from_accept_encoding("*"), Encoding::Gzip);
321	324	}
322	325
323	326

▼crates/justapi-core/src/health.rs

+5-1

@@ -79,7 +79,11 @@ impl HealthRegistry {
79	79	/// Run all registered checks and return a health report.
80	80	/// Checks are run in parallel with a per-check timeout of 5 seconds.
81	81	pub async fn check_all(&self) -> HealthReport {
82		- let check_timeout = std::time::Duration::from_secs(5);
	82	+ self.check_all_with_timeout(std::time::Duration::from_secs(5)).await
	83	+ }
	84	+
	85	+ /// Run all registered checks with a custom per-check timeout.
	86	+ pub async fn check_all_with_timeout(&self, check_timeout: std::time::Duration) -> HealthReport {
83	87	let futs: Vec<_> = self
84	88	.checks
85	89	.iter()
86	90

▼crates/justapi-core/src/openai.rs

+2-2

@@ -815,8 +815,8 @@ mod tests {
815	815	#[test]
816	816	fn test_safe_status_invalid_code_falls_back_to_500() {
817	817	assert_eq!(safe_status(0), StatusCode::INTERNAL_SERVER_ERROR);
818		- assert_eq!(safe_status(999), StatusCode::INTERNAL_SERVER_ERROR);
819		- assert_eq!(safe_status(70000), StatusCode::INTERNAL_SERVER_ERROR);
	818	+ assert_eq!(safe_status(99), StatusCode::INTERNAL_SERVER_ERROR);
	819	+ assert_eq!(safe_status(6000), StatusCode::INTERNAL_SERVER_ERROR);
820	820	assert_eq!(safe_status(u16::MAX), StatusCode::INTERNAL_SERVER_ERROR);
821	821	}
822	822	}
823	823

▼crates/justapi-core/src/resilience.rs

+2-2

@@ -752,7 +752,7 @@ mod tests {
752	752	async fn test_timeout_returns_504() {
753	753	let slow_handler: HandlerFn<TestBody> = Arc::new(\|_req: Request<TestBody>\| {
754	754	Box::pin(async {
755		- tokio::time::sleep(Duration::from_secs(10)).await;
	755	+ tokio::time::sleep(Duration::from_millis(200)).await;
756	756	Ok(json_response(StatusCode::OK, r#"{"ok":true}"#))
757	757	})
758	758	});
@@ -1451,7 +1451,7 @@ mod tests {
1451	1451
1452	1452	let slow_handler: HandlerFn<TestBody> = Arc::new(\|_req: Request<TestBody>\| {
1453	1453	Box::pin(async {
1454		- tokio::time::sleep(Duration::from_secs(10)).await;
	1454	+ tokio::time::sleep(Duration::from_millis(500)).await;
1455	1455	Ok(json_response(StatusCode::OK, "ok"))
1456	1456	})
1457	1457	});
1458	1458

▼crates/justapi-core/src/router.rs

+6-4

@@ -381,7 +381,7 @@ mod tests {
381	381	let _ = router.at(&Method::GET, "/api/v3/users/42/posts/84");
382	382	}
383	383
384		- let iterations = 100_000u64;
	384	+ let iterations = if cfg!(miri) { 50u64 } else { 100_000u64 };
385	385	let mut total_ns = 0u64;
386	386
387	387	let test_paths = [
@@ -405,8 +405,10 @@ mod tests {
405	405	println!("500-route table: {} lookups, avg {}ns", total_lookups, avg_ns,);
406	406
407	407	// Benchmark gate: average lookup should be under 2us (release) / 5us (debug)
408		- // CI runners are slower than local machines, so we use a generous tolerance
409		- let target = if cfg!(debug_assertions) { 5000 } else { 2000 };
410		- assert!(avg_ns < target, "Route lookup too slow: {}ns (target < {}ns)", avg_ns, target);
	408	+ // Skip threshold check in Miri interpreter mode
	409	+ if !cfg!(miri) {
	410	+ let target = if cfg!(debug_assertions) { 5000 } else { 2000 };
	411	+ assert!(avg_ns < target, "Route lookup too slow: {}ns (target < {}ns)", avg_ns, target);
	412	+ }
411	413	}
412	414	}
413	415

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

+74-72

@@ -1610,8 +1610,13 @@ async fn serve_connection(
1610	1610	conn_metrics.connection_closed();
1611	1611	}
1612	1612
1613		-async fn serve_http(
1614		- listener: TcpListener,
	1613	+/// Shared accept loop logic for TCP, UDS, and TLS servers.
	1614	+/// Accepts connections from the listener, spawns `serve_connection` for each,
	1615	+/// and handles graceful shutdown.
	1616	+async fn serve_generic<S>(
	1617	+ mut accept_fn: impl FnMut() -> std::pin::Pin<
	1618	+ Box<dyn std::future::Future<Output = Result<(S, Option<SocketAddr>)>> + Send>,
	1619	+ >,
1615	1620	chain: Arc<MiddlewareChain>,
1616	1621	static_dir: Option<StaticDir>,
1617	1622	static_mounts: Vec<StaticMount>,
@@ -1620,10 +1625,10 @@ async fn serve_http(
1620	1625	shutdown_timeout: std::time::Duration,
1621	1626	#[cfg(feature = "wasm")] wasm_middleware: Option<Arc<crate::wasm::WasmEngine>>,
1622	1627	#[cfg(feature = "ws")] ws_handler: Option<WsHandler>,
1623		-) -> Result<()> {
1624		- // IMPORTANT: This duplicates `serve_unix` below. A future refactoring
1625		- // should extract the common accept loop into a generic function parameterized
1626		- // over the listener type. See P2-3 in deepanalysis.md.
	1628	+) -> Result<()>
	1629	+where
	1630	+ S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
	1631	+{
1627	1632	let mut connections = tokio::task::JoinSet::new();
1628	1633	let conn_semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(max_connections()));
1629	1634
@@ -1632,33 +1637,29 @@ async fn serve_http(
1632	1637
1633	1638	let (stream, peer) = if let Some(token) = &token {
1634	1639	tokio::select! {
1635		- result = listener.accept() => result?,
	1640	+ result = accept_fn() => result?,
1636	1641	_ = token.cancelled() => {
1637	1642	tracing::info!("Shutdown signal received, stopping accept loop");
1638	1643	break;
1639	1644	}
1640	1645	}
1641	1646	} else {
1642		- listener.accept().await?
	1647	+ accept_fn().await?
1643	1648	};
1644		- tracing::debug!("Accepted connection from {}", peer);
	1649	+ tracing::debug!("Accepted connection from {:?}", peer);
1645	1650
1646		- // --- Normal HTTP via hyper -----------------------------------------
1647	1651	let chain = chain.clone();
1648	1652	let static_dir = static_dir.clone();
1649	1653	let static_mounts = static_mounts.clone();
1650	1654	#[cfg(feature = "wasm")]
1651	1655	let wasm_middleware = wasm_middleware.clone();
1652		- let peer_addr = peer;
1653		-
1654	1656	#[cfg(feature = "ws")]
1655	1657	let ws_handler = ws_handler.clone();
1656		-
1657	1658	let token_clone = shutdown.clone();
1658	1659	let conn_semaphore = conn_semaphore.clone();
1659	1660	connections.spawn(serve_connection(
1660	1661	stream,
1661		- Some(peer_addr),
	1662	+ peer,
1662	1663	chain,
1663	1664	static_dir,
1664	1665	static_mounts,
@@ -1689,6 +1690,43 @@ async fn serve_http(
1689	1690	Ok(())
1690	1691	}
1691	1692
	1693	+async fn serve_http(
	1694	+ listener: TcpListener,
	1695	+ chain: Arc<MiddlewareChain>,
	1696	+ static_dir: Option<StaticDir>,
	1697	+ static_mounts: Vec<StaticMount>,
	1698	+ metrics: Metrics,
	1699	+ shutdown: Option<CancellationToken>,
	1700	+ shutdown_timeout: std::time::Duration,
	1701	+ #[cfg(feature = "wasm")] wasm_middleware: Option<Arc<crate::wasm::WasmEngine>>,
	1702	+ #[cfg(feature = "ws")] ws_handler: Option<WsHandler>,
	1703	+) -> Result<()> {
	1704	+ let local_addr = listener.local_addr()?;
	1705	+ tracing::info!("Listening on {}", local_addr);
	1706	+
	1707	+ let listener = Arc::new(listener);
	1708	+ serve_generic(
	1709	+ move \|\| {
	1710	+ let listener = listener.clone();
	1711	+ Box::pin(async move {
	1712	+ let (stream, peer) = listener.accept().await?;
	1713	+ Ok((stream, Some(peer)))
	1714	+ })
	1715	+ },
	1716	+ chain,
	1717	+ static_dir,
	1718	+ static_mounts,
	1719	+ metrics,
	1720	+ shutdown,
	1721	+ shutdown_timeout,
	1722	+ #[cfg(feature = "wasm")]
	1723	+ wasm_middleware,
	1724	+ #[cfg(feature = "ws")]
	1725	+ ws_handler,
	1726	+ )
	1727	+ .await
	1728	+}
	1729	+
1692	1730	/// Bind a Unix domain socket, removing any stale socket file left from a
1693	1731	/// previous run (a common footgun — `bind` fails with EADDRINUSE on an existing
1694	1732	/// socket path). The socket file is unlinked when the listener is dropped.
@@ -1742,65 +1780,29 @@ async fn serve_unix(
1742	1780	#[cfg(feature = "wasm")] wasm_middleware: Option<Arc<crate::wasm::WasmEngine>>,
1743	1781	#[cfg(feature = "ws")] ws_handler: Option<WsHandler>,
1744	1782	) -> Result<()> {
1745		- let mut connections = tokio::task::JoinSet::new();
1746		- let conn_semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(max_connections()));
1747		-
1748		- loop {
1749		- let token = shutdown.as_ref().cloned();
1750		-
1751		- let (stream, _peer) = if let Some(token) = &token {
1752		- tokio::select! {
1753		- result = listener.accept() => result?,
1754		- _ = token.cancelled() => {
1755		- tracing::info!("Shutdown signal received, stopping accept loop");
1756		- break;
1757		- }
1758		- }
1759		- } else {
1760		- listener.accept().await?
1761		- };
1762		-
1763		- let chain = chain.clone();
1764		- let static_dir = static_dir.clone();
1765		- let static_mounts = static_mounts.clone();
	1783	+ tracing::info!("Listening on Unix socket");
	1784	+
	1785	+ let listener = Arc::new(listener);
	1786	+ serve_generic(
	1787	+ move \|\| {
	1788	+ let listener = listener.clone();
	1789	+ Box::pin(async move {
	1790	+ let (stream, _peer) = listener.accept().await?;
	1791	+ Ok((stream, None))
	1792	+ })
	1793	+ },
	1794	+ chain,
	1795	+ static_dir,
	1796	+ static_mounts,
	1797	+ metrics,
	1798	+ shutdown,
	1799	+ shutdown_timeout,
1766	1800	#[cfg(feature = "wasm")]
1767		- let wasm_middleware = wasm_middleware.clone();
1768		-
	1801	+ wasm_middleware,
1769	1802	#[cfg(feature = "ws")]
1770		- let ws_handler = ws_handler.clone();
1771		-
1772		- let conn_semaphore = conn_semaphore.clone();
1773		- connections.spawn(serve_connection(
1774		- stream,
1775		- None,
1776		- chain,
1777		- static_dir,
1778		- static_mounts,
1779		- conn_semaphore,
1780		- metrics.clone(),
1781		- shutdown.clone(),
1782		- #[cfg(feature = "wasm")]
1783		- wasm_middleware,
1784		- #[cfg(feature = "ws")]
1785		- ws_handler,
1786		- ));
1787		- }
1788		-
1789		- if shutdown.is_some() {
1790		- tracing::info!("Waiting for {} active connections to drain...", connections.len());
1791		- tokio::select! {
1792		- _ = async { while connections.join_next().await.is_some() {} } => {
1793		- tracing::debug!("all connections closed gracefully");
1794		- }
1795		- _ = tokio::time::sleep(shutdown_timeout) => {
1796		- tracing::warn!("Graceful shutdown timeout exceeded. Dropping remaining connections.");
1797		- }
1798		- }
1799		- } else {
1800		- while connections.join_next().await.is_some() {}
1801		- }
1802		-
1803		- Ok(())
	1803	+ ws_handler,
	1804	+ )
	1805	+ .await
1804	1806	}
1805	1807
1806	1808	// ---------------------------------------------------------------------------
@@ -2058,7 +2060,7 @@ mod tests {
2058	2060	// Bind a Unix socket, serve a trivial handler over it, and confirm a raw
2059	2061	// HTTP/1.1 request over the socket gets a 200 response. Exercises the
2060	2062	// shared `serve_connection` path through `serve_unix`.
2061		- #[cfg(unix)]
	2063	+ #[cfg(all(unix, not(miri)))]
2062	2064	#[tokio::test]
2063	2065	async fn test_unix_socket_serves_http() {
2064	2066	use tokio::io::{AsyncReadExt, AsyncWriteExt};
2065	2067

▼crates/justapi-core/tests/edge_cases.rs

+2-2

@@ -156,11 +156,11 @@ async fn test_health_check_timeout() {
156	156
157	157	// Slow check (simulated)
158	158	registry.register_fn("slow", \|\| async {
159		- tokio::time::sleep(std::time::Duration::from_secs(10)).await;
	159	+ tokio::time::sleep(std::time::Duration::from_millis(200)).await;
160	160	HealthStatus::Healthy
161	161	});
162	162
163		- let report = registry.check_all().await;
	163	+ let report = registry.check_all_with_timeout(std::time::Duration::from_millis(50)).await;
164	164
165	165	// Fast check should pass
166	166	let fast = report.components.iter().find(\|c\| c.name == "fast");
167	167

swadhinbiswas