feat(ilp): cursor I/O loop reconnect + replay

The cursor I/O loop previously treated any wire failure as terminal —
first disconnect = sender broken, every subsequent batch threw. Now,
when the sender wires a ReconnectFactory + ReconnectListener, a wire
failure triggers:

  1. WARN log
  2. Build a fresh WebSocketClient via the factory (same auth/TLS/host)
  3. Reset wire state: nextWireSeq=0, fsnAtZero = engine.ackedFsn() + 1
  4. Reposition the cursor at the first unacked FSN (replay)
  5. Notify the listener → producer's connectionGeneration bumps so
     the next encode emits full schema definitions, not refs the new
     server has never seen
  6. Outer ioLoop continues — nextWireSeq=0 starts on the new wire,
     trySendOne picks up at the repositioned cursor and replays every
     unacked frame, then continues with whatever the producer publishes
     next

Added in main:
  * CursorWebSocketSendLoop.ReconnectFactory + .ReconnectListener
    interfaces (both functional, both null-able for legacy "fail-fast"
    semantics)
  * positionCursorAt(fsn) — walks frames inside the segment containing
    fsn to find the byte offset
  * SegmentRing.findSegmentContaining(fsn) + CursorSendEngine
    pass-through — used by the cursor reposition
  * QwpWebSocketSender extracts buildAndConnect() to use both for the
    initial connect and as the reconnect factory; onWireReconnect()
    is the listener that bumps connectionGeneration

