read failover

Author: bluestreak01

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

@@ -36,6 +36,14 @@ public class QueryEvent {
     public static final int KIND_END = 1;
     public static final int KIND_ERROR = 2;
     public static final int KIND_EXEC_DONE = 3;
+    /**
+     * Synthesised on the I/O thread when the server closes the socket,
+     * receiveFrame / decode raises, or the I/O thread abnormally exits.
+     * Distinct from {@link #KIND_ERROR} (server-emitted {@code QUERY_ERROR})
+     * so {@code execute()} can decide whether to trigger failover without
+     * having to reconstruct the classification from a side-channel latch.
+     */
+    public static final int KIND_TRANSPORT_ERROR = 4;
 
     public QwpBatchBuffer buffer;     // valid for KIND_BATCH (must be released to pool by consumer)
     public byte errorStatus;          // valid for KIND_ERROR
@@ -73,4 +81,12 @@ public QueryEvent asExecDone(short opType, long rowsAffected) {
         this.rowsAffected = rowsAffected;
         return this;
     }
+
+    public QueryEvent asTransportError(byte status, String message) {
+        this.kind = KIND_TRANSPORT_ERROR;
+        this.buffer = null;
+        this.errorStatus = status;
+        this.errorMessage = message;
+        return this;
+    }
 }

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

