Skip to content

Commit 12a3aec

Browse files
glasstigerglasstiger
committed
Fix QWP store-and-forward recovery safety
Retain source rows when split preflight rejects an oversized table. Allow full dictionary frames to re-anchor ACKed recovery gaps while preserving unacked-gap safety. Bump the segment format and install a legacy-reader rollback barrier.
1 parent 78a1d32 commit 12a3aec

13 files changed

Lines changed: 283 additions & 26 deletions

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3664,7 +3664,6 @@ private void flushPendingRowsSplit(
36643664
splitFrameBodyBytes.getQuick(bodyIdx), simBaseline, currentBatchMaxSymbolId);
36653665
bodyIdx++;
36663666
if (messageSize > cap) {
3667-
resetTableBuffersAfterFlush(keys);
36683667
throw new LineSenderException("single table batch too large for server batch cap")
36693668
.put(" [table=").put(tableName)
36703669
.put(", messageSize=").put(messageSize)

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -254,6 +254,14 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
254254
PersistedSymbolDict persistedDictInProgress = null;
255255
RecoveredFrameAnalysis recoveredFrameAnalysisInProgress = null;
256256
try {
257+
// v2 segment payloads may depend on a persisted symbol-dictionary
258+
// prefix. Install the rollback barrier before recovery or any new
259+
// append so a v1-only client can never skip the v2 files, treat the
260+
// slot as empty, and silently restart at FSN 0. Current recovery
261+
// recognizes and skips the reserved guard filenames.
262+
if (!memoryMode) {
263+
SegmentRing.installLegacyReaderBarrier(sfDir);
264+
}
257265
// Disk mode: try to recover any *.sfa files left behind by a prior
258266
// session before deciding to start fresh. Without this the engine
259267
// would create a new sf-initial.sfa at baseSeq=0, overlapping FSNs

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

Lines changed: 63 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@
4646
* <p>
4747
* On-disk layout — header and frame format:
4848
* <pre>
49-
* [u32 magic 'SF01'] [u8 ver=1] [u8 flags=0] [u16 reserved=0]
49+
* [u32 magic 'SF01'] [u8 ver] [u8 flags=0] [u16 reserved=0]
5050
* [u64 baseSeq] [u64 createdMicros] 24-byte header
5151
* frame, frame, ... each frame:
5252
* [u32 crc32c]
@@ -65,7 +65,8 @@ public final class MmapSegment implements QuietCloseable {
6565
public static final int FILE_MAGIC = 0x31304653; // 'SF01' little-endian
6666
public static final int FRAME_HEADER_SIZE = 8; // u32 crc + u32 payloadLen
6767
public static final int HEADER_SIZE = 24;
68-
public static final byte VERSION = 1;
68+
public static final byte LEGACY_VERSION = 1;
69+
public static final byte VERSION = 2;
6970
private static final int[] CRC32C_TABLE = buildCrc32cTable();
7071
private static final Logger LOG = LoggerFactory.getLogger(MmapSegment.class);
7172

@@ -76,6 +77,7 @@ public final class MmapSegment implements QuietCloseable {
7677
// memory-backed segments — same cursor architecture, no disk involvement.
7778
// close() and msync() branch on this flag.
7879
private final boolean memoryBacked;
80+
private final byte version;
7981
// appendCursor: written only by the producer thread, never read by anyone else
8082
// — it's the reservation cursor. Plain field is fine.
8183
private long appendCursor;
@@ -106,7 +108,7 @@ public final class MmapSegment implements QuietCloseable {
106108

107109
private MmapSegment(String path, int fd, long mmapAddress, long sizeBytes,
108110
long baseSeq, long initialCursor, long frameCount,
109-
boolean memoryBacked, long tornTailBytes) {
111+
boolean memoryBacked, long tornTailBytes, byte version) {
110112
this.path = path;
111113
this.fd = fd;
112114
this.mmapAddress = mmapAddress;
@@ -117,6 +119,7 @@ private MmapSegment(String path, int fd, long mmapAddress, long sizeBytes,
117119
this.frameCount = frameCount;
118120
this.memoryBacked = memoryBacked;
119121
this.tornTailBytes = tornTailBytes;
122+
this.version = version;
120123
}
121124

122125
/**
@@ -164,6 +167,50 @@ public static MmapSegment create(FilesFacade ff, String path, long baseSeq, long
164167
* per-call {@code byte[]} + native-malloc the way the String overload does.
165168
*/
166169
public static MmapSegment create(FilesFacade ff, long pathPtr, String displayPath, long baseSeq, long sizeBytes) {
170+
return create(ff, pathPtr, displayPath, baseSeq, sizeBytes, VERSION);
171+
}
172+
173+
/**
174+
* Creates a tiny v1 segment containing one deliberately non-QWP payload.
175+
* Two such files with non-contiguous base sequences form the rollback
176+
* barrier installed by {@link SegmentRing}: a v1-only reader opens both
177+
* and fails its mandatory FSN-contiguity check before replay. If a process
178+
* dies after creating only one guard, that reader still adopts a non-empty
179+
* log and sends the invalid one-byte head frame first, so it cannot silently
180+
* discard the v2 files and start a fresh slot.
181+
*/
182+
static void createLegacyReaderGuard(String path, long baseSeq) {
183+
long pathPtr = FilesFacade.INSTANCE.allocNativePath(path);
184+
long payload = 0L;
185+
try (MmapSegment segment = create(
186+
FilesFacade.INSTANCE,
187+
pathPtr,
188+
path,
189+
baseSeq,
190+
HEADER_SIZE + FRAME_HEADER_SIZE + 1L,
191+
LEGACY_VERSION
192+
)) {
193+
payload = Unsafe.malloc(1, MemoryTag.NATIVE_DEFAULT);
194+
Unsafe.getUnsafe().putByte(payload, (byte) 0);
195+
if (segment.tryAppend(payload, 1) != HEADER_SIZE) {
196+
throw new MmapSegmentException("could not append legacy-reader guard frame: " + path);
197+
}
198+
} finally {
199+
if (payload != 0L) {
200+
Unsafe.free(payload, 1, MemoryTag.NATIVE_DEFAULT);
201+
}
202+
FilesFacade.INSTANCE.freeNativePath(pathPtr);
203+
}
204+
}
205+
206+
private static MmapSegment create(
207+
FilesFacade ff,
208+
long pathPtr,
209+
String displayPath,
210+
long baseSeq,
211+
long sizeBytes,
212+
byte version
213+
) {
167214
if (sizeBytes < HEADER_SIZE + FRAME_HEADER_SIZE + 1) {
168215
throw new IllegalArgumentException(
169216
"sizeBytes too small for header + one minimal frame: " + sizeBytes);
@@ -194,12 +241,13 @@ public static MmapSegment create(FilesFacade ff, long pathPtr, String displayPat
194241
}
195242
// Header goes straight into the mapping — no separate write syscall.
196243
Unsafe.getUnsafe().putInt(addr, FILE_MAGIC);
197-
Unsafe.getUnsafe().putByte(addr + 4, VERSION);
244+
Unsafe.getUnsafe().putByte(addr + 4, version);
198245
Unsafe.getUnsafe().putByte(addr + 5, (byte) 0); // flags
199246
Unsafe.getUnsafe().putShort(addr + 6, (short) 0); // reserved
200247
Unsafe.getUnsafe().putLong(addr + 8, baseSeq);
201248
Unsafe.getUnsafe().putLong(addr + 16, Os.currentTimeMicros());
202-
return new MmapSegment(displayPath, fd, addr, sizeBytes, baseSeq, HEADER_SIZE, 0, false, 0L);
249+
return new MmapSegment(displayPath, fd, addr, sizeBytes, baseSeq,
250+
HEADER_SIZE, 0, false, 0L, version);
203251
} catch (Throwable t) {
204252
if (addr != Files.FAILED_MMAP_ADDRESS) {
205253
Files.munmap(addr, sizeBytes, MemoryTag.MMAP_DEFAULT);
@@ -236,7 +284,8 @@ public static MmapSegment createInMemory(long baseSeq, long sizeBytes) {
236284
Unsafe.getUnsafe().putShort(addr + 6, (short) 0);
237285
Unsafe.getUnsafe().putLong(addr + 8, baseSeq);
238286
Unsafe.getUnsafe().putLong(addr + 16, Os.currentTimeMicros());
239-
return new MmapSegment(null, -1, addr, sizeBytes, baseSeq, HEADER_SIZE, 0, true, 0L);
287+
return new MmapSegment(null, -1, addr, sizeBytes, baseSeq,
288+
HEADER_SIZE, 0, true, 0L, VERSION);
240289
} catch (Throwable t) {
241290
Unsafe.free(addr, sizeBytes, MemoryTag.NATIVE_DEFAULT);
242291
throw t;
@@ -300,7 +349,7 @@ public static MmapSegment openExisting(FilesFacade ff, String path) {
300349
"bad magic in " + path + ": 0x" + Integer.toHexString(magic));
301350
}
302351
byte version = Unsafe.getUnsafe().getByte(addr + 4);
303-
if (version != VERSION) {
352+
if (version != LEGACY_VERSION && version != VERSION) {
304353
throw new MmapSegmentException("unsupported version in " + path + ": " + version);
305354
}
306355
long baseSeq = Unsafe.getUnsafe().getLong(addr + 8);
@@ -326,7 +375,8 @@ public static MmapSegment openExisting(FilesFacade ff, String path) {
326375
+ "Investigate disk health or unexpected writer crash.",
327376
path, tornTail, lastGood, fileSize, count);
328377
}
329-
return new MmapSegment(path, fd, addr, fileSize, baseSeq, lastGood, count, false, tornTail);
378+
return new MmapSegment(path, fd, addr, fileSize, baseSeq,
379+
lastGood, count, false, tornTail, version);
330380
} catch (Throwable t) {
331381
if (addr != Files.FAILED_MMAP_ADDRESS) {
332382
Files.munmap(addr, fileSize, MemoryTag.MMAP_DEFAULT);
@@ -440,6 +490,11 @@ public long sizeBytes() {
440490
return sizeBytes;
441491
}
442492

493+
/** On-disk format version read from or written to this segment. */
494+
public byte version() {
495+
return version;
496+
}
497+
443498
/**
444499
* Appends one frame: writes {@code [crc32c | u32 payloadLen | payload]}
445500
* starting at the current append cursor, then advances both cursors

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

Lines changed: 30 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
final class RecoveredFrameAnalysis implements QuietCloseable {
4242

4343
private static final int MAX_RAW_BYTES = Integer.MAX_VALUE - 8;
44+
private final long ackedFsn;
4445
private final int baseline;
4546
private long committedBoundaryFsn = -1L;
4647
private long committedCoverage;
@@ -51,6 +52,7 @@ final class RecoveredFrameAnalysis implements QuietCloseable {
5152
private int committedRawLen;
5253
private long framesVisited;
5354
private boolean runningGap;
55+
private boolean runningUnackedGap;
5456
private long runningMaxDeltaEnd;
5557
private long runningMaxDeltaStart;
5658
private long rawAddr;
@@ -59,8 +61,9 @@ final class RecoveredFrameAnalysis implements QuietCloseable {
5961
private int runningRawLen;
6062
private long runningCoverage;
6163

62-
RecoveredFrameAnalysis(int baseline) {
64+
RecoveredFrameAnalysis(int baseline, long ackedFsn) {
6365
this.baseline = baseline;
66+
this.ackedFsn = ackedFsn;
6467
this.runningCoverage = baseline;
6568
this.committedCoverage = baseline;
6669
}
@@ -74,7 +77,7 @@ void accept(long fsn, long payload, int payloadLen) {
7477
? Unsafe.getUnsafe().getByte(payload + QwpConstants.HEADER_OFFSET_FLAGS)
7578
: 0;
7679
if (isQwp && (flags & QwpConstants.FLAG_DELTA_SYMBOL_DICT) != 0) {
77-
foldDelta(payload + QwpConstants.HEADER_SIZE, payload + payloadLen);
80+
foldDelta(fsn, payload + QwpConstants.HEADER_SIZE, payload + payloadLen);
7881
}
7982

8083
// Only a positively identified deferred QWP frame can belong to an
@@ -179,10 +182,10 @@ private void appendRaw(long entryStart, int entryLen) {
179182
runningRawCount++;
180183
}
181184

182-
private void foldDelta(long p, long limit) {
185+
private void foldDelta(long fsn, long p, long limit) {
183186
long encodedStart = readVarint(p, limit);
184187
if (encodedStart < 0L) {
185-
runningGap = true;
188+
markGap(fsn);
186189
return;
187190
}
188191
int startLen = (int) (encodedStart & 7L);
@@ -194,7 +197,7 @@ private void foldDelta(long p, long limit) {
194197

195198
long encodedCount = readVarint(p, limit);
196199
if (encodedCount < 0L) {
197-
runningGap = true;
200+
markGap(fsn);
198201
return;
199202
}
200203
int countLen = (int) (encodedCount & 7L);
@@ -207,10 +210,21 @@ private void foldDelta(long p, long limit) {
207210
runningMaxDeltaEnd = deltaEnd;
208211
}
209212
if (runningGap) {
210-
return;
213+
// A full dictionary is a new self-sufficient epoch, but it may only
214+
// repair a gap that is entirely behind the durable ACK watermark.
215+
// If any gapped frame will replay first, accepting this reset would
216+
// hide the unsafe wire-order gap and let the server observe missing
217+
// ids before it reaches the full frame.
218+
if (deltaStart != 0L || runningUnackedGap) {
219+
return;
220+
}
221+
runningGap = false;
222+
runningCoverage = baseline;
223+
runningRawLen = 0;
224+
runningRawCount = 0;
211225
}
212226
if (deltaStart > runningCoverage) {
213-
runningGap = true;
227+
markGap(fsn);
214228
return;
215229
}
216230

@@ -219,14 +233,14 @@ private void foldDelta(long p, long limit) {
219233
long entryStart = p;
220234
long encodedLen = readVarint(p, limit);
221235
if (encodedLen < 0L) {
222-
runningGap = true;
236+
markGap(fsn);
223237
return;
224238
}
225239
int varintLen = (int) (encodedLen & 7L);
226240
long symbolLen = encodedLen >>> 3;
227241
p += varintLen;
228242
if (symbolLen > limit - p) {
229-
runningGap = true;
243+
markGap(fsn);
230244
return;
231245
}
232246
p += symbolLen;
@@ -239,6 +253,13 @@ private void foldDelta(long p, long limit) {
239253
}
240254
}
241255

256+
private void markGap(long fsn) {
257+
runningGap = true;
258+
if (fsn > ackedFsn) {
259+
runningUnackedGap = true;
260+
}
261+
}
262+
242263
/**
243264
* Returns {@code (value << 3) | encodedByteCount}, or {@code -1} for an
244265
* unterminated/oversized 32-bit protocol varint.

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

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,8 @@ public final class SegmentRing implements QuietCloseable {
6262
public static final long BACKPRESSURE_NO_SPARE = -1L;
6363
/** Sentinel: append failed because the payload doesn't fit in a fresh segment. */
6464
public static final long PAYLOAD_TOO_LARGE = -2L;
65+
static final String LEGACY_READER_GUARD_A = ".qwp-v2-guard-a.sfa";
66+
static final String LEGACY_READER_GUARD_B = ".qwp-v2-guard-b.sfa";
6567
private static final Logger LOG = LoggerFactory.getLogger(SegmentRing.class);
6668
// Tally of baseSeq comparisons performed by sortByBaseSeq across every
6769
// openExisting() recovery on this JVM. Used by SegmentRingTest to
@@ -178,7 +180,9 @@ public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) {
178180
int rc = 1;
179181
while (rc > 0) {
180182
String name = Files.utf8ToString(Files.findName(find));
181-
if (name != null && name.endsWith(".sfa")) {
183+
if (name != null
184+
&& name.endsWith(".sfa")
185+
&& !isLegacyReaderGuard(name)) {
182186
String path = sfDir + "/" + name;
183187
MmapSegment seg = null;
184188
try {
@@ -325,6 +329,23 @@ public long ackedFsn() {
325329
return ackedFsn;
326330
}
327331

332+
/**
333+
* Installs an on-disk rollback barrier before any v2 segment can be
334+
* created or appended. The guards are valid v1 segments so a legacy
335+
* reader cannot skip them as an unknown format. Their deliberately
336+
* non-contiguous FSN ranges make its recovery fail closed before replay;
337+
* current readers recognize the reserved names and omit them from the
338+
* logical ring.
339+
*/
340+
static void installLegacyReaderBarrier(String sfDir) {
341+
MmapSegment.createLegacyReaderGuard(
342+
sfDir + '/' + LEGACY_READER_GUARD_A,
343+
Long.MAX_VALUE - 4L);
344+
MmapSegment.createLegacyReaderGuard(
345+
sfDir + '/' + LEGACY_READER_GUARD_B,
346+
Long.MAX_VALUE - 1L);
347+
}
348+
328349
/**
329350
* I/O thread (or anyone tracking ACK) advances the ACK cursor. {@code seq}
330351
* is cumulative -- the server has confirmed every FSN up to and including
@@ -586,7 +607,7 @@ public synchronized long collectReplaySymbolsAbove(
586607
* active segment. The returned native suffix remains owned by the caller.
587608
*/
588609
synchronized RecoveredFrameAnalysis analyzeRecovery(int symbolBaseline) {
589-
RecoveredFrameAnalysis analysis = new RecoveredFrameAnalysis(symbolBaseline);
610+
RecoveredFrameAnalysis analysis = new RecoveredFrameAnalysis(symbolBaseline, ackedFsn);
590611
try {
591612
for (int i = 0, n = sealedSegments.size(); i < n; i++) {
592613
sealedSegments.get(i).scanRecovery(analysis);
@@ -859,6 +880,10 @@ private static void sortByBaseSeq(ObjList<MmapSegment> list, int lo, int hi) {
859880
}
860881
}
861882

883+
private static boolean isLegacyReaderGuard(String name) {
884+
return LEGACY_READER_GUARD_A.equals(name) || LEGACY_READER_GUARD_B.equals(name);
885+
}
886+
862887
private static void swap(ObjList<MmapSegment> list, int i, int j) {
863888
if (i == j) return;
864889
MmapSegment tmp = list.get(i);

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -414,7 +414,8 @@ private static int countSegmentFiles(java.nio.file.Path dir) {
414414
int n = 0;
415415
if (files != null) {
416416
for (java.io.File f : files) {
417-
if (f.getName().endsWith(".sfa")) {
417+
if (f.getName().endsWith(".sfa")
418+
&& !f.getName().startsWith(".qwp-v2-guard-")) {
418419
n++;
419420
}
420421
}
@@ -1495,7 +1496,9 @@ private static boolean hasSegmentFile(java.nio.file.Path dir) {
14951496
java.io.File[] files = dir.toFile().listFiles();
14961497
if (files != null) {
14971498
for (java.io.File f : files) {
1498-
if (f.getName().endsWith(".sfa") && f.length() > 0) {
1499+
if (f.getName().endsWith(".sfa")
1500+
&& !f.getName().startsWith(".qwp-v2-guard-")
1501+
&& f.length() > 0) {
14991502
return true;
15001503
}
15011504
}

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -154,7 +154,9 @@ private static int countPopulatedSegmentFiles(String dir) {
154154
int rc = 1;
155155
while (rc > 0) {
156156
String name = Files.utf8ToString(Files.findName(find));
157-
if (name != null && name.endsWith(".sfa")) {
157+
if (name != null
158+
&& name.endsWith(".sfa")
159+
&& !name.startsWith(".qwp-v2-guard-")) {
158160
try {
159161
try (MmapSegment seg = MmapSegment.openExisting(dir + "/" + name)) {
160162
if (seg.frameCount() > 0) n++;

0 commit comments

Comments
 (0)