fix(qwp): failed-stop teardown protocol -- never free the engine or client under a live I/O thread (C5 SEGV)

Completes the C5 fix. The landed discard-when-stopped commit removed the
routine post-teardown engine access; this eliminates the SEGV class on
the close path by construction: every native free is either ordered
after the I/O thread's last access via the shutdown latch, or performed
by the I/O thread itself.

1. CursorWebSocketSendLoop.close(): an interrupted latch await with the
   thread still live (latch count nonzero) now re-asserts the interrupt
   and THROWS instead of falling through -- falling through both closed
   the client's native buffers under a possibly mid-send thread and let
   owners unmap the engine under it. A concurrent latch-zero race is
   detected and proceeds as a normal stop. ioThread stays set on the
   throw, so a duplicate close() re-signals rather than silently
   succeeding.

2. delegateEngineClose() + delegated close in ioLoop's finally: on a
   failed stop the owner hands engine.close() to the I/O thread's exit
   path, which runs it strictly after the thread's last engine access
   and after the latch countdown. Store/load handshake (flag write then
   latch read vs countdown then flag read) guarantees no missed close
   under volatile SC; engine.close() is synchronized and idempotent, so
   a double close from both sides is benign. Unlike a deliberate leak,
   this releases the mapping and slot lock as soon as the stuck wire
   call resolves (bounded by OS timeouts) -- deferred, never abandoned.

3. BackgroundDrainer.run() finally: on a failed loop stop, skip the
   client close (the thread may be mid-send on it; its exit path closes
   the loop's current client) and delegate the engine close, falling
   back to closing locally when the thread exited in the interim. The
   mid-drain capability-gap recycle bails out on a failed stop instead
   of opening a new wire session against an engine the old thread still
   uses; the finally re-signals and applies the same protocol.

4. QwpWebSocketSender.close(): the ioThreadStopped guard -- now actually
   armed by close()'s throw, which the swallow previously made
   unreachable -- keeps leaking the microbatch buffers and client
   deliberately, but delegates the owned engine close instead of
   leaking the mapping and slot flock forever. slotLockReleased stays
   false: the pool must not reuse the slot until the delegated close
   has actually run.

Turns BackgroundDrainerInterruptedTeardownTest green; the C5 leak test
and close tests stay green. Regression net: sf/cursor + qwp/client +
impl + http/client (1249) and the sf integration package (incl.
BackgroundDrainerEndToEndTest, OrphanScanIntegrationTest) -- all green.

Residual analysis: with the connect racing close(), a client obtained
after running flips false is discarded on the I/O thread; a swapClient
already in flight completes against an engine that is now guaranteed
still mapped (owners cannot unmap while the latch is up and the thread
is live). No close-path free can race the I/O thread anymore.

Author: bluestreak01

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

@@ -1062,6 +1062,27 @@ 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");
+                // The engine, however, need not leak: delegate its close to
+                // the I/O thread's exit path, which runs it strictly after
+                // the thread's last engine access — the mapping and slot
+                // lock release as soon as the stuck wire call resolves
+                // (bounded by OS timeouts). slotLockReleased intentionally
+                // stays false: the lock is released only when the delegated
+                // close actually runs, so the pool must not reuse the slot
+                // meanwhile. A false return means the thread exited between
+                // the failed close() and now — then closing here is safe.
+                if (ownsCursorEngine && cursorEngine != null && cursorSendLoop != null
+                        && !cursorSendLoop.delegateEngineClose()) {
+                    try {
+                        cursorEngine.close();
+                    } catch (Throwable t) {
+                        LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
+                        terminalError = captureCloseError(terminalError, t);
+                    }
+                    cursorEngine = null;
+                    ownsCursorEngine = false;
+                    slotLockReleased = true;
+                }
                 rethrowTerminal(terminalError);
                 return;
             }

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

