Skip to content

Commit b9b6e2f

Browse files
bluestreak01claude
bluestreak01
and
claude
committed
feat(ilp): orphan-slot scanner + .failed sentinel + drain_orphans knob
Foundation for background-drainer adoption. The scan + visibility piece lands now; the drainer runtime that actually empties orphan slots is a follow-up. A "candidate orphan" is a sibling slot under sf_dir that: - isn't the foreground sender's own slot (filtered by sender_id) - contains at least one *.sfa segment file - doesn't have a .failed sentinel Lock state intentionally isn't part of the candidate filter — testing it requires opening + flocking the lock file, which races with concurrent acquirers. The drainer pool will attempt to lock each candidate in turn and skip ones that fail. The .failed sentinel is the "bounded automatic retry, then human-in-the- loop" hop in the spec: drainer gives up after its reconnect cap → drops .failed → exits. Future scans skip the slot until the operator clears the file. Knobs (WS-only, default off): drain_orphans=false — scan + log; future: spawn drainers max_background_drainers=4 — cap on concurrent drainers When drain_orphans=true today, the foreground sender's startup logs the count + first few orphan paths so operators have visibility while the drainer runtime is still pending. Tests: - OrphanScannerTest: empty group root, missing dir, candidate detection, empty-slot exclusion, .failed exclusion, sender_id exclusion, multiple candidates, isCandidateOrphan direct. - OrphanScanIntegrationTest: a "ghost" sender writes data with no ACKs and dies; a fresh sender with a different sender_id sees the ghost slot as an orphan via OrphanScanner.scan and its own slot is filtered out. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 40f9742 commit b9b6e2f

4 files changed

Lines changed: 662 additions & 0 deletions

File tree

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

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -699,6 +699,15 @@ public int getTimeout() {
699699
// be a hardcoded 30s constant; expose so tight-SLA users can lower
700700
// and offline-tolerant users can raise.
701701
private long sfAppendDeadlineMillis = PARAMETER_NOT_SET_EXPLICITLY;
702+
// Orphan adoption: when true, the foreground sender scans
703+
// <sf_dir>/*/ at startup for sibling slots that hold unacked data
704+
// and reports them. Default false. Spec calls for spawning
705+
// background drainers to actually empty those slots; the drainer
706+
// runtime lands in a follow-up commit. For now we surface the
707+
// count via logging so users can confirm orphans are being seen.
708+
private boolean drainOrphans = false;
709+
private int maxBackgroundDrainers = DEFAULT_MAX_BACKGROUND_DRAINERS;
710+
private static final int DEFAULT_MAX_BACKGROUND_DRAINERS = 4;
702711
private boolean tlsEnabled;
703712
private TlsValidationMode tlsValidationMode;
704713
private char[] trustStorePassword;
@@ -1064,6 +1073,25 @@ public Sender build() {
10641073
}
10651074
}
10661075
slotPath = sfDir + "/" + senderId;
1076+
// Orphan scan runs BEFORE we open our own slot — keeps
1077+
// the scan's "exclude my slot" filter conceptually
1078+
// simple. If the user opted in, log what we found so
1079+
// they have visibility on pending drain candidates
1080+
// until the drainer runtime lands.
1081+
if (drainOrphans) {
1082+
io.questdb.client.std.ObjList<String> orphans =
1083+
io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner
1084+
.scan(sfDir, senderId);
1085+
if (orphans.size() > 0) {
1086+
org.slf4j.LoggerFactory.getLogger(LineSenderBuilder.class)
1087+
.info("found {} orphan slot(s) under {} (drainer "
1088+
+ "runtime not yet implemented; "
1089+
+ "max_background_drainers={}); "
1090+
+ "first paths: {}",
1091+
orphans.size(), sfDir, maxBackgroundDrainers,
1092+
sample(orphans, 3));
1093+
}
1094+
}
10671095
}
10681096
long actualSfAppendDeadlineNanos =
10691097
sfAppendDeadlineMillis == PARAMETER_NOT_SET_EXPLICITLY
@@ -1725,6 +1753,19 @@ public LineSenderBuilder senderId(String id) {
17251753
return this;
17261754
}
17271755