This commit covers the *mechanics* (one attempt, succeed-or-fail).
The follow-up commit adds policy: exponential backoff with jitter,
per-outage time cap (reconnect_max_duration_millis, default 300s
per spec decision #2), and auth-failure detection (401/403/non-101
treated as terminal so the retry budget isn't wasted on errors that
won't fix themselves).

Two integration tests:
  * testReconnectAfterServerInducedDisconnect — server ACKs then
    closes; sender reconnects, second batch lands on the new wire
  * testReplayResendsUnackedFramesAcrossReconnect — server receives
    the first frame WITHOUT ACKing then closes; sender reconnects
    and replays the unacked frame on the new connection

Spec decisions #5 (encode-mid-reconnect race) and the core of
#1/#2 (reconnect mechanics) land here.

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

Author: bluestreak01

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

@@ -1314,25 +1314,14 @@ private void ensureConnected() {
         if (cursorEngine == null) {
             throw new LineSenderException("cursor engine must be attached before connect");
         }
-        if (tlsConfig != null) {
-            client = WebSocketClientFactory.newTlsInstance(tlsConfig);
-        } else {
-            client = WebSocketClientFactory.newPlainTextInstance();
-        }
-        try {
-            client.setQwpMaxVersion(QwpConstants.MAX_SUPPORTED_INGEST_VERSION);
-            client.setQwpClientId(QwpConstants.CLIENT_ID);
-            client.setQwpRequestDurableAck(requestDurableAck);
-            client.connect(host, port);
-            client.upgrade(WRITE_PATH, authorizationHeader);
-        } catch (Exception e) {
-            client.close();
-            client = null;
-            throw new LineSenderException("Failed to connect to " + host + ":" + port, e);
-        }
+        client = buildAndConnect();
 
         try {
-            cursorSendLoop = new CursorWebSocketSendLoop(client, cursorEngine);
+            cursorSendLoop = new CursorWebSocketSendLoop(
+                    client, cursorEngine,
+                    0L, CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
+                    this::buildAndConnect,
+                    this::onWireReconnect);
             cursorSendLoop.start();
         } catch (Throwable t) {
             client.close();
@@ -1361,6 +1350,47 @@ private void ensureConnected() {
                 host, port, inFlightWindowSize, client.getServerQwpVersion());
     }
 
+    /**
+     * Build and connect a fresh WebSocket client using the sender's
+     * persistent config (host/port/TLS/auth/durable-ack flag). Used both
+     * for the initial connect and as the reconnect factory passed to the
+     * cursor I/O loop. Throws {@link LineSenderException} on any failure
+     * — the I/O loop's reconnect path treats a throw as fatal for that
+     * attempt (and, in the follow-up commit, schedules a backoff retry
+     * within the per-outage time cap).
+     */
+    private WebSocketClient buildAndConnect() {
+        WebSocketClient newClient;
+        if (tlsConfig != null) {
+            newClient = WebSocketClientFactory.newTlsInstance(tlsConfig);
+        } else {
+            newClient = WebSocketClientFactory.newPlainTextInstance();
+        }
+        try {
+            newClient.setQwpMaxVersion(QwpConstants.MAX_SUPPORTED_INGEST_VERSION);
+            newClient.setQwpClientId(QwpConstants.CLIENT_ID);
+            newClient.setQwpRequestDurableAck(requestDurableAck);
+            newClient.connect(host, port);
+            newClient.upgrade(WRITE_PATH, authorizationHeader);
+        } catch (Exception e) {
+            newClient.close();
+            throw new LineSenderException("Failed to connect to " + host + ":" + port, e);
+        }
+        return newClient;
+    }
+
+    /**
+     * Called by the cursor I/O loop after a successful reconnect. The wire
+     * state has been reset and the cursor repositioned for replay; we bump
+     * connectionGeneration so the producer thread's next encode treats the
+     * connection as fresh (full schema definitions, not refs the new server
+     * has never seen). Single-writer (the I/O thread invokes this), so a
+     * plain volatile increment is safe.
+     */
+    private void onWireReconnect() {
+        connectionGeneration++;
+    }
+
     private void ensureNoInProgressRow() {
         if (currentTableBuffer != null && currentTableBuffer.hasInProgressRow()) {
             throw new LineSenderException(

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java

@@ -290,6 +290,11 @@ public MmapSegment firstSealed() {
         return ring.firstSealed();
     }
 
+    /** Pass-through to {@link SegmentRing#findSegmentContaining(long)}. */
+    public MmapSegment findSegmentContaining(long fsn) {
+        return ring.findSegmentContaining(fsn);
+    }
+
     /** Configured per-segment size in bytes. */
     public long segmentSizeBytes() {
         return segmentSizeBytes;

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java

@@ -73,20 +73,28 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     public static final long DEFAULT_PARK_NANOS = 50_000L; // 50us idle backoff
     private static final Logger LOG = LoggerFactory.getLogger(CursorWebSocketSendLoop.class);
 
-    private final WebSocketClient client;
     private final AtomicLong consecutiveSendErrors = new AtomicLong();
     private final CursorSendEngine engine;
-    // fsnAtZero: FSN that wireSeq=0 maps to on this connection. For a fresh
-    // connection starting from a fresh engine (no recovery), this is 0.
-    // Once recovery / reconnect lands (PR2), this is set to the first
-    // unacked FSN at connect time so wire-seq math stays aligned.
-    private final long fsnAtZero;
     private final long parkNanos;
     private final WebSocketResponse response = new WebSocketResponse();
     private final ResponseHandler responseHandler = new ResponseHandler();
     private final CountDownLatch shutdownLatch = new CountDownLatch(1);
     private final AtomicLong totalAcks = new AtomicLong();
     private final AtomicLong totalFramesSent = new AtomicLong();
+    private final AtomicLong totalReconnects = new AtomicLong();
+    // Optional reconnect plumbing. If both are non-null, a wire failure
+    // triggers a reconnect attempt instead of a terminal fail(). The factory
+    // produces a fresh, connected+upgraded WebSocketClient; the listener is
+    // notified after the wire state has been reset so the producer thread
+    // can bump its connectionGeneration.
+    private final ReconnectFactory reconnectFactory;
+    private final ReconnectListener reconnectListener;
+    private WebSocketClient client;
+    // fsnAtZero: FSN that wireSeq=0 maps to on the current connection. For
+    // a fresh connection, this is 0. After a reconnect, it's set to
+    // engine.ackedFsn() + 1 — the first frame we replay maps to wireSeq=0
+    // on the new connection so server-side ACK math stays aligned.
+    private long fsnAtZero;
     // sendingSegment: the segment we're currently consuming bytes from. Starts
     // at engine.activeSegment(); advances to newer sealed segments / the new
     // active as the producer rotates.
@@ -100,18 +108,61 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     private Thread ioThread;
 
     public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine) {
-        this(client, engine, 0L, DEFAULT_PARK_NANOS);
+        this(client, engine, 0L, DEFAULT_PARK_NANOS, null, null);
     }
 
     public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                                    long fsnAtZero, long parkNanos) {
+        this(client, engine, fsnAtZero, parkNanos, null, null);
+    }
+
+    /**
+     * Full constructor with reconnect plumbing. When {@code reconnectFactory}
+     * and {@code reconnectListener} are both non-null, the I/O thread treats
+     * wire failures (send/receive errors, server-initiated close) as
+     * recoverable: it calls the factory to obtain a fresh connected client,
+     * resets wire state, repositions its replay cursor at
+     * {@code engine.ackedFsn() + 1}, and notifies the listener so the
+     * producer can bump its {@code connectionGeneration}. Either being null
+     * disables reconnect (legacy behavior — single failure is terminal).
+     */
+    public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+                                   long fsnAtZero, long parkNanos,
+                                   ReconnectFactory reconnectFactory,
+                                   ReconnectListener reconnectListener) {
         if (client == null || engine == null) {
             throw new IllegalArgumentException("client and engine must be non-null");
         }
         this.client = client;
         this.engine = engine;
         this.fsnAtZero = fsnAtZero;
         this.parkNanos = parkNanos;
+        this.reconnectFactory = reconnectFactory;
+        this.reconnectListener = reconnectListener;
+    }
+
+    /**
+     * Factory used by the I/O loop to build a fresh, connected, upgraded
+     * {@link WebSocketClient} after a wire failure. Implementations close
+     * the old client (if needed), build a new one with the same auth/TLS
+     * config, connect, perform the WebSocket upgrade, and return it ready
+     * to send. Throw on a terminal failure (auth rejection, etc.) — the
+     * I/O loop will treat the throw as fatal.
+     */
+    @FunctionalInterface
+    public interface ReconnectFactory {
+        WebSocketClient reconnect() throws Exception;
+    }
+
+    /**
+     * Notified after a successful reconnect — wire state has been reset and
+     * the cursor repositioned for replay. Implementations typically bump a
+     * {@code connectionGeneration} counter the producer thread reads so
+     * the next encode emits full schema definitions instead of refs.
+     */
+    @FunctionalInterface
+    public interface ReconnectListener {
+        void onReconnect();
     }
 
     /**
@@ -153,6 +204,10 @@ public long getTotalFramesSent() {
         return totalFramesSent.get();
     }
 
+    public long getTotalReconnects() {
+        return totalReconnects.get();
+    }
+
     public synchronized void start() {
         if (ioThread != null) {
             throw new IllegalStateException("already started");
@@ -201,14 +256,94 @@ private MmapSegment advanceSegment() {
         return liveActive;
     }
 
+    /**
+     * Surface a wire failure. With reconnect plumbing wired (factory +
+     * listener both non-null), tries one reconnect first; success returns
+     * silently and the I/O loop continues with reset wire state. Failure
+     * (or no reconnect plumbing) records the error and stops the loop.
+     * <p>
+     * Backoff / per-outage time cap / auth-failure detection land in the
+     * follow-up commit; this commit proves the mechanics with a single
+     * attempt.
+     */
     private void fail(Throwable t) {
+        if (reconnectFactory != null && reconnectListener != null && running) {
+            LOG.warn("cursor I/O loop wire failure, attempting reconnect: {}", t.getMessage());
+            try {
+                WebSocketClient newClient = reconnectFactory.reconnect();
+                if (newClient != null) {
+                    swapClient(newClient);
+                    totalReconnects.incrementAndGet();
+                    reconnectListener.onReconnect();
+                    LOG.info("cursor I/O loop reconnected; replaying from FSN {}", fsnAtZero);
+                    return;
+                }
+            } catch (Throwable reconnectError) {
+                LOG.error("cursor I/O loop reconnect failed: {}", reconnectError.getMessage(),
+                        reconnectError);
+                t = new LineSenderException(
+                        "cursor I/O loop wire failure followed by reconnect failure: "
+                                + reconnectError.getMessage(), reconnectError);
+            }
+        }
         if (lastError == null) {
             lastError = t;
         }
         running = false;
         LOG.error("Cursor I/O loop failure: {}", t.getMessage(), t);
     }
 
+    /**
+     * Reset wire state for a fresh connection: install the new client,
+     * realign {@code fsnAtZero} to the next unacked FSN, restart wire
+     * sequencing from 0, and reposition the cursor so the next
+     * {@link #trySendOne} call replays the first unacked frame.
+     */
+    private void swapClient(WebSocketClient newClient) {
+        WebSocketClient old = this.client;
+        this.client = newClient;
+        if (old != null) {
+            try {
+                old.close();
+            } catch (Throwable ignored) {
+                // best-effort
+            }
+        }
+        long replayStart = engine.ackedFsn() + 1L;
+        this.fsnAtZero = replayStart;
+        this.nextWireSeq = 0L;
+        this.consecutiveSendErrors.set(0L);
+        positionCursorAt(replayStart);
+    }
+
+    /**
+     * Walk the engine's segments to find the one containing {@code targetFsn},
+     * and set {@code sendOffset} to the byte offset of that frame within it.
+     * If {@code targetFsn} is past everything published, park at the live
+     * active segment's published offset (caller will wait for new bytes).
+     */
+    private void positionCursorAt(long targetFsn) {
+        MmapSegment seg = engine.findSegmentContaining(targetFsn);
+        if (seg == null) {
+            // targetFsn is at or past publishedFsn — nothing to replay.
+            // Resume from the active segment's tip; producer may add more.
+            sendingSegment = engine.activeSegment();
+            sendOffset = sendingSegment.publishedOffset();
+            return;
+        }
+        sendingSegment = seg;
+        // Walk frame-by-frame from HEADER_SIZE until we land on targetFsn.
+        long offset = MmapSegment.HEADER_SIZE;
+        long fsn = seg.baseSeq();
+        long base = seg.address();
+        while (fsn < targetFsn) {
+            int payloadLen = Unsafe.getUnsafe().getInt(base + offset + 4);
+            offset += MmapSegment.FRAME_HEADER_SIZE + payloadLen;
+            fsn++;
+        }
+        sendOffset = offset;
+    }
+
     private void ioLoop() {
         try {
             while (running) {

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java

@@ -443,6 +443,35 @@ public synchronized MmapSegment firstSealed() {
         return sealedSegments.size() > 0 ? sealedSegments.get(0) : null;
     }
 
+    /**
+     * Returns the segment whose published frame range covers {@code fsn}, or
+     * {@code null} if no segment currently holds it (e.g. the FSN is past
+     * {@code publishedFsn} or has been trimmed). Used by the reconnect path
+     * to position the I/O thread's cursor at the first unacked frame for
+     * replay.
+     * <p>
+     * Walks sealed first (oldest → newest) then the active. The sealed list
+     * is small enough — and reconnects are rare enough — that the linear
+     * scan cost doesn't matter.
+     */
+    public synchronized MmapSegment findSegmentContaining(long fsn) {
+        for (int i = 0, n = sealedSegments.size(); i < n; i++) {
+            MmapSegment s = sealedSegments.get(i);
+            long base = s.baseSeq();
+            if (fsn >= base && fsn < base + s.frameCount()) {
+                return s;
+            }
+        }
+        MmapSegment a = active;
+        if (a != null) {
+            long base = a.baseSeq();
+            if (fsn >= base && fsn < base + a.frameCount()) {
+                return a;
+            }
+        }
+        return null;
+    }
+
     /**
      * Segment manager pre-creates the next segment and parks it here. The
      * producer consumes the spare on its next rotation. Throws if a spare