Merge branch 'vi_sf' of https://github.com/questdb/java-questdb-client into vi_sf

Author: bluestreak01

ci/run_tests_pipeline.yaml

@@ -84,7 +84,16 @@ stages:
                 maven | "$(Agent.OS)"
               path: $(HOME)/.m2/repository
             displayName: "Cache Maven repository"
-          - script: git clone --depth 1 https://github.com/questdb/questdb.git ./questdb
+          - bash: |
+              BRANCH="${SYSTEM_PULLREQUEST_SOURCEBRANCH:-$BUILD_SOURCEBRANCHNAME}"
+              BRANCH="${BRANCH#refs/heads/}"
+              if git ls-remote --exit-code --heads https://github.com/questdb/questdb.git "$BRANCH" >/dev/null 2>&1; then
+                echo "Cloning matching questdb branch: $BRANCH"
+                git clone --depth 1 --branch "$BRANCH" https://github.com/questdb/questdb.git ./questdb
+              else
+                echo "No matching questdb branch '$BRANCH', falling back to master"
+                git clone --depth 1 https://github.com/questdb/questdb.git ./questdb
+              fi
             displayName: git clone questdb
           - task: Maven@3
             displayName: "Update client version"

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

@@ -25,6 +25,7 @@
 package io.questdb.client.test.cutlass.qwp.client;
 
 import io.questdb.client.Sender;
+import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import org.junit.Assert;
 import org.junit.Test;
