From 6937474d233609e00e93bf42fce1cbae096f1c66 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 29 Jul 2026 15:59:16 +0200 Subject: [PATCH 1/3] fix(qwp): drop cached timestamp columns and repair accounting on buffer clear/rollback Senders cache designated-timestamp ColumnBuffer refs and derive bookkeeping from their table buffers. rollbackUncommittedColumns() (failed at()) and QwpTableBuffer.clear() close those columns, leaving dangling refs that NPE on the next at() and corrupt pending accounting. - Invalidate WebSocket timestamp caches in rollbackRow(); cancelRow() now routes through it. - Add QwpTableBuffer.Owner: clear() notifies the owning sender before releasing columns. WebSocket subtracts the buffer's exact accounted share (committedBytes - new baselineBytes, seed offsets excluded); UDP resets transient row state, datagram estimate, and headroom state (its base-estimate cache is keyed by column count alone). - Unify flush snapshot anchoring at post-reset storage so per-table contributions are exact under any reset/switch/flush history. Hot path unchanged. Regressions: clear between rows (both units, both senders), non-current/mid-row clear, accounting locality, stale headroom estimate. --- .../cutlass/qwp/client/QwpUdpSender.java | 34 ++- .../qwp/client/QwpWebSocketSender.java | 61 ++++- .../cutlass/qwp/protocol/QwpTableBuffer.java | 52 ++++- .../cutlass/qwp/client/QwpUdpSenderTest.java | 65 ++++++ .../qwp/client/QwpWebSocketSenderTest.java | 215 ++++++++++++++++++ 5 files changed, 418 insertions(+), 9 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpUdpSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpUdpSender.java index f02e87c3..f3810d3c 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpUdpSender.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpUdpSender.java @@ -67,7 +67,7 @@ * state until commit, so committed table data can be flushed without replaying * the row back into column storage. */ -public class QwpUdpSender implements Sender { +public class QwpUdpSender implements Sender, QwpTableBuffer.Owner { private static final int ADAPTIVE_HEADROOM_EWMA_SHIFT = 2; private static final Logger LOG = LoggerFactory.getLogger(QwpUdpSender.class); private static final int MAX_TABLE_NAME_LENGTH = 127; @@ -195,6 +195,27 @@ public void cancelRow() { rollbackCurrentRowToCommittedState(); } + /** + * {@link QwpTableBuffer.Owner} callback: the buffer is about to close all + * of its column buffers, so drop cached column references, in-progress + * column states, and the committed datagram estimate derived from it. + * The table's adaptive headroom state must go too: its base-estimate + * cache and EWMA samples are keyed by column count alone, which is only + * sound while columns are add-only — clear() can replace the schema with + * a different one of the same size. + */ + @Override + public void onTableBufferClear(QwpTableBuffer buffer) { + if (buffer == currentTableBuffer) { + clearTransientRowState(); + resetCommittedDatagramEstimate(); + } + TableHeadroomState headroomState = tableHeadroomStates.get(buffer.getTableName()); + if (headroomState != null) { + headroomState.reset(); + } + } + @Override public void close() { if (!closed) { @@ -580,7 +601,7 @@ public Sender table(CharSequence tableName) { currentTableName = currentTableBuffer.getTableName(); } else { currentTableName = tableName.toString(); - currentTableBuffer = new QwpTableBuffer(currentTableName); + currentTableBuffer = new QwpTableBuffer(currentTableName, this); tableBuffers.put(currentTableName, currentTableBuffer); } currentTableHeadroomState = tableHeadroomStates.get(currentTableName); @@ -1490,6 +1511,15 @@ long predictNextRowGrowth() { return Math.max(lastRowDatagramGrowth, ewmaRowDatagramGrowth); } + void reset() { + cachedBaseEstimate = -1; + cachedBaseEstimateColumnCount = -1; + committedSampleCount = 0; + ewmaRowDatagramGrowth = 0; + lastRowDatagramGrowth = 0; + schemaColumnCount = -1; + } + void recordCommittedRow(int currentSchemaColumnCount, long rowDatagramGrowth) { if (schemaColumnCount != currentSchemaColumnCount) { committedSampleCount = 0; diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java index 92de0565..a3d517e2 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java @@ -120,7 +120,7 @@ * Initial connection failures are not retained as terminal sender state; a later * operation may try to connect again. */ -public class QwpWebSocketSender implements Sender { +public class QwpWebSocketSender implements Sender, QwpTableBuffer.Owner { public static final long DEFAULT_AUTH_TIMEOUT_MS = 15_000L; // Soft per-batch byte budget. Trips a flush when raw column-buffer bytes @@ -1011,10 +1011,7 @@ public QwpWebSocketSender byteColumn(CharSequence columnName, byte value) { @Override public void cancelRow() { checkNotClosed(); - if (currentTableBuffer != null) { - currentTableBuffer.cancelCurrentRow(); - currentTableBuffer.rollbackUncommittedColumns(); - } + rollbackRow(); } /** @@ -1811,6 +1808,15 @@ public long getPendingBytes() { return pendingBytes; } + /** + * Row counterpart of {@link #getPendingBytes()}: committed-but-unflushed + * rows across all tables, as tracked for the auto-flush row threshold. + */ + @TestOnly + public int getPendingRowCount() { + return pendingRowCount; + } + /** * Server-advertised cap on the per-batch raw byte size. Zero before the * first connect; updated by every successful reconnect via @@ -1824,10 +1830,18 @@ public int getServerMaxBatchSize() { @TestOnly public QwpTableBuffer getTableBuffer(String tableName) { QwpTableBuffer buffer = tableBuffers.get(tableName); + if (buffer == currentTableBuffer && buffer != null) { + // Keep the timestamp caches and byte snapshot intact: re-snapping + // mid-row would fold in-progress bytes into the snapshot and break + // sendRow()'s delta accounting. + return buffer; + } if (buffer == null) { buffer = new QwpTableBuffer(tableName, this); tableBuffers.put(tableName, buffer); } + cachedTimestampColumn = null; + cachedTimestampNanosColumn = null; currentTableBuffer = buffer; currentTableBufferSnapshotBytes = buffer.getBufferedBytes(); currentTableName = tableName; @@ -2235,6 +2249,33 @@ public void reset() { cachedTimestampNanosColumn = null; } + /** + * {@link QwpTableBuffer.Owner} callback: the buffer is about to close all + * of its column buffers, so drop any cached column references and remove + * exactly the buffer's accounted share from the pending totals. That + * share is its committed bytes (the snapshot for the current table, which + * excludes any in-progress row; raw storage otherwise) minus its baseline + * bytes, which pendingBytes never included. Runs before the buffer wipes + * itself, so all three values still reflect the pre-clear state. + */ + @Override + public void onTableBufferClear(QwpTableBuffer buffer) { + long committedBytes; + if (buffer == currentTableBuffer) { + cachedTimestampColumn = null; + cachedTimestampNanosColumn = null; + committedBytes = currentTableBufferSnapshotBytes; + currentTableBufferSnapshotBytes = 0; + } else { + committedBytes = buffer.getBufferedBytes(); + } + pendingBytes -= committedBytes - buffer.getBaselineBytes(); + pendingRowCount -= buffer.getRowCount(); + if (pendingRowCount == 0) { + firstPendingRowTimeNanos = 0; + } + } + /** * Register an async listener for connection-state transitions: initial * connect, primary failover, endpoint attempt failures, the full address @@ -3823,7 +3864,11 @@ private void resetTableBuffersAfterFlush(ObjList keys) { } currentBatchMaxSymbolId = -1; pendingBytes = 0; - currentTableBufferSnapshotBytes = 0; + // Anchor at the post-reset storage (baseline seed bytes of var-width + // columns), matching the empty-flush path above, so per-row deltas + // accumulated from here always equal committedBytes - baselineBytes. + currentTableBufferSnapshotBytes = currentTableBuffer == null + ? 0 : currentTableBuffer.getBufferedBytes(); pendingRowCount = 0; firstPendingRowTimeNanos = 0; } @@ -3860,6 +3905,10 @@ private void resetSymbolDictStateForNewConnection() { } private void rollbackRow() { + // rollbackUncommittedColumns() may close a newly created designated + // timestamp column, so cached references must not survive rollback. + cachedTimestampColumn = null; + cachedTimestampNanosColumn = null; if (currentTableBuffer != null) { currentTableBuffer.cancelCurrentRow(); currentTableBuffer.rollbackUncommittedColumns(); diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java index 0947e5e1..a25d5b8b 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java @@ -63,8 +63,10 @@ public class QwpTableBuffer implements QuietCloseable { private static final int MAX_COLUMN_NAME_LENGTH = 127; private final LowerCaseCharSequenceIntHashMap columnNameToIndex; private final ObjList columns; + private final Owner owner; private final QwpWebSocketSender sender; private final String tableName; + private long baselineBytes; // storage retained by empty columns after reset() (e.g. var-width seed offsets) private QwpColumnDef[] cachedColumnDefs; private int columnAccessCursor; // tracks expected next column index private boolean columnDefsCacheValid; @@ -74,7 +76,7 @@ public class QwpTableBuffer implements QuietCloseable { private int rowCount; public QwpTableBuffer(String tableName) { - this(tableName, null); + this(tableName, (QwpWebSocketSender) null); } /** @@ -82,10 +84,25 @@ public QwpTableBuffer(String tableName) { * {@link ColumnBuffer#addSymbol(CharSequence)} needs the sender to * call {@link QwpWebSocketSender#getOrAddGlobalSymbol(CharSequence)}, registering * the symbol in the global dictionary shared with the server. + * The sender is also registered as the buffer's {@link Owner}. */ public QwpTableBuffer(String tableName, QwpWebSocketSender sender) { + this(tableName, sender, sender); + } + + /** + * Use this constructor overload for owners that derive state from the + * buffer but do not use the global symbol dictionary (symbol columns fall + * back to the per-buffer dictionary). + */ + public QwpTableBuffer(String tableName, Owner owner) { + this(tableName, null, owner); + } + + private QwpTableBuffer(String tableName, QwpWebSocketSender sender, Owner owner) { this.tableName = tableName; this.sender = sender; + this.owner = owner; this.columns = new ObjList<>(); this.columnNameToIndex = new LowerCaseCharSequenceIntHashMap(); this.rowCount = 0; @@ -118,8 +135,15 @@ public void cancelCurrentRow() { /** * Clears the buffer completely, including column definitions. * Frees all off-heap memory. + *

+ * The registered {@link Owner} is notified before any column state is + * released, so it can drop cached {@link ColumnBuffer} references and + * repair bookkeeping derived from the buffer's pre-clear contents. */ public void clear() { + if (owner != null) { + owner.onTableBufferClear(this); + } for (int i = 0, n = columns.size(); i < n; i++) { columns.get(i).close(); } @@ -132,6 +156,7 @@ public void clear() { rowCount = 0; columnDefsCacheValid = false; cachedColumnDefs = null; + baselineBytes = 0; } @Override @@ -139,6 +164,18 @@ public void close() { clear(); } + /** + * Bytes of storage the columns retained through the last {@link #reset()} + * (var-width columns re-seed a 4-byte initial offset entry, for example). + * These bytes are part of {@link #getBufferedBytes()} but predate any row + * committed since the reset, so an owner accounting for committed-row + * bytes must exclude them: contribution = committed bytes - baseline. + * Zero for a freshly created or {@link #clear()}ed buffer. + */ + public long getBaselineBytes() { + return baselineBytes; + } + /** * Returns the total bytes buffered across all columns. * This queries actual buffer sizes, not estimates. @@ -312,6 +349,7 @@ public void reset() { committedColumnCount = columns.size(); inProgressColumnCount = 0; rowCount = 0; + baselineBytes = getBufferedBytes(); } public void retainInProgressRow( @@ -489,6 +527,18 @@ static int elementSizeInBuffer(byte type) { } } + /** + * A sender that owns this buffer and derives state from it: cached + * {@link ColumnBuffer} references, pending-byte/row accounting, datagram + * size estimates. {@link #clear()} closes every column buffer at once, + * which invalidates all of that derived state, so it notifies the owner + * before releasing anything -- the callback may still read the + * buffer's pre-clear row count and byte sizes. + */ + public interface Owner { + void onTableBufferClear(QwpTableBuffer buffer); + } + /** * Helper class to capture array data from DoubleArray/LongArray.appendToBufPtr(). */ diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpUdpSenderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpUdpSenderTest.java index f16bc267..bc2eb9bd 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpUdpSenderTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpUdpSenderTest.java @@ -269,6 +269,45 @@ public void testAtMicrosOversizeFailureRollsBackWithoutLeakingTimestampState() t }); } + @Test + public void testAtMicrosRecoversAfterTableBufferClear() throws Exception { + assertUdpSenderRecoversAfterTableBufferClear(ChronoUnit.MICROS); + } + + @Test + public void testAtNanosRecoversAfterTableBufferClear() throws Exception { + assertUdpSenderRecoversAfterTableBufferClear(ChronoUnit.NANOS); + } + + @Test + public void testTableBufferClearInvalidatesHeadroomBaseEstimate() throws Exception { + assertMemoryLeak(() -> { + // Same column count before and after clear(), but a far longer + // column name: a stale base estimate cached for the old schema + // (keyed by column count alone) would under-estimate the new one. + String longName = repeat('n', 120); + CapturingNetworkFacade nf = new CapturingNetworkFacade(); + try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1, 1024 * 1024)) { + sender.table("t") + .longColumn("a", 1) + .atNow(); + sender.currentTableBufferForTest().clear(); + + sender.table("t") + .longColumn(longName, 2) + .atNow(); + long estimate = sender.committedDatagramEstimateForTest(); + sender.flush(); + + int actual = nf.lengths.get(nf.lengths.size() - 1); + Assert.assertTrue( + "estimate must cover the rebuilt schema [estimate=" + estimate + ", actual=" + actual + "]", + estimate >= actual + ); + } + }); + } + @Test public void testAtNanosOversizeFailureRollsBackWithoutLeakingTimestampState() throws Exception { assertMemoryLeak(() -> { @@ -1955,6 +1994,32 @@ private static void assertEstimateAtLeastActual(List rows) { } } + private static void assertUdpSenderRecoversAfterTableBufferClear(ChronoUnit unit) throws Exception { + assertMemoryLeak(() -> { + CapturingNetworkFacade nf = new CapturingNetworkFacade(); + try (QwpUdpSender sender = new QwpUdpSender(nf, 0, 0, 9000, 1, 1024 * 1024)) { + sender.table("t") + .longColumn("a", 1) + .at(1_000_000L, unit); + Assert.assertTrue(sender.committedDatagramEstimateForTest() > 0); + + // Raw clear() discards the buffered row; the sender must drop + // its cached designated-timestamp column and datagram estimate. + sender.currentTableBufferForTest().clear(); + Assert.assertEquals(0, sender.committedDatagramEstimateForTest()); + + sender.table("t") + .longColumn("a", 2) + .at(2_000_000L, unit); + sender.flush(); + } + + assertRowsEqual(Collections.singletonList( + decodedRow("t", "a", 2L, "", 2_000_000L) + ), decodeRows(nf.packets)); + }); + } + private static void assertNullableArrayNullState(QwpTableBuffer.ColumnBuffer column) { Assert.assertNotNull(column); Assert.assertEquals(1, column.getSize()); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java index f5805bb9..7b6401a1 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java @@ -166,6 +166,159 @@ public void testAtNowAfterCloseThrows() throws Exception { }); } + @Test + public void testAtMicrosRecoversAfterTableBufferClear() throws Exception { + assertTimestampCacheInvalidatedAfterTableBufferClear(ChronoUnit.MICROS); + } + + @Test + public void testAtNanosRecoversAfterTableBufferClear() throws Exception { + assertTimestampCacheInvalidatedAfterTableBufferClear(ChronoUnit.NANOS); + } + + @Test + public void testClearOfUnrelatedBufferIsLocalToThatBuffer() throws Exception { + assertMemoryLeak(() -> { + try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting( + "localhost", + 9000, + Integer.MAX_VALUE, + 0, + 0L)) { + sender.setConnectedForTest(true); + + // Give both tables var-width schemas that survive reset() + // with unaccounted seed offset bytes. + sender.table("a").stringColumn("s", "x").at(1, ChronoUnit.MICROS); + sender.table("b").stringColumn("s", "y").at(2, ChronoUnit.MICROS); + QwpTableBuffer a = sender.getTableBuffer("a"); + QwpTableBuffer b = sender.getTableBuffer("b"); + sender.reset(); + + sender.table("a").stringColumn("s", "z").at(3, ChronoUnit.MICROS); + long pendingBytesBefore = sender.getPendingBytes(); + + // Clearing rowless "b" must not disturb "a"'s accounting. + b.clear(); + Assert.assertEquals(pendingBytesBefore, sender.getPendingBytes()); + Assert.assertEquals(1, sender.getPendingRowCount()); + + // Clearing "a" must remove exactly its accounted share. + a.clear(); + Assert.assertEquals(0, sender.getPendingBytes()); + Assert.assertEquals(0, sender.getPendingRowCount()); + } + }); + } + + @Test + public void testClearOfEmptyVarWidthTableKeepsPendingBytesIntact() throws Exception { + assertMemoryLeak(() -> { + try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting( + "localhost", + 9000, + Integer.MAX_VALUE, + 0, + 0L)) { + sender.setConnectedForTest(true); + + sender.table("a") + .stringColumn("s", "x") + .at(1, ChronoUnit.MICROS); + QwpTableBuffer a = sender.getTableBuffer("a"); + + // reset() keeps "a"'s var-width column, which retains seed + // offset bytes that are not part of the pending accounting + sender.reset(); + Assert.assertTrue(a.getBufferedBytes() > 0); + + sender.table("b") + .longColumn("v", 1) + .at(2, ChronoUnit.MICROS); + long pendingBytesBefore = sender.getPendingBytes(); + + a.clear(); // contributed nothing since reset() + + Assert.assertEquals(pendingBytesBefore, sender.getPendingBytes()); + Assert.assertEquals(1, sender.getPendingRowCount()); + } + }); + } + + @Test + public void testClearOfCurrentTableBufferDiscardsUnfinishedRow() throws Exception { + assertMemoryLeak(() -> { + try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting( + "localhost", + 9000, + Integer.MAX_VALUE, + 0, + 0L)) { + sender.setConnectedForTest(true); + QwpTableBuffer buffer = sender.getTableBuffer("t"); + + sender.table("t").longColumn("a", 1); // row left unfinished + buffer.clear(); + + sender.table("t") + .longColumn("a", 2) + .at(1, ChronoUnit.MICROS); + + Assert.assertEquals(1, buffer.getRowCount()); + Assert.assertEquals(1, sender.getPendingRowCount()); + Assert.assertEquals(sender.totalBufferedBytes(), sender.getPendingBytes()); + } + }); + } + + @Test + public void testClearOfNonCurrentTableBufferRepairsPendingAccounting() throws Exception { + assertMemoryLeak(() -> { + try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting( + "localhost", + 9000, + Integer.MAX_VALUE, + 0, + 0L)) { + sender.setConnectedForTest(true); + + sender.table("t1") + .longColumn("a", 1) + .at(1, ChronoUnit.MICROS); + QwpTableBuffer t1 = sender.getTableBuffer("t1"); + + sender.table("t2") + .longColumn("b", 2) + .at(2, ChronoUnit.MICROS); + + t1.clear(); // t2 is the current table + + Assert.assertEquals(0, t1.getRowCount()); + Assert.assertEquals(1, sender.getPendingRowCount()); + Assert.assertEquals(sender.totalBufferedBytes(), sender.getPendingBytes()); + + // the cleared table stays usable + sender.table("t1") + .longColumn("a", 3) + .at(3, ChronoUnit.MICROS); + + Assert.assertEquals(1, t1.getRowCount()); + Assert.assertEquals(2, sender.getPendingRowCount()); + Assert.assertEquals(sender.totalBufferedBytes(), sender.getPendingBytes()); + } + }); + } + + @Test + public void testAtMicrosRecoversAfterOversizeRowRollback() throws Exception { + assertTimestampCacheInvalidatedAfterOversizeRowRollback(ChronoUnit.MICROS); + } + + @Test + public void testAtNanosRecoversAfterOversizeRowRollback() throws Exception { + assertTimestampCacheInvalidatedAfterOversizeRowRollback(ChronoUnit.NANOS); + } + @Test public void testBinaryColumnAfterCloseThrows() throws Exception { assertMemoryLeak(() -> { @@ -732,6 +885,68 @@ private static void assertClosed(Runnable r) { } } + private static void assertTimestampCacheInvalidatedAfterTableBufferClear(ChronoUnit unit) throws Exception { + assertMemoryLeak(() -> { + try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting( + "localhost", + 9000, + Integer.MAX_VALUE, + 0, + 0L)) { + sender.setConnectedForTest(true); + + sender.table("t") + .longColumn("a", 1) + .stringColumn("s", "before") + .at(1, unit); + + sender.getTableBuffer("t").clear(); + + sender.table("t") + .longColumn("a", 2) + .stringColumn("s", "after") + .at(2, unit); + + QwpTableBuffer tableBuffer = sender.getTableBuffer("t"); + Assert.assertEquals(1, tableBuffer.getRowCount()); + Assert.assertEquals(1, sender.getPendingRowCount()); + Assert.assertEquals(sender.totalBufferedBytes(), sender.getPendingBytes()); + } + }); + } + + private static void assertTimestampCacheInvalidatedAfterOversizeRowRollback(ChronoUnit unit) throws Exception { + assertMemoryLeak(() -> { + try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting( + "localhost", + 9000, + Integer.MAX_VALUE, + 0, + 0L)) { + sender.setConnectedForTest(true); + sender.applyServerBatchSizeLimit(17); + + try { + sender.table("t") + .longColumn("a", 1) + .longColumn("b", 2) + .at(1, unit); + Assert.fail("Expected oversized row rejection"); + } catch (LineSenderException e) { + Assert.assertTrue(e.getMessage().contains("row too large")); + } + + sender.table("t") + .longColumn("a", 3) + .at(2, unit); + + QwpTableBuffer tableBuffer = sender.getTableBuffer("t"); + Assert.assertEquals(1, tableBuffer.getRowCount()); + Assert.assertEquals(sender.totalBufferedBytes(), sender.getPendingBytes()); + } + }); + } + private static MicrobatchBuffer getMicrobatchBuffer(QwpWebSocketSender sender, String fieldName) throws Exception { Field field = QwpWebSocketSender.class.getDeclaredField(fieldName); field.setAccessible(true); From 530920ead70abd3bf709e9c0a8173a9ca4152ea0 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 29 Jul 2026 17:55:17 +0200 Subject: [PATCH 2/3] test(pool): make close interrupt deadline checks deterministic Track the bounded creation wait and handled interrupts in one atomic test witness. Handshake repeated interrupts until the original wait exits so the regression tests reject deadline resets without depending on scheduler timing. --- .../questdb/client/impl/QueryClientPool.java | 33 ++++- .../io/questdb/client/impl/SenderPool.java | 33 ++++- .../impl/QuestDBImplCloseLifecycleTest.java | 136 ++++++++++-------- 3 files changed, 132 insertions(+), 70 deletions(-) diff --git a/core/src/main/java/io/questdb/client/impl/QueryClientPool.java b/core/src/main/java/io/questdb/client/impl/QueryClientPool.java index 75323a12..fde3bc06 100644 --- a/core/src/main/java/io/questdb/client/impl/QueryClientPool.java +++ b/core/src/main/java/io/questdb/client/impl/QueryClientPool.java @@ -97,6 +97,12 @@ public final class QueryClientPool implements AutoCloseable { // DEFAULT_CLOSE_QUERY_TIMEOUT_MILLIS. Volatile because QuestDBImpl sets it // once at build time on a different thread than the borrowers that read it. private volatile long closeQueryTimeoutMillis = DEFAULT_CLOSE_QUERY_TIMEOUT_MILLIS; + // Atomic test witness for the bounded creation-wait loop. The sign bit + // means active; the remaining bits count handled InterruptedExceptions. + // Condition.hasWaiters() cannot serve this purpose because an interrupt + // transiently transfers the closer out of the condition queue. Volatile + // lets white-box tests observe active/count as one coherent snapshot. + private volatile long creationWaitState; private int inFlightCreations; public QueryClientPool( @@ -339,13 +345,21 @@ public void close() { final long creationWaitDeadlineNanos = System.nanoTime() + creationWaitNanos; long creationRemainingNanos = creationWaitNanos; boolean creationWaitInterrupted = false; - while (inFlightCreations > 0 && creationRemainingNanos > 0) { - try { - creationFinished.awaitNanos(creationRemainingNanos); - } catch (InterruptedException e) { - creationWaitInterrupted = true; + creationWaitState = inFlightCreations > 0 && creationRemainingNanos > 0 + ? Long.MIN_VALUE + : 0; + try { + while (inFlightCreations > 0 && creationRemainingNanos > 0) { + try { + creationFinished.awaitNanos(creationRemainingNanos); + } catch (InterruptedException e) { + creationWaitInterrupted = true; + creationWaitState++; + } + creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime(); } - creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime(); + } finally { + creationWaitState &= Long.MAX_VALUE; } if (creationWaitInterrupted) { Thread.currentThread().interrupt(); @@ -528,11 +542,16 @@ void release(QueryWorker w, long gen) { } } + /** + * Returns whether {@link #close()} is inside its bounded wait for + * in-flight creations. This remains true while an interrupt transfers the + * closer out of the condition queue and it re-enters with the same deadline. + */ @TestOnly public boolean hasCreationWaiterForTesting() { lock.lock(); try { - return lock.hasWaiters(creationFinished); + return creationWaitState < 0; } finally { lock.unlock(); } diff --git a/core/src/main/java/io/questdb/client/impl/SenderPool.java b/core/src/main/java/io/questdb/client/impl/SenderPool.java index 2d01059e..fcbfa2bb 100644 --- a/core/src/main/java/io/questdb/client/impl/SenderPool.java +++ b/core/src/main/java/io/questdb/client/impl/SenderPool.java @@ -238,6 +238,12 @@ public final class SenderPool implements AutoCloseable { // (cancelling recovery) WITHOUT making a later close() short-circuit the // teardown. Guarded by lock. private boolean closeStarted; + // Atomic test witness for the bounded creation-wait loop. The sign bit + // means active; the remaining bits count handled InterruptedExceptions. + // Condition.hasWaiters() cannot serve this purpose because an interrupt + // transiently transfers the closer out of the condition queue. Volatile + // lets white-box tests observe active/count as one coherent snapshot. + private volatile long creationWaitState; private int inFlightCreations; // Lease teardowns currently running on borrower threads (retireLease's // delegate-close section, outside the lock). close() counts these as @@ -1339,11 +1345,16 @@ public Thread getStartupRecoveryThreadForTesting() { return startupRecoveryThread; } + /** + * Returns whether {@link #close()} is inside its bounded wait for + * in-flight creations. This remains true while an interrupt transfers the + * closer out of the condition queue and it re-enters with the same deadline. + */ @TestOnly public boolean hasCreationWaiterForTesting() { lock.lock(); try { - return lock.hasWaiters(creationFinished); + return creationWaitState < 0; } finally { lock.unlock(); } @@ -1483,13 +1494,21 @@ public void close() { final long creationWaitDeadlineNanos = System.nanoTime() + creationWaitNanos; long creationRemainingNanos = creationWaitNanos; boolean creationWaitInterrupted = false; - while (inFlightCreations > 0 && creationRemainingNanos > 0) { - try { - creationFinished.awaitNanos(creationRemainingNanos); - } catch (InterruptedException e) { - creationWaitInterrupted = true; + creationWaitState = inFlightCreations > 0 && creationRemainingNanos > 0 + ? Long.MIN_VALUE + : 0; + try { + while (inFlightCreations > 0 && creationRemainingNanos > 0) { + try { + creationFinished.awaitNanos(creationRemainingNanos); + } catch (InterruptedException e) { + creationWaitInterrupted = true; + creationWaitState++; + } + creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime(); } - creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime(); + } finally { + creationWaitState &= Long.MAX_VALUE; } if (creationWaitInterrupted) { Thread.currentThread().interrupt(); diff --git a/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java b/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java index 6344336b..67feabd8 100644 --- a/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java +++ b/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java @@ -43,13 +43,18 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; -import java.util.function.BooleanSupplier; +import java.util.concurrent.locks.LockSupport; import java.util.function.Consumer; import java.util.function.IntFunction; +import java.util.function.LongSupplier; public class QuestDBImplCloseLifecycleTest { + private static final long QUERY_CREATION_WAIT_STATE_OFFSET = + Unsafe.getFieldOffset(QueryClientPool.class, "creationWaitState"); private static final String QUERY_CFG = "ws::addr=127.0.0.1:9000;"; + private static final long SENDER_CREATION_WAIT_STATE_OFFSET = + Unsafe.getFieldOffset(SenderPool.class, "creationWaitState"); private static final String SENDER_CFG = "http::addr=127.0.0.1:1;protocol_version=2;auto_flush=off;"; @Test(timeout = 30_000) @@ -186,18 +191,14 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr inCreation.countDown(); awaitOrFail(releaseCreation, "test never released query creation"); }; - // A 1s creation-wait budget, not 100ms: the interrupt storm below must land at least - // twice inside this window for the deadline-restart property to be exercised at all, - // and a freshly started, yielding interrupter thread is not guaranteed two scheduler - // quanta within 100ms on a saturated CI agent (observed on hosted 3-core mac agents, - // where the post-join count assert failed with the product deadline honored exactly). + // A 1s creation-wait budget, not 100ms, leaves enough room to + // establish the handshaked interrupt loop before asserting the + // original absolute deadline. QuestDBImpl db = newQuestDB( SENDER_CFG, 0, 0, 1000, slotIndex -> fakeSender(null, null, null), connectHook); QueryClientPool pool = db.getQueryPoolForTesting(); AtomicReference borrowOutcome = new AtomicReference<>(); AtomicBoolean closeReturnedInterrupted = new AtomicBoolean(); - AtomicBoolean keepInterrupting = new AtomicBoolean(true); - AtomicInteger interruptCount = new AtomicInteger(); Thread borrower = new Thread(() -> { try { db.borrowQuery(); @@ -209,13 +210,6 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr db.close(); closeReturnedInterrupted.set(Thread.currentThread().isInterrupted()); }, "interrupted-query-closer"); - Thread interrupter = new Thread(() -> { - while (keepInterrupting.get()) { - interruptCount.incrementAndGet(); - closer.interrupt(); - Thread.yield(); - } - }, "query-close-interrupter"); long nativeBaseline = Unsafe.getMemUsedByTag(MemoryTag.NATIVE_DEFAULT); try { borrower.start(); @@ -226,14 +220,17 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr closer.start(); awaitCreationWaiter(pool, "facade close did not wait while query construction was internally owned"); - interrupter.start(); - awaitRepeatedInterrupts(interruptCount, pool::hasCreationWaiterForTesting, - "query close left its creation wait before the interrupt storm landed twice"); + long handledInterrupts = interruptUntilCreationWaitEnds( + closer, + () -> getCreationWaitState(pool), + "query creation wait did not honor its absolute deadline"); closer.join(TimeUnit.SECONDS.toMillis(5)); Assert.assertFalse( "repeated interrupts restarted the query creation-wait deadline", closer.isAlive()); - Assert.assertTrue("test did not repeatedly interrupt query close", interruptCount.get() > 1); + Assert.assertTrue( + "query close did not handle both interrupts", + handledInterrupts > 1); Assert.assertTrue("facade close must restore query closer interruption", closeReturnedInterrupted.get()); Assert.assertEquals( @@ -252,9 +249,7 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr borrowOutcome.get() instanceof QueryException && String.valueOf(borrowOutcome.get().getMessage()).contains("closed")); } finally { - keepInterrupting.set(false); releaseCreation.countDown(); - interrupter.join(TimeUnit.SECONDS.toMillis(10)); db.close(); borrower.join(TimeUnit.SECONDS.toMillis(10)); closer.join(TimeUnit.SECONDS.toMillis(10)); @@ -275,15 +270,13 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th }; String senderConfig = "ws::addr=localhost:1;sf_dir=" + System.getProperty("java.io.tmpdir") + "/qdb-interrupted-pool-" + System.nanoTime() + ";"; - // 1s creation-wait budget for the same reason as the query-interrupt test above: the - // interrupt storm must land at least twice inside the window even on a saturated agent. + // Use the same 1s budget and handshaked interrupt loop as the query + // lifecycle test above. QuestDBImpl db = newQuestDB(senderConfig, 0, 0, 1000, senderFactory, client -> { }); SenderPool pool = db.getSenderPoolForTesting(); AtomicReference borrowOutcome = new AtomicReference<>(); AtomicBoolean closeReturnedInterrupted = new AtomicBoolean(); - AtomicBoolean keepInterrupting = new AtomicBoolean(true); - AtomicInteger interruptCount = new AtomicInteger(); Thread borrower = new Thread(() -> { try { db.borrowSender(); @@ -295,13 +288,6 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th db.close(); closeReturnedInterrupted.set(Thread.currentThread().isInterrupted()); }, "interrupted-sender-closer"); - Thread interrupter = new Thread(() -> { - while (keepInterrupting.get()) { - interruptCount.incrementAndGet(); - closer.interrupt(); - Thread.yield(); - } - }, "sender-close-interrupter"); try { borrower.start(); Assert.assertTrue("sender borrow never reached construction", @@ -313,14 +299,17 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th closer.start(); awaitCreationWaiter(pool, "facade close did not wait while sender construction was internally owned"); - interrupter.start(); - awaitRepeatedInterrupts(interruptCount, pool::hasCreationWaiterForTesting, - "sender close left its creation wait before the interrupt storm landed twice"); + long handledInterrupts = interruptUntilCreationWaitEnds( + closer, + () -> getCreationWaitState(pool), + "sender creation wait did not honor its absolute deadline"); closer.join(TimeUnit.SECONDS.toMillis(5)); Assert.assertFalse( "repeated interrupts restarted the sender creation-wait deadline", closer.isAlive()); - Assert.assertTrue("test did not repeatedly interrupt sender close", interruptCount.get() > 1); + Assert.assertTrue( + "sender close did not handle both interrupts", + handledInterrupts > 1); Assert.assertTrue("facade close must restore sender closer interruption", closeReturnedInterrupted.get()); Assert.assertEquals( @@ -341,9 +330,7 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th borrowOutcome.get() instanceof LineSenderException && String.valueOf(borrowOutcome.get().getMessage()).contains("closed")); } finally { - keepInterrupting.set(false); releaseCreation.countDown(); - interrupter.join(TimeUnit.SECONDS.toMillis(10)); db.close(); borrower.join(TimeUnit.SECONDS.toMillis(10)); closer.join(TimeUnit.SECONDS.toMillis(10)); @@ -529,32 +516,69 @@ private static void awaitCreationWaiter(SenderPool pool, String message) { Assert.fail(message); } + private static long getCreationWaitState(QueryClientPool pool) { + return Unsafe.getUnsafe().getLongVolatile(pool, QUERY_CREATION_WAIT_STATE_OFFSET); + } + + private static long getCreationWaitState(SenderPool pool) { + return Unsafe.getUnsafe().getLongVolatile(pool, SENDER_CREATION_WAIT_STATE_OFFSET); + } + /** - * Holds the test until the interrupt storm has landed at least twice while the facade close is - * still inside its bounded creation wait. The deadline-restart property is only exercised by - * interrupts that arrive during that wait, and the scheduler owes the interrupter thread - * nothing: with a post-join count assert alone, the run races the close budget against thread - * scheduling and can fail with the product invariant intact. Failing here instead separates - * "interrupter starved before the budget expired" from a genuine deadline bug. + * Delivers one interrupt at a time, waiting until the creation-wait loop + * handles each one before delivering the next. This prevents interrupt + * coalescing while keeping pressure on the wait until its original absolute + * deadline expires. The state snapshot encodes active in the sign bit and + * the handled-interrupt count in the remaining bits. */ - private static void awaitRepeatedInterrupts( - AtomicInteger interruptCount, - BooleanSupplier closerStillWaiting, + private static long interruptUntilCreationWaitEnds( + Thread closer, + LongSupplier creationWaitState, String message ) { - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); while (System.nanoTime() < deadline) { - // Count first: two interrupts observed while polling means the storm landed no matter - // how quickly the wait ends afterwards, so a budget expiry seen next is not a failure. - if (interruptCount.get() > 1) { - return; + long state = creationWaitState.getAsLong(); + long handledInterrupts = state & Long.MAX_VALUE; + if (state >= 0) { + Assert.assertTrue( + message + "; wait ended after handling only " + handledInterrupts + " interrupt(s)", + handledInterrupts > 1); + return handledInterrupts; } - if (!closerStillWaiting.getAsBoolean()) { - Assert.fail(message + "; interrupts landed: " + interruptCount.get()); + + closer.interrupt(); + long countBeforeInterrupt = handledInterrupts; + while (System.nanoTime() < deadline) { + state = creationWaitState.getAsLong(); + handledInterrupts = state & Long.MAX_VALUE; + if (state >= 0) { + Assert.assertTrue( + message + "; wait ended after handling only " + handledInterrupts + " interrupt(s)", + handledInterrupts > 1); + return handledInterrupts; + } + if (handledInterrupts > countBeforeInterrupt) { + // Keep enough interrupt pressure to expose a restarted + // deadline without monopolising a CI worker core. + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); + break; + } + Thread.yield(); } - Thread.yield(); } - Assert.fail(message + "; interrupts landed: " + interruptCount.get()); + + long state = creationWaitState.getAsLong(); + long handledInterrupts = state & Long.MAX_VALUE; + if (state >= 0) { + Assert.assertTrue( + message + "; wait ended after handling only " + handledInterrupts + " interrupt(s)", + handledInterrupts > 1); + return handledInterrupts; + } + Assert.fail(message + "; creation wait still active after handling " + + handledInterrupts + " interrupt(s)"); + return 0; } private static void awaitOrFail(CountDownLatch latch, String message) { From bf275aa7c818e07be9e8d505914a20eb4688d8d0 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 30 Jul 2026 09:57:39 +0200 Subject: [PATCH 3/3] test(qwp): cover clear and post-flush byte accounting Exercise current-buffer clear with committed and in-progress rows so cached timestamps and pending counters must be repaired. Add a real flush regression that excludes retained var-width seed bytes from subsequent pending-byte accounting. --- .../qwp/client/QwpWebSocketSenderTest.java | 59 ++++++++++++++++++- 1 file changed, 56 insertions(+), 3 deletions(-) diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java index 7b6401a1..131d5915 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java @@ -257,12 +257,19 @@ public void testClearOfCurrentTableBufferDiscardsUnfinishedRow() throws Exceptio sender.setConnectedForTest(true); QwpTableBuffer buffer = sender.getTableBuffer("t"); - sender.table("t").longColumn("a", 1); // row left unfinished + sender.table("t") + .stringColumn("s", "committed") + .at(1, ChronoUnit.MICROS); + sender.table("t").stringColumn("s", "unfinished"); buffer.clear(); + Assert.assertEquals(0, buffer.getRowCount()); + Assert.assertEquals(0, sender.getPendingRowCount()); + Assert.assertEquals(0, sender.getPendingBytes()); + sender.table("t") - .longColumn("a", 2) - .at(1, ChronoUnit.MICROS); + .stringColumn("s", "after") + .at(2, ChronoUnit.MICROS); Assert.assertEquals(1, buffer.getRowCount()); Assert.assertEquals(1, sender.getPendingRowCount()); @@ -309,6 +316,52 @@ public void testClearOfNonCurrentTableBufferRepairsPendingAccounting() throws Ex }); } + @Test + public void testFlushReanchorsPendingBytesAtVarWidthBaseline() throws Exception { + assertMemoryLeak(() -> { + try (TestWebSocketServer server = new TestWebSocketServer( + new TestWebSocketServer.WebSocketServerHandler() { + })) { + server.start(); + Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); + + CursorSendEngine engine = new CursorSendEngine(null, 4L * 1024 * 1024); + try (QwpWebSocketSender sender = QwpWebSocketSender.connect( + "localhost", + server.getPort(), + null, + Integer.MAX_VALUE, + 0, + 0L, + null, + false, + engine, + 0L)) { + sender.table("t") + .stringColumn("s", "before") + .at(1, ChronoUnit.MICROS); + sender.flush(); + + QwpTableBuffer buffer = sender.getTableBuffer("t"); + long baselineBytes = buffer.getBaselineBytes(); + Assert.assertTrue( + "flush must retain var-width seed storage", + baselineBytes > 0); + Assert.assertEquals(baselineBytes, sender.totalBufferedBytes()); + Assert.assertEquals(0, sender.getPendingBytes()); + + sender.table("t") + .stringColumn("s", "after") + .at(2, ChronoUnit.MICROS); + + Assert.assertEquals( + sender.totalBufferedBytes() - baselineBytes, + sender.getPendingBytes()); + } + } + }); + } + @Test public void testAtMicrosRecoversAfterOversizeRowRollback() throws Exception { assertTimestampCacheInvalidatedAfterOversizeRowRollback(ChronoUnit.MICROS);