feat(ilp): opt-in sf_fsync_on_flush, fix sf_fsync default doc lie

Sender.storeAndForwardFsync's Javadoc claimed the default sf_fsync=off
"runs fsync on rotation and on explicit flush()". In practice flush()
never called segmentLog.fsync() — the only production fsync paths were
per-append (gated on fsync_each_append, the sf_fsync=on path),
rotation (rare), and new-segment header creation (rare). With default
config a sender that flushes coarsely between rotations was leaving
all bytes in the OS page cache; an OS crash would lose them despite
the docs implying durability.

Two-part fix:

1. Doc honesty (Sender.java):
   storeAndForwardFsync rewritten to spell out exactly what
   sf_fsync=off and sf_fsync=on mean. The default leaves bytes
   between rotations in the page cache — process crashes survive,
   OS crashes don't.

2. Opt-in fsync-on-flush:
   New knob storeAndForwardFsyncOnFlush(boolean) on the builder,
   parsed as sf_fsync_on_flush=on/off in the connect string. When
   enabled, every flush() (and the implicit flush in close()) routes
   a fsync request to the I/O thread before returning. Off by
   default — small-batch + frequent-flush senders pay one disk fsync
   per call, which is unacceptable for high-rate workloads.

   The fsync runs on the I/O thread because SegmentLog is single-
   threaded (the I/O thread owns every read/write/trim/rotate).
   Calling segmentLog.fsync() from the user thread would race
   against an in-flight trim() (which may force-rotate the active
   under per-frame trim) or append() from a concurrent send. The
   signal pattern is the same one used by ping/pong: user sets
   fsyncRequested + waits on fsyncComplete; I/O thread observes the
   flag at the top of its iteration, performs the fsync, publishes
   outcome via fsyncError + fsyncComplete. Concurrent callers are
   serialised by fsyncLock so each gets its own round-trip.

Tests:
- testFlushDoesNotFsyncByDefault (SfIntegrationTest): with
  fsyncOnFlush=false the FsyncCountingFacade observes ZERO fsyncs
  during flush() — proves we did not regress the small-batch
  hot path.
- testFlushFsyncsWhenOptedIn (SfIntegrationTest): with
  fsyncOnFlush=true the same counter observes >= 1 fsync per
  flush() — proves the wiring is end-to-end.
- testSfFsyncOnFlushParses (SfFromConfigTest): connect-string
  round-trip.
- testInvalidSfFsyncOnFlushValueRejected (SfFromConfigTest): bad
  value rejected with a useful message.
- testSfFsyncOnFlushOnTcpRejected (SfFromConfigTest): TCP transport
  rejects the WebSocket-only knob.

Full suite: 1981 tests pass, zero regressions.

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

Author: bluestreak01

core/src/main/java/io/questdb/client/Sender.java

