swadhin / justapi public

swadhinbiswas

[email protected]

2026-07-26 15:21:06 +0600 committed NaN years ago

ef04ee5

commit ef04ee5

4 files changed

+438 -59

Files

crates

justapi-core

src

server

mod.rs +112 -45 lib.rs +40 -11 middleware.rs +281 -2

tests

integration.rs +5 -1

Showing 4 changed files+438-59

crates/justapi-core/src/lib.rs crates/justapi-core/src/middleware.rs crates/justapi-core/src/server/mod.rs crates/justapi-core/tests/integration.rs

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

+40-11

@@ -76,29 +76,58 @@ pub fn json_response(status: hyper::StatusCode, body: &str) -> Response<Response
76	76	.expect("Response::builder with valid inputs should never fail")
77	77	}
78	78
79		-/// Canonical error envelope. Every non-2xx response in justapi uses this single
80		-/// shape so clients have one contract to parse (see ADR-052). `detail` carries
81		-/// the human-readable message; the numeric status lives in the HTTP status
82		-/// line. Sensitive internals must never be placed in `detail` for 5xx.
	79	+/// Canonical error envelope following RFC 9457 Problem Details. Every non-2xx
	80	+/// response in justapi uses this shape so clients have one contract to parse.
	81	+/// The `type` URI identifies the error class, `title` is a short human-readable
	82	+/// label, `status` is the HTTP status code, and `detail` carries the full message.
83	83	pub fn error_response(status: hyper::StatusCode, detail: &str) -> Response<ResponseBody> {
84		- json_response(status, &serde_json::json!({ "detail": detail }).to_string())
	84	+ let status_code: u16 = status.into();
	85	+ let title = match status_code {
	86	+ 400 => "Bad Request",
	87	+ 401 => "Unauthorized",
	88	+ 403 => "Forbidden",
	89	+ 404 => "Not Found",
	90	+ 405 => "Method Not Allowed",
	91	+ 413 => "Payload Too Large",
	92	+ 422 => "Unprocessable Entity",
	93	+ 429 => "Too Many Requests",
	94	+ 500 => "Internal Server Error",
	95	+ 502 => "Bad Gateway",
	96	+ 503 => "Service Unavailable",
	97	+ 504 => "Gateway Timeout",
	98	+ _ => "Error",
	99	+ };
	100	+ let body = serde_json::json!({
	101	+ "type": format!("https://justapi.dev/errors/{}", title.to_lowercase().replace(' ', "-")),
	102	+ "title": title,
	103	+ "status": status_code,
	104	+ "detail": detail,
	105	+ })
	106	+ .to_string();
	107	+ json_response(status, &body)
85	108	}
86	109
87		-/// 422 validation error using the canonical `{"detail": ...}` envelope.
	110	+/// 422 validation error using RFC 9457 Problem Details.
88	111	pub fn validation_response(detail: &str) -> Response<ResponseBody> {
89	112	error_response(hyper::StatusCode::UNPROCESSABLE_ENTITY, detail)
90	113	}
91	114
92		-/// 503 Service Unavailable with a `Retry-After` hint, using the canonical
93		-/// `{"detail": ...}` envelope. Used for backpressure signals — e.g. a saturated
94		-/// DB connection pool that cannot serve a request within its acquire window.
	115	+/// 503 Service Unavailable with a `Retry-After` hint, using RFC 9457 Problem Details.
