test(qwp): pin close()'s drain-to-commit-boundary for uncommitted deferred tails

Regression tests for c6eadf0 (stop close() waiting on acks of
uncommitted deferred frames), which shipped with the fix in 4
production files and no client-side test -- the only coverage was
testDeferredCommitConnectionDropRollsBack in the parent server repo,
where the pre-fix symptom was a 300s close() hang.

Two tests in CloseDrainTest, both verified red on the pre-fix
behavior (target = publishedFsn) and green on the fix:

- testCloseSkipsDrainForUncommittedDeferredTail: defer-commit mode,
  publish rows, never commit, close() against a server that never
  acks (what the real server does to a deferred tail by design).
  Fixed close() drains to min(publishedFsn, commitBoundary) = -1 and
  returns immediately with the abandonment WARN; broken close()
  throws 'drain timed out' after the full close_flush_timeout.

- testCloseDrainsToCommitBoundaryAndAbandonsDeferredTail: one
  committed frame below an uncommitted deferred tail, server acks
  only the commit-bearing frame (AckFirstFrameOnlyHandler). Pins
  that the boundary drain is not an opt-out of draining altogether:
  close() still waits for the committed frame's ack, then abandons
  the tail instead of timing out on it.

Author: bluestreak01

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/CloseDrainTest.java

@@ -26,6 +26,7 @@
 
 import io.questdb.client.Sender;
 import io.questdb.client.cutlass.line.LineSenderException;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import org.junit.Assert;
 import org.junit.Test;
@@ -174,6 +175,99 @@ public void testCloseDrainTimesOutWhenAcksNeverArrive() throws Exception {
         }
     }
 
+    @Test
+    public void testCloseSkipsDrainForUncommittedDeferredTail() throws Exception {
+        // Regression test for the close()-hang on abandoned deferred
+        // transactions: the server withholds acks for FLAG_DEFER_COMMIT
+        // frames until their group-closing commit lands, so a close()-time
+        // drain that targets publishedFsn (instead of the last commit
+        // boundary) can only ever time out -- 300s hangs in the e2e suite
+        // (testDeferredCommitConnectionDropRollsBack).
+        //
+        // Same producer sequence as that e2e test, against a server that
+        // never acks (which is exactly what the real server does to an
+        // uncommitted deferred tail): defer-commit mode, publish rows, no
+        // commit, close(). Fixed close() drains to
+        // min(publishedFsn, commitBoundary) = -1, abandons the tail with a
+        // WARN, and returns immediately. Broken close() targets the deferred
+        // frame and throws "drain timed out" after the full timeout.
+        long timeoutMs = 2000;
+        SilentHandler handler = new SilentHandler();
+        try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+            int port = server.getPort();
+            String cfg = "ws::addr=localhost:" + port
+                    + ";close_flush_timeout_millis=" + timeoutMs + ";";
+            try (Sender sender = Sender.fromConfig(cfg)) {
+                ((QwpWebSocketSender) sender).setDeferCommit(true);
+                sender.table("foo").longColumn("v", 1L).atNow();
+                sender.flush(); // publishes a deferred frame; commit never sent
+
+                long t0 = System.nanoTime();
+                try {
+                    sender.close();
+                } catch (LineSenderException e) {
+                    Assert.fail("close() must not wait for acks of an uncommitted deferred "
+                            + "tail (the server withholds them by design), but threw: "
+                            + e.getMessage());
+                }
+                long elapsedMs = (System.nanoTime() - t0) / 1_000_000;
+                Assert.assertTrue("close() took " + elapsedMs + "ms -- it drained toward the "
+                                + "uncommitted deferred tail instead of stopping at the commit "
+                                + "boundary",
+                        elapsedMs < timeoutMs / 2);
+            }
+        }
+    }
+
+    @Test
+    public void testCloseDrainsToCommitBoundaryAndAbandonsDeferredTail() throws Exception {
+        // Mixed case: one committed frame followed by an uncommitted deferred
+        // tail. close() must still wait for the committed frame's ack (the
+        // commit-boundary drain is not an opt-out of draining altogether) but
+        // must not wait for the deferred tail above it.
+        //
+        // The handler acks only the first data frame (the committed one) and
+        // stays silent above it -- mirroring the real server, which acks
+        // commit-bearing frames and withholds acks for deferred ones. Broken
+        // close() targets publishedFsn (the deferred tail) and throws "drain
+        // timed out"; fixed close() returns once the boundary frame is acked.
+        long timeoutMs = 2000;
+        AckFirstFrameOnlyHandler handler = new AckFirstFrameOnlyHandler();
+        try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+            int port = server.getPort();
+            String cfg = "ws::addr=localhost:" + port
+                    + ";close_flush_timeout_millis=" + timeoutMs + ";";
+            try (Sender sender = Sender.fromConfig(cfg)) {
+                sender.table("foo").longColumn("v", 1L).atNow();
+                sender.flush(); // commit-bearing frame FSN 0 -- boundary = 0, acked
+
+                ((QwpWebSocketSender) sender).setDeferCommit(true);
+                sender.table("foo").longColumn("v", 2L).atNow();
+                sender.flush(); // deferred frame FSN 1 -- never acked, never committed
+
+                long t0 = System.nanoTime();
+                try {
+                    sender.close();
+                } catch (LineSenderException e) {
+                    Assert.fail("close() must drain to the commit boundary (FSN 0) and abandon "
+                            + "the uncommitted deferred tail above it, but threw: "
+                            + e.getMessage());
+                }
+                long elapsedMs = (System.nanoTime() - t0) / 1_000_000;
+                Assert.assertTrue("close() took " + elapsedMs + "ms -- it drained toward the "
+                                + "uncommitted deferred tail instead of stopping at the acked "
+                                + "commit boundary",
+                        elapsedMs < timeoutMs / 2);
+            }
+        }
+    }
+
     @Test
     public void testDrainBlocksUntilAckArrivesAndReturnsTrue() throws Exception {
         // Public drain(timeoutMillis): explicit pre-close drain that the
@@ -411,6 +505,27 @@ public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] dat
         }
     }
 
+    /**
+     * Acks only the first data frame, then goes silent — models a server that
+     * acks the commit-bearing frame and withholds acks for the uncommitted
+     * deferred tail above it (the FLAG_DEFER_COMMIT ack contract).
+     */
+    private static class AckFirstFrameOnlyHandler implements TestWebSocketServer.WebSocketServerHandler {
+        private final AtomicLong received = new AtomicLong(0);
+
+        @Override
+        public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+            if (received.getAndIncrement() == 0) {
+                try {
+                    client.sendBinary(buildAck(0));
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+            }
+            // frames past the first are deferred: withhold their acks
+        }
+    }
+
     // Mirrors WebSocketResponse STATUS_OK layout: status u8 | sequence u64 | table_count u16
     static byte[] buildAck(long seq) {
         byte[] buf = new byte[1 + 8 + 2];