Skip to content

fix(qwp): preserve and retire schema-rejected QWP batches - #94

Draft
jerrinot wants to merge 6 commits into
mainfrom
jh_poison_pill_release
Draft

fix(qwp): preserve and retire schema-rejected QWP batches#94
jerrinot wants to merge 6 commits into
mainfrom
jh_poison_pill_release

Conversation

@jerrinot

@jerrinot jerrinot commented Sep 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

A batch rejected for not matching a table’s structure could previously block all later data, even after restarting the client.

This change reports the rejection and removes the affected batch from the retry queue. With disk buffering enabled, it saves a copy first. Applications can continue by returning the failed sender to the pool and borrowing again, or by rebuilding a standalone sender. The previous stop-on-error behavior remains available as an option.

To restore the previous behavior:

Sender sender = Sender.builder("ws::addr=localhost:9000;sf_dir=./sender-data;")
        .schemaMismatchPolicy(SenderError.Policy.TERMINAL)
        .build();

The same option is available on QuestDB.builder() for pooled senders.

Also fixes memory buildup during outages and startup failures caused by damaged saved copies.

@jerrinot
Preserve and retire schema-rejected QWP batches
13ca3df 
Add REJECT_AND_CONTINUE with lease-scoped failures, bounded rejection notifications, and durable raw recovery copies before retirement. Preserve rejected ranges synchronously on the I/O thread and reuse existing queue watermark and shutdown cleanup machinery.

Cover transactional and recovered groups, dictionary continuity, pool recovery, copy retries, and blocked-copy shutdown. Document compatibility and operational behavior.

Validation: full core suite passed with 3,492 tests, zero failures/errors, and seven skipped; git diff --check passed.
@jerrinot jerrinot changed the title fix(Qwp): preserve and retire schema-rejected QWP batches fix(qwp): preserve and retire schema-rejected QWP batches Sep 8, 2026
@jerrinot
fix(test): keep rejection archive tests compatible with Java 8
f2c6d58 
Replace Path.of and String.repeat with Java 8-compatible equivalents, preserving the long-message round-trip coverage.

Validated on Temurin Java 8: reactor compilation, six archive tests, examples, and Javadoc packaging passed.
@jerrinot
fix(test): compare recovered archive paths portably
984ed35 
Compare Path objects instead of raw strings so equivalent Windows path separators do not fail the recovery notification assertion.

Validation: 22 related tests passed locally on Linux; Windows CI confirmation remains pending.
@jerrinot
Fix schema lease retention and rejected archive recovery
729c212
@jerrinot
fix(test): persist ACK watermark in archive recovery fixture
26e3c3a
@jerrinot

jerrinot commented Sep 8, 2026

Copy link
Copy Markdown
Contributor Author

Code review — level 3

Base 0b9b576 (merge-base with main) → head 26e3c3a. ~4,900 lines across 22 production files, 14 test files, README and a new design doc.

