test(client): replace reflection with typed test seams

Author: bluestreak01

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

@@ -1292,6 +1292,11 @@ public void setSlotLockReleaseListener(Runnable listener) {
         }
     }
 
+    @TestOnly
+    public void setSlotLockReleasedForTesting(boolean isReleased) {
+        slotLockReleased = isReleased;
+    }
+
     @Override
     public Sender decimalColumn(CharSequence name, Decimal64 value) {
         checkNotClosed();
@@ -1710,16 +1715,31 @@ public SenderConnectionDispatcher getConnectionDispatcherForTesting() {
         return connectionDispatcher;
     }
 
+    @TestOnly
+    public CursorSendEngine getCursorEngineForTesting() {
+        return cursorEngine;
+    }
+
     @TestOnly
     public SenderErrorDispatcher getErrorDispatcherForTesting() {
         return errorDispatcher;
     }
 
+    @TestOnly
+    public Sender.InitialConnectMode getInitialConnectModeForTesting() {
+        return initialConnectMode;
+    }
+
     @TestOnly
     public SenderProgressDispatcher getProgressDispatcherForTesting() {
         return progressDispatcher;
     }
 
+    @TestOnly
+    public Runnable getSlotLockReleaseListenerForTesting() {
+        return slotLockReleaseListener;
+    }
+
     /**
      * Number of {@link SenderError} notifications dropped because the
      * bounded inbox was full. Non-zero means the user-supplied
@@ -2151,6 +2171,11 @@ public void setClientForTesting(WebSocketClient client) {
         this.client = client;
     }
 
+    @TestOnly
+    public void setClosedForTesting(boolean isClosed) {
+        this.closed = isClosed;
+    }
+
     /**
      * Installs a one-shot test witness that close-time drain invokes after it
      * observes a real unacknowledged target and before it starts waiting.

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

@@ -30,6 +30,7 @@
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.QuietCloseable;
 import io.questdb.client.std.Unsafe;
+import org.jetbrains.annotations.TestOnly;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -102,6 +103,7 @@ public final class AckWatermark implements QuietCloseable {
     private boolean closed;
     private long fsn;
     private long generation;
+    private boolean isStorageReleased;
 
     private AckWatermark(FilesFacade filesFacade, int fd, long mmapAddress,
                          long generation, long fsn) {
@@ -116,12 +118,7 @@ private AckWatermark(FilesFacade filesFacade, int fd, long mmapAddress,
     public void close() {
         if (closed) return;
         closed = true;
-        if (mmapAddress != 0L && mmapAddress != Files.FAILED_MMAP_ADDRESS) {
-            Files.munmap(mmapAddress, FILE_SIZE, MemoryTag.MMAP_DEFAULT);
-        }
-        if (fd >= 0) {
-            filesFacade.close(fd);
-        }
+        releaseStorage();
     }
 
     /**
@@ -171,6 +168,16 @@ static AckWatermark open(FilesFacade filesFacade, String slotDir) {
                 : new AckWatermark(filesFacade, fd, addr, selected.generation, selected.fsn);
     }
 
+    /**
+     * Releases the native storage while deliberately leaving the logical
+     * closed flag clear. Test-only: recreates a stale racing writer without
+     * reflective access to descriptor and mapping internals.
+     */
+    @TestOnly
+    public boolean releaseStorageButKeepWritableForTesting() {
+        return releaseStorage();
+    }
+
     /**
      * Best-effort removal of a stale watermark file. Used by the engine
      * startup path when no segments are recovered.
@@ -240,6 +247,20 @@ public void write(long fsn) {
         this.fsn = fsn;
     }
 
+    private boolean releaseStorage() {
+        if (isStorageReleased) {
+            return false;
+        }
+        isStorageReleased = true;
+        if (mmapAddress != 0L && mmapAddress != Files.FAILED_MMAP_ADDRESS) {
+            Files.munmap(mmapAddress, FILE_SIZE, MemoryTag.MMAP_DEFAULT);
+        }
+        if (fd >= 0) {
+            filesFacade.close(fd);
+        }
+        return true;
+    }
+
     private static Record readRecord(long address) {
         if (Unsafe.getUnsafe().getInt(address + MAGIC_OFFSET) != FILE_MAGIC
                 || Unsafe.getUnsafe().getInt(address + 4) != VERSION) {

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

@@ -900,6 +900,16 @@ private void finishClose(boolean fullyDrained) {
         }
     }
 
+    @TestOnly
+    public SegmentManager getManagerForTesting() {
+        return manager;
+    }
+
+    @TestOnly
+    public SlotLock getSlotLockForTesting() {
+        return slotLock;
+    }
+
     /**
      * Installs a hook invoked after each failed shared-driver flock release.
      * Test-only: makes persistent retry progress observable without sleeps.

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

@@ -583,6 +583,31 @@ public void register(SegmentRing ring, String dir, AckWatermark watermark) {
         wakeWorker();
     }
 
+    @TestOnly
+    public SegmentRing getInServiceRingForTesting() {
+        RingEntry entry = inService;
+        return entry == null ? null : entry.ring;
+    }
+
+    @TestOnly
+    public long getTotalBytesForTesting() {
+        synchronized (lock) {
+            return totalBytes;
+        }
+    }
+
+    @TestOnly
+    public Thread getWorkerThreadForTesting() {
+        return workerThread;
+    }
+
+    @TestOnly
+    public boolean isPathScratchAllocatedForTesting() {
+        synchronized (lock) {
+            return !scratchFreed;
+        }
+    }
+
     @TestOnly
     public void setAfterRingCleanupHook(Runnable hook) {
         this.afterRingCleanupHook = hook;

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

@@ -1229,6 +1229,11 @@ synchronized int pendingTrimCount() {
         return pendingTrims.size();
     }
 
+    @TestOnly
+    public synchronized MmapSegment getHotSpareForTesting() {
+        return hotSpare;
+    }
+
     @TestOnly
     public synchronized int getPendingTrimCount() {
         return pendingTrims.size();

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

@@ -29,6 +29,7 @@
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.QuietCloseable;
 import io.questdb.client.std.Unsafe;
+import org.jetbrains.annotations.TestOnly;
 
 import java.nio.charset.StandardCharsets;
 
@@ -56,6 +57,7 @@
  */
 public final class SlotLock implements QuietCloseable {
 
+    private static final int DEAD_FD_FOR_TESTING = 1_000_000_000;
     private static final String LOCK_FILE_NAME = ".lock";
     private static final String LOCK_PID_FILE_NAME = ".lock.pid";
     private final String slotDir;
@@ -111,6 +113,26 @@ public static SlotLock acquire(String slotDir) {
         }
     }
 
+    /**
+     * Replaces the live descriptor with a known-dead value until the returned
+     * guard closes. Test-only: exercises release retry paths without exposing
+     * mutable descriptor state.
+     */
+    @TestOnly
+    public synchronized ReleaseFailureForTesting injectReleaseFailureForTesting() {
+        if (fd < 0 || fd == DEAD_FD_FOR_TESTING) {
+            throw new IllegalStateException("slot lock is not held by a live descriptor");
+        }
+        ReleaseFailureForTesting releaseFailure = new ReleaseFailureForTesting(fd);
+        fd = DEAD_FD_FOR_TESTING;
+        return releaseFailure;
+    }
+
+    @TestOnly
+    public synchronized boolean isReleaseFailureInjectedForTesting() {
+        return fd == DEAD_FD_FOR_TESTING;
+    }
+
     /** Slot dir this lock guards. */
     public String slotDir() {
         return slotDir;
@@ -183,6 +205,13 @@ private static String readHolder(String pidPath) {
 
     private static native int release0(int fd);
 
+    private synchronized void restoreFdForTesting(int savedFd) {
+        if (fd != DEAD_FD_FOR_TESTING) {
+            throw new IllegalStateException("slot lock release failure is not injected");
+        }
+        fd = savedFd;
+    }
+
     private static void writePid(String pidPath) {
         long pid;
         try {
@@ -212,4 +241,22 @@ private static void writePid(String pidPath) {
             Files.close(wfd);
         }
     }
+
+    @TestOnly
+    public final class ReleaseFailureForTesting implements QuietCloseable {
+        private final int savedFd;
+        private boolean isRestored;
+
+        private ReleaseFailureForTesting(int savedFd) {
+            this.savedFd = savedFd;
+        }
+
+        @Override
+        public synchronized void close() {
+            if (!isRestored) {
+                restoreFdForTesting(savedFd);
+                isRestored = true;
+            }
+        }
+    }
 }

core/src/main/java/io/questdb/client/impl/PooledSender.java

@@ -32,6 +32,7 @@
 import io.questdb.client.std.Decimal64;
 import io.questdb.client.std.bytes.DirectByteSlice;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.TestOnly;
 
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
@@ -380,6 +381,21 @@ long generation() {
         return generation;
     }
 
+    @TestOnly
+    public Sender getDelegateForTesting() {
+        return slot.delegate();
+    }
+
+    @TestOnly
+    public int getSlotIndexForTesting() {
+        return slot.slotIndex();
+    }
+
+    @TestOnly
+    public boolean hasSameSlotForTesting(PooledSender that) {
+        return slot == that.slot;
+    }
+
     SenderSlot slot() {
         return slot;
     }

core/src/main/java/io/questdb/client/impl/QueryClientPool.java

@@ -528,6 +528,16 @@ void release(QueryWorker w, long gen) {
         }
     }
 
+    @TestOnly
+    public boolean hasCreationWaiterForTesting() {
+        lock.lock();
+        try {
+            return lock.hasWaiters(creationFinished);
+        } finally {
+            lock.unlock();
+        }
+    }
+
     // White-box accessor for tests: reports the current in-flight creation count
     // under the pool lock. A non-zero value after a failed acquire() means the
     // slot reservation was never released -- the capacity-shrink bug this guards
@@ -542,6 +552,11 @@ public int inFlightCreations() {
         }
     }
 
+    @TestOnly
+    public boolean isClosedForTesting() {
+        return closed;
+    }
+
     private QueryWorker createUnlocked() {
         QwpQueryClient client = QwpQueryClient.fromConfig(configurationString);
         try {

core/src/main/java/io/questdb/client/impl/QuestDBImpl.java

@@ -215,6 +215,16 @@ public synchronized void close() {
         closeQuietly(senderPool);
     }
 
+    @TestOnly
+    public QueryClientPool getQueryPoolForTesting() {
+        return queryPool;
+    }
+
+    @TestOnly
+    public SenderPool getSenderPoolForTesting() {
+        return senderPool;
+    }
+
     private static void closeQuietly(PoolHousekeeper housekeeper) {
         if (housekeeper == null) {
             return;