diff --git a/Cargo.lock b/Cargo.lock index 6078aff..c9a3868 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -177,6 +177,45 @@ dependencies = [ "num-traits", ] +[[package]] +name = "asn1-rs" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom", + "num-traits", + "rusticata-macros", + "thiserror 2.0.21", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure 0.13.2", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "astral-tokio-tar" version = "0.6.4" @@ -384,6 +423,15 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -1090,6 +1138,12 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "deadpool" version = "0.12.3" @@ -1148,6 +1202,20 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5729f5117e208430e437df2f4843f5e5952997175992d1414f94c57d61e270b4" +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.8" @@ -1198,17 +1266,21 @@ dependencies = [ "figment", "flate2", "futures", + "hyper-rustls", + "hyper-util", "indexmap 2.14.2", "maxminddb", "memchr", "metrics", "parking_lot", "proptest", + "rcgen", "rdkafka", "regex", "reqwest 0.12.28", "reqwest 0.13.5", "rustc-hash", + "rustls", "scalo", "schemars 1.2.2", "serde", @@ -2994,6 +3066,15 @@ dependencies = [ "objc2-foundation", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -3256,6 +3337,16 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "pem" +version = "4.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d354a98a3d1251555de99e8fdd8afda05573c31b82f59063a7b0a29b5527f120" +dependencies = [ + "base64 0.23.1", + "serde_core", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -3979,6 +4070,20 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rcgen" +version = "0.14.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8774e05a7d0de114588e6a28fe7e71694b82614ed569d86d8b389dfbc98b8ad8" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", +] + [[package]] name = "rdkafka" version = "0.39.0" @@ -4239,6 +4344,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom", +] + [[package]] name = "rustix" version = "1.1.5" @@ -4967,6 +5081,17 @@ dependencies = [ "futures-core", ] +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "synstructure" version = "0.14.0" @@ -6360,6 +6485,24 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" +[[package]] +name = "x509-parser" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom", + "oid-registry", + "ring", + "rusticata-macros", + "thiserror 2.0.21", + "time", +] + [[package]] name = "xattr" version = "1.6.1" @@ -6376,6 +6519,16 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" +[[package]] +name = "yasna" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282" +dependencies = [ + "bit-vec 0.9.1", + "time", +] + [[package]] name = "yoke" version = "0.8.3" @@ -6396,7 +6549,7 @@ dependencies = [ "proc-macro2", "quote", "syn 3.0.6", - "synstructure", + "synstructure 0.14.0", ] [[package]] @@ -6437,7 +6590,7 @@ dependencies = [ "proc-macro2", "quote", "syn 3.0.6", - "synstructure", + "synstructure 0.14.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 42a4e98..fda449e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -85,12 +85,17 @@ flate2 = ">=1.1.10, <2" tar = ">=0.4.46, <0.5" # ClickHouse crate (HyperI fork with native TCP + unified client) -# rustls-tls: HTTPS for HTTP transport +# rustls-tls + rustls-tls-native-roots: HTTPS for clients built with Client::default() # native-transport + native-tls-rustls: native TCP with TLS -# rustls-tls-native-roots: uses system trust store for internal CAs clickhouse = { version = ">=0.15.0, <0.16", features = ["lz4", "time", "rustls-tls", "rustls-tls-native-roots", "tcp", "native-tls-rustls", "inserter"] } time = ">=0.3.55, <0.4" +# The HTTP client handed to clickhouse::Client::with_http_client: connect timeout, +# idle-pool eviction and HTTPS verified against the system trust store. +hyper-util = { version = ">=0.1.21, <0.2", features = ["client-legacy", "http1", "tokio"] } +hyper-rustls = { version = ">=0.27.10, <0.28", default-features = false, features = ["http1", "tls12", "native-tokio", "aws-lc-rs"] } +rustls = { version = ">=0.23.45, <0.24", default-features = false, features = ["std", "tls12", "aws_lc_rs"] } + # Utilities uuid = { version = ">=1.27.0, <2", features = ["v4", "v7"] } rustc-hash = ">=2.1.3, <3" # FxHashMap - faster hashing for internal maps @@ -136,6 +141,9 @@ rdkafka = { version = ">=0.39.0, <0.40", features = ["dynamic-linking"] } clickhouse = { version = ">=0.15.0, <0.16", features = ["lz4", "time", "rustls-tls", "rustls-tls-native-roots", "tcp", "native-tls-rustls", "inserter"] } time = ">=0.3.55, <0.4" +# Private CA and server certificates for the TLS ClickHouse tests +rcgen = ">=0.14.10, <0.15" + # Async test utilities tokio-test = ">=0.4.6, <0.5" tempfile = ">=3.27.0, <4" diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 5f5d610..4538946 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -46,7 +46,7 @@ file drives the loader's own weights and is reported as a mismatch at startup. | Setting | Values | Effect | |---------|--------|--------| -| `clickhouse.protocol` | `http` (default), `native` | HTTP (8123/8543). `native` is refused at startup: the pinned client has no TCP row fetch, and the native insert sink waits on clickhouse-rs#15 | +| `clickhouse.protocol` | `http` (default), `native` | HTTP (8123), HTTPS (8443). `native` is refused at startup: the pinned client has no TCP row fetch, and the native insert sink waits on clickhouse-rs#15 | | `clickhouse.insert_format` | `row_binary` (default), `json_each_row` | binary (server skips JSON parsing) vs self-describing JSON | | `clickhouse.hosts` | list | multi-host failover (native pool round-robins, skips a refusing endpoint) | | `clickhouse.database` | string, `dfe` (default) | database the connection targets -- DDL, schema reflection, and any table name that arrives without a `db.` prefix | @@ -69,19 +69,24 @@ Both formats insert over HTTP -- see ### Transport security (TLS, private CAs) -One trust mechanism covers both transports. By default a TLS connection trusts -the OS native root store plus the compiled Mozilla (webpki) bundle. Internal -clusters behind a private CA are supported by pointing at the CA PEM -- no proxy -termination, no `InsecureSkipVerify`. +HTTPS verifies the server's certificate and host name against the container's system trust store. A private CA is trusted by adding it to that store, or by pointing `SSL_CERT_FILE` (or `SSL_CERT_DIR`) at a bundle that holds it -- there is no CA path in the loader config and no way to turn verification off. | Setting | Values | Effect | |---------|--------|--------| -| `clickhouse.tls` | bool | TLS for the connection (HTTPS 8543 / secure native 9440) | -| `clickhouse.tls_ca_file` | path | private CA PEM (may bundle many certs; all are loaded). Augments native + webpki | -| `clickhouse.tls_ca_exclusive` | bool | trust ONLY `tls_ca_file`, ignore native + webpki | +| `clickhouse.tls.enabled` | bool | HTTPS for the connection. When `clickhouse.tls` is absent, a host on 8443/9440 or under `clickhouse.cloud` turns it on | +| `clickhouse.tls.ca_cert_file`, `cert_file`, `key_file`, `skip_verify` | -- | refused at startup: no client reads them | -A CA file with no usable certificate is a hard error (it never silently falls -back to public roots). See [clickhouse/TLS.md](clickhouse/TLS.md). +See [clickhouse/TLS.md](clickhouse/TLS.md). + +### ClickHouse timeouts + +| 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.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. ### Capture (what gets stored) @@ -119,9 +124,9 @@ Config hot-reloads on change. Some fields are safe to apply live; some require a restart and are logged as a warning rather than silently applied: - **Restart-required:** `clickhouse.protocol`, ClickHouse URL/hosts, - `clickhouse.insert_format`, TLS settings (`tls`, `tls_ca_file`, - `tls_ca_exclusive`), Kafka connection -- all of these are baked into the - `Client` at build time. + `clickhouse.insert_format`, `clickhouse.tls.enabled`, the + `clickhouse.*_timeout_ms` bounds, Kafka connection -- all of these are baked + into the `Client` at build time. - **Safe to reload:** buffer thresholds, capture modes, enrichment toggles, routing rules. diff --git a/docs/clickhouse/CLICKHOUSE-EXT.md b/docs/clickhouse/CLICKHOUSE-EXT.md index 0566bea..2d0020c 100644 --- a/docs/clickhouse/CLICKHOUSE-EXT.md +++ b/docs/clickhouse/CLICKHOUSE-EXT.md @@ -84,6 +84,8 @@ fork repo is now archived and read-only, so the branch can no longer move and the pinned rev stays reachable. It also means no fix can land on the fork: retiring the patch for crates.io `clickhouse` is the only way forward (#133). +The HTTP client is the loader's own. `build_client` in `src/clickhouse/client_http.rs` hands the fork a hyper client through `Client::with_http_client`, carrying the pool timer that closes idle connections after 2 s, the connect timeout and HTTPS verified against the system trust store. Queries and inserts carry the request and insert timeouts on top. #133 still stands for the rest of the fork. + ## What stays out of clickhouse_ext - The transport, pool, handshake, and Native block framing -- those are the diff --git a/docs/clickhouse/TLS.md b/docs/clickhouse/TLS.md index dd8bb76..874ba3b 100644 --- a/docs/clickhouse/TLS.md +++ b/docs/clickhouse/TLS.md @@ -10,91 +10,52 @@ # TLS and private CAs -ClickHouse connections can run over TLS on either transport: HTTPS (8543) for -`transport = http`, secure native (9440) for `transport = native`. One trust -mechanism serves both -- the loader does not carry two TLS code paths, and a -private-CA cluster is trusted the same way regardless of transport. +The loader reaches ClickHouse over HTTP (`clickhouse.protocol: http`; `native` is refused at startup). With TLS on it speaks HTTPS, usually on 8443, and verifies the server's certificate and host name against the container's system trust store. ```mermaid -flowchart TB - CFG["clickhouse.tls / tls_ca_file / tls_ca_exclusive"] - TRUST["Trust anchor set"] - NAT["OS native roots
(rustls-native-certs)"] - WEB["Mozilla bundle
(webpki-roots)"] - CA["Private CA PEM
(tls_ca_file)"] - HTTP["HTTPS 8543"] - NATIVE["secure native 9440"] - - CFG --> TRUST - NAT --> TRUST - WEB --> TRUST - CA --> TRUST - TRUST --> HTTP - TRUST --> NATIVE +flowchart LR + CFG["clickhouse.tls.enabled"] + STORE["System trust store
(image bundle, or SSL_CERT_FILE / SSL_CERT_DIR)"] + VERIFY["Certificate chain + host name check"] + HTTPS["HTTPS to ClickHouse"] + + CFG --> HTTPS + STORE --> VERIFY + VERIFY --> HTTPS ``` -## Default trust +## Turning TLS on -With `tls = true` and no CA file, a TLS connection trusts the OS native root -store plus the compiled-in Mozilla (webpki) bundle. Public clusters and any -cluster whose CA is already installed in the host trust store work with no extra -configuration. +| Setting | Effect | +|---------|--------| +| `clickhouse.tls.enabled` | HTTPS for the connection | +| `clickhouse.tls` absent | TLS turns on for a host on 8443 or 9440, or under `clickhouse.cloud` | + +The SNI name and the name checked against the certificate are the host in `clickhouse.hosts`, so connect by a name (or IP) the certificate lists. ## Private / internal CAs -Internal clusters behind a private CA are trusted by pointing the loader at the -CA certificate: +A private CA is trusted the way the system trusts it: add it to the image's trust store, or point `SSL_CERT_FILE` at a PEM bundle that holds it (`SSL_CERT_DIR` at a directory of them). Either variable replaces the platform store rather than adding to it, so a bundle that must also trust public CAs has to carry them too. -| Setting | Effect | -|---------|--------| -| `clickhouse.tls` | enable TLS for the connection | -| `clickhouse.tls_ca_file` | path to a private CA PEM; augments native + webpki | -| `clickhouse.tls_ca_exclusive` | trust ONLY `tls_ca_file`; ignore native + webpki | - -The CA file is read with `AppendCertsFromPEM` semantics (the clickhouse-go -model): every PEM certificate block in the file is parsed best-effort and all -valid certificates land in one trust store. A bundle of concatenated certs -(`cat root.pem intermediate.pem > ca.pem`) is loaded in full -- roots and -intermediates both become anchors, which covers servers that do not present -their full chain and multi-tier PKI. - -A CA file that yields zero usable certificates, or an unreadable path, is a hard -error. The loader never silently falls back to public roots when a CA file was -asked for -- a misconfigured trust store fails the connection loudly rather than -trusting the wrong anchors. - -## Augment vs exclusive - -- **Augment (default when `tls_ca_file` is set):** native + webpki + the private - CA. The cluster's internal CA is trusted alongside public roots. Use this - unless you have a reason to lock trust down. -- **Exclusive (`tls_ca_exclusive = true`):** only the private CA. Native and - webpki roots are ignored. Use this to pin trust to your internal PKI and - refuse anything chaining to a public CA. +A trust store with no readable CA certificate fails the client at startup instead of connecting unverified. ## What is deliberately not offered -There is no insecure / skip-verify mode. Trusting an internal cluster is done by -naming its CA, not by turning verification off. If you find yourself wanting to -disable verification, supply the CA PEM instead -- that is the supported and -auditable path. - -Client-certificate (mTLS) auth is out of scope today. +`clickhouse.tls.ca_cert_file`, `cert_file`, `key_file` and `skip_verify` are refused at startup: no client reads them. There is no skip-verify mode, and client-certificate (mTLS) auth is out of scope today. ## Example: private-CA cluster ```yaml clickhouse: - transport: native # secure native protocol - tls: true # 9440 - tls_ca_file: /etc/dfe/internal-ca.pem + protocol: http + tls: + enabled: true hosts: - - clickhouse.example.internal:9440 + - clickhouse.example.internal:8443 +``` + +```text +SSL_CERT_FILE=/etc/dfe/ca-bundle.pem # the private CA plus any public roots still needed ``` -The same `tls_ca_file` works unchanged if you switch `transport` to `http` -(8543) -- one trust config, both transports. The CA PEM is non-secret and is -distributed with the deployment; the cluster password stays in the -git-excluded environment, never in committed config. See -[../CONFIGURATION.md](../CONFIGURATION.md) for the cascade and -[../deployment/](../deployment/) for how secrets are supplied at runtime. +The CA bundle is non-secret and is distributed with the deployment; the cluster password stays in the git-excluded environment, never in committed config. See [../CONFIGURATION.md](../CONFIGURATION.md) for the cascade and [../deployment/](../deployment/) for how secrets are supplied at runtime. diff --git a/docs/config-schema.json b/docs/config-schema.json index 1e81499..4343b69 100644 --- a/docs/config-schema.json +++ b/docs/config-schema.json @@ -60,7 +60,10 @@ "protocol": "http", "insert_format": "rowbinary", "tables": [], - "tls": null + "tls": null, + "connect_timeout_ms": 5000, + "request_timeout_ms": 30000, + "insert_timeout_ms": 60000 } }, "payload": { @@ -695,6 +698,27 @@ } ], "default": null + }, + "connect_timeout_ms": { + "description": "Bound on opening a TCP connection to `ClickHouse`, in milliseconds.", + "type": "integer", + "format": "uint64", + "minimum": 0, + "default": 5000 + }, + "request_timeout_ms": { + "description": "Bound on a query, and on an insert waiting to hand over its next chunk,\nin milliseconds.", + "type": "integer", + "format": "uint64", + "minimum": 0, + "default": 30000 + }, + "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.", + "type": "integer", + "format": "uint64", + "minimum": 0, + "default": 60000 } } }, diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index 3a4c019..785da94 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -78,6 +78,9 @@ properties: insert_format: rowbinary tables: [] tls: null + connect_timeout_ms: 5000 + request_timeout_ms: 30000 + insert_timeout_ms: 60000 payload: description: Payload pipeline mode. **Restart required.** $ref: '#/$defs/PayloadConfig' @@ -617,6 +620,28 @@ $defs: - $ref: '#/$defs/TlsConfig' - type: 'null' default: null + connect_timeout_ms: + description: Bound on opening a TCP connection to `ClickHouse`, in milliseconds. + type: integer + format: uint64 + minimum: 0 + default: 5000 + request_timeout_ms: + description: |- + Bound on a query, and on an insert waiting to hand over its next chunk, + in milliseconds. + type: integer + format: uint64 + minimum: 0 + default: 30000 + insert_timeout_ms: + description: |- + 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. + type: integer + format: uint64 + minimum: 0 + default: 60000 InsertFormat: description: |- Insert format for data writes. diff --git a/src/clickhouse/client_http.rs b/src/clickhouse/client_http.rs index 20f34ae..9fdf57e 100644 --- a/src/clickhouse/client_http.rs +++ b/src/clickhouse/client_http.rs @@ -22,18 +22,31 @@ //! - Table existence checks //! - Health checks (ping) +use std::future::Future; use std::sync::Arc; +use std::time::Duration; +use hyper_rustls::HttpsConnectorBuilder; +use hyper_util::client::legacy::Client as HyperClient; +use hyper_util::client::legacy::connect::HttpConnector; +use hyper_util::rt::{TokioExecutor, TokioTimer}; use rustc_hash::FxHashMap; use serde::Deserialize; -use super::config::ClickHouseConfig; +use super::config::{ClickHouseConfig, Timeouts}; use super::error::ClickHouseError; use super::types::{ColumnInfo, ParsedType, TableSchema}; /// Result type for `ClickHouse` operations. pub type Result = std::result::Result; +/// TCP keepalive interval on `ClickHouse` connections. +const TCP_KEEPALIVE: Duration = Duration::from_secs(60); + +/// Shorter than `ClickHouse`'s keep-alive timeout, so the client closes an idle +/// connection before the server times it out and logs an error for it. +const POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(2); + /// `ClickHouse` client for DDL, schema queries, and health checks. /// /// Wraps a single `clickhouse::Client` (HTTP or TCP per config). Data inserts @@ -44,6 +57,58 @@ pub struct ClickHouseQueryClient { ch_client: clickhouse::Client, /// Database name. database: String, + /// Bounds on each query and insert. + timeouts: Timeouts, +} + +/// Run `request`, failing it with `TimedOut` once `bound` passes. +pub(crate) async fn bounded( + bound: Duration, + request: impl Future>, +) -> clickhouse::error::Result { + tokio::time::timeout(bound, request) + .await + .unwrap_or(Err(clickhouse::error::Error::TimedOut)) +} + +/// Open an insert sink for `sql` on `client`, bounded by `timeouts`. +pub(crate) fn insert_sink( + client: &clickhouse::Client, + sql: String, + timeouts: Timeouts, +) -> clickhouse::insert_formatted::InsertFormatted { + client + .insert_formatted_with(sql) + .with_timeouts(Some(timeouts.request), Some(timeouts.insert)) +} + +/// The HTTP client a `clickhouse::Client` sends through. +/// +/// HTTPS verifies the server's certificate and host name against the system +/// trust store, which honours `SSL_CERT_FILE` and `SSL_CERT_DIR`. +fn http_transport(config: &ClickHouseConfig) -> Result { + let mut connector = HttpConnector::new(); + connector.set_keepalive(Some(TCP_KEEPALIVE)); + connector.set_connect_timeout(Some(config.connect_timeout())); + let mut pool = HyperClient::builder(TokioExecutor::new()); + // hyper-util evicts idle connections only when the pool has a timer. + pool.pool_timer(TokioTimer::new()) + .pool_idle_timeout(POOL_IDLE_TIMEOUT); + if !config.tls { + return Ok(clickhouse::Client::with_http_client(pool.build(connector))); + } + connector.enforce_http(false); + let https = HttpsConnectorBuilder::new() + .with_provider_and_native_roots(rustls::crypto::aws_lc_rs::default_provider()) + .map_err(|e| { + ClickHouseError::Connection(format!( + "no CA certificate could be read from the system trust store: {e}" + )) + })? + .https_only() + .enable_http1() + .wrap_connector(connector); + Ok(clickhouse::Client::with_http_client(pool.build(https))) } /// Connection-pool statistics for the native TCP transport. @@ -70,8 +135,9 @@ pub struct PoolStats { /// /// # Errors /// -/// Returns an error if the config has no hosts. -pub(crate) fn build_client(config: &ClickHouseConfig) -> Result { +/// Returns an error if the config has no hosts, or if TLS is on and no CA +/// certificate can be read from the system trust store. +pub fn build_client(config: &ClickHouseConfig) -> Result { use super::config::Transport; let endpoint = config @@ -81,7 +147,7 @@ pub(crate) fn build_client(config: &ClickHouseConfig) -> Result { let scheme = if config.tls { "https" } else { "http" }; - let mut c = clickhouse::Client::default() + let mut c = http_transport(config)? .with_url(format!("{scheme}://{endpoint}")) .with_user(&config.username) .with_database(&config.database); @@ -163,6 +229,7 @@ impl ClickHouseQueryClient { Ok(Self { ch_client, database: config.database.clone(), + timeouts: config.timeouts(), }) } @@ -172,9 +239,7 @@ impl ClickHouseQueryClient { /// /// Returns an error if the query fails. pub async fn execute(&self, sql: &str) -> Result<()> { - self.ch_client - .query(sql) - .execute() + bounded(self.timeouts.request, self.ch_client.query(sql).execute()) .await .map_err(|e| ClickHouseError::Query(format!("{e}")))?; Ok(()) @@ -201,10 +266,12 @@ impl ClickHouseQueryClient { escape_string(&tbl) ); - let rows: Vec = - self.ch_client.query(&sql).fetch_all().await.map_err(|e| { - ClickHouseError::Schema(format!("Failed to fetch schema for {table}: {e}")) - })?; + 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}")))?; // The query succeeded and system.columns has nothing: the table really is // absent. Distinct from Schema above, which a transient outage produces. @@ -248,10 +315,12 @@ impl ClickHouseQueryClient { escape_string(&tbl) ); - let rows: Vec = - self.ch_client.query(&sql).fetch_all().await.map_err(|e| { - ClickHouseError::Schema(format!("Failed to fetch table comment: {e}")) - })?; + 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}")))?; Ok(rows.into_iter().next().map(|r| r.value).unwrap_or_default()) } @@ -277,9 +346,12 @@ impl ClickHouseQueryClient { comment: String, } - let rows: Vec = self.ch_client.query(&sql).fetch_all().await.map_err(|e| { - ClickHouseError::Schema(format!("Failed to fetch column comments: {e}")) - })?; + 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 mut comments = FxHashMap::default(); for row in rows { @@ -308,12 +380,12 @@ impl ClickHouseQueryClient { escape_string(&self.database) ); - let rows: Vec = self - .ch_client - .query(&sql) - .fetch_all() - .await - .map_err(|e| ClickHouseError::Schema(format!("Failed to list tables: {e}")))?; + 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}")))?; Ok(rows.into_iter().map(|r| r.name).collect()) } @@ -345,23 +417,24 @@ impl ClickHouseQueryClient { None => format!("SELECT COUNT(*) AS _dfe_count FROM {fq_table}"), }; - let row: CountRow = self - .ch_client - .query(&sql) - .fetch_one() - .await - .map_err(|e| ClickHouseError::Query(format!("{e}")))?; + let row: CountRow = bounded( + self.timeouts.request, + 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<()> { - self.ch_client - .query("SELECT 1") - .execute() - .await - .map_err(|e| ClickHouseError::Connection(format!("Health check failed: {e}")))?; + bounded( + self.timeouts.request, + self.ch_client.query("SELECT 1").execute(), + ) + .await + .map_err(|e| ClickHouseError::Connection(format!("Health check failed: {e}")))?; Ok(()) } @@ -408,7 +481,7 @@ impl ClickHouseQueryClient { escape_identifier(&db), escape_identifier(&tbl) ); - let mut insert = self.ch_client.insert_formatted_with(sql).buffered(); + let mut insert = insert_sink(&self.ch_client, sql, self.timeouts).buffered(); insert.write_buffered(&body); @@ -919,4 +992,153 @@ mod tests { }; assert!(ClickHouseQueryClient::new(&config).is_ok()); } + + // ============================================================ + // Connection bounds, against loopback servers that misbehave + // ============================================================ + + use std::net::SocketAddr; + use std::time::{Duration, Instant}; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::{TcpListener, TcpSocket, TcpStream}; + + /// An HTTP config pointing at `addr` with the given connect and request bounds. + fn http_config(addr: SocketAddr, connect_ms: u64, request_ms: u64) -> ClickHouseConfig { + ClickHouseConfig { + hosts: vec![addr.to_string()], + transport: super::super::config::Transport::Http, + database: "default".to_string(), + connect_timeout_ms: connect_ms, + request_timeout_ms: request_ms, + ..Default::default() + } + } + + /// Reads one request with a `Content-Length` body, so nothing is left unread. + async fn read_request(socket: &mut TcpStream) { + let mut request = Vec::new(); + let mut chunk = [0u8; 1024]; + let head_end = loop { + let n = socket.read(&mut chunk).await.unwrap(); + assert!(n > 0, "client closed before sending a request"); + request.extend_from_slice(&chunk[..n]); + if let Some(pos) = request.windows(4).position(|w| w == b"\r\n\r\n") { + break pos + 4; + } + }; + let head = String::from_utf8_lossy(&request[..head_end]).to_ascii_lowercase(); + let body_len: usize = head + .lines() + .find_map(|line| line.strip_prefix("content-length:")) + .map_or(0, |v| v.trim().parse().unwrap()); + while request.len() < head_end + body_len { + let n = socket.read(&mut chunk).await.unwrap(); + assert!(n > 0, "client closed mid-request"); + request.extend_from_slice(&chunk[..n]); + } + } + + /// A listener whose accept queue is full, so the kernel drops every further + /// SYN and a connect to it never completes. The returned streams keep it full. + async fn listener_that_never_completes_a_connect() -> (TcpListener, Vec, SocketAddr) + { + let socket = TcpSocket::new_v4().unwrap(); + socket.bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let listener = socket.listen(1).unwrap(); + let addr = listener.local_addr().unwrap(); + let mut queued = Vec::new(); + for _ in 0..4 { + if let Ok(Ok(stream)) = + tokio::time::timeout(Duration::from_millis(300), TcpStream::connect(addr)).await + { + queued.push(stream); + } + } + (listener, queued, addr) + } + + #[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(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + read_request(&mut socket).await; + socket + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") + .await + .unwrap(); + let answered = Instant::now(); + // A zero-byte read is the client closing its end. + let mut byte = [0u8; 1]; + let read = socket.read(&mut byte).await.unwrap(); + (read, answered.elapsed()) + }); + + let client = build_client(&http_config(addr, 5_000, 30_000)).unwrap(); + client.query("SELECT 1").execute().await.unwrap(); + + let (read, idle) = tokio::time::timeout(Duration::from_secs(8), server) + .await + .expect("the pool kept an idle connection open for 8 s") + .unwrap(); + assert_eq!(read, 0, "the client sent more bytes instead of closing"); + assert!( + idle >= Duration::from_millis(1_500) && idle <= Duration::from_millis(5_000), + "an idle connection closed after {idle:?}, not about 2 s" + ); + // Dropping the client also closes the pool, so it must outlive the wait above. + drop(client); + } + + #[tokio::test] + async fn connect_timeout_bounds_a_connect_that_never_completes() { + let (_listener, _queued, addr) = listener_that_never_completes_a_connect().await; + let client = build_client(&http_config(addr, 300, 30_000)).unwrap(); + + let started = Instant::now(); + let result = + tokio::time::timeout(Duration::from_secs(10), client.query("SELECT 1").execute()) + .await + .expect( + "a connect to a full accept queue ran past 10 s with a 300 ms connect timeout", + ); + let elapsed = started.elapsed(); + + assert!( + result.is_err(), + "a connect that never completed reported success" + ); + assert!( + elapsed >= Duration::from_millis(250) && elapsed < Duration::from_secs(3), + "the connect gave up after {elapsed:?}, not at its 300 ms bound" + ); + } + + #[tokio::test] + async fn request_timeout_bounds_a_query_the_server_never_answers() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let _server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + read_request(&mut socket).await; + // Holds the connection open and never answers. + let mut byte = [0u8; 1]; + let _ = socket.read(&mut byte).await; + }); + let client = ClickHouseQueryClient::new(&http_config(addr, 5_000, 300)).unwrap(); + + let started = Instant::now(); + let result = tokio::time::timeout(Duration::from_secs(10), client.health_check()) + .await + .expect("a query the server never answered ran past 10 s with a 300 ms bound"); + let elapsed = started.elapsed(); + + assert!(result.is_err(), "an unanswered query reported success"); + assert!( + elapsed >= Duration::from_millis(250) && elapsed < Duration::from_secs(3), + "the query gave up after {elapsed:?}, not at its 300 ms bound" + ); + } } diff --git a/src/clickhouse/config.rs b/src/clickhouse/config.rs index 77bea31..6e43a7d 100644 --- a/src/clickhouse/config.rs +++ b/src/clickhouse/config.rs @@ -160,14 +160,20 @@ pub struct ClickHouseConfig { #[serde(default)] pub tls: bool, - /// Connection timeout in milliseconds. + /// Bound on opening a TCP connection to `ClickHouse`, in milliseconds. #[serde(default = "default_connect_timeout")] pub connect_timeout_ms: u64, - /// Request timeout in milliseconds. + /// Bound on a query, and on an insert waiting to hand over its next chunk, + /// in milliseconds. #[serde(default = "default_request_timeout")] pub request_timeout_ms: u64, + /// Bound on an insert waiting for the server's answer once its rows are + /// sent, in milliseconds. + #[serde(default = "default_insert_timeout")] + pub insert_timeout_ms: u64, + /// Enable response compression (HTTP only). #[serde(default = "default_compression")] pub compression: bool, @@ -193,14 +199,40 @@ pub fn host_implies_tls(host: &str) -> bool { .is_some_and(|port| matches!(port, 8443 | 9440)) } -const fn default_connect_timeout() -> u64 { +pub(crate) const fn default_connect_timeout() -> u64 { 5000 } -const fn default_request_timeout() -> u64 { +pub(crate) const fn default_request_timeout() -> u64 { 30000 } +/// 60 s is ~270x the ClickHouse CPU cost of a 20,000-row flush 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. +pub(crate) const fn default_insert_timeout() -> u64 { + 60000 +} + +/// 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. + 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, +} + +impl Default for Timeouts { + fn default() -> Self { + Self { + request: std::time::Duration::from_millis(default_request_timeout()), + insert: std::time::Duration::from_millis(default_insert_timeout()), + } + } +} + const fn default_compression() -> bool { true } @@ -217,6 +249,7 @@ impl Default for ClickHouseConfig { tls: false, connect_timeout_ms: default_connect_timeout(), request_timeout_ms: default_request_timeout(), + insert_timeout_ms: default_insert_timeout(), compression: default_compression(), } } @@ -277,6 +310,21 @@ impl ClickHouseConfig { self } + /// Bound on opening a TCP connection. + #[must_use] + pub fn connect_timeout(&self) -> std::time::Duration { + std::time::Duration::from_millis(self.connect_timeout_ms) + } + + /// The request and insert bounds. + #[must_use] + pub fn timeouts(&self) -> Timeouts { + Timeouts { + request: std::time::Duration::from_millis(self.request_timeout_ms), + insert: std::time::Duration::from_millis(self.insert_timeout_ms), + } + } + /// Get the first host (primary). #[must_use] pub fn primary_host(&self) -> Option<&str> { @@ -379,6 +427,9 @@ mod tests { assert!(!config.tls); assert_eq!(config.connect_timeout_ms, 5000); assert_eq!(config.request_timeout_ms, 30000); + assert_eq!(config.insert_timeout_ms, 60000); + assert_eq!(config.connect_timeout(), std::time::Duration::from_secs(5)); + assert_eq!(config.timeouts(), Timeouts::default()); assert!(config.compression); } diff --git a/src/clickhouse/error.rs b/src/clickhouse/error.rs index 45fea52..b760e06 100644 --- a/src/clickhouse/error.rs +++ b/src/clickhouse/error.rs @@ -167,7 +167,8 @@ pub fn classify_dynamic_error(err: &crate::clickhouse_ext::DynamicError) -> Erro DynamicError::SchemaFetch { .. } | DynamicError::UnsupportedType { .. } | DynamicError::EmptySchema { .. } - | DynamicError::SchemaMismatch { .. } => ErrorCategory::Transient, + | DynamicError::SchemaMismatch { .. } + | DynamicError::TimedOut { .. } => ErrorCategory::Transient, } } diff --git a/src/clickhouse/inserter.rs b/src/clickhouse/inserter.rs index 5228257..8eefb9d 100644 --- a/src/clickhouse/inserter.rs +++ b/src/clickhouse/inserter.rs @@ -44,8 +44,8 @@ use tracing::{debug, error, info, trace, warn}; use crate::Result; use crate::buffer::{FlushBatch, KafkaOffset}; -use crate::clickhouse::client_http::escape_identifier; -use crate::clickhouse::config::InsertFormat; +use crate::clickhouse::client_http::{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, retries, server_code, @@ -293,6 +293,8 @@ 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. + timeouts: Timeouts, } impl Inserter { @@ -328,9 +330,23 @@ impl Inserter { schema_cache: None, coercer: None, refresh_on_error: true, + timeouts: Timeouts::default(), } } + /// Set the bounds on schema fetches and inserts (`clickhouse.request_timeout_ms` + /// and `clickhouse.insert_timeout_ms`). + pub fn with_timeouts(mut self, timeouts: Timeouts) -> Self { + self.timeouts = timeouts; + self + } + + /// The bounds on schema fetches and inserts. + #[cfg(test)] + pub(crate) fn timeouts(&self) -> Timeouts { + self.timeouts + } + /// Set the insert format. /// /// - `RowBinary` (default): schema-reflected binary via `DynamicInsert`. @@ -452,7 +468,14 @@ impl Inserter { let schema = match self.dynamic_schema_cache.get(table) { Some(schema) => schema, None => { - match crate::clickhouse_ext::fetch_dynamic_schema(&self.ch_client, db, tbl).await { + match crate::clickhouse_ext::fetch_dynamic_schema( + &self.ch_client, + db, + tbl, + self.timeouts.request, + ) + .await + { Ok(fetched) => { self.dynamic_schema_cache.insert(table, fetched.clone()); fetched @@ -484,7 +507,7 @@ impl Inserter { let Some(encoded_against) = encoded_against else { return false; }; - crate::clickhouse_ext::fetch_dynamic_schema(&self.ch_client, db, tbl) + crate::clickhouse_ext::fetch_dynamic_schema(&self.ch_client, db, tbl, self.timeouts.request) .await .is_ok_and(|now| now.same_columns(encoded_against)) } @@ -562,7 +585,8 @@ impl Inserter { tbl, Arc::clone(&self.dynamic_schema_cache), ) - .refresh_on_error(self.refresh_on_error); + .refresh_on_error(self.refresh_on_error) + .with_timeouts(self.timeouts); let mut write_failed = false; for (i, row) in rows.iter().enumerate() { @@ -586,6 +610,11 @@ impl Inserter { insert.write_map(row).await }; if let Err(e) = write_result { + // Another attempt inside this cycle only keeps the event loop + // blocked, so a timed-out fetch goes back to be held. + if matches!(e, crate::clickhouse_ext::DynamicError::TimedOut { .. }) { + return Err(crate::Error::ClickHouse(format!("RowBinary write: {e}"))); + } // Schema mismatch — invalidate both fork and loader caches, then retry if matches!( e, @@ -797,7 +826,7 @@ impl Inserter { for attempt in 0..=self.max_retries { // HTTP only, like the RowBinary path: `clickhouse.protocol: native` // is refused at startup, and the native sink waits on clickhouse-rs#15. - let mut insert = self.ch_client.insert_formatted_with(sql.clone()); + let mut insert = insert_sink(&self.ch_client, sql.clone(), self.timeouts); // One INSERT carries one verdict, and which call surfaces it is an // accident of buffering: the fork queues the body on a channel, so @@ -822,6 +851,12 @@ impl Inserter { }; let message = format!("JSONEachRow insert: {e}"); + // Another attempt inside this cycle only keeps the event loop + // blocked, so a timed-out insert goes back to be held. + if matches!(e, clickhouse::error::Error::TimedOut) { + warn!(table = %table, error = %e, "Insert timed out, returning the batch to be held"); + return Err(crate::Error::ClickHouse(message)); + } match classify_json_insert_error(&e) { ErrorCategory::Data => { debug!( @@ -1273,6 +1308,103 @@ mod tests { ); } + /// A loopback server that takes every connection and never answers. + async fn server_that_never_answers() -> std::net::SocketAddr { + use tokio::io::AsyncReadExt; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + tokio::spawn(async move { + let mut discard = [0u8; 4096]; + while matches!(socket.read(&mut discard).await, Ok(n) if n > 0) {} + }); + } + }); + addr + } + + /// An inserter for `default.t` at `addr` with 300 ms bounds and the table's + /// schema already cached, so the insert is the first request it sends. + fn inserter_with_short_bounds(addr: std::net::SocketAddr, format: InsertFormat) -> Inserter { + let config = crate::clickhouse::ClickHouseConfig { + hosts: vec![addr.to_string()], + transport: crate::clickhouse::Transport::Http, + database: "default".to_string(), + request_timeout_ms: 300, + insert_timeout_ms: 300, + ..Default::default() + }; + let http_client = Arc::new(ClickHouseQueryClient::new(&config).unwrap()); + let ch_client = crate::clickhouse::client_http::build_client(&config).unwrap(); + let inserter = Inserter::new(http_client, ch_client, InserterConfig::default()) + .with_insert_format(format) + .with_timeouts(config.timeouts()); + inserter.dynamic_schema_cache.insert( + "default.t", + crate::clickhouse_ext::DynamicSchema::from_columns( + "default.t", + vec![ColumnDef::with_default_kind("id", "UInt64", "")], + ), + ); + inserter + } + + #[tokio::test] + async fn an_insert_the_server_never_answers_comes_back_to_be_held_within_its_bound() { + for format in [InsertFormat::RowBinary, InsertFormat::JsonEachRow] { + let inserter = inserter_with_short_bounds(server_that_never_answers().await, format); + let batch = FlushBatch { + table: "default.t".into(), + rows: vec![ + serde_json::json!({ "id": 1 }).as_object().unwrap().clone(), + serde_json::json!({ "id": 2 }).as_object().unwrap().clone(), + ], + offsets: vec![ + KafkaOffset::new("events", 0, 100), + KafkaOffset::new("events", 0, 101), + ], + raw_payloads: Vec::new(), + reserved: vec![16, 16], + }; + + let started = std::time::Instant::now(); + let result = + tokio::time::timeout(Duration::from_secs(10), inserter.insert_with_salvage(batch)) + .await + .unwrap_or_else(|_| panic!("{format}: an unanswered insert ran past 10 s")); + let elapsed = started.elapsed(); + + assert!( + !result.is_settled(), + "{format}: an unanswered insert settled" + ); + assert!(result.failed.is_empty(), "{format}: rows went to the DLQ"); + let BatchDisposition::Retry(reason) = &result.disposition else { + panic!("{format}: expected a retry, got {:?}", result.disposition); + }; + assert!(reason.contains("time"), "{format}: reason {reason}"); + let held = result + .unsettled + .unwrap_or_else(|| panic!("{format}: the batch was not handed back to hold")); + assert_eq!( + held.rows.len(), + 2, + "{format}: rows missing from the held batch" + ); + assert_eq!( + held.offsets.iter().map(|o| o.offset).collect::>(), + vec![100, 101], + "{format}: offsets missing from the held batch" + ); + // One 300 ms attempt; retrying inside the cycle would take seconds. + assert!( + elapsed >= Duration::from_millis(250) && elapsed < Duration::from_secs(2), + "{format}: the insert came back after {elapsed:?}, not after one 300 ms bound" + ); + } + } + #[test] fn split_parallel_tolerates_a_short_or_empty_companion() { // offsets is empty on the gRPC path and raw_payloads is empty in diff --git a/src/clickhouse/mod.rs b/src/clickhouse/mod.rs index 9b7e772..c8c42d8 100644 --- a/src/clickhouse/mod.rs +++ b/src/clickhouse/mod.rs @@ -39,7 +39,7 @@ pub mod schema; // Re-export core types pub use client_http::{ClickHouseQueryClient, PoolStats, SharedQueryClient}; -pub use config::{ClickHouseConfig, InsertFormat, Transport, host_implies_tls}; +pub use config::{ClickHouseConfig, InsertFormat, Timeouts, Transport, host_implies_tls}; pub use error::{ClickHouseError, ErrorCategory}; pub use types::{ ColumnInfo, NULL_STRINGS, ParsedType, ParsedTypeExt, TableSchema, default_value_for_category, diff --git a/src/clickhouse_ext/error.rs b/src/clickhouse_ext/error.rs index 22793a1..4979240 100644 --- a/src/clickhouse_ext/error.rs +++ b/src/clickhouse_ext/error.rs @@ -29,6 +29,8 @@ pub enum DynamicError { }, /// Table has no columns (or does not exist). EmptySchema { table: String }, + /// A schema fetch or insert ran past its client-side bound. + TimedOut { table: String }, } impl fmt::Display for DynamicError { @@ -49,6 +51,9 @@ impl fmt::Display for DynamicError { Self::EmptySchema { table } => { write!(f, "table '{table}' has no columns or does not exist") } + Self::TimedOut { table } => { + write!(f, "request to ClickHouse for '{table}' timed out") + } } } } diff --git a/src/clickhouse_ext/insert.rs b/src/clickhouse_ext/insert.rs index f338509..02ba8b5 100644 --- a/src/clickhouse_ext/insert.rs +++ b/src/clickhouse_ext/insert.rs @@ -38,6 +38,8 @@ use clickhouse::Client; 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; /// 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 @@ -106,6 +108,8 @@ 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. + timeouts: Timeouts, } impl DynamicInsert { @@ -126,9 +130,17 @@ impl DynamicInsert { sink: None, rows_written: 0, refresh_on_error: true, + timeouts: Timeouts::default(), } } + /// Set the bounds on the schema fetch and the insert (default + /// [`Timeouts::default`]). + pub fn with_timeouts(mut self, timeouts: Timeouts) -> Self { + self.timeouts = timeouts; + 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. @@ -150,7 +162,13 @@ 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).await?; + let fetched = fetch_dynamic_schema( + &self.client, + &self.database, + &self.table, + self.timeouts.request, + ) + .await?; self.schema_cache.insert(&full, fetched.clone()); fetched }; @@ -209,7 +227,7 @@ impl DynamicInsert { if schema.has_json_columns() { client = client.with_setting(JSON_AS_STRING_SETTING, "1"); } - let mut sink = client.insert_formatted_with(sql).buffered(); + let mut sink = insert_sink(&client, sql, self.timeouts).buffered(); sink.write_buffered(&names_and_types_header(&columns)); self.sink = Some(sink); @@ -296,11 +314,17 @@ impl DynamicInsert { /// /// # Errors /// - /// Returns [`DynamicError`] if the server rejects the insert. + /// Returns [`DynamicError`] if the server rejects the insert, or + /// [`DynamicError::TimedOut`] if its answer does not arrive within the bound. pub async fn end(mut self) -> Result { if let Some(mut sink) = self.sink.take() && let Err(e) = sink.end().await { + if matches!(e, clickhouse::error::Error::TimedOut) { + return Err(DynamicError::TimedOut { + table: self.full_table(), + }); + } let err = classify_error(&self.full_table(), &e); if self.refresh_on_error && matches!(err, DynamicError::SchemaMismatch { .. }) { self.schema_cache.invalidate(&self.full_table()); diff --git a/src/clickhouse_ext/schema.rs b/src/clickhouse_ext/schema.rs index 2f389d0..8f3efad 100644 --- a/src/clickhouse_ext/schema.rs +++ b/src/clickhouse_ext/schema.rs @@ -122,17 +122,18 @@ struct SysColumn { /// /// # Errors /// -/// Returns [`DynamicError::SchemaFetch`] if the query fails, or -/// [`DynamicError::EmptySchema`] if the table has no columns (or does not -/// exist). +/// Returns [`DynamicError::SchemaFetch`] if the query fails or runs past +/// `timeout`, 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, ) -> Result { let full_table = format!("{database}.{table}"); - let rows = client + let query = client .query( "SELECT name, type AS col_type, default_kind \ FROM system.columns \ @@ -141,11 +142,17 @@ pub async fn fetch_dynamic_schema( ) .bind(database) .bind(table) - .fetch_all::() + .fetch_all::(); + let rows = crate::clickhouse::client_http::bounded(timeout, query) .await - .map_err(|source| DynamicError::SchemaFetch { - table: full_table.clone(), - source, + .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() { diff --git a/src/config/loader.rs b/src/config/loader.rs index 2a1e62e..5df2ec4 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -178,6 +178,14 @@ pub struct ClickHouseConfig { pub insert_format: crate::clickhouse::InsertFormat, pub tables: Vec, 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. + 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. + pub insert_timeout_ms: u64, } impl Default for ClickHouseConfig { @@ -193,6 +201,9 @@ impl Default for ClickHouseConfig { insert_format: crate::clickhouse::InsertFormat::default(), tables: Vec::new(), tls: None, + connect_timeout_ms: crate::clickhouse::config::default_connect_timeout(), + request_timeout_ms: crate::clickhouse::config::default_request_timeout(), + insert_timeout_ms: crate::clickhouse::config::default_insert_timeout(), } } } @@ -224,9 +235,10 @@ impl From<&ClickHouseConfig> for crate::clickhouse::ClickHouseConfig { username: cfg.username.clone(), password: cfg.password.expose().to_string(), tls, - connect_timeout_ms: 5000, // Default timeout - request_timeout_ms: 30000, // Default timeout - compression: true, // Enable by default for HTTP + connect_timeout_ms: cfg.connect_timeout_ms, + request_timeout_ms: cfg.request_timeout_ms, + insert_timeout_ms: cfg.insert_timeout_ms, + compression: true, // Enable by default for HTTP } } } @@ -512,6 +524,18 @@ impl Config { "At least one ClickHouse host must be configured".into(), )); } + // A zero bound fails every request before it is sent. + for (name, value) in [ + ("connect_timeout_ms", self.clickhouse.connect_timeout_ms), + ("request_timeout_ms", self.clickhouse.request_timeout_ms), + ("insert_timeout_ms", self.clickhouse.insert_timeout_ms), + ] { + if value == 0 { + return Err(crate::Error::Config(format!( + "clickhouse.{name} must be greater than 0" + ))); + } + } // Of the borrowed Kafka TlsConfig shape only `enabled` reaches a client. if let Some(ref tls) = self.clickhouse.tls { let ignored: Vec<&str> = [ @@ -1556,6 +1580,7 @@ kafka: insert_format: crate::clickhouse::InsertFormat::default(), tables: vec!["events".to_string()], tls: None, + ..ClickHouseConfig::default() }; let client_cfg: crate::clickhouse::ClickHouseConfig = (&cfg).into(); assert_eq!(client_cfg.hosts, vec!["ch:9000"]); @@ -1580,6 +1605,7 @@ kafka: insert_format: crate::clickhouse::InsertFormat::default(), tables: vec![], tls: None, + ..ClickHouseConfig::default() }; let client_cfg: crate::clickhouse::ClickHouseConfig = (&cfg).into(); assert!(matches!( @@ -1588,6 +1614,56 @@ kafka: )); } + #[test] + fn the_clickhouse_timeouts_reach_the_client_config() { + let cfg = ClickHouseConfig { + connect_timeout_ms: 1_234, + request_timeout_ms: 5_678, + insert_timeout_ms: 91_011, + ..ClickHouseConfig::default() + }; + let client_cfg: crate::clickhouse::ClickHouseConfig = (&cfg).into(); + let ms = std::time::Duration::from_millis; + assert_eq!(client_cfg.connect_timeout(), ms(1_234)); + assert_eq!( + client_cfg.timeouts(), + crate::clickhouse::Timeouts { + request: ms(5_678), + insert: ms(91_011), + } + ); + } + + #[test] + fn the_default_clickhouse_timeouts_are_5_s_connect_30_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().insert, secs(60)); + } + + #[test] + fn a_zero_clickhouse_timeout_fails_validation() { + for key in [ + "connect_timeout_ms", + "request_timeout_ms", + "insert_timeout_ms", + ] { + let mut config = Config::default(); + match key { + "connect_timeout_ms" => config.clickhouse.connect_timeout_ms = 0, + "request_timeout_ms" => config.clickhouse.request_timeout_ms = 0, + _ => config.clickhouse.insert_timeout_ms = 0, + } + let err = config.validate().unwrap_err().to_string(); + assert!( + err.contains(&format!("clickhouse.{key}")), + "a zero {key} was not named: {err}" + ); + } + } + #[test] fn test_clickhouse_insert_format_reaches_the_client_config() { // The From impl hard-coded the default, so every value of this key was diff --git a/src/pipeline/orchestrator.rs b/src/pipeline/orchestrator.rs index 0582784..b59a9d3 100644 --- a/src/pipeline/orchestrator.rs +++ b/src/pipeline/orchestrator.rs @@ -2349,7 +2349,9 @@ fn build_inserter( http_client: Arc, ch_client: clickhouse::Client, ) -> Inserter { + let ch_config: crate::clickhouse::ClickHouseConfig = (&config.clickhouse).into(); Inserter::new(http_client, ch_client, InserterConfig::default()) + .with_timeouts(ch_config.timeouts()) .with_insert_format(config.clickhouse.insert_format) .with_refresh_on_error(config.schema.refresh_on_error) .with_schema_ttl(Duration::from_secs(config.schema.cache_ttl_secs)) @@ -4214,6 +4216,23 @@ mod tests { assert!(!inserter.refreshes_on_error()); } + #[test] + fn the_clickhouse_timeouts_reach_the_inserter() { + let mut config = Config::default(); + config.clickhouse.request_timeout_ms = 1_500; + config.clickhouse.insert_timeout_ms = 2_500; + + let inserter = inserter_for(&config); + + assert_eq!( + inserter.timeouts(), + crate::clickhouse::Timeouts { + request: Duration::from_millis(1_500), + insert: Duration::from_millis(2_500), + } + ); + } + #[test] fn the_default_schema_settings_keep_300_s_and_refresh_on_error() { let inserter = inserter_for(&Config::default()); diff --git a/tests/common/containers.rs b/tests/common/containers.rs index 4a6f9c4..d67c0ef 100644 --- a/tests/common/containers.rs +++ b/tests/common/containers.rs @@ -147,6 +147,7 @@ mod testcontainers_impl { insert_format: dfe_loader::clickhouse::InsertFormat::default(), tables: Vec::new(), tls: None, + ..Default::default() }) } @@ -223,6 +224,67 @@ mod testcontainers_impl { .expect("Failed to start ClickHouse container") } + /// ClickHouse's HTTPS port inside the container. + const CLICKHOUSE_HTTPS_PORT: u16 = 8443; + + /// Server config that opens HTTPS on 8443 with the certificate copied in beside it. + const CLICKHOUSE_HTTPS_CONFIG: &str = r" + 8443 + + + /etc/clickhouse-server/certs/server.crt + /etc/clickhouse-server/certs/server.key + none + false + + + +"; + + /// Start ClickHouse serving HTTP on `http_host_port` and HTTPS with + /// `cert_pem` and `key_pem` on `https_host_port`. + /// + /// Both are mapped explicitly: one explicit mapping turns off publishing + /// the image's own ports, and the readiness probe needs 8123. + pub async fn start_clickhouse_https( + test: &str, + cert_pem: String, + key_pem: String, + http_host_port: u16, + https_host_port: u16, + ) -> ContainerAsync { + let name = crate::common::container_name(Some(test), "clickhouse"); + crate::common::reap_stale(&name); + ClickHouseImage::default() + .with_tag(format!("{CLICKHOUSE_TAG}@{CLICKHOUSE_DIGEST}")) + .with_env_var("CLICKHOUSE_SKIP_USER_SETUP", "1") + .with_copy_to( + "/etc/clickhouse-server/config.d/dfe-https.xml", + CLICKHOUSE_HTTPS_CONFIG.as_bytes().to_vec(), + ) + .with_copy_to( + "/etc/clickhouse-server/certs/server.crt", + cert_pem.into_bytes(), + ) + .with_copy_to( + "/etc/clickhouse-server/certs/server.key", + key_pem.into_bytes(), + ) + .with_mapped_port(http_host_port, CLICKHOUSE_PORT) + .with_mapped_port( + https_host_port, + testcontainers::core::ContainerPort::Tcp(CLICKHOUSE_HTTPS_PORT), + ) + .with_container_name(&name) + .with_labels(crate::common::test_labels("clickhouse")) + .with_ready_conditions(vec![clickhouse_ready()]) + .with_log_consumer(echo_logs(name.clone())) + .with_startup_timeout(Duration::from_secs(180)) + .start() + .await + .expect("Failed to start ClickHouse container with HTTPS") + } + /// Start Kafka container async fn start_kafka(test: &str) -> ContainerAsync { let name = crate::common::container_name(Some(test), "kafka"); @@ -293,3 +355,6 @@ mod testcontainers_impl { // reference it — hence the allow. #[allow(unused_imports)] pub use testcontainers_impl::TestInfrastructure; +#[cfg(feature = "testcontainers")] +#[allow(unused_imports)] +pub use testcontainers_impl::start_clickhouse_https; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 008d8c2..064ec9a 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -607,6 +607,7 @@ pub fn get_clickhouse_config() -> ClickHouseConfig { insert_format: dfe_loader::clickhouse::InsertFormat::default(), tables: Vec::new(), tls: None, + ..Default::default() } } diff --git a/tests/e2e/kafka_to_clickhouse.rs b/tests/e2e/kafka_to_clickhouse.rs index 2f37345..88502a9 100644 --- a/tests/e2e/kafka_to_clickhouse.rs +++ b/tests/e2e/kafka_to_clickhouse.rs @@ -188,6 +188,7 @@ fn ch_config_from_env() -> ClickHouseConfig { } else { None }, + ..Default::default() } } diff --git a/tests/integration/clickhouse_tls.rs b/tests/integration/clickhouse_tls.rs new file mode 100644 index 0000000..e536a78 --- /dev/null +++ b/tests/integration/clickhouse_tls.rs @@ -0,0 +1,157 @@ +// SPDX-License-Identifier: BUSL-1.1 +// Copyright (c) 2026 HYPERI PTY LIMITED + +//! HTTPS to `ClickHouse` is verified against the system trust store. +//! +//! A private CA reaches the loader's client the only way the config allows, +//! through the trust store (`SSL_CERT_FILE` here), so the same server is +//! accepted when its CA is there and refused when it is not, when another CA +//! is, and when the host name is not the one the certificate names. +//! +//! Gated behind `#[cfg(feature = "testcontainers")]`. Needs process-per-test +//! (nextest): it sets `SSL_CERT_FILE` for the whole process. + +#![cfg(feature = "testcontainers")] + +use std::path::Path; + +use dfe_loader::clickhouse::client_http::build_client; +use dfe_loader::clickhouse::{ClickHouseConfig, ClickHouseQueryClient, Transport}; +use rcgen::{ + BasicConstraints, CertificateParams, CertifiedIssuer, DnType, IsCa, KeyPair, KeyUsagePurpose, +}; + +use super::grpc_outage::free_port; +use crate::common::containers::start_clickhouse_https; +use crate::test_name; + +/// A self-signed CA named `name`. +fn private_ca(name: &str) -> CertifiedIssuer<'static, KeyPair> { + let mut params = CertificateParams::new(Vec::::new()).expect("CA params"); + params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + params.distinguished_name.push(DnType::CommonName, name); + params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign]; + CertifiedIssuer::self_signed(params, KeyPair::generate().expect("CA key")).expect("CA cert") +} + +/// A server certificate for `localhost` only, signed by `ca`, as (cert, key) PEM. +fn localhost_cert(ca: &CertifiedIssuer<'static, KeyPair>) -> (String, String) { + let key = KeyPair::generate().expect("server key"); + let mut params = CertificateParams::new(vec!["localhost".to_string()]).expect("server params"); + params + .distinguished_name + .push(DnType::CommonName, "localhost"); + let cert = params.signed_by(&key, ca).expect("server cert"); + (cert.pem(), key.serialize_pem()) +} + +/// Two distinct free host ports. +fn two_free_ports() -> (u16, u16) { + let first = free_port(); + let _held = std::net::TcpListener::bind(("127.0.0.1", first)).expect("hold the first port"); + (first, free_port()) +} + +fn config(host: &str, port: u16, tls: bool) -> ClickHouseConfig { + ClickHouseConfig { + hosts: vec![format!("{host}:{port}")], + transport: Transport::Http, + database: "default".to_string(), + tls, + ..Default::default() + } +} + +/// Point the system trust store at `bundle`, or back at the platform's own. +fn trust(bundle: Option<&Path>) { + // SAFETY: test-only; nextest runs this test in a process of its own. + unsafe { + std::env::remove_var("SSL_CERT_DIR"); + match bundle { + Some(path) => std::env::set_var("SSL_CERT_FILE", path), + None => std::env::remove_var("SSL_CERT_FILE"), + } + } +} + +/// Every message in `err`'s source chain, so a TLS refusal is visible past +/// the transport's own wording. +fn chain(err: &(dyn std::error::Error + 'static)) -> String { + let mut out = err.to_string(); + let mut next = err.source(); + while let Some(e) = next { + out.push_str(" <- "); + out.push_str(&e.to_string()); + next = e.source(); + } + out +} + +/// `SELECT 1` through the loader's client, with the trust store as it now stands. +async fn select_one(config: &ClickHouseConfig) -> Result<(), String> { + let client = build_client(config).map_err(|e| e.to_string())?; + client + .query("SELECT 1") + .execute() + .await + .map_err(|e| chain(&e)) +} + +#[tokio::test(flavor = "multi_thread")] +async fn https_is_verified_against_the_system_trust_store() { + let ca = private_ca("dfe-loader test CA"); + let other_ca = private_ca("unrelated CA"); + let (cert, key) = localhost_cert(&ca); + let (http_port, https_port) = two_free_ports(); + let _clickhouse = start_clickhouse_https(test_name!(), cert, key, http_port, https_port).await; + + let dir = tempfile::tempdir().expect("CA bundle directory"); + let ca_bundle = dir.path().join("private-ca.pem"); + let other_bundle = dir.path().join("other-ca.pem"); + std::fs::write(&ca_bundle, ca.pem()).expect("write CA bundle"); + std::fs::write(&other_bundle, other_ca.pem()).expect("write other CA bundle"); + + trust(Some(&ca_bundle)); + let with_its_ca = select_one(&config("localhost", https_port, true)).await; + let query_client = ClickHouseQueryClient::new(&config("localhost", https_port, true)) + .expect("query client") + .health_check() + .await; + let wrong_name = select_one(&config("127.0.0.1", https_port, true)).await; + let plain_http = select_one(&config("localhost", http_port, false)).await; + + trust(Some(&other_bundle)); + let wrong_ca = select_one(&config("localhost", https_port, true)).await; + + trust(None); + let no_ca = select_one(&config("localhost", https_port, true)).await; + + assert!( + with_its_ca.is_ok(), + "HTTPS was refused with the server's CA in the trust store: {with_its_ca:?}" + ); + assert!( + query_client.is_ok(), + "the query client was refused with the server's CA in the trust store: {query_client:?}" + ); + assert!( + plain_http.is_ok(), + "plain HTTP with TLS off failed: {plain_http:?}" + ); + for (case, result) in [ + ("a host name the certificate does not name", &wrong_name), + ("only an unrelated CA in the trust store", &wrong_ca), + ("the platform trust store alone", &no_ca), + ] { + let err = result + .as_ref() + .expect_err(&format!("HTTPS was accepted with {case}")); + // A platform store with no CA in it fails at the client build instead. + assert!( + err.contains("invalid peer certificate") + || err.contains("no CA certificate could be read"), + "with {case} the request failed for a reason other than certificate \ + verification: {err}" + ); + } +} diff --git a/tests/integration/kafka_offset_floor.rs b/tests/integration/kafka_offset_floor.rs index ed43383..392adfa 100644 --- a/tests/integration/kafka_offset_floor.rs +++ b/tests/integration/kafka_offset_floor.rs @@ -17,7 +17,7 @@ use std::collections::BTreeSet; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; use dfe_loader::clickhouse::InsertFormat; @@ -135,6 +135,86 @@ impl ReadRefusingProxy { } } +/// A loopback proxy to `ClickHouse` HTTP that can stall: every connection opened +/// while stalled is held open and never answered, so a request on it ends only +/// at the loader's own bound. Counts the inserts it has stalled. +struct StallingProxy { + port: u16, + stall: Arc, + stalled_inserts: Arc, + cut: watch::Sender, +} + +impl StallingProxy { + async fn start(upstream: String) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind proxy"); + let port = listener.local_addr().expect("proxy addr").port(); + let stall = Arc::new(AtomicBool::new(false)); + let stalled_inserts = Arc::new(AtomicUsize::new(0)); + let (cut, _) = watch::channel(0_u64); + + let accept_stall = Arc::clone(&stall); + let accept_count = Arc::clone(&stalled_inserts); + let accept_cut = cut.clone(); + tokio::spawn(async move { + while let Ok((client, _)) = listener.accept().await { + let stall = Arc::clone(&accept_stall); + let count = Arc::clone(&accept_count); + let upstream = upstream.clone(); + let mut cut_rx = accept_cut.subscribe(); + tokio::spawn(async move { + let mut client = client; + if stall.load(Ordering::SeqCst) { + let mut head = Vec::new(); + if read_request_line(&mut client, &mut head) + .await + .is_some_and(|line| line.contains("query=INSERT")) + { + count.fetch_add(1, Ordering::SeqCst); + } + let _ = cut_rx.changed().await; + return; + } + let Ok(mut server) = TcpStream::connect(&upstream).await else { + return; + }; + tokio::select! { + _ = tokio::io::copy_bidirectional(&mut client, &mut server) => {} + _ = cut_rx.changed() => {} + } + }); + } + }); + + Self { + port, + stall, + stalled_inserts, + cut, + } + } + + fn address(&self) -> String { + format!("127.0.0.1:{}", self.port) + } + + /// Sever every open connection and hold every new one unanswered. + fn stall(&self) { + self.stall.store(true, Ordering::SeqCst); + self.cut.send_modify(|n| *n += 1); + } + + /// Release the held connections and forward again. + fn resume(&self) { + self.stall.store(false, Ordering::SeqCst); + self.cut.send_modify(|n| *n += 1); + } + + fn stalled_inserts(&self) -> usize { + self.stalled_inserts.load(Ordering::SeqCst) + } +} + /// Read from `stream` into `head` until it holds the first request line, which /// is returned. `None` if the connection closes first or sends no line. async fn read_request_line(stream: &mut TcpStream, head: &mut Vec) -> Option { @@ -577,6 +657,85 @@ async fn a_batch_whose_insert_failed_holds_the_commit_below_it_until_it_lands() ); } +/// How long an insert waits for `ClickHouse`'s answer in the stalled-insert test. +const SHORT_INSERT_TIMEOUT_MS: u64 = 2_000; + +/// How long the stalled-insert test waits for the loader to retry a stalled +/// insert: several insert bounds plus the hold's retry schedule. +const RETRY_SEEN: Duration = Duration::from_secs(45); + +#[tokio::test(flavor = "multi_thread")] +async fn an_insert_clickhouse_never_answers_times_out_and_is_held_until_it_lands() { + let infra = TestInfrastructure::new(test_name!(), true, true).await; + let clickhouse = clickhouse_address(&infra).await; + let bootstrap = kafka_bootstrap(&infra).await; + let direct = query_client(&clickhouse); + let proxy = StallingProxy::start(clickhouse.clone()).await; + + let landing = unique_table_name("kafka_stall_landing"); + direct + .execute(&format!( + "CREATE TABLE default.{landing} (id UInt64) ENGINE = MergeTree() ORDER BY id" + )) + .await + .expect("create table"); + let topic = format!("stall-{landing}"); + let group = format!("dfe-{landing}"); + ensure_topic(&bootstrap, &topic).await; + + let mut config = kafka_loader(&bootstrap, &topic, &group, proxy.address(), &landing); + config.clickhouse.insert_timeout_ms = SHORT_INSERT_TIMEOUT_MS; + config.clickhouse.request_timeout_ms = SHORT_INSERT_TIMEOUT_MS; + let (shutdown, loader) = start_loader(config); + let producer = make_producer(&bootstrap); + wait_for_loader(&producer, &topic, &clickhouse, &landing).await; + + // From here ClickHouse takes every request and answers none. + proxy.stall(); + let held = produce(&producer, &topic, r#"{"id":1}"#).await; + + // A second stalled insert means the first ended at the loader's bound and + // the held batch was sent again. + let deadline = tokio::time::Instant::now() + RETRY_SEEN; + while proxy.stalled_inserts() < 2 && tokio::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(250)).await; + } + let stalled_inserts = proxy.stalled_inserts(); + let while_held = highest_committed_over_watch(&bootstrap, &group, &topic).await; + + // Once ClickHouse answers again the held batch lands, and a record after it + // moves the watermark past both. + proxy.resume(); + let held_landed = wait_landed(&clickhouse, &landing, &BTreeSet::from([1]), RECOVERY).await; + let last = produce(&producer, &topic, r#"{"id":2}"#).await; + let after = wait_committed(&bootstrap, &group, &topic, last + 1).await; + + shutdown.cancel(); + let _ = tokio::time::timeout(Duration::from_secs(30), loader).await; + let _ = direct + .execute(&format!("DROP TABLE IF EXISTS default.{landing}")) + .await; + + assert!( + stalled_inserts >= 2, + "the loader sent {stalled_inserts} insert(s) into the stall in {RETRY_SEEN:?}: an \ + unanswered insert never reached its {SHORT_INSERT_TIMEOUT_MS} ms bound" + ); + assert!( + while_held.is_none_or(|next| next <= held), + "the group committed {while_held:?} while the batch holding offset {held} had not \ + landed: a crash then would have lost it" + ); + assert!( + held_landed.contains(&1), + "the timed-out batch never landed once ClickHouse answered again" + ); + assert!( + after >= Some(last + 1), + "the watermark never moved past the timed-out batch once it landed: {after:?}" + ); +} + #[tokio::test(flavor = "multi_thread")] async fn a_batch_held_for_a_dropped_table_follows_new_records_to_the_default_table() { let infra = TestInfrastructure::new(test_name!(), true, true).await; diff --git a/tests/integration/mod.rs b/tests/integration/mod.rs index de3acd8..7233b50 100644 --- a/tests/integration/mod.rs +++ b/tests/integration/mod.rs @@ -6,6 +6,8 @@ mod clickhouse; #[cfg(feature = "testcontainers")] mod clickhouse_inserter_e2e; +#[cfg(feature = "testcontainers")] +mod clickhouse_tls; mod coerce_integration; mod config; mod config_reachability;