Skip to content

Commit 1374448

Browse files
bluestreak01bluestreak01
committed
fix(qwp): route AckWatermark mmap/munmap through the injected FilesFacade
AckWatermark.open(FilesFacade, ...) routed exists/length/openRW/ allocate/close/msync/fsync through the facade but mapped the watermark with static Files.mmap and released it with static Files.munmap. Harmless in production -- the facade defaults delegate verbatim -- but inconsistent with MmapSegment, which routes mmap/munmap through the facade, and it made watermark-mmap fault injection impossible from a test facade: the injected override was silently bypassed on the one mapping that lives for the watermark's whole lifetime. Route both calls through the facade. The instance already stores the mapping facade in a final field used by sync() and releaseStorage(), so map and unmap always go through the same facade on every path, and the error-cleanup ordering (mmap failure closes the fd via the facade; release unmaps then closes exactly once) is unchanged. Widen open(FilesFacade, String) to public, matching the established facade-factory precedent (MmapSegment.create/openExisting, SegmentRing.openExisting) in the same exported package. This lets SegmentManagerCrashConsistencyTest and SegmentManagerManifestFsyncTest drop their reflective Method lookups for compile-time-checked calls. Adds AckWatermarkTest coverage with a delegating facade: mapping and release must be observed by the injected facade, and a facade that rejects mmap must fail open() with the fd closed and no mapping left. Both tests fail pre-fix (the static call bypassed the facade), pass post-fix.
1 parent bd7c3cb commit 1374448

4 files changed

Lines changed: 112 additions & 13 deletions

File tree

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

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -136,7 +136,12 @@ public static AckWatermark open(String slotDir) {
136136
return open(FilesFacade.INSTANCE, slotDir);
137137
}
138138

