diff --git a/.config/nextest.toml b/.config/nextest.toml new file mode 100644 index 000000000..0abc58828 --- /dev/null +++ b/.config/nextest.toml @@ -0,0 +1,8 @@ +# GitHub-hosted ubuntu runner 当前按 4 vCPU 配置；固定线程数可避免 runner +# 规格变化时测试并发和内存峰值随之漂移。 +[profile.default] +test-threads = 4 + +# 60 秒后标记慢测试，连续两轮仍未结束则终止，避免单个卡死用例拖满整个 job。 +# 真实连接/数据库测试仍有足够时间完成；超时结果保持失败，不隐藏回归。 +slow-timeout = { period = "60s", terminate-after = 2, grace-period = "10s" } diff --git a/apps/aether-gateway/src/tests/control/admin/endpoints/keys.rs b/apps/aether-gateway/src/tests/control/admin/endpoints/keys.rs index 1043eb702..7b9afb6dd 100644 --- a/apps/aether-gateway/src/tests/control/admin/endpoints/keys.rs +++ b/apps/aether-gateway/src/tests/control/admin/endpoints/keys.rs @@ -44,21 +44,11 @@ where F: FnOnce() -> Fut + Send + 'static, Fut: std::future::Future + 'static, { - let handle = std::thread::Builder::new() - .name(test_name.to_string()) - .stack_size(PROVIDER_KEYS_TEST_STACK_BYTES) - .spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("test runtime should build"); - runtime.block_on(make_future()); - }) - .expect("provider keys test thread should spawn"); - - if let Err(payload) = handle.join() { - std::panic::resume_unwind(payload); - } + crate::tests::run_async_test_on_large_stack( + test_name, + PROVIDER_KEYS_TEST_STACK_BYTES, + make_future, + ); } struct SummaryNullingProviderCatalogReadRepository { diff --git a/apps/aether-gateway/src/tests/control/admin/endpoints/quota.rs b/apps/aether-gateway/src/tests/control/admin/endpoints/quota.rs index 37beb9b9a..2f9b61bd5 100644 --- a/apps/aether-gateway/src/tests/control/admin/endpoints/quota.rs +++ b/apps/aether-gateway/src/tests/control/admin/endpoints/quota.rs @@ -59,21 +59,11 @@ where F: FnOnce() -> Fut + Send + 'static, Fut: std::future::Future + 'static, { - let handle = std::thread::Builder::new() - .name(test_name.to_string()) - .stack_size(PROVIDER_QUOTA_TEST_STACK_BYTES) - .spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("test runtime should build"); - runtime.block_on(make_future()); - }) - .expect("provider quota test thread should spawn"); - - if let Err(payload) = handle.join() { - std::panic::resume_unwind(payload); - } + crate::tests::run_async_test_on_large_stack( + test_name, + PROVIDER_QUOTA_TEST_STACK_BYTES, + make_future, + ); } #[tokio::test] diff --git a/apps/aether-gateway/src/tests/control/admin/oauth.rs b/apps/aether-gateway/src/tests/control/admin/oauth.rs index 50e9cfc95..fea3ca978 100644 --- a/apps/aether-gateway/src/tests/control/admin/oauth.rs +++ b/apps/aether-gateway/src/tests/control/admin/oauth.rs @@ -56,21 +56,11 @@ where F: FnOnce() -> Fut + Send + 'static, Fut: std::future::Future + 'static, { - let handle = std::thread::Builder::new() - .name(test_name.to_string()) - .stack_size(ADMIN_OAUTH_TEST_STACK_BYTES) - .spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("test runtime should build"); - runtime.block_on(make_future()); - }) - .expect("admin oauth test thread should spawn"); - - if let Err(payload) = handle.join() { - std::panic::resume_unwind(payload); - } + crate::tests::run_async_test_on_large_stack( + test_name, + ADMIN_OAUTH_TEST_STACK_BYTES, + make_future, + ); } fn decrypt_persisted_provider_api_key(key: &StoredProviderCatalogKey) -> String { diff --git a/apps/aether-gateway/src/tests/control/admin/provider_ops.rs b/apps/aether-gateway/src/tests/control/admin/provider_ops.rs index ccc2fee1f..a9b2f5647 100644 --- a/apps/aether-gateway/src/tests/control/admin/provider_ops.rs +++ b/apps/aether-gateway/src/tests/control/admin/provider_ops.rs @@ -68,21 +68,11 @@ where F: FnOnce() -> Fut + Send + 'static, Fut: std::future::Future + 'static, { - let handle = std::thread::Builder::new() - .name(test_name.to_string()) - .stack_size(PROVIDER_OPS_TEST_STACK_BYTES) - .spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("test runtime should build"); - runtime.block_on(make_future()); - }) - .expect("provider ops test thread should spawn"); - - if let Err(payload) = handle.join() { - std::panic::resume_unwind(payload); - } + crate::tests::run_async_test_on_large_stack( + test_name, + PROVIDER_OPS_TEST_STACK_BYTES, + make_future, + ); } async fn start_managed_redis_or_skip() -> Option { diff --git a/apps/aether-gateway/src/tests/control/admin/provider_query.rs b/apps/aether-gateway/src/tests/control/admin/provider_query.rs index 801385e34..336063114 100644 --- a/apps/aether-gateway/src/tests/control/admin/provider_query.rs +++ b/apps/aether-gateway/src/tests/control/admin/provider_query.rs @@ -36,21 +36,11 @@ where F: FnOnce() -> Fut + Send + 'static, Fut: std::future::Future + 'static, { - let handle = std::thread::Builder::new() - .name(test_name.to_string()) - .stack_size(PROVIDER_QUERY_TEST_STACK_BYTES) - .spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("test runtime should build"); - runtime.block_on(make_future()); - }) - .expect("provider query test thread should spawn"); - - if let Err(payload) = handle.join() { - std::panic::resume_unwind(payload); - } + crate::tests::run_async_test_on_large_stack( + test_name, + PROVIDER_QUERY_TEST_STACK_BYTES, + make_future, + ); } fn crc32(data: &[u8]) -> u32 { diff --git a/apps/aether-gateway/src/tests/control/admin/security.rs b/apps/aether-gateway/src/tests/control/admin/security.rs index 2df981435..71b469cf3 100644 --- a/apps/aether-gateway/src/tests/control/admin/security.rs +++ b/apps/aether-gateway/src/tests/control/admin/security.rs @@ -239,11 +239,25 @@ async fn send_admin_security_request( method: reqwest::Method, path: &str, body: Option, +) -> (StatusCode, serde_json::Value, usize) { + let path = path.to_string(); + crate::tests::run_async_test_on_large_stack_with_result( + "admin-security-router-request", + 16 * 1024 * 1024, + move || send_admin_security_request_on_large_stack(gateway, method, path, body), + ) +} + +async fn send_admin_security_request_on_large_stack( + gateway: Router, + method: reqwest::Method, + path: String, + body: Option, ) -> (StatusCode, serde_json::Value, usize) { let upstream_hits = Arc::new(Mutex::new(0usize)); let upstream_hits_clone = Arc::clone(&upstream_hits); let upstream = Router::new().route( - path, + &path, any(move |_request: Request| { let upstream_hits_inner = Arc::clone(&upstream_hits_clone); async move { @@ -253,26 +267,52 @@ async fn send_admin_security_request( }), ); - let (upstream_url, upstream_handle) = start_server(upstream).await; - let (gateway_url, gateway_handle) = start_server(gateway).await; + let (_upstream_url, upstream_handle) = start_server(upstream).await; - let client = reqwest::Client::new(); - let mut request = client - .request(method, format!("{gateway_url}{path}")) + // 这些用例只验证本地安全路由和“不得转发”断言，不需要为 Gateway + // 再启动一个 TCP listener；send_request 会补齐 ConnectInfo，仍经过完整 Router。 + let mut request_builder = Request::builder() + .method(method.as_str()) + .uri(&path) .header(crate::constants::GATEWAY_HEADER, "rust-phase3b") .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123") .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin") .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123"); if let Some(body) = body { - request = request.json(&body); + request_builder = request_builder.header(http::header::CONTENT_TYPE, "application/json"); + let request = request_builder + .body(Body::from(body.to_string())) + .expect("request should build"); + let response = send_request(gateway, request).await; + let status = response.status(); + let payload = response + .into_body() + .collect() + .await + .expect("response body should collect") + .to_bytes(); + let payload: serde_json::Value = + serde_json::from_slice(&payload).expect("json body should parse"); + let upstream_count = *upstream_hits.lock().expect("mutex should lock"); + upstream_handle.abort(); + return (status, payload, upstream_count); } - let response = request.send().await.expect("request should succeed"); + let request = request_builder + .body(Body::empty()) + .expect("request should build"); + let response = send_request(gateway, request).await; let status = response.status(); - let payload: serde_json::Value = response.json().await.expect("json body should parse"); + let payload = response + .into_body() + .collect() + .await + .expect("response body should collect") + .to_bytes(); + let payload: serde_json::Value = + serde_json::from_slice(&payload).expect("json body should parse"); let upstream_count = *upstream_hits.lock().expect("mutex should lock"); - gateway_handle.abort(); upstream_handle.abort(); (status, payload, upstream_count) diff --git a/apps/aether-gateway/src/tests/control/admin/system_import.rs b/apps/aether-gateway/src/tests/control/admin/system_import.rs index 022a7ece6..8ad2871b1 100644 --- a/apps/aether-gateway/src/tests/control/admin/system_import.rs +++ b/apps/aether-gateway/src/tests/control/admin/system_import.rs @@ -341,21 +341,11 @@ where F: FnOnce() -> Fut + Send + 'static, Fut: std::future::Future + 'static, { - let handle = std::thread::Builder::new() - .name(test_name.to_string()) - .stack_size(ADMIN_SYSTEM_IMPORT_TEST_STACK_BYTES) - .spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("test runtime should build"); - runtime.block_on(make_future()); - }) - .expect("admin system import test thread should spawn"); - - if let Err(payload) = handle.join() { - std::panic::resume_unwind(payload); - } + crate::tests::run_async_test_on_large_stack( + test_name, + ADMIN_SYSTEM_IMPORT_TEST_STACK_BYTES, + make_future, + ); } #[test] diff --git a/apps/aether-gateway/src/tests/files/mod.rs b/apps/aether-gateway/src/tests/files/mod.rs index a9c9b1e7c..0acd7fb10 100644 --- a/apps/aether-gateway/src/tests/files/mod.rs +++ b/apps/aether-gateway/src/tests/files/mod.rs @@ -37,21 +37,7 @@ where F: FnOnce() -> Fut + Send + 'static, Fut: std::future::Future + 'static, { - let handle = std::thread::Builder::new() - .name(test_name.to_string()) - .stack_size(FILES_TEST_STACK_BYTES) - .spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("test runtime should build"); - runtime.block_on(make_future()); - }) - .expect("files test thread should spawn"); - - if let Err(payload) = handle.join() { - std::panic::resume_unwind(payload); - } + crate::tests::run_async_test_on_large_stack(test_name, FILES_TEST_STACK_BYTES, make_future); } fn hash_api_key(value: &str) -> String { diff --git a/apps/aether-gateway/src/tests/frontdoor.rs b/apps/aether-gateway/src/tests/frontdoor.rs index 15ae3a645..32a20d3e4 100644 --- a/apps/aether-gateway/src/tests/frontdoor.rs +++ b/apps/aether-gateway/src/tests/frontdoor.rs @@ -39,21 +39,7 @@ fn run_frontdoor_async_test(name: &'static str, future: F) where F: std::future::Future + Send + 'static, { - let handle = std::thread::Builder::new() - .name(name.to_string()) - .stack_size(16 * 1024 * 1024) - .spawn(move || { - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("frontdoor test runtime should build") - .block_on(future); - }) - .expect("large-stack frontdoor test thread should spawn"); - - if let Err(payload) = handle.join() { - std::panic::resume_unwind(payload); - } + crate::tests::run_async_test_on_large_stack(name, 16 * 1024 * 1024, || future); } fn hash_api_key(value: &str) -> String { diff --git a/apps/aether-gateway/src/tests/mod.rs b/apps/aether-gateway/src/tests/mod.rs index 2aa0bafef..4e93f587a 100644 --- a/apps/aether-gateway/src/tests/mod.rs +++ b/apps/aether-gateway/src/tests/mod.rs @@ -45,6 +45,50 @@ pub(super) async fn start_server(app: Router) -> (String, tokio::task::JoinHandl (format!("http://{addr}"), handle) } +/// 在独立的大栈线程中运行需要深调用栈的异步测试。 +/// +/// 这些测试仍保留 16 MiB 栈空间；这里只统一线程和 runtime 的启动逻辑， +/// 避免每个测试分区各自复制一份 helper，降低维护时误改测试执行语义的风险。 +pub(crate) fn run_async_test_on_large_stack( + test_name: &'static str, + stack_size: usize, + make_future: F, +) where + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future + 'static, +{ + run_async_test_on_large_stack_with_result(test_name, stack_size, make_future); +} + +/// 与上面的 helper 相同，但允许深栈测试返回结果，供公共请求 helper 使用。 +pub(crate) fn run_async_test_on_large_stack_with_result( + test_name: &'static str, + stack_size: usize, + make_future: F, +) -> R +where + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future + 'static, + R: Send + 'static, +{ + let handle = std::thread::Builder::new() + .name(test_name.to_string()) + .stack_size(stack_size) + .spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should build"); + runtime.block_on(make_future()) + }) + .expect("large-stack test thread should spawn"); + + match handle.join() { + Ok(result) => result, + Err(payload) => std::panic::resume_unwind(payload), + } +} + pub(super) const OPERATIONAL_ADMIN_DEVICE_ID: &str = "device-operational-admin"; pub(super) async fn start_authenticated_operational_server(