Make close() rethrow latched terminal errors

Three small changes to QwpWebSocketSender that close a silent-loss
window in the async client:

1. close() now captures the first Throwable raised during the
   flush/drain path, finishes shutdown of the I/O thread, socket
   and microbatch buffers, and then rethrows. Previously a HALT
   policy SenderError surfaced from drainOnClose was swallowed
   into a LOG.error(), so a user who only ever called close()
   (no flush() afterwards) had no way to observe the failure and
   no signal that data was lost. A static rethrowTerminal helper
   wraps any non-RuntimeException Throwable in a LineSenderException
   so close() keeps its existing throws contract.

2. drainOnClose now throws a LineSenderException on timeout
   carrying publishedFsn, ackedFsn, and the count of unacked
   batches, instead of logging a WARN and returning silently.
   The exception propagates through close() via mechanism (1)
   above. SF-mode users can recover the unacked tail by reopening
   on the same SF directory; memory-mode users have no recovery
   path and must treat this as fatal.

3. Adds two accessors so tests and user code can observe
   storage-drained progress without polling internal state:
     - getAckedFsn() returns the highest server-acknowledged FSN
       (or -1 before the I/O loop has started).
     - awaitAckedFsn(targetFsn, timeoutMillis) blocks until
       acked >= target or the timeout elapses, returning a
       boolean and surfacing latched I/O errors via
       cursorSendLoop.checkError(). Pair with
       flushAndGetSequence() to wait for a specific publish
       to land on disk server-side.

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

@@ -560,6 +560,45 @@ public void atNow() {
         }
     }
 
