Skip to content

Commit 2257daa

Browse files
glasstigerclaude
glasstiger
and
claude
committed
Drop the send loop's unused reconnect budget field
CursorWebSocketSendLoop stored reconnectMaxDurationMillis but never read it: the running connectLoop retries with no wall-clock budget (store-and-forward Invariant B), and the initial-connect budget flows through the static connectWithRetry parameter instead. Remove the dead field, its assignment, and the parameter from all eight constructor overloads, updating the delegating calls and the two production and 24 test call sites. QwpWebSocketSender and BackgroundDrainer keep their own reconnectMaxDurationMillis -- both still feed it to connectWithRetry. Also harden two PersistedSymbolDict spots for consistency with their twins: - flushChunk computes bodyLen/recLen in long like commitMappedChunk, so a near-2 GiB scratch chunk cannot wrap the CRC and write length negative on the positioned-write (fault-facade) path. - decodeLoadedSymbols decodes exactly `size` entries and throws on a truncated entry or trailing bytes instead of silently under-filling the dictionary the way a short buffer previously would. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 3314d98 commit 2257daa

16 files changed

Lines changed: 53 additions & 58 deletions

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3421,7 +3421,6 @@ private void ensureConnected() {
34213421
client, cursorEngine,
34223422
0L, CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
34233423
reconnectFactory,
3424-
reconnectMaxDurationMillis,
34253424
reconnectInitialBackoffMillis,
34263425
reconnectMaxBackoffMillis,
34273426
requestDurableAck,

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainer.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -636,7 +636,6 @@ public void run() {
636636
client, engine,
637637
0L, CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
638638
clientFactory,
639-
reconnectMaxDurationMillis,
640639
reconnectInitialBackoffMillis,
641640
reconnectMaxBackoffMillis,
642641
requestDurableAck,

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java‎

Lines changed: 11 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -252,7 +252,6 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
252252
// give-up from connectLoop. The budget still bounds the blocking (non-lazy)
253253
// initial connect via QwpWebSocketSender -> connectWithRetry, which takes it
254254
// as an explicit argument rather than reading this field.
255-
private final long reconnectMaxDurationMillis;
256255
private final WebSocketResponse response = new WebSocketResponse();
257256
private final ResponseHandler responseHandler = new ResponseHandler();
258257
private final CountDownLatch shutdownLatch = new CountDownLatch(1);
@@ -528,11 +527,10 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
528527
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
529528
long fsnAtZero, long parkNanos,
530529
ReconnectFactory reconnectFactory,
531-
long reconnectMaxDurationMillis,
532530
long reconnectInitialBackoffMillis,
533531
long reconnectMaxBackoffMillis) {
534532
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
535-
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
533+
reconnectInitialBackoffMillis,
536534
reconnectMaxBackoffMillis, false);
537535
}
538536

@@ -548,12 +546,11 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
548546
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
549547
long fsnAtZero, long parkNanos,
550548
ReconnectFactory reconnectFactory,
551-
long reconnectMaxDurationMillis,
552549
long reconnectInitialBackoffMillis,
553550
long reconnectMaxBackoffMillis,
554551
boolean durableAckMode) {
555552
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
556-
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
553+
reconnectInitialBackoffMillis,
557554
reconnectMaxBackoffMillis, durableAckMode,
558555
DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS);
559556
}
@@ -568,19 +565,18 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
568565
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
569566
long fsnAtZero, long parkNanos,
570567
ReconnectFactory reconnectFactory,
571-
long reconnectMaxDurationMillis,
572568
long reconnectInitialBackoffMillis,
573569
long reconnectMaxBackoffMillis,
574570
boolean durableAckMode,
575571
long durableAckKeepaliveIntervalMillis) {
576572
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
577-
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
573+
reconnectInitialBackoffMillis,
578574
reconnectMaxBackoffMillis, durableAckMode,
579575
durableAckKeepaliveIntervalMillis, DEFAULT_MAX_HEAD_FRAME_REJECTIONS);
580576
}
581577

