diff --git a/benchmarks/aft-search/slice-descriptors/land-flake-batch-7.json b/benchmarks/aft-search/slice-descriptors/land-flake-batch-7.json new file mode 100644 index 0000000000..398b2a4241 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-flake-batch-7.json @@ -0,0 +1,12 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "harness-goldens" + ], + "slices": [ + "flake-batch-7" + ], + "note": "Test-only: timing and race fixes for merge-queue flakes (generous positive waits, waits on real signals). No product search behavior changed and no changed path matches RANKING_FENCE_PREFIXES." +} diff --git a/crates/aft/tests/integration/aft_search_contract_test.rs b/crates/aft/tests/integration/aft_search_contract_test.rs index 144752c360..88d66af268 100644 --- a/crates/aft/tests/integration/aft_search_contract_test.rs +++ b/crates/aft/tests/integration/aft_search_contract_test.rs @@ -460,11 +460,42 @@ fn start_mock_embedding_error_server() -> (String, thread::JoinHandle<()>) { (base_url, handle) } -/// How long the recovering server holds its first response. It is well past -/// the 1s bound `slow_query_embedding_degrades_within_budget_and_next_query_retries_fresh` +/// How long the recovering server holds its first response. It is far past +/// every bound `slow_query_embedding_degrades_within_budget_and_next_query_retries_fresh` /// asserts, so a search that waits for this response instead of honouring -/// its 500ms query budget fails that bound instead of slipping under it. -const SLOW_FIRST_EMBEDDING_RESPONSE: Duration = Duration::from_millis(2_000); +/// its 500ms query budget fails those bounds instead of slipping under them, +/// while a loaded runner that is merely slow at the lexical fallback after +/// the budget fires still passes. +const SLOW_FIRST_EMBEDDING_RESPONSE: Duration = Duration::from_millis(10_000); + +/// Holds a request's connection open without answering until `hold` has +/// passed or the client closes the connection, whichever comes first. A +/// client that gave up after its timeout no longer needs a reply, so ending +/// the hold there keeps the test from idling out the full hold at teardown. +fn hold_unless_client_disconnects(stream: &mut std::net::TcpStream, hold: Duration) { + let deadline = Instant::now() + hold; + let mut scratch = [0u8; 4096]; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() || stream.set_read_timeout(Some(remaining)).is_err() { + return; + } + match stream.read(&mut scratch) { + // The client closed its side: nobody is waiting for the reply. + Ok(0) => return, + // Late request bytes (for example a body sent after the headers). + Ok(_) => {} + // The read timed out; the loop re-checks the deadline. + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) => {} + // Reset or aborted connection: the client is gone. + Err(_) => return, + } + } +} fn start_recovering_slow_embedding_server() -> (String, Arc, thread::JoinHandle<()>) { let listener = TcpListener::bind("127.0.0.1:0").expect("bind slow embedding server"); @@ -480,7 +511,7 @@ fn start_recovering_slow_embedding_server() -> (String, Arc, thread let mut request = [0u8; 4096]; let _ = stream.read(&mut request); if ordinal == 0 { - thread::sleep(SLOW_FIRST_EMBEDDING_RESPONSE); + hold_unless_client_disconnects(&mut stream, SLOW_FIRST_EMBEDDING_RESPONSE); } let body = r#"{"data":[{"embedding":[0.1,0.2,0.3],"index":0}]}"#; let response = format!( @@ -1830,13 +1861,41 @@ fn slow_query_embedding_degrades_within_budget_and_next_query_retries_fresh() { // Checked first so a budget that is not enforced reports as a budget // failure: the search would then wait for the server's held response, - // which arrives well after this bound. - assert!( - elapsed < Duration::from_millis(1_000), - "500ms query budget took {elapsed:?}; phases: {}", + // which arrives only after this bound. The bound is deliberately loose + // (half the hold) because the lexical fallback that runs after the + // budget fires can take hundreds of milliseconds on a loaded runner; + // the phase checks below pin the budget itself. + assert!( + elapsed < SLOW_FIRST_EMBEDDING_RESPONSE / 2, + "500ms query budget took {elapsed:?}, as if it waited for the held \ + embedding response; phases: {}", format_query_embedding_phases(&phases, started) ); #[cfg(any(debug_assertions, feature = "test-timing-hooks"))] + { + let offset_of = |phase: &str| { + phases + .iter() + .find(|(name, _)| *name == phase) + .map(|(_, at)| at.duration_since(started)) + }; + // The budget is what ended the embedding request: its timeout became + // observable no earlier than the 500ms budget (less a margin for + // timer granularity) and long before the server would have replied. + let deadline_fired = offset_of("deadline_fire_observed").unwrap_or_else(|| { + panic!( + "500ms query budget never fired; phases: {}", + format_query_embedding_phases(&phases, started) + ) + }); + assert!( + deadline_fired >= Duration::from_millis(400) + && deadline_fired < SLOW_FIRST_EMBEDDING_RESPONSE / 4, + "500ms query budget fired at {deadline_fired:?}, not near the budget; phases: {}", + format_query_embedding_phases(&phases, started) + ); + } + #[cfg(any(debug_assertions, feature = "test-timing-hooks"))] assert_eq!( phases.iter().map(|(name, _)| *name).collect::>(), [