perf(core): make wait_wal_table(), sleep() functions lightweight and not block IO threads

[ideoma]

Tandem https://github.com/questdb/questdb-enterprise/pull/1006

Summary

• Productionises wait_wal_table(table[, seq_txn]) and adds sleep(seconds) as the first two callers of a new worker-continuation framework. While these functions are parked, the worker carrier is released to run other queries, so the number of concurrent calls is no longer bounded by the worker pool size.

• Introduces an internal continuation runtime in io.questdb.mp.continuation built on jdk.internal.vm.Continuation. A SQL function body can yield the carrier, register a wakeup condition (target seqTxn reached, table state change, timer pop), and resume on any worker. WAL commit notification fires waiters as soon as the target txn lands; a TimerShards daemon pool drives periodic wakeups so the body can probe the circuit breaker, cancel flag and SQL timeout.

• Replaces JDK ThreadLocal with a carrier-keyed CarrierLocal along the continuation-critical paths. C2 hoists the JIT-cached thread identity across yield/resume boundaries, which means a stale value can survive a carrier migration. CarrierLocal routes reads through an opaque carrier-identity intrinsic to defeat the hoist. The analysis is in CARRIER_LOCAL.md.

SQL functions

wait_wal_table(table_name [, seq_txn])

Blocks the current query until the WAL writer for table_name has applied transactions up to a target seq_txn. Returns BOOLEAN (true on success). Useful when a client needs to confirm that a prior write has been materialised before issuing a follow-up read.

Arguments:

• table_name STRING -- name of a WAL-enabled table. For non-WAL tables the call returns true immediately.
• seq_txn LONG, optional -- target sequencer txn to wait for. When omitted (or NULL), the call captures the table's current seq_txn at start time and waits for that. The argument must be a constant or runtime constant (not a column reference).

Behaviour:

• Returns true as soon as the writer txn reaches seq_txn.
• Throws table is suspended [tableName=...] if the table becomes suspended while the call is waiting.
• Throws table does not exist if the table is dropped while the call is waiting.
• Respects the SQL circuit breaker: query timeout, explicit CANCEL QUERY, and broken client connection abort the wait promptly.
• The wait is never unbounded: a dead client, a CANCEL QUERY, or query.timeout always wins.

Concurrency: while parked, the worker that started the call is released to run other queries. The number of concurrent wait_wal_table calls is no longer bounded by the shared-worker pool size.

Examples:

INSERT INTO trades VALUES (now(), 'AAPL', 100, 150.0);
-- wait until WAL apply has caught up to the seq_txn observed now
SELECT wait_wal_table('trades');
-- same
SELECT wait_wal_table('trades', null);

-- wait for a specific seq_txn
SELECT wait_wal_table('trades', 42);

sleep(seconds)

Pauses the current query for the given number of seconds and emits a single TIMESTAMP row containing the server's wall-clock time at wake. Implemented as a table function, so it can be used as a top-level query as well as inside SELECT.

Argument:

• seconds DOUBLE -- sleep duration in seconds. Must be finite and non-negative. Maximum is 24 hours; longer values are rejected with a sleep duration exceeds 24 hour maximum error. Sub-millisecond values round down to 0 ms and return immediately.

Behaviour:

• Returns a single-row cursor with one TIMESTAMP column named sleep.
• Respects the SQL circuit breaker on every wake interval (griffin.query.continuation.wake.interval, default 1s): query timeout, explicit CANCEL QUERY, and broken client connection abort the sleep within one wake interval.
• During engine shutdown the sleep aborts with sleep aborted, connection closing.

Concurrency: while parked, the worker carrier is released. Concurrent sleep calls do not pin worker threads.

Examples:

-- top-level use
sleep(0.5);

-- pause between two statements in a script
INSERT INTO t VALUES (now(), 'x');
sleep(0.1);
SELECT * FROM t WHERE ts > dateadd('s', -1, now());

Configuration

• cairo.timer.shards -- number of TimerShards daemon threads driving deadline-based wakeups. Defaults to min(4, max(1, cpu_count / 4)). Higher values reduce DelayQueue lock contention at the cost of one always-on thread per shard.
• griffin.query.continuation.wake.interval -- millisecond interval at which a parked suspending function body wakes to probe the circuit breaker. Default 1000. Lower values trade timeout/cancel latency for more wake-up work.

Known risk: context-local reader-pool supervisor

To let the reader-pool leak supervisor survive a continuation park/resume, this PR moves it off a per-thread ThreadLocal and onto the SqlExecutionContext (read through SqlExecutionContext.getReaderPoolSupervisor()). This relies on an invariant: a single SqlExecutionContext is only ever driven by one thread at a time. Production satisfies the invariant -- each connection and query owns its context, and a parked continuation resumes on at most one carrier thread -- so the supervisor never sees concurrent access.

The supervisor's backing state (QueryProgress.readers) is not thread-safe. When two threads do share one context, concurrent getReader calls cross-wire the supervisor slot and corrupt that list, which leaks a reader; the leaked, goodbye'd-but-retained reader then pollutes later work on the same engine. Several concurrency stress tests hit exactly this by fanning one context across worker threads. Commit a0051ae gave the parallel group-by contention tests their own per-thread context, and a follow-up does the same for SampleByTest and SampleByNanoTimestampTest.

The invariant is enforced by convention, not by the type. Any future production path or test that shares one SqlExecutionContext across threads reintroduces the corruption. Making the supervisor access thread-safe, or asserting single-thread ownership on the context, would remove the risk at the source; this PR does not do that and instead relies on the ownership convention plus the test fixes above.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 2ff4cb51-8511-421d-b414-091fc805497b

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat-productionise-wait-wal-table-func

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[bluestreak01]

@ideoma — review from the review-pr skill at --level 3 (independent pass with per-finding source verification against the current branch state).

The load-bearing surface (continuation runtime, CarrierLocal/CarrierIdentity + Rust FFI, the Worker driver + job rotation, SeqTxnTracker, wait_wal_table/sleep) is unusually well-engineered and documented. The CAS state machines, the size==0 lost-wakeup fast path, the Rust FFI panic/ABI safety, and the CarrierLocal id-recycle ordering are all correct. But the job-rotation recycle machinery is inert during normal operation, which turns the advertised feature into an unbounded native-memory leak on the REST path.

