Skip to content

Commit 350e4a3

Browse files
glasstigerglasstiger
committed
Restore QWP API compatibility
Restore the previous QwpWebSocketSender connect overload and the BackgroundDrainer constructor. Delegate both to the new catch-up dwell defaults. Use a saturating conversion for catch-up dwell milliseconds so large valid values cannot wrap negative and quarantine slots prematurely.
1 parent 12a3aec commit 350e4a3

5 files changed

Lines changed: 193 additions & 1 deletion

File tree

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

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -702,6 +702,45 @@ public static QwpWebSocketSender connect(
702702
CursorWebSocketSendLoop.DEFAULT_CATCHUP_CAP_GAP_MIN_ESCALATION_WINDOW_MILLIS);
703703
}
704704

705+
/**
706+
* Compatibility overload retained for callers compiled before the
707+
* catch-up cap-gap dwell was added to the master signature.
708+
*/
709+
public static QwpWebSocketSender connect(
710+
List<Endpoint> endpoints,
711+
ClientTlsConfiguration tlsConfig,
712+
int autoFlushRows,
713+
int autoFlushBytes,
714+
long autoFlushIntervalNanos,
715+
String authorizationHeader,
716+
boolean requestDurableAck,
717+
CursorSendEngine cursorEngine,
718+
long closeFlushTimeoutMillis,
719+
long reconnectMaxDurationMillis,
720+
long reconnectInitialBackoffMillis,
721+
long reconnectMaxBackoffMillis,
722+
Sender.InitialConnectMode initialConnectMode,
723+
SenderErrorHandler errorHandler,
724+
int errorInboxCapacity,
725+
long durableAckKeepaliveIntervalMillis,
726+
long authTimeoutMs,
727+
int connectTimeoutMs,
728+
SenderConnectionListener connectionListener,
729+
int connectionListenerInboxCapacity,
730+
int maxFrameRejections,
731+
long poisonMinEscalationWindowMillis
732+
) {
733+
return connect(endpoints, tlsConfig, autoFlushRows, autoFlushBytes,
734+
autoFlushIntervalNanos, authorizationHeader, requestDurableAck,
735+
cursorEngine, closeFlushTimeoutMillis, reconnectMaxDurationMillis,
736+
reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
737+
initialConnectMode, errorHandler, errorInboxCapacity,
738+
durableAckKeepaliveIntervalMillis, authTimeoutMs, connectTimeoutMs,
739+
connectionListener, connectionListenerInboxCapacity,
740+
maxFrameRejections, poisonMinEscalationWindowMillis,
741+
CursorWebSocketSendLoop.DEFAULT_CATCHUP_CAP_GAP_MIN_ESCALATION_WINDOW_MILLIS);
742+
}
743+
705744
/**
706745
* Master connect overload — also accepts the poison-frame detector
707746
* threshold ({@code max_frame_rejections}): consecutive server-active

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

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,31 @@ public BackgroundDrainer(
160160
CursorWebSocketSendLoop.DEFAULT_CATCHUP_CAP_GAP_MIN_ESCALATION_WINDOW_MILLIS);
161161
}
162162

163+
/**
164+
* Compatibility constructor retained for callers compiled before the
165+
* catch-up cap-gap dwell was added to the master signature.
166+
*/
167+
public BackgroundDrainer(
168+
String slotPath,
169+
long segmentSizeBytes,
170+
long sfMaxTotalBytes,
171+
CursorWebSocketSendLoop.ReconnectFactory clientFactory,
172+
long reconnectMaxDurationMillis,
173+
long reconnectInitialBackoffMillis,
174+
long reconnectMaxBackoffMillis,
175+
boolean requestDurableAck,
176+
long durableAckKeepaliveIntervalMillis,
177+
int maxHeadFrameRejections,
178+
long poisonMinEscalationWindowMillis
179+
) {
180+
this(slotPath, segmentSizeBytes, sfMaxTotalBytes, clientFactory,
181+
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
182+
reconnectMaxBackoffMillis, requestDurableAck,
183+
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
184+
poisonMinEscalationWindowMillis,
185+
CursorWebSocketSendLoop.DEFAULT_CATCHUP_CAP_GAP_MIN_ESCALATION_WINDOW_MILLIS);
186+
}
187+
163188
/**
164189
* Master constructor — also accepts the poison-frame detector threshold
165190
* ({@code max_frame_rejections}) forwarded to the drain loop's

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
import java.util.Arrays;
5454
import java.util.concurrent.CountDownLatch;
5555
import java.util.concurrent.ThreadLocalRandom;
56+
import java.util.concurrent.TimeUnit;
5657
import java.util.concurrent.atomic.AtomicLong;
5758
import java.util.concurrent.locks.LockSupport;
5859

@@ -682,8 +683,12 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
682683
"catchUpCapGapMinEscalationWindowMillis must be >= 0: "
683684
+ catchUpCapGapMinEscalationWindowMillis);
684685
}
686+
// TimeUnit conversion saturates at Long.MAX_VALUE. A raw multiply
687+
// wraps large valid millisecond values negative, which makes the
688+
// elapsed-time gate appear satisfied as soon as the strike threshold
689+
// is reached and can quarantine a recoverable orphan slot prematurely.
685690
this.catchUpCapGapMinEscalationWindowNanos =
686-
catchUpCapGapMinEscalationWindowMillis * 1_000_000L;
691+
TimeUnit.MILLISECONDS.toNanos(catchUpCapGapMinEscalationWindowMillis);
687692
if (catchUpCapGapPolicy == null) {
688693
throw new IllegalArgumentException("catchUpCapGapPolicy must be non-null");
689694
}

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

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*******************************************************************************
2+
* ___ _ ____ ____
3+
* / _ \ _ _ ___ ___| |_| _ \| __ )
4+
* | | | | | | |/ _ \/ __| __| | | | _ \
5+
* | |_| | |_| | __/\__ \ |_| |_| | |_) |
6+
* \__\_\\__,_|\___||___/\__|____/|____/
7+
*
8+
* Copyright (c) 2014-2019 Appsicle
9+
* Copyright (c) 2019-2026 QuestDB
10+
*
11+
* Licensed under the Apache License, Version 2.0 (the "License");
12+
* you may not use this file except in compliance with the License.
13+
* You may obtain a copy of the License at
14+
*
15+
* http://www.apache.org/licenses/LICENSE-2.0
16+
*
17+
* Unless required by applicable law or agreed to in writing, software
18+
* distributed under the License is distributed on an "AS IS" BASIS,
19+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20+
* See the License for the specific language governing permissions and
21+
* limitations under the License.
22+
*
23+
******************************************************************************/
24+
25+
package io.questdb.client.test.cutlass.qwp.client;
26+
27+
import io.questdb.client.ClientTlsConfiguration;
28+
import io.questdb.client.Sender;
29+
import io.questdb.client.SenderConnectionListener;
30+
import io.questdb.client.SenderErrorHandler;
31+
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
32+
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
33+
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
34+
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
35+
import org.junit.Test;
36+
37+
import java.lang.reflect.Constructor;
38+
import java.lang.reflect.Method;
39+
import java.lang.reflect.Modifier;
40+
import java.util.List;
41+
42+
import static org.junit.Assert.assertEquals;
43+
import static org.junit.Assert.assertTrue;
44+
45+
public class QwpPublicApiCompatibilityTest {
46+
47+
@Test
48+
public void testBackgroundDrainerLegacyConstructorDescriptorIsPreserved() throws Exception {
49+
Constructor<BackgroundDrainer> constructor = BackgroundDrainer.class.getConstructor(
50+
String.class,
51+
long.class,
52+
long.class,
53+
CursorWebSocketSendLoop.ReconnectFactory.class,
54+
long.class,
55+
long.class,
56+
long.class,
57+
boolean.class,
58+
long.class,
59+
int.class,
60+
long.class
61+
);
62+
assertTrue(Modifier.isPublic(constructor.getModifiers()));
63+
}
64+
65+
@Test
66+
public void testQwpWebSocketSenderLegacyConnectDescriptorIsPreserved() throws Exception {
67+
Method method = QwpWebSocketSender.class.getMethod(
68+
"connect",
69+
List.class,
70+
ClientTlsConfiguration.class,
71+
int.class,
72+
int.class,
73+
long.class,
74+
String.class,
75+
boolean.class,
76+
CursorSendEngine.class,
77+
long.class,
78+
long.class,
79+
long.class,
80+
long.class,
81+
Sender.InitialConnectMode.class,
82+
SenderErrorHandler.class,
83+
int.class,
84+
long.class,
85+
long.class,
86+
int.class,
87+
SenderConnectionListener.class,
88+
int.class,
89+
int.class,
90+
long.class
91+
);
92+
assertEquals(QwpWebSocketSender.class, method.getReturnType());
93+
assertTrue(Modifier.isPublic(method.getModifiers()));
94+
assertTrue(Modifier.isStatic(method.getModifiers()));
95+
}
96+
}

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

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -397,6 +397,33 @@ public void testCatchUpCapGapStrikesAloneDoNotLatchWithinTheEscalationWindow() t
397397
});
398398
}
399399

