Add async progress callback, fast-close opt-out, and test coverage

Pair the existing SenderError async dispatch with a SenderProgressHandler
sibling so users can react to ack-watermark advances off the I/O thread.
SenderProgressDispatcher mirrors SenderErrorDispatcher: bounded inbox,
lazy-started daemon, volatile handler swap. Watermarks are monotonic, so
a full inbox evicts the oldest entry rather than the newest -- drops
compress, they don't lose information. CursorWebSocketSendLoop dispatches
on every advance the engine actually accepts; CursorSendEngine.acknowledge
and SegmentRing.acknowledge now return a boolean to give the loop a
no-cost way to skip dispatch on idempotent re-acks.

Make the close() drain genuinely opt-out. close_flush_timeout_millis<=0
now skips both drainOnClose AND the pre-drain checkError(). Previously
checkError ran unconditionally, so a HALT-policy rejection always escaped
through close() even when the user explicitly asked for a fast close.
Users opting out are expected to observe outcomes via the async progress
or error callback instead, which is exactly the surface this change adds.

setProgressHandler and setErrorHandler are post-connect-effective: they
forward to the live dispatcher's volatile handler field so builders,
tests, and reconfigurable apps can install a new handler without tearing
down the dispatcher thread. Drop the unused 1-arg
SenderProgressDispatcher constructor and the dead capacity field; the
LinkedBlockingDeque enforces its own bound, so the field had no readers.

Cover the public surface with focused unit tests:

- CursorSendEngineTest gains coverage for wasRecoveredFromDisk(),
  sealedSegments(), sealedSegmentsSnapshot(), and segmentSizeBytes().
  Snapshot tests assert empty-buffer, sized-fit, and undersize-buffer
  paths; recovery tests cover memory-mode, fresh-disk, and reopen.
- CursorWebSocketSendLoopDurableAckTest gains coverage for
  getTotalDurableAcks() and getTotalDurableTrimAdvances(), including
  the on-enqueue drain path that bumps trim advances even when the
  durable-ack frame itself ran against an empty queue.
- SegmentRingTest gains coverage for maxBytesPerSegment(), including
  the openExisting recovery path which forwards the configured value.

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

Author: bluestreak01

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

@@ -0,0 +1,76 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  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;
+
+/**
+ * User-supplied callback invoked when the asynchronous SF send loop observes a
+ * server acknowledgement that advances the per-sender ack watermark. Registered
+ * on {@code QwpWebSocketSender} via {@code setProgressHandler(...)} or on the
+ * builder via {@code LineSenderBuilder.progressHandler(...)}.
+ *
+ * <h2>Watermark semantics</h2>
+ * {@code ackedFsn} is the highest FSN whose batch is now durable on the server
+ * side (committed to WAL, or skipped past on a {@code DROP_AND_CONTINUE}
+ * rejection). The handler fires only when the watermark <em>advances</em>:
+ * <ul>
+ *   <li>delivered values are strictly increasing,</li>
+ *   <li>the handler may be called many times during the lifetime of a sender,</li>
+ *   <li>a single call may skip multiple FSNs if the server batches several
+ *       frames into one OK frame.</li>
+ * </ul>
+ *
+ * Callers polling for "is everything up to FSN N durable?" should compare
+ * {@code ackedFsn} against their target and act once the inequality is satisfied,
+ * not assume one call per sent batch.
+ *
+ * <h2>Threading</h2>
+ * Implementations are invoked on a dedicated daemon dispatcher thread, never on
+ * the I/O thread or the producer thread. Slow handlers cannot stall publishing.
+ * If the bounded inbox fills, surplus notifications are dropped — visible via
+ * {@code QwpWebSocketSender.getDroppedProgressNotifications()}. Drops are
+ * tolerable because the next delivered call carries an equal-or-higher FSN, so
+ * watchers comparing against a target threshold catch up automatically.
+ *
+ * <h2>Exceptions</h2>
+ * Any {@link Throwable} thrown by the handler is caught and logged by the
+ * dispatcher. The dispatcher and the sender continue running.
+ *
+ * <h2>What this callback is for</h2>
+ * Marking application state durable, releasing producer-side latches, fan-out
+ * to journals tagged with {@code (fsn, domainContext)} pairs returned by
+ * {@code flushAndGetSequence()}. For an "is this batch rejected" question on the
+ * producer thread, see {@link SenderErrorHandler} and
+ * {@link io.questdb.client.cutlass.line.LineSenderException}.
+ *
+ * @see SenderErrorHandler
+ */
+@FunctionalInterface
+public interface SenderProgressHandler {
+    /**
+     * Called when the server-acked watermark advances. Strictly monotonic:
+     * {@code ackedFsn} on call N+1 is greater than on call N.
+     */
+    void onAcked(long ackedFsn);
+}