Skip to content

Commit fe9546f

Browse files
glasstigerglasstiger
committed
Prevent orphan adoption during slot quarantine
1 parent 84293fa commit fe9546f

6 files changed

Lines changed: 362 additions & 104 deletions

File tree

‎core/src/main/java/io/questdb/client/Sender.java‎

Lines changed: 103 additions & 81 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
4141
import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner;
4242
import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
43+
import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLock;
4344
import io.questdb.client.cutlass.qwp.client.sf.cursor.UnreplayableSlotException;
4445
import io.questdb.client.impl.ConfStringParser;
4546
import io.questdb.client.impl.ConfigString;
@@ -1012,6 +1013,8 @@ final class LineSenderBuilder {
10121013
private static final int PROTOCOL_TCP = 0;
10131014
private static final int PROTOCOL_UDP = 3;
10141015
private static final int PROTOCOL_WEBSOCKET = 2;
1016+
@TestOnly
1017+
private static volatile Runnable quarantineAfterCloseHook;
10151018
// Suffix for a slot set aside by quarantineTornSlot. Deliberately NOT the
10161019
// sender's own slot name, so a restarted sender does not re-adopt it as its own;
10171020
// quarantineTornSlot then marks it .failed, so the orphan drainer skips it too and
@@ -1559,91 +1562,101 @@ public Sender build() {
15591562
sfAppendDeadlineMillis == PARAMETER_NOT_SET_EXPLICITLY
15601563
? CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS
15611564
: sfAppendDeadlineMillis * 1_000_000L;
1562-
CursorSendEngine cursorEngine = new CursorSendEngine(
1563-
slotPath, actualSfMaxBytes,
1564-
actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
1565-
int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
1566-
? errorInboxCapacity
1567-
: io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher.DEFAULT_CAPACITY;
1568-
int actualConnectionListenerInboxCapacity = connectionListenerInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
1569-
? connectionListenerInboxCapacity
1570-
: io.questdb.client.cutlass.qwp.client.sf.cursor.SenderConnectionDispatcher.DEFAULT_CAPACITY;
1571-
List<QwpWebSocketSender.Endpoint> wsEndpoints =
1572-
new ArrayList<>(hosts.size());
1573-
for (int i = 0, n = hosts.size(); i < n; i++) {
1574-
wsEndpoints.add(new QwpWebSocketSender.Endpoint(hosts.getQuick(i), ports.getQuick(i)));
1575-
}
1576-
// The recovery seed inside connect() is the authority on whether a recovered
1577-
// slot can be replayed: it rebuilds the dictionary from its intact prefix and
1578-
// then from the surviving frames' own delta sections, and throws
1579-
// UnreplayableSlotException only once neither source holds the missing ids.
1580-
// Quarantining on anything weaker would set aside slots that recovery can
1581-
// still rescue, so build() waits for that verdict rather than pre-judging it.
15821565
QwpWebSocketSender connected = null;
1583-
boolean quarantined = false;
1584-
while (connected == null) {
1585-
try {
1586-
connected = QwpWebSocketSender.connect(
1587-
wsEndpoints,
1588-
wsTlsConfig,
1589-
actualAutoFlushRows,
1590-
actualAutoFlushBytes,
1591-
actualAutoFlushIntervalNanos,
1592-
wsAuthHeader,
1593-
requestDurableAck,
1594-
cursorEngine,
1595-
actualCloseFlushTimeoutMillis,
1596-
actualReconnectMaxDurationMillis,
1597-
actualReconnectInitialBackoffMillis,
1598-
actualReconnectMaxBackoffMillis,
1599-
actualInitialConnectMode,
1600-
errorHandler,
1601-
actualErrorInboxCapacity,
1602-
actualDurableAckKeepaliveIntervalMillis,
1603-
authTimeoutMillis,
1604-
connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis,
1605-
connectionListener,
1606-
actualConnectionListenerInboxCapacity,
1607-
actualMaxFrameRejections,
1608-
actualPoisonMinEscalationWindowMillis,
1609-
actualCatchUpCapGapMinEscalationWindowMillis
1610-
);
1611-
} catch (UnreplayableSlotException e) {
1612-
// The one failure build() recovers from. The slot's frames reference ids
1613-
// that nothing still holds, so they can never go on the wire -- but that is
1614-
// no reason to take the producer down with them. Before this, the throw
1615-
// escaped build() and, because senderId is stable and a not-fully-drained
1616-
// slot is retained on close, every retry re-recovered the same slot and
1617-
// threw again: the application could not construct a Sender at all, so it
1618-
// could not even BUFFER new rows. An already-lost batch became an unbounded
1619-
// outage of everything after it.
1620-
//
1621-
// Set the slot aside instead, keep its bytes for forensics and resend, and
1622-
// start the producer on a clean one. Once only: a second such failure would
1623-
// mean the FRESH slot is unreplayable, which cannot happen, so let it out
1624-
// rather than loop.
1625-
if (quarantined || slotPath == null) {
1566+
// The parent-anchored logical lock is stable across a slot rename. Keep it
1567+
// from before the directory-local lock is acquired until connect() has either
1568+
// adopted that engine or quarantine has closed, renamed and recreated it.
1569+
// This closes the inode-swap window in which an already-queued orphan drainer
1570+
// could otherwise acquire the renamed directory's old .lock and later operate
1571+
// on the fresh slot through the original pathname.
1572+
try (SlotLock logicalSlotLock = slotPath == null
1573+
? null
1574+
: SlotLock.acquireLogical(slotPath)) {
1575+
CursorSendEngine cursorEngine = new CursorSendEngine(
1576+
slotPath, actualSfMaxBytes,
1577+
actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
1578+
int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
1579+
? errorInboxCapacity
1580+
: io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher.DEFAULT_CAPACITY;
1581+
int actualConnectionListenerInboxCapacity = connectionListenerInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
1582+
? connectionListenerInboxCapacity
1583+
: io.questdb.client.cutlass.qwp.client.sf.cursor.SenderConnectionDispatcher.DEFAULT_CAPACITY;
1584+
List<QwpWebSocketSender.Endpoint> wsEndpoints =
1585+
new ArrayList<>(hosts.size());
1586+
for (int i = 0, n = hosts.size(); i < n; i++) {
1587+
wsEndpoints.add(new QwpWebSocketSender.Endpoint(hosts.getQuick(i), ports.getQuick(i)));
1588+
}
1589+
// The recovery seed inside connect() is the authority on whether a recovered
1590+
// slot can be replayed: it rebuilds the dictionary from its intact prefix and
1591+
// then from the surviving frames' own delta sections, and throws
1592+
// UnreplayableSlotException only once neither source holds the missing ids.
1593+
// Quarantining on anything weaker would set aside slots that recovery can
1594+
// still rescue, so build() waits for that verdict rather than pre-judging it.
1595+
boolean quarantined = false;
1596+
while (connected == null) {
16261597
try {
1627-
cursorEngine.close();
1628-
} catch (Throwable ignored) {
1629-
// best-effort
1598+
connected = QwpWebSocketSender.connect(
1599+
wsEndpoints,
1600+
wsTlsConfig,
1601+
actualAutoFlushRows,
1602+
actualAutoFlushBytes,
1603+
actualAutoFlushIntervalNanos,
1604+
wsAuthHeader,
1605+
requestDurableAck,
1606+
cursorEngine,
1607+
actualCloseFlushTimeoutMillis,
1608+
actualReconnectMaxDurationMillis,
1609+
actualReconnectInitialBackoffMillis,
1610+
actualReconnectMaxBackoffMillis,
1611+
actualInitialConnectMode,
1612+
errorHandler,
1613+
actualErrorInboxCapacity,
1614+
actualDurableAckKeepaliveIntervalMillis,
1615+
authTimeoutMillis,
1616+
connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis,
1617+
connectionListener,
1618+
actualConnectionListenerInboxCapacity,
1619+
actualMaxFrameRejections,
1620+
actualPoisonMinEscalationWindowMillis,
1621+
actualCatchUpCapGapMinEscalationWindowMillis
1622+
);
1623+
} catch (UnreplayableSlotException e) {
1624+
// The one failure build() recovers from. The slot's frames reference ids
1625+
// that nothing still holds, so they can never go on the wire -- but that is
1626+
// no reason to take the producer down with them. Before this, the throw
1627+
// escaped build() and, because senderId is stable and a not-fully-drained
1628+
// slot is retained on close, every retry re-recovered the same slot and
1629+
// threw again: the application could not construct a Sender at all, so it
1630+
// could not even BUFFER new rows. An already-lost batch became an unbounded
1631+
// outage of everything after it.
1632+
//
1633+
// Set the slot aside instead, keep its bytes for forensics and resend, and
1634+
// start the producer on a clean one. Once only: a second such failure would
1635+
// mean the FRESH slot is unreplayable, which cannot happen, so let it out
1636+
// rather than loop.
1637+
if (quarantined || slotPath == null) {
1638+
try {
1639+
cursorEngine.close();
1640+
} catch (Throwable ignored) {
1641+
// best-effort
1642+
}
1643+
throw e;
1644+
}
1645+
quarantined = true;
1646+
cursorEngine = quarantineTornSlot(
1647+
cursorEngine, e, sfDir, senderId, slotPath, actualSfMaxBytes,
1648+
actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
1649+
} catch (Throwable t) {
1650+
// connect() failed before ownership of cursorEngine
1651+
// transferred — close it ourselves.
1652+
try {
1653+
cursorEngine.close();
1654+
} catch (Throwable ignored) {
1655+
// best-effort
1656+
}
1657+
throw t;
16301658
}
1631-
throw e;
16321659
}
1633-
quarantined = true;
1634-
cursorEngine = quarantineTornSlot(
1635-
cursorEngine, e, sfDir, senderId, slotPath, actualSfMaxBytes,
1636-
actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
1637-
} catch (Throwable t) {
1638-
// connect() failed before ownership of cursorEngine
1639-
// transferred — close it ourselves.
1640-
try {
1641-
cursorEngine.close();
1642-
} catch (Throwable ignored) {
1643-
// best-effort
1644-
}
1645-
throw t;
1646-
}
16471660
}
16481661
// connect() succeeded — `connected` now owns cursorEngine
16491662
// via setCursorEngine(engine, true). From here on, ANY
@@ -3047,6 +3060,10 @@ private static CursorSendEngine quarantineTornSlot(
30473060
// path already closed the engine; close() is idempotent, so make it explicit rather
30483061
// than depend on that.
30493062
torn.close();
3063+
Runnable hook = quarantineAfterCloseHook;
3064+
if (hook != null) {
3065+
hook.run();
3066+
}
30503067

30513068
String quarantinePath = null;
30523069
for (int i = 0; i < MAX_QUARANTINE_SLOT_ATTEMPTS; i++) {
@@ -3074,6 +3091,11 @@ private static CursorSendEngine quarantineTornSlot(
30743091
return new CursorSendEngine(slotPath, sfMaxBytes, sfMaxTotalBytes, sfAppendDeadlineNanos);
30753092
}
30763093

3094+
@TestOnly
3095+
public static void setQuarantineAfterCloseHookForTest(Runnable hook) {
3096+
quarantineAfterCloseHook = hook;
3097+
}
3098+
30773099
private static int resolveIPv4(String host) {
30783100
try {
30793101
byte[] addr = InetAddress.getByName(host).getAddress();

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

Lines changed: 45 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,9 @@
4444
* <p>
4545
* Lifecycle:
4646
* <ol>
47-
* <li>Acquire the slot's {@code .lock}; skip silently on contention.</li>
47+
* <li>Acquire the parent-anchored logical-slot lock and revalidate the
48+
* scanner snapshot; skip silently on contention or a stale snapshot.</li>
49+
* <li>Acquire the slot's {@code .lock}, then release the logical lock.</li>
4850
* <li>Open a {@link CursorSendEngine} on the slot — recovery picks up
4951
* every {@code .sfa} file already on disk.</li>
5052
* <li>Open a fresh {@link WebSocketClient} via the supplied factory
@@ -529,27 +531,54 @@ private boolean stopRequestedOrInterrupted() {
529531
@Override
530532
public void run() {
531533
runnerThread = Thread.currentThread();
534+
SlotLock logicalSlotLock = null;
532535
CursorSendEngine engine = null;
533536
WebSocketClient client = null;
534537
CursorWebSocketSendLoop loop = null;
535538
try {
536-
// The engine acquires the slot's .lock itself — we don't need
537-
// (and must not) double-lock it. If another sender or drainer
538-
// holds it, the engine constructor throws and we exit silently
539-
// (no .failed sentinel — contention is expected, not an error).
539+
// Scanner results are only snapshots. Serialize adoption against
540+
// a producer's close -> quarantine rename -> fresh-slot recreate
541+
// transition, then revalidate while that stable parent-anchored
542+
// lock is held. The slot's own .lock inode moves with a rename and
543+
// cannot provide this guarantee by itself.
544+
if (slotPath != null) {
545+
try {
546+
logicalSlotLock = SlotLock.acquireLogical(slotPath);
547+
} catch (IllegalStateException t) {
548+
if (isLockContention(t)) {
549+
LOG.info("orphan logical slot already locked, skipping: {} ({})",
550+
slotPath, t.getMessage());
551+
outcome = DrainOutcome.LOCKED_BY_OTHER;
552+
return;
553+
}
554+
throw t;
555+
}
556+
if (!OrphanScanner.isCandidateOrphan(slotPath)) {
557+
LOG.info("orphan candidate changed before adoption, skipping: {}", slotPath);
558+
outcome = DrainOutcome.SUCCESS;
559+
return;
560+
}
561+
}
562+
563+
// The engine acquires the directory-local .lock itself. Keep the
564+
// lock order logical -> local, and release the short-lived logical
565+
// lock only after the engine has secured stable ownership.
540566
try {
541567
engine = new CursorSendEngine(slotPath, segmentSizeBytes,
542568
sfMaxTotalBytes, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS);
543569
} catch (IllegalStateException t) {
544-
String msg = t.getMessage();
545-
if (msg != null && msg.contains("already in use")) {
570+
if (isLockContention(t)) {
546571
LOG.info("orphan slot already locked, skipping: {} ({})",
547-
slotPath, msg);
572+
slotPath, t.getMessage());
548573
outcome = DrainOutcome.LOCKED_BY_OTHER;
549574
return;
550575
}
551576
throw t;
552577
}
578+
if (logicalSlotLock != null) {
579+
logicalSlotLock.close();
580+
logicalSlotLock = null;
581+
}
553582
long target = engine.publishedFsn();
554583
if (engine.ackedFsn() >= target) {
555584
LOG.info("orphan slot already drained: {} (acked={} target={})",
@@ -724,12 +753,20 @@ public void run() {
724753
+ "slot lock releases when it exits", slotPath);
725754
}
726755
}
756+
if (logicalSlotLock != null) {
757+
logicalSlotLock.close();
758+
}
727759
// Don't let a later requestStop() unpark an unrelated task that
728760
// the pool's executor may have scheduled onto this same thread.
729761
runnerThread = null;
730762
}
731763
}
732764

765+
private static boolean isLockContention(IllegalStateException error) {
766+
String message = error.getMessage();
767+
return message != null && message.contains("already in use");
768+
}
769+
733770
/**
734771
* Plug an observer for durable-ack-related events. {@code null} clears
735772
* any previously installed listener. See {@link BackgroundDrainerListener}

0 commit comments

Comments
 (0)