Gates. Compiles clean on JDK 25. Full core suite: 3508 tests, 0 failures, 0 errors, 7 skipped. No submodule pointer moved (zstd unchanged). No committed binaries. No post-Java-8 syntax or APIs anywhere in the diff (TestUtils.repeat is the repo's own helper, not String.repeat).

Verdict: approve with comments. No Critical findings; correctness and test gates both pass. 0 Critical / 8 Moderate / 6 Minor.

Priority order for this PR: M1, M2, M3 — a stalled slot that notifies nobody will generate a support ticket, and the other two are small fixes that lose diagnostics on failure paths.

Moderate

M1 · A transactional open-group rejection dispatches no async error, and only the producer thread can unblock it

When reject() fires on an active transactional lease with no commit-bearing frame between the rejected FSN and publishedFsn, firstCommitFsn returns -1, so end = -1 (the owner.active arm at SchemaRejectionState.java:116-125). pending is created with error == null, finishFailure is skipped, and stopFsn is set anyway. On the I/O thread:

  • sealedRange() returns null → tryRetireSchemaRange returns false every iteration
  • trySendOne (CursorWebSocketSendLoop.java:3571) stops at stopFsn and sends nothing further
  • clampAckBeforeSchemaStop pins acks at stopFsn - 1
  • the installed == true arm (:4537-4540) calls failSchemaPaced and returns — no dispatchError

Sealing is reachable only via sealIfNeeded, called from endLease / ownedFailure — both producer-thread-driven. The I/O thread needs a seal it cannot perform itself.

At base the TERMINAL arm called dispatchError(err) synchronously with the NACK (git show 0b9b576:…CursorWebSocketSendLoop.java:4146), so a handler was notified immediately regardless of producer activity.

Scope, stated accurately: awaitAckedFsn (QwpWebSocketSender.java:1129), drain, flush and every producer API call route through checkConnectionError() → checkSchemaFailure(), which seals and throws. The stall lasts only until the producer's next touch and nothing is discarded. The exposed population is an idle or bursty transactional producer whose ingestion health is watched through the async SenderErrorHandler: that observer sees nothing, and the slot neither sends nor acks, for an unbounded wall-clock window.

Suggested fix: seal on the I/O thread using engine.publishedFsn() when the owner is active, or emit an interim un-pathed notification on the installed arm. A bare dispatchError there would double-notify, since retirement later dispatches the pathed notification.

M2 · PooledSender.close() throws from a finally, discarding the primary exception

PooledSender.java:198 throws operationalFailure from inside the finally block, which silently discards the in-flight exception from the try, with no addSuppressed.

Trigger: a pooled SF sender whose lease owns a schema failure and whose connection has a latched transport error, so checkSchemaSlotHealth() rethrows connectionError. A catch (LineSenderServerException e) around a try-with-resources borrow then receives an unrelated LineSenderException with empty getSuppressed(), losing getRejectedFsn() and getRejectedPath().

The method's own surviving comment at :204-210 — "The original throwable propagates naturally once this finally returns — no explicit rethrow needed" — is now false for this path.

Suggested fix: capture the primary, addSuppressed(operationalFailure), rethrow the primary.

M3 · An unreadable rejected/ directory reads as "no archive", then is memoized for the process lifetime

Files.findFirst's javadoc (Files.java:359-363) is explicit: -1 means "opendir / FindFirstFile failed — directory does not exist, no read permission, transient error, etc.", and "Distinguishing this from a 'real empty' success matters for recovery code paths."

RejectedMiniSlotArchive.findOverlapping:243-247 collapses both into find <= 0 → return null. (The inner if (find > 0) ff.findClose(find) is dead code inside that branch.) cleanupTemporaryDirectories:308 has the same conflation.

tryRetireOrphanTail then memoizes the null in recoveredOrphanReportLookedUp (set at :3799, never reset), so one transient EMFILE/EACCES permanently suppresses the report — and the orphan tail is retired anyway. The archive sits on disk with nothing pointing an operator at it.

Suggested fix: branch find < 0 (fail the lookup, retry later) apart from find == 0 (genuinely empty), and do not memoize a failed lookup.

M4 · dlqEnabled(true) is silently inactive when no destination exists

configureSchemaMismatch (QwpWebSocketSender.java:2928) only builds a SchemaPreserver when cursorEngine.sfDir() != null || directory != null. A plain ws::addr=host:9000; sender takes Sender.java:1573 (slotPath = null), so sfDir() is null; with no .dlqDirectory(...) the if never fires. dlqEnabled stays true, schemaPreserver stays null, and tryRetireSchemaRange takes the else branch and discards the range with no archive.

The asymmetry is the defect, not the policy: when a destination is configured but unusable, SchemaPreserver.probeDestination throws at build time (SenderPool.java:597, QwpWebSocketSender.java:2939). The one configuration that loses data with no copy is the one that gets no diagnostic. getDlqFilesWritten() stays 0, indistinguishable from "no rejections yet". The README documents the behaviour in one sentence, but nothing in the API does.

Suggested fix: LOG.warn once in configureSchemaMismatch when policy == REJECT_AND_CONTINUE && preserve but no destination resolves, or reject it at build time the way an unusable destination already is.

M5 · getRejectedFsn() returns toFsn for four pre-existing non-NACK error kinds

The public 10-arg SenderError constructor delegates ..., quarantinedPath, toFsn, null) (SenderError.java:104), so rejectedFsn = toFsn for every site not going through withRejectionSpan. Correct for handleServerRejection (which passes fsn, fsn); wrong for the four pre-existing sites that build a real span with NO_MESSAGE_SEQUENCE as the message sequence:

