From 271e67110d6f28dfbebe2e58b449b3943c8c6c21 Mon Sep 17 00:00:00 2001 From: Derek Date: Mon, 12 Oct 2026 17:20:05 +1100 Subject: [PATCH] fix: widen the ClickHouse query bound until reads answer A fixed 30 s bound on schema reads failed every attempt once ClickHouse took longer than that to answer, so nothing waiting on the schema ever landed and the gRPC push path never answered. - Queries now run under a shared bound that doubles after each timeout, up to the 300 s ClickHouse gives its own socket reads, and steps back down after a quick answer. - The inserter's schema reads share that bound with the query client. - clickhouse.request_timeout_ms defaults to 60000, the starting bound. - New test: schema reads answered slower than the request bound, and every pushed record still lands. --- docs/CONFIGURATION.md | 6 +- docs/config-schema.json | 6 +- docs/config-schema.yaml | 9 +- src/clickhouse/client_http.rs | 211 +++++++++++++++++++++++++------ src/clickhouse/config.rs | 14 +- src/clickhouse/inserter.rs | 21 +-- src/clickhouse_ext/insert.rs | 27 ++-- src/clickhouse_ext/schema.rs | 30 ++--- src/config/loader.rs | 9 +- tests/integration/grpc_outage.rs | 81 ++++++++++++ 10 files changed, 323 insertions(+), 91 deletions(-) diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 4538946..c97cbff 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -83,10 +83,12 @@ See [clickhouse/TLS.md](clickhouse/TLS.md). | Setting | Default | Bounds | |---------|---------|--------| | `clickhouse.connect_timeout_ms` | `5000` | opening a TCP connection | -| `clickhouse.request_timeout_ms` | `30000` | every query (schema reads, health), and an insert waiting to hand over its next chunk | +| `clickhouse.request_timeout_ms` | `60000` | the first try of every query (schema reads, health), and an insert waiting to hand over its next chunk. A query that times out doubles the bound for the next one, up to 300 s, so a slow but working ClickHouse still answers; a quick answer narrows it again | | `clickhouse.insert_timeout_ms` | `60000` | an insert waiting for the server's answer once its rows are sent | -A timed-out insert is held and sent again on a later flush, and its Kafka offset stays uncommitted until it lands. 60 s is about 270x the ClickHouse CPU cost of the largest default flush (20,000 rows at a measured 91,000 rows per CPU-second), and two rounds of timed-out concurrent inserts still fit Kafka's 300 s `max.poll.interval.ms`, which a blocked flush cycle counts against. Idle connections close on the client side after 2 s. +A timed-out insert is held and sent again on a later flush, and its Kafka offset stays uncommitted until it lands. 60 s is about 270x the ClickHouse CPU cost of the largest default flush (20,000 rows at a measured 91,000 rows per CPU-second), and two rounds of timed-out concurrent inserts still fit Kafka's 300 s `max.poll.interval.ms`, which a blocked flush cycle counts against. Through a full outage the query bound widens, so an insert that first has to read its table's schema can hold one flush cycle past that limit; the consumer then rejoins with the batch's offsets still uncommitted. + +Idle connections close on the client side after 2 s. ### Capture (what gets stored) diff --git a/docs/config-schema.json b/docs/config-schema.json index 4343b69..b647d5e 100644 --- a/docs/config-schema.json +++ b/docs/config-schema.json @@ -62,7 +62,7 @@ "tables": [], "tls": null, "connect_timeout_ms": 5000, - "request_timeout_ms": 30000, + "request_timeout_ms": 60000, "insert_timeout_ms": 60000 } }, @@ -707,11 +707,11 @@ "default": 5000 }, "request_timeout_ms": { - "description": "Bound on a query, and on an insert waiting to hand over its next chunk,\nin milliseconds.", + "description": "Starting bound on a query, which doubles after each timed-out query up to\n300 s, and bound on an insert waiting to hand over its next chunk, in\nmilliseconds.", "type": "integer", "format": "uint64", "minimum": 0, - "default": 30000 + "default": 60000 }, "insert_timeout_ms": { "description": "Bound on an insert waiting for the server's answer once its rows are\nsent, in milliseconds. A timed-out batch is held and retried.", diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index 785da94..470fa73 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -79,7 +79,7 @@ properties: tables: [] tls: null connect_timeout_ms: 5000 - request_timeout_ms: 30000 + request_timeout_ms: 60000 insert_timeout_ms: 60000 payload: description: Payload pipeline mode. **Restart required.** @@ -628,12 +628,13 @@ $defs: default: 5000 request_timeout_ms: description: |- - Bound on a query, and on an insert waiting to hand over its next chunk, - in milliseconds. + Starting bound on a query, which doubles after each timed-out query up to + 300 s, and bound on an insert waiting to hand over its next chunk, in + milliseconds. type: integer format: uint64 minimum: 0 - default: 30000 + default: 60000 insert_timeout_ms: description: |- Bound on an insert waiting for the server's answer once its rows are diff --git a/src/clickhouse/client_http.rs b/src/clickhouse/client_http.rs index 9fdf57e..b2f65f1 100644 --- a/src/clickhouse/client_http.rs +++ b/src/clickhouse/client_http.rs @@ -24,6 +24,7 @@ use std::future::Future; use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; use std::time::Duration; use hyper_rustls::HttpsConnectorBuilder; @@ -47,6 +48,10 @@ const TCP_KEEPALIVE: Duration = Duration::from_secs(60); /// connection before the server times it out and logs an error for it. const POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(2); +/// Widened reads stop at the 300 s `ClickHouse` gives its own socket reads and +/// writes by default. +const READ_BOUND_CEILING: Duration = Duration::from_secs(300); + /// `ClickHouse` client for DDL, schema queries, and health checks. /// /// Wraps a single `clickhouse::Client` (HTTP or TCP per config). Data inserts @@ -57,8 +62,10 @@ pub struct ClickHouseQueryClient { ch_client: clickhouse::Client, /// Database name. database: String, - /// Bounds on each query and insert. + /// Bounds on each insert. timeouts: Timeouts, + /// Bound on each query, shared with the inserter's schema reads. + read_bound: Arc, } /// Run `request`, failing it with `TimedOut` once `bound` passes. @@ -71,6 +78,67 @@ pub(crate) async fn bounded( .unwrap_or(Err(clickhouse::error::Error::TimedOut)) } +/// The bound on `ClickHouse` reads, which widens while reads keep timing out. +/// +/// A fixed bound below how long a slow but working server takes would fail +/// every attempt at a schema read, and nothing waiting on that schema would +/// ever land. So a read that times out doubles the bound for the next one, up +/// to 300 s, and a read answered in under half the next narrower bound steps +/// it back down. +#[derive(Debug)] +pub struct ReadBound { + base: Duration, + ceiling: Duration, + level: AtomicU32, +} + +impl ReadBound { + /// A bound that starts at `base` and widens no further than the ceiling, or + /// than `base` itself if that is larger. + #[must_use] + pub fn new(base: Duration) -> Self { + Self { + base, + ceiling: READ_BOUND_CEILING.max(base), + level: AtomicU32::new(0), + } + } + + /// The bound the next read runs under. + #[must_use] + pub fn current(&self) -> Duration { + self.at(self.level.load(Ordering::Relaxed)) + } + + fn at(&self, level: u32) -> Duration { + self.base + .checked_mul(1_u32.checked_shl(level).unwrap_or(u32::MAX)) + .unwrap_or(self.ceiling) + .min(self.ceiling) + } + + /// Run `read` under the current bound, widening it after a timeout and + /// narrowing it after a quick answer. + pub async fn run( + &self, + read: impl Future>, + ) -> clickhouse::error::Result { + let level = self.level.load(Ordering::Relaxed); + let started = tokio::time::Instant::now(); + let result = bounded(self.at(level), read).await; + if matches!(result, Err(clickhouse::error::Error::TimedOut)) { + if self.at(level) < self.ceiling { + self.level.fetch_max(level + 1, Ordering::Relaxed); + } + } else if level > 0 && started.elapsed() <= self.at(level - 1) / 2 { + let _ = + self.level + .compare_exchange(level, level - 1, Ordering::Relaxed, Ordering::Relaxed); + } + result + } +} + /// Open an insert sink for `sql` on `client`, bounded by `timeouts`. pub(crate) fn insert_sink( client: &clickhouse::Client, @@ -226,20 +294,29 @@ impl ClickHouseQueryClient { /// Returns an error if the config has no hosts. pub fn new(config: &ClickHouseConfig) -> Result { let ch_client = build_client(config)?; + let timeouts = config.timeouts(); Ok(Self { ch_client, database: config.database.clone(), - timeouts: config.timeouts(), + timeouts, + read_bound: Arc::new(ReadBound::new(timeouts.request)), }) } + /// The bound on queries, for the inserter's schema reads to share. + #[must_use] + pub fn read_bound(&self) -> Arc { + Arc::clone(&self.read_bound) + } + /// Execute a DDL or DML statement (CREATE, DROP, ALTER, TRUNCATE, etc.). /// /// # Errors /// /// Returns an error if the query fails. pub async fn execute(&self, sql: &str) -> Result<()> { - bounded(self.timeouts.request, self.ch_client.query(sql).execute()) + self.read_bound + .run(self.ch_client.query(sql).execute()) .await .map_err(|e| ClickHouseError::Query(format!("{e}")))?; Ok(()) @@ -266,12 +343,13 @@ impl ClickHouseQueryClient { escape_string(&tbl) ); - let rows: Vec = bounded( - self.timeouts.request, - self.ch_client.query(&sql).fetch_all(), - ) - .await - .map_err(|e| ClickHouseError::Schema(format!("Failed to fetch schema for {table}: {e}")))?; + let rows: Vec = self + .read_bound + .run(self.ch_client.query(&sql).fetch_all()) + .await + .map_err(|e| { + ClickHouseError::Schema(format!("Failed to fetch schema for {table}: {e}")) + })?; // The query succeeded and system.columns has nothing: the table really is // absent. Distinct from Schema above, which a transient outage produces. @@ -315,12 +393,11 @@ impl ClickHouseQueryClient { escape_string(&tbl) ); - let rows: Vec = bounded( - self.timeouts.request, - self.ch_client.query(&sql).fetch_all(), - ) - .await - .map_err(|e| ClickHouseError::Schema(format!("Failed to fetch table comment: {e}")))?; + let rows: Vec = self + .read_bound + .run(self.ch_client.query(&sql).fetch_all()) + .await + .map_err(|e| ClickHouseError::Schema(format!("Failed to fetch table comment: {e}")))?; Ok(rows.into_iter().next().map(|r| r.value).unwrap_or_default()) } @@ -346,12 +423,13 @@ impl ClickHouseQueryClient { comment: String, } - let rows: Vec = bounded( - self.timeouts.request, - self.ch_client.query(&sql).fetch_all(), - ) - .await - .map_err(|e| ClickHouseError::Schema(format!("Failed to fetch column comments: {e}")))?; + let rows: Vec = self + .read_bound + .run(self.ch_client.query(&sql).fetch_all()) + .await + .map_err(|e| { + ClickHouseError::Schema(format!("Failed to fetch column comments: {e}")) + })?; let mut comments = FxHashMap::default(); for row in rows { @@ -380,12 +458,11 @@ impl ClickHouseQueryClient { escape_string(&self.database) ); - let rows: Vec = bounded( - self.timeouts.request, - self.ch_client.query(&sql).fetch_all(), - ) - .await - .map_err(|e| ClickHouseError::Schema(format!("Failed to list tables: {e}")))?; + let rows: Vec = self + .read_bound + .run(self.ch_client.query(&sql).fetch_all()) + .await + .map_err(|e| ClickHouseError::Schema(format!("Failed to list tables: {e}")))?; Ok(rows.into_iter().map(|r| r.name).collect()) } @@ -417,24 +494,21 @@ impl ClickHouseQueryClient { None => format!("SELECT COUNT(*) AS _dfe_count FROM {fq_table}"), }; - let row: CountRow = bounded( - self.timeouts.request, - self.ch_client.query(&sql).fetch_one(), - ) - .await - .map_err(|e| ClickHouseError::Query(format!("{e}")))?; + let row: CountRow = self + .read_bound + .run(self.ch_client.query(&sql).fetch_one()) + .await + .map_err(|e| ClickHouseError::Query(format!("{e}")))?; Ok(row.count as usize) } /// Health check — executes `SELECT 1`. pub async fn health_check(&self) -> Result<()> { - bounded( - self.timeouts.request, - self.ch_client.query("SELECT 1").execute(), - ) - .await - .map_err(|e| ClickHouseError::Connection(format!("Health check failed: {e}")))?; + self.read_bound + .run(self.ch_client.query("SELECT 1").execute()) + .await + .map_err(|e| ClickHouseError::Connection(format!("Health check failed: {e}")))?; Ok(()) } @@ -1058,6 +1132,65 @@ mod tests { (listener, queued, addr) } + /// A read the server answers after `delay`. + async fn read_answered_after(delay: Duration) -> clickhouse::error::Result<()> { + tokio::time::sleep(delay).await; + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn a_read_slower_than_the_bound_lands_on_the_next_attempt() { + let bound = ReadBound::new(Duration::from_secs(2)); + + let first = bound.run(read_answered_after(Duration::from_secs(3))).await; + let second = bound.run(read_answered_after(Duration::from_secs(3))).await; + + assert!( + matches!(first, Err(clickhouse::error::Error::TimedOut)), + "a 3 s read under a 2 s bound answered: {first:?}" + ); + assert!(second.is_ok(), "the widened bound timed the read out again"); + assert_eq!( + bound.current(), + Duration::from_secs(4), + "a read that still needs 3 s narrowed the bound it only just fits" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_read_that_never_answers_widens_the_bound_no_further_than_the_ceiling() { + let bound = ReadBound::new(Duration::from_secs(60)); + let mut seen = Vec::new(); + for _ in 0..5 { + let _ = bound + .run(std::future::pending::>()) + .await; + seen.push(bound.current().as_secs()); + } + assert_eq!(seen, vec![120, 240, 300, 300, 300]); + } + + #[tokio::test(start_paused = true)] + async fn a_quick_answer_narrows_a_widened_bound() { + let bound = ReadBound::new(Duration::from_secs(2)); + let _ = bound.run(read_answered_after(Duration::from_secs(3))).await; + assert_eq!(bound.current(), Duration::from_secs(4)); + + let quick = bound + .run(read_answered_after(Duration::from_millis(100))) + .await; + + assert!(quick.is_ok()); + assert_eq!(bound.current(), Duration::from_secs(2)); + } + + #[test] + fn a_base_above_the_ceiling_is_its_own_ceiling() { + let bound = ReadBound::new(Duration::from_secs(400)); + assert_eq!(bound.current(), Duration::from_secs(400)); + assert_eq!(bound.at(5), Duration::from_secs(400)); + } + #[tokio::test] async fn an_idle_connection_closes_about_two_seconds_after_its_last_request() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/src/clickhouse/config.rs b/src/clickhouse/config.rs index 6e43a7d..d98c15a 100644 --- a/src/clickhouse/config.rs +++ b/src/clickhouse/config.rs @@ -164,8 +164,9 @@ pub struct ClickHouseConfig { #[serde(default = "default_connect_timeout")] pub connect_timeout_ms: u64, - /// Bound on a query, and on an insert waiting to hand over its next chunk, - /// in milliseconds. + /// Starting bound on a query, which doubles after each timed-out query up to + /// 300 s, and bound on an insert waiting to hand over its next chunk, in + /// milliseconds. #[serde(default = "default_request_timeout")] pub request_timeout_ms: u64, @@ -203,8 +204,10 @@ pub(crate) const fn default_connect_timeout() -> u64 { 5000 } +/// A schema read on a loaded server can take tens of seconds, and every timed-out +/// first try costs that wait before the widened one starts. pub(crate) const fn default_request_timeout() -> u64 { - 30000 + 60000 } /// 60 s is ~270x the ClickHouse CPU cost of a 20,000-row flush at a measured @@ -218,7 +221,8 @@ pub(crate) const fn default_insert_timeout() -> u64 { /// Client-side bounds on `ClickHouse` requests. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Timeouts { - /// Bound on a query, and on an insert waiting to hand over its next chunk. + /// Starting bound on a query, and bound on an insert waiting to hand over + /// its next chunk. pub request: std::time::Duration, /// Bound on an insert waiting for the server's answer once its rows are sent. pub insert: std::time::Duration, @@ -426,7 +430,7 @@ mod tests { assert_eq!(config.password, ""); assert!(!config.tls); assert_eq!(config.connect_timeout_ms, 5000); - assert_eq!(config.request_timeout_ms, 30000); + assert_eq!(config.request_timeout_ms, 60000); assert_eq!(config.insert_timeout_ms, 60000); assert_eq!(config.connect_timeout(), std::time::Duration::from_secs(5)); assert_eq!(config.timeouts(), Timeouts::default()); diff --git a/src/clickhouse/inserter.rs b/src/clickhouse/inserter.rs index 8eefb9d..ead753d 100644 --- a/src/clickhouse/inserter.rs +++ b/src/clickhouse/inserter.rs @@ -44,7 +44,7 @@ use tracing::{debug, error, info, trace, warn}; use crate::Result; use crate::buffer::{FlushBatch, KafkaOffset}; -use crate::clickhouse::client_http::{escape_identifier, insert_sink}; +use crate::clickhouse::client_http::{ReadBound, escape_identifier, insert_sink}; use crate::clickhouse::config::{InsertFormat, Timeouts}; use crate::clickhouse::error::{ ErrorCategory, classify_dynamic_error, classify_json_insert_error, is_schema_drift_error, @@ -293,8 +293,10 @@ pub struct Inserter { /// Whether a schema-drift insert error drops the cached schema /// (`schema.refresh_on_error`). refresh_on_error: bool, - /// Bounds on schema fetches and inserts. + /// Bounds on inserts. timeouts: Timeouts, + /// Bound on schema reads, shared with `http_client`. + read_bound: Arc, } impl Inserter { @@ -315,6 +317,7 @@ impl Inserter { }; Self { + read_bound: http_client.read_bound(), http_client, ch_client, dynamic_schema_cache: crate::clickhouse_ext::DynamicSchemaCache::new( @@ -334,14 +337,15 @@ impl Inserter { } } - /// Set the bounds on schema fetches and inserts (`clickhouse.request_timeout_ms` - /// and `clickhouse.insert_timeout_ms`). + /// Set the bounds on inserts (`clickhouse.request_timeout_ms` and + /// `clickhouse.insert_timeout_ms`). Schema reads run under the query + /// client's widening bound. pub fn with_timeouts(mut self, timeouts: Timeouts) -> Self { self.timeouts = timeouts; self } - /// The bounds on schema fetches and inserts. + /// The bounds on inserts. #[cfg(test)] pub(crate) fn timeouts(&self) -> Timeouts { self.timeouts @@ -472,7 +476,7 @@ impl Inserter { &self.ch_client, db, tbl, - self.timeouts.request, + &self.read_bound, ) .await { @@ -507,7 +511,7 @@ impl Inserter { let Some(encoded_against) = encoded_against else { return false; }; - crate::clickhouse_ext::fetch_dynamic_schema(&self.ch_client, db, tbl, self.timeouts.request) + crate::clickhouse_ext::fetch_dynamic_schema(&self.ch_client, db, tbl, &self.read_bound) .await .is_ok_and(|now| now.same_columns(encoded_against)) } @@ -586,7 +590,8 @@ impl Inserter { Arc::clone(&self.dynamic_schema_cache), ) .refresh_on_error(self.refresh_on_error) - .with_timeouts(self.timeouts); + .with_timeouts(self.timeouts) + .with_read_bound(Arc::clone(&self.read_bound)); let mut write_failed = false; for (i, row) in rows.iter().enumerate() { diff --git a/src/clickhouse_ext/insert.rs b/src/clickhouse_ext/insert.rs index 02ba8b5..eebf285 100644 --- a/src/clickhouse_ext/insert.rs +++ b/src/clickhouse_ext/insert.rs @@ -39,7 +39,7 @@ use super::encode::{ColumnDef, DynamicRow}; use super::error::DynamicError; use super::schema::{DynamicSchema, DynamicSchemaCache, fetch_dynamic_schema}; use crate::clickhouse::Timeouts; -use crate::clickhouse::client_http::insert_sink; +use crate::clickhouse::client_http::{ReadBound, insert_sink}; /// The clickhouse-rs setting that makes the server read a length-prefixed /// string as a JSON value, which is how the dynamic encoder writes JSON @@ -108,8 +108,10 @@ pub struct DynamicInsert { rows_written: u64, /// Whether a schema-mismatch error at `end` drops the cached schema. refresh_on_error: bool, - /// Bounds on the schema fetch and the insert. + /// Bounds on the insert. timeouts: Timeouts, + /// Bound on the schema fetch. + read_bound: Arc, } impl DynamicInsert { @@ -131,16 +133,23 @@ impl DynamicInsert { rows_written: 0, refresh_on_error: true, timeouts: Timeouts::default(), + read_bound: Arc::new(ReadBound::new(Timeouts::default().request)), } } - /// Set the bounds on the schema fetch and the insert (default - /// [`Timeouts::default`]). + /// Set the bounds on the insert (default [`Timeouts::default`]). pub fn with_timeouts(mut self, timeouts: Timeouts) -> Self { self.timeouts = timeouts; self } + /// Share the bound the schema fetch runs under, so a timeout here widens + /// the next read anywhere it is shared. + pub fn with_read_bound(mut self, read_bound: Arc) -> Self { + self.read_bound = read_bound; + self + } + /// Set whether a schema-mismatch error at [`end`][Self::end] drops the /// cached schema (default `true`). With `false` the stale schema stays /// until the cache's TTL expires. @@ -162,13 +171,9 @@ impl DynamicInsert { let schema = if let Some(cached) = self.schema_cache.get(&full) { cached } else { - let fetched = fetch_dynamic_schema( - &self.client, - &self.database, - &self.table, - self.timeouts.request, - ) - .await?; + let fetched = + fetch_dynamic_schema(&self.client, &self.database, &self.table, &self.read_bound) + .await?; self.schema_cache.insert(&full, fetched.clone()); fetched }; diff --git a/src/clickhouse_ext/schema.rs b/src/clickhouse_ext/schema.rs index 8f3efad..c6a46df 100644 --- a/src/clickhouse_ext/schema.rs +++ b/src/clickhouse_ext/schema.rs @@ -23,6 +23,7 @@ use clickhouse::Client; use super::encode::ColumnDef; use super::error::DynamicError; +use crate::clickhouse::client_http::ReadBound; /// Resolved schema for a single table -- an ordered list of columns plus a /// name index for O(1) lookup during encoding. @@ -122,14 +123,15 @@ struct SysColumn { /// /// # Errors /// -/// Returns [`DynamicError::SchemaFetch`] if the query fails or runs past -/// `timeout`, or [`DynamicError::EmptySchema`] if the table has no columns (or -/// does not exist). +/// Returns [`DynamicError::SchemaFetch`] if the query fails, +/// [`DynamicError::TimedOut`] if it runs past `bound`, or +/// [`DynamicError::EmptySchema`] if the table has no columns (or does not +/// exist). pub async fn fetch_dynamic_schema( client: &Client, database: &str, table: &str, - timeout: Duration, + bound: &ReadBound, ) -> Result { let full_table = format!("{database}.{table}"); @@ -143,17 +145,15 @@ pub async fn fetch_dynamic_schema( .bind(database) .bind(table) .fetch_all::(); - let rows = crate::clickhouse::client_http::bounded(timeout, query) - .await - .map_err(|source| match source { - clickhouse::error::Error::TimedOut => DynamicError::TimedOut { - table: full_table.clone(), - }, - source => DynamicError::SchemaFetch { - table: full_table.clone(), - source, - }, - })?; + let rows = bound.run(query).await.map_err(|source| match source { + clickhouse::error::Error::TimedOut => DynamicError::TimedOut { + table: full_table.clone(), + }, + source => DynamicError::SchemaFetch { + table: full_table.clone(), + source, + }, + })?; if rows.is_empty() { return Err(DynamicError::EmptySchema { table: full_table }); diff --git a/src/config/loader.rs b/src/config/loader.rs index 5df2ec4..628100a 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -180,8 +180,9 @@ pub struct ClickHouseConfig { pub tls: Option, /// Bound on opening a TCP connection to `ClickHouse`, in milliseconds. pub connect_timeout_ms: u64, - /// Bound on a query, and on an insert waiting to hand over its next chunk, - /// in milliseconds. + /// Starting bound on a query, which doubles after each timed-out query up to + /// 300 s, and bound on an insert waiting to hand over its next chunk, in + /// milliseconds. pub request_timeout_ms: u64, /// Bound on an insert waiting for the server's answer once its rows are /// sent, in milliseconds. A timed-out batch is held and retried. @@ -1635,11 +1636,11 @@ kafka: } #[test] - fn the_default_clickhouse_timeouts_are_5_s_connect_30_s_request_60_s_insert() { + fn the_default_clickhouse_timeouts_are_5_s_connect_60_s_request_60_s_insert() { let client_cfg: crate::clickhouse::ClickHouseConfig = (&ClickHouseConfig::default()).into(); let secs = std::time::Duration::from_secs; assert_eq!(client_cfg.connect_timeout(), secs(5)); - assert_eq!(client_cfg.timeouts().request, secs(30)); + assert_eq!(client_cfg.timeouts().request, secs(60)); assert_eq!(client_cfg.timeouts().insert, secs(60)); } diff --git a/tests/integration/grpc_outage.rs b/tests/integration/grpc_outage.rs index 979772a..c0e00f2 100644 --- a/tests/integration/grpc_outage.rs +++ b/tests/integration/grpc_outage.rs @@ -1324,3 +1324,84 @@ async fn a_row_clickhouse_refuses_for_its_data_is_dead_lettered_and_the_rest_lan "the dead letter does not carry the server's refusal: {reason}" ); } + +/// A loopback proxy to `ClickHouse` HTTP that holds every connection opening +/// with anything but an INSERT for `delay` before passing it on, so every read +/// the loader makes on a fresh connection answers that much later. +async fn slow_read_proxy(upstream: String, delay: Duration) -> String { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = TcpListener::bind(("127.0.0.1", free_port())) + .await + .expect("bind proxy"); + let address = listener.local_addr().expect("proxy addr").to_string(); + tokio::spawn(async move { + while let Ok((mut client, _)) = listener.accept().await { + let upstream = upstream.clone(); + tokio::spawn(async move { + let mut head = vec![0_u8; 4096]; + let Ok(n) = client.read(&mut head).await else { + return; + }; + head.truncate(n); + if !String::from_utf8_lossy(&head).contains("query=INSERT") { + tokio::time::sleep(delay).await; + } + let Ok(mut server) = TcpStream::connect(&upstream).await else { + return; + }; + if server.write_all(&head).await.is_err() { + return; + } + let _ = tokio::io::copy_bidirectional(&mut client, &mut server).await; + }); + } + }); + address +} + +/// How long the slow-read proxy holds a read in the slow-schema test. +const SLOW_READ: Duration = Duration::from_secs(3); + +/// The loader's request bound in the slow-schema test: below `SLOW_READ`. +const SHORT_REQUEST_TIMEOUT_MS: u64 = 2_000; + +#[tokio::test] +async fn every_record_lands_when_schema_reads_answer_slower_than_the_request_bound() { + let infra = TestInfrastructure::new(test_name!(), true, false).await; + let clickhouse = clickhouse_address(&infra).await; + let direct = query_client(&clickhouse); + let proxy = slow_read_proxy(clickhouse.clone(), SLOW_READ).await; + + let table = unique_table_name("grpc_slow_schema"); + create_id_table(&direct, &table, "").await; + + let listen_port = free_port(); + let mut config = grpc_loader(listen_port, proxy, &table); + config.clickhouse.request_timeout_ms = SHORT_REQUEST_TIMEOUT_MS; + let (shutdown, loader, client) = start_loader(config, listen_port).await; + + let settle = tokio::time::Instant::now() + Duration::from_secs(60); + let (accepted, refused) = push_ids(&client, 0..BEFORE, settle, id_payload).await; + let landed = wait_landed(&clickhouse, &table, &accepted, LANDING_BUDGET).await; + + shutdown.cancel(); + let _ = tokio::time::timeout(Duration::from_secs(30), loader).await; + let _ = direct + .execute(&format!("DROP TABLE IF EXISTS default.{table}")) + .await; + + assert_eq!( + accepted.len() as u64, + BEFORE, + "with every schema read {SLOW_READ:?} slow against a {SHORT_REQUEST_TIMEOUT_MS} ms \ + bound the listener accepted {} of {BEFORE} records in 60 s, after {refused} retries: \ + the bound never let the schema fetch finish", + accepted.len() + ); + let missing: Vec = accepted.difference(&landed).copied().collect(); + assert!( + missing.is_empty(), + "accepted records never landed: {missing:?}" + ); +}