Skip to content

Commit 788b85a

Browse files
bluestreak01bluestreak01
committed
test(qwp): cover the connect-walk lock policy and the pool's split stop policy (M11)
Dedicated tests for the two behaviors landed in 720d05f, plus two @testonly seams so neither test needs reflection: - QwpWebSocketSender.newBackgroundReconnectFactory(stopFlag): builds the same background ReconnectSupplier startOrphanDrainers wires into drainers. - QwpWebSocketSender.setClientFactoryOverride(factory): public setter for the existing per-attempt client factory test seam. QwpConnectWalkLockPriorityTest: while a foreground walk holds the connect-walk lock (stub client parked in connect), a background walk must yield via tryLock with a PLAIN LineSenderException (exact class asserted -- any typed terminal could quarantine a slot on contention) before reaching the client factory; once the lock is free the same factory shape proceeds into a real walk. BackgroundDrainerPoolConnectPhaseCloseTest: a drainer stuck in its connect-retry loop (seeded slot, always-throwing factory, 30s backoff) must be stop-signaled by pool.close() before the graceful window -- close under 2s, outcome STOPPED, ackedFsn untouched, no .failed sentinel. Verified red against the pre-fix pool: fails at ~2.5s (graceful-window burn) without the split stop policy.
1 parent 720d05f commit 788b85a

3 files changed

Lines changed: 440 additions & 0 deletions

File tree

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

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2038,6 +2038,30 @@ public CursorWebSocketSendLoop.ReconnectFactory newReconnectFactory() {
20382038
return new ReconnectSupplier();
20392039
}
20402040

