Skip to content

Commit 38cdf3e

Browse files
bluestreak01bluestreak01
committed
test(qwp): close the three remaining C-2 test-gate rows with verified mutant-killers
Round-3 review re-audit of e0ebdf0 found three rows not genuinely closed; each is now pinned by a test whose mutant was applied and verified to die before restoring: - C-2a: close-path unlink stop-on-first-failure was empirically unprotected -- a continue-past-first-failed-remove mutant in CursorSendEngine.unlinkAllSegmentFiles survived the whole suite (2,722 green). e0ebdf0 mapped the row to the periodic sync pass (different subsystem); the enumeration and all-unlinks-fail siblings cannot discriminate stop-vs-continue. New CursorSendEngineCloseUnlinkStopOnFirstFailureTest: legacy (manifest-less) slot, two FSN-contiguous sealed segments recovered from disk, FilesFacade refusing only the lowest .sfa removal once; asserts the higher generation is never attempted and survives, and a successor recovers the contiguous residue and completes the cleanup. Mutant now fails 'expected:<0> but was:<1>'. - C-2b: the sender-level checkDurability call sites (QwpWebSocketSender awaitAckedFsn pre-check, wait-loop, and flushAndGetSequence) had no user-visible-layer coverage -- deleting all three survived the whole suite; only the engine seam was pinned. New QwpWebSocketSenderCursorEngineAttachmentTest. testLatchedDurabilityFailureSurfacesThroughSenderFlushAndAwait: unconnected createForTesting sender + attached engine + latched failure; empty flush()/flushAndGetSequence() and awaitAckedFsn(fsn,0) must each throw the latched instance repeatedly (empty flush and <=0-timeout polls publish nothing, so the ring gate never runs -- these call sites are their only durability act, and they must fire before connection setup). Mutant now dies on the first flush(). - C-2c: testSecondBoundedJoinReapsWorkerFinishingExitCleanups passed under the exact pre-fix revert of 8d962e0: both variants null workerThread (isWorkerReaped() cannot discriminate) and the worker.join(5s) preceding the isAlive assert destroyed the liveness observable. Added the discriminating assertion at the moment close() returns: the second bounded join must hold close() until the parked cleanup finishes (+400ms), while the revert returns at ~200ms with the worker alive. Verified deterministic both ways: fails under the revert, passes with the fix. Full core suite: 2,724 green (2,722 + 2 new).
1 parent cce3091 commit 38cdf3e

3 files changed

Lines changed: 257 additions & 0 deletions

File tree

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import io.questdb.client.cutlass.line.LineSenderException;
2828
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
2929
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
30+
import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegmentException;
3031
import org.junit.Assert;
3132
import org.junit.Rule;
3233
import org.junit.Test;
@@ -126,6 +127,55 @@ public void testSameSharedEngineCannotTransferOwnership() throws Exception {
126127
});
127128
}
128129