@@ -631,6 +631,7 @@ public int getTimeout() {
         private long sfMaxBytes = PARAMETER_NOT_SET_EXPLICITLY;
         private long sfMaxTotalBytes = PARAMETER_NOT_SET_EXPLICITLY;
         private boolean sfFsync;
+        private boolean sfFsyncOnFlush;
         private boolean tlsEnabled;
         private TlsValidationMode tlsValidationMode;
         private char[] trustStorePassword;
@@ -968,7 +969,8 @@ public Sender build() {
                             wsAuthHeader,
                             actualMaxSchemasPerConnection,
                             requestDurableAck,
-                            segmentLog
+                            segmentLog,
+                            sfFsyncOnFlush
                     );
                 } catch (Throwable t) {
                     // If connect failed, the sender's close() ran and would have closed
@@ -1640,9 +1642,18 @@ public LineSenderBuilder storeAndForwardMaxTotalBytes(long maxTotalBytes) {
         /**
          * When enabled, every successful SF append calls {@code fsync} on the
          * active segment file before returning. Trades throughput for the
-         * strongest durability guarantee — captured frames survive even an OS
-         * crash, not just a process crash. Default: off (fsync runs on rotation
-         * and on explicit flush()).
+         * strongest durability guarantee — every captured frame survives an OS
+         * crash, not just a process crash.
+         * <p>
+         * Default: off. With {@code sf_fsync=off}, fsync only fires on
+         * segment rotation and new-segment header creation; bytes appended to
+         * the active segment between rotations live only in the OS page cache
+         * and may be lost in an OS crash, kernel panic, or power loss. The
+         * JVM going down is survived (the page cache outlives the process).
+         * <p>
+         * If you flush coarsely (one fsync per flush is acceptable) and want
+         * OS-crash survival without paying per-append fsync cost, set
+         * {@link #storeAndForwardFsyncOnFlush(boolean)} instead.
          */
         public LineSenderBuilder storeAndForwardFsync(boolean enabled) {
             if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
@@ -1652,6 +1663,32 @@ public LineSenderBuilder storeAndForwardFsync(boolean enabled) {
             return this;
         }
 
+        /**
+         * When enabled, every successful {@code Sender.flush()} (and the
+         * implicit flush during {@code close()}) calls {@code fsync} on the
+         * SF active segment file before returning. Trades flush latency
+         * (one fsync per flush) for OS-crash survival of every byte that
+         * the user explicitly flushed.
+         * <p>
+         * Off by default. Use this when batches are large or flushes are
+         * coarse and you want OS-crash durability without paying the
+         * per-append fsync cost of {@link #storeAndForwardFsync(boolean)}.
+         * Avoid it when batches are small and flushes are frequent — every
+         * flush blocks on a disk fsync, which is typically the slowest
+         * operation in the SF write path.
+         * <p>
+         * Combining {@code sf_fsync=on} and {@code sf_fsync_on_flush=on}
+         * is allowed but redundant: per-append fsync already covers every
+         * byte before flush returns.
+         */
+        public LineSenderBuilder storeAndForwardFsyncOnFlush(boolean enabled) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("store_and_forward is only supported for WebSocket transport");
+            }
+            this.sfFsyncOnFlush = enabled;
+            return this;
+        }
+
         /**
          * Configures the maximum time the Sender will spend retrying upon receiving a recoverable error from the server.
          * <br>
@@ -2118,6 +2155,18 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     } else {
                         throw new LineSenderException("invalid sf_fsync [value=").put(sink).put(", allowed-values=[on, off]]");
                     }
+                } else if (Chars.equals("sf_fsync_on_flush", sink)) {
+                    if (protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException("sf_fsync_on_flush is only supported for WebSocket transport");
+                    }
+                    pos = getValue(configurationString, pos, sink, "sf_fsync_on_flush");
+                    if (Chars.equalsIgnoreCase("on", sink)) {
+                        storeAndForwardFsyncOnFlush(true);
+                    } else if (Chars.equalsIgnoreCase("off", sink)) {
+                        storeAndForwardFsyncOnFlush(false);
+                    } else {
+                        throw new LineSenderException("invalid sf_fsync_on_flush [value=").put(sink).put(", allowed-values=[on, off]]");
+                    }
                 } else if (Chars.equals("max_datagram_size", sink)) {
                     pos = getValue(configurationString, pos, sink, "max_datagram_size");
                     int mds = parseIntValue(sink, "max_datagram_size");

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

@@ -172,6 +172,11 @@ public class QwpWebSocketSender implements Sender {
     // True when this sender took ownership of segmentLog (e.g. via the
     // connect-string builder); close() will then close the log too.
     private boolean ownsSegmentLog;
+    // When true, every successful flush() (including the implicit flush
+    // during close()) routes a fsync request to the I/O thread before
+    // returning. Off by default — opt-in via setSegmentLogFsyncOnFlush
+    // or sf_fsync_on_flush=on in the connect string.
+    private boolean fsyncOnFlush;
     // Set by the I/O thread after a successful SF reconnect; checked by the user
     // thread on the next flushPendingRows so the next batch re-publishes schemas
     // the new server doesn't yet know about.
@@ -350,6 +355,33 @@ public static QwpWebSocketSender connect(
             int maxSchemasPerConnection,
             boolean requestDurableAck,
             SegmentLog segmentLog
+    ) {
+        return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes,
+                autoFlushIntervalNanos, inFlightWindowSize, authorizationHeader,
+                maxSchemasPerConnection, requestDurableAck, segmentLog, false);
+    }
+
+    /**
+     * Connect overload with store-and-forward and an explicit
+     * fsync-on-flush opt-in. {@code fsyncOnFlush=true} routes a fsync of
+     * the SF active segment to the I/O thread at every {@link #flush()}
+     * (and at the implicit flush during {@link #close()}); off by default
+     * because small-batch / frequent-flush senders pay one disk fsync per
+     * call.
+     */
+    public static QwpWebSocketSender connect(
+            String host,
+            int port,
+            ClientTlsConfiguration tlsConfig,
+            int autoFlushRows,
+            int autoFlushBytes,
+            long autoFlushIntervalNanos,
+            int inFlightWindowSize,
+            String authorizationHeader,
+            int maxSchemasPerConnection,
+            boolean requestDurableAck,
+            SegmentLog segmentLog,
+            boolean fsyncOnFlush
     ) {
         QwpWebSocketSender sender = new QwpWebSocketSender(
                 host, port, tlsConfig,
@@ -361,6 +393,7 @@ public static QwpWebSocketSender connect(
             if (segmentLog != null) {
                 sender.setSegmentLog(segmentLog, true);
             }
+            sender.setSegmentLogFsyncOnFlush(fsyncOnFlush);
             sender.ensureConnected();
         } catch (Throwable t) {
             sender.close();
@@ -579,6 +612,12 @@ public void close() {
                     // SF dir will replay them.
                     if (sendQueue != null) {
                         sendQueue.flush();
+                        if (fsyncOnFlush && segmentLog != null) {
+                            // Same opt-in fsync as the public flush(): the
+                            // user asked for "data is durable on flush"
+                            // semantics, and close() implies a final flush.
+                            sendQueue.requestSegmentLogFsync();
+                        }
                         if (segmentLog == null) {
                             sendQueue.awaitPendingAcks();
                         }
@@ -866,6 +905,20 @@ public void flush() {
                 throw e;
             }
 
+            // Opt-in fsync of the SF active segment. Routed through the
+            // I/O thread (it owns SegmentLog) via the requestSegmentLogFsync
+            // signal; blocks until the syscall returns. Off by default
+            // because small-batch / frequent-flush senders pay one disk
+            // fsync per call.
+            if (fsyncOnFlush && segmentLog != null) {
+                try {
+                    sendQueue.requestSegmentLogFsync();
+                } catch (LineSenderException e) {
+                    checkConnectionError();
+                    throw e;
+                }
+            }
+
             // Under SF the durability guarantee is "data is on disk", not "data is
             // server-acked". sendQueue.flush() already waited for processingCount
             // to drain, which means SF.append has run for every queued batch. Skip
@@ -1234,6 +1287,28 @@ public void setSegmentLog(SegmentLog log, boolean takeOwnership) {
         this.ownsSegmentLog = takeOwnership && log != null;
     }
 
+    /**
+     * Opt in to fsyncing the SF active segment at every {@link #flush()}
+     * (and at the implicit flush during {@link #close()}). Off by default.
+     * <p>
+     * Useful for senders that flush coarsely and want OS-crash survival
+     * without paying the per-append fsync cost of {@code sf_fsync=on}.
+     * Avoid for high-rate small-batch + frequent-flush workloads — every
+     * flush blocks on a disk fsync.
+     * <p>
+     * Must be set before the first send (mirrors {@link #setSegmentLog}).
+     */
+    public void setSegmentLogFsyncOnFlush(boolean enabled) {
+        if (closed) {
+            throw new LineSenderException("Sender is closed");
+        }
+        if (connected) {
+            throw new LineSenderException(
+                    "setSegmentLogFsyncOnFlush must be called before the first send");
+        }
+        this.fsyncOnFlush = enabled;
+    }
+
     /**
      * Number of times an outgoing batch was stalled because the SF total disk cap
      * was reached. Each stall blocks the user thread's flush() until ACKs trim

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

@@ -159,6 +159,17 @@ public class WebSocketSendQueue implements QuietCloseable {
     private volatile boolean pingRequested;
     private volatile boolean pongReceived;
     private long pingDeadlineNanos;
+    // Signal-based fsync: user thread sets fsyncRequested + waits on
+    // fsyncComplete; I/O thread (the only thread that owns SegmentLog)
+    // observes the flag, calls segmentLog.fsync(), publishes outcome
+    // via fsyncError + fsyncComplete. Mirrors the ping handshake.
+    private volatile boolean fsyncRequested;
+    private volatile boolean fsyncComplete;
+    private volatile Throwable fsyncError;
+    // Serialises concurrent requestSegmentLogFsync callers, same idiom as
+    // pingLock — each caller gets its own round-trip so post-conditions
+    // hold per caller.
+    private final Object fsyncLock = new Object();
     // Running state
     private volatile boolean running;
     private volatile boolean shuttingDown;
@@ -490,6 +501,55 @@ public void ping() {
         }
     }
 
+    /**
+     * Asks the I/O thread to fsync the SF active segment and blocks until
+     * the syscall returns. No-op when no SegmentLog is configured.
+     * <p>
+     * SegmentLog is single-threaded — the I/O thread owns every read,
+     * write, trim and rotate. Calling {@code segmentLog.fsync()} from the
+     * user thread would race against an in-flight {@code trim()} (which
+     * may force-rotate the active segment under per-frame trim) or
+     * against {@code append()} from a concurrent send. The signal pattern
+     * keeps SegmentLog ownership clean.
+     * <p>
+     * Concurrent callers are serialised via {@link #fsyncLock} so each
+     * one gets its own round-trip — the post-condition "every byte
+     * persisted before the call returned is durable on disk" holds per
+     * caller independently.
+     */
+    public void requestSegmentLogFsync() {
+        if (segmentLog == null) {
+            return;
+        }
+        synchronized (fsyncLock) {
+            checkError();
+            synchronized (processingLock) {
+                fsyncComplete = false;
+                fsyncError = null;
+                fsyncRequested = true;
+                processingLock.notifyAll();
+                while (!fsyncComplete && running) {
+                    try {
+                        processingLock.wait(1000);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        throw new LineSenderException("SF fsync interrupted");
+                    }
+                }
+                if (!fsyncComplete) {
+                    checkError();
+                    throw new LineSenderException(
+                            "SF fsync aborted: send queue is shutting down");
+                }
+            }
+            Throwable err = fsyncError;
+            if (err != null) {
+                throw new LineSenderException("SF fsync failed: " + err.getMessage(), err);
+            }
+            checkError();
+        }
+    }
+
     /**
      * Returns the total number of batches sent.
      */
@@ -641,6 +701,30 @@ private void ioLoop() {
                     }
                 }
 
+                // Honour any pending SF fsync request. Runs before batch
+                // processing so the user's flush() observes a stable
+                // "every byte appended before the request is durable"
+                // invariant. A failure here is published to the caller
+                // via fsyncError; we do NOT failConnection because an
+                // fsync EIO is a storage problem, not a wire problem,
+                // and the user can decide whether to retry vs. close.
+                if (fsyncRequested) {
+                    fsyncRequested = false;
+                    Throwable err = null;
+                    try {
+                        if (segmentLog != null) {
+                            segmentLog.fsync();
+                        }
+                    } catch (Throwable t) {
+                        err = t;
+                    }
+                    fsyncError = err;
+                    synchronized (processingLock) {
+                        fsyncComplete = true;
+                        processingLock.notifyAll();
+                    }
+                }
+
                 MicrobatchBuffer batch = null;
                 boolean hasInFlight = (inFlightWindow != null && inFlightWindow.getInFlightCount() > 0);
                 IoState state = computeState(hasInFlight);
@@ -652,7 +736,7 @@ private void ioLoop() {
                         // Nothing to do - wait for work under lock
                         synchronized (processingLock) {
                             // Re-check under lock to avoid missed wakeup
-                            if (isPendingEmpty() && running && !pingRequested) {
+                            if (isPendingEmpty() && running && !pingRequested && !fsyncRequested) {
                                 try {
                                     processingLock.wait(100);
                                 } catch (InterruptedException e) {

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

@@ -499,6 +499,53 @@ public void testInvalidSfFsyncValueRejected() {
         }
     }
 
+    /**
+     * sf_fsync_on_flush is opt-in. Verify the connect-string parses both
+     * values and the wiring reaches the sender (basic round-trip — the
+     * actual fsync-on-flush behaviour is exercised in WebSocketSendQueueTest
+     * with a counting FilesFacade).
+     */
+    @Test
+    public void testSfFsyncOnFlushParses() throws Exception {
+        int port = TEST_PORT + 6;
+        AckHandler handler = new AckHandler();
+        try (TestWebSocketServer server = new TestWebSocketServer(port, handler)) {
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+            String config = "ws::addr=localhost:" + port
+                    + ";store_and_forward=on;sf_dir=" + sfDir
+                    + ";sf_fsync_on_flush=on;";
+            try (Sender sender = Sender.fromConfig(config)) {
+                sender.table("foo").longColumn("v", 1L).atNow();
+                sender.flush();
+            }
+            Assert.assertTrue(Files.exists(sfDir));
+        }
+    }
+
+    @Test
+    public void testInvalidSfFsyncOnFlushValueRejected() {
+        String config = "ws::addr=localhost:1;store_and_forward=on;sf_dir=" + sfDir
+                + ";sf_fsync_on_flush=maybe;";
+        try (Sender ignored = Sender.fromConfig(config)) {
+            Assert.fail("expected rejection");
+        } catch (LineSenderException expected) {
+            Assert.assertTrue(expected.getMessage(),
+                    expected.getMessage().contains("invalid sf_fsync_on_flush"));
+        }
+    }
+
+    @Test
+    public void testSfFsyncOnFlushOnTcpRejected() {
+        String config = "tcp::addr=localhost:1;sf_fsync_on_flush=on;";
+        try (Sender ignored = Sender.fromConfig(config)) {
+            Assert.fail("expected rejection");
+        } catch (LineSenderException expected) {
+            Assert.assertTrue(expected.getMessage(),
+                    expected.getMessage().contains("WebSocket"));
+        }
+    }
+
     @Test
     public void testStoreAndForwardWithSyncWindowRejected() {
         String config = "ws::addr=localhost:1;store_and_forward=on;sf_dir=" + sfDir