2041+
/**
2042+
* Test seam: a BACKGROUND reconnect factory identical to the ones
2043+
* {@link #startOrphanDrainers} hands to orphan drainers (abort gate =
2044+
* the supplied stop flag, {@code isBackground()=true}), so tests can
2045+
* exercise the background side of the connect-walk lock policy (see
2046+
* {@link #buildAndConnect}) without reflection.
2047+
*/
2048+
@TestOnly
2049+
public CursorWebSocketSendLoop.ReconnectFactory newBackgroundReconnectFactory(
2050+
java.util.function.BooleanSupplier stopFlag
2051+
) {
2052+
return new ReconnectSupplier(stopFlag, "drainer stop requested during connect");
2053+
}
2054+
2055+
/**
2056+
* Test seam: installs the per-attempt WebSocket client factory override
2057+
* consulted by {@code newWebSocketClient()} inside the connect walk.
2058+
* Production code never sets it.
2059+
*/
2060+
@TestOnly
2061+
public void setClientFactoryOverride(java.util.function.Supplier<WebSocketClient> factory) {
2062+
this.clientFactoryOverride = factory;
2063+
}
2064+
20412065
@Override
20422066
public void reset() {
20432067
checkNotClosed();

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

Lines changed: 225 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,225 @@
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.DefaultHttpClientConfiguration;
28+
import io.questdb.client.cutlass.http.client.WebSocketClient;
29+
import io.questdb.client.cutlass.line.LineSenderException;
30+
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
31+
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
32+
import io.questdb.client.network.PlainSocketFactory;
33+
import io.questdb.client.test.tools.TestUtils;
34+
import org.junit.Test;
35+
36+
import java.util.ArrayList;
37+
import java.util.Collections;
38+
import java.util.List;
39+
import java.util.concurrent.CountDownLatch;
40+
import java.util.concurrent.TimeUnit;
41+
import java.util.concurrent.atomic.AtomicInteger;
42+
import java.util.concurrent.atomic.AtomicReference;
43+
44+
import static org.junit.Assert.assertEquals;
45+
import static org.junit.Assert.assertFalse;
46+
import static org.junit.Assert.assertNotNull;
47+
import static org.junit.Assert.assertSame;
48+
import static org.junit.Assert.assertTrue;
49+
import static org.junit.Assert.fail;
50+
51+
/**
52+
* Coverage of the connect-walk lock policy (M11): {@code buildAndConnect}
53+
* holds a dedicated lock across its endpoint walk, foreground walks block
54+
* on {@code lock()}, background (drainer) walks use {@code tryLock()} and
55+
* must YIELD on contention instead of queuing behind the foreground's
56+
* network I/O.
57+
* <p>
58+
* The contract under test has two halves:
59+
* <ol>
60+
* <li>a background walk attempted while a foreground walk holds the lock
61+
* throws a plain {@link LineSenderException} (transport-shaped, never
62+
* a typed terminal) BEFORE any network attempt — the drainer's retry
63+
* loops route it to indefinite capped-backoff retry (Invariant B), so
64+
* lock contention can never quarantine a slot;</li>
65+
* <li>once the lock is free, the same background factory proceeds into a
66+
* real walk (the busy throw is contention-scoped, not sticky).</li>
67+
* </ol>
68+
*/
69+
public class QwpConnectWalkLockPriorityTest {
70+
71+
/** Tracks every stub for defensive close (close() is idempotent). */
72+
private static final List<StubClient> LIVE_STUBS =
73+
Collections.synchronizedList(new ArrayList<>());
74+
75+
@Test
76+
public void testBackgroundWalkYieldsWhileForegroundHoldsLockThenProceedsWhenFree() throws Exception {
77+
TestUtils.assertMemoryLeak(() -> {
78+
try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting("localhost", 19999)) {
79+
final CountDownLatch foregroundInConnect = new CountDownLatch(1);
80+
final CountDownLatch releaseForeground = new CountDownLatch(1);
81+
final AtomicInteger factoryCalls = new AtomicInteger();
82+
sender.setClientFactoryOverride(() -> {
83+
int call = factoryCalls.incrementAndGet();
84+
StubClient stub = new StubClient(
85+
call == 1 ? foregroundInConnect : null,
86+
call == 1 ? releaseForeground : null);
87+
LIVE_STUBS.add(stub);
88+
return stub;
89+
});
90+
91+
// Foreground walk on a helper thread: its stub connect()
92+
// parks on releaseForeground, so the walk holds the
93+
// connect-walk lock for as long as this test wants.
94+
final CursorWebSocketSendLoop.ReconnectFactory foreground =
95+
sender.newReconnectFactory();
96+
final AtomicReference<Throwable> foregroundError = new AtomicReference<>();
97+
Thread fg = new Thread(() -> {
98+
try {
99+
foreground.reconnect();
100+
} catch (Throwable e) {
101+
foregroundError.set(e);
102+
}
103+
}, "test-foreground-walk");
104+
fg.setDaemon(true);
105+
fg.start();
106+
try {
107+
assertTrue("foreground walk must reach its (blocking) connect attempt",
108+
foregroundInConnect.await(5, TimeUnit.SECONDS));
109+
110+
// HALF 1: background walk while the lock is held. Must
111+
// throw the transport-shaped busy exception without
112+
// touching the client factory (tryLock fires before any
113+
// per-attempt work).
114+
final CursorWebSocketSendLoop.ReconnectFactory background =
115+
sender.newBackgroundReconnectFactory(() -> false);
116+
try {
117+
background.reconnect();
118+
fail("background walk must yield (tryLock) while the foreground "
119+
+ "holds the connect-walk lock");
120+
} catch (Exception e) {
121+
assertSame("lock contention must surface as a PLAIN LineSenderException: "
122+
+ "the drainer retry loops route exactly this shape to "
123+
+ "indefinite backoff-retry (Invariant B); any typed "
124+
+ "terminal here could quarantine a slot on contention",
125+
LineSenderException.class, e.getClass());
126+
assertTrue("busy message must name the lock (got: " + e.getMessage() + ")",
127+
e.getMessage() != null
128+
&& e.getMessage().contains("connect walk lock is busy"));
129+
}
130+
assertEquals("background contention must not reach the client factory "
131+
+ "(the yield happens before any network attempt)",
132+
1, factoryCalls.get());
133+
} finally {
134+
releaseForeground.countDown();
135+
}
136+
fg.join(5_000);
137+
assertFalse("foreground walk thread must exit once released", fg.isAlive());
138+
Throwable fgErr = foregroundError.get();
139+
assertNotNull("foreground walk fails its (single-endpoint) round once the "
140+
+ "stub connect throws", fgErr);
141+
assertTrue("foreground failure is the ordinary end-of-round connect error",
142+
fgErr instanceof LineSenderException
143+
&& String.valueOf(fgErr.getMessage()).contains("Failed to connect"));
144+
145+
// HALF 2: lock is free now — the SAME background factory must
146+
// get past tryLock and run a real walk: the factory is
147+
// consulted (call 2) and the failure is the ordinary
148+
// end-of-round error, NOT the busy throw.
149+
final CursorWebSocketSendLoop.ReconnectFactory background2 =
150+
sender.newBackgroundReconnectFactory(() -> false);
151+
try {
152+
background2.reconnect();
153+
fail("stub connect always throws; the walk must fail its round");
154+
} catch (Exception e) {
155+
assertFalse("with the lock free the background walk must proceed past "
156+
+ "tryLock (got busy throw: " + e.getMessage() + ")",
157+
String.valueOf(e.getMessage()).contains("connect walk lock is busy"));
158+
}
159+
assertEquals("the free-lock background walk must reach the client factory",
160+
2, factoryCalls.get());
161+
} finally {
162+
closeAllStubs();
163+
}
164+
});
165+
}
166+
167+
private static void closeAllStubs() {
168+
synchronized (LIVE_STUBS) {
169+
for (StubClient c : LIVE_STUBS) {
170+
try {
171+
c.close();
172+
} catch (Throwable ignored) {
173+
// best-effort; close() is idempotent
174+
}
175+
}
176+
LIVE_STUBS.clear();
177+
}
178+
}
179+
180+
/**
181+
* Real-constructor stub (native buffers allocated and freed by the base
182+
* class; the walk closes failed-attempt clients itself). {@code connect}
183+
* optionally parks on a latch to pin the walk — and thus the
184+
* connect-walk lock — then always throws, so no walk ever "succeeds"
185+
* and reaches upgrade or lifecycle commits.
186+
*/
187+
private static final class StubClient extends WebSocketClient {
188+
private final CountDownLatch entered;
189+
private final CountDownLatch release;
190+
191+
StubClient(CountDownLatch entered, CountDownLatch release) {
192+
super(DefaultHttpClientConfiguration.INSTANCE, PlainSocketFactory.INSTANCE);
193+
this.entered = entered;
194+
this.release = release;
195+
}
196+
197+
@Override
198+
public void connect(CharSequence host, int port) {
199+
if (entered != null) {
200+
entered.countDown();
201+
}
202+
if (release != null) {
203+
try {
204+
if (!release.await(10, TimeUnit.SECONDS)) {
205+
throw new RuntimeException("stub connect never released");
206+
}
207+
} catch (InterruptedException e) {
208+
Thread.currentThread().interrupt();
209+
throw new RuntimeException("stub connect interrupted", e);
210+
}
211+
}
212+
throw new RuntimeException("stub: connection refused");
213+
}
214+
215+
@Override
216+
protected void ioWait(int timeout, int op) {
217+
throw new UnsupportedOperationException("stub: no socket");
218+
}
219+
220+
@Override
221+
protected void setupIoWait() {
222+
// no-op
223+
}
224+
}
225+
}

0 commit comments

Comments
 (0)