fix(ilp): fail fast on fatal SF errors instead of looping

Three correctness fixes from a follow-up review of the QWiP store-and-
forward client. Each is paired with a regression test that fails on the
unfixed code and passes after the change.

Critical:
- WebSocketSendQueue.retryStalled split its catch ladder so a fatal
  SfException during stall-retry (corruption, oversized frame, fsync
  EIO) is classified the same as the main-loop sendBatch catch:
  failConnection(_, true) terminal, not (_, false) reconnect. The old
  behaviour silently reconnected and recycled the buffer as if sent,
  hiding storage failures and risking infinite loops on persistent
  errors. (C1)

- SegmentLog.createActive registers the freshly-opened fd into the
  Segment before calling allocNativePath, and the try block now wraps
  the path-allocation call. The catch closes the fd and best-effort
  removes the orphan .sfa file. The previous order leaked one fd per
  failed rotation under OOM pressure. (C2)

- ResponseHandler.onBinaryMessage error branch now fails the connection
  fatally. A server-side per-batch error (parse, schema mismatch, write,
  security, internal) is a protocol-level rejection of specific bytes;
  reconnecting and re-sending the same payload produces the same error.
  Under SF the rejected frame sits on disk and replay-on-reconnect
  shipped it again, so the previous transient classification turned any
  poisoned frame into an unbounded reconnect loop. (C4)

Infrastructure:
- FilesFacade gains allocNativePath / freeNativePath. SegmentLog now
  routes all path-pointer alloc/free through the facade so tests can
  inject OOM at the exact moment between openCleanRW and the try block
  in createActive. Required for the C2 regression test.

Tests:
- testCreateActiveDoesNotLeakFdOnAllocNativePathOom (SegmentLogTest)
- testRetryStalledTreatsSfStorageErrorAsTerminal (SfIntegrationTest)
- testPoisonedFrameInSfDoesNotLoopForever (SfIntegrationTest)
- Full suite: 1971 tests pass (was 1968), zero regressions.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bluestreak01

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

@@ -970,8 +970,21 @@ private void retryStalled() {
                 // wants to wake the I/O thread cooperatively.
                 Thread.currentThread().interrupt();
             }
+        } catch (SfException sfe) {
+            // Fatal SF storage error during retry — same classification as
+            // the main-loop sendBatch catch (corruption, oversized frame,
+            // fsync EIO). Won't recover by reconnect; surface hard so the
+            // user sees it instead of looping.
+            LOG.error("Fatal SF storage error during retry [id={}]", batch.getBatchId(), sfe);
+            failConnection(new LineSenderException(
+                    "SF storage error: " + sfe.getMessage(), sfe), true);
+            if (batch.isSealed()) batch.markSending();
+            if (batch.isSending()) batch.markRecycled();
+            cleared = true;
         } catch (Throwable t) {
-            // Non-disk-full failure during retry — recycle and surface.
+            // Non-SF failure during retry (e.g. wire send error) — recycle
+            // and surface as transient so SF auto-reconnect (when configured)
+            // can absorb it.
             LOG.error("Error retrying stalled batch [id={}]", batch.getBatchId(), t);
             failConnection(new LineSenderException(
                     "Error retrying stalled batch " + batch.getBatchId() + ": " + t.getMessage(), t), false);
@@ -1165,15 +1178,22 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
                     LOG.debug("Durable ACK received [tables={}]", response.getTableEntryCount());
                 }
             } else {
-                // Error - fail the batch
+                // Server returned a per-batch error (parse, schema mismatch,
+                // write, security, internal). The bytes are the bytes —
+                // reconnecting and re-sending the same payload will produce
+                // the same error. Under SF, the rejected frame sits on disk
+                // and replay-on-reconnect would ship it again, so a
+                // transient classification turns into an unbounded
+                // reconnect loop. Treat as fatal so the user sees the
+                // failure instead of silent thrashing.
                 String errorMessage = response.getErrorMessage();
                 LOG.error("Error response [seq={}, status={}, error={}]", sequence, response.getStatusName(), errorMessage);
 
                 LineSenderException error = new LineSenderException(
                         "Server error for batch " + sequence + ": " +
                                 response.getStatusName() + " - " + errorMessage);
                 totalErrors.incrementAndGet();
-                failConnection(error, false);
+                failConnection(error, true);
             }
         }
 

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

@@ -275,7 +275,7 @@ public void trim(long ackedSeq) {
                     s.fd = -1;
                 }
                 ff.remove(s.pathPtrNative);
