diff --git a/Cargo.lock b/Cargo.lock index d37a442e..3c28aa44 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -470,6 +470,7 @@ dependencies = [ "async-stream", "async-trait", "beyond-ai-providers", + "beyond-ai-test-support", "bytes", "divan", "futures", diff --git a/crates/agent-core/Cargo.toml b/crates/agent-core/Cargo.toml index 28729f1f..2bc21ae3 100644 --- a/crates/agent-core/Cargo.toml +++ b/crates/agent-core/Cargo.toml @@ -91,6 +91,8 @@ rustls = { version = "0.23", default-features = false, features = ["ring"] } zstd = "0.13" [dev-dependencies] +# Loopback listeners for the socket tests (`ports`): the one place a test binds port 0. +beyond-ai-test-support = { path = "../test-support" } # Bench harness for the streaming decode hot path. `divan` (the workspace's `bench:unit` pin, matching # gateway/agent) measures **allocations** natively via its `AllocProfiler` global allocator alongside # timing — so the per-SSE-event allocation cost the decode findings target is visible, not inferred. diff --git a/crates/agent-core/src/client.rs b/crates/agent-core/src/client.rs index 1371b00a..5a8caaa4 100644 --- a/crates/agent-core/src/client.rs +++ b/crates/agent-core/src/client.rs @@ -2082,7 +2082,7 @@ mod tests { #[tokio::test] async fn oauth_credential_adds_anthropics_identity_headers() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -2131,7 +2131,7 @@ mod tests { #[tokio::test] async fn a_static_credential_never_sends_oauth_headers() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -2174,7 +2174,7 @@ mod tests { /// the wire, as opposed to anything this crate constructs itself. #[tokio::test] async fn a_connection_dropped_mid_body_is_tagged_as_a_mid_stream_network_error() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -2222,7 +2222,7 @@ mod tests { /// than being a dead field, since the default [`READ_TIMEOUT`] (600s) would never trip in a test. #[tokio::test] async fn with_idle_timeout_overrides_the_default_read_timeout() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -2304,7 +2304,7 @@ mod tests { /// `anthropic-version` for a self-hosted/proxied endpoint that speaks a different wire version). #[tokio::test] async fn extra_headers_are_merged_onto_the_request_and_win_on_a_name_collision() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new())); let captured_clone = captured.clone(); @@ -2366,7 +2366,7 @@ mod tests { } } - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -2424,7 +2424,7 @@ mod tests { } } - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let captured = Arc::new(std::sync::Mutex::new(Vec::new())); let captured_clone = captured.clone(); @@ -2473,7 +2473,7 @@ mod tests { } } - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let captured = Arc::new(std::sync::Mutex::new(Vec::new())); let captured_clone = captured.clone(); @@ -2519,7 +2519,7 @@ mod tests { } } - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -2567,7 +2567,7 @@ mod tests { /// this shutdown was actually a transport fault. #[tokio::test] async fn a_clean_early_close_before_the_terminal_event_is_not_tagged_as_a_network_error() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -2613,7 +2613,7 @@ mod tests { /// never fix. #[tokio::test] async fn a_429_with_quota_exhaustion_body_is_not_retried() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let request_count = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)); let count = request_count.clone(); @@ -2657,7 +2657,7 @@ mod tests { /// normal retry treatment `is_quota_exhausted` doesn't touch. #[tokio::test] async fn a_429_with_a_plain_rate_limit_body_is_retried() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let request_count = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)); let count = request_count.clone(); @@ -2719,7 +2719,7 @@ mod tests { /// headers when compat.sendSessionAffinityHeaders is enabled" case. #[tokio::test] async fn session_affinity_headers_are_sent_for_a_fireworks_chat_completions_request() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new())); let captured_clone = captured.clone(); @@ -2769,7 +2769,7 @@ mod tests { /// provider got these headers indiscriminately. #[tokio::test] async fn session_affinity_headers_are_absent_for_a_non_fireworks_chat_completions_request() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new())); let captured_clone = captured.clone(); @@ -2821,7 +2821,7 @@ mod tests { /// dedicated negative test above for that). #[tokio::test] async fn session_affinity_headers_are_suppressed_when_no_cache_is_set() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new())); let captured_clone = captured.clone(); @@ -2876,7 +2876,7 @@ mod tests { /// never reach the wire with that call still unanswered. #[tokio::test] async fn an_orphaned_tool_use_from_a_cancelled_mid_argument_stream_never_reaches_the_wire() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new())); let captured_clone = captured.clone(); @@ -2937,7 +2937,7 @@ mod tests { /// matching `tool_use` id) rather than on substrings, so it fails for the right reason. #[tokio::test] async fn an_aborted_turns_tool_use_never_reaches_the_wire_as_a_dangling_tool_result() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new())); let captured_clone = captured.clone(); @@ -3022,7 +3022,7 @@ mod tests { #[tokio::test] async fn prefixed_route_override_reaches_the_gateway_under_the_provider_prefix_with_static_headers() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3108,7 +3108,7 @@ mod tests { /// Anthropic's own OAuth identity-spoofing headers (this is a Copilot host, not Anthropic's). #[tokio::test] async fn direct_route_override_bypasses_the_gateway_base_url_and_attaches_copilot_headers() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3192,7 +3192,7 @@ mod tests { #[tokio::test] async fn a_copilot_routed_claude_model_gets_the_browser_access_header_but_not_claude_code_identity() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3259,7 +3259,7 @@ mod tests { /// Chat Completions body (`"messages"`), never a Responses body (`"input"`). #[tokio::test] async fn a_copilot_routed_gpt_4_1_builds_a_chat_completions_body_not_a_responses_body() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3318,7 +3318,7 @@ mod tests { /// Responses dialect when reached through any route but Copilot's. #[tokio::test] async fn a_non_copilot_gpt_4_1_still_builds_a_responses_body() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3352,7 +3352,7 @@ mod tests { /// proves the wiring from a real Copilot-routed credential through to that flag. #[tokio::test] async fn copilot_routed_gpt5_2_omits_the_explicit_reasoning_disable_end_to_end() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3404,7 +3404,7 @@ mod tests { /// for the header shape — this proves it also suppresses `prompt_cache_retention` end to end. #[tokio::test] async fn azure_routed_request_omits_prompt_cache_retention_end_to_end() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3459,7 +3459,7 @@ mod tests { /// reasoning-"off" wire value for) would incorrectly send the native `{"effort":"none"}` signal. #[tokio::test] async fn entra_id_shaped_config_with_only_deployment_name_still_suppresses_reasoning_disable() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3526,7 +3526,7 @@ mod tests { /// (`{base_url}/responses`, base_url already carrying `/openai/v1`). #[tokio::test] async fn direct_route_with_custom_auth_header_sends_bare_key_and_omits_authorization() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3594,7 +3594,7 @@ mod tests { /// `inferCopilotInitiator`/`hasCopilotVisionInput`. #[tokio::test] async fn copilot_dynamic_headers_reflect_the_turns_own_messages() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3665,7 +3665,7 @@ mod tests { /// `input`, with `parallel_tool_calls`/`text.verbosity` always present, on the real wire body. #[tokio::test] async fn codex_routed_request_sends_instructions_and_parallel_tool_calls_end_to_end() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3760,7 +3760,7 @@ mod tests { /// third-party provider (Kimi-Coding) would hit. #[tokio::test] async fn dialect_override_forces_anthropic_wire_for_a_model_id_that_fails_the_name_heuristic() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3817,7 +3817,7 @@ mod tests { /// Anthropic's `anthropic-version` header. #[tokio::test] async fn without_a_dialect_override_the_same_model_id_builds_chat_completions_wire() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3861,7 +3861,7 @@ mod tests { /// OpenAI-wire endpoint, a hard 400. #[tokio::test] async fn aggregator_host_from_a_byo_override_resolves_a_real_bare_id_host_collision() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3914,7 +3914,7 @@ mod tests { #[tokio::test] async fn without_an_aggregator_host_the_same_bare_id_keeps_the_host_agnostic_anthropic_default() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3947,7 +3947,7 @@ mod tests { /// URL — an Azure resource pinned to a dated `api-version` needs this on every request. #[tokio::test] async fn direct_route_query_param_is_appended_to_the_built_url() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -3999,7 +3999,7 @@ mod tests { /// Azure deployment name doesn't have to match the app-level model id. #[tokio::test] async fn deployment_name_override_replaces_the_wire_level_model_field() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; @@ -4060,7 +4060,7 @@ mod tests { /// a Bearer-prefixed value (unlike Azure's). #[tokio::test] async fn auth_header_prefix_is_prepended_to_the_credential_value() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { use std::io::{Read, Write}; diff --git a/crates/agent-core/src/codex_websocket.rs b/crates/agent-core/src/codex_websocket.rs index 4e9796fa..e3b2a5b9 100644 --- a/crates/agent-core/src/codex_websocket.rs +++ b/crates/agent-core/src/codex_websocket.rs @@ -1306,9 +1306,7 @@ mod tests { /// reads an entry's `created_at` and drops the rest, so no handshake or peer is needed — but the /// type is concrete, so *some* socket is. async fn parked_socket() -> WsSocket { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") - .await - .expect("loopback bind"); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().expect("bound port"); let client = tokio::net::TcpStream::connect(addr).await.expect("connect"); listener.accept().await.expect("accept"); diff --git a/crates/agent-core/tests/client_socket.rs b/crates/agent-core/tests/client_socket.rs index 1a56c2c0..831b297b 100644 --- a/crates/agent-core/tests/client_socket.rs +++ b/crates/agent-core/tests/client_socket.rs @@ -7,7 +7,6 @@ #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] use std::io::{Read, Write}; -use std::net::TcpListener; use std::thread; use agent_core::message::StreamEvent; @@ -17,7 +16,7 @@ use futures::StreamExt; /// Spawn a one-shot server that returns `body` as an SSE response, and return its base URL. fn spawn_sse_server(body: &'static str) -> String { - let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { @@ -102,7 +101,7 @@ data: {\"type\":\"message_stop\"}\n\ async fn gateway_client_reassembles_utf8_split_across_chunks() { use std::time::Duration; - let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { @@ -145,7 +144,7 @@ async fn gateway_client_reassembles_utf8_split_across_chunks() { #[tokio::test] async fn gateway_client_surfaces_http_error() { - let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { @@ -175,7 +174,7 @@ async fn gateway_client_gives_actionable_guidance_on_a_401() { // "gateway returned 401: " — identical formatting to any other 4xx, with no hint at what to // actually do about it. Unlike a 403 (still just a generic error, see the test above), 401 // specifically must name the actual cause and point at the fix. - let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { @@ -215,7 +214,7 @@ async fn gateway_client_bounds_memory_on_an_oversized_error_body() { // memory ahead of the display truncation. `Content-Length` is set so the body isn't itself framed // as an SSE stream — this exercises the non-2xx error path, not `LineFramer`. const BODY_LEN: usize = 3 * 1024 * 1024; // comfortably over the 1MB read cap - let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { @@ -262,7 +261,7 @@ async fn gateway_client_bounds_memory_on_an_oversized_error_body() { /// Capture the raw bytes of the first request a one-shot server receives, then answer with a minimal /// empty-body SSE response. Returns the shared buffer the caller reads after driving the request. fn spawn_request_capturing_server() -> (String, std::sync::Arc>) { - let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new())); let captured2 = captured.clone(); @@ -365,7 +364,7 @@ async fn gateway_client_retries_transient_503_then_succeeds() { // The server returns a retryable 503 on the first connection and the real SSE body on the second. // The client must transparently retry and deliver the events — a transient gateway hiccup should // not vaporize the request. - let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); thread::spawn(move || { // First attempt: 503 with a tiny Retry-After so the backoff stays fast. @@ -417,7 +416,7 @@ async fn gateway_client_retries_a_connection_reset_mid_send_then_succeeds() { // then dropped with zero bytes written back — exactly that reset-after-connect shape, not a 503 or // a timeout — so this is real coverage for a genuinely different failure mode than the sibling test // above, not a duplicate of it. - let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); thread::spawn(move || { // First attempt: accept, then drop immediately — a connection reset mid-send, not a timeout and diff --git a/crates/agent-core/tests/codex_websocket_socket.rs b/crates/agent-core/tests/codex_websocket_socket.rs index a79cbf6a..0ef7eeb8 100644 --- a/crates/agent-core/tests/codex_websocket_socket.rs +++ b/crates/agent-core/tests/codex_websocket_socket.rs @@ -20,7 +20,6 @@ use agent_core::{GatewayClient, Message, ModelRequest}; use futures::{SinkExt, StreamExt}; use serde_json::{Value, json}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::TcpListener; use tokio_tungstenite::WebSocketStream; use tokio_tungstenite::tungstenite::Message as WsMessage; @@ -104,7 +103,7 @@ async fn recv_request(ws: &mut WebSocketStream) -> Value #[tokio::test] async fn first_turn_sends_the_full_transcript_with_no_previous_response_id() { - let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().expect("addr"); let server = tokio::spawn(async move { @@ -154,7 +153,7 @@ async fn first_turn_sends_the_full_transcript_with_no_previous_response_id() { #[tokio::test] async fn second_turn_on_the_same_session_sends_only_the_new_delta() { - let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().expect("addr"); let server = tokio::spawn(async move { @@ -239,7 +238,7 @@ async fn second_turn_on_the_same_session_sends_only_the_new_delta() { #[tokio::test] async fn a_failed_websocket_handshake_falls_back_to_http_sse_transparently() { - let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().expect("addr"); let server = tokio::spawn(async move { @@ -313,7 +312,7 @@ async fn a_failed_websocket_handshake_falls_back_to_http_sse_transparently() { /// corrupt a binary zstd body) to check the header and decompress the body directly. #[tokio::test] async fn the_http_sse_fallback_sends_a_zstd_compressed_body_with_the_matching_header() { - let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().expect("addr"); let server = tokio::spawn(async move { diff --git a/crates/agent/ARCHITECTURE.md b/crates/agent/ARCHITECTURE.md index 720b0e42..17c57cef 100644 --- a/crates/agent/ARCHITECTURE.md +++ b/crates/agent/ARCHITECTURE.md @@ -4239,7 +4239,7 @@ backend an operator points it at. | `src/bin/mcp_skills_fixture_server.rs` | Hand-rolled SEP-2640 server for `tests/mcp_skills*.rs` (stdio, or `--http`): an ordinary skill with a supporting file and a nested skill, a same-name pair, a `disable-model-invocation` skill, a tampered one, a dynamic one, an unlisted one, two over the per-skill limits, and a late one published by its `publish_late` tool (with `notifications/resources/list_changed`); caching hints from env; `_meta` on every result like a `2026-07-28` server; logs every request so tests assert what was (not) fetched | | `benches/search.rs` | Criterion macro-bench: `grep` (1 vs auto threads) and `find` (sequential) over a 5,000-file tree | | `benches/serve_runtime.rs` | Process-level Tokio runtime A/B for `serve --listen`: spawns the real binary under `current_thread` vs multi-thread (`BEYOND_AI_AGENT_TOKIO_WORKER_THREADS`) and drives concurrent WebSocket sessions, reporting RSS, throughput, p95 latency, and CPU. Custom harness, not Criterion/divan — the cost is thread stacks and per-thread heaps, which an in-process alloc profiler never sees. | -| `tests/common/mod.rs` | Shared test harness: mock Anthropic-SSE model server (an in-order server answers `serve`'s session-title request itself, off the conversation script and out of its record, counting them for `title_calls`; a test about titles matches `SESSION_TITLE_MARKER` on a routed one), ports with no window for another process to take them (`spawn_listening`: `serve --listen 127.0.0.1:0`, the bound port read from its announcement; `HeldPort`: a listener the test binds and hands over by socket activation, for a port named in `serve`'s own arguments; `DeadPort`: bound, never listening; the gateway binds `127.0.0.1:0`/`127.0.0.2:0` (`GATEWAY_LISTENERS`, distinct addresses because Pingora keys listeners by address string) and `spawn_gateway` reads both ports from its own `LISTEN` sockets in `/proc`; the shared `nats-server` picks its own port (`-p -1`) and reports it via `--ports_file_dir` — no code in the workspace picks a free port and releases it; the read-back helpers (`listening_on`, `gateway_ports`, `nats_port_from`, `DeadPort`, `GATEWAY_LISTENERS`) live in `beyond_ai_test_support::ports` and are shared with the gateway, verify and fleet-sim harnesses, fail at once off Linux (they need `/proc` and `127.0.0.2`), and `serve_harness_deadlines::no_test_picks_a_port_and_releases_it` enforces it workspace-wide by what happens to a port-0 socket — its port read and the socket let go, under any name — not by a helper's name), one frame deadline for every `serve` reader (`FRAME_DEADLINE`, 60s: `child_frames` for stdout — the only way to a child's stdout pipe: `ChildGuard` takes the pipe off the `Child` at spawn, so `child.stdout` is empty however it is reached (`raw_stdout` exists for tests about the pipe itself, outside the `serve_*`/`mcp_*` suites), and `serve_harness_deadlines` lints every other route (the type, `Option` methods on the field, `mem::take`/`replace`/`swap` of it, `Child { stdout, .. }` patterns, unguarded spawns) in those suites, their directory modules and `tests/common`, `ws_next_frame` for WebSockets, and `skills_env::Serve`), so a stalled run fails promptly instead of waiting for the runner's kill — `tests/serve_harness_deadlines.rs` holds every `serve_*` suite to it; `HeldPort` starts `serve` through a shell that exports its own pid as `LISTEN_PID` and `exec`s in place, as an activator does; `serve`/`run` command builders, `read_until_response`, gateway binary locator, port/connection helpers, process-lifetime JetStream for managed-gateway tests — used by every file below | +| `tests/common/mod.rs` | Shared test harness: mock Anthropic-SSE model server (an in-order server answers `serve`'s session-title request itself, off the conversation script and out of its record, counting them for `title_calls`; a test about titles matches `SESSION_TITLE_MARKER` on a routed one), ports with no window for another process to take them (`spawn_listening`: `serve --listen 127.0.0.1:0`, the bound port read from its announcement; `HeldPort`: a listener the test binds and hands over by socket activation, for a port named in `serve`'s own arguments; `DeadPort`: bound, never listening; the gateway binds `127.0.0.1:0`/`127.0.0.2:0` (`GATEWAY_LISTENERS`, distinct addresses because Pingora keys listeners by address string) and `spawn_gateway` reads both ports from its own `LISTEN` sockets in `/proc`; the shared `nats-server` picks its own port (`-p -1`) and reports it via `--ports_file_dir` — no code in the workspace picks a free port and releases it; the read-back helpers (`listening_on`, `gateway_ports`, `nats_port_from`, `DeadPort`, `GATEWAY_LISTENERS`) live in `beyond_ai_test_support::ports` and are shared with the gateway, verify and fleet-sim harnesses, fail at once off Linux (they need `/proc` and `127.0.0.2`), and `tests/ports_lint.rs` enforces it workspace-wide with no data-flow analysis: port 0 is written only in `beyond_ai_test_support::ports` (`listener`, `tokio_listener`, `bound_socket`, `bound_reuseport_socket`, `DeadPort`), apart from the few sites that bind it and keep the socket (the MCP OAuth callback, the fixture servers, child command lines), each annotated `// port-0:` in place and counted in the lint's `ALLOWED`), one frame deadline for every `serve` reader (`FRAME_DEADLINE`, 60s: `child_frames` for stdout — the only way to a child's stdout pipe: `ChildGuard` takes the pipe off the `Child` at spawn, so `child.stdout` is empty however it is reached (`raw_stdout` exists for tests about the pipe itself, outside the `serve_*`/`mcp_*` suites), and `serve_harness_deadlines` lints every other route (the type, `Option` methods on the field, `mem::take`/`replace`/`swap` of it, `Child { stdout, .. }` patterns, unguarded spawns) in those suites, their directory modules and `tests/common`, `ws_next_frame` for WebSockets, and `skills_env::Serve`), so a stalled run fails promptly instead of waiting for the runner's kill — `tests/serve_harness_deadlines.rs` holds every `serve_*` suite to it; `HeldPort` starts `serve` through a shell that exports its own pid as `LISTEN_PID` and `exec`s in place, as an activator does; `serve`/`run` command builders, `read_until_response`, gateway binary locator, port/connection helpers, process-lifetime JetStream for managed-gateway tests — used by every file below | | `tests/run_*.rs` | `run` binary against a mock model server (no gateway in the loop), split by domain: `run_core_flow` (tool round trips, refusal exit codes, text/json mode, stdin/@file input), `run_skills_prompts` (skill/prompt-template expansion, trust gating), `run_session_management` (`--session`/`--continue`/`--name`/`--fork`, same- and cross-project), `run_cli_flags` (`--version`/`--help`/`list-models`/`export`, flags reaching the wire request, including `models.json` header/api_key overrides, `--idle-timeout-ms`, `--block-images`), `run_stdout_robustness` (Task #10: a closed stdout pipe, simulated by dropping the child's stdout handle mid-stream, must exit 0 rather than panic on `EPIPE`, in both text and `--json` mode) | | `tests/serve_*.rs` | `serve` binary NDJSON protocol round-trip, split by domain: `serve_session_lifecycle` (startup/resume/crash recovery/jsonl framing/export), `serve_state_reporting` (`get_state`/`get_session_stats`/`cwd_stale`), `serve_session_tree` (branches/forks/`get_tree`), `serve_models_thinking` (model/thinking-level/`--models`), `serve_compaction_retry` (compaction + whole-run auto-retry), `serve_trust_prompts` (trust gating + system prompt), `serve_tools_bash` (tool exclusion + host `bash`), `serve_prompt_flow` (`prompt`/`steer`/`follow_up` queuing, busy semantics, refusal, abort), `serve_lifecycle` (host-emitted run lifecycle POSTs: started/progress/terminal) — each a separate Cargo test binary, ~400-1200 lines apiece rather than one ~6,200-line file | | `tests/gateway_e2e.rs` | `run` binary → real gateway binary → mock upstream (proves key-swap + the virtual key never reaches upstream). Boots a real nats-server so the fail-closed allowance watcher can seed | @@ -4296,7 +4296,7 @@ backend an operator points it at. | `BEYOND_AI_AGENT_TEST_SLOW_SESSION_OPEN_MS` | unset (no delay) | **Debug builds only** (`#[cfg(debug_assertions)]`), and only for tests: makes every `Persistence::open` sleep this many milliseconds, standing in for a slow network-backed session directory. `tests/serve_startup_stall.rs` uses it to prove session start-up is off the shared runtime thread. A release binary contains no latency-injection path at all. | | `AI_AGENT_OAUTH_CALLBACK_HOST` | unset (`127.0.0.1`) | The host the fixed-port OAuth login callback listener (Anthropic 53692, OpenAI Codex 1455) binds. **Loopback only** — an address in `127.0.0.0/8`, `::1`, or `localhost` — because the listener receives an authorization code; anything else (`0.0.0.0`, a routable address, a name) fails the login with a clear error before any bind (`oauth::callback_server::callback_host`). Tests give a login its own loopback address so concurrent runs do not collide on the fixed port. | | `BEYOND_AI_AGENT_TEST_SLOW_CHECKPOINT_MS` | unset (no delay) | **Debug builds only** (`#[cfg(debug_assertions)]`), and only for tests: makes every mid-run checkpoint write in `serve` take this many milliseconds longer, inside the write itself, standing in for a slow disk — so an acknowledgement sent before the write would be caught. `tests/serve_session_lifecycle.rs` uses it to prove a tool does not start before its `tool_use` is on disk. A release binary contains no latency-injection path at all. | -| `BEYOND_AI_AGENT_TEST_FAIL_CHECKPOINT` | unset (no delay) | **Debug builds only** (`#[cfg(debug_assertions)]`), and only for tests: fails `serve`'s checkpoint writes — `all` of them, or only the `n`th (a number). `tests/serve_session_lifecycle.rs` uses it to prove a failed write before a tool fails closed (the tool never runs) and that a best-effort one is superseded. A release binary contains no fault-injection path and none of these names (`tests/release_seams.rs`). | +| `BEYOND_AI_AGENT_TEST_FAIL_CHECKPOINT` | unset (no delay) | **Debug builds only** (`#[cfg(debug_assertions)]`), and only for tests: fails `serve`'s checkpoint writes — `all` of them, or only the `n`th (a number), counted per session so a daemon's sessions do not shift each other's count (`serve_session_isolation`: `an_injected_checkpoint_failure_counts_each_sessions_writes`). `tests/serve_session_lifecycle.rs` uses it to prove a failed write before a tool fails closed (the tool never runs) and that a best-effort one is superseded. A release binary contains no fault-injection path and none of these names (`tests/release_seams.rs`). | | `BEYOND_AI_AGENT_TEST_SLOW_SESSION_END_MS` | unset (no delay) | **Debug builds only** (`#[cfg(debug_assertions)]`), and only for tests: delays a failed daemon session's `error` broadcast after its loop has ended. `tests/serve_service_session_end.rs` uses it to prove the reason still reaches the client before the close. A release binary contains no latency-injection path at all. | | `BEYOND_AI_AGENT_TEST_SLOW_ATTACH_MS` | unset (no delay) | **Debug builds only** (`#[cfg(debug_assertions)]`), and only for tests: delays a daemon connection's registration of its output. `tests/serve_service_session_end.rs` uses it to prove a connection registering after its session ended is still told why. A release binary contains no latency-injection path at all. | | `--idle-timeout-ms` / `AI_AGENT_IDLE_TIMEOUT_MS` (`run`-only) | `agent_core::client::READ_TIMEOUT` (~600s) | Task #19 (pi-parity feature): overrides the gateway HTTP client's idle-read timeout between response chunks (`GatewayClient::with_idle_timeout`, previously unused anywhere in this codebase) — matters most for a direct-routed/custom `models.json` `base_url` override, which bypasses the gateway's own upstream-timeout assumption entirely | diff --git a/crates/agent/benches/serve_runtime.rs b/crates/agent/benches/serve_runtime.rs index 4b537584..c1613807 100644 --- a/crates/agent/benches/serve_runtime.rs +++ b/crates/agent/benches/serve_runtime.rs @@ -359,6 +359,7 @@ fn spawn_serve(bin: &str, flavor: Flavor, gateway: &str, cwd: &Path) -> ServePro .args([ "serve", "--listen", + // port-0: the child binds it, keeps it, and announces the port it got. "127.0.0.1:0", "--gateway-url", gateway, @@ -477,7 +478,7 @@ fn parse_listen_port(line: &str) -> Option { /// Unbounded Anthropic-SSE mock: every request gets the same one-turn text reply, concurrently. /// Prompt load would serialize behind a FIFO mock and measure the mock, not `serve`. fn spawn_mock_gateway() -> String { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); listener.set_nonblocking(true).unwrap(); let addr = listener.local_addr().unwrap(); let body = anthropic_text_turn("ok"); diff --git a/crates/agent/benches/web_extract.rs b/crates/agent/benches/web_extract.rs index 8f777d91..e0925c62 100644 --- a/crates/agent/benches/web_extract.rs +++ b/crates/agent/benches/web_extract.rs @@ -20,7 +20,6 @@ use std::hint::black_box; use std::io::{Read, Write}; -use std::net::TcpListener; use std::sync::OnceLock; use agent_core::Tool; @@ -74,7 +73,7 @@ fn page_html() -> &'static str { fn fixture() -> &'static str { static BASE: OnceLock = OnceLock::new(); BASE.get_or_init(|| { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let body = page_html(); let response = format!( diff --git a/crates/agent/src/bin/mcp_fixture_events_server.rs b/crates/agent/src/bin/mcp_fixture_events_server.rs index 0b095caa..65434e61 100644 --- a/crates/agent/src/bin/mcp_fixture_events_server.rs +++ b/crates/agent/src/bin/mcp_fixture_events_server.rs @@ -1735,6 +1735,7 @@ async fn main() { forbid_left: env_u64("MCP_FIXTURE_FORBID_FIRST", 0), ..State::default() })); + // port-0: held for the server's life; the address is announced through the control file. let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let control = format!("http://{addr}"); diff --git a/crates/agent/src/bin/mcp_fixture_tasks_server.rs b/crates/agent/src/bin/mcp_fixture_tasks_server.rs index 99574259..b4c50866 100644 --- a/crates/agent/src/bin/mcp_fixture_tasks_server.rs +++ b/crates/agent/src/bin/mcp_fixture_tasks_server.rs @@ -760,6 +760,7 @@ async fn serve_stdio(server: Server) { /// (or `202` for a notification), one request per connection. No SSE stream: `GET` is `405`, which /// the transport allows for a server that never pushes. async fn serve_http(server: Arc>, port_file: &str) { + // port-0: held for the server's life; the port is announced through the port file. let Ok(listener) = tokio::net::TcpListener::bind("127.0.0.1:0").await else { return; }; diff --git a/crates/agent/src/bin/mcp_skills_fixture_server.rs b/crates/agent/src/bin/mcp_skills_fixture_server.rs index b30821d4..ee6f4dc5 100644 --- a/crates/agent/src/bin/mcp_skills_fixture_server.rs +++ b/crates/agent/src/bin/mcp_skills_fixture_server.rs @@ -736,6 +736,7 @@ async fn main() { /// life the same way it does a stdio server's. async fn serve_http() { use tokio::io::AsyncReadExt; + // port-0: held for the server's life; the port it got is announced to the test. let Ok(listener) = tokio::net::TcpListener::bind("127.0.0.1:0").await else { eprintln!("mcp_skills_fixture_server: cannot bind a loopback port"); return; diff --git a/crates/agent/src/lifecycle.rs b/crates/agent/src/lifecycle.rs index 0cfff38a..9a442abf 100644 --- a/crates/agent/src/lifecycle.rs +++ b/crates/agent/src/lifecycle.rs @@ -1131,7 +1131,7 @@ mod tests { seen: Arc>>, hits: Arc, ) -> String { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().unwrap(); tokio::spawn(async move { loop { diff --git a/crates/agent/src/main.rs b/crates/agent/src/main.rs index bb48b3bb..e495140e 100644 --- a/crates/agent/src/main.rs +++ b/crates/agent/src/main.rs @@ -2816,6 +2816,7 @@ async fn run() -> Result<(), Box> { // picking a free port, releasing it and binding it later let another process take it in // between, and the login then failed before printing its URL. let listener = + // port-0: the listener is held for the whole login; the redirect URI names its port. beyond_ai_agent::oauth::callback_server::CallbackServer::bind("127.0.0.1", 0) .map_err(|e| { format!("failed to bind the local OAuth callback listener: {e}") diff --git a/crates/agent/src/metrics.rs b/crates/agent/src/metrics.rs index 3b414cdf..903d1ec3 100644 --- a/crates/agent/src/metrics.rs +++ b/crates/agent/src/metrics.rs @@ -479,7 +479,7 @@ mod tests { #[tokio::test] async fn the_listener_serves_metrics_and_refuses_every_other_path() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().unwrap(); let m = Metrics::new().unwrap(); m.sessions_live.set(7); diff --git a/crates/agent/src/oauth/callback_server.rs b/crates/agent/src/oauth/callback_server.rs index dda077c7..d06892b7 100644 --- a/crates/agent/src/oauth/callback_server.rs +++ b/crates/agent/src/oauth/callback_server.rs @@ -236,11 +236,17 @@ fn respond(request: tiny_http::Request, success: bool) { mod tests { use super::*; + /// A server on a port the kernel picks: what `agent mcp login` binds, and what is under test. + fn on_any_port() -> CallbackServer { + // port-0: the CallbackServer under test holds its listener for its whole life. + CallbackServer::bind("127.0.0.1", 0).unwrap() + } + /// Port `0` binds a port the kernel picks, and [`CallbackServer::port`] says which: the listener is /// on it from the start, and a callback sent there arrives. #[tokio::test] async fn port_zero_binds_a_real_port_and_reports_it() { - let server = CallbackServer::bind("127.0.0.1", 0).unwrap(); + let server = on_any_port(); let port = server.port(); assert_ne!(port, 0); assert!(std::net::TcpStream::connect(("127.0.0.1", port)).is_ok()); @@ -248,7 +254,7 @@ mod tests { #[tokio::test] async fn a_matching_callback_resolves_with_the_code() { - let server = CallbackServer::bind("127.0.0.1", 0).unwrap(); + let server = on_any_port(); let port = server.port(); let client = tokio::spawn(async move { // Give the server a moment to be actively waiting. @@ -272,7 +278,7 @@ mod tests { #[tokio::test] async fn a_state_mismatch_does_not_resolve_and_the_server_keeps_listening() { - let server = CallbackServer::bind("127.0.0.1", 0).unwrap(); + let server = on_any_port(); let port = server.port(); let cancel = CancellationToken::new(); let cancel_for_client = cancel.clone(); @@ -308,7 +314,7 @@ mod tests { #[tokio::test] async fn cancellation_resolves_with_none() { - let server = CallbackServer::bind("127.0.0.1", 0).unwrap(); + let server = on_any_port(); let cancel = CancellationToken::new(); let cancel_clone = cancel.clone(); tokio::spawn(async move { @@ -356,7 +362,7 @@ mod tests { #[tokio::test] async fn binding_an_already_used_port_fails() { - let first = CallbackServer::bind("127.0.0.1", 0).unwrap(); + let first = on_any_port(); let port = first.port(); let second = CallbackServer::bind("127.0.0.1", port); assert!(matches!(second, Err(OAuthError::PortBindFailed { .. }))); diff --git a/crates/agent/src/serve.rs b/crates/agent/src/serve.rs index bb00b84b..5a0bd6d2 100644 --- a/crates/agent/src/serve.rs +++ b/crates/agent/src/serve.rs @@ -1131,7 +1131,8 @@ async fn persist_messages_blocking( messages: Arc>, ) -> (Persistence, std::io::Result<()>) { match tokio::task::spawn_blocking(move || { - let r = simulated_checkpoint_write().and_then(|()| persistence.persist_messages(&messages)); + let r = simulated_checkpoint_write(&persistence.meta.id) + .and_then(|()| persistence.persist_messages(&messages)); (persistence, r) }) .await @@ -1227,19 +1228,28 @@ async fn write_then_ack( /// A slow or failing disk under a checkpoint write, simulated, **inside** the write itself: sleeps /// `BEYOND_AI_AGENT_TEST_SLOW_CHECKPOINT_MS` milliseconds, then fails the write when /// `BEYOND_AI_AGENT_TEST_FAIL_CHECKPOINT` says so — `all` fails every checkpoint write, a number `n` -/// only the `n`th (1-based, per process). Inside the write, not before it, so an +/// only the `n`th (1-based) **of each session**, counted by session id, so on a daemon one session's +/// writes never shift which write of another's fails. Inside the write, not before it, so an /// acknowledgement sent ahead of the write would precede the slow part — which is the mistake the /// slow-disk test is there to catch. `#[cfg]`-gated like [`simulated_slow_open`], names and all: a /// release binary has no fault-injection path and carries none of these variable names. #[cfg(debug_assertions)] -fn simulated_checkpoint_write() -> std::io::Result<()> { +fn simulated_checkpoint_write(session: &str) -> std::io::Result<()> { if let Ok(ms) = std::env::var("BEYOND_AI_AGENT_TEST_SLOW_CHECKPOINT_MS") && let Ok(ms) = ms.parse::() { std::thread::sleep(std::time::Duration::from_millis(ms)); } - static WRITES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); - let n = WRITES.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1; + static WRITES: std::sync::Mutex> = + std::sync::Mutex::new(std::collections::BTreeMap::new()); + let n = { + let mut writes = WRITES + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let n = writes.entry(session.to_owned()).or_default(); + *n += 1; + *n + }; let fail = match std::env::var("BEYOND_AI_AGENT_TEST_FAIL_CHECKPOINT").as_deref() { Ok("all") => true, Ok(which) => which.parse::().is_ok_and(|which| which == n), @@ -1252,7 +1262,7 @@ fn simulated_checkpoint_write() -> std::io::Result<()> { } #[cfg(not(debug_assertions))] -fn simulated_checkpoint_write() -> std::io::Result<()> { +fn simulated_checkpoint_write(_session: &str) -> std::io::Result<()> { Ok(()) } diff --git a/crates/agent/src/tools/mcp_events/wire.rs b/crates/agent/src/tools/mcp_events/wire.rs index 556e9987..b9fff546 100644 --- a/crates/agent/src/tools/mcp_events/wire.rs +++ b/crates/agent/src/tools/mcp_events/wire.rs @@ -1347,7 +1347,7 @@ mod tests { /// A loopback server answering 401 to any request without `Bearer fresh`, 200 otherwise; /// records each request's `Authorization`. async fn bearer_server() -> (String, Arc>>) { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let url = format!("http://{}/mcp", listener.local_addr().unwrap()); let seen = Arc::new(Mutex::new(Vec::new())); let record = seen.clone(); diff --git a/crates/agent/src/tools/mcp_http_exit.rs b/crates/agent/src/tools/mcp_http_exit.rs index 3303c9a3..b7f491bf 100644 --- a/crates/agent/src/tools/mcp_http_exit.rs +++ b/crates/agent/src/tools/mcp_http_exit.rs @@ -306,7 +306,7 @@ mod tests { /// A loopback server that answers every request 200 and records each request's first line and /// `Mcp-Session-Id`. async fn server() -> (String, Arc>>) { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let url = format!("http://{}/mcp", listener.local_addr().unwrap()); let seen = Arc::new(Mutex::new(Vec::new())); let record = seen.clone(); @@ -382,7 +382,7 @@ mod tests { ..Registry::default() }); // A listener that accepts and never answers. - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let url = format!("http://{}/mcp", listener.local_addr().unwrap()); let _accepting = tokio::spawn(async move { let mut held = Vec::new(); diff --git a/crates/agent/src/tools/mcp_oauth.rs b/crates/agent/src/tools/mcp_oauth.rs index 8989fd38..b6dde431 100644 --- a/crates/agent/src/tools/mcp_oauth.rs +++ b/crates/agent/src/tools/mcp_oauth.rs @@ -930,7 +930,7 @@ mod tests { status: &'static str, extra: &'static str, ) -> (String, Arc>>) { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let url = format!("http://{}/mcp", listener.local_addr().unwrap()); let seen = Arc::new(Mutex::new(Vec::new())); let record = seen.clone(); diff --git a/crates/agent/src/tools/mcp_view_http.rs b/crates/agent/src/tools/mcp_view_http.rs index 21f9ce6c..928e59cd 100644 --- a/crates/agent/src/tools/mcp_view_http.rs +++ b/crates/agent/src/tools/mcp_view_http.rs @@ -191,7 +191,7 @@ mod tests { /// [`canned`], written in `parts` with a pause between, so each reaches the client as a chunk /// of its own (an event split across reads, as a large one always is). async fn canned_in_parts(parts: &'static [&'static [u8]]) -> String { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().unwrap(); tokio::spawn(async move { while let Ok((mut stream, _)) = listener.accept().await { diff --git a/crates/agent/src/tools/mcp_wire.rs b/crates/agent/src/tools/mcp_wire.rs index dbbe322a..c37b4f85 100644 --- a/crates/agent/src/tools/mcp_wire.rs +++ b/crates/agent/src/tools/mcp_wire.rs @@ -661,7 +661,7 @@ mod tests { /// One canned HTTP response per connection, from a loopback listener; `hold` keeps the /// connection open afterwards (an SSE stream the server never ends). async fn canned(response: &'static [u8], hold: bool) -> String { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().unwrap(); tokio::spawn(async move { while let Ok((mut stream, _)) = listener.accept().await { @@ -891,7 +891,7 @@ mod tests { async fn an_ordinary_request_is_sent_as_rmcps_own_client_would() { agent_core::ensure_provider(); async fn capture() -> (String, tokio::sync::oneshot::Receiver) { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().unwrap(); let (tx, rx) = tokio::sync::oneshot::channel(); tokio::spawn(async move { diff --git a/crates/agent/src/tools/web/mod.rs b/crates/agent/src/tools/web/mod.rs index 2eb4b847..e1714ae8 100644 --- a/crates/agent/src/tools/web/mod.rs +++ b/crates/agent/src/tools/web/mod.rs @@ -442,12 +442,11 @@ impl Tool for Web { mod tests { use super::*; use std::io::{Read, Write}; - use std::net::TcpListener; /// A one-shot loopback HTTP/1.1 server that replies with `response` to the first connection, and /// records the raw request line + headers it received. Returns `(port, join_handle_for_request)`. fn one_shot(response: &'static str) -> (u16, std::thread::JoinHandle) { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let port = listener.local_addr().unwrap().port(); let handle = std::thread::spawn(move || { let (mut stream, _) = listener.accept().unwrap(); diff --git a/crates/agent/src/tools/web/ssrf.rs b/crates/agent/src/tools/web/ssrf.rs index 8fc5324c..f83c6cfb 100644 --- a/crates/agent/src/tools/web/ssrf.rs +++ b/crates/agent/src/tools/web/ssrf.rs @@ -221,6 +221,7 @@ impl Resolve for SsrfResolver { let policy = Arc::clone(&self.policy); let host = name.as_str().to_string(); Box::pin(async move { + // port-0: a DNS lookup, not a bind — only the addresses it returns are used. let addrs: Vec = tokio::net::lookup_host((host.as_str(), 0)) .await .map_err(|e| Box::new(e) as Box)? diff --git a/crates/agent/tests/common/mcp_apps_http.rs b/crates/agent/tests/common/mcp_apps_http.rs index 18053b76..a6e89daf 100644 --- a/crates/agent/tests/common/mcp_apps_http.rs +++ b/crates/agent/tests/common/mcp_apps_http.rs @@ -57,7 +57,7 @@ fn advertises_ui(request: &Value) -> bool { } pub fn spawn_http_apps_fixture() -> HttpAppsFixture { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let ui_requests = Arc::new(AtomicUsize::new(0)); let plain_requests = Arc::new(AtomicUsize::new(0)); diff --git a/crates/agent/tests/common/mcp_fixture.rs b/crates/agent/tests/common/mcp_fixture.rs index e5b0819b..dab0a96b 100644 --- a/crates/agent/tests/common/mcp_fixture.rs +++ b/crates/agent/tests/common/mcp_fixture.rs @@ -59,7 +59,7 @@ impl HttpMcpFixture { /// Start one fixture on a loopback port. The thread runs for the test process's life. pub fn spawn_http_mcp_fixture() -> HttpMcpFixture { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let requests = Arc::new(Mutex::new(Vec::new())); let recorder = requests.clone(); diff --git a/crates/agent/tests/common/mcp_oauth_fixture.rs b/crates/agent/tests/common/mcp_oauth_fixture.rs index 829dd43d..5a8aff04 100644 --- a/crates/agent/tests/common/mcp_oauth_fixture.rs +++ b/crates/agent/tests/common/mcp_oauth_fixture.rs @@ -227,7 +227,7 @@ impl OAuthFixture { /// seconds remain, so anything below 30 is refreshed on the very next read without any waiting; /// a test that wants a token to stay put must pass something comfortably above it. pub fn spawn(expires_in_secs: u64) -> Self { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let base = format!("http://{}", listener.local_addr().unwrap()); let fixture = Self { url: format!("{base}/mcp"), diff --git a/crates/agent/tests/common/mod.rs b/crates/agent/tests/common/mod.rs index 95cfe48b..05914fc5 100644 --- a/crates/agent/tests/common/mod.rs +++ b/crates/agent/tests/common/mod.rs @@ -218,6 +218,7 @@ pub fn spawn_listening(cmd: &mut Command) -> (ChildGuard, u16) { /// [`spawn_listening`], also copying the child's stderr to `log` — for a test that reads it back. pub fn spawn_listening_logged(cmd: &mut Command, log: Option) -> (ChildGuard, u16) { + // port-0: the child binds it, keeps it, and announces the port it got (read below). cmd.args(["--listen", "127.0.0.1:0"]).stderr(Stdio::piped()); let mut child = ChildGuard::spawn(cmd); let stderr = child.stderr.take().expect("stderr is piped"); @@ -279,7 +280,7 @@ pub struct HeldPort { impl HeldPort { pub fn bind() -> Self { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let port = listener.local_addr().unwrap().port(); Self { listener, port } } diff --git a/crates/agent/tests/direct_routing_wire.rs b/crates/agent/tests/direct_routing_wire.rs index d02c0403..8adec54e 100644 --- a/crates/agent/tests/direct_routing_wire.rs +++ b/crates/agent/tests/direct_routing_wire.rs @@ -21,7 +21,7 @@ use futures::StreamExt; /// A one-shot listener that records the raw request text and answers with an empty SSE 200. fn capture_request() -> (String, Arc>) { - let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().expect("addr"); let captured = Arc::new(Mutex::new(String::new())); let sink = Arc::clone(&captured); diff --git a/crates/agent/tests/exec_endpoint.rs b/crates/agent/tests/exec_endpoint.rs index e58c9bfc..ed965ae7 100644 --- a/crates/agent/tests/exec_endpoint.rs +++ b/crates/agent/tests/exec_endpoint.rs @@ -26,7 +26,7 @@ async fn mock_provider( root: std::path::PathBuf, seen: Arc, ) -> (String, Arc) { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().unwrap(); let hits = seen.clone(); tokio::spawn(async move { @@ -235,7 +235,7 @@ async fn an_unreachable_endpoint_fails_loudly_rather_than_looking_empty() { #[tokio::test] async fn auth_headers_reach_the_endpoint() { // Every real provider needs some form of auth; which form is theirs to decide. - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().unwrap(); let (tx, rx) = std::sync::mpsc::channel::(); tokio::spawn(async move { diff --git a/crates/agent/tests/exec_endpoint_response_cap.rs b/crates/agent/tests/exec_endpoint_response_cap.rs index a7ee0857..183bbb73 100644 --- a/crates/agent/tests/exec_endpoint_response_cap.rs +++ b/crates/agent/tests/exec_endpoint_response_cap.rs @@ -28,7 +28,7 @@ where F: Fn(tokio::net::TcpStream) -> Fut + Send + Sync + 'static, Fut: std::future::Future + Send + 'static, { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let url = format!("http://{}/exec", listener.local_addr().unwrap()); let respond = Arc::new(respond); tokio::spawn(async move { diff --git a/crates/agent/tests/mcp_tasks_env/mod.rs b/crates/agent/tests/mcp_tasks_env/mod.rs index b6f74886..54bedeab 100644 --- a/crates/agent/tests/mcp_tasks_env/mod.rs +++ b/crates/agent/tests/mcp_tasks_env/mod.rs @@ -261,7 +261,7 @@ pub struct Relay { impl Relay { pub fn start() -> Self { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let port = listener.local_addr().unwrap().port(); let backend = std::sync::Arc::new(std::sync::Mutex::new(None::)); let routes = backend.clone(); diff --git a/crates/agent/tests/ports_lint.rs b/crates/agent/tests/ports_lint.rs new file mode 100644 index 00000000..4527d097 --- /dev/null +++ b/crates/agent/tests/ports_lint.rs @@ -0,0 +1,738 @@ +//! Port 0 is spelled in one module: `beyond_ai_test_support::ports`. +//! +//! Binding port 0 and then reading the port the kernel picked is how code gets a port no other +//! process holds. Releasing that socket and passing the number on for something else to bind +//! reopens the race: any process can take the port in between. A lint that tried to tell a held +//! socket from a released one had to follow the socket through bindings, wrappers, closures and +//! helpers in other files, and it kept missing new shapes. This rule needs none of that, so a new +//! shape cannot slip past it. **A zero port is not written anywhere in the workspace except +//! `crates/test-support/src/ports.rs`**, the one module allowed to bind and read a port. Every test +//! that needs a port uses its helpers: +//! +//! - a listener held for the test's life: `ports::listener()`, `ports::tokio_listener()`; +//! - a bound socket that never listens: `ports::bound_socket()`, `DeadPort`; +//! - a child that binds port 0 itself and reports what it got (`common::spawn_listening`, the +//! gateway's `GATEWAY_LISTENERS`, `nats-server --ports_file_dir`). +//! +//! Some code outside `ports.rs` binds port 0 and keeps the listener for real: the MCP OAuth +//! callback, the fixture servers that announce their port, and the command lines that ask a child +//! to bind port 0 and report it. Each such site has a `// port-0: ` comment in place +//! and is counted in [`ALLOWED`], so a new one shows up as a change to this file. +//! +//! What counts as writing a zero port: +//! +//! - an address literal with port 0: `"127.0.0.1:0"`, `"[::1]:0"`, `"localhost:0"`, +//! `"{host}:0"`, `":0"`; +//! - a zero (`0`, `0u16`, `0_u16`, `u16::MIN`, `0 as u16`, `Default::default()`, or a `const`, +//! `static` or `let` bound to one) used as the port of: +//! - a `bind` call; +//! - `SocketAddr::new`, `SocketAddrV4::new`, or `SocketAddrV6::new` (the second argument); +//! - `set_port`; +//! - an address tuple whose first element is an IP (`("127.0.0.1", 0)`, `([127, 0, 0, 1], 0)`, +//! `(Ipv4Addr::LOCALHOST, 0)`, `(ip, 0)`); +//! - a zero formatted in as an address's port: `format!("127.0.0.1:{}", 0)`, +//! `format!("{}:{}", host, 0)`, `format!("{host}:{port}", port = 0)`, or `{port}` capturing a +//! `let port = 0`; +//! - a zero in a port field: `Config { port: 0, .. }`, `listen_port: 0`. +//! +//! Comments are skipped, and string literals are only checked for addresses. Whitespace is +//! ignored, so layout cannot hide a case. +//! +//! **What a text rule cannot see.** A zero that only becomes a port later, through a value the rule +//! cannot follow, passes: +//! +//! - an environment default: `env::var("PORT").ok().and_then(|v| v.parse().ok()).unwrap_or(0)`, +//! later passed to a bind; +//! - a string default: `unwrap_or_else(|| "0".into())`, later formatted into an address; +//! - a port computed at run time: `n - n`. +//! +//! Catching these needs the value's flow, which is exactly the analysis this rule gave up. The first +//! two are plausible by accident, so review still matters where a port comes from configuration; the +//! last would be deliberate evasion. +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use std::collections::HashSet; +use std::path::{Path, PathBuf}; + +/// Every file outside `ports.rs` that binds port 0 and keeps the socket, with how many annotated +/// (`// port-0:`) sites it has. Adding one is a reviewed change here, not just a comment there. +const ALLOWED: &[(&str, usize)] = &[ + // `agent mcp login`: the callback listener is held and the redirect URI names its port. + ("agent/src/main.rs", 1), + // The `CallbackServer` under test holds its listener. + ("agent/src/oauth/callback_server.rs", 1), + // Fixture servers, run as children: each binds port 0, keeps it, and announces it. + ("agent/src/bin/mcp_fixture_events_server.rs", 1), + ("agent/src/bin/mcp_fixture_tasks_server.rs", 1), + ("agent/src/bin/mcp_skills_fixture_server.rs", 1), + // Not a bind: the resolver's lookup takes a port, and only the addresses it returns are used. + ("agent/src/tools/web/ssrf.rs", 1), + // Command lines that ask a child `serve` to bind port 0 and report the port it got. + ("agent/tests/common/mod.rs", 1), + ("agent/tests/serve_service_mode.rs", 2), + ("agent/benches/serve_runtime.rs", 1), + ("fleet-sim/src/replica.rs", 2), +]; + +/// A source file with comments blanked out (newlines kept) and its string literals located. +struct Lexed { + /// The source's characters, with every comment character replaced by a space. + code: Vec, + /// Each string literal as (start, end) indices of its contents in `code`. + literals: Vec<(usize, usize)>, +} + +fn lex(source: &str) -> Lexed { + let s: Vec = source.chars().collect(); + let mut code = s.clone(); + let mut literals = Vec::new(); + let mut i = 0; + let ident = |c: char| c.is_alphanumeric() || c == '_'; + while i < s.len() { + let c = s[i]; + let next = s.get(i + 1).copied(); + if c == '/' && next == Some('/') { + while i < s.len() && s[i] != '\n' { + code[i] = ' '; + i += 1; + } + } else if c == '/' && next == Some('*') { + let mut depth = 0; + while i < s.len() { + if s[i] == '/' && s.get(i + 1) == Some(&'*') { + depth += 1; + code[i] = ' '; + code[i + 1] = ' '; + i += 2; + } else if s[i] == '*' && s.get(i + 1) == Some(&'/') { + depth -= 1; + code[i] = ' '; + code[i + 1] = ' '; + i += 2; + if depth == 0 { + break; + } + } else { + if s[i] != '\n' { + code[i] = ' '; + } + i += 1; + } + } + } else if (c == 'r' || (c == 'b' && next == Some('r'))) && !(i > 0 && ident(s[i - 1])) && { + let mut j = i + if c == 'b' { 2 } else { 1 }; + while s.get(j) == Some(&'#') { + j += 1; + } + s.get(j) == Some(&'"') + } { + // A raw string: `r#"…"#`, closed by a quote and the same number of hashes. + let mut j = i + if c == 'b' { 2 } else { 1 }; + let mut hashes = 0; + while s[j] == '#' { + hashes += 1; + j += 1; + } + let start = j + 1; + let mut k = start; + while k < s.len() { + if s[k] == '"' && (0..hashes).all(|h| s.get(k + 1 + h) == Some(&'#')) { + break; + } + k += 1; + } + literals.push((start, k.min(s.len()))); + i = k + 1 + hashes; + } else if c == '"' { + let start = i + 1; + let mut k = start; + while k < s.len() && s[k] != '"' { + if s[k] == '\\' { + k += 1; + } + k += 1; + } + literals.push((start, k.min(s.len()))); + i = k + 1; + } else if c == '\'' { + // A char literal (`'"'`, `'\''`), or a lifetime (`'a`), which has no closing quote. + if next == Some('\\') { + // Past the escaped character, which may itself be a quote (`'\''`). + let mut k = i + 3; + while k < s.len() && s[k] != '\'' { + k += 1; + } + i = k + 1; + } else if s.get(i + 2) == Some(&'\'') { + i += 3; + } else { + i += 1; + } + } else { + i += 1; + } + } + Lexed { code, literals } +} + +/// Whether `text` (a literal's contents) holds an address with port 0. +fn literal_has_port_zero(text: &[char]) -> bool { + let t: String = text.iter().collect(); + t.match_indices(":0").any(|(at, _)| { + let after = t[at + 2..].chars().next(); + if after.is_some_and(|c| c.is_ascii_alphanumeric() || c == '.' || c == '/' || c == ':') { + return false; + } + names_a_host(&t[..at]) + }) +} + +/// The placeholders of a format string that stand for an address's port (`"127.0.0.1:{}"`, +/// `"{host}:{port}"`), each as the argument it takes: `Ok(i)` the `i`th positional argument, or +/// `Err(name)` a named one (or a captured variable). +fn port_placeholders(t: &str) -> Vec> { + let mut out = Vec::new(); + let mut positional = 0; + let mut i = 0; + while let Some(open) = t[i..].find('{').map(|o| o + i) { + if t[open + 1..].starts_with('{') { + i = open + 2; + continue; + } + let Some(close) = t[open..].find('}').map(|c| c + open) else { + break; + }; + let inner = &t[open + 1..close]; + let arg = inner.split(':').next().unwrap_or(""); + let which = if arg.is_empty() { + positional += 1; + Ok(positional - 1) + } else if let Ok(n) = arg.parse::() { + Ok(n) + } else { + Err(arg.to_string()) + }; + if open > 0 && t[..open].ends_with(':') && names_a_host(&t[..open - 1]) { + out.push(which); + } + i = close + 1; + } + out +} + +/// Whether `before` (the text up to an address's `:`) ends with a host: a dotted quad, `]`, +/// `localhost`, a `{placeholder}`, or nothing at all. +fn names_a_host(before: &str) -> bool { + { + if before.is_empty() || before.ends_with(']') || before.ends_with('}') { + return true; + } + if before.ends_with("localhost") { + return true; + } + // A dotted quad right before the colon. + let quad: String = before + .chars() + .rev() + .take_while(|c| c.is_ascii_digit() || *c == '.') + .collect::>() + .into_iter() + .rev() + .collect(); + let parts: Vec<&str> = quad.split('.').collect(); + parts.len() == 4 + && parts + .iter() + .all(|p| !p.is_empty() && p.parse::().is_ok()) + } +} + +/// The code with whitespace removed (a single space kept between two identifier characters), each +/// character with its index in the lexed source and whether it lies inside a string literal. +fn normalised(lexed: &Lexed) -> Vec<(char, usize, bool)> { + let mut in_literal = vec![false; lexed.code.len()]; + for &(a, b) in &lexed.literals { + // The quotes too, so `"…"` is one unit. + for flag in &mut in_literal[a.saturating_sub(1)..(b + 1).min(lexed.code.len())] { + *flag = true; + } + } + let ident = |c: char| c.is_alphanumeric() || c == '_'; + let mut out: Vec<(char, usize, bool)> = Vec::new(); + let mut pending_space = false; + for (i, &c) in lexed.code.iter().enumerate() { + if c.is_whitespace() && !in_literal[i] { + pending_space = true; + continue; + } + if pending_space + && ident(c) + && out.last().is_some_and(|&(p, _, lit)| ident(p) && !lit) + && !in_literal[i] + { + out.push((' ', i, false)); + } + pending_space = false; + out.push((c, i, in_literal[i])); + } + out +} + +const ZERO_SPELLINGS: &[&str] = &[ + "0u16", + "0_u16", + "u16::MIN", + "0 as u16", + "Default::default()", + "u16::default()", + "0", +]; + +/// Names bound to a zero: `const`/`static` ones (`const ANY: u16 = 0;`) are returned in `global`, +/// `let` ones in `local`. +fn zero_names(text: &str, global: &mut HashSet, local: &mut HashSet) { + for kw in ["const ", "static ", "let mut ", "let "] { + for (at, _) in text.match_indices(kw) { + if text[..at] + .chars() + .last() + .is_some_and(|c| c.is_alphanumeric() || c == '_') + { + continue; + } + let rest = &text[at + kw.len()..]; + let name: String = rest + .chars() + .take_while(|c| c.is_alphanumeric() || *c == '_') + .collect(); + if name.is_empty() { + continue; + } + let mut after = &rest[name.len()..]; + if let Some(typed) = after.strip_prefix(':') { + let ty: String = typed + .chars() + .take_while(|c| c.is_alphanumeric() || *c == '_') + .collect(); + after = &typed[ty.len()..]; + } + let Some(value) = after.strip_prefix('=') else { + continue; + }; + if ZERO_SPELLINGS + .iter() + .any(|z| value.strip_prefix(z).is_some_and(|r| r.starts_with(';'))) + { + if kw.starts_with("let") { + local.insert(name); + } else { + global.insert(name); + } + } + } + } +} + +/// Whether an address tuple's first element names an IP. +fn ip_like(first: &str) -> bool { + let first = first.trim_start_matches('&'); + if let Some(lit) = first.strip_prefix('"') { + let lit = lit.trim_end_matches('"'); + let parts: Vec<&str> = lit.split('.').collect(); + return lit == "localhost" + || lit.contains("::") + || (parts.len() == 4 && parts.iter().all(|p| p.parse::().is_ok())); + } + if first.starts_with('[') && first[1..].starts_with(|c: char| c.is_ascii_digit()) { + return true; + } + if ["Ipv4Addr", "Ipv6Addr", "IpAddr", "LOCALHOST", "UNSPECIFIED"] + .iter() + .any(|k| first.contains(k)) + { + return true; + } + first + .split(['.', ':']) + .map(|s| s.trim_end_matches("()").to_ascii_lowercase()) + .any(|s| { + ["ip", "host", "addr", "loopback", "localhost"].contains(&s.as_str()) + || s.ends_with("_ip") + || s.ends_with("_host") + }) +} + +/// The top-level, comma-separated elements of the group opened at `open`, and where it closes. +fn elements(n: &[(char, usize, bool)], open: usize) -> (Vec, usize) { + let mut depth = 0i32; + let mut out = vec![String::new()]; + for (i, &(c, _, lit)) in n.iter().enumerate().skip(open + 1) { + if !lit { + match c { + '(' | '[' | '{' => depth += 1, + ')' | ']' | '}' => { + if depth == 0 { + // A trailing comma (`(ip, 0,)`, as rustfmt lays out a long tuple). + if out.len() > 1 && out.last().is_some_and(String::is_empty) { + out.pop(); + } + return (out, i); + } + depth -= 1; + } + ',' if depth == 0 => { + out.push(String::new()); + continue; + } + _ => {} + } + } + out.last_mut().unwrap().push(c); + } + (out, n.len()) +} + +/// The path called right before `open` (`TcpListener::bind`, `SocketAddr::new`), or empty for a +/// bare tuple. +fn callee(n: &[(char, usize, bool)], open: usize) -> String { + let mut s: Vec = n[..open] + .iter() + .rev() + .take_while(|&&(c, _, lit)| { + !lit && (c.is_alphanumeric() || c == '_' || c == ':' || c == '.') + }) + .map(|&(c, ..)| c) + .collect(); + s.reverse(); + s.into_iter().collect() +} + +/// Each place in `source` that writes a zero port, as the index in `source`'s characters. +fn port_zero_sites(source: &str, global_zero: &HashSet) -> Vec { + let lexed = lex(source); + let mut sites = Vec::new(); + for &(a, b) in &lexed.literals { + if literal_has_port_zero(&lexed.code[a..b]) { + sites.push(a); + } + } + let n = normalised(&lexed); + let text: String = n.iter().map(|&(c, ..)| c).collect(); + let mut local_zero = HashSet::new(); + let mut ignored = HashSet::new(); + zero_names(&text, &mut ignored, &mut local_zero); + let is_zero = + |e: &str| ZERO_SPELLINGS.contains(&e) || global_zero.contains(e) || local_zero.contains(e); + for open in (0..n.len()).filter(|&i| n[i].0 == '(' && !n[i].2) { + let (els, close) = elements(&n, open); + let called = callee(&n, open); + let zero_at = |k: usize| els.get(k).is_some_and(|e| is_zero(e)); + let site = n.get(close.saturating_sub(1)).map_or(0, |&(_, at, _)| at); + let macro_call = open > 0 && n[open - 1].0 == '!' && !n[open - 1].2; + let hit = if macro_call { + // `format!("127.0.0.1:{}", 0)`: an address format string whose port placeholder takes + // a zero, positional, named (`port = 0`) or captured (`{port}` with `let port = 0`). + let lit = els + .first() + .map_or("", |f| f.trim_start_matches('r').trim_matches('#')); + let args = els.get(1..).unwrap_or_default(); + let positional: Vec<&String> = args.iter().filter(|e| !e.contains('=')).collect(); + lit.starts_with('"') + && port_placeholders(lit.trim_matches('"')) + .iter() + .any(|which| match which { + Ok(i) => positional.get(*i).is_some_and(|e| is_zero(e)), + // A captured variable counts only when it is named as a port: a + // `{seed}:{attempt}` key whose counter starts at 0 is not an address. + Err(name) => { + ((name == "port" || name.ends_with("_port")) && is_zero(name)) + || args.iter().any(|e| { + e.split_once('=') + .is_some_and(|(k, v)| k == name && is_zero(v)) + }) + } + }) + } else if called.ends_with("SocketAddrV6::new") { + zero_at(1) + } else if called.ends_with("set_port") { + els.len() == 1 && zero_at(0) + } else if els.len() == 2 && zero_at(1) { + called.ends_with("bind") + || called.ends_with("SocketAddr::new") + || called.ends_with("SocketAddrV4::new") + || ip_like(&els[0]) + } else { + false + }; + if hit { + sites.push(site); + } + } + // A port field set to zero: `Config { port: 0, .. }`, `listen_port: 0`. + for (at, _) in text.match_indices(':') { + let name: String = text[..at] + .chars() + .rev() + .take_while(|c| c.is_alphanumeric() || *c == '_') + .collect::>() + .into_iter() + .rev() + .collect(); + let opens_field = text[..at - name.len()].ends_with(['{', ',']); + let value = &text[at + 1..]; + let zero = ZERO_SPELLINGS.iter().any(|z| { + value + .strip_prefix(z) + .is_some_and(|r| r.starts_with([',', '}'])) + }); + if (name == "port" || name.ends_with("_port")) + && opens_field + && zero + && !n[at].2 + && !text[at + 1..].starts_with(':') + && !text[..at].ends_with(':') + { + sites.push(n[at].1); + } + } + sites +} + +/// Whether the line holding character `at`, or the comment lines right above it, carry a +/// `port-0:` annotation. +fn annotated(source: &str, at: usize) -> bool { + let lines: Vec<&str> = source.lines().collect(); + let line = source.chars().take(at).filter(|&c| c == '\n').count(); + if lines.get(line).is_some_and(|l| l.contains("port-0:")) { + return true; + } + lines[..line] + .iter() + .rev() + .take_while(|l| l.trim_start().starts_with("//")) + .any(|l| l.contains("port-0:")) +} + +fn line_of(source: &str, at: usize) -> (usize, String) { + let line = source.chars().take(at).filter(|&c| c == '\n').count(); + ( + line + 1, + source.lines().nth(line).unwrap_or("").trim().to_string(), + ) +} + +/// The unannotated zero-port sites in one source, for the pinned cases. +fn findings(source: &str) -> Vec { + let lexed = lex(source); + let n = normalised(&lexed); + let text: String = n.iter().map(|&(c, ..)| c).collect(); + let mut global = HashSet::new(); + zero_names(&text, &mut global, &mut HashSet::new()); + port_zero_sites(source, &global) + .into_iter() + .filter(|&at| !annotated(source, at)) + .map(|at| line_of(source, at).1) + .collect() +} + +fn rust_files(dir: &Path, out: &mut Vec) { + for entry in std::fs::read_dir(dir).unwrap() { + let path = entry.unwrap().path(); + let name = path.file_name().unwrap().to_string_lossy().into_owned(); + if path.is_dir() { + if name != "target" && !name.starts_with('.') { + rust_files(&path, out); + } + } else if name.ends_with(".rs") { + out.push(path); + } + } +} + +#[test] +fn port_zero_is_written_only_in_the_ports_module() { + let crates = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap(); + let mut files = Vec::new(); + rust_files(crates, &mut files); + for krate in [ + "agent", + "agent-core", + "gateway", + "verify", + "fleet-sim", + "test-support", + ] { + assert!( + files.iter().any(|f| f.starts_with(crates.join(krate))), + "the scan covers the {krate} crate" + ); + } + let this = Path::new(file!()).file_name().unwrap(); + let ports = crates.join("test-support/src/ports.rs"); + assert!(files.contains(&ports), "the ports module is where it was"); + let files: Vec = files + .into_iter() + .filter(|f| *f != ports && f.file_name() != Some(this)) + .collect(); + let sources: Vec = files + .iter() + .map(|f| std::fs::read_to_string(f).unwrap()) + .collect(); + // `const ANY: u16 = 0` in one file is a zero port when another file uses it. + let mut global = HashSet::new(); + for s in &sources { + let n = normalised(&lex(s)); + let text: String = n.iter().map(|&(c, ..)| c).collect(); + zero_names(&text, &mut global, &mut HashSet::new()); + } + let mut unannotated = Vec::new(); + let mut held: Vec<(String, usize)> = Vec::new(); + for (file, source) in files.iter().zip(&sources) { + let rel = file + .strip_prefix(crates) + .unwrap() + .to_string_lossy() + .into_owned(); + let mut count = 0; + for at in port_zero_sites(source, &global) { + if annotated(source, at) { + count += 1; + } else { + let (line, text) = line_of(source, at); + unannotated.push(format!("{rel}:{line}: {text}")); + } + } + if count > 0 { + held.push((rel, count)); + } + } + unannotated.sort(); + unannotated.dedup(); + assert!( + unannotated.is_empty(), + "port 0 written outside beyond_ai_test_support::ports — use its helpers (`listener()`, \ + `tokio_listener()`, `bound_socket()`, `DeadPort`, or a child that reports its port):\n{}", + unannotated.join("\n") + ); + held.sort(); + let mut allowed: Vec<(String, usize)> = + ALLOWED.iter().map(|&(f, n)| (f.to_string(), n)).collect(); + allowed.sort(); + assert_eq!( + held, allowed, + "the annotated (`// port-0:`) sites differ from ALLOWED: update it in the same change" + ); +} + +/// Every way of releasing a picked port that the old data-flow lint had to chase is now caught, +/// because each one writes a zero port. So are listeners that are held, if they sit outside +/// `ports.rs` without an annotation. +#[test] +fn the_rule_catches_every_spelling_of_port_zero() { + for bad in [ + // A temporary. + "let p = TcpListener::bind(\"127.0.0.1:0\").unwrap().local_addr().unwrap().port();", + "let p = std::net::TcpListener::bind(\"127.0.0.1:0\")\n .and_then(|l| l.local_addr())\n .unwrap()\n .port();", + "let p = TcpListener::bind((\"127.0.0.1\", 0)).await?.local_addr()?.port();", + // Named, in a helper or inline. + "fn pick() -> u16 { let l = TcpListener::bind(\"127.0.0.1:0\").unwrap(); l.local_addr().unwrap().port() }", + "let sock = std::net::TcpListener::bind(\"127.0.0.1:0\")?;\nlet port = sock.local_addr()?.port();\ndrop(sock);\nspawn_child(port);", + "let reserve = TcpListener::bind(\"0.0.0.0:0\").unwrap();\nlet n = reserve.local_addr().unwrap().port();\n}\nchild(n);", + // A socket bound after creation. + "let s = TcpSocket::new_v4()?;\ns.set_reuseaddr(true)?;\ns.bind(\"127.0.0.1:0\".parse().unwrap())?;\nlet p = s.local_addr()?.port();\ndrop(s);", + // A port-0 address held in a variable or const. + "let a = \"127.0.0.1:0\";\nlet l = TcpListener::bind(a).unwrap();\nlet p = l.local_addr().unwrap().port();\ndrop(l);", + "const ANY: &str = \"127.0.0.1:0\";\nfn pick() -> u16 { TcpListener::bind(ANY).unwrap().local_addr().unwrap().port() }", + // A typed zero. + "let p = TcpListener::bind((\"127.0.0.1\", 0u16)).unwrap().local_addr().unwrap().port();", + "let p = TcpListener::bind((Ipv4Addr::LOCALHOST, 0_u16)).unwrap().local_addr().unwrap().port();", + "let p = TcpListener::bind((Ipv4Addr::LOCALHOST, u16::MIN)).unwrap();", + "let p = TcpListener::bind((ip, 0 as u16)).unwrap();", + // The socket handed to a helper that only reads its address. + "fn port_of(l: &TcpListener) -> u16 { l.local_addr().unwrap().port() }\nfn pick() -> u16 { let l = TcpListener::bind(\"127.0.0.1:0\").unwrap(); port_of(&l) }", + // A tuple pattern, `let _ =`, a self-rebind. + "let (l, _) = (TcpListener::bind(\"127.0.0.1:0\").unwrap(), 1);\nlet p = l.local_addr().unwrap().port();\ndrop(l);", + "let l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet p = l.local_addr().unwrap().port();\nlet _ = l;", + "let l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet l = l;\nlet p = l.local_addr().unwrap().port();\ndrop(l);", + // A closure that only reads the address. + "let port_of = |l: &TcpListener| l.local_addr().unwrap().port();\nlet l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet p = port_of(&l);\ndrop(l);", + // A listener held in an `Option`. + "let l = Some(TcpListener::bind(\"127.0.0.1:0\").unwrap());\nlet p = l.as_ref().unwrap().local_addr().unwrap().port();\ndrop(l);", + "let l = Some(TcpListener::bind(\"127.0.0.1:0\").unwrap());\nlet p = l.as_ref().expect(\"bound\").local_addr().unwrap().port();", + // A zero port in a const or a let. + "const ANY: u16 = 0;\nlet p = TcpListener::bind((\"127.0.0.1\", ANY)).unwrap().local_addr().unwrap().port();", + "let mut port = 0;\nlet l = TcpListener::bind((host, port)).unwrap();", + // A helper in another file: the bind is still written here. + "let l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet p = common::port_of(&l);\ndrop(l);", + // socket2. + "let s = Socket::new(Domain::IPV4, Type::STREAM, None)?;\ns.bind(&\"127.0.0.1:0\".parse::()?.into())?;\nlet p = s.local_addr()?.as_socket().unwrap().port();\ndrop(s);", + "s.bind(&SockAddr::from(SocketAddr::from(([127, 0, 0, 1], 0))))?;", + // SocketAddr constructors and setters. + "let a = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0);", + "let a = SocketAddrV4::new(loopback, 0);", + "let a = SocketAddrV6::new(Ipv6Addr::LOCALHOST, 0, 0, 0);", + "addr.set_port(0);", + "let a: SocketAddr = ([127, 0, 0, 1], 0).into();", + // IPv6, names and formatted addresses. + "TcpListener::bind(\"[::1]:0\").unwrap();", + "TcpListener::bind(\"localhost:0\").unwrap();", + "TcpListener::bind(format!(\"{host}:0\")).unwrap();", + "TcpListener::bind(host.to_string() + \":0\").unwrap();", + "TcpListener::bind(r#\"127.0.0.1:0\"#).unwrap();", + // A held listener is still a zero port written outside `ports.rs`. + "let listener = TcpListener::bind(\"127.0.0.1:0\").await.unwrap();\ntokio::spawn(serve(listener));", + "CallbackServer::bind(\"127.0.0.1\", 0).unwrap();", + // A zero formatted in as the port. + "let a = format!(\"127.0.0.1:{}\", 0);", + "let a = format!(\"{host}:{}\", 0u16);", + "let a = format!(\"{host}:{port}\", port = 0);", + "let a = format!(\"{}:{}\", host, 0);", + "let port = 0;\nlet a = format!(\"127.0.0.1:{port}\");", + "TcpListener::bind(format!(\"localhost:{}\", ANY)).unwrap();\nconst ANY: u16 = 0;", + // A zero in a port field. + "let cfg = Config { host: \"127.0.0.1\".into(), port: 0 };", + "let cfg = ListenConfig {\n listen_port: 0,\n ..Default::default()\n};", + // Layout cannot hide it. + "TcpListener::bind((\n \"127.0.0.1\",\n 0,\n))", + ] { + assert!(!findings(bad).is_empty(), "missed: {bad:?}"); + } + for fine in [ + "let l = TcpListener::bind(\"127.0.0.1:8080\").unwrap();", + "let listener = beyond_ai_test_support::ports::listener();", + "let dead = DeadPort::bind();", + "// let p = TcpListener::bind(\"127.0.0.1:0\").unwrap();", + "/* TcpListener::bind(\"127.0.0.1:0\") */", + // A port-0 address that is not one. + "let model = \"us.anthropic.claude-haiku-4-5-20251001-v1:0\";", + "let id = \"functions.read:0\";", + "assert!(line.contains(r#\"\"input_tokens\":0\"#));", + "let t = \"12:00:01\";", + "let ratio = \"1:0\";", + // A URL naming port 0 is a connect target, not a bind. + "let url = \"http://127.0.0.1:0/x\";", + // Zeros that are not ports. + "assert_eq!(n, 0);", + "let pair = (name, 0);", + "call.bind(handle.id.clone());", + "TcpListener::bind((\"127.0.0.1\", held.port())).unwrap();", + "let x = vec![(1, 0)];", + "let c = '\"'; let l = TcpListener::bind(addr).unwrap();", + // A format string that is not an address, or a port that is not zero. + "let t = format!(\"{} of {}\", 12, 0);", + "let mut attempt = 0;\nlet key = format!(\"{seed_base}:{attempt}\");", + "let a = format!(\"127.0.0.1:{}\", port);", + "let line = format!(\"line {}:{}\", 0, col);", + "let a = format!(\"{host}:{port}\", host = 0, port = p);", + // Fields that are not ports, and port fields that are not zero or are types. + "let p = Point { x: 0, y: 0 };", + "let cfg = Config { port: held.port() };", + "struct Config { port: u16 }", + "fn f(port: u16) {}", + // Annotated, on the line or above it. + "TcpListener::bind(\"127.0.0.1:0\").unwrap(); // port-0: held by the server for its life", + "// port-0: the child binds it and reports the port it got.\n\"127.0.0.1:0\",", + ] { + assert_eq!(findings(fine), Vec::::new(), "{fine:?}"); + } +} diff --git a/crates/agent/tests/run_cli_flags.rs b/crates/agent/tests/run_cli_flags.rs index 92fd3f15..4e1db714 100644 --- a/crates/agent/tests/run_cli_flags.rs +++ b/crates/agent/tests/run_cli_flags.rs @@ -2456,7 +2456,7 @@ fn run_binary_idle_timeout_ms_flag_causes_a_stalled_response_to_fail_quickly() { // Task #19 (pi-parity feature): `with_idle_timeout` previously had zero callers. A server that // answers headers plus one partial event, then goes silent well past a deliberately shrunk // `--idle-timeout-ms`, must fail quickly instead of hanging on the default (~600s) read timeout. - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); std::thread::spawn(move || { use std::io::{Read as _, Write as _}; @@ -2522,7 +2522,7 @@ fn run_binary_retry_max_backoff_ms_flag_caps_the_exponential_backoff_between_ret // third succeeds. With `--retry-base-delay-ms` deliberately large (500ms) but // `--retry-max-backoff-ms` deliberately small (20ms), both waits between attempts must be capped // at ~20ms — total elapsed well under the ~1500ms (500ms + 1000ms) the uncapped schedule would take. - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let success_body = turn_text("ok"); std::thread::spawn(move || { diff --git a/crates/agent/tests/serve_compaction_retry.rs b/crates/agent/tests/serve_compaction_retry.rs index f561751a..d803746d 100644 --- a/crates/agent/tests/serve_compaction_retry.rs +++ b/crates/agent/tests/serve_compaction_retry.rs @@ -1198,7 +1198,7 @@ fn serve_idle_timeout_ms_flag_causes_a_stalled_response_to_fail_quickly() { // partial event, then goes silent well past a deliberately shrunk `--idle-timeout-ms`, must fail // the `prompt` quickly instead of hanging on the default (~600s) read timeout — `run`'s identical // `run_binary_idle_timeout_ms_flag_causes_a_stalled_response_to_fail_quickly`. - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); std::thread::spawn(move || { use std::io::Read as _; diff --git a/crates/agent/tests/serve_harness_deadlines.rs b/crates/agent/tests/serve_harness_deadlines.rs index a15b67d8..2a1c6ec3 100644 --- a/crates/agent/tests/serve_harness_deadlines.rs +++ b/crates/agent/tests/serve_harness_deadlines.rs @@ -36,7 +36,7 @@ fn a_silent_stdout_fails_the_read_at_its_deadline() { /// A WebSocket that stays open and silent fails the read at the deadline. #[tokio::test] async fn a_silent_websocket_fails_the_read_at_its_deadline() { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let server = tokio::spawn(async move { let (tcp, _) = listener.accept().await.unwrap(); @@ -287,580 +287,3 @@ fn only_child_frames_touches_a_childs_stdout() { "the suites read through child_frames ({readers})" ); } - -/// Every place `source` binds a socket to port 0, reads the port the kernel gave it, and lets the -/// socket go without using it for anything else — the port then handed on for something else to bind, -/// which any process can take first. Recognised by what happens to the socket, not by a helper's -/// name: -/// -/// - a **temporary**: `TcpListener::bind(…:0)…local_addr()` in one expression, the listener dropped -/// at the end of it (`.unwrap()`, `?`, `.and_then(|l| l.local_addr())` in between, any spelling); -/// - a **named** socket (`let l = TcpListener::bind(…:0)?` or `let s = TcpSocket::new_v4()?` then -/// `s.bind(…0…)`) whose only uses until the end of its block — or until the name is bound again — -/// are reading its address, configuring it, binding it, or `drop(l)`. -/// -/// A socket that is also accepted on, listened on, converted, stored or passed anywhere is held, and -/// fine. Comments are dropped and whitespace normalised first ([`normalised`]), so formatting cannot -/// hide a case. -fn released_ports(source: &str) -> Vec { - released_ports_with(source, &[]) -} - -/// Every source's findings, with address readers ([`address_readers`]) collected across all of them -/// first — so a socket handed to a helper defined in another file (`common::port_of(&l)`) is still a -/// port read, not a socket held. -fn released_ports_in(sources: &[&str]) -> Vec { - let readers: Vec = sources - .iter() - .flat_map(|s| address_readers(&normalised(s))) - .collect(); - sources - .iter() - .flat_map(|s| released_ports_with(s, &readers)) - .collect() -} - -fn released_ports_with(source: &str, other_readers: &[String]) -> Vec { - let code = normalised(source); - // Names holding a port-0 address (`let a = "127.0.0.1:0";`) or a zero port (`const ANY: u16 = - // 0;`), and functions or closures that only read the address of the socket they are given - // (`fn port_of(l: &TcpListener) -> u16`, `let port_of = |l| l.local_addr()…`). - let zero_names = zero_address_names(&code); - let zero_ports = zero_port_names(&code); - let mut readers = address_readers(&code); - readers.extend(other_readers.iter().cloned()); - let is_zero = |args: &str| { - binds_port_zero(args) - || zero_names.iter().any(|n| mentions_ident(args, n)) - || zero_ports.iter().any(|n| { - [")", "))"] - .iter() - .any(|end| args.contains(&format!(",{n}{end}"))) - }) - }; - let mut found = Vec::new(); - // Temporaries: the bind's own statement reads the address. - for ty in ["TcpListener::bind(", "TcpSocket::bind(", "UdpSocket::bind("] { - for (at, _) in code.match_indices(ty) { - let stmt = &code[at..]; - let stmt = &stmt[..stmt.find(';').unwrap_or(stmt.len())]; - let close = matching_paren(stmt, stmt.find('(').unwrap()); - if is_zero(&stmt[..close]) && stmt[close..].contains(".local_addr()") { - found.push(format!("a temporary listener's port ({})", excerpt(stmt))); - } - } - } - // Named sockets — one name, or each name of a tuple pattern: follow it to the end of its scope. - for (at, _) in code.match_indices("let") { - if code[..at].chars().last().is_some_and(is_ident) { - continue; - } - // `let x` keeps its space; `let (a, b)` loses it in `normalised`. - let after = &code[at + 3..]; - let Some(after) = after - .strip_prefix(' ') - .or_else(|| after.starts_with('(').then_some(after)) - else { - continue; - }; - let Some((names, init, after_stmt)) = let_binding(after) else { - continue; - }; - let bound_zero_here = (init.contains("TcpListener::bind(") - || init.contains("UdpSocket::bind(")) - && is_zero(init); - let socket_here = init.contains("TcpSocket::new_v4(") - || init.contains("TcpSocket::new_v6(") - || init.contains("Socket::new("); - if !bound_zero_here && !socket_here { - continue; - } - for name in names { - let rest = scope_after(after_stmt, &name); - let uses: Vec<&str> = uses_of(rest, &name, &readers) - .into_iter() - .map(through_option) - .collect(); - let reads_port = uses - .iter() - .any(|u| u.starts_with(".local_addr()") || *u == "read"); - let bound_zero = bound_zero_here - || uses - .iter() - .any(|u| u.starts_with(".bind(") && is_zero(&u[..matching_paren(u, 5)])); - let let_go = uses.iter().all(|u| { - u.starts_with(".local_addr()") - || u.starts_with(".bind(") - || u.starts_with(".set_") - || *u == "drop" - || *u == "read" - }); - if bound_zero && reads_port && let_go { - found.push(format!( - "`{name}`: bound to port 0, its port read, then let go" - )); - } - } - } - found -} - -fn is_ident(c: char) -> bool { - c.is_alphanumeric() || c == '_' -} - -/// A `let` binding, from just after `let `: the names it binds (one, or each non-`_` name of a tuple -/// pattern), its initialiser, and the code after its `;`. -fn let_binding(after: &str) -> Option<(Vec, &str, &str)> { - let after = after.strip_prefix("mut ").unwrap_or(after); - let (names, pattern_end) = if after.starts_with('(') { - let close = matching_paren(after, 0); - let names = after[1..close - 1] - .split(',') - .map(|n| n.trim_start_matches("mut ").trim_start_matches('&')) - .filter(|n| !n.is_empty() && *n != "_" && n.chars().all(is_ident)) - .map(str::to_owned) - .collect(); - (names, close) - } else { - let name: String = after.chars().take_while(|c| is_ident(*c)).collect(); - if name.is_empty() { - return None; - } - let len = name.len(); - (vec![name], len) - }; - let eq = after.find('=')?; - if eq < pattern_end - || !after[pattern_end..eq] - .chars() - .all(|c| is_ident(c) || "<>_:& ".contains(c)) - { - return None; - } - let stmt_end = after.find(';').unwrap_or(after.len()); - if stmt_end < eq { - return None; - } - Some((names, &after[eq + 1..stmt_end], &after[stmt_end..])) -} - -/// Whether identifier `name` occurs in `code` as a whole identifier. -fn mentions_ident(code: &str, name: &str) -> bool { - code.match_indices(name).any(|(i, m)| { - !code[..i].chars().last().is_some_and(is_ident) - && !code[i + m.len()..].chars().next().is_some_and(is_ident) - }) -} - -/// Names a port-0 address is bound to: `let a = "127.0.0.1:0";`, `const A: &str = "…:0";`, -/// `let a = SocketAddr::from(([127, 0, 0, 1], 0));` — anything not itself a bind. -fn zero_address_names(code: &str) -> Vec { - let mut names = Vec::new(); - for kw in ["let ", "const ", "static "] { - for (at, _) in code.match_indices(kw) { - if code[..at].chars().last().is_some_and(is_ident) { - continue; - } - let Some((found, init, _)) = let_binding(&code[at + kw.len()..]) else { - continue; - }; - if binds_port_zero(init) && !init.contains("bind(") { - names.extend(found); - } - } - } - names -} - -/// Names bound to a zero port number: `const ANY: u16 = 0;`, `let p = 0u16;`. -fn zero_port_names(code: &str) -> Vec { - let mut names = Vec::new(); - for kw in ["let ", "const ", "static "] { - for (at, _) in code.match_indices(kw) { - if code[..at].chars().last().is_some_and(is_ident) { - continue; - } - let Some((found, init, _)) = let_binding(&code[at + kw.len()..]) else { - continue; - }; - if ["0", "0u16", "0_u16"].contains(&init) { - names.extend(found); - } - } - } - names -} - -/// A use with `Option` access in front of it seen through: `l.as_ref().unwrap().local_addr()` reads -/// the address as surely as `l.local_addr()` does, for a listener held in an `Option`. -fn through_option(mut u: &str) -> &str { - loop { - if let Some(rest) = [".as_ref()", ".as_mut()", ".as_deref()", ".unwrap()"] - .iter() - .find_map(|p| u.strip_prefix(p)) - { - u = rest; - } else if u.starts_with(".expect(") { - u = &u[matching_paren(u, 7)..]; - } else { - return u; - } - } -} - -/// Functions of one parameter — and closures bound to a name — that do nothing with it but read its -/// address: passing a socket to one (`port_of(&l)`) is reading its port, not holding it. -fn address_readers(code: &str) -> Vec { - let mut readers = Vec::new(); - for (at, _) in code.match_indices("fn ") { - if code[..at].chars().last().is_some_and(is_ident) { - continue; - } - let rest = &code[at + 3..]; - let name: String = rest.chars().take_while(|c| is_ident(*c)).collect(); - let Some(open) = rest.find('(') else { continue }; - if open != name.len() { - continue; - } - let close = matching_paren(rest, open); - let params = &rest[open + 1..close - 1]; - if params.contains(',') { - continue; - } - let param: String = params - .trim_start_matches("mut ") - .chars() - .take_while(|c| is_ident(*c)) - .collect(); - let Some(body_open) = rest[close..].find('{').map(|i| close + i) else { - continue; - }; - let mut depth = 0; - let mut body_end = rest.len(); - for (i, c) in rest[body_open..].char_indices() { - match c { - '{' => depth += 1, - '}' => { - depth -= 1; - if depth == 0 { - body_end = body_open + i; - break; - } - } - _ => {} - } - } - let uses = uses_of(&rest[body_open..body_end], ¶m, &[]); - if !param.is_empty() - && !uses.is_empty() - && uses.iter().all(|u| u.starts_with(".local_addr()")) - { - readers.push(name); - } - } - // `let port_of = |l| l.local_addr()…;` (or `|l: &TcpListener| { … }`). - for (at, _) in code.match_indices("let ") { - if code[..at].chars().last().is_some_and(is_ident) { - continue; - } - let rest = &code[at + 4..]; - let name: String = rest.chars().take_while(|c| is_ident(*c)).collect(); - let Some(after) = rest[name.len()..] - .strip_prefix("=|") - .or_else(|| rest[name.len()..].strip_prefix("=move|")) - else { - continue; - }; - let Some(bar) = after.find('|') else { continue }; - let param: String = after[..bar].chars().take_while(|c| is_ident(*c)).collect(); - let body = &after[bar + 1..]; - let body = &body[..body.find(';').unwrap_or(body.len())]; - let uses = uses_of(body, ¶m, &[]); - if !name.is_empty() - && !param.is_empty() - && !after[..bar].contains(',') - && !uses.is_empty() - && uses.iter().all(|u| u.starts_with(".local_addr()")) - { - readers.push(name); - } - } - readers -} - -/// The index just past the `)` matching the `(` at `open`. -fn matching_paren(s: &str, open: usize) -> usize { - let mut depth = 0; - for (i, c) in s[open..].char_indices() { - match c { - '(' => depth += 1, - ')' => { - depth -= 1; - if depth == 0 { - return open + i + 1; - } - } - _ => {} - } - } - s.len() -} - -/// Whether a bind's arguments (or an address expression) ask for port 0: `"…:0"`, or a `0` (`0u16`, -/// `0_u16`) as the last element of an address tuple. -fn binds_port_zero(args: &str) -> bool { - if args.contains(":0\"") { - return true; - } - args.match_indices(",0").any(|(i, m)| { - let rest = &args[i + m.len()..]; - let rest = rest - .strip_prefix("_u16") - .or_else(|| rest.strip_prefix("u16")) - .unwrap_or(rest); - rest.starts_with(')') - }) -} - -/// `source` without comments, and with whitespace removed except a single space between two -/// identifier characters (`for x in listener.incoming()` keeps `in listener` apart), so a pattern -/// matches however the code is laid out. -fn normalised(source: &str) -> String { - let ident = |c: char| c.is_alphanumeric() || c == '_'; - let code: String = source - .lines() - .map(|line| line.split("//").next().unwrap_or("")) - .collect::>() - .join("\n"); - let mut out = String::with_capacity(code.len()); - let mut pending_space = false; - for c in code.chars() { - if c.is_whitespace() { - pending_space = true; - continue; - } - if pending_space && out.chars().last().is_some_and(ident) && ident(c) { - out.push(' '); - } - pending_space = false; - out.push(c); - } - out -} - -/// The code from a binding to the end of its block, or through the statement that binds `name` again -/// (whose right-hand side may still use the old binding: `let l = from_std(l)`). -fn scope_after<'a>(code: &'a str, name: &str) -> &'a str { - let mut depth = 0i32; - for (i, c) in code.char_indices() { - match c { - '{' => depth += 1, - '}' => { - depth -= 1; - if depth < 0 { - return &code[..i]; - } - } - _ => {} - } - let rebinds = [ - format!("let {name}="), - format!("let mut {name}="), - format!("let {name}:"), - ]; - if let Some(r) = rebinds.iter().find(|r| code[i..].starts_with(r.as_str())) { - let end = code[i..].find(';').map_or(code.len(), |e| i + e); - // `let l = l;` moves the same socket under the same name: keep following it. - if code[i + r.len()..end] == *name { - continue; - } - return &code[..end]; - } - } - code -} - -/// Every use of identifier `name` in `code`, as the text from the identifier's end — or `drop` for -/// `drop(name)` / `let _ = name;`, or `read` for passing it to an address reader (`port_of(&name)`). -/// Binding sites (`let name =`, the right-hand side of a self-rebind `let name = name;`) are not uses. -fn uses_of<'a>(code: &'a str, name: &str, readers: &[String]) -> Vec<&'a str> { - code.match_indices(name) - .filter(|(i, _)| !code[..*i].chars().last().is_some_and(is_ident)) - .filter(|(i, m)| !code[i + m.len()..].chars().next().is_some_and(is_ident)) - .filter(|(i, _)| !code[..*i].ends_with('.')) - .filter(|(i, _)| !code[..*i].ends_with("let ") && !code[..*i].ends_with("let mut ")) - .filter(|(i, m)| { - let self_rebind = code[..*i].ends_with(&format!("let {name}=")) - || code[..*i].ends_with(&format!("let mut {name}=")); - !(self_rebind && code[i + m.len()..].starts_with(';')) - }) - .map(|(i, m)| { - let after = &code[i + m.len()..]; - let before = &code[..i]; - if (before.ends_with("drop(") && after.starts_with(')')) - || (before.ends_with("let _=") && after.starts_with(';')) - { - "drop" - } else if after.starts_with(')') - && readers.iter().any(|r| { - [format!("{r}(&"), format!("{r}(")].iter().any(|call| { - before.ends_with(call.as_str()) - && !before[..before.len() - call.len()] - .chars() - .last() - .is_some_and(is_ident) - }) - }) - { - "read" - } else { - after - } - }) - .collect() -} - -fn excerpt(s: &str) -> &str { - &s[..s.len().min(60)] -} - -/// A helper that only reads the address, defined in one file (`tests/common`) and called from -/// another, is a port read like a local one. -#[test] -fn the_released_port_lint_sees_helpers_in_other_files() { - let common = "pub fn port_of(l: &TcpListener) -> u16 { l.local_addr().unwrap().port() }"; - let suite = "let l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet p = common::port_of(&l);\ndrop(l);"; - assert!( - released_ports(suite).is_empty(), - "alone, the helper is unknown" - ); - assert!( - !released_ports_in(&[common, suite]).is_empty(), - "with the file that defines it, the release is caught" - ); -} - -/// The lint recognises the shape — a port-0 socket whose port is read and which is then let go — -/// however it is written, and leaves a socket that is held, served or passed on alone. -#[test] -fn the_released_port_lint_catches_what_it_should() { - for bad in [ - // A temporary. - "let p = TcpListener::bind(\"127.0.0.1:0\").unwrap().local_addr().unwrap().port();", - "let p = std::net::TcpListener::bind(\"127.0.0.1:0\")\n .and_then(|l| l.local_addr())\n .unwrap()\n .port();", - "let p = TcpListener::bind((\"127.0.0.1\", 0)).await?.local_addr()?.port();", - // Named, under any name, in a helper or inline. - "fn pick() -> u16 { let l = TcpListener::bind(\"127.0.0.1:0\").unwrap(); l.local_addr().unwrap().port() }", - "let sock = std::net::TcpListener::bind(\"127.0.0.1:0\")?;\nlet port = sock.local_addr()?.port();\ndrop(sock);\nspawn_child(port);", - "let reserve = TcpListener::bind(\"0.0.0.0:0\").unwrap();\nlet n = reserve.local_addr().unwrap().port();\n}\nchild(n);", - // A socket bound to port 0 after creation. - "let s = TcpSocket::new_v4()?;\ns.set_reuseaddr(true)?;\ns.bind(\"127.0.0.1:0\".parse().unwrap())?;\nlet p = s.local_addr()?.port();\ndrop(s);", - // A port-0 address held in a variable. - "let a = \"127.0.0.1:0\";\nlet l = TcpListener::bind(a).unwrap();\nlet p = l.local_addr().unwrap().port();\ndrop(l);", - "const ANY: &str = \"127.0.0.1:0\";\nfn pick() -> u16 { TcpListener::bind(ANY).unwrap().local_addr().unwrap().port() }", - // A typed zero. - "let p = TcpListener::bind((\"127.0.0.1\", 0u16)).unwrap().local_addr().unwrap().port();", - "let p = TcpListener::bind((Ipv4Addr::LOCALHOST, 0_u16)).unwrap().local_addr().unwrap().port();", - // The socket handed to a helper that only reads its address. - "fn port_of(l: &TcpListener) -> u16 { l.local_addr().unwrap().port() }\nfn pick() -> u16 { let l = TcpListener::bind(\"127.0.0.1:0\").unwrap(); port_of(&l) }", - // A tuple pattern. - "let (l, _) = (TcpListener::bind(\"127.0.0.1:0\").unwrap(), 1);\nlet p = l.local_addr().unwrap().port();\ndrop(l);", - // Dropped by `let _ =`. - "let l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet p = l.local_addr().unwrap().port();\nlet _ = l;", - // A self-rebind on the way. - "let l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet l = l;\nlet p = l.local_addr().unwrap().port();\ndrop(l);", - // A closure that only reads the address. - "let port_of = |l: &TcpListener| l.local_addr().unwrap().port();\nlet l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet p = port_of(&l);\ndrop(l);", - // A listener held in an `Option`. - "let l = Some(TcpListener::bind(\"127.0.0.1:0\").unwrap());\nlet p = l.as_ref().unwrap().local_addr().unwrap().port();\ndrop(l);", - "let l = Some(TcpListener::bind(\"127.0.0.1:0\").unwrap());\nlet p = l.as_ref().expect(\"bound\").local_addr().unwrap().port();", - // A zero port in a const. - "const ANY: u16 = 0;\nlet p = TcpListener::bind((\"127.0.0.1\", ANY)).unwrap().local_addr().unwrap().port();", - // socket2. - "let s = Socket::new(Domain::IPV4, Type::STREAM, None)?;\ns.bind(&\"127.0.0.1:0\".parse::()?.into())?;\nlet p = s.local_addr()?.as_socket().unwrap().port();\ndrop(s);", - ] { - assert!(!released_ports(bad).is_empty(), "missed: {bad:?}"); - } - for fine in [ - // Held and served. - "let listener = TcpListener::bind(\"127.0.0.1:0\").await.unwrap();\nlet port = listener.local_addr().unwrap().port();\ntokio::spawn(serve(listener));", - // Held in a struct for the test's life. - "let socket = TcpSocket::new_v4().unwrap();\nsocket.bind(addr0).unwrap();\nlet port = socket.local_addr().unwrap().port();\nSelf { _socket: socket, port }", - "let l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet port = l.local_addr().unwrap().port();\nlet (conn, _) = l.accept().unwrap();", - // A fixed port: nothing picked. - "let l = TcpListener::bind(\"127.0.0.1:8080\").unwrap();\nlet a = l.local_addr().unwrap();\ndrop(l);", - "// let p = TcpListener::bind(\"127.0.0.1:0\").unwrap().local_addr().unwrap().port();", - "let dead = DeadPort::bind();", - // Handed to a helper that serves it, not one that only reads its address. - "fn serve(l: TcpListener) { for c in l.incoming() { drop(c); } }\nlet l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet p = l.local_addr().unwrap().port();\nserve(l);", - // A port-0 address that is bound and kept. - "let a = \"127.0.0.1:0\";\nlet l = TcpListener::bind(a).unwrap();\nlet p = l.local_addr().unwrap().port();\nspawn(l);", - // Used right after a keyword, which a whitespace-blind match would glue to it. - "let listener = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet addr = listener.local_addr().unwrap();\nthread::spawn(move || { for s in listener.incoming() { drop(s); } });", - // Re-bound by a statement that consumes the old binding. - "let listener = std::net::TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet port = listener.local_addr().unwrap().port();\nlet listener = tokio::net::TcpListener::from_std(listener).unwrap();", - // A helper whose name merely ends in a reader's name. - "fn port_of(l: &TcpListener) -> u16 { l.local_addr().unwrap().port() }\nfn my_port_of(l: &TcpListener) { serve(l) }\nlet l = TcpListener::bind(\"127.0.0.1:0\").unwrap();\nlet p = l.local_addr().unwrap().port();\nmy_port_of(&l);", - ] { - assert_eq!(released_ports(fine), Vec::::new(), "{fine:?}"); - } -} - -/// No code in the workspace — any crate's sources, tests or benches — picks a port, releases it and -/// hands it to something else to bind: every child binds its own (port 0, read back: `serve`'s -/// announcement, the gateway's own `LISTEN` sockets, `nats-server --ports_file_dir`), is handed a -/// socket the test holds (`HeldPort`, the forwarder in front of a restartable `nats-server`), or is -/// pointed at one held bound (`DeadPort`). The released-port helpers are gone from every crate; this -/// keeps the pattern from coming back. -#[test] -fn no_test_picks_a_port_and_releases_it() { - let crates = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) - .parent() - .unwrap() - .to_path_buf(); - let mut files = Vec::new(); - let mut dirs = vec![crates.clone()]; - while let Some(dir) = dirs.pop() { - for entry in std::fs::read_dir(&dir).unwrap() { - let path = entry.unwrap().path(); - let name = path.file_name().unwrap().to_string_lossy().into_owned(); - if path.is_dir() { - if name != "target" && !name.starts_with('.') { - dirs.push(path); - } - } else if name.ends_with(".rs") && name != "serve_harness_deadlines.rs" { - files.push(path); - } - } - } - for krate in ["agent", "gateway", "verify", "fleet-sim", "test-support"] { - assert!( - files.iter().any(|f| f.starts_with(crates.join(krate))), - "the scan covers the {krate} crate" - ); - } - let sources: Vec = files - .iter() - .map(|path| std::fs::read_to_string(path).unwrap_or_default()) - .collect(); - // Readers from every file: a helper in `tests/common` is called from the suites. - let readers: Vec = sources - .iter() - .flat_map(|s| address_readers(&normalised(s))) - .collect(); - let found: Vec = files - .iter() - .zip(&sources) - .flat_map(|(path, source)| { - released_ports_with(source, &readers) - .into_iter() - .map(move |what| format!("{}: {what}", path.display())) - }) - .collect(); - assert!( - found.is_empty(), - "a port picked free and released for something else to bind:\n{}", - found.join("\n") - ); -} diff --git a/crates/agent/tests/serve_lifecycle.rs b/crates/agent/tests/serve_lifecycle.rs index 4916dc44..c17d9a49 100644 --- a/crates/agent/tests/serve_lifecycle.rs +++ b/crates/agent/tests/serve_lifecycle.rs @@ -9,7 +9,6 @@ mod common; use std::io::{Read, Write}; -use std::net::TcpListener; use std::process::Stdio; use std::sync::{Arc, Mutex}; use std::thread; @@ -67,7 +66,7 @@ impl Collector { } fn spawn_tcp(held: bool) -> Self { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let posts = Arc::new(Mutex::new(Vec::new())); let hold = Arc::new(Hold { diff --git a/crates/agent/tests/serve_service_mode.rs b/crates/agent/tests/serve_service_mode.rs index 0e970d20..bc8384be 100644 --- a/crates/agent/tests/serve_service_mode.rs +++ b/crates/agent/tests/serve_service_mode.rs @@ -69,6 +69,7 @@ fn service_cmd(k: &Keys) -> Command { "--model", "claude-test", "--listen", + // port-0: the child binds it and keeps it (and every use is a startup meant to fail). "127.0.0.1:0", "--grant-key", &k.flag, @@ -91,6 +92,7 @@ fn service_mode_demands_a_verifier_a_shard_and_a_listener() { "--model".into(), "claude-test".into(), ]; + // port-0: the child binds it and keeps it (and every use is a startup meant to fail). let listen = ["--listen".to_string(), "127.0.0.1:0".to_string()]; let verifier = [ "--grant-key".to_string(), diff --git a/crates/agent/tests/serve_session_isolation.rs b/crates/agent/tests/serve_session_isolation.rs index 3c2c150a..fc64636a 100644 --- a/crates/agent/tests/serve_session_isolation.rs +++ b/crates/agent/tests/serve_session_isolation.rs @@ -521,3 +521,61 @@ async fn concurrent_sessions_persist_to_distinct_files() { let _ = child.kill(); let _ = child.wait(); } + +/// `BEYOND_AI_AGENT_TEST_FAIL_CHECKPOINT=n` counts each session's checkpoint writes on its own. On a +/// daemon a process-wide count would let one session's writes decide which write of another's fails. +/// Here each session's second write (the one before its tool) must fail, so neither tool runs. With +/// a shared count, bravo's writes would be the daemon's third and fourth: bravo's tool would run, +/// and its `prompt` would succeed. +#[tokio::test] +async fn an_injected_checkpoint_failure_counts_each_sessions_writes() { + let dir = tempfile::tempdir().unwrap(); + let marker = |name: &str| dir.path().join(format!("{name}-ran")); + let touch = |name: &str| { + turn_tool_use( + "toolu_1", + "bash", + &json!({ "command": format!("touch {}", marker(name).display()) }).to_string(), + ) + }; + let (base, _requests) = spawn_model_server(vec![touch("alpha"), touch("bravo")]); + let sessions = tempfile::tempdir().unwrap(); + let (_daemon, port) = spawn_listening( + Command::new(BIN) + .args([ + "serve", + "--gateway-url", + &base, + "--key", + "bai_v1.test", + "--model", + "claude-test", + "--session-dir", + &sessions.path().to_string_lossy(), + ]) + // Write 1 of a session is its prompt's checkpoint; write 2 is the one before its tool. + .env("BEYOND_AI_AGENT_TEST_FAIL_CHECKPOINT", "2") + .env("HOME", ISOLATED_HOME) + .stdin(Stdio::null()) + .stdout(Stdio::null()), + ); + // One after the other, so the writes are ordered: alpha's two, then bravo's two. + for name in ["alpha", "bravo"] { + let mut ws = ws_connect(port, Some(&format!("ckpt-{name}"))).await; + ws_send(&mut ws, json!({ "type": "prompt", "message": "go" })).await; + let frames = ws_read_until_response(&mut ws, "prompt").await; + let response = frames.last().unwrap(); + assert_eq!(response["success"], false, "{name}: {}", dump(&frames)); + assert!( + response["error"] + .as_str() + .is_some_and(|e| e.contains("simulated checkpoint write failure")), + "{name}: {}", + dump(&frames) + ); + assert!( + !marker(name).exists(), + "{name}'s tool ran: its own second checkpoint write was not the one that failed" + ); + } +} diff --git a/crates/agent/tests/serve_systemd.rs b/crates/agent/tests/serve_systemd.rs index e00a785d..d7677c31 100644 --- a/crates/agent/tests/serve_systemd.rs +++ b/crates/agent/tests/serve_systemd.rs @@ -131,7 +131,7 @@ fn serve_refuses_passed_sockets_without_its_own_listen_pid() { let (base, _requests) = spawn_model_server(vec![]); for listen_pid in [None, Some("1")] { let dir = tempfile::tempdir().unwrap(); - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let mut cmd = common::serve_dir_cmd( env!("CARGO_BIN_EXE_beyond-ai-agent"), &base, diff --git a/crates/agent/tests/smoke.rs b/crates/agent/tests/smoke.rs index e459a1cf..e1a4dea2 100644 --- a/crates/agent/tests/smoke.rs +++ b/crates/agent/tests/smoke.rs @@ -18,7 +18,7 @@ mod common; use std::io::{BufRead, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::TcpStream; use std::path::Path; use std::process::{Command, Output, Stdio}; use std::thread; @@ -1977,7 +1977,7 @@ fn smoke_split_turn_compaction_fires_mid_turn_live() { /// the proxy's own base URL. Blocking/threaded, matching this file's existing synchronous style (no /// tokio runtime is otherwise running in this test binary). fn spawn_flaky_proxy(real_target: &str, fail_first_n: usize) -> String { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let real_addr = real_target .trim_start_matches("http://") diff --git a/crates/agent/tests/web_tool.rs b/crates/agent/tests/web_tool.rs index 4a5fadef..7b46719b 100644 --- a/crates/agent/tests/web_tool.rs +++ b/crates/agent/tests/web_tool.rs @@ -10,7 +10,6 @@ mod common; use std::io::{Read, Write}; -use std::net::TcpListener; use std::sync::{Arc, Mutex}; use common::{run_cmd, spawn_model_server, turn_text, turn_tool_use}; @@ -22,7 +21,7 @@ const BIN: &str = env!("CARGO_BIN_EXE_beyond-ai-agent"); /// HTTP response including status line and headers). Records each raw request for assertions. Returns /// `(base_url, requests)`; the thread lives for the test process. fn web_fixture(raw_response: String) -> (String, Arc>>) { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = beyond_ai_test_support::ports::listener(); let addr = listener.local_addr().unwrap(); let requests = Arc::new(Mutex::new(Vec::new())); let recorder = requests.clone(); diff --git a/crates/fleet-sim/src/replica.rs b/crates/fleet-sim/src/replica.rs index 113e2e7b..e22cb7fa 100644 --- a/crates/fleet-sim/src/replica.rs +++ b/crates/fleet-sim/src/replica.rs @@ -101,9 +101,9 @@ impl Replica { gateway_url, "--model", "claude-test", - // A port the kernel picks, read back below: never one picked free and released, which - // another process can take before the replica binds it. "--listen", + // port-0: the replica binds it and keeps it; the port is read back below. Never one + // picked free and released, which another process can take before the replica binds it. "127.0.0.1:0", "--grant-key", grant_key_flag, @@ -121,6 +121,7 @@ impl Replica { } if with_metrics { // Loopback only, which the flag enforces — the scrape describes every tenant here. + // port-0: the replica binds it and keeps it. c.arg("--metrics-listen").arg("127.0.0.1:0"); } // The replica's own `$HOME` must not be reachable: in service mode a host default is a diff --git a/crates/gateway/tests/cancellation.rs b/crates/gateway/tests/cancellation.rs index dbca1424..9eba38ef 100644 --- a/crates/gateway/tests/cancellation.rs +++ b/crates/gateway/tests/cancellation.rs @@ -230,7 +230,7 @@ async fn closes_reused_connections_unread() -> (u16, std::sync::Arc (u16, std::sync::Arc) { use tokio::io::AsyncReadExt; - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let accepted = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let counter = accepted.clone(); diff --git a/crates/gateway/tests/claims_reliability.rs b/crates/gateway/tests/claims_reliability.rs index 1a8ab089..771b26ac 100644 --- a/crates/gateway/tests/claims_reliability.rs +++ b/crates/gateway/tests/claims_reliability.rs @@ -75,7 +75,7 @@ struct RawUpstream { impl RawUpstream { async fn start() -> Self { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let events = Arc::new(Mutex::new(Vec::new())); let conns = Arc::new(AtomicUsize::new(0)); @@ -568,13 +568,9 @@ async fn expect_100_continue_never_stalls() { /// without reuseport is refused. (A server restarted on its old port could find it taken; closing the /// listener without the holder would release it.) /// -/// **Linux semantics.** That holding relies on Linux's `SO_REUSEPORT`: a socket may share a port only -/// if it too set the option (and runs as the same user), and the kernel's port-0 search skips ports a -/// reuseport group holds. The gateway suites run on Linux CI, which is where this is relied on. On -/// the BSDs and macOS the option means something different (it lets any later socket with the option -/// share the port, and the last one bound may take the traffic), so the hold would be weaker there — -/// and there is no portable non-reuseport alternative: an ordinary socket bound to the port cannot -/// coexist with the listener that has to be bound to it while the server is up. +/// **Linux semantics.** The holder is `ports::bound_reuseport_socket`, which relies on Linux's +/// `SO_REUSEPORT`; its documentation says what differs elsewhere, and why there is no portable +/// alternative. struct RestartableNats { child: Option, /// The forwarder's port: what the gateway and the test dial. @@ -595,11 +591,7 @@ impl RestartableNats { let dir = std::env::temp_dir().join(format!("beyond-ai-rel13-{}-{n}", std::process::id())); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap(); - let holder = tokio::net::TcpSocket::new_v4().unwrap(); - holder.set_reuseport(true).unwrap(); - holder - .bind(std::net::SocketAddr::from(([127, 0, 0, 1], 0))) - .unwrap(); + let holder = beyond_ai_test_support::ports::bound_reuseport_socket(); let port = holder.local_addr().unwrap().port(); RestartableNats { child: None, @@ -1461,7 +1453,7 @@ async fn counting_h2_upstream( use hyper::service::service_fn; use hyper_util::rt::{TokioExecutor, TokioIo}; let _ = rustls::crypto::ring::default_provider().install_default(); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let ck = rcgen::generate_simple_self_signed(vec!["127.0.0.1".to_string()]).unwrap(); let key = rustls::pki_types::PrivateKeyDer::Pkcs8(ck.key_pair.serialize_der().into()); @@ -1637,7 +1629,7 @@ async fn goaway_upstream() -> (u16, Arc, tokio::task::JoinHandle<() use hyper::service::service_fn; use hyper_util::rt::{TokioExecutor, TokioIo}; let _ = rustls::crypto::ring::default_provider().install_default(); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let ck = rcgen::generate_simple_self_signed(vec!["127.0.0.1".to_string()]).unwrap(); let key = rustls::pki_types::PrivateKeyDer::Pkcs8(ck.key_pair.serialize_der().into()); @@ -1736,7 +1728,7 @@ async fn refusing_h2_upstream( tokio::task::JoinHandle<()>, ) { let _ = rustls::crypto::ring::default_provider().install_default(); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let ck = rcgen::generate_simple_self_signed(vec!["127.0.0.1".to_string()]).unwrap(); let key = rustls::pki_types::PrivateKeyDer::Pkcs8(ck.key_pair.serialize_der().into()); diff --git a/crates/gateway/tests/claims_security.rs b/crates/gateway/tests/claims_security.rs index a118cb8f..a52cda0a 100644 --- a/crates/gateway/tests/claims_security.rs +++ b/crates/gateway/tests/claims_security.rs @@ -112,7 +112,7 @@ struct EchoUpstream { impl EchoUpstream { async fn start(mode: Echo) -> Self { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let task = tokio::spawn(async move { while let Ok((stream, _)) = listener.accept().await { diff --git a/crates/gateway/tests/common/mod.rs b/crates/gateway/tests/common/mod.rs index a8160bf8..8807357e 100644 --- a/crates/gateway/tests/common/mod.rs +++ b/crates/gateway/tests/common/mod.rs @@ -25,7 +25,6 @@ use hyper::{Request, Response}; use hyper_util::rt::{TokioExecutor, TokioIo}; use hyper_util::server::conn::auto; use store::Connection; -use tokio::net::TcpListener; use tokio::time::{sleep, timeout}; use tokio_rustls::TlsAcceptor; @@ -168,15 +167,26 @@ pub fn test_keypair(seed: u8) -> (Vec, ed25519_dalek::SigningKey) { // --- nats-server (JetStream) ------------------------------------------------ pub struct Nats { - child: Child, - /// The port it chose (`-p -1`) and reported (`--ports_file_dir`); `0` until [`Nats::spawn_ready`] - /// has read it. + server: NatsProcess, + /// The port it chose (`-p -1`) and reported (`--ports_file_dir`). pub port: u16, +} + +/// A started `nats-server`, before its port is known: stopped, and its store removed, on drop. +struct NatsProcess { + child: Child, store_dir: std::path::PathBuf, /// The `server_name` it was started with, unique to this test process. name: String, } +impl Drop for NatsProcess { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = std::fs::remove_dir_all(&self.store_dir); + } +} + impl Nats { /// A store directory (holding the ports file directory) and a server name for one `nats-server`. fn layout(store_prefix: &str) -> (std::path::PathBuf, String) { @@ -194,7 +204,7 @@ impl Nats { store_dir.join("ports") } - fn spawn(store_prefix: &str) -> Self { + fn spawn(store_prefix: &str) -> NatsProcess { let (store_dir, name) = Self::layout(store_prefix); let child = Command::new("nats-server") .args(["-js", "-a", "127.0.0.1", "-p", "-1", "-n", &name]) @@ -206,9 +216,8 @@ impl Nats { .stderr(std::process::Stdio::null()) .spawn() .expect("spawn nats-server (on PATH? run via mise)"); - Nats { + NatsProcess { child, - port: 0, store_dir, name, } @@ -220,7 +229,7 @@ impl Nats { /// busy machine). A shell watchdog polls this test process and stops the server, and removes its /// store, within a second of the process exiting, however it exits. The shell exits as soon as /// the server does, so a server that failed to start shows as an exited `child` here too. - fn spawn_reaped(store_prefix: &str) -> Self { + fn spawn_reaped(store_prefix: &str) -> NatsProcess { const WATCHDOG: &str = r#"nats-server -js -a 127.0.0.1 -p -1 --ports_file_dir "$2" -sd "$3" -n "$4" >/dev/null 2>&1 & server=$! echo "$server" > "$3/server.pid" @@ -243,9 +252,8 @@ rm -rf "$3""#; .stderr(std::process::Stdio::null()) .spawn() .expect("spawn nats-server (on PATH? run via mise)"); - Nats { + NatsProcess { child, - port: 0, store_dir, name, } @@ -255,7 +263,7 @@ rm -rf "$3""#; /// writes only after it is listening. The port was never anyone else's, so there is nothing to /// lose to another process and nothing to respawn: a server that exits, or that is not up within /// [`STARTUP_BUDGET`], fails the test. - fn spawn_ready(spawn: fn(&str) -> Self, store_prefix: &str) -> Self { + fn spawn_ready(spawn: fn(&str) -> NatsProcess, store_prefix: &str) -> Self { let mut nats = spawn(store_prefix); let ports = Self::ports_dir_of(&nats.store_dir); let deadline = std::time::Instant::now() + STARTUP_BUDGET; @@ -267,8 +275,7 @@ rm -rf "$3""#; .and_then(|p| p.trim().parse().ok()) .unwrap_or_else(|| nats.child.id()); if let Some(port) = beyond_ai_test_support::ports::nats_port_from(&ports, server_pid) { - nats.port = port; - return nats; + return Nats { server: nats, port }; } if let Ok(Some(status)) = nats.child.try_wait() { panic!( @@ -295,15 +302,8 @@ rm -rf "$3""#; impl Nats { /// Kill the server mid-test (for fail-open coverage). Idempotent with `Drop`. pub fn stop(&mut self) { - let _ = self.child.kill(); - let _ = self.child.wait(); - } -} - -impl Drop for Nats { - fn drop(&mut self) { - let _ = self.child.kill(); - let _ = std::fs::remove_dir_all(&self.store_dir); + let _ = self.server.child.kill(); + let _ = self.server.child.wait(); } } @@ -873,7 +873,7 @@ impl MockUpstream { pub async fn start(mode: Mode) -> Self { // Bind `:0` and read the port back, keeping the listener open the whole time — no // pick-release-rebind window for another test to slip into (this is an in-process server). - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let captured: Arc>> = Arc::new(Mutex::new(None)); let hits = Arc::new(std::sync::atomic::AtomicUsize::new(0)); @@ -919,7 +919,7 @@ impl MockUpstream { // no default), pick ring to match the gateway. Idempotent across multiple mocks in one process. let _ = rustls::crypto::ring::default_provider().install_default(); - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let ck = rcgen::generate_simple_self_signed(vec![ @@ -1733,7 +1733,7 @@ impl ScriptedUpstream { /// `script(body, n)` decides the reply to the `n`th (0-based) fully received request. pub async fn start(script: impl Fn(&[u8], usize) -> Vec + Send + Sync + 'static) -> Self { let script: Script = Arc::new(script); - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let bodies: Arc>>> = Arc::new(Mutex::new(Vec::new())); let seen = Arc::clone(&bodies); @@ -2111,7 +2111,7 @@ impl ReplyUpstream { script: impl Fn(usize, &ScriptReq) -> Reply + Send + Sync + 'static, ) -> Self { let script: ReplyScript = Arc::new(script); - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let hits = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let counter = hits.clone(); diff --git a/crates/gateway/tests/reliability_early_response.rs b/crates/gateway/tests/reliability_early_response.rs index f9f487d3..80d43298 100644 --- a/crates/gateway/tests/reliability_early_response.rs +++ b/crates/gateway/tests/reliability_early_response.rs @@ -34,7 +34,7 @@ enum AfterAnswer { /// headers arrive, without reading the body. Returns its port and how many requests it got. async fn early_answer_upstream(after: AfterAnswer) -> (u16, Arc) { let _ = rustls::crypto::ring::default_provider().install_default(); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let ck = rcgen::generate_simple_self_signed(vec!["127.0.0.1".to_string()]).unwrap(); let key = rustls::pki_types::PrivateKeyDer::Pkcs8(ck.key_pair.serialize_der().into()); diff --git a/crates/gateway/tests/reliability_large_body.rs b/crates/gateway/tests/reliability_large_body.rs index 10da8845..c7caa9c4 100644 --- a/crates/gateway/tests/reliability_large_body.rs +++ b/crates/gateway/tests/reliability_large_body.rs @@ -237,7 +237,7 @@ async fn mid_upload_resetter() -> ( tokio::task::JoinHandle<()>, ) { use tokio::io::AsyncReadExt; - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let addr = listener.local_addr().unwrap(); let resets = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let counter = resets.clone(); @@ -315,7 +315,7 @@ async fn a_reset_during_the_upload_fails_over() { async fn a_reset_before_a_small_body_is_sent_fails_over() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let (pubkey, sk) = test_keypair(1); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let primary = listener.local_addr().unwrap(); let resets = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let counter = resets.clone(); diff --git a/crates/gateway/tests/reliability_lifecycle.rs b/crates/gateway/tests/reliability_lifecycle.rs index 28ea302f..2a2f6e69 100644 --- a/crates/gateway/tests/reliability_lifecycle.rs +++ b/crates/gateway/tests/reliability_lifecycle.rs @@ -144,7 +144,7 @@ async fn h2_upstream(head_delay: Duration, freeze: bool) -> (u16, tokio::task::J use hyper_util::rt::{TokioExecutor, TokioIo}; use std::sync::Arc; let _ = rustls::crypto::ring::default_provider().install_default(); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let ck = rcgen::generate_simple_self_signed(vec!["127.0.0.1".to_string()]).unwrap(); let key = rustls::pki_types::PrivateKeyDer::Pkcs8(ck.key_pair.serialize_der().into()); diff --git a/crates/gateway/tests/request_deadlines.rs b/crates/gateway/tests/request_deadlines.rs index d17c602c..d2135ffe 100644 --- a/crates/gateway/tests/request_deadlines.rs +++ b/crates/gateway/tests/request_deadlines.rs @@ -260,7 +260,7 @@ async fn request_max_secs_zero_disables_the_ceiling() { /// and how many connections it has accepted. async fn silent_upstream() -> (u16, Arc) { use tokio::io::AsyncReadExt; - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr().unwrap().port(); let accepted = Arc::new(AtomicUsize::new(0)); let counter = accepted.clone(); @@ -547,8 +547,7 @@ async fn a_large_body_re_run_keeps_the_requests_deadline() { /// A listener whose accept queue is full: a connect to it hangs (the kernel drops the SYN) until /// the connector's own timeout. Returns its authority and the connections that filled it. async fn unaccepting_upstream() -> (String, Vec) { - let sock = tokio::net::TcpSocket::new_v4().unwrap(); - sock.bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let sock = beyond_ai_test_support::ports::bound_socket(); let addr = sock.local_addr().unwrap(); let listener = sock.listen(0).unwrap(); let mut held = Vec::new(); diff --git a/crates/test-support/src/exec_mock.rs b/crates/test-support/src/exec_mock.rs index df68e052..dccd5652 100644 --- a/crates/test-support/src/exec_mock.rs +++ b/crates/test-support/src/exec_mock.rs @@ -65,7 +65,12 @@ impl ExecMock { /// A sandbox: commands run with `cwd` defaulting to `root` and `HOME` set to `home`, so `$HOME` /// (what `serve --service`'s startup probe asks for) is the sandbox's, never the replica's. pub async fn start_with_home(root: &Path, home: Option<&Path>, honors_stdin: bool) -> Self { - Self::start_on("127.0.0.1:0", root, home, honors_stdin).await + Self::serve( + crate::ports::tokio_listener().await, + root, + home, + honors_stdin, + ) } /// As [`Self::start_with_home`], bound where the caller says. @@ -81,6 +86,16 @@ impl ExecMock { honors_stdin: bool, ) -> Self { let listener = tokio::net::TcpListener::bind(bind).await.unwrap(); + Self::serve(listener, root, home, honors_stdin) + } + + /// Serve on `listener`, which the endpoint holds for as long as it runs. + fn serve( + listener: tokio::net::TcpListener, + root: &Path, + home: Option<&Path>, + honors_stdin: bool, + ) -> Self { let url = format!("http://{}/exec", listener.local_addr().unwrap()); let requests = Arc::new(Mutex::new(Vec::new())); let root = root.to_path_buf(); diff --git a/crates/test-support/src/lib.rs b/crates/test-support/src/lib.rs index bea201f0..fbae0206 100644 --- a/crates/test-support/src/lib.rs +++ b/crates/test-support/src/lib.rs @@ -254,7 +254,7 @@ fn serve_titles_only(listener: TcpListener, recorder: Option) -> (String, Arc>>) { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = crate::ports::listener(); let addr = listener.local_addr().unwrap(); let requests = Arc::new(Mutex::new(Vec::new())); let recorder = requests.clone(); @@ -311,7 +311,8 @@ pub fn spawn_model_server_routed_unrecorded_on( routes: Vec<(String, String)>, fallback: String, ) -> String { - spawn_model_server_routed_at(bind, routes, fallback, false).0 + let listener = TcpListener::bind(bind).unwrap(); + spawn_model_server_routed_at(listener, routes, fallback, false).0 } fn spawn_model_server_routed_inner( @@ -319,16 +320,15 @@ fn spawn_model_server_routed_inner( fallback: String, record: bool, ) -> (String, Arc>>) { - spawn_model_server_routed_at("127.0.0.1:0", routes, fallback, record) + spawn_model_server_routed_at(crate::ports::listener(), routes, fallback, record) } fn spawn_model_server_routed_at( - bind: &str, + listener: TcpListener, routes: Vec<(String, String)>, fallback: String, record: bool, ) -> (String, Arc>>) { - let listener = TcpListener::bind(bind).unwrap(); let addr = listener.local_addr().unwrap(); let requests = Arc::new(Mutex::new(Vec::new())); let recorder = requests.clone(); @@ -419,7 +419,7 @@ pub fn spawn_model_server_with_stalled_response( stall: std::time::Duration, after: Vec, ) -> String { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let listener = crate::ports::listener(); let addr = listener.local_addr().unwrap(); thread::spawn(move || { // Session-title requests do not consume a scripted response (see [`accept_scripted`]). diff --git a/crates/test-support/src/ports.rs b/crates/test-support/src/ports.rs index c2610f1b..461edaba 100644 --- a/crates/test-support/src/ports.rs +++ b/crates/test-support/src/ports.rs @@ -11,14 +11,66 @@ //! - **`nats-server`** is started with `-p -1 --ports_file_dir ` and writes the port it got to a //! file ([`nats_port_from`]). //! - **A port where nothing listens** is held bound, never listening ([`DeadPort`]). +//! - **An in-process server** listens on a port it holds for its whole life ([`listener`], +//! [`tokio_listener`]); a socket configured before it listens starts from [`bound_socket`]. +//! +//! This is the one module that writes port 0: everywhere else, the workspace lint +//! (`crates/agent/tests/ports_lint.rs`) fails on a zero port outside an annotated, allow-listed site. +//! A port is only ever read from a socket that stays held, so nothing here can hand out a port that +//! another process takes first. //! //! Reading a child's ports back needs Linux: its `LISTEN` sockets come from `/proc`, and the //! gateway's metrics listener sits on `127.0.0.2`, which Linux routes to loopback with no setup. //! Elsewhere the read-back fails at once and says so ([`require_proc`]), never a timeout. -use std::net::{SocketAddr, SocketAddrV4}; +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::path::Path; +/// A loopback listener on a port the kernel picks, held for as long as the value lives. +#[track_caller] +pub fn listener() -> std::net::TcpListener { + std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).expect("bind a loopback port") +} + +/// [`listener`], for tokio. Call it inside a runtime. +pub async fn tokio_listener() -> tokio::net::TcpListener { + tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)) + .await + .expect("bind a loopback port") +} + +/// A TCP socket bound to a loopback port the kernel picks, not yet listening — for a test that sets +/// its own backlog (`listen(0)`) or holds a port that refuses connections ([`DeadPort`]). +#[track_caller] +pub fn bound_socket() -> tokio::net::TcpSocket { + let socket = tokio::net::TcpSocket::new_v4().expect("a TCP socket"); + socket + .bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + .expect("bind a loopback port"); + socket +} + +/// [`bound_socket`] with `SO_REUSEPORT` set before the bind, so that a second socket that also sets it +/// can bind and listen on the same port while this one keeps holding it. A test uses this to take a +/// listener down and bring it back on the same port, with no gap in which another process could take +/// the port. +/// +/// **Linux semantics.** A socket may share the port only if it too set the option (and runs as the +/// same user), and the kernel's port-0 search skips ports a reuseport group holds. The suites that +/// use this run on Linux CI, which is where it is relied on. On the BSDs and macOS the option means +/// something different (any later socket with the option may share the port, and the last one bound +/// may take the traffic), so the hold would be weaker there. There is no portable alternative: an +/// ordinary socket bound to the port cannot coexist with the listener that has to bind it too. +#[track_caller] +pub fn bound_reuseport_socket() -> tokio::net::TcpSocket { + let socket = tokio::net::TcpSocket::new_v4().expect("a TCP socket"); + socket.set_reuseport(true).expect("SO_REUSEPORT"); + socket + .bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + .expect("bind a loopback port"); + socket +} + /// The gateway's two listeners, as config lines: each on a port the kernel picks. Each at its own /// loopback address, because Pingora keys a server's listeners by their address string — two /// `127.0.0.1:0` collide (one listener comes up, and shutdown closes its descriptor twice) — and the @@ -143,10 +195,7 @@ pub struct DeadPort { impl DeadPort { pub fn bind() -> Self { - let socket = tokio::net::TcpSocket::new_v4().expect("a TCP socket"); - socket - .bind(SocketAddr::from(([127, 0, 0, 1], 0))) - .expect("bind a loopback port"); + let socket = bound_socket(); let port = socket.local_addr().expect("its address").port(); Self { _socket: socket, diff --git a/crates/verify/tests/common/fault_proxy.rs b/crates/verify/tests/common/fault_proxy.rs index 04f344a1..a049c634 100644 --- a/crates/verify/tests/common/fault_proxy.rs +++ b/crates/verify/tests/common/fault_proxy.rs @@ -26,7 +26,7 @@ use std::time::{Duration, Instant}; use serde_json::Value; use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _}; -use tokio::net::{TcpListener, TcpStream}; +use tokio::net::TcpStream; use tokio_rustls::{TlsAcceptor, TlsConnector}; /// What the proxy does with one request (or, for [`Fault::ResetOnConnect`], one connection). @@ -189,7 +189,7 @@ impl FaultProxy { .with_no_client_auth(); client.alpn_protocols = vec![b"http/1.1".to_vec()]; - let listener = TcpListener::bind("127.0.0.1:0").await?; + let listener = beyond_ai_test_support::ports::tokio_listener().await; let port = listener.local_addr()?.port(); let state = Arc::new(State { upstream_host: upstream_host.to_owned(),