feat(facade): single cluster config for the QuestDB handle

Collapse the dual ingest/query config surface on the QuestDB facade into a
single configuration string for the whole cluster. A QuestDB cluster is one
logical target reached over QWP for both ingest and query, so one ws/wss
string -- listing every node in a single `addr` server list -- now drives
both the sender and query pools.

- QuestDBBuilder: drop ingestConfig()/queryConfig(); fromConfig() sets the
  one cluster config. Remove the cross-side pool-key conflict resolution
  (no two strings to reconcile) -- resolvePoolInt/Long read one ConfigView.
  build() validates the single string with both the ingest and egress
  validators; each side applies the keys it owns and ignores the rest.
- QuestDB: remove the connect(ingest, query) overload; connect(config) and
  builder() now document the one-config/server-list model.
- QuestDBImpl is unchanged: the builder passes the one config to both pool
  slots, preserving the white-box reflection seam.

Tests: TestWebSocketServer now serves both pools from one config like a real
node -- SERVER_INFO is emitted only on the egress /read path (the ingest
/write ACK stream would choke on it), plus a setRejectReadUpgrade() toggle to
fail just the query upgrade. Rewrote QuestDBBuilderTest and updated the
facade callback/recovery/write-only tests and the examples accordingly.

Author: bluestreak01

core/src/main/java/io/questdb/client/QuestDB.java

@@ -38,10 +38,10 @@
  * and the {@link Completion} associated with each query is a field on that
  * cached handle.
  * <p>