95	116	pub fn service_unavailable_response(detail: &str) -> Response<ResponseBody> {
	117	+ let body = serde_json::json!({
	118	+ "type": "https://justapi.dev/errors/service-unavailable",
	119	+ "title": "Service Unavailable",
	120	+ "status": 503,
	121	+ "detail": detail,
	122	+ })
	123	+ .to_string();
96	124	Response::builder()
97	125	.status(hyper::StatusCode::SERVICE_UNAVAILABLE)
98		- .header("content-type", "application/json")
	126	+ .header("content-type", "application/problem+json")
99	127	.header("retry-after", "1")
	128	+ .header("content-length", body.len().to_string())
100	129	.body(UnsyncBoxBody::new(
101		- Full::new(Bytes::from(serde_json::json!({ "detail": detail }).to_string()))
	130	+ Full::new(Bytes::from(body))
102	131	.map_err(\|e: std::convert::Infallible\| -> anyhow::Error { match e {} }),
103	132	))
104	133	.expect("Response::builder with valid inputs should never fail")
105	134

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

+281-2

@@ -768,11 +768,149 @@ impl<B: Send + 'static> Middleware<B> for IpRateLimiter {
768	768	}
769	769	}
770	770
	771	+// ---------------------------------------------------------------------------
	772	+// Per-route rate limiting
	773	+// ---------------------------------------------------------------------------
	774	+
	775	+/// A single rate-limit rule matching a path prefix.
	776	+struct RateLimitRule {
	777	+ /// Path prefix to match (e.g. "/api/search"). Empty string matches all paths.
	778	+ prefix: String,
	779	+ /// Optional method filter. `None` means all methods.
	780	+ method: Option<Method>,
	781	+ limiter: governor::RateLimiter<
	782	+ governor::state::direct::NotKeyed,
	783	+ governor::state::InMemoryState,
	784	+ governor::clock::DefaultClock,
	785	+ >,
	786	+}
	787	+
	788	+/// Per-route rate limiter. Applies different rate limits to different path
	789	+/// prefixes and optionally HTTP methods.
	790	+///
	791	+/// # Example
	792	+///
	793	+/// ```ignore
	794	+/// use std::time::Duration;
	795	+///
	796	+/// let rl = PerRouteRateLimiter::new()
	797	+/// .add_rule("/api/search", None, Duration::from_secs(1), 100)
	798	+/// .add_rule("/api/export", Some(Method::POST), Duration::from_secs(1), 10);
	799	+/// ```
	800	+pub struct PerRouteRateLimiter {
	801	+ rules: Vec<RateLimitRule>,
	802	+ /// Fallback limiter applied when no rule matches. `None` means no limit.
	803	+ fallback: Option<RateLimiter>,
	804	+}
	805	+
	806	+impl PerRouteRateLimiter {
	807	+ pub fn new() -> Self {
	808	+ Self { rules: Vec::new(), fallback: None }
	809	+ }
	810	+
	811	+ /// Add a rate-limit rule for a path prefix.
	812	+ ///
	813	+ /// - `prefix`: Match paths starting with this string. Use `"/"` for all paths.
	814	+ /// - `method`: Optional HTTP method filter. `None` matches all methods.
	815	+ /// - `duration`: Time window for the rate limit (e.g. 1 second).
	816	+ /// - `max_burst`: Maximum burst size within the time window.
	817	+ pub fn add_rule(
	818	+ mut self,
	819	+ prefix: &str,
	820	+ method: Option<Method>,
	821	+ duration: Duration,
	822	+ max_burst: u32,
	823	+ ) -> Self {
	824	+ let burst = NonZeroU32::new(max_burst).unwrap_or(NonZeroU32::MIN);
	825	+ let quota = governor::Quota::with_period(duration)
	826	+ .expect("valid rate-limit duration for PerRouteRateLimiter")
	827	+ .allow_burst(burst);
	828	+ self.rules.push(RateLimitRule {
	829	+ prefix: prefix.to_string(),
	830	+ method,
	831	+ limiter: governor::RateLimiter::direct(quota),
	832	+ });
	833	+ self
	834	+ }
	835	+
	836	+ /// Set a fallback rate limiter for paths that don't match any rule.
	837	+ pub fn with_fallback(mut self, limiter: RateLimiter) -> Self {
	838	+ self.fallback = Some(limiter);
	839	+ self
	840	+ }
	841	+
	842	+ fn find_rule(&self, path: &str, method: &Method) -> Option<&RateLimitRule> {
	843	+ self.rules
	844	+ .iter()
	845	+ .find(\|r\| path.starts_with(&r.prefix) && r.method.as_ref().is_none_or(\|m\| m == method))
	846	+ }
	847	+}
	848	+
	849	+impl Default for PerRouteRateLimiter {
	850	+ fn default() -> Self {
	851	+ Self::new()
	852	+ }
	853	+}
	854	+
	855	+#[async_trait]
	856	+impl<B: Send + 'static> Middleware<B> for PerRouteRateLimiter {
	857	+ async fn handle(&self, req: Request<B>, next: Next<'_, B>) -> Result<Response<ResponseBody>> {
	858	+ let path = req.uri().path().to_string();
	859	+ let method = req.method().clone();
	860	+
	861	+ if let Some(rule) = self.find_rule(&path, &method) {
	862	+ match rule.limiter.check() {
	863	+ Ok(()) => next.run(req).await,
	864	+ Err(negative) => {
	865	+ let clock = governor::clock::DefaultClock::default();
	866	+ let now = clock.now();
	867	+ let wait = negative.wait_time_from(now);
	868	+ let retry_after_secs = wait.as_secs().max(1);
	869	+ let mut resp = json_error(StatusCode::TOO_MANY_REQUESTS, "rate limit exceeded");
	870	+ let retry_val = HeaderValue::from_str(&retry_after_secs.to_string())
	871	+ .unwrap_or_else(\|_\| HeaderValue::from_static("1"));
	872	+ resp.headers_mut().insert("retry-after", retry_val.clone());
	873	+ resp.headers_mut().insert("x-ratelimit-reset", retry_val);
	874	+ Ok(resp)
	875	+ }
	876	+ }
	877	+ } else if let Some(ref fallback) = self.fallback {
	878	+ // Delegate to the fallback global limiter
	879	+ fallback.handle(req, next).await
	880	+ } else {
	881	+ // No rule matches and no fallback — allow through
	882	+ next.run(req).await
	883	+ }
	884	+ }
	885	+}
	886	+
