fix(qwp): free native path-scratch sink when engine/manager construction fails

Two constructor-throw paths leaked SegmentManager's 256-byte native
pathScratch sink (NATIVE_DIRECT_UTF8_SINK) per failed construction:

- CursorSendEngine: the delegating constructors evaluate
  new SegmentManager(...) BEFORE the private ctor body runs, so a
  pre-try throw (slot lock collision -- 'already in use', or empty
  sfDir) orphaned the owned manager. A submit/close race in the
  drainer pool leaked 1546 sinks (395,776 bytes) in a single test run;
  any client retrying fromConfig against a locked slot leaks 256 bytes
  per attempt. The pre-try section now closes an owned manager before
  propagating, and the main catch closes it regardless of whether
  start() was reached (close on an unstarted manager is safe -- no
  worker to join, just frees the scratch).

- SegmentManager: the pathScratch field initializer has already
  allocated by the time the ctor's validation throws fire
  (segmentSizeBytes/maxTotalBytes too small). Both throws now close
  the sink first.

Found by tightening the test leak-check (follow-up commits); pinned by
the existing suite going green under strict per-tag accounting.

Author: bluestreak01

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

@@ -145,15 +145,30 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
         boolean memoryMode = sfDir == null;
         SlotLock acquiredLock = null;
         if (!memoryMode) {
-            if (sfDir.isEmpty()) {
-                throw new IllegalArgumentException("sfDir must not be empty");
+            try {
+                if (sfDir.isEmpty()) {
+                    throw new IllegalArgumentException("sfDir must not be empty");
+                }
+                // Acquire the slot lock BEFORE we touch any *.sfa files. Two
+                // engines pointed at the same slot would otherwise race on
+                // recovery and create overlapping FSN ranges. SlotLock.acquire
+                // also creates the slot dir if it doesn't exist yet — no
+                // separate mkdir step needed here.
+                acquiredLock = SlotLock.acquire(sfDir);
+            } catch (Throwable t) {
+                // The delegating constructors evaluate `new SegmentManager(...)`
+                // BEFORE this body runs, so on a pre-try throw (e.g. slot lock
+                // collision) an owned manager is already alive and would leak
+                // its native path-scratch sink -- 256 bytes per failed
+                // construction attempt. Close it before propagating.
+                if (ownsManager) {
+                    try {
+                        manager.close();
+                    } catch (Throwable ignored) {
+                    }
+                }
+                throw t;
             }
-            // Acquire the slot lock BEFORE we touch any *.sfa files. Two
-            // engines pointed at the same slot would otherwise race on
-            // recovery and create overlapping FSN ranges. SlotLock.acquire
-            // also creates the slot dir if it doesn't exist yet — no
-            // separate mkdir step needed here.
-            acquiredLock = SlotLock.acquire(sfDir);
         }
         this.slotLock = acquiredLock;
         this.sfDir = sfDir;
@@ -168,7 +183,6 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
         // reference instead of orphaning the mmap'd segments + fds.
         SegmentRing ringInProgress = null;
         AckWatermark watermarkInProgress = null;
-        boolean managerStarted = false;
         try {
             // Disk mode: try to recover any *.sfa files left behind by a prior
             // session before deciding to start fresh. Without this the engine
@@ -277,7 +291,6 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
 
             if (ownsManager) {
                 manager.start();
-                managerStarted = true;
             }
             manager.register(ringInProgress, sfDir, watermarkInProgress);
             // All construction succeeded — commit the ring and
@@ -288,7 +301,10 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
             // Stop an owned manager before freeing the ring and watermark it may
             // touch, then release the slot lock. Each cleanup is in its own
             // try/catch so a single failure doesn't strand later cleanups.
-            if (ownsManager && managerStarted) {
+            // Closing an owned-but-never-started manager is safe (no worker to
+            // join) and required: skipping it leaked the manager's native
+            // path-scratch sink whenever construction failed before start().
+            if (ownsManager) {
                 try {
                     manager.close();
                 } catch (Throwable ignored) {

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

@@ -143,10 +143,16 @@ public SegmentManager(long segmentSizeBytes, long pollNanos) {
      *                         hold an initial active plus one hot spare.
      */
     public SegmentManager(long segmentSizeBytes, long pollNanos, long maxTotalBytes) {
+        // The pathScratch field initializer has already allocated its native
+        // buffer by the time this body runs, so a validation throw must free
+        // it or every failed construction leaks 256 bytes of native memory
+        // (e.g. a drainer retry loop hitting the same bad config).
         if (segmentSizeBytes < MmapSegment.HEADER_SIZE + MmapSegment.FRAME_HEADER_SIZE + 1) {
+            pathScratch.close();
             throw new IllegalArgumentException("segmentSizeBytes too small: " + segmentSizeBytes);
         }
         if (maxTotalBytes < segmentSizeBytes) {
+            pathScratch.close();
             throw new IllegalArgumentException(
                     "maxTotalBytes (" + maxTotalBytes + ") must allow at least one segment of "
                             + segmentSizeBytes + " bytes");