fix(qwp): interrupted drainer-pool close leaves drainers busy-spinning at 100% CPU

BackgroundDrainerPool.close()'s InterruptedException arm called only
executor.shutdownNow() -- never requestStop() on the active drainers. The
pre-shutdown split-stop sweep deliberately spares actively-draining
drainers (ackedFsn >= 0), and the graceful-timeout sweep never runs when
awaitTermination throws instead of returning false, so on this path an
actively-draining drainer kept stopRequested == false while shutdownNow's
interrupt turned every LockSupport.parkNanos in its poll/backoff loops
into an immediate return (park does not clear the status): a 100% CPU
busy-spin for as long as the outage lasted, with the slot's on-disk lock
pinned against re-adoption.

Fixed in two layers:
- pool: the interrupted arm sweeps active -> requestStop() before
  shutdownNow -- the authoritative signal, synchronous with close();
- drainer: the park loops fold a pending interrupt into stopRequested
  (stopRequestedOrInterrupted), so ANY interrupt reaching a drainer
  thread without the flag routes through the normal STOPPED exit instead
  of the spin -- the pairing is no longer caller discipline. The status
  is deliberately left set (isInterrupted, not Thread.interrupted): the
  interrupted-teardown delegation protocol depends on loop.close()'s
  latch await throwing under a wedged I/O thread
  (BackgroundDrainerInterruptedTeardownTest).

A drainer cut down mid-drain exits STOPPED with its unacked rows still in
SF -- re-adopted by the next orphan scan, never dropped -- matching the
graceful-timeout sweep's semantics.

Both regression tests are red pre-fix: the pool test fails the
synchronous stop-signal pin after an interrupted close(), and the
drainer test joins out after 10s against a spinning connect-retry loop.

Author: bluestreak01

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

