Skip to content

Commit a531164

Browse files
bluestreak01bluestreak01
committed
fix(qwp): harden SF crash recovery and cleanup
Make acknowledged segment deletion crash-consistent, bound retry and cleanup paths, improve pool recovery complexity, and add deterministic lifecycle/native regression coverage.
1 parent cad1bb8 commit a531164

30 files changed

Lines changed: 2464 additions & 360 deletions

‎core/src/main/c/share/files.c‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,21 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsync
124124
return res;
125125
}
126126

127+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsyncDir0
128+
(JNIEnv *e, jclass cl, jlong lpszName) {
129+
int fd;
130+
RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_RDONLY), fd);
131+
if (fd < 0) {
132+
return -1;
133+
}
134+
int res;
135+
RESTARTABLE(fsync(fd), res);
136+
int saved_errno = errno;
137+
close(fd);
138+
errno = saved_errno;
139+
return res;
140+
}
141+
127142
JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_truncate
128143
(JNIEnv *e, jclass cl, jint fd, jlong size) {
129144
int res;

‎core/src/main/c/windows/files.c‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -215,6 +215,25 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsync
215215
return 0;
216216
}
217217

218+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsyncDir0
219+
(JNIEnv *e, jclass cl, jlong lpszName) {
220+
jint fd = open_file((const char *) (uintptr_t) lpszName,
221+
GENERIC_READ | GENERIC_WRITE,
222+
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
223+
OPEN_EXISTING,
224+
FILE_FLAG_BACKUP_SEMANTICS);
225+
if (fd < 0) {
226+
return -1;
227+
}
228+
if (!FlushFileBuffers(FD_TO_HANDLE(fd))) {
229+
SaveLastError();
230+
CloseHandle(FD_TO_HANDLE(fd));
231+
return -1;
232+
}
233+
CloseHandle(FD_TO_HANDLE(fd));
234+
return 0;
235+
}
236+
218237
JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_truncate
219238
(JNIEnv *e, jclass cl, jint fd, jlong size) {
220239
FILE_END_OF_FILE_INFO eof;

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

Lines changed: 138 additions & 125 deletions
Original file line numberDiff line numberDiff line change
@@ -195,6 +195,8 @@ public class QwpWebSocketSender implements Sender {
195195
// close() drain timeout in millis. Default applied at construction.
196196
// 0 or -1 means "fast close" (skip the drain); otherwise close blocks
197197
// up to this many millis for ackedFsn to catch up to publishedFsn.
198+
private volatile boolean closeCleanupComplete;
199+
private boolean closeCleanupStarted;
198200
private long closeFlushTimeoutMillis = 5_000L;
199201
private volatile boolean closed;
200202
// Test-only lifecycle witness. close() invokes and clears it strictly after
@@ -1206,136 +1208,26 @@ public void close() {
12061208
}
12071209

12081210
if (!ioThreadStopped) {
1209-
// The I/O thread may still be using the socket and microbatch
1210-
// buffers. Freeing them would risk SIGSEGV.
1211-
LOG.error("I/O thread is still running, leaking WebSocket client and microbatch buffers");
1212-
// The engine, however, need not leak: delegate its close to
1213-
// the I/O thread's exit path, which runs it strictly after
1214-
// the thread's last engine access — the mapping and slot
1215-
// lock release as soon as the stuck wire call resolves
1216-
// (bounded by OS timeouts). slotLockReleased intentionally
1217-
// stays false: the lock is released only when the delegated
1218-
// close actually runs, so the pool must not reuse the slot
1219-
// meanwhile. A false return means the thread exited between
1220-
// the failed close() and now — then closing here is safe.
1221-
if (ownsCursorEngine && cursorEngine != null && cursorSendLoop != null) {
1222-
if (cursorSendLoop.delegateEngineClose()) {
1223-
// The I/O thread adopted the engine close and runs it on
1224-
// its exit path. Keep the engine visible for re-probing:
1225-
// isSlotLockReleased() flips true once that close
1226-
// completes, letting a pool recover the retired slot.
1227-
retainedEngine = cursorEngine;
1228-
} else {
1229-
CursorSendEngine engine = cursorEngine;
1230-
try {
1231-
engine.close();
1232-
} catch (Throwable t) {
1233-
LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
1234-
terminalError = captureCloseError(terminalError, t);
1235-
}
1236-
if (engine.isCloseCompleted()) {
1237-
cursorEngine = null;
1238-
ownsCursorEngine = false;
1239-
slotLockReleased = true;
1240-
} else {
1241-
// Engine cleanup is pending on its manager worker's
1242-
// exit path (or leaked, for a shared manager). Report
1243-
// the retained flock now; the getter re-probes the
1244-
// retained engine so a late release is not lost.
1245-
slotLockReleased = false;
1246-
retainedEngine = engine;
1247-
}
1248-
}
1211+
// The worker may still touch every resource below. Hand the
1212+
// complete sender-owned tail to its exit path rather than
1213+
// permanently leaking everything except the engine. The
1214+
// callback is idempotence-gated by closeRemainingResources().
1215+
if (ownsCursorEngine && cursorEngine != null) {
1216+
retainedEngine = cursorEngine;
12491217
}
1250-
rethrowTerminal(terminalError);
1251-
return;
1252-
}
1253-
1254-
if (buffer0 != null) {
1255-
try {
1256-
buffer0.close();
1257-
} catch (Throwable t) {
1258-
LOG.error("Error closing buffer0: {}", String.valueOf(t));
1259-
terminalError = captureCloseError(terminalError, t);
1260-
}
1261-
}
1262-
if (buffer1 != null) {
1263-
try {
1264-
buffer1.close();
1265-
} catch (Throwable t) {
1266-
LOG.error("Error closing buffer1: {}", String.valueOf(t));
1267-
terminalError = captureCloseError(terminalError, t);
1268-
}
1269-
}
1270-
1271-
if (client != null) {
1272-
try {
1273-
client.close();
1274-
} catch (Throwable t) {
1275-
LOG.error("Error closing WebSocket client: {}", String.valueOf(t));
1276-
terminalError = captureCloseError(terminalError, t);
1277-
}
1278-
client = null;
1279-
}
1280-
1281-
if (ownsCursorEngine && cursorEngine != null) {
1282-
CursorSendEngine engine = cursorEngine;
1283-
try {
1284-
engine.close();
1285-
} catch (Throwable t) {
1286-
LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
1287-
terminalError = captureCloseError(terminalError, t);
1288-
}
1289-
if (engine.isCloseCompleted()) {
1290-
cursorEngine = null;
1291-
ownsCursorEngine = false;
1292-
slotLockReleased = true;
1293-
} else {
1294-
// The manager worker did not quiesce. Preserve ownership
1295-
// and report the retained flock so pools retire this slot.
1296-
// Repeated Sender.close() calls remain no-ops by contract.
1297-
// Engine cleanup was handed to a safe manager-worker path:
1298-
// owned-manager exit or shared-manager ring-pass completion.
1299-
// The getter re-probes the retained engine so the pool can
1300-
// reclaim the slot once cleanup actually completes.
1301-
slotLockReleased = false;
1302-
retainedEngine = engine;
1218+
Runnable closeCallback = () -> closeRemainingResources(null);
1219+
if (cursorSendLoop != null && cursorSendLoop.delegateClose(closeCallback)) {
1220+
rethrowTerminal(terminalError);
1221+
return;
13031222
}
1223+
// The worker exited between close() failing and delegation.
1224+
// Cleanup is safe here and its failures remain suppressed on
1225+
// the original close error.
1226+
terminalError = closeRemainingResources(terminalError);
13041227
} else {
1305-
// This sender owns no cursor engine holding an SF flock.
1306-
slotLockReleased = true;
1228+
terminalError = closeRemainingResources(terminalError);
13071229
}
13081230

1309-
// Shutdown order: dispatcher last, after the I/O loop has stopped
1310-
// producing into it. close() drains pending entries with a short
1311-
// deadline so any final errors land in the user's handler.
1312-
if (errorDispatcher != null) {
1313-
try {
1314-
errorDispatcher.close();
1315-
} catch (Throwable t) {
1316-
LOG.error("Error closing error dispatcher: {}", String.valueOf(t));
1317-
terminalError = captureCloseError(terminalError, t);
1318-
}
1319-
}
1320-
if (progressDispatcher != null) {
1321-
try {
1322-
progressDispatcher.close();
1323-
} catch (Throwable t) {
1324-
LOG.error("Error closing progress dispatcher: {}", String.valueOf(t));
1325-
terminalError = captureCloseError(terminalError, t);
1326-
}
1327-
}
1328-
if (connectionDispatcher != null) {
1329-
try {
1330-
connectionDispatcher.close();
1331-
} catch (Throwable t) {
1332-
LOG.error("Error closing connection dispatcher: {}", String.valueOf(t));
1333-
terminalError = captureCloseError(terminalError, t);
1334-
}
1335-
}
1336-
1337-
LOG.info("QwpWebSocketSender closed");
1338-
13391231
// If close() ended up holding the same instance the user already
13401232
// caught earlier, suppress the rethrow. The user's catch block
13411233
// wraps close() (try-with-resources), and Throwable refuses
@@ -1347,6 +1239,11 @@ public void close() {
13471239
}
13481240
}
13491241

1242+
@TestOnly
1243+
public boolean isCloseCleanupComplete() {
1244+
return closeCleanupComplete;
1245+
}
1246+
13501247
/**
13511248
* True once the store-and-forward slot flock has been released. False
13521249
* means an I/O or manager worker did not stop and close() retained the
@@ -1808,6 +1705,21 @@ public long getDroppedConnectionNotifications() {
18081705
return d == null ? 0L : d.getDroppedNotifications();
18091706
}
18101707

1708+
@TestOnly
1709+
public SenderConnectionDispatcher getConnectionDispatcherForTesting() {
1710+
return connectionDispatcher;
1711+
}
1712+
1713+
@TestOnly
1714+
public SenderErrorDispatcher getErrorDispatcherForTesting() {
1715+
return errorDispatcher;
1716+
}
1717+
1718+
@TestOnly
1719+
public SenderProgressDispatcher getProgressDispatcherForTesting() {
1720+
return progressDispatcher;
1721+
}
1722+
18111723
/**
18121724
* Number of {@link SenderError} notifications dropped because the
18131725
* bounded inbox was full. Non-zero means the user-supplied
@@ -2234,6 +2146,11 @@ public void setClientFactoryOverride(java.util.function.Supplier<WebSocketClient
22342146
this.clientFactoryOverride = factory;
22352147
}
22362148

2149+
@TestOnly
2150+
public void setClientForTesting(WebSocketClient client) {
2151+
this.client = client;
2152+
}
2153+
22372154
/**
22382155
* Installs a one-shot test witness that close-time drain invokes after it
22392156
* observes a real unacknowledged target and before it starts waiting.
@@ -2344,6 +2261,25 @@ public void setCursorEngine(CursorSendEngine engine, boolean takeOwnership) {
23442261
}
23452262
}
23462263

2264+
@TestOnly
2265+
public void setCursorSendLoopForTesting(CursorWebSocketSendLoop loop) {
2266+
cursorSendLoop = loop;
2267+
if (connectionDispatcher == null) {
2268+
connectionDispatcher = new SenderConnectionDispatcher(
2269+
connectionListener, connectionListenerInboxCapacity);
2270+
}
2271+
if (errorDispatcher == null) {
2272+
errorDispatcher = new SenderErrorDispatcher(errorHandler, errorInboxCapacity);
2273+
}
2274+
if (progressDispatcher == null) {
2275+
progressDispatcher = new SenderProgressDispatcher(
2276+
progressHandler, SenderProgressDispatcher.DEFAULT_CAPACITY);
2277+
}
2278+
loop.setConnectionDispatcher(connectionDispatcher);
2279+
loop.setErrorDispatcher(errorDispatcher);
2280+
loop.setProgressDispatcher(progressDispatcher);
2281+
}
2282+
23472283
/**
23482284
* Register an async observer for background orphan-slot drainer events.
23492285
* May be called either before or after {@link #startOrphanDrainers} —
@@ -3211,6 +3147,83 @@ private void checkTableSelected() {
32113147
}
32123148
}
32133149

3150+
private synchronized Throwable closeRemainingResources(Throwable terminalError) {
3151+
if (closeCleanupStarted) {
3152+
return terminalError;
3153+
}
3154+
closeCleanupStarted = true;
3155+
try {
3156+
try {
3157+
buffer0.close();
3158+
} catch (Throwable t) {
3159+
LOG.error("Error closing buffer0: {}", String.valueOf(t));
3160+
terminalError = captureCloseError(terminalError, t);
3161+
}
3162+
try {
3163+
buffer1.close();
3164+
} catch (Throwable t) {
3165+
LOG.error("Error closing buffer1: {}", String.valueOf(t));
3166+
terminalError = captureCloseError(terminalError, t);
3167+
}
3168+
if (client != null) {
3169+
try {
3170+
client.close();
3171+
} catch (Throwable t) {
3172+
LOG.error("Error closing WebSocket client: {}", String.valueOf(t));
3173+
terminalError = captureCloseError(terminalError, t);
3174+
}
3175+
client = null;
3176+
}
3177+
if (ownsCursorEngine && cursorEngine != null) {
3178+
CursorSendEngine engine = cursorEngine;
3179+
try {
3180+
engine.close();
3181+
} catch (Throwable t) {
3182+
LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
3183+
terminalError = captureCloseError(terminalError, t);
3184+
}
3185+
if (engine.isCloseCompleted()) {
3186+
cursorEngine = null;
3187+
ownsCursorEngine = false;
3188+
slotLockReleased = true;
3189+
} else {
3190+
slotLockReleased = false;
3191+
retainedEngine = engine;
3192+
}
3193+
} else {
3194+
slotLockReleased = true;
3195+
}
3196+
if (errorDispatcher != null) {
3197+
try {
3198+
errorDispatcher.close();
3199+
} catch (Throwable t) {
3200+
LOG.error("Error closing error dispatcher: {}", String.valueOf(t));
3201+
terminalError = captureCloseError(terminalError, t);
3202+
}
3203+
}
3204+
if (progressDispatcher != null) {
3205+
try {
3206+
progressDispatcher.close();
3207+
} catch (Throwable t) {
3208+
LOG.error("Error closing progress dispatcher: {}", String.valueOf(t));
3209+
terminalError = captureCloseError(terminalError, t);
3210+
}
3211+
}
3212+
if (connectionDispatcher != null) {
3213+
try {
3214+
connectionDispatcher.close();
3215+
} catch (Throwable t) {
3216+
LOG.error("Error closing connection dispatcher: {}", String.valueOf(t));
3217+
terminalError = captureCloseError(terminalError, t);
3218+
}
3219+
}
3220+
LOG.info("QwpWebSocketSender closed");
3221+
return terminalError;
3222+
} finally {
3223+
closeCleanupComplete = true;
3224+
}
3225+
}
3226+
32143227
private int countNonEmptyTables(ObjList<CharSequence> keys) {
32153228
int tableCount = 0;
32163229
for (int i = 0, n = keys.size(); i < n; i++) {

0 commit comments

Comments
 (0)