feat(ilp): close() drain — bounded ACK wait via close_flush_timeout_millis

QwpWebSocketSender.close() previously stopped the cursor I/O loop the
moment it was called: any frames already published into the engine but
not yet sent (or sent but not yet ACK'd) were silently dropped on
JVM exit. In memory mode that means data loss; in SF mode the next
sender recovers from disk, but the durability claim of close() was
weaker than the spec promised.

Closes the gap with one knob from the durability spec
(design/qwp-cursor-durability.md, decision #3):

  close_flush_timeout_millis (default 5000)
    > 0:  close() blocks until ackedFsn >= publishedFsn or timeout
    0/-1: fast close — no drain, opt-in to legacy fast-exit behavior

On timeout, log WARN and proceed with shutdown. SF-mode pending data
is recoverable; memory-mode pending data is not.

Wired through:
  - LineSenderBuilder.closeFlushTimeoutMillis(long)
  - connect-string key close_flush_timeout_millis
  - new QwpWebSocketSender.connect overload that takes the timeout

Tests cover all three regimes:
  - delayed-ACK server: close blocks ~ack delay
  - timeout=0: close returns immediately
  - silent server: close times out at the configured cap, logs WARN

This is decision #3 of the spec; subsequent commits add the
connectionGeneration foundation, reconnect/replay, slot dirs,
and background drainers.

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

Author: bluestreak01

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

@@ -658,6 +658,10 @@ public int getTimeout() {
         // never approaches it. SF mode = 10 GiB (2560 segments at default
         // size). Users can lower this on space-constrained hosts.
         private static final long DEFAULT_MAX_BYTES_SF = 10L * 1024 * 1024 * 1024;
+        // Default close() drain timeout: block up to 5s waiting for the
+        // server to ACK everything published into the engine before
+        // shutting down the I/O loop.
+        private static final long DEFAULT_CLOSE_FLUSH_TIMEOUT_MILLIS = 5_000L;
         // Store-and-forward (WebSocket only). SF is enabled iff sfDir is non-null —
         // there is no separate on/off flag (presence of the directory is the switch).
         // null sfDir → memory-only async ingest (same lock-free architecture, no disk).
@@ -668,6 +672,10 @@ public int getTimeout() {
         // implemented; FLUSH and APPEND are deferred follow-ups (cursor needs
         // to learn fsync first).
         private SfDurability sfDurability = SfDurability.MEMORY;
+        // close() drain timeout. Default applied at build() time. 0 or -1
+        // means "fast close" (skip the drain entirely); any positive value
+        // bounds the wait for ackedFsn to catch up to publishedFsn.
+        private long closeFlushTimeoutMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private boolean tlsEnabled;
         private TlsValidationMode tlsValidationMode;
         private char[] trustStorePassword;
@@ -996,6 +1004,9 @@ public Sender build() {
                 long actualSfMaxTotalBytes = sfMaxTotalBytes == PARAMETER_NOT_SET_EXPLICITLY
                         ? Math.max(defaultMaxTotal, actualSfMaxBytes * 2)
                         : sfMaxTotalBytes;
+                long actualCloseFlushTimeoutMillis = closeFlushTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY
+                        ? DEFAULT_CLOSE_FLUSH_TIMEOUT_MILLIS
+                        : closeFlushTimeoutMillis;
 
                 CursorSendEngine cursorEngine = new CursorSendEngine(
                         sfDir, actualSfMaxBytes,
@@ -1012,7 +1023,8 @@ public Sender build() {
                             wsAuthHeader,
                             actualMaxSchemasPerConnection,
                             requestDurableAck,
-                            cursorEngine
+                            cursorEngine,
+                            actualCloseFlushTimeoutMillis
                     );
                 } catch (Throwable t) {
                     try {
@@ -1662,6 +1674,26 @@ public LineSenderBuilder storeAndForwardMaxTotalBytes(long maxTotalBytes) {
             return this;
         }
 
+        /**
+         * close() drain timeout in milliseconds. The sender's {@code close()}
+         * method blocks up to this many millis waiting for the server to ACK
+         * every batch already published into the engine before shutting down
+         * the I/O loop. Default {@code 5000}.
+         * <p>
+         * Set to {@code 0} or {@code -1} to opt out — close() will not wait
+         * at all (fast close). Pending data is then lost in memory mode and
+         * recovered by the next sender in SF mode.
+         * <p>
+         * WebSocket transport only.
+         */
+        public LineSenderBuilder closeFlushTimeoutMillis(long timeoutMillis) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("close_flush_timeout_millis is only supported for WebSocket transport");
+            }
+            this.closeFlushTimeoutMillis = timeoutMillis;
+            return this;
+        }
+
         /**
          * Selects the durability contract for SF appends and flushes. See
          * {@link SfDurability} for the value semantics.
@@ -2190,6 +2222,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     }
                     pos = getValue(configurationString, pos, sink, "sf_durability");
                     storeAndForwardDurability(parseDurabilityValue(sink));
+                } else if (Chars.equals("close_flush_timeout_millis", sink)) {
+                    if (protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException("close_flush_timeout_millis is only supported for WebSocket transport");
+                    }
+                    pos = getValue(configurationString, pos, sink, "close_flush_timeout_millis");
+                    closeFlushTimeoutMillis(parseLongValue(sink, "close_flush_timeout_millis"));
                 } else if (Chars.equals("max_datagram_size", sink)) {
                     pos = getValue(configurationString, pos, sink, "max_datagram_size");
                     int mds = parseIntValue(sink, "max_datagram_size");

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

@@ -160,6 +160,10 @@ public class QwpWebSocketSender implements Sender {
     private CursorSendEngine cursorEngine;
     private boolean ownsCursorEngine;
     private CursorWebSocketSendLoop cursorSendLoop;
+    // close() drain timeout in millis. Default applied at construction.
+    // 0 or -1 means "fast close" (skip the drain); otherwise close blocks
+    // up to this many millis for ackedFsn to catch up to publishedFsn.
+    private long closeFlushTimeoutMillis = 5_000L;
 
     private QwpWebSocketSender(
             String host,
@@ -267,6 +271,31 @@ public static QwpWebSocketSender connect(
             int maxSchemasPerConnection,
             boolean requestDurableAck,
             CursorSendEngine cursorEngine
+    ) {
+        return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos,
+                inFlightWindowSize, authorizationHeader, maxSchemasPerConnection,
+                requestDurableAck, cursorEngine, 5_000L);
+    }
+
+    /**
+     * Connect overload that also configures the {@code close()} drain
+     * timeout. {@code 0} or {@code -1} disables the drain (fast close);
+     * any positive value bounds the wait for {@code ackedFsn} to catch
+     * up to {@code publishedFsn} during {@code close()}.
+     */
+    public static QwpWebSocketSender connect(
+            String host,
+            int port,
+            ClientTlsConfiguration tlsConfig,
+            int autoFlushRows,
+            int autoFlushBytes,
+            long autoFlushIntervalNanos,
+            int inFlightWindowSize,
+            String authorizationHeader,
+            int maxSchemasPerConnection,
+            boolean requestDurableAck,
+            CursorSendEngine cursorEngine,
+            long closeFlushTimeoutMillis
     ) {
         QwpWebSocketSender sender = new QwpWebSocketSender(
                 host, port, tlsConfig,
@@ -275,6 +304,7 @@ public static QwpWebSocketSender connect(
         );
         try {
             sender.requestDurableAck = requestDurableAck;
+            sender.closeFlushTimeoutMillis = closeFlushTimeoutMillis;
             if (cursorEngine != null) {
                 sender.setCursorEngine(cursorEngine, true);
             }
@@ -485,16 +515,21 @@ public void close() {
                 // up — close() is also called from createForTesting() teardown
                 // and from connect() rollback paths where one or both may be null.
                 if (connectionError.get() == null && cursorEngine != null && cursorSendLoop != null) {
-                    // Flush user-thread state (accumulated rows -> microbatch ->
-                    // engine append). The cursor engine append is durable
-                    // (on-disk in SF mode, in-RAM in memory mode); we don't
-                    // wait for the cursor I/O loop to acknowledge — the data
-                    // is already past the in-flight handoff.
+                    // 1) Flush user-thread state into the engine (encoded
+                    //    rows → mmap'd / malloc'd ring). After this, the
+                    //    cursor engine's publishedFsn reflects the final
+                    //    target the I/O loop must drive ackedFsn up to.
                     flushPendingRows();
                     if (activeBuffer != null && activeBuffer.hasData()) {
                         sealAndSwapBuffer();
                     }
                     cursorSendLoop.checkError();
+                    // 2) Bounded drain: block until the server has ACK'd
+                    //    everything we just published, or until the
+                    //    configured timeout elapses. closeFlushTimeoutMillis
+                    //    <= 0 opts out (fast close, may lose memory-mode
+                    //    data on JVM exit).
+                    drainOnClose();
                 }
             } catch (Exception e) {
                 LOG.error("Error during close: {}", String.valueOf(e));
@@ -1417,6 +1452,33 @@ private void resetSchemaStateForNewConnection() {
         }
     }
 
+    /**
+     * Bounded drain on close: block until {@code ackedFsn >= publishedFsn}
+     * or until {@code closeFlushTimeoutMillis} elapses. {@code <= 0} skips
+     * the drain (fast close). On timeout, log a WARN and proceed with
+     * shutdown — pending data is lost in memory mode and recovered by
+     * the next sender in SF mode.
+     */
+    private void drainOnClose() {
+        if (closeFlushTimeoutMillis <= 0L) {
+            return;
+        }
+        long target = cursorEngine.publishedFsn();
+        if (cursorEngine.ackedFsn() >= target) {
+            return;
+        }
+        long deadlineNanos = System.nanoTime() + closeFlushTimeoutMillis * 1_000_000L;
+        while (cursorEngine.ackedFsn() < target) {
+            cursorSendLoop.checkError();
+            if (System.nanoTime() >= deadlineNanos) {
+                LOG.warn("close() drain timed out after {}ms [target={} acked={}] — pending data may be lost",
+                        closeFlushTimeoutMillis, target, cursorEngine.ackedFsn());
+                return;
+            }
+            java.util.concurrent.locks.LockSupport.parkNanos(50_000L);
+        }
+    }
+
     private void rollbackRow() {
         if (currentTableBuffer != null) {
             currentTableBuffer.cancelCurrentRow();

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/CloseDrainTest.java

@@ -0,0 +1,175 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  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.test.cutlass.qwp.client;
+
+import io.questdb.client.Sender;
+import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Regression tests for the close() drain semantics specified in
+ * design/qwp-cursor-durability.md.
+ * <p>
+ * Without {@code close_flush_timeout_millis}, close() returned as soon as
+ * the cursor I/O loop's {@code running} flag flipped — meaning frames
+ * still queued in the engine could be dropped when the JVM exited
+ * immediately after close(). The drain timeout makes close() wait for
+ * the server to ACK everything published before shutting the loop down.
+ */
+public class CloseDrainTest {
+
+    private static final int TEST_PORT = 19_700 + (int) (System.nanoTime() % 100);
+
+    @Test
+    public void testCloseBlocksUntilAckArrives() throws Exception {
+        // Server delays every ACK by 800ms. With the default
+        // close_flush_timeout_millis=5000, close() must wait for that ACK
+        // before returning. Pre-fix close() returned within milliseconds.
+        int port = TEST_PORT + 1;
+        long ackDelayMs = 800;
+        DelayingAckHandler handler = new DelayingAckHandler(ackDelayMs);
+        try (TestWebSocketServer server = new TestWebSocketServer(port, handler)) {
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+            String cfg = "ws::addr=localhost:" + port + ";";  // memory mode
+            long elapsedMs;
+            try (Sender sender = Sender.fromConfig(cfg)) {
+                sender.table("foo").longColumn("v", 1L).atNow();
+                sender.flush();
+                long t0 = System.nanoTime();
+                sender.close();
+                elapsedMs = (System.nanoTime() - t0) / 1_000_000;
+            }
+            Assert.assertTrue(
+                    "close() took only " + elapsedMs + "ms — did not wait for ACK; "
+                            + "drain timeout is broken or never enabled",
+                    elapsedMs >= ackDelayMs / 2);
+        }
+    }
+
+    @Test
+    public void testCloseFastWhenTimeoutIsZero() throws Exception {
+        // Same delayed-ACK server, but with close_flush_timeout_millis=0
+        // (fast close). close() must return immediately, well before the
+        // ACK delay would have elapsed.
+        int port = TEST_PORT + 2;
+        long ackDelayMs = 1500;
+        DelayingAckHandler handler = new DelayingAckHandler(ackDelayMs);
+        try (TestWebSocketServer server = new TestWebSocketServer(port, handler)) {
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+            String cfg = "ws::addr=localhost:" + port
+                    + ";close_flush_timeout_millis=0;";
+            long elapsedMs;
+            try (Sender sender = Sender.fromConfig(cfg)) {
+                sender.table("foo").longColumn("v", 1L).atNow();
+                sender.flush();
+                long t0 = System.nanoTime();
+                sender.close();
+                elapsedMs = (System.nanoTime() - t0) / 1_000_000;
+            }
+            Assert.assertTrue(
+                    "close() with timeout=0 took " + elapsedMs + "ms — fast close is broken",
+                    elapsedMs < ackDelayMs / 2);
+        }
+    }
+
+    @Test
+    public void testCloseDrainTimesOutWhenAcksNeverArrive() throws Exception {
+        // Server that buffers frames silently and never ACKs. close() must
+        // return after roughly the configured timeout — not hang forever
+        // and not return immediately.
+        int port = TEST_PORT + 3;
+        long timeoutMs = 500;
+        SilentHandler handler = new SilentHandler();
+        try (TestWebSocketServer server = new TestWebSocketServer(port, handler)) {
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+            String cfg = "ws::addr=localhost:" + port
+                    + ";close_flush_timeout_millis=" + timeoutMs + ";";
+            long elapsedMs;
+            try (Sender sender = Sender.fromConfig(cfg)) {
+                sender.table("foo").longColumn("v", 1L).atNow();
+                sender.flush();
+                long t0 = System.nanoTime();
+                sender.close();
+                elapsedMs = (System.nanoTime() - t0) / 1_000_000;
+            }
+            Assert.assertTrue("close() returned too early: " + elapsedMs + "ms",
+                    elapsedMs >= timeoutMs);
+            Assert.assertTrue("close() exceeded the bounded timeout by too much: " + elapsedMs + "ms",
+                    elapsedMs < timeoutMs * 4);
+        }
+    }
+
+    /** Acks every binary frame after a fixed delay, so we can observe close() blocking. */
+    private static class DelayingAckHandler implements TestWebSocketServer.WebSocketServerHandler {
+        private final long delayMs;
+        private final AtomicLong nextSeq = new AtomicLong(0);
+
+        DelayingAckHandler(long delayMs) {
+            this.delayMs = delayMs;
+        }
+
+        @Override
+        public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+            try {
+                Thread.sleep(delayMs);
+                client.sendBinary(buildAck(nextSeq.getAndIncrement()));
+            } catch (IOException | InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+    /** Receives but never ACKs — used to verify close() honors its timeout cap. */
+    private static class SilentHandler implements TestWebSocketServer.WebSocketServerHandler {
+        @Override
+        public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+            // intentionally drop the frame on the floor
+        }
+    }
+
+    // Mirrors WebSocketResponse STATUS_OK layout: status u8 | sequence u64 | table_count u16
+    static byte[] buildAck(long seq) {
+        byte[] buf = new byte[1 + 8 + 2];
+        ByteBuffer bb = ByteBuffer.wrap(buf).order(ByteOrder.LITTLE_ENDIAN);
+        bb.put((byte) 0x00); // STATUS_OK
+        bb.putLong(seq);
+        bb.putShort((short) 0);
+        return buf;
+    }
+}