fix(qwp): robust pooled store-and-forward slot lifecycle — distinct ids, startup recovery, prompt close() - #53
Merged
Merged
Conversation
A SenderPool builds every sender from one immutable config string, so every store-and-forward (SF) sender inherited the same sender_id (default "default"), pointed at the same slot dir <sf_dir>/default, and the second borrow() died on the slot flock with "sf slot already in use by another process". An SF pool could therefore hold at most one live sender. Fix: when SF is enabled, the pool now hands each sender a distinct, stable slot id <base>-<index>, where <base> is the configured sender_id (default "default") and <index> is a pool slot index in [0, maxSize). Indices are recycled lowest-free-first, so a slot dir is reused deterministically and across a restart the same dirs are re-adopted and their unacked data is recovered on creation. A slot is only returned to the free set once its delegate has released the flock, tracked via a new closingSlots counter so the cap check (all + inFlight + closingSlots < maxSize) and the slot allocator stay consistent and no concurrent borrow can reclaim a slot dir whose lock is still held. Non-SF (HTTP/TCP/memory-mode) paths are unchanged: slotIndex is -1 and no slot bookkeeping runs. The cross-writer guard the slot lock exists for is preserved: two pools sharing one sf_dir with the same base still fail fast. Changes: - Sender.LineSenderBuilder: add isStoreAndForwardEnabled() and getConfiguredSenderId() introspection accessors (additive, no behavior change). The pool also probes the config once at construction so a bad config fails eagerly even when minSize == 0. - PooledSender: carry a stable slotIndex (-1 when SF is off). - SenderPool: per-slot sender_id derivation, slot-index allocator, and flock-safe slot lifecycle via closingSlots. Tests: new SenderPoolSfTest (12 tests) covering the two-concurrent-sender repro, grow-to-max coexistence, configured-base slot naming, slot reuse, reap-and-reuse, repeated-saturation invariant, end-to-end ingest, the cross-pool flock guard, distinct-base sharing, recovery replay through a re-adopted slot, and a concurrent borrow/return stress test. Existing non-SF SenderPoolTest (17) and the WS lifecycle suites remain green.
bluestreak01
changed the title
…m orphan drain
Pool teardown Error-safety (M1):
Best-effort teardown paths caught only RuntimeException, but a delegate
close()/worker shutdown() can throw an Error (e.g. an -ea AssertionError,
rethrown as-is by QwpWebSocketSender.close via rethrowTerminal). Such an
Error escaped and caused a permanent capacity leak:
- SenderPool.discardBroken: skipped freeSlotIndex + closingSlots--, so
an SF slot stayed reserved forever (slotInUse stuck true).
- SenderPool.reapIdle: aborted the close loop (sibling flocks leaked) and
skipped the slot-release block (all reaped indices stranded).
- PoolHousekeeper: the Error killed the daemon thread, stopping all
future reaping.
The understated capacity (cap check uses closingSlots) eventually forced
borrow() to only ever time out.
Fix: catch Throwable at every delegate close()/worker shutdown() site and
move the slot accounting into a finally so it always runs. Widen the
housekeeper's reapIdle guards to Throwable so an Error can never kill the
thread. Apply the same discipline to QueryClientPool (its close() loop had
no guard at all). Add regression tests that inject an AssertionError into
delegate teardown via a Proxy delegate swap and assert the reap/close loops
survive it, close every delegate, and leave the pool fully usable.
Orphan-drain namespace exclusion:
Add LineSenderBuilder.orphanDrainExcludePrefix / OrphanScanner.scan overload
so a pooled SF sender's startup drainer never adopts a sibling pool slot's
dir/lock (which would resurrect "sf slot already in use"). The pool sets the
exclude prefix to its own "<base>-" namespace; foreign leftovers are still
drained.
The pool's invariant was "delegate.close() returned => SF flock released", so discardBroken()/close() called freeSlotIndex() in a finally regardless. But QwpWebSocketSender.close() has an early-return path (ioThreadStopped becomes false when cursorSendLoop.close() throws) that returns normally before reaching cursorEngine.close() -- the only flock release. The pool would then free the index and hand a still-locked dir to the next borrow, resurrecting "sf slot already in use" permanently. Make the flock-release a queryable postcondition and retire (not reuse) a slot whose delegate did not release it: - QwpWebSocketSender: add slotLockReleased flag (set true only after the cursorEngine.close() block) + isSlotLockReleased() accessor. - SenderPool: add leakedSlots counter (folded into the borrow cap check); freeSlotIndex only when flockReleased() confirms the lock dropped, otherwise move the slot from closingSlots to leakedSlots and keep the index reserved permanently so no borrow ever reuses a locked dir. Releasing the flock early would be worse (a concurrent borrower could corrupt the slot dir the zombie I/O thread is still writing), so the slot is leaked rather than reused. Add SenderPoolSfTest.testSlotLeakedWhenDelegateCloseDoesNotReleaseFlock: forges the not-released symptom, routes the wrapper through discardBroken, and asserts the index stays reserved, a fresh dir is used, and capacity is permanently reduced. Verified it fails when the gate is neutered.
…erver API main #47 (a02732c) reworked TestWebSocketServer to bind an OS-assigned ephemeral port at construction and dropped the (int port, handler) ctor in favour of (handler) + getPort(). This PR's new SenderPoolSfTest predated that change and still called new TestWebSocketServer(port, handler), so the PR-merge-into-main build failed testCompile with 18x 'int cannot be converted to WebSocketServerHandler'. Merge main and switch all 18 call sites to the new API: drop TestPorts.findUnusedPort(), construct with the handler only, and read the bound port via server.getPort() inside the try block. Removes the now-unused TestPorts import. All 15 SenderPoolSfTest tests pass.
bluestreak01
enabled auto-merge (squash)
June 18, 2026 12:09
When a pooled store-and-forward sender's delegate close() returns with the slot flock still held (the I/O thread refused to stop), SenderPool retires the slot index permanently. Until now this happened silently: SenderPool had no logger, so a pool slowly bleeding capacity degraded to "every borrow() times out" with nothing in the logs to explain why. - SenderPool: add an SLF4J logger and emit a WARN at both leakedSlots++ sites (discardBroken + reapIdle) naming the retired slot index and the reduced effective capacity. - SenderPool: add public leakedSlotCount() accessor for metrics/tests. - Sender.build(): treat a benign concurrent mkdir(sf_dir) race (EEXIST, dir now exists) as success instead of failing the loser. - Tests: testLeakedSlotIsObservable (captures the WARN via a logback appender + asserts leakedSlotCount()) and testConcurrentFirstBorrowsWithMinZeroRaceOnSfDir (min=0 first-borrow race on sf_dir). module-info: test reads ch.qos.logback.core.
…meException The teardown-hardening work widened best-effort cleanup catches to Throwable but left the four outer catches that wrap the heavy native build/connect path (createUnlocked/build/connect/start) at catch (RuntimeException). The project runs with -ea, so those paths can throw an Error (AssertionError, OOM) that skips cleanup: - SenderPool ctor prewarm: an Error stranded already-built SF delegates (flock + mmap'd ring + I/O thread), resurrecting 'sf slot already in use'. - QueryClientPool ctor prewarm: an Error stranded pre-warmed worker threads. - QueryClientPool.acquire(): an Error skipped inFlightCreations--, permanently shrinking capacity until every acquire() timed out. - QueryClientPool.createUnlocked(): an Error skipped client.close(), leaking the field-initialised NATIVE_DEFAULT scratch. Widen all four to catch (Throwable e), matching SenderPool.borrow(). The inner cleanup loops were already Throwable-safe and the bodies already rethrow. Add red/green regression tests for each site. QwpQueryClient is concrete, so a package-private connectHook seam injects an Error at the real connect step (fromConfig still allocates, so NATIVE_DEFAULT leak assertions are meaningful); Sender is an interface, faked via Proxy behind a package-private senderFactory seam. Both seams are reached by reflection through package-private constructors (main module is open); production callers pass null for real behaviour. Each test was confirmed to go red when its catch is reverted to RuntimeException.
…axSize doesn't strand SF data A pooled SF sender excluded its whole "<base>-" namespace from orphan draining, while the pool only ever issues slot indices in [0, maxSize). If a deployment restarted with a smaller maxSize (e.g. 4->2) while default-2/default-3 still held unacked .sfa data, those slots were neither re-created (out of index range) nor drained (matched the excluded prefix) -- so even with drain_orphans=on the data was silently never recovered. Replace the coarse prefix exclusion with an exact managed-slot exclusion bounded by maxSize: - OrphanScanner.scan(sfDir, exclude, managedBase, managedSlotCount) excludes only <base>-<i> for canonical i in [0, managedSlotCount). Same-base indices >= count, and non-canonical suffixes (default-007, default-foo), are treated like foreign leftovers and drained through the existing flock-safe background drainer. - Sender builder hook orphanDrainExcludePrefix(String) -> orphanDrainExcludeManagedSlots(String base, int slotCount). - SenderPool passes (slotBaseId, maxSize). In-range live slots are still excluded, preserving the anti-collision guarantee the per-slot ids were added for. A base change keeps working (old dirs become foreign and drain). Tests (verified red without the fix): - OrphanScannerTest: bounded in-range exclusion, out-of-range drain after shrink, non-canonical names drained, disabled-when-count<=0, isManagedSlot predicate matrix. - SenderPoolSfTest#testShrinkingMaxSizeDrainsStrandedOutOfRangeSlots: seeds unacked data into default-0..3 via a maxSize=4 pool, restarts at maxSize=2 with drain_orphans=on, asserts default-2/3 are recovered (no .sfa left, no .failed sentinel).
PooledSender.close() wrapped delegate.flush() in catch (RuntimeException), so an Error (AssertionError under -ea, OutOfMemoryError, ...) left broken unset and the failed sender was returned to the pool via giveBack() as healthy. Since Sender does not clear its buffer on flush failure, the next borrower would inherit the unsent rows or a dead connection. Invert the flag to track normal completion instead: flush() must complete to set flushed=true, otherwise discardBroken() runs. This treats any abnormal exit (RuntimeException or Error) as unrecyclable, needs no explicit rethrow (the original throwable propagates naturally), and matches the catch-Throwable posture already used elsewhere in the pool.
The QuestDBImpl constructor orchestrates SenderPool and QueryClientPool construction: it builds the SenderPool first, then the QueryClientPool, then starts the housekeeper, with a cleanup catch that closes whatever was already built on failure. That catch was catch (RuntimeException). Both pool constructors run heavy native build/connect paths (mmap, flock, WebSocket connect) that can throw an Error under -ea (AssertionError) or via OOM in production. The teardown-hardening work already widened SenderPool, QueryClientPool, and their prewarm catches to catch (Throwable) for exactly this reason, but the orchestrator that ties them together was missed: an Error from new QueryClientPool(...) skipped the narrow catch, so the already-built SenderPool was never closed -- stranding its prewarmed delegates' flocks, mmap'd rings, and I/O threads. This is the precise leak class the hardening work exists to kill, reachable under -ea (how QuestDB tests run) and via OOM in production. Widen the cleanup catch to catch (Throwable e), matching the pools it calls. Add a red/green regression test. A package-private QuestDBImpl seam constructor (mirroring the SenderPool/QueryClientPool seams) threads the senderFactory and connectHook through to the pools; production callers pass null for both. The test fakes a Sender via Proxy (close() flips a flag, senderMin=1) and injects an AssertionError from the QueryClientPool connect hook (queryMin=1); the fake delegate's close() flag is the discriminator. Confirmed red against catch (RuntimeException) and green against catch (Throwable).
Every pooled SF sender's orphan drainer excludes the pool's whole [0, maxSize) managed slot range (orphanDrainExcludeManagedSlots) so a sibling never adopts a slot dir/lock the pool intends to (re)create -- that exclusion is what keeps the per-slot ids from resurfacing "sf slot already in use". The side effect: an in-range slot left holding unacked data by a previous run was recovered ONLY when the pool happened to (re)create that index. Because the pool pre-warms [0, minSize) and builds [minSize, maxSize) lazily at the lowest free index on demand, a high in-range slot under steady low load was never rebuilt -- neither drained (excluded) nor recovered -- so its store-and-forward data sat stranded on disk (durable, but undelivered) until a restart or a load spike. This was strictly weaker than the drain_orphans=on contract a standalone sender gives, and the Javadoc quietly assumed the slot would be recreated. Close the gap without touching the exclusion (widening it would re-open the cannibalization race). The pool now recovers its own stranded managed slots once, at construction, while it is still single-threaded and unpublished: for each in-range slot dir that holds unacked data and is not already live (prewarmed), it reserves the index, builds a sender on it (which re-adopts and recovers the slot), drains the recovered frames, then closes it and frees the index. Reserving the index for the duration means no concurrent borrow can target the dir, so the cannibalization race the exclusion guards against cannot occur here either. Every step is best-effort: a clean slot is a cheap directory probe; an unreachable server, slow drain, or build/close Error is logged and never fails construction (the data stays durable on disk for a later attempt), and a build/connect failure short-circuits the scan to avoid paying a connect timeout per slot. Add getConfiguredSfDir() to LineSenderBuilder so the pool can locate its group root, mirroring the existing getConfiguredSenderId() hook. Add a red/green regression test: seed default-0..3 in a busy maxSize=4 run, then restart at the SAME maxSize=4 with minSize=0 and steady low load (a single borrow). Before the fix all four in-range slots stayed stranded (test waits 15s and fails); after the fix startup recovery empties every slot and replays the frames to the server. Confirmed red with the recovery call disabled and green with it enabled; full SenderPool/SF/orphan suites pass unchanged.
Commit 7118264 made PooledSender.close() route any abnormal flush() exit (RuntimeException OR Error) to discardBroken instead of recycling, but shipped without a test that exercises the Error branch -- the existing flush-failure tests drive a LineSenderException (a RuntimeException), which the pre-fix catch (RuntimeException) already handled, so they pass against both pre- and post-fix code. Add a white-box test via the package-private senderFactory seam: inject a delegate whose flush() throws an AssertionError, then assert the wrapper is discarded (next borrow() returns a fresh instance) rather than recycled. Verified RED against 7118264^ (fails on assertNotSame -- the broken wrapper is handed back to the next borrower) and GREEN on the current tree.
SenderPool.borrow() reserves an SF slot index (allocateSlotIndex) before building the delegate and must return it (freeSlotIndex) if createUnlocked() throws, or slotInUse[idx] stays stuck true: capacity is permanently lowered and a later borrow() trips "no free SF slot index" / eventually only times out -- the failure mode this PR fixes. The pre-existing error-injection test fails in the constructor pre-warm loop with a non-SF config (slotIndex == -1), so the borrow path and the SF (slotIndex >= 0) case were uncovered. Inject (via the senderFactory seam) an SF-config pool whose factory throws on the first borrow-triggered build and succeeds afterwards; assert the throwable propagates AND a subsequent borrow() reuses the slot (capacity intact). Verified RED by removing freeSlotIndex(slotIndex) from the borrow catch (2nd borrow() throws IllegalStateException "no free SF slot index"), GREEN on the current tree.
recoverStrandedManagedSlots() ran one full acquireTimeoutMillis drain per stranded slot, so a reachable-but-not-acking server could block QuestDB construction for (maxSize - minSize) * acquireTimeoutMillis. Cap the whole scan at a single shared acquire-timeout budget and short-circuit the moment a drain fails to ack; remaining slots stay durable on disk for a later attempt. Covered by testStartupRecoveryIsBoundedByASharedBudget (M1).
…rload The 3-arg scan(sfDir, exclude, prefix) form was dead public surface: the 4-arg managed-slot overload is its documented precise replacement and the only form production uses (Sender.build). Fold the directory walk back into the 2-arg body, delete the overload and its three prefix-only tests, and repair the now-dangling javadoc cross-reference (M2).
Address three review findings on the SF slot-collision branch. C1 (correctness): recoverStrandedManagedSlots() freed the slot index unconditionally in its finally, ignoring whether the recoverer's close() released the flock. When close() early-returns with the I/O thread still running (flock held), a later borrow re-picked the index and threw "sf slot already in use", permanently poisoning the slot. The recovery finally now mirrors discardBroken/reapIdle: retire permanently (leakedSlots++, slotInUse stays set) when flockReleased() is false. Adds SenderPoolSfTest regression driving the forged-flock-held recovery. M4 (maintainability): collapse the production-dead 2-arg OrphanScanner.scan into a thin delegate to the managed-aware 4-arg form (removes ~30 duplicated lines and the literal-vs-\u2014 em-dash drift), and extract the duplicated flock-release/retire+WARN block from discardBroken and reapIdle into a single reclaimSlot() helper (behaviour and per-path WARN text preserved). M2 (test gap): add SlotLockReleasedContractTest under the qwp test tree, driving the real QwpWebSocketSender.close() paths instead of forging the slotLockReleased field: a clean close => isSlotLockReleased()==true, and a forced !ioThreadStopped early-return => isSlotLockReleased()==false. Verified non-vacuous: a getter inversion fails both tests and a wrongly-set slotLockReleased on the leak path fails the leak-path test.
…tests - SenderPoolTest: failing-close proxy now forwards close() to the real delegate before throwing the injected Error, so the resource-safety teardown tests no longer leak the delegate's native HttpClient while still exercising the pool's Error-survival path. - SenderPoolSfTest: add coverage for the previously-untested reapIdle() leaked-slot branch (reclaimSlot during idle reaping with the flock still held) and a deterministic complement that asserts build() tolerates a pre-existing sf_dir (no reliance on a probabilistic mkdir race). - PoolHousekeeper: document the catch(Throwable) guards as intentionally defensive and restore the rationale comment dropped from the second catch. - QueryClientPool/QueryWorker: pool teardown hardening + error-safety tests.
QuestDBImpl ran its three teardown steps -- housekeeper.stop(), queryPool.close(), senderPool.close() -- as bare sequential calls in both the ctor cleanup catch and close(). A Throwable from an earlier step (e.g. an OutOfMemoryError from QueryClientPool.close()'s `new ArrayList<>(all)`, or an Error from the housekeeper join) would skip the remaining closes. SenderPool runs last and owns the flock/mmap/I-O-thread resources, so the worst-ordered step was the one most likely to be skipped -- exactly the leak class the surrounding teardown-hardening work (and the widened ctor catch) exists to kill. Wrap each cleanup step in its own try/catch(Throwable) via closeQuietly helpers in both sites, so senderPool.close() always runs. Make QueryClientPool implement AutoCloseable (mirroring SenderPool) so it is usable through the typed helper.
testReapIdleSurvivesDelegateCloseError and testCloseSurvivesDelegateCloseError borrow real DEAD_HTTP_CONFIG senders (native HttpClient) but had no native-memory accounting. The prior commit fixed a real native leak in these very tests (the failing-close proxy now forwards real.close() before throwing); without a leak guard a future regression would pass green and go unnoticed. Wrap both bodies in TestUtils.assertMemoryLeak(...), matching the convention already used by SenderPoolSfTest. Verified the guard bites: re-introducing the leak fails with a ~270KB native HttpClient delta.
QuestDB.build() ran recoverStrandedManagedSlots() synchronously in the SenderPool constructor, so against a reachable-but-not-acking server the first stranded slot's drain consumed the whole shared acquire-timeout budget before construction could return -- a multi-second stall even for minSize == 0, undocumented on build(). Move recovery off the constructor onto the existing PoolHousekeeper daemon (no new thread): it now drives a one-shot runStartupRecoveryOnce() up front, before its first interval wait. build() returns without blocking; recovered data stays durable on disk and is delivered once the server acks. - SenderPool: add an 8-arg constructor with a deferStartupRecovery flag (6-/7-arg constructors, incl. the reflection test seam, delegate with false -> unchanged inline behavior); add idempotent runStartupRecoveryOnce(); make the in-range pass concurrency-safe by counting its slot reservation in a new recoveringSlots field included in the borrow() capacity check, so deferred recovery overlapping live borrows can neither over-allocate past maxSize nor target a dir being recovered. - QuestDBImpl: construct the sender pool with deferStartupRecovery=true. - PoolHousekeeper: run runStartupRecoveryOnce() once at thread start, Throwable-guarded like the reap calls. - QuestDBBuilder.build(): document that startup recovery is deferred and build() does not block on it. - OrphanScanner: add listStrandedOutOfRangeManagedSlots() backing the pass-2 recovery of same-base slots a larger previous run left out of the current index range. Tests: add deferred-path coverage (non-blocking construction; delivery + idempotency when driven) and the out-of-range listing test.
Deferred SF recovery ran as a blocking one-shot prelude on the PoolHousekeeper thread (runStartupRecoveryOnce) that re-checked `closed` only at entry. Against a reachable-but-slow-to-ack server -- the exact scenario the deferral exists for -- a close() landing while a recoverer was mid-drain could not stop the scan: housekeeper.stop()'s join(2s) timed out, yet the daemon kept holding the slot flock (plus a live WebSocket + I/O thread) and kept building senders on a logically-closed pool for up to the remaining acquire-timeout budget (~3s). The in-flight recoverer is a local PooledSender never added to `all`, so the pool's close() snapshot loop could not reach it either. Two consequences: close() returned while a background thread still owned native resources, and a reopen on the same sf_dir+base during that residual window could collide on the still-held flock -> "sf slot already in use", the precise failure this work set out to eliminate. Restructure recovery into a resumable, per-slot scan driven on the existing reap-loop thread (no new thread): - SenderPool: replace the blocking recoverStrandedManagedSlots() with a cursor-based recoverOneSlotStep() that recovers at most one stranded slot per call (skipping live/empty slots cheaply), preserving the two-pass in-range/out-of-range semantics, reservation accounting and leak-retire logic. Two drivers sit over it: runStartupRecoveryToCompletion() (inline construction + test drives, shared acquire-timeout budget) and runStartupRecoveryStep() (the housekeeper's per-tick unit). Each step's drain is capped at RECOVERY_DRAIN_BUDGET_MILLIS (1s), kept below PoolHousekeeper.STOP_TIMEOUT_MILLIS (2s) so a drain in flight when close() arrives cannot outlive the housekeeper join. - SenderPool: add markClosing() to raise the `closed` shutdown signal without tearing down delegates, and split the teardown guard into a new closeStarted flag so the early signal never short-circuits close()'s delegate teardown. Every recovery step re-checks `closed`. - PoolHousekeeper: drive recovery one slot per loop iteration, back-to-back while work remains (skipping the idle wait so the backlog drains promptly), re-checking stop between every step; name the stop/join budget STOP_TIMEOUT_MILLIS. - QuestDBImpl.close(): call senderPool.markClosing() before stopping the housekeeper, so an in-flight recovery step bails between slots and the join cannot time out on a fresh slot's drain. Trade-off: per-slot drains are now capped at 1s each (vs the old 5s shared budget), trading some per-run recovery headroom on slow links for prompt, race-free shutdown -- recovery data stays durable on disk and resumes on the next start. Tests: add testCloseDuringDeferredRecoveryStopsBuildingOnClosingPool, a deterministic guard (a fake recoverer parks slot-0's drain; the shutdown signal is raised mid-drain; slot 1 must never be built). Verified it bites: neutering the closed re-checks fails with builtSlots=[0, 1]. Repoint the existing deferred-drive test seam to runStartupRecoveryToCompletion.
bluestreak01
changed the title
Recovery delegates are built and connected on the PoolHousekeeper thread. The build went through the normal sender factory, so it inherited the user's initial-connect mode -- and any reconnect_* knob auto-promotes that to SYNC, which retries the connect for the whole reconnect budget (default 5 min) inside build(). RECOVERY_DRAIN_BUDGET_MILLIS only caps the drain that follows, not this build. A close() landing during such a build made housekeeper.stop()'s join(2s) time out, detached the housekeeper thread, and let the recoverer take/release the slot flock after close() returned -- reopening the "sf slot already in use" window the PR set out to close (M1). Route recovery through a dedicated recoverySenderFactory whose production impl (defaultRecoverySender) forces initial_connect_mode=OFF, so a recovery build makes at most one connect attempt instead of a SYNC reconnect-budget retry loop. An injected test factory still drives both paths, preserving the white-box recovery seam. One residual window remains and is now documented honestly rather than implied closed: a single in-flight connect to a black-holed/firewalled host still blocks on the OS connect timeout, because the transport exposes no application-level connect timeout. Updated the boundedness comments on RECOVERY_DRAIN_BUDGET_MILLIS, recoverOneSlotStep, runStartupRecoveryStep, and PoolHousekeeper accordingly. Adds testRecoveryDelegateForcesOffInitialConnectMode: with a SYNC-promoting config, asserts the normal delegate resolves to SYNC while the recovery delegate is forced to OFF (RED before the fix: expected OFF but was SYNC).
Member
Author
Review (level 3, mission-critical pass)Read all changed source in full, built the change-surface map with real searches, traced the cap-check invariant and all reclaim/recovery paths, compiled the module, and ran all 72 new/changed tests (green). Change surface (summary)
Callsite inventory confirms the new symbols are reached only from CriticalNone. I specifically tried to break the central invariant and the recovery/close races and could not. The cap-check invariant is airtight: every ModerateM1 — No test exercises the production SF-recovery path (real
Minor
Downgraded (verified false positives)
SummaryVerdict: approve — with M1 (production-path test coverage) worth addressing before merge for a mission-critical close path; minors at the author's discretion.
🤖 Generated with review-pr (level 3) |
The QueryClientPool constructor null-defaults the field this.startHook (line 129), but the prewarm loop read the unqualified name startHook, which resolves to the constructor parameter that shadows the field. Production builds the pool via the 7-arg constructor, which delegates with startHook == null, so any QuestDB.connect() prewarming a query pool (minSize >= 1) threw a NullPointerException out of QueryClientPool.<init>. Qualify the call as this.startHook.accept(pending) so it uses the null-defaulted field (QueryWorker::start) instead of the raw parameter. The other startHook use (in acquire()) and the connectHook use (in createUnlocked()) sit in separate methods where no shadowing applies and were already correct. This was caught by QuestDBFacadeE2ETest, whose 10 methods all errored on the same NPE; they pass after the fix.
The javadoc <source> was hardcoded to 1.8, but on JDK 11+ the active
java11+ profile passes --add-exports as a javadoc option. javadoc rejects
--add-exports with -source 8 ("option --add-exports not allowed with
target 8"), breaking the javadoc/release-artifacts builds.
Use ${javac.compile.source} (already used by the compiler plugin) so the
javadoc source level tracks the JDK: 11 on JDK 11+ (where --add-exports is
valid) and 1.8 on JDK 8 (where compilerArg1/2 are no-op -Xlint flags).
Add a CI workflow (PR + push to main) that matches how the client is
shipped and consumed:
* build-jdk8: the client ships as a Java 8 artifact and is released from
JDK 8, so JDK 8 is the source of truth -- compile, run the full test
suite against the committed native libs, and build the javadoc jar
(mvn -P javadoc clean install).
* compile-jdk25: the client is a submodule of the JDK 25 questdb repo, so
guard against JDK 25 compile breakage (main + test sources) without
running tests (the parent runs them against a real server).
Also fix the JDK 8 javadoc build: the javadoc config reused the javac
compilerArgs, which on JDK 8 are -Xlint:none -- a valid javac flag but an
invalid javadoc option ('javadoc: error - invalid flag: -Xlint:none').
Give javadoc its own per-profile J-options: --add-exports on JDK 11+ (to
resolve the src/main/java11 jdk.internal.math.FDBigInteger bridge) and a
harmless -quiet no-op on JDK 8 (where the bridge uses the open
sun.misc.FDBigInteger and needs no export).
Extend the JDK 25 smoke to build the javadoc jar (mvn -P javadoc package, no tests) in addition to compiling main + test sources, so javadoc is covered on both JDK 8 and JDK 25.
A startup-recovery delegate is built from the shared config string, so with drain_orphans=on it inherited the flag. Since recovery forces initial_connect_mode=OFF (one connect attempt), against a reachable server the build succeeds and reaches startOrphanDrainers(), standing up a BackgroundDrainerPool for any foreign/out-of-range orphan. Its close() -- called from drainCandidateSlotForRecovery() on the PoolHousekeeper thread, BEFORE cursorEngine.close() releases the slot flock -- then blocks up to GRACEFUL_DRAIN_MILLIS + STOP_GRACE_MILLIS (3s) against a reachable-but-not-acking server. That makes one recovery step ~1s drain + ~3s drainerPool.close ~= 4s, overrunning RECOVERY_DRAIN_BUDGET_MILLIS and PoolHousekeeper.STOP_TIMEOUT_MILLIS (2s); a close() landing mid-step then returns with the flock still held, resurfacing the "sf slot already in use" window this PR exists to eliminate. Recovery delegates now force drain_orphans=off: their sole job is to drain their own slot. Sibling in-range slots are covered by recoverOneSlotStep's own passes; foreign/out-of-range orphans are covered by the live pooled senders' drainers (whose close() senderPool.close() awaits synchronously, so they release their flock before close() returns). Adds testRecoveryStepStaysBoundedWithDrainOrphansAgainstNonAckingServer: seeds an in-range stranded slot + a foreign orphan, restarts with drain_orphans=on against a reachable-but-silent server, drives one recovery step, and asserts it returns within STOP_TIMEOUT_MILLIS with the foreign data preserved (not lost, not .failed). Fails pre-fix (~3.5s), passes post-fix (~1s).
The build now ships as a Java 8 artifact (JDK 8 is the source of truth, CI builds/tests/releases on it) while still compiling on JDK 11+. Update the project guidance to describe the dual-target profiles and the Java 8 language/API floor, and list the newer-than-8 features that are off-limits (var, List.of/copyOf, String.repeat, ProcessHandle, Thread.onSpinWait, the Java 9 Provider(String,String,String) ctor, enhanced switch, text blocks, instanceof pattern vars).
Contributor
[PR Coverage check]😍 pass : 313 / 385 (81.30%) file detail
|
Member
Author
Review (level 3, mission-critical pass)Built the change-surface map with real searches, traced the cap-check invariant and every reclaim/recovery/close path, compiled on JDK 25, ran the 72 new/changed tests (green), confirmed CI green on HEAD ( This is a careful, well-reasoned PR. I reproduced the central conclusions of the existing self-review independently and found two substantive items it missed (a hot-path allocation regression, and the exact test count), plus corroboration of its coverage gap. CriticalNone. I specifically tried to break:
ModerateM1 —
M2 — Production SF-recovery wiring is untested end-to-end (corroborates the self-review's M1). M3 — Scope: the JDK 8 target-lowering is bundled with the SF fix. The build-target change (the Minor
Downgraded (false positives, verified against source)
SummaryVerdict: approve — with M1 (hot-path allocation) and M2 (production-path test) worth addressing before merge; M3 (split the JDK-8 work) at the author's discretion; minors trivial.
🤖 Generated with review-pr (level 3) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
SenderPoolbuilds every sender from one immutable config string. When store-and-forward (SF) is enabled (sf_dirset), every sender therefore inherited the samesender_id(default"default"), pointed at the same slot dir<sf_dir>/default, and took an exclusiveflockthere for its lifetime. The secondborrow()died on the lock:Net effect: an SF pool could hold at most one live sender — pooling + SFA was effectively broken. The code even documented the footgun (
Sender.java: "multi-sender setups must set this explicitly or the second sender will fail") but the pool had no way to satisfy it with a single config string.Closing the collision exposed a chain of adjacent robustness gaps in the pooled SF slot lifecycle — flock release on teardown, orphan-drain coexistence, recovery of data stranded across restarts, and prompt shutdown — which this PR addresses as one coherent change.
Fix
When SF is enabled, the pool now hands each sender a distinct, stable slot id
<base>-<index>:<base>= the configuredsender_id(default"default"), so two pools/processes can share onesf_dirby using distinctsender_ids.<index>= a pool slot index in[0, maxSize), recycled lowest-free-first.Because indices are deterministic and reused, a slot dir is re-adopted across restarts and any unacked data it holds is recovered on creation (recovery semantics rely on stable ids — random UUIDs would orphan the data).
Non-SF paths (HTTP / TCP / memory-mode) are unchanged —
slotIndexis-1and no slot bookkeeping runs. The cross-writer guard the slot lock exists for is preserved: two pools sharing onesf_dirwith the same base still fail fast.Flock-safe slot lifecycle
A slot index is only returned to the free set after its delegate releases the
flock. The pool tracks the in-between states with counters, all folded into the capacity check:closingSlots— a slot whose delegate is mid-close()and still releasing its flock. Transient: once the flock drops, the index goes back to the free set. This keeps a concurrentborrow()from reclaiming a dir whose lock is still held.leakedSlots— a slot whose delegateclose()returned with the flock still held (the I/O thread refused to stop, observed viaQwpWebSocketSender.isSlotLockReleased()).QwpWebSocketSender.close()has an early-return path (whencursorSendLoop.close()throws, it returns before the only flock-release), so "close() returned" does not imply "flock released". The index is retired permanently — never freed, never reused — so no borrow ever hands out a still-locked dir. Releasing the flock early would be worse (a concurrent borrower could corrupt the dir the zombie I/O thread is still writing). The lost capacity is accounted for in the cap math, so the pool degrades gracefully instead of resurrecting "sf slot already in use". The release/retire logic is shared by the three close-and-reclaim paths (discardBroken,reapIdle, startup recovery) via onereclaimSlot()helper.recoveringSlots— a slot held by the startup-recovery pass (see below); counted so deferred recovery overlapping live borrows can neither over-allocate pastmaxSizenor target a dir being recovered.A permanent retirement is no longer silent:
SenderPoolhas a logger and emits aWARNat every retire site naming the slot index and the reduced effective capacity, and exposes the count via a publicleakedSlotCount()accessor for metrics. Previously the pool had no logger at all, so a pool bleeding capacity degraded to "everyborrow()times out" with nothing in the logs to explain why.Orphan-drain coexistence
The pool co-manages every
<base>-<index>slot and recovers each slot's unacked data when it (re)creates it. To stop a sibling sender's startup drainer from adopting another live pool slot's dir/lock (which would resurface "sf slot already in use"), the pool fences off exactly its own managed slots from orphan draining:LineSenderBuilder.orphanDrainExcludeManagedSlots(String base, int slotCount)builder option,OrphanScanner.scan(...)managed-slot exclusion that excludes only the canonical pool-minted names<base>-<i>foriin[0, slotCount),build()passes the configured base + slot count into the orphan scan.The exclusion is bounded, not a coarse
<base>-prefix: a same-base leftover whose index falls outside the current[0, maxSize)range — e.g.<base>-3stranded after amaxSizeshrink across restarts — is treated as a foreign leftover and drained, instead of being silently fenced off and stranded forever. Non-canonical suffixes (default-007,default-foo) are likewise drained. This is a no-op unlessdrain_orphans=on; foreign leftovers under other names are still drained.Startup recovery of stranded managed slots
Because the orphan drainer excludes the pool's live
[0, maxSize)range, an in-range slot left holding unacked data by a previous run would otherwise be recovered only when the pool happens to (re)create that index — so under steady low load it could sit durable-but-undelivered on disk indefinitely. The pool closes this gap by recovering its own stranded managed slots, in two passes:[0, maxSize)slots that hold unacked data and are not already live: reserve the index (underrecoveringSlots, so no concurrent borrow can target it), build a sender that re-adopts and drains the slot, then close it and free the index. A recoverer whoseclose()leaves the flock held retires the slot permanently, exactly like the other reclaim paths.<base>-i,i >= maxSize, viaOrphanScanner.listStrandedOutOfRangeManagedSlots). These are recovered regardless ofdrain_orphans, so the default config never strands them.Recovery never blocks
build()and never delays a promptclose():QuestDBhandle constructs its pool withdeferStartupRecovery=true, so recovery does not run in theSenderPoolconstructor (against a reachable-but-not-acking server an inline drain would otherwise stall construction for seconds). It is driven instead on the existingPoolHousekeeperreap-loop thread — no new thread.runStartupRecoveryStep), back-to-back while work remains (skipping the idle wait so the backlog drains promptly), re-checking itsstopflag between every step. Direct (non-pooled)SenderPoolconstruction still recovers inline to completion (runStartupRecoveryToCompletion), single-threaded and unpublished, bounded by one shared acquire-timeout budget so a non-acking server can't pay a per-slot timeout(maxSize - minSize) ×over.RECOVERY_DRAIN_BUDGET_MILLIS(1s), kept belowPoolHousekeeper.STOP_TIMEOUT_MILLIS(2s).QuestDBImpl.close()raises the pool's shutdown signal (markClosing()) before stopping the housekeeper — a separatecloseStartedflag guards the actual delegate teardown so the early signal never short-circuits it — so aclose()landing while a recoverer is mid-drain only ever waits out a single bounded drain. The housekeeper join cannot time out on a fresh slot, the recoverer's flock is released beforeclose()returns, and the residual-window "sf slot already in use" on reopen is gone.Trade-off: per-slot recovery drains are capped at 1s each (vs. the old 5s shared budget), trading some per-run recovery headroom on slow links for prompt, race-free shutdown — stranded data stays durable on disk and resumes on the next start.
Error-safe teardown (run under
-ea)createUnlocked()runs a heavy native build path (mmap, flock, WebSocket connect) and delegateclose()/flush()can throw anError(e.g. an-eaAssertionError,OutOfMemoryError), not just aRuntimeException. Every best-effort teardown and creation site is thereforecatch (Throwable)so a strayErrorcan't strand sibling flocks, over-count slot bookkeeping, kill the housekeeper daemon, or leak a half-built pool:SenderPool/QueryClientPoolprewarm,borrow()/acquire(),createUnlocked(), anddiscardBroken/reapIdleslot accounting (moved intofinallyso it always runs).PooledSender.close()tracks normalflush()completion (flushed=true) and routes any abnormal exit —RuntimeExceptionorError— throughdiscardBroken, so a sender that failed to flush is never recycled with unsent rows.QuestDBImplorchestration: the ctor cleanup catch isThrowable, and each teardown step (housekeeper.stop(),queryPool.close(),senderPool.close()) is independently guarded viacloseQuietlyhelpers so a Throwable from an earlier step can't skip closing theSenderPool, which runs last and owns the flock/mmap/I-O-thread resources.Construction-race tolerance
The pool builds senders outside its lock, so two first-borrows sharing one
sf_dircan race intomkdir(sf_dir); the loser previously failed with a spuriouscould not create sf_dir.build()now treats a benign creation race (the dir exists afterwards) as success.What changes for consumers
io.questdb:questdb-clientnow compiles and runs on Java 8+(previously Java 11+). Java 8 is the release source of truth — the
artifact is built and released from JDK 8.
minimum runtime/build JDK. Existing Java 11+ consumers are unaffected.
1.8bytecode; building on JDK 11+ emits${javac.target}(default11).How the multi-JDK build works
build — no flag needed:
java8(activated on JDK 1.8):source/target=1.8, excludesmodule-info.java, drops the--add-exportsflags, logback1.3.x.java11+(activated on JDK ≥ 11):source/target=${javac.target}(default 11), keeps
--add-exports java.base/jdk.internal.math, logback1.5.x.src/main/java8andsrc/main/java11, added to thecompile path by
build-helper-maven-pluginvia thecompat.src.dirproperty. Each holds a JDK-specific copy of two tiny bridge classes:
Compat— shims APIs missing on Java 8 (e.g.Thread.onSpinWait()becomes a no-op; PID lookup uses the Java 8 path).
FdBig— bridge over the JDK's internalFDBigIntegerused by thedouble→string fast path: the java8 copy targetssun.misc.FDBigInteger,the java11 copy targets
jdk.internal.math.FDBigInteger(reached via thecompile-time
--add-exports).Numbersdouble-to-string path was rerouted from a directjdk.internal.math.FDBigIntegerreference to theFdBigbridge so itcompiles on both JDKs. This is the high-blast-radius change — it sits on
the hot path for every serialized
double. Behaviour is unchanged; only theclass it routes through differs per JDK.
Unsafe.addExportsremoved (along withJAVA_BASE_MODULE/implAddExports): the old runtime reflective self-injection ofjdk.internal.mathis gone. On JDK 11+ the export is now a compile-timeflag in the
java11+profile; on JDK 8 it is not needed at all.1.3.xline on Java 8(
1.5.xrequires Java 11).New CI (
.github/workflows/ci.yml)clean installwith-P javadoc: compiles, runs the fulltest suite against the committed native libs, and builds the javadoc jar.
This is the source-of-truth job.
-DskipTests clean package -P javadoc) ofmain + test sources and the javadoc jar, to guard against breakage when the
client is consumed as a submodule of the JDK-25
questdbparent repo.Also bundled (flagged by M-B, listed for completeness)
QueryClientPoolstartHookNPE fix on the prewarm-from-production path.for the SF lifecycle work above).
Changes
Sender.LineSenderBuilder— additive introspection accessorsisStoreAndForwardEnabled(),getConfiguredSenderId(),getConfiguredSfDir(); theorphanDrainExcludeManagedSlots(base, slotCount)option;build()wires the managed base + slot count into the orphan scan and tolerates a benign concurrentmkdir(sf_dir)race. The pool probes the config once at construction, so a bad config fails eagerly even whenminSize == 0.QwpWebSocketSender—isSlotLockReleased()(set true only after the flock-release block) so the pool can tell whetherclose()actually dropped the flock.OrphanScanner— bounded managed-slot exclusion (canonical<base>-<i>foriin[0, slotCount)only);listStrandedOutOfRangeManagedSlots()backing pass-2 recovery.PooledSender— stableslotIndex(-1when SF off);close()discards on any abnormalflush()exit.SenderPool— per-slotsender_idderivation; slot-index allocator; flock-safe lifecycle viaclosingSlots/leakedSlots/reclaimSlot(); bounded managed-slot orphan-drain fencing; resumable per-slot startup recovery (recoverOneSlotStep,runStartupRecoveryStep,runStartupRecoveryToCompletion) withrecoveringSlotscap accounting;markClosing()+closeStartedfor prompt, cancellable shutdown; Throwable-safe teardown; logger +WARNon permanent retirement +leakedSlotCount()accessor.PoolHousekeeper— drives recovery one slot per loop iteration on its reap thread, re-checkingstopbetween steps; namedSTOP_TIMEOUT_MILLIS; Throwable-safe reap loop.QuestDBImpl— constructs the sender pool withdeferStartupRecovery=true; signalsmarkClosing()before stopping the housekeeper; independently guarded teardown steps; Throwable ctor cleanup catch. Package-private seam constructor (senderFactory + connectHook) for white-box error-injection tests.QueryClientPool/QueryWorker— Throwable-safe teardown / prewarm / acquire;QueryClientPoolimplementsAutoCloseable.QuestDBBuilder.build()— documents that startup recovery is deferred andbuild()does not block on it.core/pom.xmlgains the auto-activatedjava8/java11+profiles,build-helper-maven-plugin(per-JDK source roots viacompat.src.dir), JDK-drivensource/target, and a profile-scopedlogback.version; newsrc/main/java8+src/main/java11Compat/FdBigbridges;Numbersreroutes thedouble→string path throughFdBig;UnsafedropsaddExports/JAVA_BASE_MODULE/implAddExports(thejdk.internal.mathexport is now compile-time, JDK 11+ only); new.github/workflows/ci.yml(JDK 8 build+test+javadoc, JDK 25 compile smoke).Tests
New / expanded white-box coverage (each behavioural guard verified red→green by neutering the fix):
SenderPoolSfTest(27) — two concurrent SF senders (original repro); grow-to-max coexistence + extra borrow times out; configured-base slot naming; slot reuse and reap-and-reuse index recycling; repeated-saturation invariant; end-to-end ingest; cross-pool flock guard; distinct-base sharing;close()releases all slots; leaked slot retired permanently when delegateclose()leaves the flock held (viadiscardBrokenandreapIdle); leaked slot observable (WARN +leakedSlotCount()); recovery replay through a re-adopted slot; concurrent borrow/return stress;drain_orphansdoes not cannibalize sibling slots but still drains foreign orphans; shrinkingmaxSizedrains stranded out-of-range slots; in-range idle slot recovered at startup under steady low load; default config recovers out-of-range slots after shrink; startup recovery bounded by a shared budget; deferred construction does not block + delivers when driven (idempotent); retire-on-recovery when the recoverer keeps the flock; benignsf_dirmkdirrace (concurrent + deterministic pre-existing-dir);close()mid-recovery stops building on a closing pool (the C1 race — fake recoverer parks a drain, shutdown signalled mid-drain, next slot must not be built).SenderPoolTest(19) — non-SF coverage incl. Error-safe reap/close teardown wrapped inassertMemoryLeak; borrow-path SF slot-index release on creation failure; flush-Error→discardBroken.SenderPoolErrorSafetyTest(3) /QuestDBImplErrorSafetyTest(1) /QueryClientPoolErrorSafetyTest(5) —Throwable(notRuntimeException) on every creation/teardown path, via Proxy/connectHookseams.OrphanScannerTest(14) — bounded in-range exclusion, out-of-range drain after shrink, non-canonical names drained,isManagedSlotmatrix.SlotLockReleasedContractTest(2) — realQwpWebSocketSender.close()paths: clean close ⇒isSlotLockReleased()==true; forced!ioThreadStoppedearly-return ⇒false.Verification
SenderPoolSfTest27/27 ✅ ·SenderPoolTest19/19 ✅ ·SenderPoolErrorSafetyTest3/3 ✅ ·QuestDBImplErrorSafetyTest1/1 ✅ ·QueryClientPoolErrorSafetyTest5/5 ✅ ·OrphanScannerTest14/14 ✅ ·SlotLockReleasedContractTest2/2 ✅QuestDBBuilderTest/LineSenderBuilderTestand the WS lifecycle /sf.**/sf.cursor.**suites remain green (no regression).