@@ -66,6 +66,26 @@ public interface QwpColumnBatchHandler {
      */
     void onError(byte status, String message);
 
+    /**
+     * Invoked when {@link QwpQueryClient#execute} has transparently reconnected
+     * to another endpoint after a transport failure and is about to re-submit
+     * the query (with {@code failover=on}, the default).
+     * <p>
+     * {@code newNode} is the {@link QwpServerInfo} of the endpoint the client
+     * just bound to, or {@code null} if the new server negotiated the v1
+     * protocol (no SERVER_INFO frame).
+     * <p>
+     * After this callback fires, {@link #onBatch} will be invoked again with
+     * {@code batch_seq} restarting at 0 on the new connection. Handlers that
+     * accumulate rows across batches should discard whatever they built up
+     * from the previous attempt. The default implementation is a no-op, safe
+     * for handlers that don't care (for example, simple row-count aggregators
+     * that are idempotent against replays) and for applications that set
+     * {@code failover=off}.
+     */
+    default void onFailoverReset(QwpServerInfo newNode) {
+    }
+
     /**
      * Invoked in place of {@link #onBatch} + {@link #onEnd} when the query was
      * a non-SELECT (DDL, INSERT, UPDATE, etc.). No batches are delivered for

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

@@ -163,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) {
-            emitTerminalError("server sent truncated frame");
+            emitTerminalTransportError("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;
@@ -183,16 +183,17 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
             decodeAndEmitError(payloadPtr, payloadLen);
             currentQueryDone = true;
         } else {
-            emitTerminalError("unknown msg_kind 0x" + Integer.toHexString(msgKind & 0xFF));
+            emitTerminalTransportError("unknown msg_kind 0x" + Integer.toHexString(msgKind & 0xFF));
             currentQueryDone = true;
         }
     }
 
     @Override
     public void onClose(int code, String reason) {
-        notifyTerminalFailure("server closed connection: code=" + code + " reason=" + reason);
+        String msg = "server closed connection: code=" + code + " reason=" + reason;
+        notifyTerminalFailure(msg);
         if (!currentQueryDone) {
-            emitError(WebSocketResponse.STATUS_INTERNAL_ERROR, "server closed connection: code=" + code + " reason=" + reason);
+            events.offer(new QueryEvent().asTransportError(WebSocketResponse.STATUS_INTERNAL_ERROR, msg));
             currentQueryDone = true;
         }
     }
@@ -261,8 +262,9 @@ public void run() {
                 }
             }
         } catch (Throwable t) {
-            notifyTerminalFailure("I/O thread failure: " + t.getMessage());
-            emitErrorBlocking("I/O thread failure: " + t.getMessage());
+            String msg = "I/O thread failure: " + t.getMessage();
+            notifyTerminalFailure(msg);
+            emitTransportErrorBlocking(WebSocketResponse.STATUS_INTERNAL_ERROR, msg);
         } 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
@@ -272,11 +274,15 @@ public void run() {
                 String msg = shutdown
                         ? "I/O thread shut down 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.
+                if (shutdown) {
+                    // User-driven close. Emit a plain query error (KIND_ERROR): the
+                    // user initiated the shutdown, we do not want failover to fire on
+                    // their behalf.
+                    emitError(WebSocketResponse.STATUS_INTERNAL_ERROR, msg);
+                } else {
                     notifyTerminalFailure(msg);
+                    emitTransportErrorBlocking(WebSocketResponse.STATUS_INTERNAL_ERROR, msg);
                 }
-                emitErrorBlocking(msg);
                 currentQueryDone = true;
             }
         }
@@ -335,7 +341,7 @@ private void decodeAndEmitExecDone(long payload, int payloadLen) {
         long p = payload + QwpConstants.HEADER_SIZE + 1 + 8;
         long limit = payload + payloadLen;
         if (p + 1 > limit) {
-            emitTerminalError("EXEC_DONE frame truncated before op_type");
+            emitTerminalTransportError("EXEC_DONE frame truncated before op_type");
             return;
         }
         byte opType = Unsafe.getUnsafe().getByte(p++);
@@ -347,7 +353,7 @@ private void decodeAndEmitExecDone(long payload, int payloadLen) {
             if ((b & 0x80) == 0) break;
             shift += 7;
             if (shift > 63) {
-                emitTerminalError("EXEC_DONE rows_affected varint overflow");
+                emitTerminalTransportError("EXEC_DONE rows_affected varint overflow");
                 return;
             }
         }
@@ -408,12 +414,13 @@ private void emitError(byte status, String message) {
     }
 
     /**
-     * Like {@link #emitError} but retries until the event is enqueued. Used on
-     * shutdown and fatal-error paths where dropping the event would leave the
-     * user thread blocked on {@link #takeEvent} indefinitely.
+     * Like {@link #emitError} but emits a {@code KIND_TRANSPORT_ERROR} event
+     * rather than {@code KIND_ERROR}, and retries until the event is enqueued.
+     * Used on transport-failure paths where the WebSocket is torn down and
+     * the user thread needs to be woken so failover can kick in.
      */
-    private void emitErrorBlocking(String message) {
-        QueryEvent ev = new QueryEvent().asError(WebSocketResponse.STATUS_INTERNAL_ERROR, message);
+    private void emitTransportErrorBlocking(byte status, String message) {
+        QueryEvent ev = new QueryEvent().asTransportError(status, message);
         while (!events.offer(ev)) {
             // Events queue is bounded; the consumer side (user thread) drains
             // steadily under a cooperative shutdown. Yield between retries so
@@ -428,13 +435,20 @@ private void emitErrorBlocking(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(String message) {
+    /**
+     * Emits a {@code KIND_TRANSPORT_ERROR} event AND latches the client's
+     * terminal-failure state. Use for transport- or protocol-level faults
+     * where the WebSocket is no longer usable (server close, send/recv
+     * exception, decoder out of sync). Per-query {@code QUERY_ERROR} still
+     * goes through {@link #emitError}, since the connection remains healthy
+     * after one. The event-kind split means {@code execute()} classifies
+     * the failure by the event kind, not by peeking at the terminal-failure
+     * latch -- the latch stays strictly for short-circuiting subsequent
+     * {@code execute()} calls on a broken client.
+     */
+    private void emitTerminalTransportError(String message) {
         notifyTerminalFailure(message);
-        emitError(WebSocketResponse.STATUS_INTERNAL_ERROR, message);
+        events.offer(new QueryEvent().asTransportError(WebSocketResponse.STATUS_INTERNAL_ERROR, message));
     }
 
     private void handleResultBatch(long payloadPtr, int payloadLen) {
@@ -467,7 +481,7 @@ private void handleResultBatch(long payloadPtr, int payloadLen) {
             }
             // A decode failure leaves the client-side decoder out of step with
             // the server's byte stream: the next frame cannot be trusted.
-            emitTerminalError("decode failure: " + e.getMessage());
+            emitTerminalTransportError("decode failure: " + e.getMessage());
             currentQueryDone = true;
             return;
         }

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

@@ -41,9 +41,34 @@ public final class QwpEgressMsgKind {
     public static final byte QUERY_REQUEST = 0x10;
     public static final byte RESULT_BATCH = 0x11;
     public static final byte RESULT_END = 0x12;
-
-    // Egress-specific QUERY_ERROR status codes. Extend the ingress
-    // QwpConstants.STATUS_* namespace (0x00-0x09).
+    /**
+     * Role value on {@code SERVER_INFO.role}: the authoritative write node.
+     */
+    public static final byte ROLE_PRIMARY = 1;
+    /**
+     * Role value on {@code SERVER_INFO.role}: promotion-in-progress. Clients
+     * insisting on primary-only reads may still route here; clients needing
+     * write-visible reads should wait for {@link #ROLE_PRIMARY}.
+     */
+    public static final byte ROLE_PRIMARY_CATCHUP = 3;
+    /**
+     * Role value on {@code SERVER_INFO.role}: the node is a read-only replica
+     * that pulls WAL segments from the shared object store. Reads may lag the
+     * primary by the replication poll interval plus transport time.
+     */
+    public static final byte ROLE_REPLICA = 2;
+    /**
+     * Role value on {@code SERVER_INFO.role}: no replication is configured. The
+     * standalone OSS default; behaves like a primary for routing purposes.
+     */
+    public static final byte ROLE_STANDALONE = 0;
+    /**
+     * Server -> client. Unsolicited frame delivered as the first QWP message
+     * on every v2 WebSocket connection. Body (little-endian): {@code
+     * msg_kind:u8, role:u8, epoch:u64, capabilities:u32, server_wall_ns:i64,
+     * cluster_id:u16_len+utf8, node_id:u16_len+utf8}.
+     */
+    public static final byte SERVER_INFO = 0x17;
     /**
      * Status byte on a {@code QUERY_ERROR} frame: the query was cancelled,
      * either by a client {@code CANCEL} frame or by explicit server-side cancel.