@@ -534,7 +534,21 @@ public void run() {
                                     slotPath, t.getMessage());
                             try {
                                 loop.close();
-                            } catch (Throwable ignored) {
+                            } catch (Throwable closeFailure) {
+                                // Interrupted shutdown mid-recycle (pool
+                                // shutdownNow): the old I/O thread is still
+                                // alive, so opening a new wire session against
+                                // the same engine would race its exit — and
+                                // closing the client under a possibly mid-send
+                                // thread risks SEGV. Bail out; the finally
+                                // re-runs loop.close(), which re-signals the
+                                // failed stop and routes client/engine
+                                // teardown to the delegation protocol there.
+                                LOG.warn("drainer slot {}: stop requested mid-recycle and the "
+                                                + "I/O thread did not stop ({}); abandoning recycle",
+                                        slotPath, closeFailure.getMessage());
+                                outcome = stopRequested ? DrainOutcome.STOPPED : DrainOutcome.FAILED;
+                                return;
                             }
                             try {
                                 client.close();
@@ -584,23 +598,51 @@ public void run() {
             }
             outcome = DrainOutcome.FAILED;
         } finally {
+            boolean ioThreadStopped = true;
             if (loop != null) {
                 try {
                     loop.close();
-                } catch (Throwable ignored) {
+                } catch (Throwable e) {
+                    // The loop's I/O thread would not stop — close() was
+                    // interrupted (the pool's shutdownNow path) while the
+                    // thread sat in a blocking native connect/send that
+                    // neither unpark nor interrupt cancels. Freeing the
+                    // client's buffers or unmapping the engine now would
+                    // race the live thread (C5 SEGV); both are delegated to
+                    // the thread's own exit path below.
+                    ioThreadStopped = false;
+                    LOG.warn("drainer slot {}: I/O thread did not stop during close ({}); "
+                                    + "delegating client/engine teardown to its exit path",
+                            slotPath, e.getMessage());
                 }
             }
-            if (client != null) {
+            if (client != null && ioThreadStopped) {
+                // Skipped on a failed stop: the thread may be mid-send on
+                // this very client; ioLoop's finally closes the loop's
+                // current client (this one, unless a reconnect swapped it —
+                // in which case swapClient already closed this reference).
                 try {
                     client.close();
                 } catch (Throwable ignored) {
                 }
             }
             if (engine != null) {
-                try {
-                    // engine.close() releases the slot lock too.
-                    engine.close();
-                } catch (Throwable ignored) {
+                // Failed-stop hand-off: delegateEngineClose() makes the I/O
+                // thread run engine.close() strictly after its last engine
+                // access, releasing the slot lock as soon as the stuck wire
+                // call resolves — deferred teardown, never abandoned. The
+                // false return covers the race where the thread exited
+                // between the failed close() and now: then it is safe (and
+                // necessary) to close the engine here.
+                if (ioThreadStopped || !loop.delegateEngineClose()) {
+                    try {
+                        // engine.close() releases the slot lock too.
+                        engine.close();
+                    } catch (Throwable ignored) {
+                    }
+                } else {
+                    LOG.warn("drainer slot {}: engine close delegated to the I/O thread; "
+                            + "slot lock releases when it exits", slotPath);
                 }
             }
             // Don't let a later requestStop() unpark an unrelated task that

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

@@ -217,6 +217,12 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // terminalError: the only writer runs on the I/O thread under the same
     // first-writer-wins latch.
     private volatile QwpDurableAckMismatchException capabilityGapTerminal;
+    // Failed-stop hand-off flag: set by delegateEngineClose() when an owner's
+    // close() could not stop the I/O thread and the engine close is therefore
+    // performed by the I/O thread's exit path. Write-once, owner thread only;
+    // read by the I/O thread strictly after its shutdown-latch countdown (see
+    // the handshake contract on delegateEngineClose).
+    private volatile boolean engineCloseDelegated;
     // The latched terminal failure — THE exception every checkError() call
     // rethrows. Write-once for the loop's lifetime: the only writer is
     // recordFatal on the I/O thread (first-writer-wins). The whole
@@ -576,8 +582,31 @@ public synchronized void close() {
             if (t.isAlive()) {
                 try {
                     shutdownLatch.await();
-                } catch (InterruptedException ignored) {
+                } catch (InterruptedException e) {
+                    // Re-assert the flag for the caller's stack, then decide.
+                    // If the I/O thread has genuinely not exited (latch still
+                    // up — it may be inside a blocking native connect/send
+                    // that neither unpark nor interrupt cancels), touching the
+                    // client here would free native buffers under a possibly
+                    // mid-send thread, and returning quietly would let the
+                    // owner unmap the engine under it (C5 SEGV). Signal the
+                    // failed stop loudly instead: QwpWebSocketSender.close()
+                    // keys its ioThreadStopped guard on this throw, and
+                    // BackgroundDrainer switches to delegateEngineClose().
+                    // The I/O thread's own exit path (ioLoop's finally)
+                    // disposes of the client either way. ioThread stays set,
+                    // so a duplicate close() re-signals rather than silently
+                    // succeeding against a still-live thread.
                     Thread.currentThread().interrupt();
+                    if (shutdownLatch.getCount() != 0L) {
+                        throw new LineSenderException(
+                                "cursor I/O thread did not stop: close() was interrupted "
+                                        + "while awaiting shutdown; client/engine teardown "
+                                        + "is delegated to the I/O thread's exit path");
+                    }
+                    // Latch hit zero concurrently with the interrupt: the
+                    // thread is past its last client/engine access — proceed
+                    // with normal teardown.
                 }
             }
             ioThread = null;
@@ -602,6 +631,34 @@ public synchronized void close() {
         }
     }
 
+    /**
+     * Failed-stop hand-off for the engine. Called by an owner whose
+     * {@link #close()} threw because the I/O thread would not stop: the owner
+     * must not free the engine (munmap/Unsafe.free of segment memory) while
+     * the thread may still touch it with raw {@code Unsafe} reads. Setting
+     * the delegation flag makes the I/O thread run {@code engine.close()} on
+     * its exit path, strictly after its last engine access and after the
+     * shutdown-latch countdown — releasing the slot lock as soon as the
+     * stuck wire call resolves (bounded by OS timeouts) instead of leaking
+     * the mapping and lock forever.
+     * <p>
+     * Returns {@code true} when the I/O thread is still live and has adopted
+     * the engine close; {@code false} when the thread has already exited —
+     * the caller must close the engine itself.
+     * <p>
+     * Memory model — the classic store/load handshake: this method writes the
+     * volatile flag, then reads the latch count; the exit path counts the
+     * latch down, then reads the flag. Under the sequential consistency of
+     * volatile (and AQS latch state) accesses, if this method observes the
+     * latch still up, the exit path is guaranteed to observe the flag — no
+     * missed close. If both sides act, {@link CursorSendEngine#close()} is
+     * synchronized and idempotent, so the double close is benign.
+     */
+    public boolean delegateEngineClose() {
+        engineCloseDelegated = true;
+        return shutdownLatch.getCount() != 0L;
+    }
+
     /**
      * Typed server-rejection payload of the latched terminal error, or
      * {@code null} when the loop latched a wire-level failure (or nothing).
@@ -1232,6 +1289,21 @@ private void ioLoop() {
                 }
             }
             shutdownLatch.countDown();
+            // Failed-stop hand-off (see delegateEngineClose): the owner could
+            // not free the engine safely while this thread was alive, so the
+            // engine close — and with it the slot-lock release — happens
+            // here, strictly after this thread's last engine access. The flag
+            // is read only after the countDown: the store/load pairing with
+            // delegateEngineClose's flag-write-then-latch-read guarantees
+            // either this branch or the owner's fallback runs (or both —
+            // engine.close() is idempotent).
+            if (engineCloseDelegated) {
+                try {
+                    engine.close();
+                } catch (Throwable ignored) {
+                    // best-effort
+                }
+            }
         }
     }