Critical

C1. Unbounded selector + HTTP-handler leak on the REST path — the rotation recycle pool is never replenished except at pool halt (in-diff)

Locations:

• core/src/main/java/io/questdb/mp/Worker.java:236,267 — recycleJobList(cont) is reachable only from the two cont.isDone() branches.
• core/src/main/java/io/questdb/mp/Worker.java:326-420 — loopBody has no return/break except the while (lifecycle.get() == RUNNING) guard, so a continuation reaches isDone() only at pool halt.
• core/src/main/java/io/questdb/mp/Worker.java:443-461 — mintNextGen cold-allocates and appends every stateful clone to ownedJobClones (line 459).
• core/src/main/java/io/questdb/mp/WorkerPool.java:209 — ownedJobClones is freed only by halt().
• core/src/main/java/io/questdb/cutlass/http/HttpServer.java:435-470, 524-526, 610-614 — HttpRequestJob.cloneInstance() → selectorFactory.acquire() → create() mints a fresh HttpRequestProcessorSelectorImpl populated with the entire REST handler set (factory.newInstance() per registered URL: JsonQueryProcessor, import/export processors, etc.). recycleInstance() → selectorFactory.release() is the only thing that refills recyclePool.
• core/src/main/java/io/questdb/griffin/engine/functions/date/SleepFunctionFactory.java:198-214 — sleep re-suspends once per wakeIntervalMillis (default 1000ms); WaitWalFunction.getBool re-parks the same way.

Trace.

1. A continuation wraps the worker's entire loopBody loop, not a single query. loopBody only returns when the worker is HALTED, so cont.isDone() is never true during RUNNING.
2. recycleJobList() — the sole feeder of Worker.snapshotPool and the sole caller of Job.recycleInstance() — runs only from the isDone() branches. Therefore during RUNNING it never runs: snapshotPool stays empty and HttpRequestProcessorSelectorFactory.recyclePool is never replenished.
3. Every time a body deep-parks (a wait_wal_table/sleep suspend()), the outer driver takes the handoff == null path and calls mintNextGen. With snapshotPool empty, mintNextGen always cold-allocates and cloneInstance() → acquire() → create() builds a fresh selector + full handler set, then ownedJobClones.add(clone) pins it (strong ref) until halt.
4. Magnitude is per-wake, not per-call: sleep(3600) re-parks ~3600 times (one per 1s wake), so a single hour-long REST sleep leaks ~3600 complete HTTP handler sets. wait_wal_table against a lagging writer behaves identically.

The HttpRequestProcessorSelectorFactory comment ("steady-state rotation amortises to 'pop a selector from the queue' instead of allocating a fresh selector + handler set per suspend") and DESIGN_NOTES.md ("pool size converges to the workload's concurrent-suspend count") are false: the pool is write-only at halt, so steady state always allocates. Tests miss it because everything is freed at halt, before assertMemoryLeak's native check, and result assertions are insensitive to heap growth.

Impact. Unbounded native + heap growth on the HTTP/REST interface, proportional to the number of wait_wal_table/sleep suspend episodes, reclaimed only on server shutdown — i.e. triggered by exactly the long-poll/monitoring workloads these functions exist for. PGWire is unaffected (its dispatcher Job uses the default cloneInstance() returning this, so clones are not pinned), and the shared/ILP pools never run suspending functions, so the leak is HTTP-specific — but the REST path is a primary QuestDB interface.

Suggested fix. The recycle pool must be replenished during RUNNING, not only at halt. Options: (a) when a deep-parked body is resumed and completes its current job iteration, return the abandoned generation's stateful clones to the pool at that point rather than waiting for cont.isDone(); (b) decouple selector lifetime from cont/gen lifetime — acquire a selector lazily per run() and release it back to recyclePool at the end of each non-suspending iteration; (c) at minimum, bound ownedJobClones. Until fixed, the steady-state-amortisation claims in HttpServer and DESIGN_NOTES.md are incorrect and should not ship.

Moderate

M2. Shutdown is not fenced behind a guaranteed ContinuationQueue drain (in-diff)

ServerMain.close() calls engine.signalClose() then immediately workerPoolManager.halt(). signalClose() → timerShards.shutdown() only fires parked entries (cont.scheduleResume() enqueues into the per-pool ContinuationQueue) and returns; it does not wait for any cont to remount and unwind. The resume queue is drained only inside Worker.loopBody's tryDequeue, gated by lifecycle == RUNNING. If halt() flips workers to HALTED before they dequeue a just-fired cont, those conts are stranded in ContinuationQueue and the in-flight connection/query native resources they own (socket fd, native buffers, any open TableReader) are never released. Bounded and shutdown-only (harmless at process exit; OS reclaims), but it matters for embedded/repeated-engine-restart use and would surface as an intermittent native leak in leak-sensitive tests. The CairoEngine.close() assert (closing || timerShards.size() == 0) covers the timer heap but not the resume queue. Consider a bounded "drain ContinuationQueue until empty (or deadline) while workers RUNNING" step before halt(), or document the residual as accepted-at-process-exit.

M3. TimerShards.shutdown()/halt() ordering can silently skip the drain (in-diff)

TimerShards.halt() never drains; shutdown() early-returns without draining when !running. So if halt() is ever called before shutdown() — both are documented as idempotent and order-independent — the entry-drain (entry.shutdown() on each parked waiter) is silently skipped. Production is safe today because CairoEngine always drives shutdown() (via signalClose) before halt() (via close), and the partial-init path is guarded by the assert in CairoEngine.close(). But the class Javadoc advertises an order-independence the implementation doesn't honor. Make shutdown() always drain (the toArray/clear path is heap-safe regardless of daemon state), or have CairoEngine.close() call shutdown() before halt().

M4. Job.run() Javadoc contradicts the implementation about workerId identity (in-diff)

