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/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/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); 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) {