400+
@Test
401+
public void testCatchUpCapGapDwellConversionSaturatesInsteadOfOverflowing() throws Exception {
402+
TestUtils.assertMemoryLeak(() -> {
403+
long maxExactMillis = Long.MAX_VALUE / 1_000_000L;
404+
try (CursorSendEngine engine = newEngine()) {
405+
CursorWebSocketSendLoop exactLoop = newLoop(
406+
engine, new CatchUpCapturingClient(0), maxExactMillis);
407+
try {
408+
assertEquals(maxExactMillis * 1_000_000L,
409+
readLong(exactLoop, "catchUpCapGapMinEscalationWindowNanos"));
410+
} finally {
411+
exactLoop.close();
412+
}
413+
414+
CursorWebSocketSendLoop saturatedLoop = newLoop(
415+
engine, new CatchUpCapturingClient(0), maxExactMillis + 1L);
416+
try {
417+
assertEquals("an oversized dwell must become effectively infinite, not negative",
418+
Long.MAX_VALUE,
419+
readLong(saturatedLoop, "catchUpCapGapMinEscalationWindowNanos"));
420+
} finally {
421+
saturatedLoop.close();
422+
}
423+
}
424+
});
425+
}
426+
400427
@Test
401428
public void testCapGapEpisodeWithANegativeAnchorStillEscalates() throws Exception {
402429
// A cap-gap episode anchored at a NEGATIVE nanoTime instant must escalate like any

0 commit comments

Comments
 (0)