refactor(qwp): rename sf_max_bytes to sf_max_segment_bytes (#70)

Renames the store-and-forward connect-string key sf_max_bytes to sf_max_segment_bytes: the key caps a single SF segment, and the new name says so; it also pairs cleanly with sf_max_total_bytes. The builder method follows: storeAndForwardMaxBytes() is now storeAndForwardMaxSegmentBytes(). Internal identifiers, error messages, and test assertions rename in step, so git grep -iE 'sf[_-]?max[_-]?bytes|storeandforwardmaxbytes' returns zero.

Author: mtopolnik

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

@@ -1128,7 +1128,7 @@ public int getConnectTimeout() {
         // Durability contract for SF append/flush. FLUSH and APPEND remain
         // deferred follow-ups; PERIODIC uses the segment manager.
         private SfDurability sfDurability = SfDurability.MEMORY;
-        private long sfMaxBytes = PARAMETER_NOT_SET_EXPLICITLY;
+        private long sfMaxSegmentBytes = PARAMETER_NOT_SET_EXPLICITLY;
         private long sfMaxTotalBytes = PARAMETER_NOT_SET_EXPLICITLY;
         private long sfSyncIntervalMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private boolean shouldDestroyPrivKey;
@@ -1452,17 +1452,17 @@ public Sender build() {
                 // (same lock-free architecture, no disk involvement).
                 // Durability-combination validation lives in validateParameters
                 // so build() and no-connect validation apply the same rules.
-                long actualSfMaxBytes = sfMaxBytes == PARAMETER_NOT_SET_EXPLICITLY
+                long actualSfMaxSegmentBytes = sfMaxSegmentBytes == PARAMETER_NOT_SET_EXPLICITLY
                         ? DEFAULT_SEGMENT_BYTES
-                        : sfMaxBytes;
+                        : sfMaxSegmentBytes;
                 // Default cap depends on backing: RAM (memory mode) is tight
                 // by default; disk (SF mode) is cheap so the default is
                 // generous enough that normal traffic never hits it.
                 long defaultMaxTotal = sfDir == null
                         ? DEFAULT_MAX_BYTES_MEMORY
                         : DEFAULT_MAX_BYTES_SF;
                 long actualSfMaxTotalBytes = sfMaxTotalBytes == PARAMETER_NOT_SET_EXPLICITLY
-                        ? Math.max(defaultMaxTotal, actualSfMaxBytes * 2)
+                        ? Math.max(defaultMaxTotal, actualSfMaxSegmentBytes * 2)
                         : sfMaxTotalBytes;
                 long actualCloseFlushTimeoutMillis = closeFlushTimeoutMillis == CLOSE_FLUSH_TIMEOUT_NOT_SET
                         ? DEFAULT_CLOSE_FLUSH_TIMEOUT_MILLIS
@@ -1556,7 +1556,7 @@ public Sender build() {
                         ? DEFAULT_SF_SYNC_INTERVAL_MILLIS : sfSyncIntervalMillis) * 1_000_000L
                         : 0L;
                 CursorSendEngine cursorEngine = new CursorSendEngine(
-                        slotPath, actualSfMaxBytes,
+                        slotPath, actualSfMaxSegmentBytes,
                         actualSfMaxTotalBytes, actualSfAppendDeadlineNanos,
                         actualSfSyncIntervalNanos);
                 int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
@@ -1638,7 +1638,7 @@ public Sender build() {
                             connected.startOrphanDrainers(
                                     orphans,
                                     maxBackgroundDrainers,
-                                    actualSfMaxBytes,
+                                    actualSfMaxSegmentBytes,
                                     actualSfMaxTotalBytes,
                                     actualSfSyncIntervalNanos);
                         }
@@ -2770,19 +2770,21 @@ public LineSenderBuilder storeAndForwardDurability(SfDurability durability) {
         }
 
         /**
-         * Maximum bytes per segment file before rotation. Defaults to
-         * {@code DEFAULT_SEGMENT_BYTES}
-         * (4 MiB). Smaller segments mean faster trim of acked data; larger
-         * segments mean fewer rotations.
+         * Maximum bytes per segment file before rotation, the builder form of
+         * the {@code sf_max_segment_bytes} connect-string key. Smaller segments
+         * mean faster trim of acked data; larger segments mean fewer rotations.
+         * Default: {@code 4 MiB}. WebSocket transport only.
+         *
+         * @param maxSegmentBytes per-segment cap in bytes; must be positive
          */
-        public LineSenderBuilder storeAndForwardMaxBytes(long maxBytes) {
+        public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxSegmentBytes) {
             if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
                 throw new LineSenderException("store_and_forward is only supported for WebSocket transport");
             }
-            if (maxBytes <= 0) {
-                throw new LineSenderException("sf_max_bytes must be positive: ").put(maxBytes);
+            if (maxSegmentBytes <= 0) {
+                throw new LineSenderException("sf_max_segment_bytes must be positive: ").put(maxSegmentBytes);
             }
-            this.sfMaxBytes = maxBytes;
+            this.sfMaxSegmentBytes = maxSegmentBytes;
             return this;
         }
 
@@ -3387,12 +3389,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     }
                     pos = getValue(configurationString, pos, sink, "sender_id");
                     senderId(sink.toString());
-                } else if (Chars.equals("sf_max_bytes", sink)) {
+                } else if (Chars.equals("sf_max_segment_bytes", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {
-                        throw new LineSenderException("sf_max_bytes is only supported for WebSocket transport");
+                        throw new LineSenderException("sf_max_segment_bytes is only supported for WebSocket transport");
                     }
-                    pos = getValue(configurationString, pos, sink, "sf_max_bytes");
-                    storeAndForwardMaxBytes(parseSizeValue(sink, "sf_max_bytes"));
+                    pos = getValue(configurationString, pos, sink, "sf_max_segment_bytes");
+                    storeAndForwardMaxSegmentBytes(parseSizeValue(sink, "sf_max_segment_bytes"));
                 } else if (Chars.equals("sf_max_total_bytes", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {
                         throw new LineSenderException("sf_max_total_bytes is only supported for WebSocket transport");
@@ -3730,8 +3732,8 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString)
                 if (view.has("sf_append_deadline_millis")) {
                     sfAppendDeadlineMillis(wsLong(view, v, "sf_append_deadline_millis"));
                 }
-                if (view.has("sf_max_bytes")) {
-                    storeAndForwardMaxBytes(wsSize(view, v, "sf_max_bytes"));
+                if (view.has("sf_max_segment_bytes")) {
+                    storeAndForwardMaxSegmentBytes(wsSize(view, v, "sf_max_segment_bytes"));
                 }
                 if (view.has("sf_max_total_bytes")) {
                     storeAndForwardMaxTotalBytes(wsSize(view, v, "sf_max_total_bytes"));
@@ -3892,7 +3894,7 @@ public java.util.Map<String, Object> wsConfigSnapshotForTest() {
             m.put("request_durable_ack", requestDurableAck);
             m.put("sender_id", senderId);
             m.put("sf_dir", sfDir);
-            m.put("sf_max_bytes", sfMaxBytes);
+            m.put("sf_max_segment_bytes", sfMaxSegmentBytes);
             m.put("sf_max_total_bytes", sfMaxTotalBytes);
             m.put("sf_durability", sfDurability == null ? null : sfDurability.name());
             m.put("sf_append_deadline_millis", sfAppendDeadlineMillis);

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

@@ -183,7 +183,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // drained from the head every time a STATUS_DURABLE_ACK frame advances
     // any watermark; an entry pops when every (name, seqTxn) it carries is
     // covered by durableTableWatermarks. Bounded in practice by the SF on-disk
-    // cap: once the producer hits sf_max_bytes it blocks, which caps how far
+    // cap: once the producer hits sf_max_segment_bytes it blocks, which caps how far
     // the durable watermark can lag behind the OK watermark.
     private final ArrayDeque<PendingDurableEntry> pendingDurable = new ArrayDeque<>();
     private final ArrayDeque<PendingDurableEntry> pendingDurablePool = new ArrayDeque<>();

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

@@ -219,7 +219,7 @@ static MmapSegment create(FilesFacade ff, long pathPtr, String displayPath, long
         // the JVM).
         if (!ff.allocate(fd, sizeBytes)) {
             ff.close(fd);
-            // Unlink the partially-created file so a sf_max_bytes-sized
+            // Unlink the partially-created file so a sf_max_segment_bytes-sized
             // empty file does not survive the failure. Under sustained
             // disk-full pressure with the manager polling, hundreds would
             // otherwise accumulate.

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

@@ -86,7 +86,7 @@ public final class ConfigSchema {
         str("sf_append_deadline_millis", Side.INGRESS);
         str("sf_dir", Side.INGRESS);
         str("sf_durability", Side.INGRESS);
-        str("sf_max_bytes", Side.INGRESS);
+        str("sf_max_segment_bytes", Side.INGRESS);
         str("sf_max_total_bytes", Side.INGRESS);
         str("sf_sync_interval_millis", Side.INGRESS);
         str("transaction", Side.INGRESS);

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

@@ -38,7 +38,7 @@
 /**
  * Tests for WebSocket transport support in the Sender.builder() API.
  * These tests verify the builder configuration and validation,
- * not actual WebSocket connectivity (which requires a running server).
+ * not actual WebSocket connectivity (that requires a running server).
  */
 public class LineSenderBuilderWebSocketTest extends AbstractTest {
 
@@ -634,6 +634,39 @@ public void testRetryTimeout_notSupportedForWebSocket() {
                 "not supported for WebSocket");
     }
 
+    @Test
+    public void testStoreAndForwardMaxSegmentBytes() {
+        Sender.LineSenderBuilder builder = Sender.builder(Sender.Transport.WEBSOCKET);
+
+        Assert.assertSame(builder, builder.storeAndForwardMaxSegmentBytes(64 * 1024L));
+        Assert.assertEquals(
+                64 * 1024L,
+                ((Number) builder.wsConfigSnapshotForTest().get("sf_max_segment_bytes")).longValue()
+        );
+    }
+
+    @Test
+    public void testStoreAndForwardMaxSegmentBytesRejectsNonPositiveValues() {
+        long[] rejected = {0L, -1L};
+        for (long value : rejected) {
+            assertThrows("sf_max_segment_bytes must be positive: " + value,
+                    () -> Sender.builder(Sender.Transport.WEBSOCKET).storeAndForwardMaxSegmentBytes(value));
+        }
+    }
+
+    @Test
+    public void testStoreAndForwardMaxSegmentBytesRejectedForNonWebSocketTransports() {
+        Sender.Transport[] rejected = {
+                Sender.Transport.HTTP,
+                Sender.Transport.TCP,
+                Sender.Transport.UDP
+        };
+        for (Sender.Transport transport : rejected) {
+            assertThrows("store_and_forward is only supported for WebSocket transport",
+                    () -> Sender.builder(transport).storeAndForwardMaxSegmentBytes(64 * 1024L));
+        }
+    }
+
     @Test
     public void testSyncModeAutoFlushDefaults() throws Exception {
         // Regression test: connect() must not hardcode autoFlush to 0.
@@ -809,6 +842,15 @@ public void testWsConfigString_withPath_fails() {
                 "unknown configuration key: path");
     }
 
+    @Test
+    public void testWsConfigString_withSfMaxBytes_fails() {
+        // sf_max_bytes is the spelling this key carries in the 1.2.1-1.3.5 jars.
+        // sf_max_segment_bytes replaces it with no alias, so the old spelling
+        // rejects as an unknown key rather than silently configuring nothing.
+        assertBadConfig("ws::addr=localhost:9000;sf_max_bytes=4096;",
+                "unknown configuration key: sf_max_bytes");
+    }
+
     @Test
     public void testWsConfigString_withEgressOnlyKeysSilentlyAccepted() {
         // connect-string.md "Query client keys" and "Multi-host failover": these

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

@@ -720,7 +720,7 @@ public void testIngressOnlyKeysSilentlyAcceptedOnEgress() {
                 "sf_append_deadline_millis=30000",
                 "sf_dir=/var/lib/qdb-sf",
                 "sf_durability=memory",
-                "sf_max_bytes=4m",
+                "sf_max_segment_bytes=4m",
                 "sf_max_total_bytes=10g",
                 "transaction=on",
         };

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

@@ -86,7 +86,7 @@ public void testRestartReplaysSealedSegmentsAgainstFreshServer() throws Exceptio
             String pad = repeat("x", 64);
             String cfg1 = "ws::addr=localhost:" + port1
                     + ";sf_dir=" + sfDir
-                    + ";sf_max_bytes=4096"
+                    + ";sf_max_segment_bytes=4096"
                     + ";close_flush_timeout_millis=0;";
             try (Sender s1 = Sender.fromConfig(cfg1)) {
                 for (int i = 0; i < 50; i++) {

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

@@ -93,7 +93,7 @@ public void testSfDirOnTcpRejected() throws Exception {
     }
 
     @Test
-    public void testSfMaxBytesParsing() throws Exception {
+    public void testSfMaxSegmentBytesParsing() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
             AckHandler handler = new AckHandler();
             try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
@@ -102,7 +102,7 @@ public void testSfMaxBytesParsing() throws Exception {
 
                 int port = server.getPort();
                 String config = "ws::addr=localhost:" + port
-                        + ";sf_dir=" + sfDir + ";sf_max_bytes=131072;";
+                        + ";sf_dir=" + sfDir + ";sf_max_segment_bytes=131072;";
                 try (Sender sender = Sender.fromConfig(config)) {
                     // Write enough data that segments rotate at ~128 KiB boundary.
                     for (int i = 0; i < 50; i++) {
@@ -140,7 +140,7 @@ public void testNoSfDirMeansNoSf() throws Exception {
     }
 
     /**
-     * Regression test for the connect-string {@code sf_max_bytes} /
+     * Regression test for the connect-string {@code sf_max_segment_bytes} /
      * {@code sf_max_total_bytes} parser accepting values larger than
      * {@code Integer.MAX_VALUE}. The pre-cursor parser used parseInt which
      * artificially capped the SF size from the connect string at ~2 GiB.
@@ -349,7 +349,7 @@ public void testSfSyncIntervalRejectsOnNonWebSocketTransport() throws Exception
     }
 
     @Test
-    public void testSfMaxBytesAcceptsSizeSuffixes() throws Exception {
+    public void testSfMaxSegmentBytesAcceptsSizeSuffixes() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
             AckHandler handler = new AckHandler();
             try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
@@ -360,7 +360,7 @@ public void testSfMaxBytesAcceptsSizeSuffixes() throws Exception {
                 // 64m / 4g should parse identically to their byte-count equivalents.
                 String config = "ws::addr=localhost:" + port
                         + ";sf_dir=" + sfDir
-                        + ";sf_max_bytes=64m"
+                        + ";sf_max_segment_bytes=64m"
                         + ";sf_max_total_bytes=4g;";
                 try (Sender sender = Sender.fromConfig(config)) {
                     sender.table("foo").longColumn("v", 1L).atNow();
@@ -473,14 +473,14 @@ public void testSenderIdInvalidCharRejected() throws Exception {
     }
 
     @Test
-    public void testSfMaxBytesInvalidSizeSuffixRejected() throws Exception {
+    public void testSfMaxSegmentBytesInvalidSizeSuffixRejected() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
-            String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";sf_max_bytes=64x;";
+            String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";sf_max_segment_bytes=64x;";
             try (Sender ignored = Sender.fromConfig(config)) {
                 Assert.fail("expected rejection of unknown unit suffix");
             } catch (LineSenderException expected) {
                 Assert.assertTrue(expected.getMessage(),
-                        expected.getMessage().contains("invalid sf_max_bytes"));
+                        expected.getMessage().contains("invalid sf_max_segment_bytes"));
             }
         });
     }

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

@@ -1004,7 +1004,7 @@ public void testNextSealedAfterSurvivesConcurrentRotationAndHeadTrim() throws Ex
 
     /**
      * Open-time sort regression: at the documented {@code sf_max_total_bytes
-     * / sf_max_bytes} ceiling (~16K segments), an O(N²) sort over the
+     * / sf_max_segment_bytes} ceiling (~16K segments), an O(N²) sort over the
      * recovered segments delays the I/O thread. This test guards the sort
      * with a deterministic comparison count.
      * <p>

core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java

@@ -61,7 +61,7 @@ public void testEveryIngressKeyIsHonored() {
         assertHonored("request_durable_ack=on", "request_durable_ack", true);
         assertHonored("sender_id=probe-1", "sender_id", "probe-1");
         assertHonored("sf_dir=/var/probe", "sf_dir", "/var/probe");
-        assertHonored("sf_max_bytes=4096", "sf_max_bytes", 4096L);
+        assertHonored("sf_max_segment_bytes=4096", "sf_max_segment_bytes", 4096L);
         assertHonored("sf_max_total_bytes=8192", "sf_max_total_bytes", 8192L);
         assertHonored("sf_durability=periodic", "sf_durability", "PERIODIC");
         assertHonored("sf_append_deadline_millis=1500", "sf_append_deadline_millis", 1500L);