+    /**
+     * Blocks until {@code ackedFsn() >= targetFsn}, or until {@code timeoutMillis}
+     * elapses. Polls the cursor engine on a 50us park; surfaces I/O loop errors
+     * synchronously via {@code cursorSendLoop.checkError()}.
+     * <p>
+     * Useful for tests and user code that need to confirm a specific publish
+     * has been server-acknowledged. Pair with {@link #flushAndGetSequence()} to
+     * obtain {@code targetFsn}.
+     *
+     * @param targetFsn     FSN to wait for; typically {@link #flushAndGetSequence()}'s return value
+     * @param timeoutMillis upper bound on the wait; {@code <= 0} returns immediately
+     * @return {@code true} if {@code ackedFsn() >= targetFsn} on return, {@code false} on timeout
+     * @throws LineSenderException if the I/O loop has latched a terminal error
+     */
+    public boolean awaitAckedFsn(long targetFsn, long timeoutMillis) {
+        checkNotClosed();
+        if (cursorEngine == null) {
+            return targetFsn < 0L;
+        }
+        if (cursorEngine.ackedFsn() >= targetFsn) {
+            return true;
+        }
+        if (timeoutMillis <= 0L) {
+            return false;
+        }
+        long deadlineNanos = System.nanoTime() + timeoutMillis * 1_000_000L;
+        while (cursorEngine.ackedFsn() < targetFsn) {
+            if (cursorSendLoop != null) {
+                cursorSendLoop.checkError();
+            }
+            checkConnectionError();
+            if (System.nanoTime() >= deadlineNanos) {
+                return false;
+            }
+            java.util.concurrent.locks.LockSupport.parkNanos(50_000L);
+        }
+        return true;
+    }
+
     @Override
     public QwpWebSocketSender boolColumn(CharSequence columnName, boolean value) {
         checkNotClosed();
@@ -641,6 +680,12 @@ public void close() {
         if (!closed) {
             closed = true;
             boolean ioThreadStopped = true;
+            // Captures the first error from the flush/drain path so it can be
+            // rethrown after all cleanup is done. Silently swallowing it here
+            // would hide latched terminal SenderError HALTs (server-side
+            // rejections like MESSAGE_TOO_BIG, SCHEMA_MISMATCH HALT) from
+            // users who only call close() and never call flush() afterwards.
+            Throwable terminalError = null;
 
             try {
                 // Only drain when both the engine and the I/O loop are wired
@@ -663,8 +708,8 @@ public void close() {
                     //    data on JVM exit).
                     drainOnClose();
                 }
-            } catch (Exception e) {
-                LOG.error("Error during close: {}", String.valueOf(e));
+            } catch (Throwable t) {
+                terminalError = t;
             }
 
             // Shut down the I/O thread before closing the socket or buffers
@@ -705,6 +750,7 @@ public void close() {
                 // The I/O thread may still be using the socket and microbatch
                 // buffers. Freeing them would risk SIGSEGV.
                 LOG.error("I/O thread is still running, leaking WebSocket client and microbatch buffers");
+                rethrowTerminal(terminalError);
                 return;
             }
 
@@ -742,6 +788,8 @@ public void close() {
             }
 
             LOG.info("QwpWebSocketSender closed");
+
+            rethrowTerminal(terminalError);
         }
     }
 
@@ -982,6 +1030,18 @@ public long flushAndGetSequence() {
         return cursorEngine != null ? cursorEngine.publishedFsn() : -1L;
     }
 
+    /**
+     * Highest FSN that has been server-acknowledged (or skipped past on a
+     * {@link SenderError.Policy#DROP_AND_CONTINUE} rejection). {@code -1} if
+     * the I/O loop has not yet started or no batch has been published.
+     * <p>
+     * Snapshot accessor — for a bounded wait, use
+     * {@link #awaitAckedFsn(long, long)}.
+     */
+    public long getAckedFsn() {
+        return cursorEngine != null ? cursorEngine.ackedFsn() : -1L;
+    }
+
     /**
      * Returns the auto-flush byte threshold.
      */
@@ -1856,9 +1916,12 @@ 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.
+     * the drain (fast close). On timeout, throw a {@link LineSenderException}
+     * so the caller cannot silently lose data — close() collects the
+     * exception, finishes shutdown, and rethrows it from close() itself.
+     * SF-mode users can recover the unacked tail by reopening a sender on
+     * the same SF directory; memory-mode users have no recovery path and
+     * must treat this as fatal.
      */
     private void drainOnClose() {
         if (closeFlushTimeoutMillis <= 0L) {
@@ -1872,14 +1935,37 @@ private void drainOnClose() {
         while (cursorEngine.ackedFsn() < target) {
             cursorSendLoop.checkError();
             if (System.nanoTime() >= deadlineNanos) {
+                long acked = cursorEngine.ackedFsn();
                 LOG.warn("close() drain timed out after {}ms [target={} acked={}] — pending data may be lost",
-                        closeFlushTimeoutMillis, target, cursorEngine.ackedFsn());
-                return;
+                        closeFlushTimeoutMillis, target, acked);
+                throw new LineSenderException("close() drain timed out after ")
+                        .put(closeFlushTimeoutMillis).put(" ms [publishedFsn=")
+                        .put(target).put(", ackedFsn=").put(acked)
+                        .put("] - server did not acknowledge ")
+                        .put(target - acked)
+                        .put(" pending batches; data may be lost (use larger closeFlushTimeoutMillis or smaller batches)");
             }
             java.util.concurrent.locks.LockSupport.parkNanos(50_000L);
         }
     }
 
+    private static void rethrowTerminal(Throwable t) {
+        if (t == null) {
+            return;
+        }
+        if (t instanceof RuntimeException) {
+            throw (RuntimeException) t;
+        }
+        if (t instanceof Error) {
+            throw (Error) t;
+        }
+        // Wrap any checked Throwable so close() stays declared without a
+        // throws clause. flush/drain only ever raises RuntimeException
+        // subclasses today, but defending against future changes here is
+        // cheaper than chasing a leaked checked throw later.
+        throw new LineSenderException("close failed: " + t);
+    }
+
     private void rollbackRow() {
         if (currentTableBuffer != null) {
             currentTableBuffer.cancelCurrentRow();