Test deferred close caller fallbacks

Add deterministic gray-box coverage for BackgroundDrainer and the cursor WebSocket send loop. The tests stall SegmentManager quiescence, verify callers return without releasing the slot lock, and confirm deferred cleanup releases the slot after quiescence resumes.

Add narrow test-only seams for engine injection and I/O thread joining so the tests avoid a live server and wait for the complete exit path.

Author: bluestreak01

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

@@ -108,6 +108,7 @@ public final class BackgroundDrainer implements Runnable {
     private final String slotPath;
     /** Latest known {@code engine.ackedFsn()}; published for visibility. */
     private volatile long ackedFsn = -1L;
+    private CursorSendEngine engineForTesting;
     private volatile String lastErrorMessage;
     /**
      * Optional observer for durable-ack-unavailable transients and the
@@ -530,8 +531,10 @@ public void run() {
             // holds it, the engine constructor throws and we exit silently
             // (no .failed sentinel — contention is expected, not an error).
             try {
-                engine = new CursorSendEngine(slotPath, segmentSizeBytes,
-                        sfMaxTotalBytes, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS);
+                engine = engineForTesting != null
+                        ? engineForTesting
+                        : new CursorSendEngine(slotPath, segmentSizeBytes,
+                                sfMaxTotalBytes, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS);
             } catch (IllegalStateException t) {
                 String msg = t.getMessage();
                 if (msg != null && msg.contains("already in use")) {
@@ -725,6 +728,15 @@ public void run() {
         }
     }
 
+    /**
+     * Replaces the engine created by {@link #run()} so tests can drive the
+     * production teardown path with a controlled manager.
+     */
+    @TestOnly
+    public void setEngineForTesting(CursorSendEngine engineForTesting) {
+        this.engineForTesting = engineForTesting;
+    }
+
     /**
      * Plug an observer for durable-ack-related events. {@code null} clears
      * any previously installed listener. See {@link BackgroundDrainerListener}

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

@@ -842,6 +842,15 @@ public boolean delegateEngineClose() {
         return shutdownLatch.getCount() != 0L;
     }
 
+    /**
+     * Returns the I/O thread so lifecycle tests can wait for its exit path,
+     * which continues beyond the shutdown-latch countdown.
+     */
+    @TestOnly
+    public Thread getIoThreadForTesting() {
+        return ioThread;
+    }
+
     /**
      * Typed server-rejection payload of the latched terminal error, or
      * {@code null} when the loop latched a wire-level failure (or nothing).