582578
/**
583-
* Eleven-arg overload — omits the poison-escalation dwell window, which
579+
* Ten-arg overload — omits the poison-escalation dwell window, which
584580
* defaults to {@code 0} (legacy: escalate as soon as maxHeadFrameRejections
585581
* strikes accrue, with no minimum wall-clock). The user-facing 5s default
586582
* ({@link #DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS}) is applied at the
@@ -591,43 +587,41 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
591587
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
592588
long fsnAtZero, long parkNanos,
593589
ReconnectFactory reconnectFactory,
594-
long reconnectMaxDurationMillis,
595590
long reconnectInitialBackoffMillis,
596591
long reconnectMaxBackoffMillis,
597592
boolean durableAckMode,
598593
long durableAckKeepaliveIntervalMillis,
599594
int maxHeadFrameRejections) {
600595
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
601-
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
596+
reconnectInitialBackoffMillis,
602597
reconnectMaxBackoffMillis, durableAckMode,
603598
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections, 0L);
604599
}
605600

606601
/**
607-
* Twelve-arg overload — omits the symbol-dict cap-gap escalation dwell, which
608-
* defaults to {@code 0}. This and the thirteen-arg overload use the foreground-safe
602+
* Eleven-arg overload — omits the symbol-dict cap-gap escalation dwell, which
603+
* defaults to {@code 0}. This and the twelve-arg overload use the foreground-safe
609604
* retry-forever policy; orphan drainers opt into count-and-dwell escalation through
610605
* the policy-aware master constructor.
611606
*/
612607
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
613608
long fsnAtZero, long parkNanos,
614609
ReconnectFactory reconnectFactory,
615-
long reconnectMaxDurationMillis,
616610
long reconnectInitialBackoffMillis,
617611
long reconnectMaxBackoffMillis,
618612
boolean durableAckMode,
619613
long durableAckKeepaliveIntervalMillis,
620614
int maxHeadFrameRejections,
621615
long poisonMinEscalationWindowMillis) {
622616
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
623-
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
617+
reconnectInitialBackoffMillis,
624618
reconnectMaxBackoffMillis, durableAckMode,
625619
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
626620
poisonMinEscalationWindowMillis, 0L);
627621
}
628622

