feat(ilp): multi-host ws ingress + egress failover hardening

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

@@ -134,10 +134,8 @@ public abstract class WebSocketClient implements QuietCloseable {
     // setQwpRequestDurableAck) is the early-fail signal.
     private boolean serverDurableAckEnabled;
     private int serverQwpVersion = 1;
-    // Set during upgrade response validation from the X-QuestDB-Role header
-    // on every server response (101 success and 421 role-mismatch alike).
-    // Null when the server omitted the header (legacy build or non-QWP path).
-    private String serverRole;
+    private String upgradeRejectRole;
+    private int upgradeStatusCode;
     private boolean upgraded;
 
     public WebSocketClient(HttpClientConfiguration configuration, SocketFactory socketFactory) {
@@ -302,14 +300,19 @@ public int getServerQwpVersion() {
     }
 
     /**
-     * Returns the server's replication role as advertised in the
-     * {@code X-QuestDB-Role} header during the upgrade handshake, or
-     * {@code null} when the server omitted the header. Possible values:
-     * {@code STANDALONE}, {@code PRIMARY}, {@code REPLICA},
-     * {@code PRIMARY_CATCHUP}, {@code UNKNOWN}.
+     * Role from {@code X-QuestDB-Role} on the most recent rejected upgrade,
+     * or null when no such header was present.
      */
-    public String getServerRole() {
-        return serverRole;
+    public String getUpgradeRejectRole() {
+        return upgradeRejectRole;
+    }
+
+    /**
+     * HTTP status code from the most recent rejected upgrade, or 0 if no
+     * upgrade rejection has been observed yet.
+     */
+    public int getUpgradeStatusCode() {
+        return upgradeStatusCode;
     }
 
     /**
@@ -520,6 +523,8 @@ public void upgrade(CharSequence path, int timeout, CharSequence authorizationHe
         if (upgraded) {
             return; // Already upgraded
         }
+        upgradeRejectRole = null;
+        upgradeStatusCode = 0;
 
         // Generate random key
         byte[] keyBytes = new byte[16];
@@ -640,36 +645,40 @@ private static boolean extractDurableAckEnabled(String response) {
         return false;
     }
 
-    private static int parseStatusCode(String response) {
-        // "HTTP/1.1 NNN ..." — find the first space, then read three ASCII digits.
-        int sp = response.indexOf(' ');
-        if (sp < 0 || sp + 4 > response.length()) {
-            return WebSocketUpgradeException.STATUS_NONE;
+    private static int parseStatusCode(String statusLine) {
+        int sp1 = statusLine.indexOf(' ');
+        if (sp1 < 0 || sp1 + 4 > statusLine.length()) return 0;
+        if (sp1 + 4 < statusLine.length()) {
+            char afterCode = statusLine.charAt(sp1 + 4);
+            if (afterCode != ' ' && afterCode != '\r' && afterCode != '\n') {
+                return 0;
+            }
         }
         int code = 0;
-        for (int i = sp + 1; i < sp + 4; i++) {
-            char c = response.charAt(i);
-            if (c < '0' || c > '9') {
-                return WebSocketUpgradeException.STATUS_NONE;
-            }
+        for (int i = sp1 + 1; i < sp1 + 4; i++) {
+            char c = statusLine.charAt(i);
+            if (c < '0' || c > '9') return 0;
             code = code * 10 + (c - '0');
         }
         return code;
     }
 
-    private static String extractServerRole(String response) {
+    private static String extractRoleHeader(String response) {
         int headerLen = QUESTDB_ROLE_HEADER_NAME.length();
         int responseLen = response.length();
-        for (int i = 0; i <= responseLen - headerLen; i++) {
-            if (response.regionMatches(true, i, QUESTDB_ROLE_HEADER_NAME, 0, headerLen)) {
-                int valueStart = i + headerLen;
+        int lineStart = response.indexOf("\r\n");
+        while (lineStart >= 0 && lineStart + 2 + headerLen <= responseLen) {
+            int hStart = lineStart + 2;
+            if (response.regionMatches(true, hStart, QUESTDB_ROLE_HEADER_NAME, 0, headerLen)) {
+                int valueStart = hStart + headerLen;
                 int lineEnd = response.indexOf('\r', valueStart);
                 if (lineEnd < 0) {
                     lineEnd = responseLen;
                 }
                 String value = response.substring(valueStart, lineEnd).trim();
                 return value.isEmpty() ? null : value;
             }
+            lineStart = response.indexOf("\r\n", hStart);
         }
         return null;
     }
@@ -1078,22 +1087,16 @@ private void validateUpgradeResponse(int headerEnd) {
         }
         String response = new String(responseBytes, StandardCharsets.US_ASCII);
 
-        // Parse the X-QuestDB-Role header off any response (101 success or
-        // 421 role-mismatch), so callers that walk a multi-host list can
-        // log/diagnose which role the unsuitable node reported.
-        serverRole = extractServerRole(response);
-
-        // Check status line. Parse the numeric status code so callers can
-        // distinguish a transient role-mismatch (421) from a terminal upgrade
-        // rejection (401, 403, 426) without sniffing the message text.
         if (!response.startsWith("HTTP/1.1 101")) {
-            int lineEnd = response.indexOf('\r');
-            String statusLine = lineEnd < 0 ? response : response.substring(0, lineEnd);
-            int statusCode = parseStatusCode(response);
-            throw new WebSocketUpgradeException(
-                    statusCode,
-                    serverRole,
-                    "WebSocket upgrade failed: " + statusLine);
+            String statusLine = response.split("\r\n")[0];
+            upgradeStatusCode = parseStatusCode(statusLine);
+            if (upgradeStatusCode == 421) {
+                upgradeRejectRole = extractRoleHeader(response);
+            }
+            WebSocketUpgradeException ex = new WebSocketUpgradeException(
+                    upgradeStatusCode, upgradeRejectRole, "WebSocket upgrade failed: ");
+            ex.put(statusLine);
+            throw ex;
         }
 
         // Verify Upgrade: websocket (case-insensitive value per RFC 6455 Section 4.1)

core/src/main/java/io/questdb/client/cutlass/qwp/client/QueryEvent.java

@@ -44,6 +44,13 @@ public class QueryEvent {
      * having to reconstruct the classification from a side-channel latch.
      */
     public static final int KIND_TRANSPORT_ERROR = 4;
+    /**
+     * Permanent protocol-level disagreement (unsupported version, framing
+     * corruption that won't recover). {@code execute()} surfaces this directly
+     * instead of triggering failover -- version mismatch is cluster-wide and
+     * retrying against another endpoint masks the disagreement.
+     */
+    public static final int KIND_PROTOCOL_ERROR = 5;
 
     public QwpBatchBuffer buffer;     // valid for KIND_BATCH (must be released to pool by consumer)
     public String errorMessage;       // valid for KIND_ERROR
@@ -82,6 +89,14 @@ public QueryEvent asExecDone(short opType, long rowsAffected) {
         return this;
     }
 
+    public QueryEvent asProtocolError(byte status, String message) {
+        this.kind = KIND_PROTOCOL_ERROR;
+        this.buffer = null;
+        this.errorStatus = status;
+        this.errorMessage = message;
+        return this;
+    }
+
     public QueryEvent asTransportError(byte status, String message) {
         this.kind = KIND_TRANSPORT_ERROR;
         this.buffer = null;

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpAuthFailedException.java

@@ -0,0 +1,60 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client.cutlass.qwp.client;
+
+import io.questdb.client.cutlass.http.client.HttpClientException;
+
+/**
+ * WebSocket upgrade rejected with {@code 401} or {@code 403}. Terminal across all
+ * configured endpoints: a rejected credential is uniformly rejected across the
+ * cluster, so failing fast surfaces the configuration error immediately. Path
+ * mismatches ({@code 404}) are NOT routed through this exception because a single
+ * misconfigured node mid-deploy can return 404 while peers are healthy.
+ */
+public final class QwpAuthFailedException extends HttpClientException {
+    private final String host;
+    private final int port;
+    private final int statusCode;
+
+    public QwpAuthFailedException(int statusCode, String host, int port) {
+        super("WebSocket upgrade rejected with HTTP ");
+        put(statusCode).put(" for ").put(host).put(':').put(port);
+        this.statusCode = statusCode;
+        this.host = host;
+        this.port = port;
+    }
+
+    public String getHost() {
+        return host;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public int getStatusCode() {
+        return statusCode;
+    }
+}

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressIoThread.java

@@ -530,6 +530,12 @@ private void emitError(byte status, String message) {
      * latch -- the latch stays strictly for short-circuiting subsequent
      * {@code execute()} calls on a broken client.
      */
+    private void emitTerminalProtocolError(String message) {
+        notifyTerminalFailure(message, true);
+        events.offer(new QueryEvent().asProtocolError(WebSocketResponse.STATUS_INTERNAL_ERROR, message));
+        shutdown = true;
+    }
+
     private void emitTerminalTransportError(String message) {
         notifyTerminalFailure(message);
         events.offer(new QueryEvent().asTransportError(WebSocketResponse.STATUS_INTERNAL_ERROR, message));
@@ -594,15 +600,17 @@ private void handleResultBatch(long payloadPtr, int payloadLen) {
         // buffer) directly, skipping the previous per-batch memcpy into buf.scratchAddr.
         try {
             decoder.decode(buf, payloadPtr, payloadLen);
+        } catch (QwpProtocolVersionException e) {
+            if (!freeBuffers.offer(buf)) {
+                buf.close();
+            }
+            emitTerminalProtocolError(e.getMessage());
+            currentQueryDone = true;
+            return;
         } catch (QwpDecodeException e) {
-            // Same invariant as releaseBuffer: a slot is always free for a buf
-            // we took out of the pool moments ago. Close-on-failure is a
-            // defensive guard against future refactors breaking that invariant.
             if (!freeBuffers.offer(buf)) {
                 buf.close();
             }
-            // A decode failure leaves the client-side decoder out of step with
-            // the server's byte stream: the next frame cannot be trusted.
             emitTerminalTransportError("decode failure: " + e.getMessage());
             currentQueryDone = true;
             return;
@@ -643,9 +651,13 @@ private void handleResultBatch(long payloadPtr, int payloadLen) {
     }
 
     private void notifyTerminalFailure(String message) {
+        notifyTerminalFailure(message, false);
+    }
+
+    private void notifyTerminalFailure(String message, boolean isProtocol) {
         if (terminalFailureListener != null) {
             try {
-                terminalFailureListener.onTerminalFailure(WebSocketResponse.STATUS_INTERNAL_ERROR, message);
+                terminalFailureListener.onTerminalFailure(WebSocketResponse.STATUS_INTERNAL_ERROR, message, isProtocol);
             } catch (Throwable ignored) {
                 // Listener must not bring down the I/O thread. A first-failure-wins
                 // CAS in the listener cannot throw in practice; defensive anyway.
@@ -741,7 +753,7 @@ void closePool() {
 
     @FunctionalInterface
     public interface TerminalFailureListener {
-        void onTerminalFailure(byte status, String message);
+        void onTerminalFailure(byte status, String message, boolean isProtocol);
     }
 
     private static final class QueryRequest {