- * Configuration: use {@link #connect(CharSequence)} when the same address list
- * and credentials serve both ingest and egress -- the most common case.
- * Use {@link #connect(CharSequence, CharSequence)} or {@link #builder()} when
- * ingest and egress endpoints differ.
+ * Configuration: one {@code ws}/{@code wss} string describes the whole cluster
+ * (a single {@code addr} server list) and both the ingest and query pools
+ * connect across it. Use {@link #connect(CharSequence)} for the common case, or
+ * {@link #builder()} for pool sizing, the ingest callbacks, and write-only mode.
  * <p>
  * Thread safety: instances are safe to share. {@link #borrowSender()} and
  * {@link #query()} may be called concurrently from any thread; the pool
@@ -51,20 +51,21 @@ public interface QuestDB extends Closeable {
 
     /**
      * Builder for advanced configuration (pool sizes, acquisition timeouts,
-     * differing ingest/egress configs).
+     * ingest callbacks, write-only mode).
      */
     static QuestDBBuilder builder() {
         return new QuestDBBuilder();
     }
 
     /**
-     * Connects with a single configuration string used for both ingest and
-     * egress. The schema must be {@code ws} or {@code wss}: QuestDB ingests and
-     * queries over QWP (the QuestDB WebSocket protocol), so one string
-     * configures both clients.
+     * Connects with a single configuration string for the whole QuestDB cluster,
+     * used for both ingest and egress. The schema must be {@code ws} or
+     * {@code wss}: QuestDB ingests and queries over QWP (the QuestDB WebSocket
+     * protocol), so one string configures both clients. List every cluster node
+     * in a single {@code addr} server list and both pools connect across it.
      * <p>
-     * Use {@link #connect(CharSequence, CharSequence)} or {@link #builder()}
-     * when ingest and egress use different addresses or credentials.
+     * Use {@link #builder()} for pool sizing, the ingest callbacks, and
+     * write-only mode.
      *
      * @param configurationString a {@code ws}/{@code wss} config string (see
      *                            {@link Sender#fromConfig} or
@@ -75,22 +76,6 @@ static QuestDB connect(CharSequence configurationString) {
         return builder().fromConfig(configurationString).build();
     }
 
-    /**
-     * Connects with explicit ingest and egress configuration strings.
-     *
-     * @param ingestConfigurationString config for the {@link Sender} pool
-     *                                  ({@link Sender#fromConfig} format)
-     * @param queryConfigurationString  config for the query pool
-     *                                  ({@link io.questdb.client.cutlass.qwp.client.QwpQueryClient#fromConfig} format)
-     * @return a connected QuestDB handle
-     */
-    static QuestDB connect(CharSequence ingestConfigurationString, CharSequence queryConfigurationString) {
-        return builder()
-                .ingestConfig(ingestConfigurationString)
-                .queryConfig(queryConfigurationString)
-                .build();
-    }
-
     /**
      * Borrows a {@link Sender} from the pool. The caller MUST call
      * {@link Sender#close()} on the returned instance to release it back to

core/src/main/java/io/questdb/client/QuestDBBuilder.java

@@ -35,14 +35,15 @@
 
 /**
  * Builder for {@link QuestDB}. Most callers use {@link QuestDB#connect(CharSequence)};
- * this builder is for pool sizing, idle/lifetime knobs, acquire timeout,
- * and the case where ingest and egress configs differ.
+ * this builder adds pool sizing, idle/lifetime knobs, the acquire timeout, the
+ * ingest callbacks, and write-only mode.
  * <p>
- * Both configs must use the {@code ws} or {@code wss} schema (QWP over
- * WebSocket). A pool key (e.g. {@code sender_pool_min}) may be carried in the
- * connect string or set with an explicit builder call; an explicit call always
- * wins. When both connect strings carry the same pool key with different values,
- * {@link #build()} fails.
+ * One configuration string describes the whole QuestDB cluster (see
+ * {@link #fromConfig}): list every node in a single {@code addr} server list and
+ * both the ingest and query pools connect across it. The schema must be
+ * {@code ws} or {@code wss} (QWP over WebSocket). A pool key (e.g.
+ * {@code sender_pool_min}) may be carried in the connect string or set with an
+ * explicit builder call; an explicit call always wins.
  */
 public final class QuestDBBuilder {
 
@@ -64,10 +65,9 @@ public final class QuestDBBuilder {
     private SenderConnectionListener connectionListener;
     private SenderErrorHandler errorHandler;
     private long housekeeperIntervalMillis = UNSET;
+    private String config;
     private long idleTimeoutMillis = UNSET;
-    private String ingestConfig;
     private long maxLifetimeMillis = UNSET;
-    private String queryConfig;
     private int queryPoolMax = UNSET;
     private int queryPoolMin = UNSET;
     private int senderPoolMax = UNSET;
@@ -157,42 +157,36 @@ public QuestDBBuilder errorHandler(SenderErrorHandler handler) {
      * and is delivered once the server acks; until then it stays preserved.
      */
     public QuestDB build() {
-        if (ingestConfig == null) {
-            throw new IllegalStateException("ingest configuration is required; call fromConfig() or ingestConfig()");
+        if (config == null) {
+            throw new IllegalStateException("configuration is required; call fromConfig()");
         }
         if (writeOnly) {
             return buildWriteOnly();
         }
-        if (queryConfig == null) {
-            throw new IllegalStateException("query configuration is required; call fromConfig() or queryConfig()");
-        }
-        ConfigString ingestCs = ConfigString.parse(ingestConfig);
-        ConfigString queryCs = ConfigString.parse(queryConfig);
-        ConfigView ingestView = new ConfigView(ingestCs);
-        ConfigView queryView = new ConfigView(queryCs);
-        // Validate both connect strings exactly as the pools will, but without
-        // connecting. The ingest string runs the full Sender parse plus
-        // validateParameters -- ingress value keys are registry-STRING, so only
-        // the real parse validates their values. The egress string runs the
-        // typed validateConfig. A malformed config therefore fails here even
-        // when a pool min is 0 and nothing connects.
-        Sender.LineSenderBuilder.validateWsConfigString(ingestConfig);
-        QwpQueryClient.validateConfig(queryView, "wss".equals(queryCs.schema()));
-
-        // A view carries no side; getInt/getLong read any key, so the ingest
-        // and query views also serve the POOL reads.
-        resolvePoolInt(senderPoolMin, "sender_pool_min", ingestView, queryView, DEFAULT_POOL_MIN, this::senderPoolMin);
-        resolvePoolInt(senderPoolMax, "sender_pool_max", ingestView, queryView, DEFAULT_POOL_MAX, this::senderPoolMax);
-        resolvePoolInt(queryPoolMin, "query_pool_min", ingestView, queryView, DEFAULT_POOL_MIN, this::queryPoolMin);
-        resolvePoolInt(queryPoolMax, "query_pool_max", ingestView, queryView, DEFAULT_POOL_MAX, this::queryPoolMax);
-        resolvePoolLong(acquireTimeoutMillis, "acquire_timeout_ms", ingestView, queryView, DEFAULT_ACQUIRE_TIMEOUT_MILLIS, this::acquireTimeoutMillis);
-        resolvePoolLong(idleTimeoutMillis, "idle_timeout_ms", ingestView, queryView, DEFAULT_IDLE_TIMEOUT_MILLIS, this::idleTimeoutMillis);
-        resolvePoolLong(maxLifetimeMillis, "max_lifetime_ms", ingestView, queryView, DEFAULT_MAX_LIFETIME_MILLIS, this::maxLifetimeMillis);
-        resolvePoolLong(housekeeperIntervalMillis, "housekeeper_interval_ms", ingestView, queryView, DEFAULT_HOUSEKEEPER_INTERVAL_MILLIS, this::housekeeperIntervalMillis);
+        ConfigString cs = ConfigString.parse(config);
+        ConfigView view = new ConfigView(cs);
+        // Validate the single cluster config exactly as both pools will, but
+        // without connecting: the full Sender parse plus validateParameters
+        // (ingress value keys are registry-STRING, so only the real parse
+        // validates their values), then the typed egress validateConfig. Each
+        // side applies the keys it owns and silently ignores the rest, so one
+        // string drives both. A malformed config therefore fails here even when
+        // a pool min is 0 and nothing connects.
+        Sender.LineSenderBuilder.validateWsConfigString(config);
+        QwpQueryClient.validateConfig(view, "wss".equals(cs.schema()));
+
+        resolvePoolInt(senderPoolMin, "sender_pool_min", view, DEFAULT_POOL_MIN, this::senderPoolMin);
+        resolvePoolInt(senderPoolMax, "sender_pool_max", view, DEFAULT_POOL_MAX, this::senderPoolMax);
+        resolvePoolInt(queryPoolMin, "query_pool_min", view, DEFAULT_POOL_MIN, this::queryPoolMin);
+        resolvePoolInt(queryPoolMax, "query_pool_max", view, DEFAULT_POOL_MAX, this::queryPoolMax);
+        resolvePoolLong(acquireTimeoutMillis, "acquire_timeout_ms", view, DEFAULT_ACQUIRE_TIMEOUT_MILLIS, this::acquireTimeoutMillis);
+        resolvePoolLong(idleTimeoutMillis, "idle_timeout_ms", view, DEFAULT_IDLE_TIMEOUT_MILLIS, this::idleTimeoutMillis);
+        resolvePoolLong(maxLifetimeMillis, "max_lifetime_ms", view, DEFAULT_MAX_LIFETIME_MILLIS, this::maxLifetimeMillis);
+        resolvePoolLong(housekeeperIntervalMillis, "housekeeper_interval_ms", view, DEFAULT_HOUSEKEEPER_INTERVAL_MILLIS, this::housekeeperIntervalMillis);
 
         return new QuestDBImpl(
-                ingestConfig,
-                queryConfig,
+                config,
+                config,
                 senderPoolMin,
                 senderPoolMax,
                 queryPoolMin,
@@ -210,19 +204,17 @@ public QuestDB build() {
     // knobs from the ingest config alone and constructs a QuestDBImpl with no
     // query pool. Never touches the read side, so a down server cannot fail it.
     private QuestDB buildWriteOnly() {
-        ConfigString ingestCs = ConfigString.parse(ingestConfig);
-        ConfigView ingestView = new ConfigView(ingestCs);
-        Sender.LineSenderBuilder.validateWsConfigString(ingestConfig);
-        // Only the ingest view applies; pass it for both sides so the existing
-        // resolvers (which cross-check ingest vs query) read a single source.
-        resolvePoolInt(senderPoolMin, "sender_pool_min", ingestView, ingestView, DEFAULT_POOL_MIN, this::senderPoolMin);
-        resolvePoolInt(senderPoolMax, "sender_pool_max", ingestView, ingestView, DEFAULT_POOL_MAX, this::senderPoolMax);
-        resolvePoolLong(acquireTimeoutMillis, "acquire_timeout_ms", ingestView, ingestView, DEFAULT_ACQUIRE_TIMEOUT_MILLIS, this::acquireTimeoutMillis);
-        resolvePoolLong(idleTimeoutMillis, "idle_timeout_ms", ingestView, ingestView, DEFAULT_IDLE_TIMEOUT_MILLIS, this::idleTimeoutMillis);
-        resolvePoolLong(maxLifetimeMillis, "max_lifetime_ms", ingestView, ingestView, DEFAULT_MAX_LIFETIME_MILLIS, this::maxLifetimeMillis);
-        resolvePoolLong(housekeeperIntervalMillis, "housekeeper_interval_ms", ingestView, ingestView, DEFAULT_HOUSEKEEPER_INTERVAL_MILLIS, this::housekeeperIntervalMillis);
+        ConfigString cs = ConfigString.parse(config);
+        ConfigView view = new ConfigView(cs);
+        Sender.LineSenderBuilder.validateWsConfigString(config);
+        resolvePoolInt(senderPoolMin, "sender_pool_min", view, DEFAULT_POOL_MIN, this::senderPoolMin);
+        resolvePoolInt(senderPoolMax, "sender_pool_max", view, DEFAULT_POOL_MAX, this::senderPoolMax);
+        resolvePoolLong(acquireTimeoutMillis, "acquire_timeout_ms", view, DEFAULT_ACQUIRE_TIMEOUT_MILLIS, this::acquireTimeoutMillis);
+        resolvePoolLong(idleTimeoutMillis, "idle_timeout_ms", view, DEFAULT_IDLE_TIMEOUT_MILLIS, this::idleTimeoutMillis);
+        resolvePoolLong(maxLifetimeMillis, "max_lifetime_ms", view, DEFAULT_MAX_LIFETIME_MILLIS, this::maxLifetimeMillis);
+        resolvePoolLong(housekeeperIntervalMillis, "housekeeper_interval_ms", view, DEFAULT_HOUSEKEEPER_INTERVAL_MILLIS, this::housekeeperIntervalMillis);
         return new QuestDBImpl(
-                ingestConfig,
+                config,
                 senderPoolMin,
                 senderPoolMax,
                 acquireTimeoutMillis,
@@ -235,14 +227,15 @@ private QuestDB buildWriteOnly() {
     }
 
     /**
-     * Sets a single configuration string used for both ingest and egress. The
-     * schema must be {@code ws} or {@code wss}.
+     * Sets the single configuration string for the whole QuestDB cluster --
+     * used for both ingest and egress. List every cluster node in one
+     * {@code addr} (comma-separated, or by repeating the key); the ingest and
+     * query pools each connect across that one server list. The schema must be
+     * {@code ws} or {@code wss}.
      */
     public QuestDBBuilder fromConfig(CharSequence configurationString) {
-        requireWebSocketSchema(configurationString, "connection");
-        String s = configurationString.toString();
-        this.ingestConfig = s;
-        this.queryConfig = s;
+        requireWebSocketSchema(configurationString, "cluster");
+        this.config = configurationString.toString();
         return this;
     }
 
@@ -272,16 +265,6 @@ public QuestDBBuilder idleTimeoutMillis(long millis) {
         return this;
     }
 
-    /**
-     * Sets the ingest-side configuration. The schema must be {@code ws} or
-     * {@code wss}.
-     */
-    public QuestDBBuilder ingestConfig(CharSequence configurationString) {
-        requireWebSocketSchema(configurationString, "ingest");
-        this.ingestConfig = configurationString.toString();
-        return this;
-    }
-
     /**
      * Maximum age of a pooled connection before the housekeeper recycles it
      * (next time it is idle). Useful for picking up DNS / load-balancer
@@ -295,16 +278,6 @@ public QuestDBBuilder maxLifetimeMillis(long millis) {
         return this;
     }
 
-    /**
-     * Sets the query-side configuration. The schema must be {@code ws} or
-     * {@code wss}.
-     */
-    public QuestDBBuilder queryConfig(CharSequence configurationString) {
-        requireWebSocketSchema(configurationString, "query");
-        this.queryConfig = configurationString.toString();
-        return this;
-    }
-
     /**
      * Maximum query-pool size. Defaults to 4.
      */
@@ -406,53 +379,17 @@ private static void requireWebSocketSchema(CharSequence config, String role) {
         }
     }
 
-    private void resolvePoolInt(int current, String key, ConfigView ingest, ConfigView query, int dflt, IntConsumer setter) {
+    private void resolvePoolInt(int current, String key, ConfigView view, int dflt, IntConsumer setter) {
         if (current != UNSET) {
-            return; // explicit builder call wins; skip the conflict check
-        }
-        boolean inIngest = ingest.has(key);
-        boolean inQuery = query.has(key);
-        int value;
-        if (inIngest && inQuery) {
-            int vi = ingest.getInt(key, UNSET);
-            int vq = query.getInt(key, UNSET);
-            if (vi != vq) {
-                throw new IllegalArgumentException(
-                        "conflicting pool config: " + key + " (ingest=" + vi + ", query=" + vq + ")");
-            }
-            value = vi;
-        } else if (inIngest) {
-            value = ingest.getInt(key, UNSET);
-        } else if (inQuery) {
-            value = query.getInt(key, UNSET);
-        } else {
-            value = dflt;
+            return; // explicit builder call wins
         }
-        setter.accept(value);
+        setter.accept(view.has(key) ? view.getInt(key, UNSET) : dflt);
     }
 
-    private void resolvePoolLong(long current, String key, ConfigView ingest, ConfigView query, long dflt, LongConsumer setter) {
+    private void resolvePoolLong(long current, String key, ConfigView view, long dflt, LongConsumer setter) {
         if (current != UNSET) {
-            return; // explicit builder call wins; skip the conflict check
-        }
-        boolean inIngest = ingest.has(key);
-        boolean inQuery = query.has(key);
-        long value;
-        if (inIngest && inQuery) {
-            long vi = ingest.getLong(key, UNSET);
-            long vq = query.getLong(key, UNSET);
-            if (vi != vq) {
-                throw new IllegalArgumentException(
-                        "conflicting pool config: " + key + " (ingest=" + vi + ", query=" + vq + ")");
-            }
-            value = vi;
-        } else if (inIngest) {
-            value = ingest.getLong(key, UNSET);
-        } else if (inQuery) {
-            value = query.getLong(key, UNSET);
-        } else {
-            value = dflt;
+            return; // explicit builder call wins
         }
-        setter.accept(value);
+        setter.accept(view.has(key) ? view.getLong(key, UNSET) : dflt);
     }
 }