629623
/**
630-
* Thirteen-arg overload. Catch-up cap gaps are retried indefinitely, which is the
624+
* Twelve-arg overload. Catch-up cap gaps are retried indefinitely, which is the
631625
* required policy for a foreground store-and-forward sender. Orphan drainers use the
632626
* policy-aware overload below to opt into bounded terminal escalation.
633627
* <p>
@@ -647,7 +641,6 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
647641
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
648642
long fsnAtZero, long parkNanos,
649643
ReconnectFactory reconnectFactory,
650-
long reconnectMaxDurationMillis,
651644
long reconnectInitialBackoffMillis,
652645
long reconnectMaxBackoffMillis,
653646
boolean durableAckMode,
@@ -656,7 +649,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
656649
long poisonMinEscalationWindowMillis,
657650
long catchUpCapGapMinEscalationWindowMillis) {
658651
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
659-
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
652+
reconnectInitialBackoffMillis,
660653
reconnectMaxBackoffMillis, durableAckMode,
661654
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
662655
poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
@@ -671,7 +664,6 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
671664
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
672665
long fsnAtZero, long parkNanos,
673666
ReconnectFactory reconnectFactory,
674-
long reconnectMaxDurationMillis,
675667
long reconnectInitialBackoffMillis,
676668
long reconnectMaxBackoffMillis,
677669
boolean durableAckMode,
@@ -827,7 +819,6 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
827819
this.fsnAtZero = fsnAtZero;
828820
this.parkNanos = parkNanos;
829821
this.reconnectFactory = reconnectFactory;
830-
this.reconnectMaxDurationMillis = reconnectMaxDurationMillis;
831822
this.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
832823
this.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
833824
this.durableAckMode = durableAckMode;
@@ -871,7 +862,6 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
871862
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
872863
long fsnAtZero, long parkNanos,
873864
ReconnectFactory reconnectFactory,
874-
long reconnectMaxDurationMillis,
875865
long reconnectInitialBackoffMillis,
876866
long reconnectMaxBackoffMillis,
877867
boolean durableAckMode,
@@ -881,7 +871,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
881871
long catchUpCapGapMinEscalationWindowMillis,
882872
ReconnectPolicy reconnectPolicy) {
883873
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
884-
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
874+
reconnectInitialBackoffMillis,
885875
reconnectMaxBackoffMillis, durableAckMode,
886876
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
887877
poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/PersistedSymbolDict.java‎

Lines changed: 21 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -601,26 +601,30 @@ public ObjList<String> readLoadedSymbols() {
601601
private void decodeLoadedSymbols(GlobalSymbolDictionary target, ObjList<String> out) {
602602
long p = loadedEntriesAddr;
603603
long limit = p + loadedEntriesLen;
604-
for (int i = 0; i < size && p < limit; i++) {
604+
// open() CRC-validated these bytes and copyRecoveredEntries laid down exactly
605+
// `size` entries, so this decode runs on trusted, self-consistent input. Any
606+
// mismatch means an internal invariant broke: fail loud like the rest of this
607+
// file (copyRecoveredEntries throws too) instead of silently under-populating
608+
// the dictionary, which would shift every id above the short point.
609+
for (int i = 0; i < size; i++) {
605610
long len = 0;
606611
int shift = 0;
612+
boolean terminated = false;
607613
while (p < limit) {
608614
byte b = Unsafe.getUnsafe().getByte(p++);
609615
len |= (long) (b & 0x7F) << shift;
610616
if ((b & 0x80) == 0) {
617+
terminated = true;
611618
break;
612619
}
613620
shift += 7;
614621
if (shift > 35) {
615-
// Bound the varint like the other decoders: a canonical length is
616-
// <= 5 bytes. open() already CRC-validated these bytes, so this is
617-
// defensive only; the p + len > limit check below rejects the
618-
// over-long run.
619-
break;
622+
break; // over-long run -- a canonical length varint is <= 5 bytes
620623
}
621624
}
622-
if (p + len > limit) {
623-
break; // defensive: torn tail (should not happen past parse in open)
625+
if (!terminated || p + len > limit) {
626+
throw new IllegalStateException("truncated loaded symbol dictionary entry to "
627+
+ FILE_NAME + " [entry=" + i + ", size=" + size + ']');
624628
}
625629
String symbol = Utf8s.stringFromUtf8Bytes(p, p + len);
626630
if (target != null) {
@@ -630,6 +634,10 @@ private void decodeLoadedSymbols(GlobalSymbolDictionary target, ObjList<String>
630634
}
631635
p += len;
632636
}
637+
if (p != limit) {
638+
throw new IllegalStateException("loaded symbol dictionary has trailing bytes to "
639+
+ FILE_NAME + " [consumed=" + (p - loadedEntriesAddr) + ", length=" + loadedEntriesLen + ']');
640+
}
633641
}
634642

635643
/**
@@ -985,8 +993,11 @@ private void ensureScratch(long required) {
985993
* same offset.
986994
*/
987995
private void flushChunk(long recStart, int hdrLen, int entriesLen, int count) {
988-
int bodyLen = hdrLen + entriesLen;
989-
int recLen = bodyLen + CRC_SIZE;
996+
// long math, matching commitMappedChunk: hdrLen + entriesLen can reach
997+
// Integer.MAX_VALUE (entriesLen is bounded only by MAX_SCRATCH_BYTES), and an
998+
// int sum would wrap negative and hand a bogus length to the CRC and the write.
999+
long bodyLen = (long) hdrLen + entriesLen;
1000+
long recLen = bodyLen + CRC_SIZE;
9901001
Unsafe.getUnsafe().putInt(recStart + bodyLen, Crc32c.update(Crc32c.INIT, recStart, bodyLen));
9911002
appendWriteCount++;
9921003
long written = ff.write(fd, recStart, recLen, appendOffset);

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CloseOwnershipRaceTest.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,6 @@ public void closeOwnershipSnapshotNeverClaimsAnUnsurfacedError() {
7070
() -> {
7171
throw new QwpAuthFailedException(401, "localhost", 1);
7272
},
73-
0,
7473
1, 1,
7574
false,
7675
0,

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopCatchUpAlignmentTest.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -787,7 +787,7 @@ private void assertUnrelatedReconnectStateRestartsCapGapEpisode(boolean roleReje
787787
}
788788
throw new AssertionError("unexpected reconnect call " + call);
789789
},
790-
5_000L, 0L, 0L, false,
790+
0L, 0L, false,
791791
CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS,
792792
CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS,
793793
0L, TimeUnit.HOURS.toMillis(1),
@@ -904,7 +904,7 @@ private CursorWebSocketSendLoop newLoop(
904904
() -> {
905905
throw new UnsupportedOperationException("test loop is never started");
906906
},
907-
5_000L, 100L, 5_000L, false,
907+
100L, 5_000L, false,
908908
CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS,
909909
CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS,
910910
0L, capGapWindowMillis,
@@ -921,7 +921,7 @@ private CursorWebSocketSendLoop newForegroundLoop(
921921
() -> {
922922
throw new UnsupportedOperationException("test loop is never started");
923923
},
924-
5_000L, 100L, 5_000L, false,
924+
100L, 5_000L, false,
925925
CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS,
926926
CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS,
927927
0L, 0L);

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckFuzzTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -197,7 +197,7 @@ private static void runOneIteration(Rnd rnd, int iter) throws Exception {
197197
() -> {
198198
throw new UnsupportedOperationException();
199199
},
200-
5_000L, 100L, 5_000L, true);
200+
100L, 5_000L, true);
201201
Field f = CursorWebSocketSendLoop.class.getDeclaredField("nextWireSeq");
202202
f.setAccessible(true);
203203
f.setLong(loop, frames);

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -668,7 +668,7 @@ private CursorWebSocketSendLoop newDefaultLoop(CursorSendEngine engine) {
668668
() -> {
669669
throw new UnsupportedOperationException("test loop is never started");
670670
},
671-
5_000L, 100L, 5_000L, false);
671+
100L, 5_000L, false);
672672
}
673673

