Skip to content

Commit e868354

Browse files
bluestreak01bluestreak01
committed
test(qwp): fuzz both retired-slot drain-liveness shapes
Extends the drain-liveness fuzz with a runtime-wedge action -- a borrow built against the silent sink, written through, then wedged-closed and discarded, so reclaimSlot retires it with its rows durably unacked -- and a second pinned iteration for the post-latch residual: the scan completes legitimately over clean dirs first, then the runtime wedge lands, and only the end-of-schedule heal may deliver the data back through the re-armed scan. Random iterations draw the runtime wedge freely so runtime retires also interleave with a still-running scan (that interleaving is what surfaced the mid-cycle release gap this change's sibling fix closes). The fault injector switches off at the end-of-schedule heal (faultsHealed): without it, a wedge[idx] flag whose first recovery build only happened DURING the quiescent convergence would forge a brand-new wedge after "every fault healed" -- one the schedule can never heal -- and the scan would correctly refuse to complete, failing the audit for the wrong reason. Discrimination matrix: both fixes green; residual fix removed reds at the pinned iteration 1; both fixes removed reds at the pinned iteration 0.
1 parent 94ddc60 commit e868354

1 file changed

Lines changed: 124 additions & 18 deletions

File tree

‎core/src/test/java/io/questdb/client/test/impl/SenderPoolSfFuzzTest.java‎

Lines changed: 124 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -88,17 +88,31 @@
8888
* The schedule is single-threaded on purpose: with one driver the whole
8989
* iteration is a pure function of the seed, so any failure replays exactly
9090
* with {@code TestUtils.generateRandom(null, s0, s1)} (seeds are printed by
91-
* the harness and repeated in the failure message). Iteration 0 is pinned to
92-
* the worst case -- every slot stranded, slot 0 wedged, and a deterministic
93-
* two-step prologue that drives the scan into the wedged-retire shape BEFORE
94-
* any random traffic (a random schedule could otherwise draw a borrow first,
95-
* adopt slot 0 and never materialize the wedge) -- so the suite cannot go
96-
* green on an unlucky seed while the bug is present; later iterations
97-
* randomize freely.
91+
* the harness and repeated in the failure message). Two iterations are pinned
92+
* so the suite cannot go green on an unlucky seed while either bug shape is
93+
* present; the rest randomize freely:
94+
* <ul>
95+
* <li><b>Iteration 0 -- scan abandonment.</b> Every slot stranded, slot 0
96+
* wedged, and a deterministic two-step prologue drives the scan into
97+
* the wedged-retire shape BEFORE any random traffic (a random schedule
98+
* could otherwise draw a borrow first, adopt slot 0 and never
99+
* materialize the wedge). Pre-fix, the scan skipped the retired index
100+
* without a deferral and latched {@code recoveryComplete} past the
101+
* stranded data.</li>
102+
* <li><b>Iteration 1 -- post-latch runtime retire (the residual).</b> The
103+
* scan completes LEGITIMATELY over clean dirs first; only then does a
104+
* borrowed sender -- built against the silent sink so its rows stay
105+
* durably unacked -- take the wedged-close discard path
106+
* ({@code reclaimSlot} retires it). Pre-fix, the late flock release
107+
* restored capacity but nothing re-armed the latched scan, so the
108+
* data waited for a restart or a lucky borrow of that index.</li>
109+
* </ul>
110+
* Random iterations (2+) additionally draw the runtime-wedge op freely, so
111+
* runtime retires also interleave with a still-running scan.
98112
*/
99113
public class SenderPoolSfFuzzTest {
100114

101-
private static final int ITERATIONS = 4;
115+
private static final int ITERATIONS = 5;
102116
private static final long CONVERGE_BUDGET_MILLIS = 20_000;
103117

104118
@Test
@@ -121,11 +135,13 @@ private void runOneIteration(Rnd rnd, int iter) throws Exception {
121135
try {
122136
TestUtils.assertMemoryLeak(() -> {
123137
int maxSize = 1 + rnd.nextInt(3); // 1..3
124-
// Iteration 0 pins the guaranteed-red shape: every slot
125-
// stranded so the wedged recovery build below has data behind
126-
// it. Later iterations may strand any subset (0 = a plain
127-
// clean-scan iteration, still a valid latch/borrow interplay).
128-
int stranded = iter == 0 ? maxSize : rnd.nextInt(maxSize + 1);
138+
// Iteration 0 pins the guaranteed-red scan-abandonment shape:
139+
// every slot stranded so the wedged recovery build below has
140+
// data behind it. Iteration 1 pins the residual and needs a
141+
// CLEAN start so the scan latches legitimately first. Later
142+
// iterations may strand any subset (0 = a plain clean-scan
143+
// iteration, still a valid latch/borrow interplay).
144+
int stranded = iter == 0 ? maxSize : iter == 1 ? 0 : rnd.nextInt(maxSize + 1);
129145

130146
// Phase 1: strand unacked data under default-0..(stranded-1).
131147
if (stranded > 0) {
@@ -168,6 +184,8 @@ private void runOneIteration(Rnd rnd, int iter) throws Exception {
168184
for (int i = 0; i < maxSize; i++) {
169185
wedge[i] = i < stranded && (iter == 0 ? i == 0 : rnd.nextBoolean());
170186
}
187+
// Iteration 1's fault is a RUNTIME wedge, not an in-scan one.
188+
boolean forceRuntimeWedge = iter == 1;
171189
CountingAckHandler handler = new CountingAckHandler();
172190
try (TestWebSocketServer ack = new TestWebSocketServer(handler);
173191
TestWebSocketServer wedgeSink = new TestWebSocketServer(new SilentHandler())) {
@@ -196,12 +214,27 @@ private void runOneIteration(Rnd rnd, int iter) throws Exception {
196214
+ ";close_flush_timeout_millis=0;";
197215

198216
boolean[] inRecoveryStep = new boolean[1];
217+
// Latched by the end-of-schedule heal: from that point the
218+
// fault injector is OFF. Without this, a wedge[idx] flag
219+
// whose first recovery build only happens DURING the
220+
// quiescent convergence would forge a brand-new wedge
221+
// after "every fault healed" -- a fault the schedule can
222+
// never heal -- and the scan would (correctly!) refuse to
223+
// complete, failing the audit for the wrong reason.
224+
boolean[] faultsHealed = new boolean[1];
225+
boolean[] wedgeNextBorrow = new boolean[1];
226+
Sender[] lastWedgeBuild = new Sender[1];
199227
Sender[] forged = new Sender[maxSize];
200228
List<Sender> unhealed = new ArrayList<>();
201229
IntFunction<Sender> factory = idx -> {
202-
boolean forgeNow = inRecoveryStep[0] && idx < maxSize
230+
boolean forgeNow = inRecoveryStep[0] && !faultsHealed[0] && idx < maxSize
203231
&& wedge[idx] && forged[idx] == null;
204-
Sender real = Sender.builder(forgeNow ? cfgWedge : cfg)
232+
// A runtime-wedge borrow is also built against the
233+
// silent sink (rows must stay durably unacked) but is
234+
// NOT forged closed at build: the op writes through it
235+
// first, then forges and discards.
236+
boolean wedgeBorrow = !inRecoveryStep[0] && wedgeNextBorrow[0];
237+
Sender real = Sender.builder(forgeNow || wedgeBorrow ? cfgWedge : cfg)
205238
.senderId("default-" + idx).build();
206239
if (forgeNow) {
207240
try {
@@ -211,6 +244,8 @@ private void runOneIteration(Rnd rnd, int iter) throws Exception {
211244
}
212245
forged[idx] = real;
213246
unhealed.add(real);
247+
} else if (wedgeBorrow) {
248+
lastWedgeBuild[0] = real;
214249
}
215250
return real;
216251
};
@@ -243,11 +278,49 @@ private void runOneIteration(Rnd rnd, int iter) throws Exception {
243278
Assert.assertTrue("iter 0 prologue must leave slot 0 stranded",
244279
hasSegmentFile(sfDir + "/default-0"));
245280
}
281+
if (forceRuntimeWedge) {
282+
// Pinned residual prologue: latch the scan
283+
// legitimately FIRST, then retire a live
284+
// borrow whose rows are durably unacked. No
285+
// heal until the end of the schedule, so the
286+
// post-heal re-arm (or its absence) is the
287+
// only thing that decides the audit.
288+
long latchDeadline = System.currentTimeMillis() + 10_000;
289+
while (!pool.isRecoveryCompleteForTesting()
290+
&& System.currentTimeMillis() < latchDeadline) {
291+
inRecoveryStep[0] = true;
292+
try {
293+
pool.runStartupRecoveryStepForTesting();
294+
} finally {
295+
inRecoveryStep[0] = false;
296+
}
297+
}
298+
Assert.assertTrue("iter 1 prologue: scan must latch over clean dirs",
299+
pool.isRecoveryCompleteForTesting());
300+
wedgeNextBorrow[0] = true;
301+
PooledSender ps = pool.borrow();
302+
wedgeNextBorrow[0] = false;
303+
Assert.assertSame("iter 1 prologue: borrow must create the wedge target",
304+
lastWedgeBuild[0], ps.getDelegateForTesting());
305+
lastWedgeBuild[0] = null;
306+
for (int r = 0; r < 2; r++) {
307+
ps.table("fuzz").longColumn("resid", r).atNow();
308+
ps.flush();
309+
}
310+
Sender delegate = ps.getDelegateForTesting();
311+
((QwpWebSocketSender) delegate).setClosedForTesting(true);
312+
pool.discardBrokenForTesting(ps);
313+
unhealed.add(delegate);
314+
Assert.assertEquals("iter 1 prologue must retire the runtime wedge",
315+
1, pool.leakedSlotCount());
316+
Assert.assertTrue("iter 1 prologue must leave its slot stranded",
317+
hasSegmentFile(sfDir + "/default-0"));
318+
}
246319
// The randomized schedule. Single-threaded, so the
247320
// iteration replays exactly from the seed.
248321
int ops = 8 + rnd.nextInt(12);
249322
for (int op = 0; op < ops; op++) {
250-
switch (rnd.nextInt(5)) {
323+
switch (rnd.nextInt(iter >= 2 ? 6 : 5)) {
251324
case 0:
252325
case 1: // bias toward driving the scan
253326
inRecoveryStep[0] = true;
@@ -291,17 +364,50 @@ private void runOneIteration(Rnd rnd, int iter) throws Exception {
291364
// held, no borrow can ever own slot 0, so
292365
// post-heal the SCAN is the only possible
293366
// deliverer (fixed) versus nobody (bug).
294-
if (iter != 0 && !unhealed.isEmpty()) {
367+
if (iter >= 2 && !unhealed.isEmpty()) {
295368
Sender s = unhealed.remove(rnd.nextInt(unhealed.size()));
296369
((QwpWebSocketSender) s).setClosedForTesting(false);
297370
s.close();
298371
}
299372
break;
373+
case 5: { // runtime wedge: retire a live borrow undelivered
374+
wedgeNextBorrow[0] = true;
375+
PooledSender ps = null;
376+
try {
377+
ps = pool.borrow();
378+
} catch (LineSenderException e) {
379+
// Capacity-starved: legal, move on.
380+
break;
381+
} finally {
382+
// Disarm regardless: a reused borrow
383+
// must not leave the flag armed for
384+
// an unrelated later creation.
385+
wedgeNextBorrow[0] = false;
386+
}
387+
Sender delegate = ps.getDelegateForTesting();
388+
if (delegate == lastWedgeBuild[0]) {
389+
lastWedgeBuild[0] = null;
390+
ps.table("fuzz").longColumn("wedged", op).atNow();
391+
ps.flush();
392+
((QwpWebSocketSender) delegate).setClosedForTesting(true);
393+
pool.discardBrokenForTesting(ps);
394+
unhealed.add(delegate);
395+
} else {
396+
// Borrow reused an idle (ack-server)
397+
// sender; nothing to wedge.
398+
lastWedgeBuild[0] = null;
399+
ps.close();
400+
}
401+
break;
402+
}
300403
}
301404
}
302405

303406
// Heal every remaining wedge: the "workers" exit and
304-
// the flocks genuinely drop.
407+
// the flocks genuinely drop. Also switch the fault
408+
// injector off -- no new wedge may be born after
409+
// this point (see faultsHealed).
410+
faultsHealed[0] = true;
305411
while (!unhealed.isEmpty()) {
306412
Sender s = unhealed.remove(unhealed.size() - 1);
307413
((QwpWebSocketSender) s).setClosedForTesting(false);

0 commit comments

Comments
 (0)