Reconnect stale pooled connections via ping_before_query

README.md

@@ -101,6 +101,9 @@ client.close
 | `password:`   | `""`          |                                                     |
 | `compression:`| `:none`       | `:none`, `:lz4`, or `:zstd`                         |
 | `logger:`     | `nil`         | any `Logger`-compatible object (see [Logging](#logging)) |
+| `ping_before_query:` | `true` | ping and transparently reconnect a dead socket before each query (see [Connection resilience](#connection-resilience)) |
+| `tcp_keepalive:` | `true`     | enable OS-level TCP keepalive on the socket          |
+| `retry_timeout:` | `1`        | seconds to back off before each reconnect attempt    |
 
 `Pool.new` additionally accepts:
 
@@ -110,6 +113,20 @@ client.close
 | `pool_timeout:` | `5`     | seconds                                                  |
 | `settings:`     | `{}`    | session settings applied to every client in the pool (see [Session settings](#session-settings)) |
 
+## Connection resilience
+
+Long-lived pooled connections get closed out from under you — ClickHouse's `idle_connection_timeout`, a load balancer recycling idle sockets, or a server restart. The next use of such a socket would otherwise fail with `ConnectionError: closed: ...` (the errno in the message is whatever stale value was left in the thread — `Success`, `Operation now in progress`, etc. — and is not meaningful).
+
+By default the driver defends against this:
+
+- **`ping_before_query`** sends a lightweight ping before each query. If the socket is dead, the driver reconnects (re-running the handshake) and retries the ping before the real query runs, so a stale connection is repaired transparently instead of surfacing an error.
+- **`tcp_keepalive`** enables OS-level TCP keepalive so idle sockets are kept warm and dead peers are detected sooner.
+- **`retry_timeout`** bounds the back-off (in seconds) before each reconnect attempt.
+
+`Pool#with` additionally retries once on `ConnectionError` as a backstop for the residual race where a socket dies between the ping and the query.
+
+The per-query ping adds one round-trip per query. Set `ping_before_query: false` to opt out and manage staleness yourself (e.g. via `#reset_connection`).
+
 ## API
 
 ### `#execute(sql, settings: {})`
@@ -165,7 +182,7 @@ client.execute("OPTIMIZE TABLE events FINAL", settings: { mutations_sync: 2 })
 
 Keys may be Symbols or Strings. Values go through `#to_s` on the wire; `true` / `false` render as `"1"` / `"0"`. The same setting is also available pool-wide via `Pool.new(settings: ...)` for session-default values — per-call `settings:` is the right tool when only a specific query needs the override.
 
-`#insert` does not accept a `settings:` argument: clickhouse-cpp's block-insert API has no hook for per-request settings. For insert-time tuning, set the setting in the pool's session defaults, or use `#execute` to run an explicit `INSERT ... SETTINGS k=v VALUES ...` statement.
+`#insert` does not accept a per-call `settings:` argument, but a pool's `Pool.new(settings: ...)` defaults *are* carried into inserts (they're applied to the generated INSERT query), so insert-time tuning like `insert_quorum` or `max_insert_threads` belongs there. For a one-off override, use `#execute` to run an explicit `INSERT ... SETTINGS k=v VALUES ...` statement.
 
 ### `#insert(table, rows, columns: nil, db_name: nil, types: nil)`
 
@@ -247,7 +264,7 @@ end
 
 ### Session settings
 
-Pass `settings:` to `Pool.new` to apply ClickHouse session settings to every client the pool creates. Each checked-out connection starts from the same session state — equivalent to sending `?key=value` URL params on every HTTP request.
+Pass `settings:` to `Pool.new` to apply ClickHouse settings to every query the pool runs — equivalent to sending `?key=value` URL params on every HTTP request. They ride each query as per-query settings rather than a one-time session `SET`, so they survive a transparent [reconnect](#connection-resilience) instead of reverting to server defaults on the new socket.
 
 ```ruby
 pool = ClickhouseNative::Pool.new(
@@ -260,7 +277,7 @@ pool = ClickhouseNative::Pool.new(
 )
 ```
 
-Integer and Float values render bare (`SET allow_experimental_analyzer = 1`); anything else is quoted as a SQL string literal.
+An unknown setting name is rejected by the server (the settings are sent as important), so a typo fails loudly rather than being silently ignored. A per-call `settings:` on an individual `#query` / `#execute` overrides the pool default for that query.
 
 ## Type mapping
 

ext/clickhouse_native/client.cpp

@@ -17,11 +17,13 @@
 #include <clickhouse/exceptions.h>
 #include <clickhouse/types/types.h>
 
+#include <chrono>
 #include <cstdint>
 #include <exception>
 #include <memory>
 #include <string>
 #include <system_error>
+#include <utility>
 #include <vector>
 
 using namespace clickhouse;
@@ -666,6 +668,11 @@ static void append_value(const ColumnRef& col, VALUE value) {
 
 struct CHClient {
     std::unique_ptr<Client> client;
+    // Session settings applied to *every* query as per-query settings, rather
+    // than a one-time session `SET`. Per-query settings ride each query packet,
+    // so they survive a transparent ping_before_query reconnect (a fresh socket
+    // would otherwise lose a session-level SET and fall back to server defaults).
+    std::vector<std::pair<std::string, std::string>> default_settings;
 };
 
 static void ch_client_free(void* p) {
@@ -707,6 +714,18 @@ static uint16_t kwarg_uint16(VALUE kwargs, const char* key, uint16_t fallback) {
     return static_cast<uint16_t>(NUM2UINT(v));
 }
 
+static bool kwarg_bool(VALUE kwargs, const char* key, bool fallback) {
+    VALUE v = rb_hash_lookup2(kwargs, ID2SYM(rb_intern(key)), Qundef);
+    if (v == Qundef) return fallback;
+    return RTEST(v);
+}
+
+static unsigned int kwarg_uint(VALUE kwargs, const char* key, unsigned int fallback) {
+    VALUE v = rb_hash_lookup2(kwargs, ID2SYM(rb_intern(key)), Qundef);
+    if (v == Qundef || NIL_P(v)) return fallback;
+    return NUM2UINT(v);
+}
+
 static CompressionMethod kwarg_compression(VALUE kwargs) {
     VALUE v = rb_hash_lookup2(kwargs, ID2SYM(rb_intern("compression")), Qundef);
     if (v == Qundef || NIL_P(v)) return CompressionMethod::None;
@@ -742,6 +761,16 @@ static int apply_settings_cb(VALUE key, VALUE val, VALUE arg) {
     return ST_CONTINUE;
 }
 
+// Collect a `settings:` Hash into a client's default_settings vector at
+// construction. Values are stringified the same way per-query settings are.
+static int collect_setting_cb(VALUE key, VALUE val, VALUE arg) {
+    auto* vec = reinterpret_cast<std::vector<std::pair<std::string, std::string>>*>(arg);
+    VALUE k = SYMBOL_P(key) ? rb_sym2str(key) : key;
+    StringValue(k);
+    vec->emplace_back(std::string(RSTRING_PTR(k), RSTRING_LEN(k)), stringify_setting_value(val));
+    return ST_CONTINUE;
+}
+
 // Read a `settings:` Hash out of the parsed kwargs and stamp each entry
 // onto `q` as a per-query setting. No-op if kwargs is nil or settings is
 // missing/empty. Raises TypeError if settings is not a Hash.
@@ -765,6 +794,16 @@ static void apply_read_settings(Query& q, VALUE kwargs) {
     apply_settings(q, kwargs);
 }
 
+// Stamp the client's construction-time `settings:` onto `q`. Marked important
+// (flag 1) so an unknown setting name errors — matching the old session `SET`,
+// which failed loudly instead of silently ignoring a typo'd setting. Applied
+// before per-call settings so an explicit `settings:` on the call still wins.
+static void apply_default_settings(Query& q, CHClient* c) {
+    for (const auto& kv : c->default_settings) {
+        q.SetSetting(kv.first, QuerySettingsField{kv.second, 1});
+    }
+}
+
 // Client.new(host:, port:, database:, user:, password:)
 static VALUE ch_client_initialize(int argc, VALUE* argv, VALUE self) {
     VALUE kwargs = Qnil;
@@ -778,12 +817,35 @@ static VALUE ch_client_initialize(int argc, VALUE* argv, VALUE self) {
     std::string password = kwarg_str(kwargs, "password", "");
     CompressionMethod compression = kwarg_compression(kwargs);
 
+    // Connection-resilience defaults. Long-lived pooled connections get
+    // silently closed by the server (idle_connection_timeout) or an LB, so
+    // the next use of a checked-out client would otherwise hit recv()==0 and
+    // raise ConnectionError("closed: ..."). ping_before_query makes the driver
+    // ping first and transparently reconnect on a dead socket; tcp_keepalive
+    // keeps idle sockets healthy at the OS level. retry_timeout is the backoff
+    // before each reconnect attempt — kept low (1s) since a pooled reconnect
+    // to a live server should be quick and we don't want to stall queries.
+    bool ping_before_query = kwarg_bool(kwargs, "ping_before_query", true);
+    bool tcp_keepalive = kwarg_bool(kwargs, "tcp_keepalive", true);
+    unsigned int retry_timeout = kwarg_uint(kwargs, "retry_timeout", 1);
+
     CHClient* c = as_client(self);
+
+    VALUE settings = rb_hash_lookup2(kwargs, ID2SYM(rb_intern("settings")), Qnil);
+    if (!NIL_P(settings)) {
+        Check_Type(settings, T_HASH);
+        rb_hash_foreach(settings, collect_setting_cb,
+                        reinterpret_cast<VALUE>(&c->default_settings));
+    }
+
     try {
         ClientOptions opts;
         opts.SetHost(host).SetPort(port)
             .SetDefaultDatabase(database).SetUser(user).SetPassword(password)
-            .SetCompressionMethod(compression);
+            .SetCompressionMethod(compression)
+            .SetPingBeforeQuery(ping_before_query)
+            .TcpKeepAlive(tcp_keepalive)
+            .SetRetryTimeout(std::chrono::seconds(retry_timeout));
         c->client = std::make_unique<Client>(opts);
     } catch (const std::exception& e) {
         raise_mapped_ex(e);
@@ -792,6 +854,9 @@ static VALUE ch_client_initialize(int argc, VALUE* argv, VALUE self) {
     rb_ivar_set(self, rb_intern("@host"), rb_utf8_str_new(host.data(), host.size()));
     rb_ivar_set(self, rb_intern("@port"), UINT2NUM(port));
     rb_ivar_set(self, rb_intern("@database"), rb_utf8_str_new(database.data(), database.size()));
+    rb_ivar_set(self, rb_intern("@ping_before_query"), ping_before_query ? Qtrue : Qfalse);
+    rb_ivar_set(self, rb_intern("@tcp_keepalive"), tcp_keepalive ? Qtrue : Qfalse);
+    rb_ivar_set(self, rb_intern("@retry_timeout"), UINT2NUM(retry_timeout));
 
     VALUE logger = rb_hash_lookup2(kwargs, ID2SYM(rb_intern("logger")), Qnil);
     rb_ivar_set(self, rb_intern("@logger"), logger);
@@ -835,6 +900,7 @@ static VALUE ch_client_execute(int argc, VALUE* argv, VALUE self) {
     if (!c->client) rb_raise(err_connection, "clickhouse-native: client is closed");
 
     Query q(std::string(RSTRING_PTR(rb_sql), RSTRING_LEN(rb_sql)));
+    apply_default_settings(q, c);
     apply_settings(q, kwargs);
 
     ExecuteNoGVL args{c->client.get(), &q, nullptr};
@@ -866,6 +932,7 @@ static VALUE ch_client_query(int argc, VALUE* argv, VALUE self) {
     try {
         std::vector<ID> col_ids;
         Query q(std::string(RSTRING_PTR(rb_sql), RSTRING_LEN(rb_sql)));
+        apply_default_settings(q, c);
         apply_read_settings(q, kwargs);
         q.OnData([&](const Block& block) {
             size_t ncols = block.GetColumnCount();
@@ -911,6 +978,7 @@ static VALUE ch_client_query_value(int argc, VALUE* argv, VALUE self) {
         VALUE out = Qnil;
         bool seen = false;
         Query q(std::string(RSTRING_PTR(rb_sql), RSTRING_LEN(rb_sql)));
+        apply_default_settings(q, c);
         apply_read_settings(q, kwargs);
         q.OnData([&](const Block& block) {
             if (seen) return;
@@ -936,14 +1004,15 @@ struct InsertNoGVL {
     Client* client;
     const std::string* table;
     const Block* block;
+    const std::vector<std::pair<std::string, std::string>>* settings;
     std::exception_ptr err;
 };
 }  // namespace
 
 static void* insert_no_gvl(void* data) {
     auto* a = static_cast<InsertNoGVL*>(data);
     try {
-        a->client->Insert(*a->table, *a->block);
+        a->client->Insert(*a->table, *a->block, *a->settings);
     } catch (...) {
         a->err = std::current_exception();
     }
@@ -1009,7 +1078,7 @@ static VALUE ch_client_insert_block(VALUE self, VALUE rb_table, VALUE rb_columns
             block.AppendColumn(names[i], cols[i]);
         }
 
-        InsertNoGVL args{c->client.get(), &table, &block, nullptr};
+        InsertNoGVL args{c->client.get(), &table, &block, &c->default_settings, nullptr};
         rb_thread_call_without_gvl(insert_no_gvl, &args, insert_unblock, &args);
         if (args.err) {
             try { c->client->ResetConnection(); } catch (...) {}
@@ -1110,6 +1179,7 @@ static VALUE ch_client_query_each(int argc, VALUE* argv, VALUE self) {
 
     QueryEachState state{rb_block_proc(), {}, 0, false};
     Query q(std::string(RSTRING_PTR(rb_sql), RSTRING_LEN(rb_sql)));
+    apply_default_settings(q, c);
     apply_read_settings(q, kwargs);
     q.OnDataCancelable([&state](const Block& block) -> bool {
         if (state.aborted) return false;

ext/clickhouse_native/patches/0002-carry-settings-into-insert.patch

@@ -0,0 +1,82 @@
+diff --git a/clickhouse/client.cpp b/clickhouse/client.cpp
+index 34a733d..c9732f4 100644
+--- a/clickhouse/client.cpp
++++ b/clickhouse/client.cpp
+@@ -211,7 +211,8 @@ public:
+
+     bool IsSelecting() const { return state_ == State::Selecting; }
+
+-    void Insert(const std::string& table_name, const std::string& query_id, const Block& block);
++    void Insert(const std::string& table_name, const std::string& query_id, const Block& block,
++                const std::vector<std::pair<std::string, std::string>>& settings = {});
+
+     Block BeginInsert(Query query);
+
+@@ -485,7 +486,8 @@ std::string NameToQueryString(const std::string &input)
+     return output;
+ }
+
+-void Client::Impl::Insert(const std::string& table_name, const std::string& query_id, const Block& block) {
++void Client::Impl::Insert(const std::string& table_name, const std::string& query_id, const Block& block,
++                          const std::vector<std::pair<std::string, std::string>>& settings) {
+     if (state_ == State::Inserting) {
+         throw ValidationError("cannot execute query while inserting, use SendInsertData instead");
+     }
+@@ -511,6 +513,9 @@ void Client::Impl::Insert(const std::string& table_name, const std::string& quer
+     }
+
+     Query query("INSERT INTO " + table_name + " ( " + fields_section.str() + " ) VALUES", query_id);
++    for (const auto& kv : settings) {
++        query.SetSetting(kv.first, QuerySettingsField{kv.second, 1});
++    }
+     SendQuery(query);
+
+     // Wait for a data packet and return
+@@ -1365,12 +1370,14 @@ bool Client::IsSelecting() const
+     return impl_->IsSelecting();
+ }
+
+-void Client::Insert(const std::string& table_name, const Block& block) {
+-    impl_->Insert(table_name, Query::default_query_id, block);
++void Client::Insert(const std::string& table_name, const Block& block,
++                    const std::vector<std::pair<std::string, std::string>>& settings) {
++    impl_->Insert(table_name, Query::default_query_id, block, settings);
+ }
+
+-void Client::Insert(const std::string& table_name, const std::string& query_id, const Block& block) {
+-    impl_->Insert(table_name, query_id, block);
++void Client::Insert(const std::string& table_name, const std::string& query_id, const Block& block,
++                    const std::vector<std::pair<std::string, std::string>>& settings) {
++    impl_->Insert(table_name, query_id, block, settings);
+ }
+
+ Block Client::BeginInsert(const std::string& query) {
+diff --git a/clickhouse/client.h b/clickhouse/client.h
+index a55fd74..6c6ff53 100644
+--- a/clickhouse/client.h
++++ b/clickhouse/client.h
+@@ -23,6 +23,8 @@
+ #include "columns/bool.h"
+
+ #include <chrono>
++#include <utility>
++#include <vector>
+ #include <cstdint>
+ #include <memory>
+ #include <ostream>
+@@ -298,8 +300,13 @@ public:
+     bool IsSelecting() const;
+
+     /// Intends for insert block of data into a table \p table_name.
+-    void Insert(const std::string& table_name, const Block& block);
+-    void Insert(const std::string& table_name, const std::string& query_id, const Block& block);
++    /// \p settings are applied to the generated INSERT query as per-query
++    /// settings (patched-in for clickhouse-native: lets a pooled client carry
++    /// its session settings into inserts without a session-level SET).
++    void Insert(const std::string& table_name, const Block& block,
++                const std::vector<std::pair<std::string, std::string>>& settings = {});
++    void Insert(const std::string& table_name, const std::string& query_id, const Block& block,
++                const std::vector<std::pair<std::string, std::string>>& settings = {});
+
+     /// Start an \p INSERT statement, insert batches of data, then finish the insert.
+     Block BeginInsert(const std::string& query);

lib/clickhouse_native/client.rb

@@ -2,7 +2,7 @@
 
 module ClickhouseNative
   class Client
-    attr_reader :host, :port, :database
+    attr_reader :host, :port, :database, :ping_before_query, :tcp_keepalive, :retry_timeout
 
     def describe_table(table, db_name: nil)
       fq = db_name ? "#{db_name}.#{table}" : table