130+
@Test
131+
public void testLatchedDurabilityFailureSurfacesThroughSenderFlushAndAwait() throws Exception {
132+
assertMemoryLeak(() -> {
133+
// The ring gate (SegmentRing.appendOrFsn) guards every publish
134+
// path, but an EMPTY flush()/flushAndGetSequence() and a
135+
// timeoutMillis <= 0 awaitAckedFsn() poll publish nothing: the
136+
// sender-level cursorEngine.checkDurability() call sites are
137+
// their ONLY durability act. Without them an empty flush
138+
// silently succeeds against an unsyncable slot and a poll loop
139+
// spins on "not yet" forever instead of surfacing the latched
140+
// barrier failure. Deleting any of the three sender call sites
141+
// previously survived the whole suite -- this pins the
142+
// user-visible layer (the engine seam has its own pin in
143+
// CursorSendEngineTest).
144+
CursorSendEngine engine = new CursorSendEngine(null, SEGMENT_SIZE);
145+
QwpWebSocketSender sender = QwpWebSocketSender.createForTesting("localhost", 1);
146+
try {
147+
sender.setCursorEngine(engine, false);
148+
MmapSegmentException failure = new MmapSegmentException("injected data-sync failure");
149+
engine.getRingForTesting().recordDurabilityFailureForTesting(failure);
150+
for (int i = 0; i < 2; i++) { // repeatable until cleared, not one-shot
151+
try {
152+
sender.flush();
153+
Assert.fail("empty flush() must surface the latched durability failure, call #" + i);
154+
} catch (MmapSegmentException expected) {
155+
Assert.assertSame("the latched instance itself must surface", failure, expected);
156+
}
157+
try {
158+
sender.flushAndGetSequence();
159+
Assert.fail("empty flushAndGetSequence() must surface the latched durability "
160+
+ "failure, call #" + i);
161+
} catch (MmapSegmentException expected) {
162+
Assert.assertSame(failure, expected);
163+
}
164+
try {
165+
sender.awaitAckedFsn(0L, 0L);
166+
Assert.fail("awaitAckedFsn(fsn, 0) must throw instead of polling 'not yet', "
167+
+ "call #" + i);
168+
} catch (MmapSegmentException expected) {
169+
Assert.assertSame(failure, expected);
170+
}
171+
}
172+
} finally {
173+
sender.close();
174+
engine.close();
175+
}
176+
});
177+
}
178+
129179
private static void assertSecondAttachmentRejected(
130180
QwpWebSocketSender sender,
131181
CursorSendEngine engine,

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

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
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.sf.cursor;
26+
27+
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
28+
import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment;
29+
import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentManager;
30+
import io.questdb.client.std.Files;
31+
import io.questdb.client.std.FilesFacade;
32+
import io.questdb.client.std.MemoryTag;
33+
import io.questdb.client.std.Unsafe;
34+
import io.questdb.client.test.tools.TestUtils;
35+
import org.junit.After;
36+
import org.junit.Assert;
37+
import org.junit.Before;
38+
import org.junit.Test;
39+
40+
import java.lang.reflect.InvocationTargetException;
41+
import java.lang.reflect.Proxy;
42+
import java.util.concurrent.atomic.AtomicBoolean;
43+
import java.util.concurrent.atomic.AtomicInteger;
44+
45+
/**
46+
* Close-time unlink STOP-ON-FIRST-FAILURE on a legacy (manifest-less) slot.
47+
* Removal runs in ascending generation order and must stop at the first
48+
* failed unlink so the residue is always a contiguous top slice that passes
49+
* FSN-contiguity at the next recovery. Continuing past a failed
50+
* low-generation remove deletes higher generations and can leave
51+
* non-contiguous residue -- a startup brick on a slot that lost nothing.
52+
* <p>
53+
* The siblings cover the OTHER close-cleanup contracts:
54+
* {@link CursorSendEngineClosePartialEnumerationTest} proves a torn
55+
* enumeration drives zero unlinks, and the unlink-failure sibling fails ALL
56+
* removals (permission trick), which cannot discriminate stop-vs-continue
57+
* (every file survives either way). A continue-past-first-failure mutant in
58+
* {@code unlinkAllSegmentFiles} previously survived the whole suite; this
59+
* test kills it via the higher-generation-survival assertions. Same
60+
* determinism trick as the siblings: the shared manager is never started, so
61+
* no worker touches the slot and no concurrent cleanup can trip the armed
62+
* fault.
63+
*/
64+
public class CursorSendEngineCloseUnlinkStopOnFirstFailureTest {
65+
66+
private static final long SEGMENT_SIZE = 4096L;
67+
68+
private String tmpDir;
69+
70+
@Before
71+
public void setUp() {
72+
tmpDir = TestUtils.createTmpDir("qdb-engine-close-unlink-stop-");
73+
}
74+
75+
@After
76+
public void tearDown() {
77+
TestUtils.removeTmpDir(tmpDir);
78+
}
79+
80+
@Test(timeout = 20_000L)
81+
public void testCloseUnlinkStopsAtFirstFailedRemoveOnLegacySlot() throws Exception {
82+
TestUtils.assertMemoryLeak(() -> {
83+
final String slot = tmpDir + "/legacy-slot";
84+
final String lowPath = slot + "/sf-initial.sfa";
85+
final String highPath = slot + "/sf-0000000000000002.sfa";
86+
final AtomicBoolean armLowestRemoveFailure = new AtomicBoolean();
87+
final AtomicInteger lowRemoveAttempts = new AtomicInteger();
88+
final AtomicInteger highRemoveAttempts = new AtomicInteger();
89+
FilesFacade faultFacade = (FilesFacade) Proxy.newProxyInstance(
90+
FilesFacade.class.getClassLoader(),
91+
new Class<?>[]{FilesFacade.class},
92+
(proxy, method, args) -> {
93+
if ("remove".equals(method.getName()) && args[0] instanceof String) {
94+
String path = (String) args[0];
95+
if (path.equals(lowPath)) {
96+
lowRemoveAttempts.incrementAndGet();
97+
if (armLowestRemoveFailure.get()) {
98+
// EBUSY-style transient refusal: no unlink happens.
99+
return false;
100+
}
101+
} else if (path.equals(highPath)) {
102+
highRemoveAttempts.incrementAndGet();
103+
}
104+
}
105+
try {
106+
return method.invoke(FilesFacade.INSTANCE, args);
107+
} catch (InvocationTargetException e) {
108+
throw e.getCause();
109+
}
110+
});
111+
112+
Assert.assertEquals(0, Files.mkdir(slot, Files.DIR_MODE_DEFAULT));
113+
// Legacy slot: two FSN-contiguous segments on disk, NO manifest.
114+
long buf = Unsafe.malloc(32, MemoryTag.NATIVE_DEFAULT);
115+
SegmentManager manager = null;
116+
try {
117+
MmapSegment low = MmapSegment.create(lowPath, 0L, SEGMENT_SIZE);
118+
Assert.assertTrue("setup: append must land", low.tryAppend(buf, 32) >= 0);
119+
Assert.assertTrue("setup: append must land", low.tryAppend(buf, 32) >= 0);
120+
low.close();
121+
MmapSegment high = MmapSegment.create(highPath, 2L, SEGMENT_SIZE);
122+
Assert.assertTrue("setup: append must land", high.tryAppend(buf, 32) >= 0);
123+
high.close();
124+
125+
manager = new SegmentManager(
126+
SEGMENT_SIZE,
127+
SegmentManager.DEFAULT_POLL_NANOS,
128+
SEGMENT_SIZE * 4L,
129+
faultFacade,
130+
System::nanoTime);
131+
132+
CursorSendEngine engine = new CursorSendEngine(slot, SEGMENT_SIZE, manager);
133+
boolean engineClosed = false;
134+
try {
135+
Assert.assertTrue("engine must recover the legacy two-segment chain",
136+
engine.wasRecoveredFromDisk());
137+
Assert.assertEquals("recovered chain must publish FSNs 0..2",
138+
2L, engine.publishedFsn());
139+
Assert.assertTrue(engine.acknowledge(2L));
140+
141+
// Fully drained close with the LOWEST-generation removal
142+
// refused once: ascending-order removal must STOP there.
143+
armLowestRemoveFailure.set(true);
144+
engine.close();
145+
engineClosed = true;
146+
Assert.assertTrue("close must complete despite the aborted cleanup",
147+
engine.isCloseCompleted());
148+
Assert.assertEquals("cleanup must attempt the lowest generation first",
149+
1, lowRemoveAttempts.get());
150+
Assert.assertEquals(
151+
"removal must STOP at the first failed unlink -- continuing would "
152+
+ "delete higher generations and could leave non-contiguous "
153+
+ "residue that fails FSN-contiguity at the next recovery",
154+
0, highRemoveAttempts.get());
155+
Assert.assertTrue("the refused lowest segment must survive",
156+
Files.exists(lowPath));
157+
Assert.assertTrue("higher-generation segment must survive the stopped cleanup",
158+
Files.exists(highPath));
159+
} finally {
160+
if (!engineClosed) {
161+
engine.close();
162+
}
163+
}
164+
165+
// Heal the fault; a successor adopts the contiguous residue
166+
// and its fully-drained close completes the cleanup.
167+
armLowestRemoveFailure.set(false);
168+
CursorSendEngine successor = new CursorSendEngine(slot, SEGMENT_SIZE, manager);
169+
boolean successorClosed = false;
170+
try {
171+
Assert.assertTrue("successor must recover the contiguous residue",
172+
successor.wasRecoveredFromDisk());
173+
successor.close();
174+
successorClosed = true;
175+
Assert.assertTrue(successor.isCloseCompleted());
176+
Assert.assertFalse("successor's fully-drained close must complete the unlink",
177+
Files.exists(lowPath));
178+
Assert.assertFalse("successor's fully-drained close must complete the unlink",
179+
Files.exists(highPath));
180+
} finally {
181+
if (!successorClosed) {
182+
successor.close();
183+
}
184+
}
185+
} finally {
186+
Unsafe.free(buf, 32, MemoryTag.NATIVE_DEFAULT);
187+
if (manager != null) {
188+
manager.close();
189+
}
190+
}
191+
});
192+
}
193+
}

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -766,6 +766,20 @@ public void testSecondBoundedJoinReapsWorkerFinishingExitCleanups() throws Excep
766766
manager.setWorkerJoinTimeoutMillis(200L);
767767
manager.close();
768768

769+
// The discriminating observable: with the second bounded join,
770+
// close() blocks until the parked cleanup finishes (released at
771+
// +400ms) and reaps a DEAD worker; the pre-fix code reaped on
772+
// observing workerLoopExited and returned at ~200ms with the
773+
// worker still alive in its cleanup. Assert liveness at the
774+
// moment close() returns -- before any join in this test can
775+
// mask it. (isWorkerReaped() alone cannot discriminate: both
776+
// variants null workerThread.)
777+
Assert.assertFalse(
778+
"close() must not return while the worker is still alive in its exit "
779+
+ "cleanups -- the second bounded join has to hold close() until "
780+
+ "they finish",
781+
worker.isAlive());
782+
769783
Assert.assertEquals("worker must have been parked in its exit cleanups",
770784
0, cleanupEntered.getCount());
771785
Assert.assertTrue(

0 commit comments

Comments
 (0)