139-
static AckWatermark open(FilesFacade filesFacade, String slotDir) {
139+
/**
140+
* Facade-aware variant of {@link #open(String)}. Every filesystem call,
141+
* including the lifetime mapping, goes through {@code filesFacade} so
142+
* tests can observe or fault-inject the watermark mmap.
143+
*/
144+
public static AckWatermark open(FilesFacade filesFacade, String slotDir) {
140145
String filePath = slotDir + "/" + FILE_NAME;
141146
long existing = filesFacade.exists(filePath) ? filesFacade.length(filePath) : -1L;
142147
int fd;
@@ -156,7 +161,7 @@ static AckWatermark open(FilesFacade filesFacade, String slotDir) {
156161
LOG.warn("ack watermark {} could not be opened (rc={})", filePath, fd);
157162
return null;
158163
}
159-
long addr = Files.mmap(fd, FILE_SIZE, 0, Files.MAP_RW, MemoryTag.MMAP_DEFAULT);
164+
long addr = filesFacade.mmap(fd, FILE_SIZE, 0, Files.MAP_RW, MemoryTag.MMAP_DEFAULT);
160165
if (addr == Files.FAILED_MMAP_ADDRESS) {
161166
LOG.warn("ack watermark {} could not be mmapped", filePath);
162167
filesFacade.close(fd);
@@ -253,7 +258,7 @@ private boolean releaseStorage() {
253258
}
254259
isStorageReleased = true;
255260
if (mmapAddress != 0L && mmapAddress != Files.FAILED_MMAP_ADDRESS) {
256-
Files.munmap(mmapAddress, FILE_SIZE, MemoryTag.MMAP_DEFAULT);
261+
filesFacade.munmap(mmapAddress, FILE_SIZE, MemoryTag.MMAP_DEFAULT);
257262
}
258263
if (fd >= 0) {
259264
filesFacade.close(fd);

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

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626

2727
import io.questdb.client.cutlass.qwp.client.sf.cursor.AckWatermark;
2828
import io.questdb.client.std.Files;
29+
import io.questdb.client.std.FilesFacade;
2930
import io.questdb.client.std.MemoryTag;
3031
import io.questdb.client.std.Unsafe;
3132
import io.questdb.client.test.tools.TestUtils;
@@ -38,6 +39,7 @@
3839
import static org.junit.Assert.assertEquals;
3940
import static org.junit.Assert.assertFalse;
4041
import static org.junit.Assert.assertNotNull;
42+
import static org.junit.Assert.assertNull;
4143
import static org.junit.Assert.assertTrue;
4244

4345
public class AckWatermarkTest {
@@ -78,6 +80,22 @@ public void testCrossSessionPersistence() throws Exception {
7880
});
7981
}
8082

83+
@Test
84+
public void testFacadeMmapFaultFailsOpenAndClosesFd() throws Exception {
85+
// The watermark mapping must be reachable from an injected facade so
86+
// tests can fault-inject mmap. On a rejected mapping, open() must
87+
// fail cleanly and release the fd through the same facade.
88+
TestUtils.assertMemoryLeak(() -> {
89+
MappingFilesFacade ff = new MappingFilesFacade(true);
90+
assertNull("open must fail when the injected facade rejects mmap",
91+
AckWatermark.open(ff, slotDir));
92+
assertEquals("facade must receive the watermark mmap call", 1, ff.mmapCalls);
93+
assertEquals("failed open must close the watermark fd via the facade",
94+
1, ff.watermarkFdCloseCalls);
95+
assertEquals("a rejected mapping must not be munmapped", 0, ff.munmapCalls);
96+
});
97+
}
98+
8199
@Test
82100
public void testFallsBackFromTornPlausibleHighValue() throws Exception {
83101
TestUtils.assertMemoryLeak(() -> {
@@ -143,6 +161,24 @@ public void testNegativeFsnRoundTrips() throws Exception {
143161
});
144162
}
145163

164+
@Test
165+
public void testOpenAndCloseRouteMappingThroughFacade() throws Exception {
166+
// The lifetime mapping and its release must go through the injected
167+
// FilesFacade, matching MmapSegment, so test facades can observe them.
168+
TestUtils.assertMemoryLeak(() -> {
169+
MappingFilesFacade ff = new MappingFilesFacade(false);
170+
try (AckWatermark w = AckWatermark.open(ff, slotDir)) {
171+
assertNotNull(w);
172+
assertEquals("open must mmap through the injected facade", 1, ff.mmapCalls);
173+
w.write(7L);
174+
assertEquals(7L, w.read());
175+
}
176+
assertEquals("close must munmap through the injected facade", 1, ff.munmapCalls);
177+
assertEquals("close must release the watermark fd via the facade",
178+
1, ff.watermarkFdCloseCalls);
179+
});
180+
}
181+
146182
@Test
147183
public void testPhysicalReleaseIsIdempotentAcrossTestingSeamAndClose() throws Exception {
148184
TestUtils.assertMemoryLeak(() -> {
@@ -241,4 +277,68 @@ public void testWriteReadInSameSession() throws Exception {
241277
}
242278
});
243279
}
280+
281+
/**
282+
* Delegating facade that counts mmap/munmap traffic and watermark-fd
283+
* closes, optionally rejecting the mapping to exercise the open()
284+
* failure path.
285+
*/
286+
private static final class MappingFilesFacade implements FilesFacade {
287+
private final boolean failMmap;
288+
private int mmapCalls;
289+
private int munmapCalls;
290+
private int watermarkFd = -1;
291+
private int watermarkFdCloseCalls;
292+
293+
private MappingFilesFacade(boolean failMmap) {
294+
this.failMmap = failMmap;
295+
}
296+
297+
@Override public boolean allocate(int fd, long size) { return INSTANCE.allocate(fd, size); }
298+
@Override public long allocNativePath(String path) { return INSTANCE.allocNativePath(path); }
299+
@Override public int close(int fd) {
300+
if (fd >= 0 && fd == watermarkFd) watermarkFdCloseCalls++;
301+
return INSTANCE.close(fd);
302+
}
303+
@Override public boolean exists(String path) { return INSTANCE.exists(path); }
304+
@Override public void findClose(long findPtr) { INSTANCE.findClose(findPtr); }
305+
@Override public long findFirst(String dir) { return INSTANCE.findFirst(dir); }
306+
@Override public long findName(long findPtr) { return INSTANCE.findName(findPtr); }
307+
@Override public int findNext(long findPtr) { return INSTANCE.findNext(findPtr); }
308+
@Override public int findType(long findPtr) { return INSTANCE.findType(findPtr); }
309+
@Override public void freeNativePath(long pathPtr) { INSTANCE.freeNativePath(pathPtr); }
310+
@Override public int fsync(int fd) { return INSTANCE.fsync(fd); }
311+
@Override public long length(int fd) { return INSTANCE.length(fd); }
312+
@Override public long length(String path) { return INSTANCE.length(path); }
313+
@Override public long length(long pathPtr) { return INSTANCE.length(pathPtr); }
314+
@Override public int lock(int fd) { return INSTANCE.lock(fd); }
315+
@Override public int mkdir(String path, int mode) { return INSTANCE.mkdir(path, mode); }
316+
@Override public long mmap(int fd, long len, long offset, int flags, int memoryTag) {
317+
mmapCalls++;
318+
if (failMmap) return Files.FAILED_MMAP_ADDRESS;
319+
return INSTANCE.mmap(fd, len, offset, flags, memoryTag);
320+
}
321+
@Override public void munmap(long address, long len, int memoryTag) {
322+
munmapCalls++;
323+
INSTANCE.munmap(address, len, memoryTag);
324+
}
325+
@Override public int openCleanRW(String path) {
326+
int fd = INSTANCE.openCleanRW(path);
327+
if (path.endsWith(AckWatermark.FILE_NAME)) watermarkFd = fd;
328+
return fd;
329+
}
330+
@Override public int openCleanRW(long pathPtr) { return INSTANCE.openCleanRW(pathPtr); }
331+
@Override public int openRW(String path) {
332+
int fd = INSTANCE.openRW(path);
333+
if (path.endsWith(AckWatermark.FILE_NAME)) watermarkFd = fd;
334+
return fd;
335+
}
336+
@Override public int openRW(long pathPtr) { return INSTANCE.openRW(pathPtr); }
337+
@Override public long read(int fd, long addr, long len, long offset) { return INSTANCE.read(fd, addr, len, offset); }
338+
@Override public boolean remove(String path) { return INSTANCE.remove(path); }
339+
@Override public boolean remove(long pathPtr) { return INSTANCE.remove(pathPtr); }
340+
@Override public int rename(String oldPath, String newPath) { return INSTANCE.rename(oldPath, newPath); }
341+
@Override public boolean truncate(int fd, long size) { return INSTANCE.truncate(fd, size); }
342+
@Override public long write(int fd, long addr, long len, long offset) { return INSTANCE.write(fd, addr, len, offset); }
343+
}
244344
}

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

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
import org.junit.Assert;
2626
import org.junit.Test;
2727

28-
import java.lang.reflect.Method;
2928
import java.nio.file.Paths;
3029
import java.util.ArrayList;
3130
import java.util.Arrays;
@@ -77,10 +76,8 @@ private static SegmentRing createRing(String root, long segmentSize, long payloa
7776
}
7877
}
7978

80-
private static AckWatermark openWatermark(FilesFacade ff, String root) throws Exception {
81-
Method method = AckWatermark.class.getDeclaredMethod("open", FilesFacade.class, String.class);
82-
method.setAccessible(true);
83-
return (AckWatermark) method.invoke(null, ff, root);
79+
private static AckWatermark openWatermark(FilesFacade ff, String root) {
80+
return AckWatermark.open(ff, root);
8481
}
8582

8683
private static void removeRecursive(String dir) {

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

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,6 @@
3737
import org.junit.Before;
3838
import org.junit.Test;
3939

40-
import java.lang.reflect.Method;
4140
import java.util.concurrent.TimeUnit;
4241
import java.util.concurrent.atomic.AtomicInteger;
4342

@@ -110,10 +109,8 @@ private static void awaitTrimmed(SegmentRing ring) {
110109
}
111110
}
112111

113-
private static AckWatermark openWatermark(FilesFacade ff, String root) throws Exception {
114-
Method method = AckWatermark.class.getDeclaredMethod("open", FilesFacade.class, String.class);
115-
method.setAccessible(true);
116-
return (AckWatermark) method.invoke(null, ff, root);
112+
private static AckWatermark openWatermark(FilesFacade ff, String root) {
113+
return AckWatermark.open(ff, root);
117114
}
118115

119116
private static void writeSegmentWithFrames(String path, long baseSeq, int frames) {

0 commit comments

Comments
 (0)