Job.java:107-117 states "workerId is the calling carrier's globally-unique id (see CarrierIdentity)… The Worker loop body fills this argument with CarrierIdentity.current() at every call site." The implementation passes the pool-local id: Worker.java:337 reads currentWorkerId = WORKER_ID.get() (the pool-local CarrierLocal<Integer>) and passes that to myJobs.get(i).run(currentWorkerId, …). Pool-local is the correct choice (it's what PerWorkerLocks expects), but the Javadoc asserts the opposite and a non-existent uniqueness guarantee. This is the exact misunderstanding that enables M5.

M5. Latent cross-carrier workerId aliasing (in-diff design risk; not demonstrably reachable today)

Because currentWorkerId is captured in the frozen frame, when a body suspends mid-iteration on carrier A (pool-local id a) and resumes on carrier B, the remainder of that iteration runs on B's thread with workerId == a, while A's fresh continuation also runs workerId == a. Two carriers then hold the same pool-local workerId concurrently. The mint-new-generation policy isolates per-job-instance state, but workerId-indexed global state (PerWorkerLocks, per-worker scratch arrays, RingQueue single-producer slots) is not isolated. I could not construct a reachable corruption with the current suspending functions — wait_wal_table/sleep are runtime-constant/table functions evaluated on the IO worker that do not touch pool-local-workerId-indexed shared scratch — so this is not a live bug. But it is a real hazard for any future suspending function, and M4's incorrect Javadoc makes it likely someone trips it. The design docs acknowledge migration but do not enumerate which Jobs are safe under cross-carrier remount; that enumeration should exist.

M6. TxnWaiter.cancel() is an unconditional write but its contract describes a CAS (in-diff)

TxnWaiter.cancel() (:121-123) is state = STATE_CANCELLED; (void, no CAS), yet its own Javadoc (:100-120) and DESIGN_NOTES.md describe a tryCancel that "CASes PENDING → CANCELLED… on iteration 2+ tryCancel's CAS fails harmlessly." There is no CAS; on the happy path it clobbers FIRED → CANCELLED. The current behavior is benign (after the finally the body is already unwinding and never re-suspends, so no resume is lost), but the safety argument the design leans on is false and one refactor away from breaking. Make it a real CAS matching the doc, or rewrite the doc + DESIGN_NOTES to justify an unconditional terminal write.

M7. Per-iteration FFI downcall added to the hot worker loop, including legacy pools (in-diff, perf)

Worker.loopBody:337 reads WORKER_ID.get() every loop iteration — a CarrierLocal.get() → CarrierIdentity.current() critical FFI downcall plus an open-addressed map lookup — replacing master's plain workerId field read. This runs on every pool, including legacy ILP TCP IO/writer pools that never migrate and gain nothing. It is a measurable per-iteration regression on the busy-spin/cheap-job path. On legacy pools (continuationQueue == null) use the stable workerId field directly; only non-legacy pools need the carrier-aware read.

M8. Test coverage gaps in the new concurrency machinery (in-diff)

• The carrier-pinned / suspend-refused fallback (suspend() returns false → markParkRefused/consumeParkRefused phantom-dequeue protocol) is entirely untested — no test forces suspend() to return false. This is the trickiest path and a regression there (peer busy-spin or lost/duplicated resume) would slip through.
• The Job recycle-on-completion path (attachJobs/takeJobs/recycleInstance) has no focused test — and given C1, a test that asserts the recycle pool actually refills during operation would have caught the leak.
• ServerMainSleepTest.testFuzzConcurrentSleeps computes its anti-serialization metric (sumSleptMillis vs busyWallMillis) but never asserts it. Add the assertion or soften the comment.

Minor

• ColumnTypeConverter has no registered CarrierLocal cleaner despite DESIGN_NOTES.md:225-238 naming it as the example requiring one. Harmless today (every conversion eagerly unmaps via detachFdClose() in finally), but the doc tells future authors to rely on a cleaner that doesn't exist. Register a cleaner or correct the note.
• WaitWalTableSeqTxnFunctionFactory is missing from function_list.txt. Harmless — that file is only an ordering map; discovery is by classpath/module scan, and the 2-arg form has no overload-ordering conflict. Cosmetic inconsistency only (the PR did add SleepFunctionFactory).
• AsyncLogRecord.getDejaVu() is dead code (zero callers) with a comment claiming the writer thread clears dejaVu — it never does. The AsyncLogRecord(Clock, name) name parameter is also dead. Delete both.
• Side-effecting assertion assert recyclePool.pop() == null in HttpServer.java:607 — pop() mutates under -ea. Invariant holds, but prefer a non-destructive peek.
• SeqTxnTracker.initTxns() advances writerTxn without fireWaiters() (unlike updateWriterTxns). Bounded extra latency (≤ wake interval), not a hang — worth a comment.
• CarrierLocal.set(null) no longer re-runs the factory (diverges from the old io.questdb.std.ThreadLocal, matches java.lang.ThreadLocal). Dormant today, but a trap for future call sites.
• Test nits: SeqTxnTrackerTest imports not alphabetized; TimerShardsTest.testShardDistribution dead empty loop + terminalCount.get() == 1 ? 1 : 0 assertion that hides the real value; ServerMainSleepTest comments contradict the actual config.

Downgraded (false positives / already resolved on the branch)

• @TestOnly on SeqTxnTracker.getWriterTxn(): already fixed — @TestOnly is on getWaiterRegistrationCount(); getWriterTxn() is unannotated.
• .returnsOnce(...) on a deterministic query: not present on the current branch; the only .returns(...) use is correct.
• Shared IODispatcher running concurrently on two carriers: verified safe — dispatch is via an MCSequence (multi-consumer, by design), the epoll loop is a SynchronizedJob, and per-handler scratch is isolated by per-generation selector cloning.
• DelayHeap.take() infinite wait(0,0): unreachable — delay <= 0 is handled separately, so when millis == 0 then nanos > 0.
• CairoException allocating per throw: not a regression — matches prior -ea behavior, no caller depends on instance reuse/identity, and it removes a real catch-rethrow aliasing hazard.
• fireWaiters size==0 vs registerWaiter lost wakeup, double-resume / concurrent two-worker mount, CarrierLocal id-recycle window, Rust FFI panic/ABI safety: all verified correct.

Summary

Verdict: request changes — one confirmed Critical (C1) blocks merge.

C1 is an unbounded native + heap leak on the HTTP/REST path: the job-rotation recycle pool is replenished only at pool halt, so every wait_wal_table/sleep suspend (and each wake re-park) mints and pins a full HTTP handler set until shutdown. It directly contradicts the PR's own amortisation claims and is reachable by the workloads the feature targets. The remaining issues are non-blocking but several touch the design's stated correctness story (M3, M4, M6) and are worth addressing alongside C1.

Draft findings verified: ~11 confirmed (1 Critical, 7 Moderate, plus Minors), ~7 dropped as false positives/already-resolved. All confirmed findings are in-diff; the only out-of-diff-reachable concern (the IODispatcher sharing) was verified safe. Nice, unusually well-documented change otherwise.

[ideoma]

Pushed two commits on top of the review:

• 7e558d7056 "fix jobs leak" — C1
• 5ea7e6145b "make job workerId a lazy pull ... Address other comments" — M4, M6, M7, M5 (mitigated), M8 (recycle test)

Per-finding status below.

C1 — Unbounded selector + HTTP-handler leak on the REST path — fixed (7e558d7056 + 5ea7e6145b)

recycleJobList() now runs during RUNNING on the spent handoff cont, not only at isDone()/halt; mintNextGen() moved to after the handoff loop, and HttpRequestJob.run() re-acquires a selector lazily when selector == null. The recycle pool now refills in steady state, so the amortisation claim in HttpServer/DESIGN_NOTES holds.

Regression test JobRotationTest.testStatefulCloneReusedAcrossReparksDuringRunning asserts clonesCreated <= 5 and clonesRecycled >= reparks/2 across 60 re-parks; fails on master, passes now.

M2 — Shutdown not fenced behind a ContinuationQueue drain — decision (already covered)

Accepting the ContinuationQueue residual as process-exit-only (OS reclaims) rather than adding a timed drain that the shutdown path otherwise avoids. Same shutdown-ordering family was already raised and characterized as "Bounded, reclaimed via freeOnExit" in #issuecomment-4603508329, and the ordering itself was answered "by design" in #issuecomment-4601174665 (C1).

M3 — TimerShards.shutdown()/halt() ordering can skip the drain — will not fix (already answered)

Same finding as RaphDal's C1 (#issuecomment-4508535952), already answered "by design" in #issuecomment-4601174665 (C1): shutdown() must run while pools are RUNNING so entry.shutdown() -> scheduleResume() lands on live workers; by close() time pools are halted, so making shutdown() "always drain" would scheduleResume() into dead pools. The close()-without-signalClose() path only hits partial-init (no entries parked) and test harnesses (assertMemoryLeak covers them). No code change.

M4 — Job.run() Javadoc contradicted the implementation on workerId — fixed (5ea7e6145b)

run(int, RunStatus) replaced by run(WorkerContext). Javadoc now documents carrierId() as the pool-local index in [0, poolWorkerCount), not JVM-unique and not CarrierIdentity.current(); the false uniqueness claim is removed.

M5 — Latent cross-carrier workerId aliasing — mitigated (5ea7e6145b)

The lazy-pull removes the per-iteration currentWorkerId capture, so carrierId() now reads the live carrier at use-site rather than a frozen value; an alias requires a future job to deliberately cache it across a suspend. Will add a remount-safety rule to DESIGN_NOTES.

M6 — TxnWaiter.cancel() contract said CAS but it is an unconditional write — fixed (5ea7e6145b)

tryCancel renamed to cancel(); Javadoc + DESIGN_NOTES now describe it as an unconditional terminal write (the FIRED-clobber is unobservable, it is the last op on the unwinding tail), with the CAS-and-parkRefused discipline kept in abortContinuation().

M7 — Per-iteration FFI downcall in the hot worker loop — fixed (5ea7e6145b)

The hot loop passes one reusable WorkerContext; carrierId() is pulled lazily. Legacy pools (continuationQueue == null) return the stable workerId field with no CarrierLocal/FFI read; only cont-aware pools read WORKER_ID, on demand.

M8 — Test coverage gaps — partial

• recycle-on-completion: covered by the new JobRotationTest above.
• suspend-refused fallback (parkRefused phantom-dequeue): already answered won't-test in #issuecomment-4601174665 (m4 coverage) — "Not reachable without a white-box context; Matches the Won't fix". Every client mounts a continuation so suspend() always succeeds on the reachable path; no black-box seam forces suspend() == false.
• ServerMainSleepTest anti-serialization metric: still unasserted; will add the assertion or soften the comment.

Minor — ColumnTypeConverter has no registered CarrierLocal cleaner — already answered

Already answered "Not a real leak" in #issuecomment-4601174665 (M1), replying to RaphDal's M1 (#issuecomment-4508535952): every use does .of(fd) then detachFdClose() within the same convertColumn() call, so nothing native outlives the call and there is nothing for a cleaner to free. Residual is only the DESIGN_NOTES wording that names it as the example — will correct the note.

[ideoma]

Code review — OSS #7031 (level 3, tandem with questdb-enterprise#1006)

Mission-critical adversarial review of the worker-continuation framework. 11 review passes across the new primitives (CarrierLocal + Rust FFI, the continuation worker loop, TimerShards, TxnWaiter, the reader-pool-supervisor relocation), the broad mechanical sweep, and the tests. ENT-side findings are posted on #1006. ~30 draft findings → 2 false positives removed.

The novel core was traced to ground truth and is sound: the C2-hoist defense, carrier-id recycle ordering, job-clone rotation (no double-close), the timer state machine (single-CAS, no double-fire), and the WAL-waiter register/fire happens-before all verified.

Critical

None. No data race, lost-wakeup-forever, use-after-free, double-free, native leak, or unclosed LOG chain found in production code.

Moderate

1. Per-iteration FFI cost on the shared reduce pool — core/src/main/java/io/questdb/cairo/sql/async/PageFrameReduceJob.java:180 (also UnorderedPageFrameReduceJob, GroupByVectorAggregateJob, GroupByMergeShardJob, GroupByLongTopKJob) — in-diff.
run() calls workerContext.carrierId() = WORKER_ID.get() = a CarrierLocal.get() (FFI critical downcall ~4 ns + open-addressed map probe + volatile rows[] read; measured ~6 ns single-thread, ~8 ns under 8 workers). Before the PR the pool-local worker id arrived as a free int argument. It is pulled once per run() (not per row — verified), but the shared reduce pool spins run() continuously, so this is net-new per-loop-iteration cost on the busiest pool, paid even on idle spins. The "lazy workerId" commit already moved the pull out of the driver loop; this is the residual.
Fix: stamp the pool-local id as a plain field on the mounted WorkerContinuation/context at mount/remount (re-stamped on migration) so jobs read a field instead of FFI+probe each iteration. Reduce jobs never suspend mid-run(), so a once-per-generation cached field suffices.

2. Reader-pool supervisor now belongs to SqlExecutionContext — core/src/main/java/io/questdb/griffin/engine/QueryProgress.java / SqlExecutionContextImpl.java — in-diff. (Downgraded from Moderate to a design note after discussion with the author — captured here for the record.)
Verified the single-thread-per-context assumption holds in every production path (parallel reduce/group-by/filter/sample-by workers never touch the context supervisor; the IODispatcher single-thread handoff serializes carriers across a parked stack; mat-view/view-compiler jobs each own a fresh per-generation context; ENT replication/backup/cold-storage are single-threaded). Not a live bug.

Design intent (author): the previous ThreadLocal supervisor was never about thread safety — it was a coding convenience. Being per-thread it actually hid leaked readers when a query opened readers across multiple threads: each thread's supervisor saw only its own borrows, so a reader opened on one thread and not returned was invisible to the leak check. The single-threaded reader-open assumption was always there; moving the supervisor onto the execution context just applies that same assumption differently — readers belong to the context, and leak detection is now per-context instead of per-thread. SqlExecutionContext is non-thread-safe by design across the board, so the supervisor is consistent with the rest of it, and it can be made thread-safe later if a use case ever needs it.

Note on guarding it: a -ea-only check would have to detect concurrency, not thread identity — the context legitimately changes carrier across a continuation park/resume (different thread, never concurrent), so a same-thread assert would false-fire on every resume. So if anything, a debug "busy" re-entrancy flag, not a thread-id check; optional given production honors the contract.

(Positive: the PR also fixes a real pre-existing reader-leak-on-cleanup bug — the old Misc.freeObjListAndClear skipped every other leaked reader due to mid-walk resize; the new back-to-front detach-then-free loop is correct.)

3. Unreachable dead code in Worker.loopBody — the else at core/src/main/java/io/questdb/mp/Worker.java:417-422 — in-diff.
The code is correct (a handoff-surrendering cont has no waiter bound, is recycled at Worker.java:287, and is never remounted — no leak, no double-resume). But WorkerContinuation.suspend() (WorkerContinuation.java:115) called at Worker.java:399 can never return true: a cont that yields there is always recycled and dropped by the outer driver, and the only remount path (scheduleResume, WorkerContinuation.java:192) carries deep-park conts that resume at Worker.java:367, not at Worker.java:399. So the else (the "resumed after a successful yield" branch) is dead, and its comment at Worker.java:423-426 ("execution lands on the else branch ... only when this cont is later remounted") describes a remount that cannot happen — contradicting the accurate comment at Worker.java:282 ("it is spent").
Fix applied on the branch: dropped the dead else (collapsed to a bare if), replaced the misleading comment, and documented the full control flow with two sequence diagrams in DESIGN_NOTES.md ("Worker loop control flow: handoff suspend vs deep park").

Minor

• wait_wal_table not flagged non-deterministic — WaitWalFunction.java:185 (isRuntimeConstant, default isNonDeterministic()==false) — in-diff. Result depends on live WAL state but won't be rejected by the !allowNonDeterministicFunctions() guard (e.g. materialized-view definitions), where a parked wait could stall refresh. Override isNonDeterministic() to return true.
• ParallelCsvFileImporter.java:1387 — in-diff. Passes CarrierIdentity.current() (global carrier id, or −1) where a pool-local worker id (0) was expected. Harmless today (CopyImportJob ignores the id), but a latent namespace trap. Pass 0 or document that the import job ignores the worker id.
• Dead/misleading code. AsyncLogRecord.getDejaVu() (:508-516) Javadoc claims the set "is cleared by the writer thread" — it has zero callers and is never cleared (pre-existing, not a regression). TimerShardsTest.java:117-119 has an empty 1000-iteration loop with a misleading comment. ServerMainSleepTest.java:537,584 comments say "12 threads × 25 = 300" / "8 buckets" but the test runs 64×8=512. Clean up while touching the files.
• Stale design docs. CARRIER_LOCAL.md "Files" section still points to src/carrier.rs (now src/ffi/carrier.rs) and claims "≥2 current() calls per log line" (shipped AbstractLogRecord resolves it once per chain, not per $() segment). DESIGN_NOTES.md lists O3PartitionPurgeJob as "Deferred" though it was converted, and describes registerWaiter as "enqueues a fresh one" each cycle though the waiter is registered once and reused.
• DelayHeap sub-millisecond deadlines round up to ~1 ms via Object.wait(millis, nanos) (never early; bounded by the 1 s wake interval). Add a one-line comment noting the rounding floor.
• Test coverage gaps (non-blocking). No deterministic unit test for the continuation-level resume-before-suspend lost-wakeup (WorkerContinuationTest always parks before scheduleResume()); covered probabilistically by the WAL fuzz suite. The carrier-pinned legacy-polling fallback (suspend()==false) is also untested.

Downgraded (false positives)

• ACL gap on wait_wal_table — dismissed. wait_wal_table/sleep are read-only and consistent with existing ungated introspection peers (table_partitions, wal_transactions, table_columns all resolve a table name and expose state with no per-table authorize*; only HydrateTableMetadata authorizes, via authorizeSystemAdmin, because it triggers a privileged system op). No new ACL surface; the tandem state-mutation rule doesn't apply since nothing mutates state.
• CarrierLocal.set(null) semantics divergence — dismissed as a live bug. Differs from the deleted io.questdb.std.ThreadLocal (remembers null instead of re-initializing), but the only production set(null) callsite (MatViewGraph.java:281) always sets non-null before the next get(). Worth a one-line Javadoc note; no behavioral break.

Summary

• Verdict: approve with nits. Address the Moderate perf item and the supervisor assert as follow-ups; neither blocks merge.
• Unusually well-defended change — DESIGN_NOTES.md/CARRIER_LOCAL.md pre-empt most hazards, the core invariants are verified sound, and the PR even fixes a latent reader-leak bug along the way.
• Tradeoffs to surface: the carrierId() FFI cost on the shared reduce pool (Moderate 1). The reader-pool-supervisor move (Ignore unsupported field types #2) is an intentional, consistent design choice — the context is single-threaded by contract, and the supervisor now follows that contract — not a tradeoff to mitigate.
• Counts: ~30 draft findings; 2 false positives removed, no Critical confirmed, 2 Moderate (+1 downgraded to a design note), ~9 Minor. All confirmed findings are in-diff or in-diff-adjacent; the cross-context pass cleared the high-blast-radius out-of-diff callsites (parallel workers, ENT replication/backup, mat-view refresh, all 10 JVM launch paths).

_{🤖 Level-3 adversarial review via Claude Code (/review-pr). Each finding verified against source; false positives moved to Downgraded.}

[ideoma]

/azp run macwin

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[bluestreak01]

Approving after an independent level-3 review pass (full continuation runtime, carrier-identity FFI, worker loop, both SQL functions, WAL notification path, Job rotation, reader-pool supervisor relocation, and HTTP/PG wiring).

Validation: compiled the core module clean across all 272 changed files; ran the focused new suites — DelayHeapTest, TimerShardsTest, TimerContTest, WorkerContinuationTest, CarrierLocalTest, JobRotationTest, SeqTxnTrackerTest, SleepFunctionFactoryTest, WaitWalTableFunctionFactoryTest, ReaderLeakTest, ServerMainSleepTest, ServerMainWaitWalTableTest — all green, including timeout/cancel/shutdown-while-parked scenarios.

No blocking defect found in the current head (60e139d). The two Criticals from the earlier review were written against an older revision and are now resolved: workerId aliasing is fixed by the carrierId lazy-pull redesign (the <=N-concurrent-distinct-id invariant is preserved), and the timer-drain ordering is documented + asserted.

Non-blocking follow-ups worth addressing (none gate the merge):

• TimerShards.shutdown() early-returns on !running; make it always drain and mark halt() synchronized, so a close() not preceded by signalClose() cannot skip the drain.
• Register an assignThreadLocalCleaner for ColumnTypeConverter's native-backed CarrierLocal slots (the case DESIGN_NOTES.md already calls out). Pre-existing, equal severity to master, but worth fixing here.
• Reader-pool supervisor on SqlExecutionContext relies on the unenforced single-thread-per-context invariant; consider an ownership assert.
• Surface the bundled execute() retry cap (infinite-retry -> throw after 1000) and the standing tradeoffs (always-on TimerShards thread per shard, per-suspend native stack chunk, ~1-3 ns per CarrierLocal.get()) in the PR description.
• Minor: add WaitWalTableSeqTxnFunctionFactory to function_list.txt for consistency; non-destructive check instead of the destructive assert recyclePool.pop() == null in bind(); add Performance/SQL/New feature/rust labels.

[bluestreak01]

/azp run macwin

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[questdb-butler]

⚠️ Enterprise CI Failed

The enterprise test suite failed for this PR.

Build: View Details
Tested Commit: 3f5be07bce304d43b6a1c7ab331c485e156398a6

Please investigate the failure before merging.

[mtopolnik]

[PR Coverage check]

😍 pass : 1457 / 1695 (85.96%)

file detail

	path	covered line	new line	coverage
🔵	io/questdb/griffin/CharacterStore.java	0	3	00.00%
🔵	io/questdb/griffin/engine/groupby/GroupByUtf8Sink.java	0	3	00.00%
🔵	io/questdb/cutlass/line/tcp/ArrayBinaryFormatParser.java	0	1	00.00%
🔵	io/questdb/std/str/FlyweightDirectUtf16Sink.java	0	3	00.00%
🔵	io/questdb/griffin/engine/functions/bin/Base64DecodeFunctionFactory.java	0	3	00.00%
🔵	io/questdb/std/BytecodeAssembler.java	0	3	00.00%
🔵	io/questdb/cutlass/pgwire/PGCleartextPasswordAuthenticator.java	0	3	00.00%
🔵	io/questdb/griffin/engine/groupby/GroupByCharSink.java	0	3	00.00%
🔵	io/questdb/std/str/Utf8StringSinkList.java	0	3	00.00%
🔵	io/questdb/log/NullLogRecord.java	0	3	00.00%
🔵	io/questdb/network/TlsSessionInitFailedException.java	0	1	00.00%
🔵	io/questdb/std/str/StdoutSink.java	0	3	00.00%
🔵	io/questdb/std/str/DirectUtf16Sink.java	0	3	00.00%
🔵	io/questdb/cairo/LogRecordSinkAdapter.java	0	3	00.00%
🔵	io/questdb/cutlass/line/udp/LineUdpLexer.java	0	3	00.00%
🔵	io/questdb/griffin/engine/ops/InsertAsSelectOperationImpl.java	0	7	00.00%
🔵	io/questdb/std/str/DirectUtf8StringList.java	0	3	00.00%
🔵	io/questdb/cutlass/http/client/HttpClient.java	0	3	00.00%
🔵	io/questdb/griffin/engine/groupby/GroupByLongTopKJob.java	0	1	00.00%
🔵	io/questdb/log/LogFactory.java	1	4	25.00%
🔵	io/questdb/std/ConcurrentLongHashMap.java	1	4	25.00%
🔵	io/questdb/std/ConcurrentIntHashMap.java	1	4	25.00%
🔵	io/questdb/cutlass/http/HttpResponseSink.java	3	9	33.33%
🔵	io/questdb/cairo/DefaultCairoConfiguration.java	1	2	50.00%
🔵	io/questdb/griffin/engine/groupby/vect/GroupByVectorAggregateJob.java	1	2	50.00%
🔵	io/questdb/std/str/Digest.java	8	16	50.00%
🔵	io/questdb/std/str/Path.java	6	9	66.67%
🔵	io/questdb/cutlass/text/ParallelCsvFileImporter.java	2	3	66.67%
🔵	io/questdb/mp/CarrierIdentity.java	32	43	74.42%
🔵	io/questdb/mp/continuation/TimerShards.java	77	93	82.80%
🔵	io/questdb/cutlass/pgwire/PGServer.java	22	26	84.62%
🔵	io/questdb/mp/continuation/TimerCont.java	22	26	84.62%
🔵	io/questdb/std/CarrierLocal.java	203	240	84.58%
🔵	io/questdb/cairo/CairoEngine.java	28	33	84.85%
🔵	io/questdb/mp/Worker.java	106	126	84.13%
🔵	io/questdb/log/AsyncLogRecord.java	165	190	86.84%
🔵	io/questdb/cairo/pool/ReaderPool.java	7	8	87.50%
🔵	io/questdb/log/AbstractLogRecord.java	8	9	88.89%
🔵	io/questdb/mp/continuation/TxnWaiter.java	46	52	88.46%
🔵	io/questdb/mp/continuation/ContinuationQueue.java	16	18	88.89%
🔵	io/questdb/griffin/engine/functions/date/SleepFunctionFactory.java	64	72	88.89%
🔵	io/questdb/mp/WorkerPool.java	22	24	91.67%
🔵	io/questdb/cairo/wal/seq/SeqTxnTracker.java	52	56	92.86%
🔵	io/questdb/mp/continuation/WorkerContinuation.java	32	34	94.12%
🔵	io/questdb/griffin/engine/functions/table/WaitWalFunction.java	65	69	94.20%
🔵	io/questdb/cutlass/Services.java	17	18	94.44%
🔵	io/questdb/cutlass/http/HttpServer.java	93	94	98.94%
🔵	io/questdb/cutlass/http/ex/NotEnoughLinesException.java	1	1	100.00%
🔵	io/questdb/std/datetime/nanotime/NanosFormatCompiler.java	1	1	100.00%
🔵	io/questdb/std/Misc.java	5	5	100.00%
🔵	io/questdb/cairo/view/ViewStateStoreImpl.java	1	1	100.00%
🔵	io/questdb/cairo/pool/TracingResourcePoolSupervisor.java	5	5	100.00%
🔵	io/questdb/cutlass/text/CopyImportJob.java	6	6	100.00%
🔵	io/questdb/mp/continuation/DelayHeap.java	20	20	100.00%
🔵	io/questdb/cutlass/line/tcp/DecimalBinaryFormatParser.java	1	1	100.00%
🔵	io/questdb/griffin/engine/functions/table/WaitWalTableSeqTxnFunctionFactory.java	9	9	100.00%
🔵	io/questdb/cutlass/qwp/server/egress/QwpRowExceedsBufferException.java	2	2	100.00%
🔵	io/questdb/cutlass/line/tcp/auth/EllipticCurveAuthenticator.java	1	1	100.00%
🔵	io/questdb/cutlass/json/JsonException.java	1	1	100.00%
🔵	io/questdb/cutlass/qwp/protocol/QwpParseException.java	1	1	100.00%
🔵	io/questdb/griffin/engine/PerWorkerLocks.java	2	2	100.00%
🔵	io/questdb/cairo/ImplicitCastException.java	1	1	100.00%
🔵	io/questdb/mp/ConcurrentQueue.java	1	1	100.00%
🔵	io/questdb/std/Decimal256.java	3	3	100.00%
🔵	io/questdb/griffin/engine/table/SortedSymbolIndexRowCursorFactory.java	1	1	100.00%
🔵	io/questdb/mp/WorkerPoolConfiguration.java	1	1	100.00%
🔵	io/questdb/cairo/sql/async/PageFrameReduceJob.java	9	9	100.00%
🔵	io/questdb/cutlass/pgwire/PGMessageProcessingException.java	1	1	100.00%
🔵	io/questdb/griffin/engine/QueryProgress.java	12	12	100.00%
🔵	io/questdb/cairo/wal/WalApplySqlExecutionContext.java	2	2	100.00%
🔵	io/questdb/cairo/CommitFailedException.java	1	1	100.00%
🔵	io/questdb/griffin/model/IntervalUtils.java	9	9	100.00%
🔵	io/questdb/cairo/O3PartitionJob.java	2	2	100.00%
🔵	io/questdb/griffin/SqlExecutionContextImpl.java	4	4	100.00%
🔵	io/questdb/std/Decimal128.java	3	3	100.00%
🔵	io/questdb/std/NumericException.java	1	1	100.00%
🔵	io/questdb/network/NoSpaceLeftInResponseBufferException.java	1	1	100.00%
🔵	io/questdb/cairo/O3PartitionPurgeJob.java	11	11	100.00%
🔵	io/questdb/cutlass/line/tcp/LineTcpLegacyWriterJob.java	1	1	100.00%
🔵	io/questdb/PropertyKey.java	2	2	100.00%
🔵	io/questdb/std/datetime/microtime/MicrosFormatFactory.java	1	1	100.00%
🔵	io/questdb/cairo/view/ViewCompilerJob.java	12	12	100.00%
🔵	io/questdb/cutlass/auth/AuthUtils.java	2	2	100.00%
🔵	io/questdb/griffin/engine/functions/date/ToStrTimestampFunctionFactory.java	1	1	100.00%
🔵	io/questdb/cairo/pool/AbstractMultiTenantPool.java	6	6	100.00%
🔵	io/questdb/griffin/SqlExecutionContext.java	4	4	100.00%
🔵	io/questdb/cairo/ColumnTypeConverter.java	6	6	100.00%
🔵	io/questdb/jit/CompiledCountOnlyFilter.java	1	1	100.00%
🔵	io/questdb/cairo/CairoConfigurationWrapper.java	2	2	100.00%
🔵	io/questdb/cutlass/line/tcp/TableStructureAdapter.java	1	1	100.00%
🔵	io/questdb/cairo/mv/MatViewGraph.java	3	3	100.00%
🔵	io/questdb/cutlass/pgwire/PGConnectionContext.java	3	3	100.00%
🔵	io/questdb/cutlass/text/TextImportException.java	1	1	100.00%
🔵	io/questdb/mp/WorkerLifecycle.java	2	2	100.00%
🔵	io/questdb/cutlass/qwp/server/QwpIngressHttpProcessor.java	4	4	100.00%
🔵	io/questdb/cutlass/http/HttpCookieHandlerImpl.java	2	2	100.00%
🔵	io/questdb/std/str/Utf8s.java	1	1	100.00%
🔵	io/questdb/cairo/wal/WalTxnDetails.java	1	1	100.00%
🔵	io/questdb/mp/AbstractQueueConsumerJob.java	1	1	100.00%
🔵	io/questdb/cairo/sql/async/UnorderedPageFrameReduceJob.java	7	7	100.00%
🔵	io/questdb/griffin/engine/functions/math/LeastNumericFunctionFactory.java	1	1	100.00%
🔵	io/questdb/cutlass/qwp/server/QwpUdpReceiver.java	1	1	100.00%
🔵	io/questdb/PropServerConfiguration.java	5	5	100.00%
🔵	io/questdb/jit/CompiledFilter.java	1	1	100.00%
🔵	io/questdb/griffin/engine/functions/date/GenerateSeriesTimestampStringRecordCursorFactory.java	1	1	100.00%
🔵	io/questdb/griffin/engine/table/GroupByShardingContext.java	2	2	100.00%
🔵	io/questdb/griffin/engine/groupby/GroupByMergeShardJob.java	2	2	100.00%
🔵	io/questdb/std/datetime/nanotime/NanosFormatFactory.java	1	1	100.00%
🔵	io/questdb/griffin/engine/LimitOverflowException.java	1	1	100.00%
🔵	io/questdb/cairo/pool/ex/EntryLockedException.java	1	1	100.00%
🔵	io/questdb/cutlass/parquet/CopyExportRequestJob.java	6	6	100.00%
🔵	io/questdb/cairo/EntryUnavailableException.java	1	1	100.00%
🔵	io/questdb/network/IOContextFactoryImpl.java	2	2	100.00%
🔵	io/questdb/griffin/engine/functions/conditional/CaseCommon.java	2	2	100.00%
🔵	io/questdb/griffin/engine/functions/rnd/SharedRandom.java	2	2	100.00%
🔵	qdbr/src/ffi/carrier.rs	25	25	100.00%
🔵	io/questdb/griffin/engine/functions/date/ToStrDateFunctionFactory.java	1	1	100.00%
🔵	io/questdb/griffin/engine/groupby/vect/SumLong256VectorAggregateFunction.java	1	1	100.00%
🔵	io/questdb/cairo/wal/seq/TableTransactionLogV2.java	1	1	100.00%
🔵	io/questdb/log/LogRecordUtf8Sink.java	3	3	100.00%
🔵	io/questdb/cairo/mv/MatViewRefreshJob.java	13	13	100.00%
🔵	io/questdb/mp/Job.java	14	14	100.00%
🔵	io/questdb/std/Decimal64.java	3	3	100.00%
🔵	io/questdb/mp/WorkerPoolUtils.java	4	4	100.00%
🔵	io/questdb/cairo/mv/MatViewStateStoreImpl.java	2	2	100.00%
🔵	io/questdb/cutlass/text/TextException.java	1	1	100.00%
🔵	io/questdb/std/str/StringSink.java	4	4	100.00%
🔵	io/questdb/std/ConcurrentHashMap.java	4	4	100.00%
🔵	io/questdb/ServerMain.java	4	4	100.00%
🔵	io/questdb/griffin/engine/functions/math/GreatestNumericFunctionFactory.java	1	1	100.00%
🔵	io/questdb/griffin/SqlTimeoutException.java	1	1	100.00%
🔵	io/questdb/std/datetime/millitime/DateFormatCompiler.java	1	1	100.00%
🔵	io/questdb/griffin/engine/functions/table/WaitWalTableFunctionFactory.java	4	4	100.00%
🔵	io/questdb/cutlass/parquet/CopyExportException.java	1	1	100.00%
🔵	io/questdb/cairo/wal/ApplyWal2TableJob.java	9	9	100.00%
🔵	io/questdb/griffin/SqlException.java	1	1	100.00%
🔵	io/questdb/std/str/FileNameExtractorUtf8Sequence.java	1	1	100.00%
🔵	io/questdb/network/NetworkError.java	1	1	100.00%
🔵	io/questdb/std/str/DirectUtf8Sink.java	3	3	100.00%
🔵	io/questdb/cutlass/http/HttpException.java	1	1	100.00%
🔵	io/questdb/cairo/sql/TableReferenceOutOfDateException.java	5	5	100.00%
🔵	io/questdb/griffin/QueryRegistry.java	1	1	100.00%
🔵	io/questdb/std/SecurePath.java	2	2	100.00%
🔵	io/questdb/cutlass/line/tcp/LineProtocolException.java	1	1	100.00%
🔵	io/questdb/std/Numbers.java	3	3	100.00%
🔵	io/questdb/mp/CountedConcurrentQueue.java	12	12	100.00%
🔵	io/questdb/cairo/wal/seq/TableTransactionLog.java	2	2	100.00%
🔵	io/questdb/cairo/mv/MatViewRefreshSqlExecutionContext.java	4	4	100.00%
🔵	io/questdb/std/str/Utf8StringSink.java	3	3	100.00%
🔵	io/questdb/griffin/engine/functions/eq/EqLong256StrFunctionFactory.java	1	1	100.00%
🔵	io/questdb/cairo/wal/seq/TableTransactionLogV1.java	1	1	100.00%
🔵	io/questdb/std/str/SingleCharCharSequence.java	1	1	100.00%
🔵	io/questdb/std/datetime/microtime/MicrosFormatCompiler.java	1	1	100.00%
🔵	io/questdb/cairo/CairoException.java	1	1	100.00%
🔵	io/questdb/griffin/SqlUtil.java	1	1	100.00%
🔵	io/questdb/std/datetime/millitime/DateFormatFactory.java	1	1	100.00%