Skip to content

Commit 4a9a6aa

Browse files
bluestreak01bluestreak01
committed
review
1 parent 54e6239 commit 4a9a6aa

5 files changed

Lines changed: 170 additions & 60 deletions

File tree

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

Lines changed: 17 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1806,6 +1806,7 @@ private boolean shouldAutoFlush() {
18061806
private void syncPing() {
18071807
client.sendPing(1000);
18081808
long deadline = System.currentTimeMillis() + InFlightWindow.DEFAULT_TIMEOUT_MS;
1809+
LineSenderException pingError = null;
18091810
while (System.currentTimeMillis() < deadline) {
18101811
sawPong = false;
18111812
sawBinaryAck = false;
@@ -1820,19 +1821,23 @@ private void syncPing() {
18201821
} else {
18211822
// Server-side error on a pending batch (parse /
18221823
// schema / security / internal / write error).
1823-
// Route through inFlightWindow.fail so the next
1824-
// waitForAck / flush surfaces it, matching the
1825-
// normal waitForAck error-handling path. Do not
1826-
// throw from syncPing: that would hide any
1827-
// durable/committed progress already observed in
1828-
// this ping round.
1829-
inFlightWindow.fail(
1830-
ackResponse.getSequence(),
1831-
new LineSenderException(ackResponse.getErrorMessage())
1832-
);
1824+
// Route through inFlightWindow.fail so subsequent
1825+
// waitForAck / flush calls also see it, capture the
1826+
// first error and throw it after PONG so the caller
1827+
// of ping() can react. We finish draining the round
1828+
// before throwing so durable/committed progress
1829+
// observed in this ping is preserved.
1830+
LineSenderException err = new LineSenderException(ackResponse.getErrorMessage());
1831+
inFlightWindow.fail(ackResponse.getSequence(), err);
1832+
if (pingError == null) {
1833+
pingError = err;
1834+
}
18331835
}
18341836
}
18351837
if (sawPong) {
1838+
if (pingError != null) {
1839+
throw pingError;
1840+
}
18361841
return;
18371842
}
18381843
}
@@ -1942,14 +1947,12 @@ private static class AckFrameHandler implements WebSocketFrameHandler {
19421947
@Override
19431948
public void onBinaryMessage(long payloadPtr, int payloadLen) {
19441949
sender.sawBinaryAck = true;
1945-
if (!WebSocketResponse.isStructurallyValid(payloadPtr, payloadLen)) {
1950+
// readFrom validates inline; a single pass parses and bounds-checks.
1951+
if (!sender.ackResponse.readFrom(payloadPtr, payloadLen)) {
19461952
throw new LineSenderException(
19471953
"Invalid ACK response payload [length=" + payloadLen + ']'
19481954
);
19491955
}
1950-
if (!sender.ackResponse.readFrom(payloadPtr, payloadLen)) {
1951-
throw new LineSenderException("Failed to parse ACK response");
1952-
}
19531956
}
19541957

19551958
@Override

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/WebSocketResponse.java‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -370,15 +370,28 @@ public int writeTo(long ptr) {
370370
return offset;
371371
}
372372

373+
// Validates inline as it parses; returns false on truncation, lying-length
374+
// entries, empty table names, or trailing garbage. On false, tableNames /
375+
// tableSeqTxns may hold partial state, but the caller (readFrom) clears
376+
// both lists at the start of every call so partial state never leaks.
373377
private boolean readTableEntries(long ptr, int remaining) {
374-
if (!validateTableEntries(ptr, remaining)) {
378+
if (remaining < 2) {
375379
return false;
376380
}
377381
int tableCount = Unsafe.getUnsafe().getShort(ptr) & 0xFFFF;
378382
int offset = 2;
379383
for (int i = 0; i < tableCount; i++) {
384+
if (remaining < offset + 2) {
385+
return false;
386+
}
380387
int nameLen = Unsafe.getUnsafe().getShort(ptr + offset) & 0xFFFF;
381388
offset += 2;
389+
// Empty table names are rejected as structurally invalid - a valid
390+
// table name is never zero bytes, and accepting empty names would
391+
// let a misbehaving server poison the per-table tracker with "" entries.
392+
if (nameLen == 0 || remaining < offset + nameLen + 8) {
393+
return false;
394+
}
382395
long nameLo = ptr + offset;
383396
long nameHi = nameLo + nameLen;
384397
offset += nameLen;
@@ -387,7 +400,7 @@ private boolean readTableEntries(long ptr, int remaining) {
387400
tableNames.add(internTableName(nameLo, nameHi));
388401
tableSeqTxns.add(seqTxn);
389402
}
390-
return true;
403+
return remaining == offset;
391404
}
392405

393406
private String internTableName(long lo, long hi) {

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

Lines changed: 34 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,11 @@ public class WebSocketSendQueue implements QuietCloseable {
8383

8484
// The I/O thread for async send/receive
8585
private final Thread ioThread;
86+
// Serializes concurrent ping() callers so each one gets its own PING/PONG
87+
// round-trip. Without this, two callers can race on pingComplete and the
88+
// second caller can return on the first caller's PONG, observing a stale
89+
// durable watermark.
90+
private final Object pingLock = new Object();
8691
// Counter for batches currently being processed by the I/O thread
8792
// This tracks batches that have been dequeued but not yet fully sent
8893
private final AtomicInteger processingCount = new AtomicInteger(0);
@@ -380,32 +385,39 @@ public long getDurableSeqTxn(CharSequence tableName) {
380385
* The server flushes pending durable ACKs before sending the PONG,
381386
* so after this method returns {@code getDurableSeqTxn()} reflects
382387
* all durable progress up to the moment the server processed the PING.
388+
* <p>
389+
* Concurrent ping callers are serialized: each caller gets its own
390+
* PING / PONG round-trip so the post-condition holds for every caller
391+
* independently. A second caller may wait up to {@code pingTimeoutMs}
392+
* for an in-flight ping to complete before its own ping starts.
383393
*/
384394
public void ping() {
385-
checkError();
386-
synchronized (processingLock) {
387-
pingComplete = false;
388-
pingRequested = true;
389-
processingLock.notifyAll();
390-
long deadline = System.nanoTime() + pingTimeoutMs * 1_000_000L;
391-
while (!pingComplete && running) {
392-
long remaining = (deadline - System.nanoTime()) / 1_000_000L;
393-
if (remaining <= 0) {
394-
throw new LineSenderException("Ping timed out");
395+
synchronized (pingLock) {
396+
checkError();
397+
synchronized (processingLock) {
398+
pingComplete = false;
399+
pingRequested = true;
400+
processingLock.notifyAll();
401+
long deadline = System.nanoTime() + pingTimeoutMs * 1_000_000L;
402+
while (!pingComplete && running) {
403+
long remaining = (deadline - System.nanoTime()) / 1_000_000L;
404+
if (remaining <= 0) {
405+
throw new LineSenderException("Ping timed out");
406+
}
407+
try {
408+
processingLock.wait(remaining);
409+
} catch (InterruptedException e) {
410+
Thread.currentThread().interrupt();
411+
throw new LineSenderException("Ping interrupted");
412+
}
395413
}
396-
try {
397-
processingLock.wait(remaining);
398-
} catch (InterruptedException e) {
399-
Thread.currentThread().interrupt();
400-
throw new LineSenderException("Ping interrupted");
414+
if (!pingComplete) {
415+
checkError();
416+
throw new LineSenderException("Ping aborted: send queue is shutting down");
401417
}
402418
}
403-
if (!pingComplete) {
404-
checkError();
405-
throw new LineSenderException("Ping aborted: send queue is shutting down");
406-
}
419+
checkError();
407420
}
408-
checkError();
409421
}
410422

411423
/**
@@ -756,7 +768,8 @@ private class ResponseHandler implements WebSocketFrameHandler {
756768

757769
@Override
758770
public void onBinaryMessage(long payloadPtr, int payloadLen) {
759-
if (!WebSocketResponse.isStructurallyValid(payloadPtr, payloadLen)) {
771+
// readFrom validates inline; a single pass parses and bounds-checks.
772+
if (!response.readFrom(payloadPtr, payloadLen)) {
760773
LineSenderException error = new LineSenderException(
761774
"Invalid ACK response payload [length=" + payloadLen + ']'
762775
);
@@ -765,14 +778,6 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
765778
return;
766779
}
767780

768-
// Parse response from binary payload
769-
if (!response.readFrom(payloadPtr, payloadLen)) {
770-
LineSenderException error = new LineSenderException("Failed to parse ACK response");
771-
LOG.error("Failed to parse response");
772-
failConnection(error);
773-
return;
774-
}
775-
776781
long sequence = response.getSequence();
777782

778783
if (response.isSuccess()) {

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderStateTest.java‎

Lines changed: 22 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -63,10 +63,9 @@ public class QwpWebSocketSenderStateTest extends AbstractTest {
6363
@Test
6464
public void testConnectionFailureIsSenderLevelTerminalState() throws Exception {
6565
assertMemoryLeak(() -> {
66-
QwpWebSocketSender sender = QwpWebSocketSender.createForTesting(
66+
try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting(
6767
"localhost", 0, 10_000, 0, 0L, 8
68-
);
69-
try {
68+
)) {
7069
LineSenderException failure = new LineSenderException(
7170
"Server error for batch 7: WRITE_ERROR - disk full"
7271
);
@@ -90,8 +89,6 @@ public void testConnectionFailureIsSenderLevelTerminalState() throws Exception {
9089
Assert.assertSame(failure, e);
9190
assertStackContains(e, "flush");
9291
}
93-
} finally {
94-
sender.close();
9592
}
9693
});
9794
}
@@ -259,19 +256,23 @@ public void testSyncPingSurfacesServerErrorFrame() throws Exception {
259256
// landed" would get a false affirmative; the error only surfaced
260257
// on the next flush's waitForAck.
261258
//
262-
// Fix: route the error through inFlightWindow.fail so it is
263-
// observable via getLastError() and the next waitForAck raises it.
259+
// Fix: capture the first error during the ping round and throw it
260+
// after PONG so ping() callers see the failure directly. Also route
261+
// through inFlightWindow.fail so subsequent waitForAck / flush
262+
// calls re-observe it. Frames arriving between the error and PONG
263+
// are still processed so durable/committed progress is preserved.
264264
assertMemoryLeak(() -> {
265265
// inFlightWindowSize=1 routes ping() through syncPing (the code under test).
266266
// The injected inFlightWindow can still hold multiple batches.
267267
QwpWebSocketSender sender = QwpWebSocketSender.createForTesting("localhost", 0, 1);
268268
PingTestClient client = new PingTestClient();
269269
try {
270-
// Server sends an error frame for seq=2, then PONG.
270+
// Server sends an error frame for seq=2, a durable ack, then PONG.
271271
client.frameSequence.add(handler -> emitBinaryResponse(
272272
handler,
273273
WebSocketResponse.error(2L, WebSocketResponse.STATUS_SCHEMA_MISMATCH, "column type mismatch")
274274
));
275+
client.frameSequence.add(handler -> emitBinaryResponse(handler, WebSocketResponse.durableAck("trades", 9)));
275276
client.frameSequence.add(handler -> handler.onPong(0, 0));
276277

277278
setField(sender, "client", client);
@@ -282,19 +283,27 @@ public void testSyncPingSurfacesServerErrorFrame() throws Exception {
282283
window.addInFlight(2);
283284
setField(sender, "inFlightWindow", window);
284285

285-
sender.ping();
286+
try {
287+
sender.ping();
288+
Assert.fail("syncPing must throw on server error frame");
289+
} catch (LineSenderException expected) {
290+
Assert.assertTrue(
291+
"error message must be propagated from the server frame",
292+
expected.getMessage() != null && expected.getMessage().contains("column type mismatch")
293+
);
294+
}
286295

287296
Assert.assertTrue(client.pingSent);
288-
// Fix: error recorded on the window, visible to the next
289-
// waitForAck / flush. Without the fix, getLastError() is null.
297+
// Durable progress observed before the throw must be preserved.
298+
Assert.assertEquals(9L, sender.getHighestDurableSeqTxn("trades"));
299+
// Error is also recorded on the window so the next waitForAck / flush sees it.
290300
Throwable err = window.getLastError();
291301
Assert.assertNotNull(
292-
"syncPing must surface server error frames via inFlightWindow.fail",
302+
"syncPing must also record the error on the inFlightWindow",
293303
err
294304
);
295305
Assert.assertTrue(err instanceof LineSenderException);
296306
Assert.assertTrue(
297-
"error message must be propagated from the server frame",
298307
err.getMessage() != null && err.getMessage().contains("column type mismatch")
299308
);
300309
} finally {

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/WebSocketSendQueueTest.java‎

Lines changed: 82 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@
5050
import static org.junit.Assert.*;
5151

5252
public class WebSocketSendQueueTest {
53-
53+
5454
@Test
5555
public void testEnqueueTimeoutWhenPendingSlotOccupied() throws Exception {
5656
assertMemoryLeak(() -> {
@@ -671,6 +671,86 @@ public void testPingSurfacesTransportError() throws Exception {
671671
});
672672
}
673673

674+
@Test
675+
public void testConcurrentPingCallersEachGetTheirOwnPing() throws Exception {
676+
// Without serialization, two concurrent ping() callers can both wake up on
677+
// the same PONG and return — the second caller observes a durable watermark
678+
// taken before its own PING was processed. The pingLock around ping()
679+
// guarantees each caller sends its own PING and waits for its own PONG.
680+
//
681+
// To trigger the bug deterministically the I/O thread is held inside the
682+
// first sendPing call until all caller threads are parked, so the buggy
683+
// code has all of them in the synchronized(processingLock) block before
684+
// any PONG is processed and only one or two PINGs are emitted in total.
685+
assertMemoryLeak(() -> {
686+
InFlightWindow window = new InFlightWindow(8, 5_000);
687+
WebSocketSendQueue queue = null;
688+
try (FakeWebSocketClient client = new FakeWebSocketClient()) {
689+
AtomicInteger pingsSent = new AtomicInteger();
690+
AtomicInteger pendingPongs = new AtomicInteger();
691+
CountDownLatch firstPingBarrier = new CountDownLatch(1);
692+
client.setPingSendBehavior(() -> {
693+
int n = pingsSent.incrementAndGet();
694+
pendingPongs.incrementAndGet();
695+
if (n == 1) {
696+
try {
697+
firstPingBarrier.await();
698+
} catch (InterruptedException e) {
699+
Thread.currentThread().interrupt();
700+
}
701+
}
702+
});
703+
client.setTryReceiveBehavior(handler -> {
704+
if (pendingPongs.get() > 0 && pendingPongs.decrementAndGet() >= 0) {
705+
handler.onPong(0, 0);
706+
return true;
707+
}
708+
return false;
709+
});
710+
711+
queue = new WebSocketSendQueue(client, window, 5_000, 500);
712+
final WebSocketSendQueue q = queue;
713+
714+
int callerCount = 3;
715+
CountDownLatch ready = new CountDownLatch(callerCount);
716+
CountDownLatch start = new CountDownLatch(1);
717+
AtomicReference<Throwable> err = new AtomicReference<>();
718+
Thread[] threads = new Thread[callerCount];
719+
for (int i = 0; i < callerCount; i++) {
720+
threads[i] = new Thread(() -> {
721+
try {
722+
ready.countDown();
723+
start.await();
724+
q.ping();
725+
} catch (Throwable t) {
726+
err.set(t);
727+
}
728+
}, "ping-caller-" + i);
729+
threads[i].start();
730+
}
731+
ready.await();
732+
start.countDown();
733+
// Wait until every caller is parked: either in processingLock.wait()
734+
// (buggy path) or BLOCKED on pingLock (fixed path).
735+
for (Thread t : threads) {
736+
awaitThreadBlocked(t);
737+
}
738+
firstPingBarrier.countDown();
739+
for (Thread t : threads) {
740+
t.join(10_000);
741+
assertFalse("ping caller " + t.getName() + " did not complete", t.isAlive());
742+
}
743+
if (err.get() != null) {
744+
throw new AssertionError("ping caller threw", err.get());
745+
}
746+
assertEquals("each concurrent caller must send its own PING",
747+
callerCount, pingsSent.get());
748+
} finally {
749+
closeQuietly(queue);
750+
}
751+
});
752+
}
753+
674754
@Test
675755
public void testDurableSeqTxnInitiallyMinusOne() throws Exception {
676756
assertMemoryLeak(() -> {
@@ -689,7 +769,7 @@ private static void awaitThreadBlocked(Thread thread) {
689769
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
690770
while (System.nanoTime() < deadline) {
691771
Thread.State state = thread.getState();
692-
if (state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING) {
772+
if (state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING || state == Thread.State.BLOCKED) {
693773
return;
694774
}
695775
Os.sleep(1);

0 commit comments

Comments
 (0)