1756+
private static String sample(io.questdb.client.std.ObjList<String> list, int n) {
1757+
int take = Math.min(n, list.size());
1758+
StringBuilder sb = new StringBuilder("[");
1759+
for (int i = 0; i < take; i++) {
1760+
if (i > 0) sb.append(", ");
1761+
sb.append(list.get(i));
1762+
}
1763+
if (list.size() > take) {
1764+
sb.append(", ...(").append(list.size() - take).append(" more)");
1765+
}
1766+
return sb.append("]").toString();
1767+
}
1768+
17281769
private static void validateSenderId(String id) {
17291770
if (id == null || id.isEmpty()) {
17301771
throw new LineSenderException("sender_id must not be empty");
@@ -1887,6 +1928,48 @@ public LineSenderBuilder sfAppendDeadlineMillis(long millis) {
18871928
return this;
18881929
}
18891930

1931+
/**
1932+
* Opt in to scanning {@code <sf_dir>/*} at startup for sibling slots
1933+
* that hold unacked data left behind by a crashed sender or a
1934+
* different sender_id. Default {@code false}. WebSocket only;
1935+
* requires {@code sf_dir} to be set.
1936+
* <p>
1937+
* The scan is read-only — slots flagged with the {@code .failed}
1938+
* sentinel are skipped (manual reset required), and the foreground
1939+
* sender's own slot is never reported.
1940+
* <p>
1941+
* <b>Status:</b> the scan + visibility (via logs) lands in this
1942+
* release; the background drainer runtime that actually empties
1943+
* orphan slots is a follow-up. Setting {@code drain_orphans=true}
1944+
* today logs the count and paths of orphans found at startup so
1945+
* users can monitor + manually drain pending slots.
1946+
*/
1947+
public LineSenderBuilder drainOrphans(boolean enabled) {
1948+
if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
1949+
throw new LineSenderException("drain_orphans is only supported for WebSocket transport");
1950+
}
1951+
this.drainOrphans = enabled;
1952+
return this;
1953+
}
1954+
1955+
/**
1956+
* Cap on concurrent background drainer threads when
1957+
* {@link #drainOrphans(boolean)} is on. Default {@code 4}. Each
1958+
* drainer carries one segment-manager thread + one I/O thread +
1959+
* one socket, so users running many senders per JVM should set
1960+
* this low.
1961+
*/
1962+
public LineSenderBuilder maxBackgroundDrainers(int n) {
1963+
if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
1964+
throw new LineSenderException("max_background_drainers is only supported for WebSocket transport");
1965+
}
1966+
if (n < 0) {
1967+
throw new LineSenderException("max_background_drainers must be >= 0: ").put(n);
1968+
}
1969+
this.maxBackgroundDrainers = n;
1970+
return this;
1971+
}
1972+
18901973
/**
18911974
* Selects the durability contract for SF appends and flushes. See
18921975
* {@link SfDurability} for the value semantics.
@@ -2457,6 +2540,24 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
24572540
}
24582541
pos = getValue(configurationString, pos, sink, "sf_append_deadline_millis");
24592542
sfAppendDeadlineMillis(parseLongValue(sink, "sf_append_deadline_millis"));
2543+
} else if (Chars.equals("drain_orphans", sink)) {
2544+
if (protocol != PROTOCOL_WEBSOCKET) {
2545+
throw new LineSenderException("drain_orphans is only supported for WebSocket transport");
2546+
}
2547+
pos = getValue(configurationString, pos, sink, "drain_orphans");
2548+
if (Chars.equalsIgnoreCase("on", sink) || Chars.equalsIgnoreCase("true", sink)) {
2549+
drainOrphans(true);
2550+
} else if (Chars.equalsIgnoreCase("off", sink) || Chars.equalsIgnoreCase("false", sink)) {
2551+
drainOrphans(false);
2552+
} else {
2553+
throw new LineSenderException("invalid drain_orphans [value=").put(sink).put(", allowed-values=[on, off, true, false]]");
2554+
}
2555+
} else if (Chars.equals("max_background_drainers", sink)) {
2556+
if (protocol != PROTOCOL_WEBSOCKET) {
2557+
throw new LineSenderException("max_background_drainers is only supported for WebSocket transport");
2558+
}
2559+
pos = getValue(configurationString, pos, sink, "max_background_drainers");
2560+
maxBackgroundDrainers(parseIntValue(sink, "max_background_drainers"));
24602561
} else if (Chars.equals("reconnect_max_backoff_millis", sink)) {
24612562
if (protocol != PROTOCOL_WEBSOCKET) {
24622563
throw new LineSenderException("reconnect_max_backoff_millis is only supported for WebSocket transport");

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

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
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.cutlass.qwp.client.sf.cursor;
26+
27+
import io.questdb.client.std.Files;
28+
import io.questdb.client.std.ObjList;
29+
30+
/**
31+
* Reads the SF group root and reports sibling slots that look like they
32+
* still hold unacked data — candidates for background-drainer adoption.
33+
* <p>
34+
* A slot is a "candidate orphan" iff:
35+
* <ul>
36+
* <li>It's a child directory of {@code sfDir}.</li>
37+
* <li>It's NOT the caller's own slot (filtered by name).</li>
38+
* <li>It contains at least one {@code *.sfa} segment file.</li>
39+
* <li>It does NOT contain a {@link #FAILED_SENTINEL_NAME} file —
40+
* that flag means a previous drainer gave up and the data needs
41+
* human attention before automation tries again.</li>
42+
* </ul>
43+
* <p>
44+
* Lock state is intentionally not part of the candidate filter — testing
45+
* it requires actually opening + flocking the lock file, which races
46+
* with concurrent drainers/senders. The drainer pool attempts to acquire
47+
* each candidate's lock in turn and skips ones that fail; this keeps the
48+
* scanner pure and read-only.
49+
* <p>
50+
* Empty slot dirs (no {@code .sfa} files but a stale {@code .lock} from
51+
* a clean shutdown) are NOT candidates — there's nothing to drain. Spec
52+
* decision #13 ("no automatic cleanup of empty slot dirs") leaves them
53+
* in place; scanning past them is fine.
54+
*/
55+
public final class OrphanScanner {
56+
57+
/** Name of the sentinel that disqualifies a slot from auto-drain. */
58+
public static final String FAILED_SENTINEL_NAME = ".failed";
59+
60+
private OrphanScanner() {
61+
}
62+
63+
/**
64+
* Walks {@code sfDir}'s children once and returns the candidate
65+
* orphan slot paths. {@code excludeSlotName} (typically the
66+
* foreground sender's {@code sender_id}) is filtered out so we
67+
* don't list our own slot as an orphan.
68+
* <p>
69+
* Returns an empty list if {@code sfDir} doesn't exist or is empty —
70+
* never throws on missing directory; the caller wants a clean
71+
* "no orphans" answer in that case.
72+
*/
73+
public static ObjList<String> scan(String sfDir, String excludeSlotName) {
74+
ObjList<String> orphans = new ObjList<>();
75+
if (sfDir == null || !Files.exists(sfDir)) {
76+
return orphans;
77+
}
78+
long find = Files.findFirst(sfDir);
79+
if (find == 0) {
80+
return orphans;
81+
}
82+
try {
83+
int rc = 1;
84+
while (rc > 0) {
85+
String name = Files.utf8ToString(Files.findName(find));
86+
rc = Files.findNext(find);
87+
if (name == null || ".".equals(name) || "..".equals(name)) {
88+
continue;
89+
}
90+
if (excludeSlotName != null && excludeSlotName.equals(name)) {
91+
continue;
92+
}
93+
String slotPath = sfDir + "/" + name;
94+
if (!isCandidateOrphan(slotPath)) {
95+
continue;
96+
}
97+
orphans.add(slotPath);
98+
}
99+
} finally {
100+
Files.findClose(find);
101+
}
102+
return orphans;
103+
}
104+
105+
/**
106+
* True iff {@code slotPath} looks like a slot dir with unacked data
107+
* and no failure sentinel. Visible for testing.
108+
*/
109+
public static boolean isCandidateOrphan(String slotPath) {
110+
if (!Files.exists(slotPath)) {
111+
return false;
112+
}
113+
if (Files.exists(slotPath + "/" + FAILED_SENTINEL_NAME)) {
114+
return false;
115+
}
116+
return hasAnySegmentFile(slotPath);
117+
}
118+
119+
/**
120+
* Drops a {@link #FAILED_SENTINEL_NAME} file in {@code slotPath}.
121+
* Idempotent — touching an existing sentinel is a no-op (its presence
122+
* is the signal; contents don't matter to scanning logic, though we
123+
* write a one-line reason for human readers).
124+
*/
125+
public static void markFailed(String slotPath, String reason) {
126+
String path = slotPath + "/" + FAILED_SENTINEL_NAME;
127+
int fd = Files.openRW(path);
128+
if (fd < 0) {
129+
// Best-effort — even if we can't write the sentinel, the
130+
// drainer is exiting anyway, and the next scan will retry.
131+
return;
132+
}
133+
try {
134+
byte[] payload = (reason == null ? "drainer failed" : reason)
135+
.getBytes(java.nio.charset.StandardCharsets.UTF_8);
136+
Files.truncate(fd, 0L);
137+
long addr = io.questdb.client.std.Unsafe.malloc(
138+
payload.length,
139+
io.questdb.client.std.MemoryTag.NATIVE_DEFAULT);
140+
try {
141+
for (int i = 0; i < payload.length; i++) {
142+
io.questdb.client.std.Unsafe.getUnsafe().putByte(addr + i, payload[i]);
143+
}
144+
Files.write(fd, addr, payload.length, 0L);
145+
} finally {
146+
io.questdb.client.std.Unsafe.free(
147+
addr, payload.length,
148+
io.questdb.client.std.MemoryTag.NATIVE_DEFAULT);
149+
}
150+
} finally {
151+
Files.close(fd);
152+
}
153+
}
154+
155+
private static boolean hasAnySegmentFile(String slotPath) {
156+
long find = Files.findFirst(slotPath);
157+
if (find == 0) {
158+
return false;
159+
}
160+
try {
161+
int rc = 1;
162+
while (rc > 0) {
163+
String name = Files.utf8ToString(Files.findName(find));
164+
rc = Files.findNext(find);
165+
if (name != null && name.endsWith(".sfa")) {
166+
return true;
167+
}
168+
}
169+
} finally {
170+
Files.findClose(find);
171+
}
172+
return false;
173+
}
174+
}

0 commit comments

Comments
 (0)