Site Category
CursorWebSocketSendLoop.java:1922 SECURITY_ERROR — ws-upgrade failed
CursorWebSocketSendLoop.java:1962 PROTOCOL_VIOLATION — durable-ack mismatch
CursorWebSocketSendLoop.java:2128 RETRIABLE — to = max(from, publishedFsn)
CursorWebSocketSendLoop.java:2247 PROTOCOL_VIOLATION

LineSenderServerException.java:68 now prints it unconditionally, so a TLS/auth upgrade failure spanning [0,5000] reads SECURITY_ERROR rejectedFsn=5000 fsn=[0,5000] — asserting frame 5000 was rejected when no NACK occurred. The accessor's javadoc says "Local FSN named by the NACK."

No test asserts the exception message text; the only getRejectedFsn() assertion is SchemaRejectionPoolTest.java:247, on the genuine NACK path.

Suggested fix: pass NO_MESSAGE_SEQUENCE at those four sites, and print rejectedFsn= only when meaningful.

M6 · Stranded .tmp- archive trees in memory mode

For a memory-backed engine the epoch is UUID.randomUUID() per sender instance (QwpWebSocketSender.java:2931). cleanupTemporaryDirectories's match prefix embeds that epoch (RejectedMiniSlotArchive.java:309), so a new instance can never reclaim a temp tree left by an earlier one, and no epoch-agnostic sweep exists — grep -rn '"\.tmp-' over core/src/main returns exactly two hits, and all four cleanupTemporaryDirectories call sites pass the epoch.

Each stranded tree holds rejected.sfa pre-allocated to the full span size plus any dictionary snapshot. Not counted by sf_max_total_bytes, not visible in getDlqBytesWritten() (bumped only on a completed publish, CursorWebSocketSendLoop.java:3872), never deleted.

Disk-backed slots are safe: the epoch is persisted in .slot-epoch, so an intra-epoch retry or a restart under the same epoch cleans up.

Suggested fix: derive a stable epoch for memory mode, or add an age-based sweep of .tmp-* under rejected/ that ignores the epoch.

M7 · Nothing pins the new default policy, and it is duplicated across five sites

REJECT_AND_CONTINUE is independently hardcoded as the default in Sender.java:1093, QuestDBBuilder.java:75, QwpWebSocketSender.java:411, and the SenderPool / QuestDBImpl delegating constructors. Every test that exercises the policy sets it explicitly (BackgroundDrainerEndToEndTest.java:117; SchemaRejectionPoolTest.java:71, 291, 332, 394, 429). Reverting any or all five leaves the full 3508-test suite green.

This is the single line deciding whether the default population's rejected rows are retained or discarded.

Related: the PR description does not mention the default flip at all — it reads as a pure feature-add. The README and design/schema-mismatch-terminal-resolution.md document it properly (design doc phase 2: "Flip the schema default to REJECT_AND_CONTINUE only when this ships"), but the description is what feeds release notes for a client library whose users inherit the new behaviour on upgrade.

Suggested fix: one shared constant, plus one test that builds with no policy call and asserts the effective retire-vs-halt behaviour. Add the default flip and the awaitAckedFsn/getAckedFsn semantic change to the PR description.

M8 · Required cross-repo tandems are absent

Recorded:

gh pr view 94 --json body | rg -i 'questdb/questdb(-enterprise)?|tandem|e2e'   -> no match
gh pr list --repo questdb/questdb            --head jh_poison_pill_release --state all -> empty
gh pr list --repo questdb/questdb-enterprise --head jh_poison_pill_release --state all -> empty

Both repos are reachable from this machine (gh pr list --state open --limit 2 returns rows for each), so this is a genuine absence, not a permissions gap.