@@ -142,8 +143,8 @@ public void testCloseFastWhenTimeoutIsMinusOne() throws Exception {
     @Test
     public void testCloseDrainTimesOutWhenAcksNeverArrive() throws Exception {
         // Server that buffers frames silently and never ACKs. close() must
-        // return after roughly the configured timeout — not hang forever
-        // and not return immediately.
+        // throw a drain-timeout LineSenderException after roughly the
+        // configured timeout — not hang forever and not return immediately.
         int port = TEST_PORT + 3;
         long timeoutMs = 500;
         SilentHandler handler = new SilentHandler();
@@ -154,12 +155,21 @@ public void testCloseDrainTimesOutWhenAcksNeverArrive() throws Exception {
             String cfg = "ws::addr=localhost:" + port
                     + ";close_flush_timeout_millis=" + timeoutMs + ";";
             long elapsedMs;
-            try (Sender sender = Sender.fromConfig(cfg)) {
+            Sender sender = Sender.fromConfig(cfg);
+            try {
                 sender.table("foo").longColumn("v", 1L).atNow();
                 sender.flush();
                 long t0 = System.nanoTime();
-                sender.close();
+                try {
+                    sender.close();
+                    Assert.fail("close() should have thrown a drain-timeout error");
+                } catch (LineSenderException e) {
+                    Assert.assertTrue("expected drain-timeout message, got: " + e.getMessage(),
+                            e.getMessage().contains("drain timed out"));
+                }
                 elapsedMs = (System.nanoTime() - t0) / 1_000_000;
+            } finally {
+                sender.close(); // idempotent — closed flag is set on first call
             }
             Assert.assertTrue("close() returned too early: " + elapsedMs + "ms",
                     elapsedMs >= timeoutMs);

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

@@ -25,6 +25,7 @@
 package io.questdb.client.test.cutlass.qwp.client;
 
 import io.questdb.client.Sender;
+import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import org.junit.Assert;
 import org.junit.Test;
@@ -119,7 +120,8 @@ public void testReconnectGivesUpAfterCap() throws Exception {
                     + ";reconnect_initial_backoff_millis=10"
                     + ";reconnect_max_backoff_millis=50"
                     + ";close_flush_timeout_millis=0;";
-            try (Sender sender = Sender.fromConfig(cfg)) {
+            Sender sender = Sender.fromConfig(cfg);
+            try {
                 sender.table("foo").longColumn("v", 1L).atNow();
                 sender.flush();
 
@@ -151,6 +153,13 @@ public void testReconnectGivesUpAfterCap() throws Exception {
                         msg.contains("reconnect failed")
                                 || msg.contains("I/O thread failed")
                                 || msg.contains("Failed to connect"));
+            } finally {
+                // close() rethrows the latched terminal reconnect-cap error
+                // (commit 052f6ee). Already observed and asserted above.
+                try {
+                    sender.close();
+                } catch (LineSenderException ignored) {
+                }
             }
         } finally {
             try {
@@ -177,7 +186,8 @@ public void testTerminalUpgradeErrorAbortsReconnect() throws Exception {
             String cfg = "ws::addr=localhost:" + port
                     + ";reconnect_max_duration_millis=10000"
                     + ";close_flush_timeout_millis=0;";
-            try (Sender sender = Sender.fromConfig(cfg)) {
+            Sender sender = Sender.fromConfig(cfg);
+            try {
                 sender.table("foo").longColumn("v", 1L).atNow();
                 sender.flush();
                 // Wait for first connection to ACK + close
@@ -209,6 +219,13 @@ public void testTerminalUpgradeErrorAbortsReconnect() throws Exception {
                         msg.contains("WebSocket upgrade failed")
                                 || msg.contains("I/O thread failed")
                                 || msg.contains("401"));
+            } finally {
+                // close() rethrows the latched terminal upgrade error
+                // (commit 052f6ee). Already observed and asserted above.
+                try {
+                    sender.close();
+                } catch (LineSenderException ignored) {
+                }
             }
         }
     }

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

@@ -82,7 +82,8 @@ public void testServerErrorAckIsTerminalAndDoesNotBurnReconnectBudget() throws E
                     + ";reconnect_max_backoff_millis=50"
                     + ";";
 
-            try (Sender sender = Sender.fromConfig(cfg)) {
+            Sender sender = Sender.fromConfig(cfg);
+            try {
                 sender.table("foo").longColumn("v", 1L).atNow();
                 sender.flush();
 
@@ -125,6 +126,14 @@ public void testServerErrorAckIsTerminalAndDoesNotBurnReconnectBudget() throws E
                         thrown.getMessage() != null
                                 && (thrown.getMessage().contains("rejected")
                                     || thrown.getMessage().contains("error")));
+            } finally {
+                // close() rethrows the latched terminal server-rejection error
+                // (commit 052f6ee). Swallow it here — the test has already
+                // observed and asserted on that error via flush() above.
+                try {
+                    sender.close();
+                } catch (LineSenderException ignored) {
+                }
             }
         }
     }

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

@@ -119,13 +119,16 @@ public void testScanFindsOrphanFromPriorSenderUnderSameGroupRoot() throws Except
                     primary.table("foo").longColumn("v", 8L).atNow();
                     primary.flush();
                 }
-                // Primary's slot now exists too; scanner with primary
-                // excluded must still return the ghost (and nothing else
-                // among the two slots).
+                // With drain_orphans=true, the background drainer pool adopts
+                // the ghost slot, replays its unacked frames against the now-
+                // ACKing primaryServer, and removes the drained slot dir.
+                // Primary's own slot drains cleanly on close() and is filtered
+                // out by sender_id. Net: scanner sees neither.
                 ObjList<String> postRun = OrphanScanner.scan(sfDir, "primary");
-                Assert.assertEquals("only ghost should appear; primary excluded",
-                        1, postRun.size());
-                Assert.assertEquals(sfDir + "/ghost", postRun.get(0));
+                Assert.assertEquals(
+                        "drain_orphans=true should have drained + removed the "
+                                + "ghost slot; primary's own slot is sender_id-filtered",
+                        0, postRun.size());
             } finally {
                 try {
                     primaryServer.close();

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

@@ -25,6 +25,8 @@
 package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
 
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentManager;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentRing;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLock;
 import io.questdb.client.std.Files;
 import io.questdb.client.test.tools.TestUtils;
@@ -119,17 +121,36 @@ public void testSlotLockReleasedEvenIfRingCloseThrows() throws Exception {
             // Sabotage: zero out ring so engine.close() NPEs before reaching
             // the slotLock cleanup. Any close-path exception (manager.close,
             // ring.close, unlinkAllSegmentFiles) lands in the same place.
+            //
+            // Capture the ring + manager references first so we can free
+            // their native resources ourselves after the sabotage — engine.close()
+            // can no longer reach ring.close() / manager.close() once we null
+            // the ring field, and assertMemoryLeak (+ the manager's worker
+            // thread) would otherwise trip.
             Field ringField = CursorSendEngine.class.getDeclaredField("ring");
             ringField.setAccessible(true);
+            SegmentRing capturedRing = (SegmentRing) ringField.get(engine);
+
+            Field managerField = CursorSendEngine.class.getDeclaredField("manager");
+            managerField.setAccessible(true);
+            SegmentManager capturedManager = (SegmentManager) managerField.get(engine);
+
             ringField.set(engine, null);
 
             try {
                 engine.close();
             } catch (Throwable t) {
-                // Expected — close() walks ring.close() and trips an NPE.
+                // Expected — close() walks ring.publishedFsn() and trips an NPE.
                 // The fix must release slotLock anyway, in finally.
             }
 
+            // Manually release the ring + manager resources that engine.close()
+            // skipped because of the NPE. The slotLock contract is the only
+            // thing the test is verifying; the rest of the close-path resources
+            // are an artifact of the sabotage.
+            capturedRing.close();
+            capturedManager.close();
+
             // The user-visible test: can a fresh SlotLock acquire the
             // same slot? If the original lock fd is still held, the
             // kernel's flock blocks this acquire and we throw.