Skip to content

Commit be79e84

Browse files
jerrinotjerrinot
committed
fix(ilp): report actionable QWP websocket frame-size errors
Report oversized WebSocket frames with the received frame size, configured max size, and guidance to decrease batch size. Also propagate terminal WebSocket sender failures at Sender level so async ACK errors surface promptly.
1 parent 77dbd53 commit be79e84

5 files changed

Lines changed: 199 additions & 43 deletions

File tree

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

Lines changed: 46 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151
import java.time.Instant;
5252
import java.time.temporal.ChronoUnit;
5353
import java.util.concurrent.TimeUnit;
54+
import java.util.concurrent.atomic.AtomicReference;
5455

5556
/**
5657
* QWP v1 WebSocket client sender for streaming data to QuestDB.
@@ -112,6 +113,7 @@ public class QwpWebSocketSender implements Sender {
112113
private final String host;
113114
// Flow control configuration
114115
private final int inFlightWindowSize;
116+
private final AtomicReference<LineSenderException> connectionError = new AtomicReference<>();
115117
private final int maxSchemasPerConnection;
116118
private final int port;
117119
private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
@@ -482,7 +484,7 @@ public void close() {
482484

483485
// Flush any remaining data
484486
try {
485-
if (inFlightWindowSize > 1) {
487+
if (connectionError.get() == null && inFlightWindowSize > 1) {
486488
// Async mode (window > 1): flush accumulated rows in table buffers first
487489
flushPendingRows();
488490

@@ -496,7 +498,7 @@ public void close() {
496498
} else if (inFlightWindow != null) {
497499
inFlightWindow.awaitEmpty();
498500
}
499-
} else {
501+
} else if (connectionError.get() == null) {
500502
// Sync mode (window=1): flush pending rows synchronously
501503
if (pendingRowCount > 0 && client != null && client.isConnected()) {
502504
flushSync();
@@ -744,10 +746,21 @@ public void flush() {
744746
}
745747

746748
// Wait for all pending batches to be sent to the server
747-
sendQueue.flush();
749+
try {
750+
sendQueue.flush();
751+
} catch (LineSenderException e) {
752+
checkConnectionError();
753+
throw e;
754+
}
748755

749756
// Wait for all in-flight batches to be acknowledged by the server
750-
sendQueue.awaitPendingAcks();
757+
try {
758+
sendQueue.awaitPendingAcks();
759+
} catch (LineSenderException e) {
760+
checkConnectionError();
761+
throw e;
762+
}
763+
checkConnectionError();
751764

752765
if (LOG.isDebugEnabled()) {
753766
LOG.debug("Flush complete [totalBatches={}, totalBytes={}, totalAcked={}]", sendQueue.getTotalBatchesSent(), sendQueue.getTotalBytesSent(), inFlightWindow.getTotalAcked());
@@ -1158,6 +1171,14 @@ private void checkNotClosed() {
11581171
if (closed) {
11591172
throw new LineSenderException("Sender is closed");
11601173
}
1174+
checkConnectionError();
1175+
}
1176+
1177+
private void checkConnectionError() {
1178+
LineSenderException error = connectionError.get();
1179+
if (error != null) {
1180+
throw error;
1181+
}
11611182
}
11621183

11631184
private void checkTableSelected() {
@@ -1200,9 +1221,7 @@ private void ensureActiveBufferReady() {
12001221
}
12011222

12021223
private void ensureConnected() {
1203-
if (closed) {
1204-
throw new LineSenderException("Sender is closed");
1205-
}
1224+
checkNotClosed();
12061225
if (!connected) {
12071226
// Create WebSocket client using factory (zero-GC native implementation)
12081227
if (tlsConfig != null) {
@@ -1232,7 +1251,8 @@ private void ensureConnected() {
12321251
try {
12331252
sendQueue = new WebSocketSendQueue(client, inFlightWindow,
12341253
WebSocketSendQueue.DEFAULT_ENQUEUE_TIMEOUT_MS,
1235-
WebSocketSendQueue.DEFAULT_SHUTDOWN_TIMEOUT_MS);
1254+
WebSocketSendQueue.DEFAULT_SHUTDOWN_TIMEOUT_MS,
1255+
this::recordConnectionFailure);
12361256
} catch (Throwable t) {
12371257
inFlightWindow = null;
12381258
client.close();
@@ -1248,6 +1268,7 @@ private void ensureConnected() {
12481268
// Server starts fresh on each connection, so any sender-local schema
12491269
// IDs retained from a prior connection must be discarded as well.
12501270
resetSchemaStateForNewConnection();
1271+
connectionError.set(null);
12511272

12521273
connected = true;
12531274
LOG.info("Connected to WebSocket [host={}, port={}, windowSize={}, qwpVersion={}]",
@@ -1264,12 +1285,16 @@ private void ensureNoInProgressRow() {
12641285
}
12651286
}
12661287

1267-
private void failExpectedIfNeeded(long expectedSequence, LineSenderException error) {
1268-
if (inFlightWindow != null && inFlightWindow.getLastError() == null) {
1269-
inFlightWindow.fail(expectedSequence, error);
1288+
private void failConnectionIfNeeded(LineSenderException error) {
1289+
if (recordConnectionFailure(error) && inFlightWindow != null) {
1290+
inFlightWindow.failAll(error);
12701291
}
12711292
}
12721293

1294+
private boolean recordConnectionFailure(LineSenderException error) {
1295+
return connectionError.compareAndSet(null, error);
1296+
}
1297+
12731298
/**
12741299
* Flushes pending rows by encoding and sending them.
12751300
* All non-empty tables are encoded into a single QWP v1 message and sent as one WebSocket frame.
@@ -1451,6 +1476,7 @@ private void flushSync() {
14511476

14521477
// Track batch in InFlightWindow before sending
14531478
long batchSequence = nextBatchSequence++;
1479+
checkConnectionError();
14541480
inFlightWindow.addInFlight(batchSequence);
14551481

14561482
if (LOG.isDebugEnabled()) {
@@ -1462,11 +1488,11 @@ private void flushSync() {
14621488
try {
14631489
client.sendBinary(buffer.getBufferPtr(), messageSize);
14641490
} catch (LineSenderException e) {
1465-
failExpectedIfNeeded(batchSequence, e);
1491+
failConnectionIfNeeded(e);
14661492
throw e;
14671493
} catch (Throwable t) {
14681494
LineSenderException error = new LineSenderException("Failed to send batch " + batchSequence, t);
1469-
failExpectedIfNeeded(batchSequence, error);
1495+
failConnectionIfNeeded(error);
14701496
throw error;
14711497
}
14721498

@@ -1583,6 +1609,7 @@ private void sealAndSwapBuffer() {
15831609
if (toSend.isSealed()) {
15841610
toSend.rollbackSealForRetry();
15851611
}
1612+
checkConnectionError();
15861613
throw e;
15871614
}
15881615
}
@@ -1701,24 +1728,22 @@ private void waitForAck(long expectedSequence) {
17011728
LineSenderException error = new LineSenderException(
17021729
"Server error for batch " + sequence + ": " +
17031730
ackResponse.getStatusName() + " - " + errorMessage);
1704-
inFlightWindow.fail(sequence, error);
1705-
if (sequence == expectedSequence) {
1706-
throw error;
1707-
}
1731+
failConnectionIfNeeded(error);
1732+
throw error;
17081733
}
17091734
}
17101735
} catch (LineSenderException e) {
1711-
failExpectedIfNeeded(expectedSequence, e);
1736+
failConnectionIfNeeded(e);
17121737
throw e;
17131738
} catch (Exception e) {
17141739
LineSenderException wrapped = new LineSenderException("Error waiting for ACK: " + e.getMessage(), e);
1715-
failExpectedIfNeeded(expectedSequence, wrapped);
1740+
failConnectionIfNeeded(wrapped);
17161741
throw wrapped;
17171742
}
17181743
}
17191744

17201745
LineSenderException timeout = new LineSenderException("Timeout waiting for ACK for batch " + expectedSequence);
1721-
failExpectedIfNeeded(expectedSequence, timeout);
1746+
failConnectionIfNeeded(timeout);
17221747
throw timeout;
17231748
}
17241749

@@ -1744,7 +1769,7 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
17441769

17451770
@Override
17461771
public void onClose(int code, String reason) {
1747-
throw new LineSenderException("WebSocket closed while waiting for ACK: " + reason);
1772+
throw new LineSenderException("WebSocket closed while waiting for ACK [code=" + code + ", reason=" + reason + ']');
17481773
}
17491774
}
17501775
}

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

Lines changed: 45 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,8 @@ public class WebSocketSendQueue implements QuietCloseable {
7373
private final WebSocketClient client;
7474
// Configuration
7575
private final long enqueueTimeoutMs;
76+
@Nullable
77+
private final ConnectionFailureListener connectionFailureListener;
7678
// Optional InFlightWindow for tracking sent batches awaiting ACK
7779
@Nullable
7880
private final InFlightWindow inFlightWindow;
@@ -121,6 +123,21 @@ public class WebSocketSendQueue implements QuietCloseable {
121123
*/
122124
public WebSocketSendQueue(WebSocketClient client, @Nullable InFlightWindow inFlightWindow,
123125
long enqueueTimeoutMs, long shutdownTimeoutMs) {
126+
this(client, inFlightWindow, enqueueTimeoutMs, shutdownTimeoutMs, null);
127+
}
128+
129+
/**
130+
* Creates a new send queue with custom configuration.
131+
*
132+
* @param client the WebSocket client for I/O
133+
* @param inFlightWindow the window to track sent batches awaiting ACK (may be null)
134+
* @param enqueueTimeoutMs timeout for enqueue operations (ms)
135+
* @param shutdownTimeoutMs timeout for graceful shutdown (ms)
136+
* @param connectionFailureListener notified once when the queue detects a terminal connection failure
137+
*/
138+
public WebSocketSendQueue(WebSocketClient client, @Nullable InFlightWindow inFlightWindow,
139+
long enqueueTimeoutMs, long shutdownTimeoutMs,
140+
@Nullable ConnectionFailureListener connectionFailureListener) {
124141
if (client == null) {
125142
throw new IllegalArgumentException("client cannot be null");
126143
}
@@ -129,6 +146,7 @@ public WebSocketSendQueue(WebSocketClient client, @Nullable InFlightWindow inFli
129146
this.inFlightWindow = inFlightWindow;
130147
this.enqueueTimeoutMs = enqueueTimeoutMs;
131148
this.shutdownTimeoutMs = shutdownTimeoutMs;
149+
this.connectionFailureListener = connectionFailureListener;
132150
this.running = true;
133151
this.shuttingDown = false;
134152
this.shutdownLatch = new CountDownLatch(1);
@@ -226,20 +244,20 @@ public boolean enqueue(MicrobatchBuffer buffer) {
226244
throw new LineSenderException("Buffer must be sealed before enqueue, state=" +
227245
MicrobatchBuffer.stateName(buffer.getState()));
228246
}
247+
checkError();
229248
if (!running || shuttingDown) {
249+
checkError();
230250
throw new LineSenderException("Send queue is not running");
231251
}
232252

233-
// Check for errors from I/O thread
234-
checkError();
235-
236253
final long deadline = System.currentTimeMillis() + enqueueTimeoutMs;
237254
synchronized (processingLock) {
238255
while (true) {
256+
checkError();
239257
if (!running || shuttingDown) {
258+
checkError();
240259
throw new LineSenderException("Send queue is not running");
241260
}
242-
checkError();
243261

244262
if (offerPending(buffer)) {
245263
processingLock.notifyAll();
@@ -368,12 +386,20 @@ private IoState computeState(boolean hasInFlight) {
368386
}
369387
}
370388

371-
private void failTransport(LineSenderException error) {
389+
private void failConnection(LineSenderException error) {
372390
Throwable rootError = lastError;
391+
boolean firstFailure = rootError == null;
373392
if (rootError == null) {
374393
lastError = error;
375394
rootError = error;
376395
}
396+
if (firstFailure && connectionFailureListener != null) {
397+
try {
398+
connectionFailureListener.onConnectionFailure(error);
399+
} catch (Throwable t) {
400+
LOG.error("Error notifying connection failure listener", t);
401+
}
402+
}
377403
running = false;
378404
shuttingDown = true;
379405
if (inFlightWindow != null) {
@@ -532,7 +558,7 @@ private void safeSendBatch(MicrobatchBuffer batch) {
532558
sendBatch(batch);
533559
} catch (Throwable t) {
534560
LOG.error("Error sending batch [id={}]{}", batch.getBatchId(), "", t);
535-
failTransport(new LineSenderException("Error sending batch " + batch.getBatchId() + ": " + t.getMessage(), t));
561+
failConnection(new LineSenderException("Error sending batch " + batch.getBatchId() + ": " + t.getMessage(), t));
536562
// Mark as recycled even on error to allow cleanup
537563
if (batch.isSealed()) {
538564
batch.markSending();
@@ -608,7 +634,7 @@ private boolean tryReceiveAcks() {
608634
} catch (Exception e) {
609635
if (running) {
610636
LOG.error("Error receiving response: {}", e.getMessage());
611-
failTransport(new LineSenderException("Error receiving response: " + e.getMessage(), e));
637+
failConnection(new LineSenderException("Error receiving response: " + e.getMessage(), e));
612638
}
613639
}
614640
return received;
@@ -626,6 +652,11 @@ private enum IoState {
626652
IDLE, ACTIVE, DRAINING
627653
}
628654

655+
@FunctionalInterface
656+
public interface ConnectionFailureListener {
657+
void onConnectionFailure(LineSenderException error);
658+
}
659+
629660
/**
630661
* Handler for received WebSocket frames (ACKs from server).
631662
*/
@@ -638,15 +669,15 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
638669
"Invalid ACK response payload [length=" + payloadLen + ']'
639670
);
640671
LOG.error("Invalid ACK response payload [length={}]", payloadLen);
641-
failTransport(error);
672+
failConnection(error);
642673
return;
643674
}
644675

645676
// Parse response from binary payload
646677
if (!response.readFrom(payloadPtr, payloadLen)) {
647678
LineSenderException error = new LineSenderException("Failed to parse ACK response");
648679
LOG.error("Failed to parse response");
649-
failTransport(error);
680+
failConnection(error);
650681
return;
651682
}
652683

@@ -670,20 +701,18 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
670701
String errorMessage = response.getErrorMessage();
671702
LOG.error("Error response [seq={}, status={}, error={}]", sequence, response.getStatusName(), errorMessage);
672703

673-
if (inFlightWindow != null) {
674-
LineSenderException error = new LineSenderException(
675-
"Server error for batch " + sequence + ": " +
676-
response.getStatusName() + " - " + errorMessage);
677-
inFlightWindow.fail(sequence, error);
678-
}
704+
LineSenderException error = new LineSenderException(
705+
"Server error for batch " + sequence + ": " +
706+
response.getStatusName() + " - " + errorMessage);
679707
totalErrors.incrementAndGet();
708+
failConnection(error);
680709
}
681710
}
682711

683712
@Override
684713
public void onClose(int code, String reason) {
685714
LOG.info("WebSocket closed by server [code={}, reason={}]", code, reason);
686-
failTransport(new LineSenderException("WebSocket closed by server [code=" + code + ", reason=" + reason + ']'));
715+
failConnection(new LineSenderException("WebSocket closed by server [code=" + code + ", reason=" + reason + ']'));
687716
}
688717
}
689718
}

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -461,8 +461,8 @@ public void testHighThroughputWithManyBatches() throws Exception {
461461

462462
/**
463463
* The server ACKs the first batch but returns a WRITE_ERROR for the
464-
* second. {@link WebSocketSendQueue#flush()} completes (both batches
465-
* were sent) but {@link InFlightWindow#awaitEmpty()} surfaces the error.
464+
* second. The error is treated as a terminal connection failure and is
465+
* surfaced by the next queue operation.
466466
*/
467467
@Test
468468
public void testServerErrorPropagatesOnFlush() throws Exception {
@@ -471,6 +471,7 @@ public void testServerErrorPropagatesOnFlush() throws Exception {
471471
FakeWebSocketClient client = new FakeWebSocketClient();
472472
AtomicLong highestSent = new AtomicLong(-1);
473473
AtomicLong highestDelivered = new AtomicLong(-1);
474+
CountDownLatch errorDelivered = new CountDownLatch(1);
474475

475476
client.setSendBehavior((ptr, len) -> highestSent.incrementAndGet());
476477
client.setTryReceiveBehavior(handler -> {
@@ -481,6 +482,7 @@ public void testServerErrorPropagatesOnFlush() throws Exception {
481482
highestDelivered.set(next);
482483
if (next == 1) {
483484
emitDiskFullError(handler, next);
485+
errorDelivered.countDown();
484486
} else {
485487
emitAck(handler, next);
486488
}
@@ -506,12 +508,10 @@ public void testServerErrorPropagatesOnFlush() throws Exception {
506508
buf1.seal();
507509
queue.enqueue(buf1);
508510

509-
// flush() waits for the queue to drain (both batches sent).
510-
queue.flush();
511+
assertTrue("Expected server error ACK", errorDelivered.await(2, TimeUnit.SECONDS));
511512

512-
// awaitEmpty() surfaces the server error for batch 1.
513513
try {
514-
window.awaitEmpty();
514+
queue.flush();
515515
fail("Expected server error to propagate");
516516
} catch (LineSenderException e) {
517517
assertTrue("Error should mention server failure",

0 commit comments

Comments
 (0)