The change trips all three triggers: server-observable wire behaviour (retire-vs-halt on a NACK, post-retirement catch-up realignment, ack clamping) needs an OSS e2e tandem; the SF drainer, pool startup and reconnect pacing are HA-facing and need an Enterprise tandem; and the durability claim — bytes on disk before they are retired — rests on the temp → fsync → rename → fsync → validate → ack protocol surviving a kill -9, which no JVM unit test can express. RejectedArchiveRecoveryTest constructs directories; it does not kill a process.

Not Critical, and the reason matters: the ordering is correct by construction — preserve() fully publishes and validates before engine.acknowledge(range.lastFsn) (CursorWebSocketSendLoop.java:3897), a crash before the ack replays the source range, ff.exists(finalDir) makes re-preservation idempotent, and metadata plus per-frame CRCs make a torn archive detectable rather than silently wrong.

Reverting the whole tryRetireSchemaRange hunk does fail six tests, including CursorWebSocketSendLoopCatchUpAlignmentTest.java:164-167, which drives the real production caller trySendOneForTest() and asserts ackedFsn, getSchemaFramesRetired() and stopFsn() together. The hunk as a whole is well covered; the remaining gaps are per-arm:

  • the loop's retry re-entry at CursorWebSocketSendLoop.java:3573 — testPreservationFailureRetriesThenRetiresWithoutTerminal calls tryRetireSchemaRangeForTest() twice by hand and never calls loop.start()
  • the predecessor-walk terminal arm at :4498-4504
  • the terminal (non-retry) preserve arm at :3848-3859, and dispatcher == null at :3828

Minor

  • SchemaRejectionPoolTest.java:130 throws AssertionError("orphan frames must be retired without sending") from a TestWebSocketServer handler. That runs on the server's readThread, whose lambda catches only IOException (TestWebSocketServer.java:754), so the named invariant is unenforced. The harness swallow is pre-existing (CloseDrainTest.java:890, BackgroundDrainerMidDrainCapabilityGapTest.java:582 do the same at base); what is new is using the idiom to assert a data-path invariant rather than a latch timeout. Same shape at :274, :376.
  • SenderPoolSfTest.java:2634-2636 — Files.delete(...) then assertFalse(Files.exists(...)) cannot fail (delete throws instead). It replaced an assertion that did depend on production behaviour, and nothing asserts the new eager-probe footprint. Note the probe now creates sf_dir at pool construction — itself an untested behaviour change.
  • Four new test files (SchemaRejectionStateTest, RejectedMiniSlotArchiveTest, RejectedArchiveRecoveryTest, SchemaRejectionPoolTest) allocate native memory with zero assertMemoryLeak, against 61/78 in the same package.
  • SchemaRejectionPoolTest.java:410 — three-level private-field reflection where b.flushAndGetSequence() is used four times in the same file (:193, :208, :241, :337). The field(...) helper at :99 also uses getDeclaredField, which does not walk superclasses.
  • RejectedMiniSlotArchiveTest.java:43-57 hand-rolls temp-dir setup/teardown where TestUtils.createTmpDir (:154) / removeTmpDirRec (:233) exist. Hardcoded '/' path fragments in the FilesFacade fakes (SchemaPreservationCloseTest.java:125, two in PoisonFrameTest) on a branch that already carries 984ed35 "compare recovered archive paths portably".
  • Javadoc continuation lines indented four spaces past the opening /** on all six new builder methods (QuestDBBuilder.java:172-176, 183-186, 192-196; Sender.java:2144-2148, 2156-2159, 2165-2169). Private test helpers interleaved between @Test methods (SchemaRejectionStateTest.java:146, SchemaRejectionPoolTest.java:98).

Hypotheses chased and disproved

Recording these so the next review does not re-spend the effort:

  • Archived tail frame corrupted by the flag mutation. QWP headers carry no checksum over the flags byte, and MmapSegment.tryAppend computes its frame CRC over the already-mutated scratch bytes (MmapSegment.java:724).
  • Stale liveLookupIndex/liveLookupOffset serving the wrong frame. Every route reaches MmapSegment.liveFrameOffset through the three synchronized SegmentRing delegates (:1302/1307/1312) — full caller inventory checked. rebaseSeq throws when frameCount > 0 (MmapSegment.java:667), so it only runs on an empty segment whose memo is still (0, HEADER_SIZE). The producer's tryAppend never touches the memo.
  • Lost update in acknowledgedThrough. All three writers are I/O-thread-only (CursorWebSocketSendLoop.java:2299, :4245, SchemaRejectionState.java:145). Single writer.
  • schemaPending leaking and stalling retirement forever. The closed || !schemaInbox.offer(...) short-circuit never enqueues on a closed dispatcher, and dispatchLoop's condition drains schemaInbox before exiting.
  • Reconnect budget imposed on the running drainer. failSchemaPaced mirrors failPaced exactly — per-attempt backoff capped, retry loop unbounded, counter reset on a real ACK (:4207). No new terminal.
  • transactional clobbered by the legacy connectWithCredentialSupplier overload. The only production caller is Sender.java:1705 on the new 27-arg overload, and Sender.java:1781 still sets it post-connect exactly as at base.
  • preparedSchemaFirstFsn guard skipping dictionary catch-up after a recycle. swapClient does not reset sentDictCount; the mirror survives via setWireBaselineWithCatchUp.
  • Unsafe publication of the plain pool-side schema fields. beginSchemaLease / endSchemaLease / prepareSchemaPoolSlot are all under the pool's ReentrantLock (SenderPool.java:1286, :1355, :1709, :1351).
  • Native-memory and fd leaks in the new archive code. preserve0's maxPayload is provably > 0; writeMetadata, validate and releaseSkippedFrameScratch pair correctly on every path; snapshotTo's fd read is page-cache-coherent with the MAP_SHARED mapping.

Tradeoffs worth confirming consciously

These are the team's decisions, documented in the README and design doc. Not findings — flagging them only so the release note is deliberate:

  1. The default schema-mismatch policy flips from halt-and-retain to retire-and-continue. Existing users inherit row discards on upgrade unless they set .schemaMismatchPolicy(TERMINAL).
  2. awaitAckedFsn / getAckedFsn now return true / advance for locally-retired frames. Code shaped like if (awaitAckedFsn(fsn, 30_000)) { /* ingested */ } silently changes meaning. The javadoc was updated; the compiler cannot help.
  3. Background orphan-drainer schema reports now carry the orphan queue's local FSN span to the same user handler that receives live-sender reports, where the base code deliberately stripped it (BackgroundDrainer.java:1111-1117 vs the surviving else if at :1119-1133). getRejectedPath() != null is the only discriminator.

