From da40aa6c44b0eaf5f2003e5837e1a4f5e7c3799b Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Fri, 13 Nov 2026 02:05:23 +0800 Subject: [PATCH] test: isolate SQL reader timeout from submission latency (#5000) The reader timeout regression can fail on a busy runner while submitting its real loopback Flight exchange, before it ever calls `reader()`. Freeze Tokio time and inhibit automatic clock advancement during submission, then let the existing 200ms result timeout fire. Preserve the exact result-stage error through both describe APIs and a second reader, and verify that the timeout closes the server response stream. With a diagnostic-only 300ms wall-clock delay in request-header preparation, main fails with `SQL query submission timed out`; the repaired fixture passes the same scenario through the real Flight/reader path. The focused regression and all 54 adjacent SQL tests pass locally. Hosted Linux and macOS CLI tests and the complete PR CI pass on `fb96931be004166a479cc031aca4b1db05f4b78d`. --- rust/lancedb/src/remote/sql_test.rs | 33 +++++++++++++++++++++++------ 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/rust/lancedb/src/remote/sql_test.rs b/rust/lancedb/src/remote/sql_test.rs index 471966731..d94defac4 100644 --- a/rust/lancedb/src/remote/sql_test.rs +++ b/rust/lancedb/src/remote/sql_test.rs @@ -607,7 +607,15 @@ impl FlightService for TestSqlService { )); } if query == "SELECT exchange stall" { - return Ok(Response::new(futures::stream::pending().boxed())); + let dropped = DropSignal(self.exchange_drops.clone()); + return Ok(Response::new( + futures::stream::pending() + .map(move |message| { + let _dropped = &dropped; + message + }) + .boxed(), + )); } let rows = parameters.ok_or_else(|| Status::invalid_argument("exchange without parameters"))?; @@ -990,13 +998,26 @@ async fn an_abandoned_reader_leaves_the_rows_for_the_next_one() { async fn a_reader_that_times_out_reports_the_failure() { let mut client_config = ClientConfig::default(); client_config.timeout_config.timeout = Some(Duration::from_millis(200)); - let (client, _shutdown) = start_sql_service(TestSqlService::default(), client_config).await; + let service = TestSqlService::default(); + let drops = service.exchange_drops.clone(); + let (client, _shutdown) = start_sql_service(service, client_config).await; - let stalled = client - .execute(parameterized("SELECT exchange stall", parameter_row())) - .await - .unwrap(); + // Real loopback I/O must finish before the clock can advance to the timeout. + // A blocking task inhibits Tokio's automatic clock advance during submission. + tokio::time::pause(); + let submission_client = client.clone(); + let stalled = tokio::task::spawn_blocking(move || { + tokio::runtime::Handle::current().block_on( + submission_client.execute(parameterized("SELECT exchange stall", parameter_row())), + ) + }) + .await + .unwrap() + .unwrap(); + assert_eq!(drops.load(Ordering::SeqCst), 0); assert_overall_timeout(stalled.reader().await, "result"); + tokio::time::resume(); + wait_for(&drops, 1, "a timed-out reader must end the call").await; let Err(retried) = stalled.reader().await else { panic!("the rows are gone, so a second reader cannot succeed"); };