From 72174f90bc80e33c11f19545d57ca8b030978d4e Mon Sep 17 00:00:00 2001 From: victor Date: Fri, 30 Oct 2026 01:32:59 -0700 Subject: [PATCH] Tolerate terminal errors latched before flush returns QwpWebSocketSender.flushAndGetSequence() publishes the frame and only then checks the I/O loop for a latched error, so a server that rejects the frame at once can make flush() throw the terminal it caused. Five tests called flush() unguarded against such a server and failed when the rejection won that race: - SymbolDictRecycleFsnContinuityTest: the latched-error and the sender-error-span tests - ConnectionHealthTest: the terminal-rejection health test - ServerErrorAckTerminalTest: both poison-escalation tests - IoThreadErrorSurfacedOnRowApiTest: its outer catch swallowed the early throw, so the test passed without asserting anything Each now catches the early throw at flush(), as ServerErrorAckTerminalTest already does since #111; the error is sticky, so the assertions that follow still see it. --- .../cutlass/qwp/client/ConnectionHealthTest.java | 7 ++++++- .../client/IoThreadErrorSurfacedOnRowApiTest.java | 7 ++++++- .../qwp/client/ServerErrorAckTerminalTest.java | 12 ++++++++++-- .../client/SymbolDictRecycleFsnContinuityTest.java | 14 ++++++++++++-- 4 files changed, 34 insertions(+), 6 deletions(-) diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ConnectionHealthTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ConnectionHealthTest.java index 5bf57bc97..d2472653a 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ConnectionHealthTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ConnectionHealthTest.java @@ -32,6 +32,7 @@ import io.questdb.client.cutlass.http.client.HttpClientException; import io.questdb.client.cutlass.http.client.WebSocketClientFactory; import io.questdb.client.cutlass.http.client.WebSocketUpgradeException; +import io.questdb.client.cutlass.line.LineSenderException; import io.questdb.client.cutlass.qwp.client.QwpColumnBatch; import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler; import io.questdb.client.cutlass.qwp.client.QwpConnectionHealthTracker; @@ -461,7 +462,11 @@ public void testSenderHealthIsFailedWithAnOutageAfterATerminalRejection() throws Assert.assertEquals(ConnectionHealth.NONE, h.getOutageSinceEpochMillis()); sender.table("t").longColumn("v", 1).atNow(); - sender.flush(); + try { + sender.flush(); + } catch (LineSenderException ignored) { + // The terminal NACK may arrive before flush() checks for errors. + } // the I/O thread latches FAILED before it hands the terminal error to the handler await(() -> errors.terminalCount() >= 1, 10_000, "the terminal rejection"); h = sender.health(); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/IoThreadErrorSurfacedOnRowApiTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/IoThreadErrorSurfacedOnRowApiTest.java index f9d401f6b..d14c3a7c7 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/IoThreadErrorSurfacedOnRowApiTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/IoThreadErrorSurfacedOnRowApiTest.java @@ -77,7 +77,12 @@ public void testRowApiMethodSurfacesIoThreadTerminalError() throws Exception { // handler routes the rejection through recordFatal, marking // the loop terminal. sender.table("foo").longColumn("v", 1L).atNow(); - sender.flush(); + try { + sender.flush(); + } catch (LineSenderException ignored) { + // The error ACK may arrive before flush() checks for errors; without this catch + // the outer catch would swallow it and skip every assertion below. + } // Wait for the I/O thread to record the error. After this, // cursorSendLoop.terminalError is populated and the loop has diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ServerErrorAckTerminalTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ServerErrorAckTerminalTest.java index fdc9c90d7..04fd0048c 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ServerErrorAckTerminalTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ServerErrorAckTerminalTest.java @@ -178,7 +178,11 @@ public void testRetriableNackReplaysThenPoisonEscalates() throws Exception { .build(); try { sender.table("foo").longColumn("v", 1L).atNow(); - sender.flush(); + try { + sender.flush(); + } catch (LineSenderException ignored) { + // The poison escalation may latch before flush() checks for errors. + } // Each NACK counts one poison strike and triggers a // reconnect+replay; after DEFAULT_MAX_HEAD_FRAME_REJECTIONS strikes @@ -267,7 +271,11 @@ public void testMaxFrameRejectionsConfigurableFromConnectString() throws Excepti Sender sender = Sender.fromConfig(cfg); try { sender.table("foo").longColumn("v", 1L).atNow(); - sender.flush(); + try { + sender.flush(); + } catch (LineSenderException ignored) { + // The poison escalation may latch before flush() checks for errors. + } waitFor(() -> handler.totalBinaryReceived.get() >= 2, 10_000); QwpWebSocketSender wss = (QwpWebSocketSender) sender; diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java index 5c77f3403..81c6c68bf 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java @@ -247,7 +247,12 @@ public void testSenderErrorSpansCarryExternalFsns() throws Exception { sender.setErrorHandler(e -> asyncError.compareAndSet(null, e)); try { sender.table("foo").longColumn("v", 1L).atNow(); - sender.flush(); + try { + sender.flush(); + } catch (LineSenderServerException ignored) { + // The terminal NACK may arrive before flush() checks for errors. + // The error is sticky, so the next batch below must still throw it. + } waitFor(() -> handler.totalBinaryReceived.get() >= 1, 5_000); waitFor(() -> sender.getLastTerminalError() != null, 5_000); @@ -349,7 +354,12 @@ public void testLatchedErrorStillThrowsForOldEpochTarget() throws Exception { QwpWebSocketSender sender2 = createRolledSender(server, fsn1); try { sender2.table("foo").longColumn("v", 2L).atNow(); - sender2.flush(); + try { + sender2.flush(); + } catch (LineSenderException ignored) { + // The terminal NACK may arrive before flush() checks for errors. + // The error is sticky, so awaitAckedFsn() below must still throw. + } waitFor(() -> sender2.getLastTerminalError() != null, 5_000); LineSenderException thrown = null;