Submodule provenance: no pointer moved — core/src/main/c/share/zstd unchanged. Scope is this repo's source only. Admitted split: 8 in-diff, 0 out-of-diff-breakage.

🤖 Generated with Claude Code

@mtopolnik

mtopolnik commented Sep 9, 2026

Copy link
Copy Markdown
Contributor

[PR Coverage check]

😍 pass : 765 / 894 (85.57%)

file detail

path covered line new line coverage
🔵 io/questdb/client/impl/PooledSender.java 10 21 47.62%
🔵 io/questdb/client/Sender.java 7 12 58.33%
🔵 io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java 89 117 76.07%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java 186 232 80.17%
🔵 io/questdb/client/QuestDBBuilder.java 10 12 83.33%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/SenderErrorDispatcher.java 19 22 86.36%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java 44 49 89.80%
🔵 io/questdb/client/impl/SenderPool.java 28 31 90.32%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainer.java 56 61 91.80%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/RejectedMiniSlotArchive.java 176 193 91.19%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/SchemaRejectionState.java 104 108 96.30%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java 3 3 100.00%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java 6 6 100.00%
🔵 io/questdb/client/LineSenderServerException.java 1 1 100.00%
🔵 io/questdb/client/cutlass/qwp/client/sf/cursor/DefaultSenderErrorHandler.java 4 4 100.00%
🔵 io/questdb/client/impl/SenderSlot.java 8 8 100.00%
🔵 io/questdb/client/SenderError.java 12 12 100.00%
🔵 io/questdb/client/impl/QuestDBImpl.java 2 2 100.00%

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@jerrinot @mtopolnik