-                Files.freeNativePath(s.pathPtrNative);
+                ff.freeNativePath(s.pathPtrNative);
                 s.pathPtrNative = 0;
                 bytesOnDiskCache -= s.writePos;
             } else {
@@ -330,7 +330,7 @@ public void close() {
                 s.fd = -1;
             }
             if (s.pathPtrNative != 0) {
-                Files.freeNativePath(s.pathPtrNative);
+                ff.freeNativePath(s.pathPtrNative);
                 s.pathPtrNative = 0;
             }
         }
@@ -555,7 +555,7 @@ private void rotate() {
         if (old.frameCount == 0) {
             // empty segment shouldn't happen via rotate, but be defensive: drop it
             ff.remove(old.pathPtrNative);
-            Files.freeNativePath(old.pathPtrNative);
+            ff.freeNativePath(old.pathPtrNative);
             old.pathPtrNative = 0;
             bytesOnDiskCache -= old.writePos;
             segments.remove(segments.size() - 1);
@@ -567,9 +567,9 @@ private void rotate() {
             throw new SfException("failed to seal segment by rename " + old.path + " -> " + sealedPath);
         }
         // Path changed — free old native ptr and re-encode for the sealed name.
-        Files.freeNativePath(old.pathPtrNative);
+        ff.freeNativePath(old.pathPtrNative);
         old.path = sealedPath;
-        old.pathPtrNative = Files.allocNativePath(sealedPath);
+        old.pathPtrNative = ff.allocNativePath(sealedPath);
         old.sealed = true;
         old.lastSeqOnDisk = lastSeq;
         createActive(lastSeq + 1);
@@ -581,18 +581,19 @@ private void createActive(long baseSeq) {
         if (fd < 0) {
             throw new SfException("cannot create active segment: " + path);
         }
+        // The fd and pathPtrNative are owned locally until segments.add(s)
+        // below; close()'s cleanup loop only walks the segments list, so
+        // anything that throws between the openCleanRW above and segments.add
+        // must release them here or they leak. Note ff.allocNativePath can
+        // throw CairoException on OOM — keep it inside the try.
         Segment s = new Segment();
         s.baseSeq = baseSeq;
         s.path = path;
-        s.pathPtrNative = Files.allocNativePath(path);
         s.fd = fd;
         s.sealed = false;
         s.frameCount = 0;
-        // The fd and pathPtrNative are owned locally until segments.add(s)
-        // below; close()'s cleanup loop only walks the segments list, so
-        // anything that throws between the openCleanRW above and segments.add
-        // must release them here or they leak.
         try {
+            s.pathPtrNative = ff.allocNativePath(path);
             writeHeader(s);
             s.writePos = HEADER_SIZE;
             if (ff.fsync(fd) != 0) {
@@ -601,8 +602,18 @@ private void createActive(long baseSeq) {
         } catch (Throwable t) {
             ff.close(fd);
             s.fd = -1;
-            Files.freeNativePath(s.pathPtrNative);
-            s.pathPtrNative = 0;
+            if (s.pathPtrNative != 0) {
+                ff.freeNativePath(s.pathPtrNative);
+                s.pathPtrNative = 0;
+            }
+            // Best-effort cleanup of the orphan .sfa file. If this also
+            // throws (e.g. another OOM during path encoding), let it
+            // propagate — the original failure is already on the way out.
+            try {
+                ff.remove(path);
+            } catch (Throwable ignored) {
+                // best-effort
+            }
             throw t;
         }
         segments.add(s);
@@ -698,7 +709,7 @@ private Segment parseFilename(String name) {
                 Segment s = new Segment();
                 s.baseSeq = Long.parseUnsignedLong(body, 16);
                 s.path = dir + "/" + name;
-                s.pathPtrNative = Files.allocNativePath(s.path);
+                s.pathPtrNative = ff.allocNativePath(s.path);
                 s.sealed = false;
                 return s;
             }
@@ -712,7 +723,7 @@ private Segment parseFilename(String name) {
                 s.baseSeq = Long.parseUnsignedLong(body.substring(0, 16), 16);
                 s.lastSeqOnDisk = Long.parseUnsignedLong(body.substring(17), 16);
                 s.path = dir + "/" + name;
-                s.pathPtrNative = Files.allocNativePath(s.path);
+                s.pathPtrNative = ff.allocNativePath(s.path);
                 s.sealed = true;
                 return s;
             }

core/src/main/java/io/questdb/client/std/DefaultFilesFacade.java

@@ -31,6 +31,11 @@
  */
 final class DefaultFilesFacade implements FilesFacade {
 
+    @Override
+    public long allocNativePath(String path) {
+        return Files.allocNativePath(path);
+    }
+
     @Override
     public int close(int fd) {
         return Files.close(fd);
@@ -66,6 +71,11 @@ public int findType(long findPtr) {
         return Files.findType(findPtr);
     }
 
+    @Override
+    public void freeNativePath(long pathPtr) {
+        Files.freeNativePath(pathPtr);
+    }
+
     @Override
     public int fsync(int fd) {
         return Files.fsync(fd);

core/src/main/java/io/questdb/client/std/FilesFacade.java

@@ -35,6 +35,15 @@
 public interface FilesFacade {
     FilesFacade INSTANCE = new DefaultFilesFacade();
 
+    /**
+     * Allocate a native UTF-8 path pointer. Test injection point: a wrapping
+     * facade can throw to simulate OOM without depending on actual memory
+     * pressure. Production callers must release the returned pointer via
+     * {@link #freeNativePath(long)}. Default delegates to
+     * {@link Files#allocNativePath(String)}.
+     */
+    long allocNativePath(String path);
+
     int close(int fd);
 
     boolean exists(String path);
@@ -49,6 +58,12 @@ public interface FilesFacade {
 
     int findType(long findPtr);
 
+    /**
+     * Release a pointer returned by {@link #allocNativePath(String)}.
+     * Default delegates to {@link Files#freeNativePath(long)}.
+     */
+    void freeNativePath(long pathPtr);
+
     int fsync(int fd);
 
     long length(int fd);

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

@@ -24,6 +24,7 @@
 
 package io.questdb.client.test.cutlass.qwp.client.sf;
 
+import io.questdb.client.cairo.CairoException;
 import io.questdb.client.cutlass.qwp.client.sf.SegmentLog;
 import io.questdb.client.cutlass.qwp.client.sf.SfDiskFullException;
 import io.questdb.client.cutlass.qwp.client.sf.SfException;
@@ -564,6 +565,78 @@ public void testOldestSeqAfterTrim() throws Exception {
      * each opened fd was {@code close}d before the {@link SfException}
      * propagated out of {@code SegmentLog.open}.
      */
+    /**
+     * Red test for the fd-leak gap between {@code openCleanRW} and the
+     * {@code try} block in {@code SegmentLog.createActive}.
+     * <p>
+     * Production order at lines 580-595:
+     * <pre>
+     *   int fd = ff.openCleanRW(path, 0);                  // fd opened
+     *   ...
+     *   s.pathPtrNative = ff.allocNativePath(path);        // CAN throw OOM
+     *   s.fd = fd;                                          // never reached on throw
+     *   try { ... } catch { ff.close(fd); ... }             // try not entered
+     * </pre>
+     * If {@code allocNativePath} throws (the {@code Unsafe.malloc} inside
+     * {@link io.questdb.client.std.Files#pathPtr(String)} wraps {@link OutOfMemoryError}
+     * in {@link CairoException}), the local {@code fd} is leaked: {@code s} was
+     * never added to {@code segments}, so {@code close()}'s cleanup loop never
+     * sees it. The orphan {@code .sfa} file also remains on disk and trips the
+     * "multiple active segments" guard on the next process restart that
+     * legitimately rotates.
+     * <p>
+     * On a long-running spacecraft client under intermittent memory pressure,
+     * each failed rotation leaks one fd; sustained loops will exhaust the
+     * process fd table.
+     * <p>
+     * The fix: register {@code s.fd = fd} BEFORE the throwing call, and
+     * extend the {@code try/catch} cleanup to cover the path allocation
+     * (and {@code ff.remove(path)} the orphan file).
+     */
+    @Test
+    public void testCreateActiveDoesNotLeakFdOnAllocNativePathOom() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            FdTrackingFacade tracker = new FdTrackingFacade();
+            tracker.failNextActiveAllocNativePath = true;
+            try {
+                SegmentLog.open(tmpDir, tracker, 4096, 4096, false);
+                fail("expected open to fail because allocNativePath was forced to throw");
+            } catch (Throwable expected) {
+                String msg = expected.getMessage() == null ? "" : expected.getMessage();
+                String causeMsg = expected.getCause() == null || expected.getCause().getMessage() == null
+                        ? "" : expected.getCause().getMessage();
+                assertTrue(
+                        "wrong failure surfaced: " + expected + " / cause=" + expected.getCause(),
+                        msg.contains("simulated") || msg.contains("OOM")
+                                || causeMsg.contains("simulated") || causeMsg.contains("OOM"));
+            }
+            Set<Integer> leaked = new HashSet<>(tracker.opened);
+            leaked.removeAll(tracker.closed);
+            assertEquals(
+                    "createActive must close every fd it opened when allocNativePath throws "
+                            + "between openCleanRW and the try-block; leaked=" + leaked,
+                    0, leaked.size());
+
+            // Also: no orphan .sfa file should remain on disk. The fix should
+            // ff.remove the half-created file so the next open sees a clean dir.
+            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 && name.endsWith(".sfa")) {
+                            fail("orphan .sfa file remains after partial-init failure: " + name);
+                        }
+                        rc = Files.findNext(find);
+                    }
+                } finally {
+                    Files.findClose(find);
+                }
+            }
+        });
+    }
+
     @Test
     public void testCreateActiveDoesNotLeakFdOnFsyncFailure() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
@@ -947,6 +1020,21 @@ private static class FdTrackingFacade implements FilesFacade {
         // Set true to fault the NEXT fsync that targets a fd which was just
         // opened (i.e., not yet closed). Auto-reset after firing once.
         volatile boolean failNextFsyncOnNewFd;
+        // Set true to fault the NEXT allocNativePath whose path ends in
+        // ACTIVE_SUFFIX. Simulates an OOM at the exact moment between
+        // openCleanRW and the try-block in createActive. Auto-reset.
+        volatile boolean failNextActiveAllocNativePath;
+
+        @Override
+        public long allocNativePath(String path) {
+            // ".sfa" is SegmentLog.ACTIVE_SUFFIX (package-private, hardcoded here).
+            if (failNextActiveAllocNativePath && path.endsWith(".sfa")) {
+                failNextActiveAllocNativePath = false;
+                throw CairoException.nonCritical()
+                        .put("simulated OOM in allocNativePath: ").put(path);
+            }
+            return FilesFacade.INSTANCE.allocNativePath(path);
+        }
 
         @Override
         public int close(int fd) {
@@ -987,6 +1075,11 @@ public int findType(long findPtr) {
             return FilesFacade.INSTANCE.findType(findPtr);
         }
 
+        @Override
+        public void freeNativePath(long pathPtr) {
+            FilesFacade.INSTANCE.freeNativePath(pathPtr);
+        }
+
         @Override
         public int fsync(int fd) {
             if (failNextFsyncOnNewFd && opened.contains(fd) && !closed.contains(fd)) {
@@ -1069,6 +1162,11 @@ private static class FaultyLengthFacade implements FilesFacade {
         int lengthFaultsTriggered;
         private int lengthCalls;
 
+        @Override
+        public long allocNativePath(String path) {
+            return FilesFacade.INSTANCE.allocNativePath(path);
+        }
+
         @Override
         public int close(int fd) {
             return FilesFacade.INSTANCE.close(fd);
@@ -1104,6 +1202,11 @@ public int findType(long findPtr) {
             return FilesFacade.INSTANCE.findType(findPtr);
         }
 
+        @Override
+        public void freeNativePath(long pathPtr) {
+            FilesFacade.INSTANCE.freeNativePath(pathPtr);
+        }
+
         @Override
         public int fsync(int fd) {
             return FilesFacade.INSTANCE.fsync(fd);
@@ -1181,6 +1284,11 @@ private static class ShortPayloadWriteFacade implements FilesFacade {
         // are larger. Use length to disambiguate without inspecting content.
         volatile boolean failNextPayloadWrite;
 
+        @Override
+        public long allocNativePath(String path) {
+            return FilesFacade.INSTANCE.allocNativePath(path);
+        }
+
         @Override
         public int close(int fd) {
             return FilesFacade.INSTANCE.close(fd);
@@ -1216,6 +1324,11 @@ public int findType(long findPtr) {
             return FilesFacade.INSTANCE.findType(findPtr);
         }
 
+        @Override
+        public void freeNativePath(long pathPtr) {
+            FilesFacade.INSTANCE.freeNativePath(pathPtr);
+        }
+
         @Override
         public int fsync(int fd) {
             return FilesFacade.INSTANCE.fsync(fd);