Skip to content

Commit b3d25d0

Browse files
committed
perf: replace fixed async poll interval with step-up backoff
The async query poll endpoint is a server-side long-poll: each GET blocks until the query completes or the server-side timeout elapses, so the client normally does not need to sleep. The previous fixed 10s sleep only added latency when a GET returned in-progress quickly.
1 parent 8183028 commit b3d25d0

2 files changed

Lines changed: 58 additions & 6 deletions

File tree

src/statement/client.rs

Lines changed: 55 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -83,15 +83,14 @@ impl StatementApiClient {
8383
timeout: Duration,
8484
query_id: Arc<str>,
8585
) -> QueryScopedResult<SnowflakeResponse> {
86-
const POLL_INTERVAL: Duration = Duration::from_secs(10);
87-
8886
let poll_url = match resolve_poll_url(&self.shared.base_url, poll_relative_url) {
8987
Ok(url) => url,
9088
Err(err) => {
9189
return Err(QueryScopedError::new(query_id, err));
9290
}
9391
};
9492
let deadline = Instant::now() + timeout;
93+
let mut backoff = PollBackoff::new();
9594

9695
loop {
9796
let resp = self
@@ -151,14 +150,45 @@ impl StatementApiClient {
151150
if remaining.is_zero() {
152151
return Err(QueryScopedError::new(query_id, TimeoutError::query()));
153152
}
154-
sleep(remaining.min(POLL_INTERVAL)).await;
153+
sleep(remaining.min(backoff.next_delay())).await;
155154
}
156155
Err(err) => return Err(QueryScopedError::new(query_id, err)),
157156
}
158157
}
159158
}
160159
}
161160

161+
/// Client-side backoff between consecutive poll GETs for an async query.
162+
///
163+
/// The poll endpoint is a server-side long-poll: each GET blocks until the query completes or the server-side
164+
/// timeout elapses, so the client normally does not sleep at all. This backoff only takes effect when a GET
165+
/// returns in-progress quickly, bounding how tightly the poll loop can spin in that case.
166+
struct PollBackoff {
167+
step: usize,
168+
}
169+
170+
impl PollBackoff {
171+
const STEPS: [Duration; 7] = [
172+
Duration::from_millis(500),
173+
Duration::from_millis(500),
174+
Duration::from_secs(1),
175+
Duration::from_millis(1500),
176+
Duration::from_secs(2),
177+
Duration::from_secs(4),
178+
Duration::from_secs(5),
179+
];
180+
181+
fn new() -> Self {
182+
Self { step: 0 }
183+
}
184+
185+
fn next_delay(&mut self) -> Duration {
186+
let delay = Self::STEPS[self.step];
187+
self.step = (self.step + 1).min(Self::STEPS.len() - 1);
188+
delay
189+
}
190+
}
191+
162192
fn resolve_poll_url(
163193
base_url: &Url,
164194
poll_relative_url: &str,
@@ -308,6 +338,28 @@ mod tests {
308338
let _ = server.await;
309339
}
310340

341+
#[test]
342+
fn poll_backoff_steps_up_then_plateaus_at_five_seconds() {
343+
let mut backoff = PollBackoff::new();
344+
let observed: Vec<Duration> = (0..9).map(|_| backoff.next_delay()).collect();
345+
346+
assert_eq!(
347+
observed,
348+
vec![
349+
Duration::from_millis(500),
350+
Duration::from_millis(500),
351+
Duration::from_secs(1),
352+
Duration::from_millis(1500),
353+
Duration::from_secs(2),
354+
Duration::from_secs(4),
355+
Duration::from_secs(5),
356+
// Plateaus at the 5s tail once the table is exhausted.
357+
Duration::from_secs(5),
358+
Duration::from_secs(5),
359+
]
360+
);
361+
}
362+
311363
#[test]
312364
fn validate_wire_rejects_empty_named_bind_keys() {
313365
let parts = into_statement_parts(Statement::new("SELECT :id").bind_named("", 1_i64));

tests/cases/test_async.rs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,19 +7,19 @@ use snowflake_connector_rs::{QueryConfig, Result};
77
#[tokio::test]
88
async fn test_async_query_picked_up_in_final_polling_window() -> Result<()> {
99
let query_config =
10-
QueryConfig::default().with_async_query_completion_timeout(Duration::from_secs(25));
10+
QueryConfig::default().with_async_query_completion_timeout(Duration::from_secs(15));
1111
let client = common::connect_with_query(query_config)?;
1212
let session = client.create_session().await?;
1313

1414
let table = session
15-
.query(r#"CALL SYSTEM$WAIT(65)"#)
15+
.query(r#"CALL SYSTEM$WAIT(50)"#)
1616
.await?
1717
.collect_table()
1818
.await?;
1919
assert_eq!(table.row_count(), 1);
2020

2121
let value = table.rows::<(String,)>()?.next().unwrap()?.0;
22-
assert_eq!(value, "waited 65 seconds");
22+
assert_eq!(value, "waited 50 seconds");
2323

2424
Ok(())
2525
}

0 commit comments

Comments
 (0)