Latch terminal failures on QwpQueryClient

Transport- or protocol-level faults detected by the egress I/O thread
(server close, truncated/unknown frames, decode failures, send/recv
exceptions) now latch a sticky terminal state on the client. The next
QwpQueryClient.execute() short-circuits via handler.onError with the
stored status and message instead of dispatching another query to the
broken connection.

QwpEgressIoThread gains a TerminalFailureListener that the client wires
to a first-failure-wins CAS (AtomicReference). The emitTerminalError
helper calls the listener before offering the per-query QueryEvent so
the sender-thread latch is visible before the user's handler reacts.

Per-query QUERY_ERROR responses are explicitly NOT terminal: the
connection stays healthy and the next execute() runs normally. The
same applies to user-driven shutdown paths, which leave the latch
clear so close() followed by connect() on a new client starts fresh.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bluestreak01

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

@@ -81,6 +81,11 @@ public class QwpEgressIoThread implements Runnable, WebSocketFrameHandler {
     // Single-slot request queue (Phase-1 allows one in-flight query).
     private final BlockingQueue<QueryRequest> requests = new ArrayBlockingQueue<>(1);
     private final NativeBufferWriter sendScratch = new NativeBufferWriter();
+    // Notified once when the I/O thread detects a transport- or protocol-level
+    // terminal failure (onClose, truncated/unknown frames, send/receive
+    // throwing). Distinct from per-query QUERY_ERROR, which leaves the
+    // connection healthy and is delivered only through the events queue.
+    private final TerminalFailureListener terminalFailureListener;
     private final WebSocketClient wsClient;
     // Per-query credit state (accessed only from the I/O thread).
     // creditEnabled == (initialCredit > 0); controls whether we emit CREDIT
@@ -98,7 +103,12 @@ public class QwpEgressIoThread implements Runnable, WebSocketFrameHandler {
     private volatile boolean shutdown;
 
     public QwpEgressIoThread(WebSocketClient wsClient, int bufferPoolSize) {
+        this(wsClient, bufferPoolSize, null);
+    }
+
+    public QwpEgressIoThread(WebSocketClient wsClient, int bufferPoolSize, TerminalFailureListener terminalFailureListener) {
         this.wsClient = wsClient;
+        this.terminalFailureListener = terminalFailureListener;
         this.freeBuffers = new ArrayBlockingQueue<>(bufferPoolSize);
         // +2 reserves slots for a trailing RESULT_END and a synthetic error
         // frame the I/O thread may emit on shutdown, so a full buffer pool
@@ -153,7 +163,7 @@ public static QueryEvent decodeError(long payload, int payloadLen) {
     @Override
     public void onBinaryMessage(long payloadPtr, int payloadLen) {
         if (payloadLen < QwpConstants.HEADER_SIZE + 1) {
-            emitError((byte) 0, "server sent truncated frame");
+            emitTerminalError((byte) 0, "server sent truncated frame");
             // Stop the receive loop; the framing is broken and any further bytes
             // would be misinterpreted relative to the expected message boundary.
             currentQueryDone = true;
@@ -173,13 +183,14 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
             decodeAndEmitError(payloadPtr, payloadLen);
             currentQueryDone = true;
         } else {
-            emitError((byte) 0, "unknown msg_kind 0x" + Integer.toHexString(msgKind & 0xFF));
+            emitTerminalError((byte) 0, "unknown msg_kind 0x" + Integer.toHexString(msgKind & 0xFF));
             currentQueryDone = true;
         }
     }
 
     @Override
     public void onClose(int code, String reason) {
+        notifyTerminalFailure((byte) 0, "server closed connection: code=" + code + " reason=" + reason);
         if (!currentQueryDone) {
             emitError((byte) 0, "server closed connection: code=" + code + " reason=" + reason);
             currentQueryDone = true;
@@ -244,16 +255,22 @@ public void run() {
                 }
             }
         } catch (Throwable t) {
+            notifyTerminalFailure((byte) 0, "I/O thread failure: " + t.getMessage());
             emitErrorBlocking((byte) 0, "I/O thread failure: " + t.getMessage());
         } finally {
             // Wake any user thread blocked on events.take(). Without this, a close()
             // (or any abnormal exit) while a user thread is mid-execute() would let
             // takeEvent() block forever -- once the I/O thread is gone, no further
             // events arrive on the queue.
             if (!currentQueryDone) {
-                emitErrorBlocking((byte) 0, shutdown
+                String msg = shutdown
                         ? "I/O thread shut down with query in flight"
-                        : "I/O thread terminated with query in flight");
+                        : "I/O thread terminated with query in flight";
+                if (!shutdown) {
+                    // Shutdown is a user-driven close; don't mark the client terminal for it.
+                    notifyTerminalFailure((byte) 0, msg);
+                }
+                emitErrorBlocking((byte) 0, msg);
                 currentQueryDone = true;
             }
         }
@@ -297,7 +314,7 @@ private void decodeAndEmitExecDone(long payload, int payloadLen) {
         long p = payload + QwpConstants.HEADER_SIZE + 1 + 8;
         long limit = payload + payloadLen;
         if (p + 1 > limit) {
-            emitError((byte) 0, "EXEC_DONE frame truncated before op_type");
+            emitTerminalError((byte) 0, "EXEC_DONE frame truncated before op_type");
             return;
         }
         byte opType = Unsafe.getUnsafe().getByte(p++);
@@ -309,7 +326,7 @@ private void decodeAndEmitExecDone(long payload, int payloadLen) {
             if ((b & 0x80) == 0) break;
             shift += 7;
             if (shift > 63) {
-                emitError((byte) 0, "EXEC_DONE rows_affected varint overflow");
+                emitTerminalError((byte) 0, "EXEC_DONE rows_affected varint overflow");
                 return;
             }
         }
@@ -390,6 +407,15 @@ private void emitErrorBlocking(byte status, String message) {
         }
     }
 
+    // Emits a per-query error event AND latches the client terminal-failure
+    // state. Use for transport- or protocol-level faults where the WebSocket
+    // is no longer usable; per-query QUERY_ERROR must still go through
+    // plain emitError, since the connection remains healthy after it.
+    private void emitTerminalError(byte status, String message) {
+        notifyTerminalFailure(status, message);
+        emitError(status, message);
+    }
+
     private void handleResultBatch(long payloadPtr, int payloadLen) {
         QwpBatchBuffer buf;
         try {
@@ -413,7 +439,9 @@ private void handleResultBatch(long payloadPtr, int payloadLen) {
             decoder.decode(buf, payloadPtr, payloadLen);
         } catch (QwpDecodeException e) {
             freeBuffers.offer(buf);
-            emitError((byte) 0, "decode failure: " + e.getMessage());
+            // A decode failure leaves the client-side decoder out of step with
+            // the server's byte stream: the next frame cannot be trusted.
+            emitTerminalError((byte) 0, "decode failure: " + e.getMessage());
             currentQueryDone = true;
             return;
         }
@@ -437,6 +465,17 @@ private void handleResultBatch(long payloadPtr, int payloadLen) {
         }
     }
 
+    private void notifyTerminalFailure(byte status, String message) {
+        if (terminalFailureListener != null) {
+            try {
+                terminalFailureListener.onTerminalFailure(status, message);
+            } 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.
+            }
+        }
+    }
+
     /**
      * Builds and transmits a CANCEL frame on the WebSocket. Wire format:
      * {@code msg_kind(0x14) + request_id(u64)}.
@@ -517,6 +556,11 @@ void closePool() {
         events.offer(new QueryEvent().asError((byte) 0, "QwpQueryClient closed"));
     }
 
+    @FunctionalInterface
+    public interface TerminalFailureListener {
+        void onTerminalFailure(byte status, String message);
+    }
+
     private static final class QueryRequest {
         final long initialCredit;
         final long requestId;

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

@@ -38,6 +38,7 @@
 
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * QWP egress (query results) client.
@@ -51,6 +52,15 @@
  * Thread safety: not thread-safe for concurrent queries on the same client.
  * One {@link #execute} at a time. Opening one client per query-issuing thread
  * is the recommended pattern.
+ * <p>
+ * Terminal-failure model: transport- or protocol-level faults detected by the
+ * I/O thread (server close, truncated/unknown frames, send/recv exceptions)
+ * latch a sticky terminal failure on the client. Subsequent {@link #execute}
+ * calls short-circuit via {@link QwpColumnBatchHandler#onError} with the
+ * stored status/message rather than dispatching to the now-broken connection.
+ * Per-query {@code QUERY_ERROR} responses are NOT terminal -- the connection
+ * remains usable for the next query. To retry after a terminal failure the
+ * caller must {@link #close} this client and open a fresh one.
  */
 public class QwpQueryClient implements QuietCloseable {
 
@@ -117,6 +127,11 @@ public class QwpQueryClient implements QuietCloseable {
     // hit the timeout branch in under a second instead of spending five.
     @SuppressWarnings("FieldMayBeFinal")
     private volatile long shutdownJoinMs = 5_000;
+    // Latched when the I/O thread reports a transport- or protocol-level
+    // failure (see QwpEgressIoThread notifyTerminalFailure callers). Checked
+    // on every user-facing method so a broken connection short-circuits
+    // instead of submitting work that will only collect an error event.
+    private final AtomicReference<TerminalFailure> terminalFailure = new AtomicReference<>();
     private boolean tlsEnabled;
     // Only meaningful when tlsEnabled. Default is full validation against the JVM's trust store.
     private int tlsValidationMode = ClientTlsConfiguration.TLS_VALIDATION_MODE_FULL;
@@ -478,7 +493,8 @@ public void connect() {
             probeZstdAvailable();
         }
 
-        ioThread = new QwpEgressIoThread(webSocketClient, bufferPoolSize);
+        terminalFailure.set(null);
+        ioThread = new QwpEgressIoThread(webSocketClient, bufferPoolSize, this::recordTerminalFailure);
         ioThreadHandle = new Thread(ioThread, "qwp-egress-io");
         ioThreadHandle.setDaemon(true);
         ioThreadHandle.start();
@@ -491,6 +507,11 @@ public void connect() {
      * Blocks the calling thread until the server sends {@code RESULT_END} or
      * {@code QUERY_ERROR}. While the user thread is inside {@code handler.onBatch},
      * the I/O thread keeps reading and decoding ahead up to the configured buffer-pool depth.
+     * <p>
+     * Short-circuits when a terminal failure has already been latched by the I/O
+     * thread (see class Javadoc): {@code handler.onError} fires immediately with
+     * the stored status and message, and no query is dispatched. The caller must
+     * {@link #close} the client and open a new one to retry.
      */
     public void execute(String sql, QwpColumnBatchHandler handler) {
         if (!connected) {
@@ -506,6 +527,16 @@ public void execute(String sql, QwpColumnBatchHandler handler) {
             handler.onError((byte) 0, "QwpQueryClient is closed");
             return;
         }
+        TerminalFailure tf = terminalFailure.get();
+        if (tf != null) {
+            // I/O thread has reported a transport- or protocol-level failure.
+            // Submitting another query would only accumulate work the I/O thread
+            // cannot process. Surface the stored failure via the handler
+            // immediately -- users who want to retry must close() and create a
+            // fresh client.
+            handler.onError(tf.status, tf.message);
+            return;
+        }
         long requestId = nextRequestId++;
         currentRequestId = requestId;
         try {
@@ -731,6 +762,10 @@ public QwpQueryClient withTrustStore(String trustStorePath, char[] trustStorePas
         return this;
     }
 
+    void recordTerminalFailure(byte status, String message) {
+        terminalFailure.compareAndSet(null, new TerminalFailure(status, message));
+    }
+
     private static String defaultClientId() {
         return "questdb-java-egress/1.0.0";
     }
@@ -789,4 +824,14 @@ private void probeZstdAvailable() {
         }
         Zstd.freeDCtx(dctx);
     }
+
+    private static final class TerminalFailure {
+        final String message;
+        final byte status;
+
+        TerminalFailure(byte status, String message) {
+            this.status = status;
+            this.message = message;
+        }
+    }
 }