feat(qwp)!: stop flushing pooled senders on close, add a commit window and a lease-scoped reset

README.md

@@ -32,7 +32,7 @@ try (QuestDB db = QuestDB.connect("ws::addr=localhost:9000;")) {
                 .doubleColumn("price", 2615.54)
                 .doubleColumn("amount", 0.00044)
                 .atNow();
-        // close() flushes pending rows and returns the Sender to the pool.
+        // close() returns the Sender to the pool; rows ship once an auto-flush threshold trips (age, rows or bytes), or when the idle slot is settled.
     }
 }
 ```
@@ -98,8 +98,15 @@ snippets below borrow from its pools.
 
 ### Ingest Data
 
-Borrow a `Sender`, write rows, and close it. `close()` flushes pending rows and returns the sender to the pool — the
-underlying connection is only torn down when the `QuestDB` handle itself is closed.
+Borrow a `Sender`, write rows, and close it. Close releases, flush ships, drain waits: `close()` returns the sender to
+the pool and ships the slot's pending rows once the oldest is older than `auto_flush_interval` (100 ms by default) or
+another auto-flush threshold trips, otherwise it only releases; rows left in the slot go out when an auto-flush
+threshold trips on a later append or `close()`, or when the pool's housekeeper settles the idle slot, typically within
+one `auto_flush_interval`. A lease that holds the slot without appending or closing, for example one that waits for input
+inside the `try` block, holds those rows back until it does: borrow once the data is in hand. Borrowing per
+message is the intended pattern and batches like a sender held for the whole run. Call `flush()` to ship now and
+`drain()` to wait for the server. The underlying connection is only torn down when the `QuestDB` handle itself is
+closed.
 
 ```java
 import io.questdb.client.Sender;
@@ -110,15 +117,29 @@ try (Sender sender = db.borrowSender()) {
             .doubleColumn("price", 42_500.50)
             .longColumn("size", 100)
             .atNow();
-    // No explicit flush() needed: close() flushes for you.
+    // close() returns the sender; the pool ships the row by threshold or on its next settle.
 }
 ```
 
 ### Sending Batches of Rows
 
 The sender buffers rows and sends them in batches. For high throughput, write many rows on one borrowed sender and
-flush per batch rather than per row: each flush has a fixed cost, so batching raises throughput. `close()` flushes the
-final partial batch.
+flush per batch rather than per row: each flush has a fixed cost, so batching raises throughput. The final partial batch ships on the
+pool's next settle; call `flush()` before `close()` if you need it on the wire now.
+
+If you borrow per message and want each message in the store-and-forward ring the moment `close()` returns, set
+`flush_on_close=on` together with a commit window (`commit_rows`, `commit_interval_ms`): every close ships one frame
+flagged defer-commit, the server appends it, and the pool commits once per window instead of once per close. A rejected
+frame then rolls back the whole open group, so every lease since the last commit loses its rows.
+
+A server rejection covers a whole frame. With the default `flush_on_close=off` one frame carries the rows of every
+lease that closed since the last ship, so one rejected row discards the other leases' rows in that frame as well,
+after their `close()` calls returned. The default error handler logs the rejection with the frame's sequence span, and
+the error is thrown by the next call on that slot, possibly to a lease that wrote none of the rejected rows. A
+retriable rejection (write error, internal error) replays the frame; a terminal one (schema mismatch, parse error,
+security error) retires the slot, and with `sf_dir` its unacked frames stay in the slot directory for recovery. A
+lease whose rows must not share a frame calls `flush()` before `close()`, or the handle sets `flush_on_close=on`.
+`reset()` on a borrowed sender discards only the rows appended through that lease.
 
 ```java
 try (Sender sender = db.borrowSender()) {
@@ -134,7 +155,7 @@ try (Sender sender = db.borrowSender()) {
             sender.flush();
         }
     }