674674
private CursorWebSocketSendLoop newDurableLoop(CursorSendEngine engine) {
@@ -677,7 +677,7 @@ private CursorWebSocketSendLoop newDurableLoop(CursorSendEngine engine) {
677677
() -> {
678678
throw new UnsupportedOperationException("test loop is never started");
679679
},
680-
5_000L, 100L, 5_000L, true);
680+
100L, 5_000L, true);
681681
}
682682

683683
private static int pendingSize(CursorWebSocketSendLoop loop) throws Exception {

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopForegroundReconnectPolicyTest.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -109,7 +109,6 @@ private void assertAsyncInitialForegroundSurfacesTerminal(
109109
0L,
110110
CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
111111
factory,
112-
100L,
113112
1L,
114113
4L,
115114
durableAck,
@@ -167,7 +166,6 @@ private void assertForegroundRecovers(boolean durableAck, FailureSupplier failur
167166
0L,
168167
CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
169168
factory,
170-
100L,
171169
1L,
172170
4L,
173171
durableAck,

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopInterruptedCloseLeakTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,7 @@ public void testC5_interruptedCloseMustNotLeakClientInstalledByInFlightReconnect
107107
null /* async-initial-connect: the I/O thread drives the connect */,
108108
engine, 0L, 1_000L,
109109
stuckConnect,
110-
5_000L, 100L, 5_000L, false);
110+
100L, 5_000L, false);
111111
loop.start();
112112
Assert.assertTrue("I/O thread never reached the reconnect factory",
113113
enteredReconnect.await(5, TimeUnit.SECONDS));

0 commit comments

Comments
 (0)