@@ -435,7 +435,7 @@ public WebSocketClient connectWithDurableAckRetry() {
             if (sleepMillis > 0L && !stopRequested) {
                 long parkDeadlineNanos = System.nanoTime() + sleepMillis * 1_000_000L;
                 long remaining;
-                while (!stopRequested
+                while (!stopRequestedOrInterrupted()
                         && (remaining = parkDeadlineNanos - System.nanoTime()) > 0L) {
                     LockSupport.parkNanos(Math.min(remaining, STOP_CHECK_PARK_CHUNK_NANOS));
                 }
@@ -480,6 +480,37 @@ public boolean isStopRequested() {
         return stopRequested;
     }
 
+    /**
+     * Stop check for the runner thread's park loops that also folds a
+     * pending thread interrupt into the stop protocol. The pool delivers
+     * cancellation as an interrupt ({@code shutdownNow}) and pairs it with a
+     * {@link #requestStop()} sweep — but that pairing is caller discipline.
+     * An interrupt arriving WITHOUT the flag would otherwise be pathological:
+     * every wait here is a {@code LockSupport.parkNanos}, which returns
+     * immediately while the status is pending and never clears it, so the
+     * backoff/poll loops would degrade into a 100% CPU busy-spin (for as
+     * long as an outage lasts) with the slot lock pinned. Mapping the
+     * interrupt onto {@code stopRequested} routes them through the normal
+     * STOPPED exit instead.
+     * <p>
+     * The status is deliberately left set ({@code isInterrupted()}, not
+     * {@code Thread.interrupted()}): the teardown in {@link #run}'s finally
+     * relies on it — {@code loop.close()}'s latch await must throw rather
+     * than block under a wedged I/O thread, routing engine teardown through
+     * the delegation protocol (pinned by
+     * {@code BackgroundDrainerInterruptedTeardownTest}).
+     * <p>
+     * Called on the runner thread only. The unsynchronized check-then-set is
+     * safe against a concurrent {@link #requestStop()}: both writers only
+     * ever transition {@code stopRequested} false→true.
+     */
+    private boolean stopRequestedOrInterrupted() {
+        if (!stopRequested && Thread.currentThread().isInterrupted()) {
+            stopRequested = true;
+        }
+        return stopRequested;
+    }
+
     @Override
     public void run() {
         runnerThread = Thread.currentThread();
@@ -539,7 +570,7 @@ public void run() {
                         maxHeadFrameRejections);
                 loop.start();
 
-                while (!stopRequested) {
+                while (!stopRequestedOrInterrupted()) {
                     long acked = engine.ackedFsn();
                     this.ackedFsn = acked;
                     if (acked >= target) {

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

@@ -54,7 +54,11 @@
  * that don't exit in time (typically parked in a blocking native connect
  * that neither unpark nor interrupt cancels) are left to finish on their
  * own — the pool's underlying executor uses daemon threads so they don't
- * block JVM exit.
+ * block JVM exit. An interrupted {@code close()} skips the graceful
+ * window: every active drainer is stop-signaled immediately, then the
+ * executor is shut down hard. A drainer cut down mid-drain exits STOPPED
+ * with its unacked rows still in SF — re-adopted by the next orphan scan,
+ * never dropped.
  */
 public final class BackgroundDrainerPool implements QuietCloseable {
 
@@ -177,6 +181,22 @@ public void close() {
             }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
+            // Signal stop BEFORE shutdownNow's interrupts land. Actively-
+            // draining drainers were spared by the split-stop above, and the
+            // graceful-timeout sweep never runs on this path -- without this
+            // sweep they would rely solely on the drainer-side fallback that
+            // folds a pending interrupt into stopRequested
+            // (BackgroundDrainer.stopRequestedOrInterrupted). This sweep is
+            // the authoritative signal: it is synchronous with close() (the
+            // caller observes fully stop-signaled drainers on return) and
+            // covers a drainer that is between park-loop checks when the
+            // interrupt lands. Cutting a mid-drain drainer short here is
+            // deliberate -- an interrupted close() means "leave now", and
+            // its unacked rows stay in SF for the next orphan scan (no data
+            // loss), exactly like the graceful-timeout sweep.
+            for (BackgroundDrainer d : active) {
+                d.requestStop();
+            }
             executor.shutdownNow();
         }
     }

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerInterruptIsStopSignalTest.java

@@ -0,0 +1,157 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  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.sf.cursor;
+
+import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.std.Files;
+import io.questdb.client.std.MemoryTag;
+import io.questdb.client.std.Unsafe;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Pins the drainer-side half of the interrupted-shutdown hardening: a bare
+ * thread interrupt — no {@link BackgroundDrainer#requestStop()} — must act
+ * as a stop signal, not degrade the drainer into a busy-spin.
+ * <p>
+ * {@code BackgroundDrainerPool.close()} pairs every executor interrupt
+ * ({@code shutdownNow}) with a {@code requestStop()} sweep, but that pairing
+ * is caller discipline: any interrupt reaching a drainer thread without the
+ * flag (a future shutdownNow call site, a stray interrupt on the pool's
+ * threads) would otherwise hit loops whose only wait primitive is
+ * {@code LockSupport.parkNanos} — which returns immediately while the
+ * interrupt status is pending and never clears it. With the status pending
+ * and {@code stopRequested} false, the backoff/poll loops spin at 100% CPU
+ * for as long as the outage lasts, holding the slot lock. The drainer
+ * therefore folds a pending interrupt into {@code stopRequested} at its
+ * park sites (status deliberately left set — the interrupted-teardown
+ * delegation protocol depends on it, see
+ * {@code BackgroundDrainerInterruptedTeardownTest}).
+ */
+public class BackgroundDrainerInterruptIsStopSignalTest {
+
+    private static final long SEGMENT_BYTES = 1L << 20;
+
+    private String tmpDir;
+
+    @Before
+    public void setUp() {
+        tmpDir = Paths.get(System.getProperty("java.io.tmpdir"),
+                "qdb-int-stop-" + System.nanoTime()).toString();
+        Assert.assertEquals(0, Files.mkdir(tmpDir, Files.DIR_MODE_DEFAULT));
+    }
+
+    @After
+    public void tearDown() {
+        if (tmpDir == null) return;
+        long find = Files.findFirst(tmpDir);
+        if (find > 0) {
+            try {
+                int rc = 1;
+                while (rc > 0) {
+                    String name = Files.utf8ToString(Files.findName(find));
+                    if (name != null && !".".equals(name) && !"..".equals(name)) {
+                        Files.remove(tmpDir + "/" + name);
+                    }
+                    rc = Files.findNext(find);
+                }
+            } finally {
+                Files.findClose(find);
+            }
+        }
+        Files.remove(tmpDir);
+    }
+
+    @Test(timeout = 60_000)
+    public void bareInterruptStopsConnectPhaseDrainer() throws Exception {
+        // Slot with one published, unacked frame so the drainer opens a real
+        // engine (takes the slot lock) and enters its connect-retry loop.
+        long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
+        try {
+            CursorSendEngine prep = new CursorSendEngine(tmpDir, SEGMENT_BYTES);
+            try {
+                for (int i = 0; i < 16; i++) {
+                    Unsafe.getUnsafe().putByte(buf + i, (byte) i);
+                }
+                Assert.assertEquals(0L, prep.appendBlocking(buf, 16));
+            } finally {
+                prep.close();
+            }
+        } finally {
+            Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
+        }
+
+        // Unreachable cluster: plain transport failures retried indefinitely
+        // under Invariant B -- absent a stop signal the drainer never exits.
+        final CountDownLatch firstAttempt = new CountDownLatch(1);
+        final BackgroundDrainer drainer = new BackgroundDrainer(
+                tmpDir, SEGMENT_BYTES, Long.MAX_VALUE,
+                () -> {
+                    firstAttempt.countDown();
+                    throw new IOException("connection refused (test)");
+                },
+                5_000L, 1L, 10L, false, 0L);
+
+        Thread runner = new Thread(drainer::run, "drainer-runner");
+        runner.setDaemon(true);
+        runner.start();
+        try {
+            Assert.assertTrue("drainer never reached its connect loop",
+                    firstAttempt.await(10, TimeUnit.SECONDS));
+
+            // Bare interrupt -- deliberately NOT paired with requestStop().
+            runner.interrupt();
+
+            // The mapping must fold the pending status into stopRequested and
+            // exit STOPPED. Pre-hardening: parks are no-ops, the flag never
+            // rises, outcome stays PENDING while the loop spins -- this join
+            // times out.
+            runner.join(10_000L);
+            Assert.assertFalse(
+                    "a bare interrupt did not stop the drainer: with the status pending "
+                            + "every parkNanos is an immediate return, so the connect-retry "
+                            + "loop busy-spins at 100% CPU holding the slot lock",
+                    runner.isAlive());
+            Assert.assertEquals("interrupt must map onto the normal STOPPED exit",
+                    BackgroundDrainer.DrainOutcome.STOPPED, drainer.outcome());
+            Assert.assertTrue("interrupt must be folded into stopRequested",
+                    drainer.isStopRequested());
+            // Engine closed on the way out: the slot lock must be free.
+            new CursorSendEngine(tmpDir, SEGMENT_BYTES).close();
+        } finally {
+            // Red-run cleanup: unhook a still-spinning drainer so a failing
+            // assertion doesn't leave a hot daemon thread behind.
+            drainer.requestStop();
+        }
+    }
+}