-    // Remaining rows are flushed by close().
+    // Remaining rows ship on the pool's next settle; flush() ships them now.
 }
 ```
 
@@ -542,6 +563,9 @@ schema::key1=value1;key2=value2;
 | `acquire_timeout_ms`     | `5000`  | How long `borrowSender()`/`borrowQuery()` waits for a free slot                |
 | `idle_timeout_ms`        | `60000` | How long a pooled connection may stay idle before it is reaped                 |
 | `max_lifetime_ms`        | `1800000` | Maximum lifetime of a pooled connection before it is recycled                |
+| `flush_on_close`         | `off`   | `on` makes a pooled `close()` flush before releasing (one frame and one commit per close, the pre-1.4 behaviour). Note: "flush" means published to the store-and-forward ring, not acknowledged. Close-time errors are wrapped; the delegate's error is the cause; with `on` and no commit window each lease ships its own frame, so a server rejection cannot discard other leases' rows; with `off` leases share frames and a rejection covers all of them; with a commit window a rejection rolls back the whole open group, every lease since the last commit |
+| `commit_rows`            | _(unset)_ | Commit window: with `flush_on_close=on`, closes ship frames flagged defer-commit and the server commits once per window of this many rows. Acks arrive at the commit; `drain()` after `flush()` |
+| `commit_interval_ms`     | _(unset)_ | Commit window age in milliseconds; setting either window key enables the window, the other defaults to 1000 rows / 100 ms. An idle slot's open group is committed at the settle cadence, so the interval only bounds groups on busy slots |
 
 ### Query routing keys
 

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

@@ -137,8 +137,25 @@ static QuestDB connect(CharSequence configurationString, HttpTokenProvider token
     /**
      * Borrows a {@link Sender} from the pool. The caller MUST call
      * {@link Sender#close()} on the returned instance to release it back to
-     * the pool. {@code close()} on a pooled Sender flushes pending rows
-     * before returning to the pool; a real disconnect only happens at
+     * the pool. Close releases, flush ships, drain waits: {@code close()} on a
+     * pooled Sender returns the slot and by default ships the slot's pending
+     * rows once the oldest is older than {@code auto_flush_interval} or another
+     * auto-flush threshold trips, otherwise it only releases; rows left in the
+     * slot go out when an auto-flush threshold trips on a later append or
+     * release, or when the housekeeper settles the idle slot, within about one
+     * {@code auto_flush_interval} of it going idle. A lease that holds the slot
+     * without appending or closing holds those rows back until it does. Call
+     * {@link Sender#flush()} to ship now and
+     * {@link Sender#drain(long)} to wait for the server. Set
+     * {@code flush_on_close=on} to have {@code close()} flush, as before 1.4.
+     * With the default, one frame carries the rows of every lease that closed
+     * since the last ship, so a server rejection of that frame discards those
+     * leases' rows together; the error is logged by the error handler and thrown
+     * by a later call on that slot. Call {@link Sender#flush()} before close, or
+     * set {@code flush_on_close=on}, when a lease's rows must not share a frame.
+     * {@link Sender#reset()} on a pooled Sender discards only that lease's rows.
+     * Note that in this client "flush" means published to the store-and-forward
+     * ring, not acknowledged by the server. A real disconnect only happens at
      * {@link #close()} on this {@code QuestDB} handle.
      * <p>
      * Allocation: zero at steady state -- the returned instance is a

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

@@ -54,24 +54,34 @@
 public final class QuestDBBuilder {
 
     static final long DEFAULT_ACQUIRE_TIMEOUT_MILLIS = 5_000;
+    static final long DEFAULT_COMMIT_INTERVAL_MILLIS = 100;
+    static final int DEFAULT_COMMIT_ROWS = 1000;
     static final long DEFAULT_HOUSEKEEPER_INTERVAL_MILLIS = 5_000;
     static final long DEFAULT_IDLE_TIMEOUT_MILLIS = 60_000;
     static final long DEFAULT_MAX_LIFETIME_MILLIS = 30 * 60_000L;
     static final int DEFAULT_POOL_MAX = 4;
     static final int DEFAULT_POOL_MIN = 1;
     static final long DEFAULT_QUERY_CLOSE_TIMEOUT_MILLIS = 5_000;
+    // Settle cadence when auto_flush_interval is not set: the WebSocket
+    // sender's own default auto-flush interval, so rows left in an idle slot
+    // wait no longer than rows in a busy one.
+    static final long DEFAULT_SETTLE_INTERVAL_MILLIS = 100;
 
     // Every valid pool value is >= 0, so -1 unambiguously marks "not set
     // explicitly". The public pool setters are the only writers of these
     // fields, so field != UNSET is exactly the "set explicitly" bit.
     private static final int UNSET = -1;
 
     private long acquireTimeoutMillis = UNSET;
+    private long commitIntervalMillis = UNSET;
+    private int commitRows = UNSET;
     // Optional ingest-side async callbacks. Null -> each pooled Sender uses its
     // loud-not-silent default. Applied to every Sender the pool builds.
     private SenderConnectionListener connectionListener;
     private BackgroundDrainerListener drainerListener;
     private SenderErrorHandler errorHandler;
+    // null = not set explicitly; the string key or the default applies.
+    private Boolean flushOnClose;
     private long housekeeperIntervalMillis = UNSET;
     private HttpTokenProvider httpTokenProvider;
     private String config;
@@ -80,6 +90,9 @@ public final class QuestDBBuilder {
     private long queryCloseTimeoutMillis = UNSET;
     private int queryPoolMax = UNSET;
     private int queryPoolMin = UNSET;
+    private int resolvedCommitRows;
+    private long resolvedCommitIntervalMillis;
+    private String resolvedIngestConfig;
     private int senderPoolMax = UNSET;
     private int senderPoolMin = UNSET;
 
@@ -114,6 +127,35 @@ public QuestDBBuilder queryCloseTimeoutMillis(long millis) {
         return this;
     }
 
+    /**
+     * Age of a commit window in milliseconds. Setting a commit window makes
+     * pooled closes ship frames flagged defer-commit: the server appends them
+     * and commits once per window, when either {@link #commitRows(int)} rows
+     * or this many milliseconds have accumulated since the last commit. Acks
+     * then arrive at the commit, so a {@link Sender#drain(long)} should follow
+     * a {@link Sender#flush()}. Requires {@code flush_on_close=on}. When only
+     * one of the two window settings is given, the other takes the auto-flush
+     * default (1000 rows, 100 ms).
+     */
+    public QuestDBBuilder commitIntervalMillis(long millis) {
+        if (millis < 1) {
+            throw new IllegalArgumentException("commitIntervalMillis must be >= 1");
+        }
+        this.commitIntervalMillis = millis;
+        return this;
+    }
+
+    /**
+     * Rows per commit window. See {@link #commitIntervalMillis(long)}.
+     */
+    public QuestDBBuilder commitRows(int rows) {
+        if (rows < 1) {
+            throw new IllegalArgumentException("commitRows must be >= 1");
+        }
+        this.commitRows = rows;
+        return this;
+    }
+
     /**
      * Sets the async connection-event listener applied to every pooled ingest
      * {@link Sender}. The listener observes connect / disconnect / failover
@@ -166,6 +208,27 @@ public QuestDBBuilder errorHandler(SenderErrorHandler handler) {
         return this;
     }
 
+    /**
+     * Whether {@code close()} on a borrowed sender flushes its rows before
+     * returning the slot. Off by default: close releases, flush ships, drain
+     * waits, and rows left in a slot ship within one auto-flush interval of the slot going
+     * idle, or at the next append or close on it after that interval. On
+     * restores the pre-1.4 behaviour, one frame and one WAL commit per close,
+     * which is the slow path for applications that borrow per message. In this
+     * client "flush" means published to the store-and-forward ring, not
+     * acknowledged by the server; use {@link Sender#drain(long)} for that.
+     * With {@code on} and no commit window every lease ships its own frame, so
+     * a server rejection cannot discard other leases' rows. With {@code off}
+     * leases share frames and a rejection covers all of them. With a commit
+     * window a rejection rolls back the whole open group, every lease since the
+     * last commit.
+     * Overrides {@code flush_on_close} in the config string.
+     */
+    public QuestDBBuilder flushOnClose(boolean flushOnClose) {
+        this.flushOnClose = flushOnClose;
+        return this;
+    }
+
     /**
      * Builds the {@link QuestDB} handle. Validates both connect strings up
      * front -- so a malformed config fails here even when both pools have
@@ -222,6 +285,53 @@ public QuestDB build() {
         resolvePoolLong(idleTimeoutMillis, "idle_timeout_ms", view, DEFAULT_IDLE_TIMEOUT_MILLIS, this::idleTimeoutMillis);
         resolvePoolLong(maxLifetimeMillis, "max_lifetime_ms", view, DEFAULT_MAX_LIFETIME_MILLIS, this::maxLifetimeMillis);
         resolvePoolLong(housekeeperIntervalMillis, "housekeeper_interval_ms", view, DEFAULT_HOUSEKEEPER_INTERVAL_MILLIS, this::housekeeperIntervalMillis);
+        if (flushOnClose == null) {
+            flushOnClose = view.getBool("flush_on_close", false);
+        }
+        int windowRows = commitRows;
+        if (windowRows == UNSET && view.has("commit_rows")) {
+            windowRows = view.getInt("commit_rows", UNSET);
+            if (windowRows < 1) {
+                throw new IllegalArgumentException("commit_rows must be >= 1, got " + windowRows);
+            }
+        }
+        long windowIntervalMillis = commitIntervalMillis;
+        if (windowIntervalMillis == UNSET && view.has("commit_interval_ms")) {
+            windowIntervalMillis = view.getLong("commit_interval_ms", UNSET);
+            if (windowIntervalMillis < 1) {
+                throw new IllegalArgumentException("commit_interval_ms must be >= 1, got " + windowIntervalMillis);
+            }
+        }
+        if (windowRows != UNSET || windowIntervalMillis != UNSET) {
+            if (windowRows == UNSET) {
+                windowRows = DEFAULT_COMMIT_ROWS;
+            }
+            if (windowIntervalMillis == UNSET) {
+                windowIntervalMillis = DEFAULT_COMMIT_INTERVAL_MILLIS;
+            }
+            if (!flushOnClose) {
+                throw new IllegalArgumentException(
+                        "conflicting configuration: commit_rows/commit_interval_ms need flush_on_close=on in this version. "
+                                + "The buffered variant (flush_on_close=off with a commit window) is not available yet; "
+                                + "set flush_on_close=on or remove the commit window keys.");
+            }
+            String transaction = view.getStr("transaction");
+            if (transaction != null && !"on".equalsIgnoreCase(transaction)) {
+                throw new IllegalArgumentException(
+                        "conflicting configuration: a commit window needs transaction=on on the pooled senders, but "
+                                + "transaction=" + transaction + " was set. Remove transaction= (the pool sets it) or set it to on.");
+            }
+            if (transaction == null) {
+                ingestConfig = injectAfterSchema(ingestConfig, "transaction=on;");
+            }
+        } else {
+            windowRows = 0;
+            windowIntervalMillis = 0;
+        }
+        resolvedCommitRows = windowRows;
+        resolvedCommitIntervalMillis = windowIntervalMillis;
+        resolvedIngestConfig = ingestConfig;
+        long settleIntervalMillis = resolveSettleIntervalMillis(view, windowIntervalMillis);
 
         return new QuestDBImpl(
                 ingestConfig,
@@ -238,7 +348,11 @@ public QuestDB build() {
                 httpTokenProvider,
                 errorHandler,
                 connectionListener,
-                drainerListener
+                drainerListener,
+                flushOnClose,
+                settleIntervalMillis,
+                windowRows,
+                windowIntervalMillis
         );
     }
 
@@ -440,6 +554,11 @@ public QuestDBBuilder senderPoolSize(int size) {
         return this;
     }
 
+    @TestOnly
+    public String ingestConfigForTest() {
+        return resolvedIngestConfig;
+    }
+
     /**
      * Snapshot of the resolved pool config, keyed by connect-string key name.
      * Valid after {@link #build()} has run pool-key resolution. Drives the
@@ -457,18 +576,44 @@ public java.util.Map<String, Object> poolConfigSnapshotForTest() {
         m.put("idle_timeout_ms", idleTimeoutMillis);
         m.put("max_lifetime_ms", maxLifetimeMillis);
         m.put("housekeeper_interval_ms", housekeeperIntervalMillis);
+        m.put("flush_on_close", flushOnClose);
+        m.put("commit_rows", resolvedCommitRows);
+        m.put("commit_interval_ms", resolvedCommitIntervalMillis);
         return m;
     }
 
+    // Inserts a key right after the "ws::" / "wss::" schema separator so the
+    // sender parse sees it as part of the ingest config.
+    private static String injectAfterSchema(String config, String keyValue) {
+        int sep = config.indexOf("::") + 2;
+        return config.substring(0, sep) + keyValue + config.substring(sep);
+    }
+
     // Inject a non-blocking async initial connect right after the schema
     // separator so lazy_connect's build never blocks or fail-fast on a down
     // server. Only used when the user set no initial_connect_retry of their own
     // (resolveLazyConnect rejects an explicit blocking mode rather than silently
     // overriding it), so placement is immaterial -- there is no competing value.
     private static String withDefaultAsyncConnect(String config) {
-        int sep = config.indexOf("::");
-        // sep >= 0: fromConfig() validated a ws/wss schema, so "::" is present.
-        return config.substring(0, sep + 2) + "initial_connect_retry=async;" + config.substring(sep + 2);
+        // fromConfig() validated a ws/wss schema, so "::" is present.
+        return injectAfterSchema(config, "initial_connect_retry=async;");
+    }
+
+    /**
+     * The housekeeper's settle cadence: {@code auto_flush_interval} when set,
+     * {@link #DEFAULT_SETTLE_INTERVAL_MILLIS} when unset. The WebSocket
+     * transport rejects both {@code auto_flush=off} and
+     * {@code auto_flush_interval=off}, so the interval is a number or unset.
+     * With a commit window, the shorter of that and {@code commitIntervalMillis}
+     * (0 = no window).
+     * Public only for the unit test, which lives in another package.
+     */
+    @TestOnly
+    public static long resolveSettleIntervalMillis(ConfigView view, long commitIntervalMillis) {
+        // getLong uses the connect-string number syntax (underscore separators
+        // included), the same one the Sender parse already accepted.
+        long fromAutoFlush = Math.max(1L, view.getLong("auto_flush_interval", DEFAULT_SETTLE_INTERVAL_MILLIS));
+        return commitIntervalMillis > 0 ? Math.min(fromAutoFlush, commitIntervalMillis) : fromAutoFlush;
     }
 
     private static void requireWebSocketSchema(CharSequence config, String role) {