feat(qwp): recycle the symbol dictionary at a threshold to lift the cardinality cap - #91
Draft
jovfer wants to merge 30 commits into
Draft
feat(qwp): recycle the symbol dictionary at a threshold to lift the cardinality cap#91jovfer wants to merge 30 commits into
jovfer wants to merge 30 commits into
Conversation
Pull the lock/quarantine engine-construction block out of build() into LineSenderBuilder.constructEngineOnSlotLocked()/constructEngineOnSlot(), and expose a QwpWebSocketSender.EngineRebuildFactory seam that build() installs on the connected sender once connect() succeeds. Pure refactor, zero behavior change: build() keeps its wide logical-lock scope spanning the connect loop; the standalone constructEngineOnSlot() acquires the lock only around construction, for a later symbol-dictionary epoch rebuild to reuse the identical construct/quarantine code path.
constructEngineOnSlotLocked() now returns a small ConstructedEngine result (engine + whether construction itself quarantined the slot) instead of a bare CursorSendEngine. build() seeds its own quarantined local from that verdict, restoring the pre-refactor invariant that a construction-time quarantine counts toward the one quarantine per build() attempt the connect loop's retry guard allows. Without this, a construction-time quarantine followed by an UnreplayableSlotException from connect() would take a second quarantineTornSlot pass instead of the original close-and-rethrow. constructEngineOnSlot(), the public factory entry Task 5 consumes, keeps its 8-arg signature and CursorSendEngine return type: it just unwraps ConstructedEngine.engine and discards the quarantined verdict, since the recycle path latches terminal on connect failure rather than quarantining.
Adds three connect-string keys for the upcoming QWP symbol-dictionary recycle feature: symbol_dict_reset (on/off, default on), symbol_dict_reset_threshold (distinct-symbol count that triggers a recycle, default 100_000, bounded by QwpConstants.MAX_SYMBOL_DICTIONARY_SIZE), and symbol_dict_reset_max_wait_millis (upper bound on how long a triggered recycle waits for an opportunistic window before forcing, default 30_000, 0 means opportunistic-only). This is config plumbing only: the three values land on QwpWebSocketSender as resetEnabled/resetThresholdSymbols/resetMaxWaitMillis fields with @testonly getters, following the catch_up_cap_gap_min_escalation_window_millis knob end-to-end (ConfigSchema registry, builder methods with WS-transport guards, both connect-string parse paths, wsConfigSnapshotForTest). Actual recycle behavior is a follow-up task.
Adds Sender.resetSymbolDictionary(), an advisory request to start a fresh symbol-dictionary epoch (default no-op; QwpWebSocketSender overrides it). QwpWebSocketSender.armIfEligible() re-evaluates arming at the tail of resetTableBuffersAfterFlush() -- the shared exit point for the plain flush, split flush, and close-path callers -- so the recycle arms once symbol_dict_reset is enabled and either the global dictionary reaches symbol_dict_reset_threshold distinct entries or a caller requested a manual reset. Arming deliberately ignores deltaDictEnabled: a sender degraded to full self-sufficient frames still benefits from bounding dictionary growth, and a manual request is honoured regardless of mode. Covered by SymbolDictRecycleArmingTest: threshold crossing, symbol_dict_reset=off never arming, the manual advisory API (both immediate and mid-batch-deferred arming), the split-flush path sharing the same arming tail, arming in full-dict (degraded) mode, and the no-op default on a non-WebSocket sender.
testArmsInFullDictMode previously substituted the manual resetSymbolDictionary() advisory request for crossing symbol_dict_reset_threshold, which never touches globalSymbolDictionary and so left a future deltaDictEnabled-conditioned regression in threshold-based arming undetected. Rewrite it to construct the sender through the widest connect(List<Endpoint>, ...) overload, which accepts both a custom symbolDictResetThresholdSymbols and the fault-injecting CursorSendEngine, and genuinely cross the threshold (registering a, b, c) while the sender stays degraded to full self-sufficient frames. No manual reset call remains in the test.
- rollFsnEpochBase now throws IllegalStateException when cursorSendLoop is non-null: the loop's externalFsnBase is a construction-time snapshot, never updated on a live loop, so rolling with a loop attached would silently desync sender-level FSN accessors from loop-level FSN emission. - testPreRollTargetAnswersTrueAfterRoll rolled the same already-connected sender that produced fsn1, so its raw engine watermark never reset and the pre-fix comparison (ackedFsn() == fsn1 >= fsn1) was also true -- the test could not fail. Rebuilt around a fresh rolled sender/engine via the existing createRolledSender helper, matching tests 3/5/6. Tests 2, 4, and 7 also rolled an already-connected sender (now rejected by the new guard) and are rewritten the same way.
Implement the table() barrier hook and recycleForDictReset() swap: once the symbol-dictionary recycle is armed (Task 3) and the ring is proven drained, table() tears the cursor I/O loop and engine down, rolls the FSN epoch base (Task 4), replaces the producer's symbol dictionary, and rebuilds the engine via the Task 1 EngineRebuildFactory before reconnecting -- all synchronously inside a single call. A failed rebuild latches recycleFailure as a terminal state: every frame existed before the swap was already proven acked, so no data is at risk, but the sender that observed the torn-down engine/loop refuses further use. checkRecycleFailure() covers table() and the flush-family entry points (flush, flushAndGetSequence, drain, awaitAckedFsn) -- deliberately not close(), which must still be able to tear down a latched sender.
Fix round 1 from review: - maybeRecycleForDictReset() now refuses before any teardown when engineRebuildFactory is null (every public connect() overload leaves it unset -- only Sender.build() installs one) or the cursor engine isn't owned by this sender (setCursorEngine(engine, false)'s contract). Without this, a connect()-built sender with the default-on recycle feature armed (via resetSymbolDictionary() or a threshold crossing) would reach step 6, NPE against the null factory, and latch itself terminal for no reason. - Step 5 also resets lastCommitBoundaryFsn -- it held a raw old-epoch FSN that does not survive the roll. - checkRecycleFailure() now also guards sendRow(), closing the fluent-chain corner where a caller continues .symbol(...).atNow() against a currentTableBuffer selected before the latch, without an intervening table() call. - Step 6 rewires cursorEngine.setSlotLockReleaseListener(...) on the rebuilt engine, restoring the pool early-wakeup notification that bypassing setCursorEngine had dropped. - testPostRecycleSlotContents now asserts the post-recycle slot directory listing against the exact expected fresh-state file set, replacing a bare Files.exists(...) check that proved nothing (the outgoing engine had a same-named file too). - New testConnectBuiltSenderNeverRecyclesWithoutFactory covers both ways such a sender can arm (manual request, threshold crossing): neither may recycle, throw, or stop the sender from working.
Adds SymbolDictRecycleMemoryModeTest, the sf_dir-omitted counterpart of SymbolDictRecycleTest: threshold-triggered recycle at an empty backlog, a content oracle proving the epoch boundary loses (and duplicates) nothing acked, and the same recycle under initial_connect_retry=async. All three pass unmodified against the existing recycle swap -- zero production changes -- confirming the factory's slotPath == null arm, CursorSendEngine's file-less close, and the table() barrier are already mode-agnostic.
Fills the maybeBlockForStarvedReset() stub: when a symbol-dict recycle is armed but the ring is not yet drained, opportunistically waits (parked, awaitAckedFsn-shaped) up to symbol_dict_reset_max_wait_millis for the outstanding acks before giving up for this armed window. resetMaxWaitMillis<=0 disables the wait entirely; at most one blocking wait runs per armed window (starvationWaitDoneThisArm); an open deferred-commit group is never waited on, since the server withholds its acks by design until the closing commit lands and this producer thread is the only one that could ever send that commit -- blocking there would just run out the clock every time. A timeout increments the new symbolDictResetStarvationTimeouts counter and leaves the recycle armed so a later drained table() call can still fire it. Verified the deferred-commit guard is load-bearing by temporarily removing it and confirming the test fails (blocks the full deadline instead of returning immediately) before restoring it.
Javadoc for symbol_dict_reset_max_wait_millis (Sender.java builder method, and the DEFAULT_.../field comments in QwpWebSocketSender.java) said the knob controls how long the recycle waits "before forcing the rebuild" and that 0 means "never forces". That is backwards: nothing is ever forced through. On timeout the wait gives up, increments the starvation counter, logs a warning, and stays armed -- the dictionary threshold is the only actual backstop. Rewrite all three to state the real policy: once the armed window exceeds the knob, the NEXT table(...) call may block the calling thread for up to the knob's value waiting for the backlog to drain; 0 disables blocking entirely. WARN log wording "re-arming opportunistically" -> "staying armed": the arm is never consumed on a timeout, so nothing re-arms. Test fixes in SymbolDictRecycleStarvationTest: - testTimeoutLogsAndReArms: the "second table() must not re-block" probe was placed after a row had been queued (pendingRowCount==1), so it short-circuited on maybeRecycleForDictReset()'s precondition guard before ever reaching the wait -- vacuously true regardless of starvationWaitDoneThisArm. Moved the probe earlier, to a table() call with pendingRowCount==0, so it actually exercises the "at most one blocking wait per armed window" guard. - testBlocksThenRecyclesWhenAcksArrive: raised maxWaitMillis from 400 to 700ms (keeping the 150ms release delay) so the full recycle path (I/O loop join, engine close, rebuild, fresh handshake) has real headroom under the elapsedMs<maxWaitMillis assertion instead of ~250ms. - Both testBlocksThenRecyclesWhenAcksArrive and testLatchedErrorDuringWaitThrows now join their helper thread (releaser/poisoner) in a finally block, so an assertion failure can no longer leak a non-daemon thread. Re-ran SymbolDictRecycleStarvationTest (5/5) and the full SymbolDictRecycle* battery (27/27), all green.
Adds SymbolDictRecycleOutageTest covering two interleavings between the symbol-dictionary recycle swap and events outside the producer's own control: a real connection outage on its own stream (recycle triggers while the pre-recycle I/O thread is mid-reconnect against a killed server; step 2's close() joins it, step 7's fresh connect recovers once the endpoint accepts again), and a sibling orphan drainer mid-drain (the recycle only tears down the foreground sender's own cursor engine/I/O loop, leaving a concurrently-gated BackgroundDrainer untouched and able to complete afterward). The third scenario from the task brief -- OrphanScanner.isCandidateOrphan rejecting an empty slot directory -- turned out to already be pinned by OrphanScannerTest#testIsCandidateOrphanDirect and #testEmptySlotDirIsNotAnOrphan, so no new test was added for it. Test-only change; no production code touched.
SymbolDictRecycleCatchUpSkipTest pins that recycleForDictReset()'s step 7 reconnect never pays for a delta-dictionary catch-up frame: the rebuilt engine sits on a freshly-emptied slot, so the new loop's sentDictCount mirror seeds from PersistedSymbolDict.recoveredSize() == 0 and setWireBaselineWithCatchUp's gate stays false for the whole first post-recycle connection. The core scenario chains the negative and positive observations in one test so the zero count is provably a property, not a handler blind spot: after the recycle sends zero zero-table frames and tiles ids from 0, the handler force-drops the connection, and the resulting UNPLANNED reconnect does catch up -- bounded to exactly the new epoch's symbols, never replaying the retired epoch's. A follow-up symbol then ships with a delta start above 0, pinning that resetSymbolDictStateForNewConnection() on the plain-reconnect path preserves sentMaxSymbolId rather than folding the recycle's baseline reset into itself. A second test repeats the zero-catch-up observation under initial_connect_retry=async, where step 7's reconnect funnels through ensureConnected()'s ASYNC arm and the handshake completes on the I/O thread. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
rev-t10 traced resetSymbolDictStateForNewConnection() to its single call site in ensureConnected(), which the unplanned I/O-thread reconnect (swapClient) never reaches -- so the old comment credited a function that does not even run on this path. sentMaxSymbolId survives the plain reconnect because nothing touches it there; only recycleForDictReset()'s step 5 ever zeroes the baseline. The class javadoc's failure-mode attribution carried the same imprecision and is reworded to match. The assertion itself was traced sound (symbolDeltaBaseline() -> encoder.beginMessage -> deltaStart) and is unchanged; comment-only diff. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
Adds SymbolDictRecycleCrashWindowsTest, pinning what a restarted sender recovers if the process crashes at each of four points around QwpWebSocketSender.recycleForDictReset()'s 8-step symbol-dictionary recycle swap: - (a) before step 2 (the barrier that starts the swap): the pre-recycle epoch's slot holds a fully-acked batch on disk. Recovery must find that residue, recognize it as already acked (nothing to replay) and resume the SAME dictionary rather than starting fresh. Constructed by closing fast against a server that never acks (so the fully-drained unlink never fires) and then stamping the ack watermark directly to declare the batch acked retroactively, mirroring DeltaDictRecoveryTest#writeAckWatermark. - (b) between step 3 (fully-drained close of the old engine) and step 6 (rebuild): the slot is empty. Constructed by driving a real recycle to completion and then closing immediately, before any flush touches the freshly-rebuilt engine -- finishClose treats "nothing published yet" as fully drained too, so this unlinks everything step 6 just created, leaving the same empty state step 3 alone would have left. - (c) after step 7 (reconnect), before the new epoch's first flush: the slot holds a freshly-rebuilt engine's own state files but no data. Constructed by snapshotting the rebuilt slot's bytes before closing (there is no supported way to release just the slot's OS flock without running finishClose's unlink), closing for real so nothing leaks, then restoring the snapshot on top of the vacated directory. - (d) an ordinary mid-operation crash one epoch into the post-recycle steady state, to prove the epoch swap does not corrupt normal backlog recovery. Uses the same close-fast-against-a-non-acking-server idiom as RecoveryReplayTest, but only after the recycle's fresh connection is established. Arms (b) and (c) both replay nothing and both look "empty" at first glance, but they are not the same recoverable state and a restarted engine can tell them apart: (c)'s slot carries a manifest with collapsed boundaries alongside a same-based, zero-frame active segment, which SegmentRing.recover()'s chain-building accepts as a RECOVERED (if empty) chain, while (b)'s slot carries no engine state at all and recovers as EMPTY. wasRecoveredFromDisk() is the pinned, distinguishing observable between the two, asserted explicitly instead of writing two assertion-for-assertion duplicate tests. Every arm's oracle: the recovered sender keeps ingesting; the symbols it and its predecessor registered are exactly and correctly reconstructable from the wire (each fresh server handler rebuilds the per-connection delta dictionary); and no data frame is delivered more than the at-least-once contract allows (each handler counts data frames so a spurious re-send would show up as an unexpected count). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
Applies review findings 1, 2, 3, 4, 6, 7, 8 and 10 from the task-11 review (findings 5, 9, 11 are deferred to the whole-branch review): - Arm (b) never actually pinned the "empty slot" disk image it exists to test -- it inferred emptiness from wasRecoveredFromDisk()==false on the successor instead of asserting the crashed sender's own slot dir. Adds an explicit listDir(slot) == [".lock", ".lock.pid"] assertion right after the crashed sender closes (symmetric to arm (c)'s pre-existing file-set assertion), plus a second one after the successor's own fully-drained close, proving the empty-slot state is stable rather than a one-shot coincidence. This also replaces arm (b)'s closing assertion, whose message previously credited the successor's "recovery" for cleaning up a stale manifest that the crashed sender's own close had already removed. - Arm (c) snapshotted the freshly-rebuilt slot before waiting for the manager worker's asynchronous hot-spare provisioning to settle, so a mid-provision snapshot could race-capture a zero-magic spare that recovery would hard-fail on. Adds a bounded poll (awaitExactFileSet) before the snapshot, reusing the same expected file list (now a shared FRESH_REBUILD_FILES constant) as the post-restore assertion so the two can never drift apart. - Arms (a) and (d) loosely asserted recoveredMaxSymbolId() >= 1, which a leaked epoch-0-plus-epoch-1 dictionary would also satisfy. Tightens both to the deterministic exact value (1L), keeping their explanatory messages as-is now that the assertion actually proves what the message claims. - Widens QwpWireTestUtils.tableCount to public (its sibling frame helpers already are) and deletes this suite's local copy plus its now-unnecessary justification javadoc. - Inlines the AckAllHandler bindings in arms (b) and (c) that were never read (dict()/dataFrameCount() were only meaningful on the "fresh" server's handler, not the "crashed" one). - Class javadoc: notes that arm (c)'s snapshot/restore does not cover the logical slot lock (lives outside the slot dir; sender.close() reclaims it, acquireLogical recreates it -- benign), and corrects three mechanism imprecisions the review traced: the "never published" fully-drained check lives in close(boolean), not finishClose; a fully-drained close does not remove .lock/.lock.pid; and arm (b)'s SegmentRing.recover() counterpart is the no-manifest fall-through to Recovery.empty(), not the manifest-present collapse branch. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
A sender that degrades to full self-sufficient frames after a symbol-dict persistence fault (disableDeltaDict) used to stay degraded for the rest of its life -- deltaDictEnabled was set once at engine construction and never re-evaluated. The symbol-dictionary recycle already rebuilds the cursor engine from scratch on every swap; make that rebuild also re-derive deltaDictEnabled from the fresh engine instead of carrying the old one's verdict forward. If the underlying fault has cleared, the next recycle heals the sender back into delta mode; if it has not, the fresh engine degrades again on its own first append, the same ordinary catchable LineSenderException as any other persistence fault -- a degrade, never a latched recycleFailure terminal state. Promote the epoch and starvation-timeout counters from @testonly accessors to permanent public API: getSymbolDictEpochForTest() becomes getSymbolDictEpoch(), and getSymbolDictResetStarvationTimeoutsForTest() becomes getSymbolDictResetStarvationTimeouts(). Both counters already existed; this only changes their visibility and documents their thread-safety contract (producer-thread-written, so a read from another thread is an eventually-consistent snapshot). Add a third counter, getSymbolDictResetsPerformed(), incremented alongside the epoch inside recycleForDictReset() -- the two move together today but are defined and incremented independently, since a future change could roll the epoch by some path other than a completed recycle swap. Migrate every existing call site of the two renamed getters (7 test files, 48 + 15 occurrences) to the new public names. SymbolDictRecycleHealingTest covers all of this: a fault-then-heal recycle proves deltaDictEnabled and wire framing both return to proper delta encoding (the second post-recycle frame's delta starts where the first left off and carries only the newly-added symbol, the shape only delta mode produces); a fault-persists variant proves the fresh engine degrades again without escaping as a raw error or latching the sender terminal; and a two-recycle run asserts the three metrics getters track correctly in lockstep while the starvation counter stays untouched. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
Applies task-12-review.md findings Q1-Q6.
Q1: the three symbol-dictionary-recycle metrics fields (symbolDictEpoch,
symbolDictResetsPerformed, symbolDictResetStarvationTimeouts) become
volatile -- they are permanent public API now, and their obvious reader
is a monitoring thread on some other thread, unlike every other
symbol-dictionary-recycle field they were modelled on. A plain,
non-volatile long gives a cross-thread reader no visibility guarantee at
all under the JMM (a polling loop can legally observe 0 forever), whereas
the javadoc claimed an "eventually-consistent snapshot". Reworded the
three javadocs to state the guarantee volatile actually gives: each read
sees the latest write the producer thread completed, with no atomicity
across the three counters -- a concurrent reader can see the epoch
already advanced while resets-performed still reflects the prior value,
even though the producer thread writes them on adjacent lines.
Q3: getSymbolDictResetStarvationTimeouts()'s javadoc named
recycleForDictReset() as the writer by importing getSymbolDictEpoch()'s
caveat sentence verbatim; the starvation counter is actually written in
maybeBlockForStarvedReset(). Named that method directly instead.
Q4: getSymbolDictResetsPerformed() said swaps this sender has
"completed" without stating that the counter advances at step 5, before
the engine rebuild (step 6) and reconnect (step 7) -- so a recycle that
later latches recycleFailure at step 6/7 still counts. Made that
explicit, mirroring the precision getSymbolDictEpoch() already had.
Q2: SymbolDictRecycleHealingTest's healing test asserted
isDeltaDictEnabledForTest() right after the recycle and attributed the
result to the healed facade, but the same assertion holds unconditionally
-- a fresh engine's construction never touches mmap, so it reports true
whether or not the facade was healed (the persistent-fault sibling test
proves this directly). Reworded the message to state what that assertion
actually pins, and added the discriminating check: re-assert after the
first post-recycle flush, the fresh engine's first real append -- a
still-armed facade would degrade it there, so staying true is real
evidence of healing.
Q5: AckAllHandler never reset its ack sequence per connection, unlike
CapturingAckHandler right below it in the same file. A rebuilt engine
restarts its raw FSNs at 0 after a recycle, so the stale, unreset
sequence could satisfy awaitAckedFsn with an ack for a frame that was
never published on the new connection -- a weaker gate than it looks,
even though the specific sequences in these tests happened not to
collide. Reset nextSeq on every new connection, matching the sibling
handler and existing repo precedent for connection-aware ack sequencing.
Q6: both fault-injection tests caught LineSenderException with a comment
claiming parity with MmapFaultDegradesTest's guard, but never actually
asserted the message MmapFaultDegradesTest checks
("failed to persist symbol dictionary before publish") -- so the catch
could equally have swallowed an unrelated connection-level exception.
Added the same message assertion to all three catch sites (two in the
persistent-fault test, one in the healing test) so the comment's claim
now holds.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
The symbol-dictionary cap error told callers to close the sender and build a new one, but gave no way to avoid hitting the cap in the first place. Append a sentence pointing at the automatic dictionary reset knobs (symbol_dict_reset, symbol_dict_reset_threshold) and the manual Sender.resetSymbolDictionary() escape hatch, so the error message matches the recycle feature this client now ships. Raise QwpConstants.MAX_SYMBOL_DICTIONARY_SIZE from 1_000_000 to 2_000_000 to mirror the server-side constant of the same name, which moved to 2_000_000 in questdb OSS commit 306062e243 (#7468). That commit is contained in release tag 10.0.0, the support floor for this client, so client <= server holds across the whole supported fleet. Update DeltaDictCeilingTest and GlobalSymbolDictionaryTest, which pinned the old 1,000,000 value and message text, to the new cap and add a case that drives the dictionary to the cap with automatic reset disabled (symbol_dict_reset=off) and a threshold configured at the cap, confirming the refusal still fires and still names the reset valve even on a sender that has it switched off. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
The new cap-reached-while-armed test never calls sender.flush(): the 2M fill goes through the raw GlobalSymbolDictionary test accessor, and the one Sender-routed call throws inside symbol() before a row completes. armIfEligible() only runs from the tail of a completed flush(), so it never had a code path available to flip isResetArmed() to true regardless of whether symbol_dict_reset was on or off -- the two assertFalse(ws.isResetArmed()) checks passed identically either way and proved nothing about the knob under test. Drop both assertions and note in the test's javadoc that arming semantics are out of scope here and are pinned instead by SymbolDictRecycleArmingTest.testArmsAtThreshold, which drives real rows through flush() and is the idiom that actually exercises armIfEligible(). The test's brief-mandated assertion -- the cap refusal still fires, with the new reset-valve message, when reset is disabled -- is untouched and remains the load-bearing check here. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
recycleForDictReset() no longer latches the sender terminal when the
recycle's reconnect fails. Step 7 moves out of the latching try block:
steps 1-6 still latch recycleFailure (a half-swapped sender genuinely
cannot make progress), but a failed ensureConnected() logs a warning and
rethrows to the triggering caller without latching. By step 7 the swap has
committed, so the sender is coherent - connected == false, loop and client
already closed and nulled by ensureConnected's own catch, the fresh engine
attached, the step-5 epoch and swap counters correctly left incremented -
and the ordinary sendRow() -> ensureConnected() path retries the connect,
and only the connect, on the next send. Nothing re-runs a teardown step,
and nothing can fire a second swap meanwhile: the fresh dictionary sits
below the threshold, manualResetRequested was consumed at step 5, and
maybeRecycleForDictReset requires connected.
This removes a default-configuration brick. A sender built with no
reconnect_* knob resolves initialConnectMode to OFF, so step 7 is a
single-shot connect; a server restart or an LB blip across a drained,
armed sender therefore latched every later table()/flush() call forever,
including seconds later once the endpoint was back.
Deferring the connect to the next send exposed a second defect, which this
commit also fixes. ensureConnected() calls
resetSymbolDictStateForNewConnection(), which cleared
currentBatchMaxSymbolId unconditionally. That watermark is batch-scoped,
not connection-scoped - a flush ships exactly [sentMaxSymbolId+1 ..
currentBatchMaxSymbolId] - and clearing it was harmless only because
build() connects before the application can register a symbol. On the
deferred path symbol() runs first, so the clear made the next flush ship
an empty delta while its rows referenced symbol id 0: rows on the wire
pointing at ids the server never received. The reset now runs only from
the drained state the old code assumed (no pending rows, no row in
progress).
testDefaultConfigRecycleSurvivesFailedReconnect pins both. It drives a
default-config sender (no reconnect knobs), kills the listener at a
drained instant, asserts the recycle's throw reaches the caller, asserts
recycleFailure is not latched by ingesting successfully once the endpoint
returns on the same port, and asserts the recovered stream defines every
symbol its rows reference. Against the pre-fix code it fails on the latch;
with the latch fixed but the watermark clear restored it fails on the
empty dictionary.
Documentation and hygiene alongside:
- getTotalFramesReplayed/getTotalFramesSent/getTotalReconnectAttempts/
getTotalReconnectsSucceeded/getTotalServerErrors now state that they
read the live send loop and therefore restart at 0 on every recycle
("since the last recycle"), and point at the lifetime-scoped
getSymbolDictEpoch/getSymbolDictResetsPerformed for correlation.
- Sender.resetSymbolDictionary(), its QwpWebSocketSender override and
LineSenderBuilder.symbolDictReset() now say that the manual valve is a
permanent no-op while symbol_dict_reset is off, because armIfEligible
gates on that knob.
- armIfEligible's javadoc names both call sites; resetSymbolDictionary()
calls it too, and the stale single-call-site claim invited a wrong
inlining refactor.
- The builder's symbolDictReset default references
DEFAULT_SYMBOL_DICT_RESET_ENABLED instead of a hardcoded true.
- SymbolDictRecycleOutageTest joins its trigger thread in a finally so an
assert failure cannot leave a thread inside the sender, and records why
the revived server's handshake count stays >= 1 rather than == 1.
- SymbolDictRecycleCrashWindowsTest inlines an unread handler binding, and
SymbolDictRecycleHealingTest's AckAllHandler carries a warning that its
per-connection ack reset assumes every connection change is a recycle
and must not be copied into a plain-reconnect test.
Client suite: 3045 run, 0 failures, 0 errors, 2 skipped.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn
Review C1: recycleForDictReset step 3 re-acquired the slot flock without waiting for a deferred close to release it. When the SF worker is wedged in a syscall past SegmentManager's bounded join, CursorSendEngine.close() returns with the flock retained and isCloseCompleted() false, releasing both from the worker's exit path; the step-6 rebuild then threw SlotLockContentionException on the retained flock and latched the sender permanently terminal -- converting exactly the transient disk stall the deferred-close machinery exists to survive into a hard sender death on the default (SF + recycle-on) path. recycleForDictReset now mirrors close()'s deferred-close discipline: awaitDeferredEngineClose parks (awaitAckedFsn-shaped) until the deferred cleanup confirms the flock release, re-arming the shared flock-release retry driver each pass like isSlotLockReleased() does. Only exhausting the 30 s budget -- a genuinely dead worker -- latches terminal, and that path hands the still-locked engine to retainedEngine so a pool re-probe recovers the slot's capacity if the worker ever exits; close() no longer clobbers slotLockReleased to true while such an engine is pending. SymbolDictRecycleDeferredCloseTest pins both branches with a wedged- worker harness (red-proofed: without the await, the survival test dies with the exact SlotLockContentionException the review traced). Review Mo1: getAckedFsn and awaitAckedFsn snapshot cursorEngine into a local (the recycle transitions it non-null -> null -> non-null), and while it is null they report the durable watermark the recycle barrier proved (new lastRecycleDurableFsn) instead of collapsing to -1. Review Mo2: PooledSender forwards resetSymbolDictionary() to the live delegate instead of inheriting the interface's no-op default; pinned by a pooled-path test. Review Mo5 (verified: released 9.4.x servers cap the dictionary at 1M): document the pre-10.0.0 compatibility constraint on the 2M client cap at QwpConstants.MAX_SYMBOL_DICTIONARY_SIZE and symbolDictReset(boolean). Review mi7/mi8: step 6 refuses a rebuilt engine that recovered from disk (the empties-the-slot contract was breached) and resets slotLockReleased after a successful swap. Review mi3: recycle javadoc now says seven steps / steps 2-6, the resetArmed and resetSymbolDictionary docs match the code, and EngineRebuildFactory moves out of the field block. Review Mo4: SymbolDictRecycleFsnContinuityTest wraps every test in assertMemoryLeak, and LineSenderBuilderWebSocketTest pins the config boundaries: threshold == 2M accepted on both config paths, the symbol_dict_reset=on parse branch, the invalid-value message, and the three fluent-setter transport guards. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
[PR Coverage check]😍 pass : 251 / 271 (92.62%) file detail
|
getAckedFsn()/awaitAckedFsn() read cursorEngine, fsnEpochBase and lastRecycleDurableFsn from monitoring threads while the recycle reassigns all three on the producer thread. The sibling observability counters went volatile in an earlier round for exactly this reader; these three carry the same contract, so a monitor could observe a fresh engine with a stale epoch base and report an FSN dip to -1. Also document resetSymbolDictionary() as producer-thread-only: it mutates unsynchronized producer state, and the fire-and-forget javadoc framing invited cross-thread calls (notably via PooledSender).
Red first: the outage tests now assert the store-and-forward contract (producer never sees a transport error nor a reconnect budget after the initial connect) instead of pinning the step-7 foreground connect they used to. The production change lands in the next commit.
Step 7 of the symbol-dict recycle re-ran the initial-connect policy on the producer thread: OFF senders got a single-shot connect whose failure made every subsequent send throw before buffering until the endpoint returned, and SYNC senders blocked the producer up to reconnect_max_duration_millis. Both violate the store-and-forward contract (post-init, the client never exposes transport problems and never imposes a reconnect budget on the producer). ensureConnected() now latches hasConnectedOnce on its first completion and routes every later entry through the existing ASYNC (deferred) branch: the loop is built and started synchronously, the socket connect happens on the I/O thread with indefinite retry, and the producer keeps buffering into the fresh epoch's slot. This is the same path ASYNC-mode senders already took at step 7. The server clears its per-connection dictionary on disconnect, so the buffered fresh-epoch frames (deltaStart=0) replay correctly on reconnect.
A symbol-dict recycle's step 7 hands the rebuilt CursorWebSocketSendLoop a null client (ensureConnected's deferred branch), and the loop's constructor seeds hasEverConnected = (client != null) = false. The fresh loop then believes it has never connected, even when a prior loop instance of the same sender already reached the server. That misclassification breaks two contracts. First, endpointPolicyFailureIsTerminal() treats "never connected" as a startup condition and takes the terminal branch on an auth, upgrade, or durable-ack rejection instead of Invariant B's retry-and-ride-it- out contract for a FOREGROUND sender past initialization. Second, the public wasEverConnected() -- documented "sticky, once true stays true" -- delegates to the loop's flag, so it reports false for the whole post-recycle outage window. QwpWebSocketSender now tracks hasLoopEverConnected, a sender-lifetime sticky OR across every loop instance it has owned: latched on a successful foreground connect, and OR'd in from the outgoing loop's own hasEverConnected() at recycle step 2 (covers an ASYNC-initial sender, whose only connect ever happened on the I/O thread). ensureConnected() seeds it into a freshly built loop via the new CursorWebSocketSendLoop.markEverConnected(), called before start(), restoring Invariant B's classification and wasEverConnected()'s stickiness across the rebuild. Also: renamed hasConnectedOnce to hasInitialConnectRun -- it collided in name, but not meaning, with CursorWebSocketSendLoop's own hasEverConnected field; fixed the deferred-connect log message and comment, which said "initial connect" even on a post-recycle re-entry; and pinned wasEverConnected()'s stickiness across the outage window in SymbolDictRecycleOutageTest.
The recycle's reconnect is now deferred to the I/O thread, so handshake/connection-count assertions that ran synchronously after the triggering table() call became racy. Each moves after the next awaitAckedFsn(): an acked post-recycle frame proves the fresh connection is up, making the count deterministic again. Refusal-path assertions (no recycle, no reconnect) are unchanged.
awaitAckedFsn now snapshots cursorSendLoop into a local before each null-check, the same way it already snapshots cursorEngine. The field is non-volatile and the recycle nulls it on the producer thread, so a monitor thread calling awaitAckedFsn could previously read non-null, then have the recycle null the field before the second read, NPE'ing inside checkError(). The snapshot closes that window at both call sites in the method. recycleForDictReset's javadoc and its step-7 catch's LOG.warn still described a foreground connect retry, but ensureConnected() now defers the socket connect to the I/O thread, so step-7 failures are only dispatcher construction, loop build, or start() -- environmental, not transport. Reworded both to describe the deferred reconnect that actually runs.
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
2 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.
The WebSocket sender no longer has a hard symbol-cardinality stop. When its global symbol dictionary reaches a configurable threshold (
symbol_dict_reset_threshold, default 100 000), the sender recycles its send stack at the next provably-safe instant — all published frames acknowledged, no row in progress, no open transaction: it closes the send loop, retires the store-and-forward slot with the existing fully-drained close (emptying it), starts a fresh dictionary epoch on the same slot, rebuilds the engine through the same code pathbuild()uses, and reconnects. The server needs no change: supported servers (10.x+) already clear per-connection dictionary state on disconnect, and this branch pins that behavior as a wire contract. Zero wire change, zero on-disk format change; all production code is client-side (~335 net lines).User-visible surface
Three connect-string keys / builder methods —
symbol_dict_reset(default on),symbol_dict_reset_threshold,symbol_dict_reset_max_wait_millis(bounded blocking wait for a starved reset, default 30 s, 0 = opportunistic-only); an advisorySender.resetSymbolDictionary()(no-op on non-QWP transports); three observability getters onQwpWebSocketSender(getSymbolDictEpoch,getSymbolDictResetsPerformed,getSymbolDictResetStarvationTimeouts). Externally visible frame sequence numbers stay strictly monotone across recycles via an epoch-base offset applied on every public surface (flush sequence, ack watermark, drain, progress stream, error spans). The client-side dictionary cap rises 1M → 2M (still under the server cap, with an OSS ordering test), and the cap error now points at the reset valve. A recycle also naturally heals a mid-life full-dict degrade (the fresh engine re-derives delta capability). Senders built via the publicconnect(...)overloads (no rebuild factory) and senders that don't own their engine never recycle.Release-note items:
CursorWebSocketSendLooppublic ctor and theQwpWebSocketSendermasterconnect(...)overload gained parameters in place.getTotalFramesSent,getTotalFramesReplayed,getTotalReconnectAttempts,getTotalReconnectsSucceeded,getTotalServerErrors, …) read the live cursor loop, which the recycle rebuilds, so they restart at 0 on every recycle — they are scoped "since the last recycle". Long-lived monitors that differenced them as lifetime counters should track the lifetime-scoped epoch counters (getSymbolDictEpoch,getSymbolDictResetsPerformed) alongside, or treat a negative delta as a recycle boundary.symbol_dict_reset=off(or a threshold above 1M) against a pre-10.0.0 server must keep its symbol cardinality below 1M — the raised 2M client cap only matches 10.0.0+ servers.Tests
10 new client suites (arming, FSN continuity, swap, memory mode, starvation, refusals, outage/orphans, catch-up skip, crash windows, healing/metrics — 45 tests), OSS disconnect-clear contract pin + real-client e2e (SF and memory modes) + seeded recycle×reconnect fuzz, ENT failover losslessness with recycles in flight. Verified green from a clean reactor build: client 3044/0/0/2, OSS cutlass/qwp 1343/0/0/3, ENT failover pair 3/0/0/0.
Review round 2 (06dcb6d)
Addresses the level-3 tandem review:
SegmentManager's bounded join (a stalled disk/NFS — exactly what the deferred-close machinery exists to survive); the step-6 rebuild then threwSlotLockContentionExceptionand latched the sender permanently terminal.awaitDeferredEngineClosenow parks until the worker's exit path confirms the release (30 s budget); only exhausting the budget — a genuinely dead worker — latches terminal, and that path hands the still-locked engine to the pool re-probe surface so the slot's capacity stays recoverable.SymbolDictRecycleDeferredCloseTestpins both branches with a wedged-worker harness; red-proofed against the pre-fix code (dies with the exactSlotLockContentionExceptiontrace).getAckedFsn()/awaitAckedFsn()snapshotcursorEngineinto a local and, while it is null (mid-swap, or after a failed swap), report the durable watermark the recycle barrier proved instead of collapsing to −1.PooledSenderforwardsresetSymbolDictionary()to the live delegate (it inherited the interface's no-op default); pooled-path test added.SymbolDictRecycleFsnContinuityTestnow wraps every test inassertMemoryLeak; config-boundary pins added (threshold == 2M accepted on both config paths,symbol_dict_reset=onparse, invalid-value message, the three fluent-setter transport guards).slotLockReleasedresets after a successful swap.Not taken this round (adjacent/pre-existing per the review): the
timeoutMillis * 1e6overflow at absurd inputs (endemic pattern, fails safe), the threshold-near-cap footgun, and theAckAllHandler/ENT-harness test dedup.Round-2 verification: full QWP client package + pool/facade suites re-run green (1 832 tests fresh, 0 failures, 0 errors,
assertMemoryLeakon); OSS core recompiled clean against this head.Companion PRs
🤖 Generated with Claude Code
https://claude.ai/code/session_01K2Hw6TfgX9sBB2L8oSdaWn