771	887	fn json_error(status: StatusCode, msg: &str) -> Response<ResponseBody> {
772		- let body = format!(r#"{{"error":"{}"}}"#, msg);
	888	+ let status_code: u16 = status.into();
	889	+ let title = match status_code {
	890	+ 400 => "Bad Request",
	891	+ 401 => "Unauthorized",
	892	+ 403 => "Forbidden",
	893	+ 404 => "Not Found",
	894	+ 405 => "Method Not Allowed",
	895	+ 413 => "Payload Too Large",
	896	+ 422 => "Unprocessable Entity",
	897	+ 429 => "Too Many Requests",
	898	+ 500 => "Internal Server Error",
	899	+ 502 => "Bad Gateway",
	900	+ 503 => "Service Unavailable",
	901	+ 504 => "Gateway Timeout",
	902	+ _ => "Error",
	903	+ };
	904	+ let body = serde_json::json!({
	905	+ "type": format!("https://justapi.dev/errors/{}", title.to_lowercase().replace(' ', "-")),
	906	+ "title": title,
	907	+ "status": status_code,
	908	+ "detail": msg,
	909	+ })
	910	+ .to_string();
773	911	Response::builder()
774	912	.status(status)
775		- .header("content-type", "application/json")
	913	+ .header("content-type", "application/problem+json")
776	914	.header("content-length", body.len().to_string())
777	915	.body(UnsyncBoxBody::new(
778	916	http_body_util::Full::new(Bytes::from(body))
@@ -1774,4 +1912,145 @@ mod tests {
1774	1912	let resp = chain.run(req).await.unwrap();
1775	1913	assert_eq!(resp.status(), StatusCode::OK);
1776	1914	}
	1915	+
	1916	+ // --- Per-route rate limiter tests (Phase 55.1) ---
	1917	+
	1918	+ #[tokio::test]
	1919	+ async fn per_route_rate_limit_allows_matching_prefix() {
	1920	+ let rl = PerRouteRateLimiter::new().add_rule(
	1921	+ "/api/",
	1922	+ None,
	1923	+ std::time::Duration::from_secs(1),
	1924	+ 10,
	1925	+ );
	1926	+ let handler: HandlerFn<TestBody> =
	1927	+ Arc::new(\|_\| Box::pin(async { Ok(json_response(StatusCode::OK, r#"{"ok":true}"#)) }));
	1928	+ let mut chain = MiddlewareChain::new(handler);
	1929	+ chain.add(rl);
	1930	+
	1931	+ let req = test_req(Method::GET, "/api/search");
	1932	+ let resp = chain.run(req).await.unwrap();
	1933	+ assert_eq!(resp.status(), StatusCode::OK);
	1934	+ }
	1935	+
	1936	+ #[tokio::test]
	1937	+ async fn per_route_rate_limit_rejects_when_exceeded() {
	1938	+ let rl = PerRouteRateLimiter::new().add_rule(
	1939	+ "/api/",
	1940	+ None,
	1941	+ std::time::Duration::from_secs(1),
	1942	+ 1,
	1943	+ );
	1944	+ let handler: HandlerFn<TestBody> =
	1945	+ Arc::new(\|_\| Box::pin(async { Ok(json_response(StatusCode::OK, r#"{"ok":true}"#)) }));
	1946	+ let mut chain = MiddlewareChain::new(handler);
	1947	+ chain.add(rl);
	1948	+
	1949	+ // First request should pass
	1950	+ let req = test_req(Method::GET, "/api/search");
	1951	+ let resp = chain.run(req).await.unwrap();
	1952	+ assert_eq!(resp.status(), StatusCode::OK);
	1953	+
	1954	+ // Second request should be rate limited
	1955	+ let req = test_req(Method::GET, "/api/search");
	1956	+ let resp = chain.run(req).await.unwrap();
	1957	+ assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
	1958	+ }
	1959	+
	1960	+ #[tokio::test]
	1961	+ async fn per_route_rate_limit_different_limits_per_prefix() {
	1962	+ let rl = PerRouteRateLimiter::new()
	1963	+ .add_rule("/api/fast", None, std::time::Duration::from_secs(1), 100)
	1964	+ .add_rule("/api/slow", None, std::time::Duration::from_secs(1), 1);
	1965	+ let handler: HandlerFn<TestBody> =
	1966	+ Arc::new(\|_\| Box::pin(async { Ok(json_response(StatusCode::OK, r#"{"ok":true}"#)) }));
	1967	+ let mut chain = MiddlewareChain::new(handler);
	1968	+ chain.add(rl);
	1969	+
	1970	+ // /api/fast should allow many requests
	1971	+ for _ in 0..5 {
	1972	+ let req = test_req(Method::GET, "/api/fast");
	1973	+ let resp = chain.run(req).await.unwrap();
	1974	+ assert_eq!(resp.status(), StatusCode::OK);
	1975	+ }
	1976	+
	1977	+ // /api/slow should limit after 1
	1978	+ let req = test_req(Method::GET, "/api/slow");
	1979	+ let resp = chain.run(req).await.unwrap();
	1980	+ assert_eq!(resp.status(), StatusCode::OK);
	1981	+ let req = test_req(Method::GET, "/api/slow");
	1982	+ let resp = chain.run(req).await.unwrap();
	1983	+ assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
	1984	+ }
	1985	+
	1986	+ #[tokio::test]
	1987	+ async fn per_route_rate_limit_method_filter() {
	1988	+ let rl = PerRouteRateLimiter::new().add_rule(
	1989	+ "/api/",
	1990	+ Some(Method::POST),
	1991	+ std::time::Duration::from_secs(1),
	1992	+ 1,
	1993	+ );
	1994	+ let handler: HandlerFn<TestBody> =
	1995	+ Arc::new(\|_\| Box::pin(async { Ok(json_response(StatusCode::OK, r#"{"ok":true}"#)) }));
	1996	+ let mut chain = MiddlewareChain::new(handler);
	1997	+ chain.add(rl);
	1998	+
	1999	+ // GET should pass (no rule matches)
	2000	+ let req = test_req(Method::GET, "/api/data");
	2001	+ let resp = chain.run(req).await.unwrap();
	2002	+ assert_eq!(resp.status(), StatusCode::OK);
	2003	+
	2004	+ // First POST should pass
	2005	+ let req = test_req(Method::POST, "/api/data");
	2006	+ let resp = chain.run(req).await.unwrap();
	2007	+ assert_eq!(resp.status(), StatusCode::OK);
	2008	+
	2009	+ // Second POST should be limited
	2010	+ let req = test_req(Method::POST, "/api/data");
	2011	+ let resp = chain.run(req).await.unwrap();
	2012	+ assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
	2013	+ }
	2014	+
	2015	+ #[tokio::test]
	2016	+ async fn per_route_rate_limit_no_fallback_allows_unmatched() {
	2017	+ let rl = PerRouteRateLimiter::new().add_rule(
	2018	+ "/api/",
	2019	+ None,
	2020	+ std::time::Duration::from_secs(1),
	2021	+ 1,
	2022	+ );
	2023	+ let handler: HandlerFn<TestBody> =
	2024	+ Arc::new(\|_\| Box::pin(async { Ok(json_response(StatusCode::OK, r#"{"ok":true}"#)) }));
	2025	+ let mut chain = MiddlewareChain::new(handler);
	2026	+ chain.add(rl);
	2027	+
	2028	+ // Unmatched path should always pass (no fallback, no rule)
	2029	+ for _ in 0..10 {
	2030	+ let req = test_req(Method::GET, "/other/path");
	2031	+ let resp = chain.run(req).await.unwrap();
	2032	+ assert_eq!(resp.status(), StatusCode::OK);
	2033	+ }
	2034	+ }
	2035	+
	2036	+ #[tokio::test]
	2037	+ async fn per_route_rate_limit_fallback_applies_to_unmatched() {
	2038	+ let rl = PerRouteRateLimiter::new()
	2039	+ .add_rule("/api/", None, std::time::Duration::from_secs(1), 1)
	2040	+ .with_fallback(RateLimiter::new(std::time::Duration::from_secs(1), 1));
	2041	+ let handler: HandlerFn<TestBody> =
	2042	+ Arc::new(\|_\| Box::pin(async { Ok(json_response(StatusCode::OK, r#"{"ok":true}"#)) }));
	2043	+ let mut chain = MiddlewareChain::new(handler);
	2044	+ chain.add(rl);
	2045	+
	2046	+ // First request to unmatched path passes (uses fallback)
	2047	+ let req = test_req(Method::GET, "/other/path");
	2048	+ let resp = chain.run(req).await.unwrap();
	2049	+ assert_eq!(resp.status(), StatusCode::OK);
	2050	+
	2051	+ // Second request to unmatched path is limited by fallback
	2052	+ let req = test_req(Method::GET, "/other/path");
	2053	+ let resp = chain.run(req).await.unwrap();
	2054	+ assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
	2055	+ }
1777	2056	}
1778	2057

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

+112-45

@@ -19,13 +19,14 @@ use crate::memory::BufferPool;
19	19	use crate::memory::SharedArena;
20	20	use crate::metrics::{self, Metrics};
21	21	use crate::middleware::{
22		- ApiKeyAuth, Cors, JwtAuth, MiddlewareChain, OAuth2Password, RateLimiter, SecurityHeaders,
	22	+ ApiKeyAuth, Cors, JwtAuth, MiddlewareChain, OAuth2Password, PerRouteRateLimiter, RateLimiter,
	23	+ SecurityHeaders,
23	24	};
24	25	use crate::openapi;
25	26	use crate::plugin::PluginRegistry;
26	27	use crate::router::Router;
27	28	use crate::static_files::{StaticDir, StaticMount};
28		-use crate::{json_response, streaming_response, ResponseBody};
	29	+use crate::{error_response, json_response, streaming_response, ResponseBody};
29	30	#[cfg(feature = "ws")]
30	31	use futures::StreamExt;
31	32	use serde_json;
@@ -354,6 +355,9 @@ pub struct Server {
354	355	/// Maximum accepted request body size in bytes (DoS guard). Configurable
355	356	/// via `with_max_body_size`; defaults to `DEFAULT_MAX_BODY_SIZE`.
356	357	max_body_size: usize,
	358	+ /// Graceful shutdown drain timeout. Configurable via `with_shutdown_timeout`;
	359	+ /// defaults to 30 seconds.
	360	+ shutdown_timeout: std::time::Duration,
357	361	}
358	362
359	363	impl Server {
@@ -394,6 +398,7 @@ impl Server {
394	398	default_routes_set: false,
395	399	openapi_spec: None,
396	400	max_body_size: DEFAULT_MAX_BODY_SIZE,
	401	+ shutdown_timeout: std::time::Duration::from_secs(30),
397	402	}
398	403	}
399	404
@@ -412,6 +417,15 @@ impl Server {
412	417	self
413	418	}
414	419
	420	+ /// Set the graceful shutdown drain timeout. When a shutdown signal is
	421	+ /// received, the server stops accepting new connections and waits up to
	422	+ /// this duration for in-flight requests to complete before dropping them.
	423	+ /// Defaults to 30 seconds.
	424	+ pub fn with_shutdown_timeout(mut self, timeout: std::time::Duration) -> Self {
	425	+ self.shutdown_timeout = timeout;
	426	+ self
	427	+ }
	428	+
415	429	/// Register a WebSocket handler. When a WebSocket upgrade is accepted on any
416	430	/// path, the handler is invoked with the path and the split stream halves.
417	431	#[cfg(feature = "ws")]
@@ -484,10 +498,7 @@ impl Server {
484	498	router.insert(Method::GET, "/events", Handler::Sse).expect("valid");
485	499	router.insert(Method::GET, "/graphql", Handler::GraphQL).expect("valid");
486	500	router.insert(Method::POST, "/graphql", Handler::GraphQL).expect("valid");
487		- router.set_fallback(Handler::Static {
488		- status: StatusCode::NOT_FOUND,
489		- body: r#"{"detail":"not found"}"#,
490		- });
	501	+ router.set_fallback(Handler::Static { status: StatusCode::NOT_FOUND, body: "not found" });
491	502	self.router = Some(router);
492	503	self.default_routes_set = true;
493	504	self
@@ -730,6 +741,26 @@ impl Server {
730	741	self
731	742	}
732	743
	744	+ /// Add a per-route rate limiter with different limits for different paths.
	745	+ ///
	746	+ /// # Example
	747	+ ///
	748	+ /// ```ignore
	749	+ /// use std::time::Duration;
	750	+ /// use hyper::Method;
	751	+ ///
	752	+ /// let server = Server::new("0.0.0.0:8080")
	753	+ /// .add_per_route_rate_limit(
	754	+ /// PerRouteRateLimiter::new()
	755	+ /// .add_rule("/api/search", None, Duration::from_secs(1), 100)
	756	+ /// .add_rule("/api/export", Some(Method::POST), Duration::from_secs(1), 10)
	757	+ /// );
	758	+ /// ```
	759	+ pub fn add_per_route_rate_limit(mut self, rl: PerRouteRateLimiter) -> Self {
	760	+ self.chain.add(rl);
	761	+ self
	762	+ }
	763	+
733	764	/// Add JWT authentication middleware.
734	765	pub fn add_jwt(mut self, jwt: JwtAuth) -> Self {
735	766	self.chain.add(jwt);
@@ -968,6 +999,7 @@ impl Server {
968	999	self.static_mounts.clone(),
969	1000	metrics,
970	1001	self.shutdown,
	1002	+ self.shutdown_timeout,
971	1003	wasm_middleware,
972	1004	ws_handler,
973	1005	)
@@ -980,6 +1012,7 @@ impl Server {
980	1012	self.static_mounts.clone(),
981	1013	metrics,
982	1014	self.shutdown,
	1015	+ self.shutdown_timeout,
983	1016	wasm_middleware,
984	1017	)
985	1018	.await;
@@ -1073,6 +1106,7 @@ impl Server {
1073	1106	self.static_mounts.clone(),
1074	1107	metrics,
1075	1108	self.shutdown,
	1109	+ self.shutdown_timeout,
1076	1110	wasm_middleware,
1077	1111	ws_handler,
1078	1112	)
@@ -1085,6 +1119,7 @@ impl Server {
1085	1119	self.static_mounts.clone(),
1086	1120	metrics,
1087	1121	self.shutdown,
	1122	+ self.shutdown_timeout,
1088	1123	wasm_middleware,
1089	1124	)
1090	1125	.await;
@@ -1099,10 +1134,7 @@ impl Server {
1099	1134
1100	1135	pub async fn serve(listener: TcpListener) -> Result<()> {
1101	1136	let mut router: Router<Handler> = Router::new();
1102		- router.set_fallback(Handler::Static {
1103		- status: StatusCode::NOT_FOUND,
1104		- body: r#"{"detail":"not found"}"#,
1105		- });
	1137	+ router.set_fallback(Handler::Static { status: StatusCode::NOT_FOUND, body: "not found" });
1106	1138	router
1107	1139	.insert(
1108	1140	hyper::Method::GET,
@@ -1126,11 +1158,30 @@ pub async fn serve(listener: TcpListener) -> Result<()> {
1126	1158	let chain = Arc::new(MiddlewareChain::new(handler));
1127	1159
1128	1160	#[cfg(feature = "ws")]
1129		- let res =
1130		- serve_http(listener, chain, None, Vec::new(), metrics, None, None, Some(default_ws_echo()))
1131		- .await;
	1161	+ let res = serve_http(
	1162	+ listener,
	1163	+ chain,
	1164	+ None,
	1165	+ Vec::new(),
	1166	+ metrics,
	1167	+ None,
	1168	+ std::time::Duration::from_secs(30),
	1169	+ None,
	1170	+ Some(default_ws_echo()),
	1171	+ )
	1172	+ .await;
1132	1173	#[cfg(not(feature = "ws"))]
1133		- let res = serve_http(listener, chain, None, Vec::new(), metrics, None, None).await;
	1174	+ let res = serve_http(
	1175	+ listener,
	1176	+ chain,
	1177	+ None,
	1178	+ Vec::new(),
	1179	+ metrics,
	1180	+ None,
	1181	+ std::time::Duration::from_secs(30),
	1182	+ None,
	1183	+ )
	1184	+ .await;
1134	1185	res
1135	1186	}
1136	1187
@@ -1174,10 +1225,9 @@ fn make_handler(
1174	1225	)
1175	1226	.await
1176	1227	}
1177		- Err(crate::router::RouterError::MethodNotAllowed) => Ok(json_response(
1178		- StatusCode::METHOD_NOT_ALLOWED,
1179		- r#"{"detail":"method not allowed"}"#,
1180		- )),
	1228	+ Err(crate::router::RouterError::MethodNotAllowed) => {
	1229	+ Ok(error_response(StatusCode::METHOD_NOT_ALLOWED, "method not allowed"))
	1230	+ }
1181	1231	Err(crate::router::RouterError::NotFound) => match router.fallback() {
1182	1232	Some(fb) => {
1183	1233	execute_handler(
@@ -1193,7 +1243,7 @@ fn make_handler(
1193	1243	)
1194	1244	.await
1195	1245	}
1196		- None => Ok(json_response(StatusCode::NOT_FOUND, r#"{"detail":"not found"}"#)),
	1246	+ None => Ok(error_response(StatusCode::NOT_FOUND, "not found")),
1197	1247	},
1198	1248	}
1199	1249	})
@@ -1500,7 +1550,7 @@ async fn serve_connection(
1500	1550	{
1501	1551	return Ok::<_, anyhow::Error>(resp);
1502	1552	}
1503		- Ok(json_response(StatusCode::NOT_FOUND, r#"{"detail":"not found"}"#))
	1553	+ Ok(error_response(StatusCode::NOT_FOUND, "not found"))
1504	1554	}
1505	1555	},
1506	1556	Err(_) => {
@@ -1514,10 +1564,7 @@ async fn serve_connection(
1514	1564	path,
1515	1565	request_timeout()
1516	1566	);
1517		- Ok(json_response(
1518		- StatusCode::GATEWAY_TIMEOUT,
1519		- r#"{"detail":"request timeout"}"#,
1520		- ))
	1567	+ Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "request timeout"))
1521	1568	}
1522	1569	}
1523	1570	}
@@ -1559,6 +1606,7 @@ async fn serve_http(
1559	1606	static_mounts: Vec<StaticMount>,
1560	1607	metrics: Metrics,
1561	1608	shutdown: Option<CancellationToken>,
	1609	+ shutdown_timeout: std::time::Duration,
1562	1610	wasm_middleware: Option<Arc<crate::wasm::WasmEngine>>,
1563	1611	#[cfg(feature = "ws")] ws_handler: Option<WsHandler>,
1564	1612	) -> Result<()> {
@@ -1614,7 +1662,7 @@ async fn serve_http(
1614	1662	_ = async { while connections.join_next().await.is_some() {} } => {
1615	1663	tracing::debug!("all connections closed gracefully");
1616	1664	}
1617		- _ = tokio::time::sleep(std::time::Duration::from_secs(30)) => {
	1665	+ _ = tokio::time::sleep(shutdown_timeout) => {
1618	1666	tracing::warn!("Graceful shutdown timeout exceeded. Dropping remaining connections.");
1619	1667	}
1620	1668	}
@@ -1670,6 +1718,7 @@ async fn serve_unix(
1670	1718	static_mounts: Vec<StaticMount>,
1671	1719	metrics: Metrics,
1672	1720	shutdown: Option<CancellationToken>,
	1721	+ shutdown_timeout: std::time::Duration,
1673	1722	wasm_middleware: Option<Arc<crate::wasm::WasmEngine>>,
1674	1723	#[cfg(feature = "ws")] ws_handler: Option<WsHandler>,
1675	1724	) -> Result<()> {
@@ -1721,7 +1770,7 @@ async fn serve_unix(
1721	1770	_ = async { while connections.join_next().await.is_some() {} } => {
1722	1771	tracing::debug!("all connections closed gracefully");
1723	1772	}
1724		- _ = tokio::time::sleep(std::time::Duration::from_secs(30)) => {
	1773	+ _ = tokio::time::sleep(shutdown_timeout) => {
1725	1774	tracing::warn!("Graceful shutdown timeout exceeded. Dropping remaining connections.");
1726	1775	}
1727	1776	}
@@ -1765,7 +1814,13 @@ async fn execute_handler(
1765	1814	let _handler_span = tracing::debug_span!("handler.execute", name = handler_name);
1766	1815
1767	1816	match handler {
1768		- Handler::Static { status, body } => Ok(json_response(status, body)),
	1817	+ Handler::Static { status, body } => {
	1818	+ if status.is_success() {
	1819	+ Ok(json_response(status, body))
	1820	+ } else {
	1821	+ Ok(crate::error_response(status, body))
	1822	+ }
	1823	+ }
1769	1824	Handler::Echo => {
1770	1825	let body_bytes = match http_body_util::Limited::new(req.into_body(), max_body_size)
1771	1826	.collect()
@@ -1777,17 +1832,9 @@ async fn execute_handler(
1777	1832	bytes
1778	1833	}
1779	1834	Err(e) if e.to_string().contains("length limit") => {
1780		- return Ok(json_response(
1781		- StatusCode::PAYLOAD_TOO_LARGE,
1782		- r#"{"detail":"payload too large"}"#,
1783		- ))
1784		- }
1785		- Err(_) => {
1786		- return Ok(json_response(
1787		- StatusCode::BAD_REQUEST,
1788		- r#"{"detail":"bad request"}"#,
1789		- ))
	1835	+ return Ok(error_response(StatusCode::PAYLOAD_TOO_LARGE, "payload too large"))
1790	1836	}
	1837	+ Err(_) => return Ok(error_response(StatusCode::BAD_REQUEST, "bad request")),
1791	1838	};
1792	1839	let mut buf = pool.acquire(body_bytes.len());
1793	1840	buf.extend_from_slice(&body_bytes);
@@ -2147,9 +2194,9 @@ async fn serve_with_tls(
2147	2194	if let Some(resp) = try_serve_static(&path, &static_dir, &static_mounts).await {
2148	2195	return Ok::<_, anyhow::Error>(resp);
2149	2196	}
2150		- Ok(json_response(
	2197	+ Ok(error_response(
2151	2198	StatusCode::NOT_FOUND,
2152		- r#"{"detail":"not found"}"#,
	2199	+ "not found",
2153	2200	))
2154	2201	},
2155	2202	},
@@ -2163,9 +2210,9 @@ async fn serve_with_tls(
2163	2210	path,
2164	2211	request_timeout()
2165	2212	);
2166		- Ok(json_response(
	2213	+ Ok(error_response(
2167	2214	StatusCode::GATEWAY_TIMEOUT,
2168		- r#"{"detail":"request timeout"}"#
	2215	+ "request timeout",
2169	2216	))
2170	2217	}
2171	2218	}
@@ -2210,7 +2257,7 @@ async fn serve_with_tls(
2210	2257	_ = async { while connections.join_next().await.is_some() {} } => {
2211	2258	tracing::debug!("all connections closed after fallback drain");
2212	2259	}
2213		- _ = tokio::time::sleep(std::time::Duration::from_secs(30)) => {
	2260	+ _ = tokio::time::sleep(shutdown_timeout) => {
2214	2261	tracing::warn!("Graceful shutdown timeout exceeded. Dropping remaining TLS connections.");
2215	2262	}
2216	2263	}
@@ -2324,10 +2371,30 @@ mod tests {
2324	2371
2325	2372	let server = tokio::spawn(async move {
2326	2373	#[cfg(feature = "ws")]
2327		- let _ = serve_unix(listener, chain, None, Vec::new(), Metrics::new(), None, None, None)
2328		- .await;
	2374	+ let _ = serve_unix(
	2375	+ listener,
	2376	+ chain,
	2377	+ None,
	2378	+ Vec::new(),
	2379	+ Metrics::new(),
	2380	+ None,
	2381	+ std::time::Duration::from_secs(30),
	2382	+ None,
	2383	+ None,
	2384	+ )
	2385	+ .await;
2329	2386	#[cfg(not(feature = "ws"))]
2330		- let _ = serve_unix(listener, chain, None, Vec::new(), Metrics::new(), None, None).await;
	2387	+ let _ = serve_unix(
	2388	+ listener,
	2389	+ chain,
	2390	+ None,
	2391	+ Vec::new(),
	2392	+ Metrics::new(),
	2393	+ None,
	2394	+ std::time::Duration::from_secs(30),
	2395	+ None,
	2396	+ )
	2397	+ .await;
2331	2398	});
2332	2399
2333	2400	// Give the acceptor a moment to start.
2334	2401

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

+5-1

@@ -81,7 +81,11 @@ async fn test_loopback_404() {
81	81	let addr = start_server().await;
82	82	let (status, body) = send_request(addr, Method::GET, "/nonexistent", None).await;
83	83	assert_eq!(status, StatusCode::NOT_FOUND);
84		- assert_eq!(&body[..], br#"{"detail":"not found"}"#);
	84	+ let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
	85	+ assert_eq!(v["status"], 404);
	86	+ assert_eq!(v["title"], "Not Found");
	87	+ assert_eq!(v["detail"], "not found");
	88	+ assert!(v["type"].as_str().unwrap().contains("not-found"));
85	89	}
86	90
87	91	#[tokio::test]
88	92