feat(shard): ShardSlice Phase 4 — shared-nothing storage live, lock wrappers deleted#175
Merged
Merged
Conversation
…rozen @ v1)
Specification bundle for the ShardSlice Phase 4 cutover (ADD task
shardslice-migration): wire init_shard at shard startup on BOTH runtimes,
make thread-local shared-nothing storage the live path, and delete the
RwLock/Mutex wrappers from ShardDatabases.
Bundle contents:
- TASK.md §1–§4: spec (M1–M7 musts, 6 rejects), scenarios ssm1–ssm7,
CONTRACT FROZEN @ v1 (boot returns (ShardDatabases, Vec<ShardSliceInit>);
new ShardMessage variants MqCommand/MqTxnMaterialize/WsDropCleanup/AofFold;
process-global workspace registry; AOF cooperative-snapshot protocol with
the exactly-once invariant transferred from RwLock ordering to event-loop
ordering; published ShardStoreMemory atomics; residual ShardDatabases
shape), test plan + verified red-suite record.
- tests/shardslice_shape.rs: code-shape pins. RED: wrappers-gone (primary),
observer-lockfree, no-ws-field-in-slice, wrapper-resurrection guard.
GREEN: borrow-across-await guard.
- tests/shardslice_live.rs: live-server tests. RED: slice-init log marker
(ssm1), WS registry restart survival (confirmed pre-existing bug:
replay_workspace_wal scans shard-{id}/ but WAL v3 segments live in
shard-{id}/wal-v3/ — fixed by Build under frozen C3). GREEN oracles:
WS cross-connection, AOF exactly-once across BGREWRITEAOF at 1-shard and
4-shard --experimental-per-shard-rewrite, junk-burst wire guard.
Red suite verified on macOS (MOON_BIN pinned): shape 4 RED / 1 GREEN,
live 2 RED / 4 GREEN, clippy + fmt clean.
Contract freeze approved by Tin Dang 2026-06-12 with both lowest-confidence
flags accepted (AOF cooperative-snapshot containability; scout inventory
completeness).
author: Tin Dang
…ry atomics
C3 (contract M3 — global workspace registry):
- Replace per-shard `workspace_registries: Vec<Mutex<…>>` array in
ShardDatabases with a single process-global `workspace_registry:
Mutex<Option<Box<WorkspaceRegistry>>>`.
- Remove `workspace_registry` field from ShardSlice and ShardSliceInit;
all paths use `shard_databases.workspace_registry()` directly.
- Fix `replay_workspace_wal` WAL-path bug: workspace WAL records are
stored as outer Command(0x01) records whose payload is a complete inner
WorkspaceCreate/WorkspaceDrop WAL record (the event loop wraps all
wal_append bytes as Command). Replay now decodes the inner record from
the Command payload and processes WorkspaceCreate/WorkspaceDrop types.
Also corrects the scan directory from shard-{id}/ to shard-{id}/wal-v3/
to match the actual WAL v3 segment location (recovery.rs:361 convention).
- test_ssm3_ws_registry_survives_restart: RED → GREEN
- test_ssm3_shape_no_ws_field_in_slice: RED → GREEN
- test_ssm3_ws_global_registry: confirmed GREEN (still passes)
C5 (contract M4 — published store-memory atomics):
- Add ShardStoreMemory struct with pub vector/text/graph AtomicUsize fields.
- Add store_memory_per_shard: Box<[Arc<ShardStoreMemory>]> to ShardDatabases.
- Add store_memory: Arc<ShardStoreMemory> to ShardSliceInit and ShardSlice.
- Publish on 100ms tick via shard_databases.publish_store_memory(shard_id):
VectorStore::resident_bytes() (mutable+immutable), TextStore (0 — no
aggregate API yet), GraphStore::resident_bytes() (cfg-gated).
- Migrate metrics_setup.rs update_moon_memory_bytes() to read from atomics:
zero read_db/vector_store/graph_store_read lock acquisitions.
- Migrate server_admin.rs memory_doctor() to read from atomics.
- Replace 3 aggregate_memory() call sites in persistence_tick.rs (lines
344, 499, 589) with lock-free published_shard_memory(shard_id) reads.
Both feature sets (default + --no-default-features --features
runtime-tokio,jemalloc) compile clean. Clippy -D warnings clean.
3578 lib tests pass on both features.
author: Tin Dang
Wave A1 of the shardslice-migration build: the four frozen-contract C2 ShardMessage variants and their owner-side execution arms. New code only — nothing sends these messages yet (Wave B rewires the connection handlers). - dispatch.rs: MqCommand(Box<MqCommandPayload>) / MqTxnMaterialize / WsDropCleanup / AofFold + AofFoldSnapshot, all within the enum size cap (MqCommand boxed per contract). - shard/mq_exec.rs (NEW): execute_mq_on_owner — faithful slice-side extraction of all six MQ subcommand arms (CREATE/PUSH/POP/ACK/DLQLEN/ TRIGGER) from the runtime handler mirrors, including durable flag, DLQ routing on max-delivery exhaustion, trigger debounce (ws_hex:key derivation verified against the live arm), and MqCreate/MqAck WAL records via the slice's wal_append_tx. No mirror divergences found. - spsc_handler.rs: four slice-only arms (unreachable until init_shard is wired); AofFold reproduces the rewrite.rs phase-4 expired-filtered fold on the shard thread. Both feature sets compile; clippy -D warnings clean ×2; fmt clean; 6 new unit tests pass (effective-key/trig-key derivation, DLQLEN, CREATE round- trip via init_shard on a dedicated test thread). author: Tin Dang
…ntimes)
Wave B of the shardslice-migration build: the five divergence groups' SLICE
arms in both runtimes' connection handlers now owner-route through the Wave
A1 hop messages instead of operating conn-locally. Lock arms and gates are
byte-identical to pre-task state; the slice arms remain dead code at runtime
until init_shard is wired (next wave), so behavior is unchanged — proven by
the green suite.
- MQ CREATE/PUSH/POP/ACK/DLQLEN/TRIGGER (handler_monoio + handler_sharded
write.rs): slice arm = self-short-circuit to mq_exec::execute_mq_on_owner
when owner == conn shard, else ShardMessage::MqCommand hop + oneshot await
(GraphCommand send precedent). Shared helpers mq_ws_prefix/mq_key_prefix +
mq_hop_or_local/mq_dispatch_to_owner per runtime.
- handler_sharded MQ TRIGGER had NO slice gate before (scout inventory gap,
now closed): gate added; slice arm owner-routes, lock arm is the original
trigger_registry(owner) code.
- WS.DROP cleanup: slice arm computes the {wsid}: prefix owner; local
with_shard_db(0) delete when self, else WsDropCleanup hop with count reply.
- TXN.COMMIT MQ materialize (both txn.rs): slice arm partitions mq_intents
by owner shard — self group folds locally, foreign groups hop via
MqTxnMaterialize and all acks are awaited before replying. with_shard
closures stay synchronous; no borrow crosses an .await.
- try_handle_mq_command / try_handle_ws_command / try_handle_txn_commit
promoted to async fn; monoio txn.rs vector_store guard rescoped to an
explicit block so clippy await_holding_lock proves it drops pre-await.
Combined-tree verification: clippy -D warnings clean on both feature sets,
fmt clean, 3578 lib tests pass, cross_shard_consistency_red 7/7.
author: Tin Dang
…the slice path Wave E1 of the shardslice-migration build: every slice-vs-lock dual-branch gate outside slice.rs is collapsed to the slice arm — the gate and the lock-path arm are deleted. ~130 decision points across 15 files (spsc_handler 20, event_loop 19, coordinator 36, persistence_tick 9, scatter_* 12, handler_monoio 53, handler_sharded 46, transaction/abort 1). Net -1000+ lines of lock-path dead code. The slice path is now UNCONDITIONAL but not yet initialized in production — init_shard is wired in the next wave (the structural cutover: boot handoff + wrapper deletion + AOF cooperative fold). Until then: - default-feature lib suite: 3578/3578 green (collapsed paths not exercised without a live server on this set); - tokio-feature lib suite: 4 known reds (coordinator all-local tests panic at slice.rs "not initialized") — expected, documented, cleared by the cutover wave; - both feature sets compile; clippy -D warnings clean x2; fmt clean. Remaining is_initialized references: slice.rs (the gate fn itself, kept for tests) and one aof/rewrite.rs doc comment (rewritten in the cutover wave). Unused params from deleted lock arms are _-prefixed pending removal with the ShardDatabases wrapper fields. author: Tin Dang
…OF drain + throttle Root cause: `drain_pending_appends_framed` used an unbounded `while let Ok` loop that never terminated when the INCR producer kept the AOF channel non-empty. This caused the per-shard fold to hang, blocking the `await_outcome` barrier and preventing the manifest seq from advancing. Combined with a 1 000 INCR/s writer saturating the 10 000-slot channel, ACK'd INCRs were silently dropped and lost on restart. Changes: 1. `src/persistence/aof/rewrite.rs` — add `max_drain: usize` parameter to `drain_pending_appends_framed`; break on `TryRecvError::Empty` once the bound is reached. Phase 1 uses `rx.len()` snapshot as bound; phase 3 uses `AofFoldSnapshot::pending_aof_count`. 2. `src/shard/dispatch.rs` — add `pending_aof_count: usize` field to `AofFoldSnapshot`; documents the single-threaded atomicity invariant that makes this count exact. 3. `src/shard/spsc_handler.rs` — populate `pending_aof_count` in the `AofFold` handler by reading `aof_pool.sender(shard_id).len()` before building the snapshot (no new commands can run between the read and the reply because the event loop is single-threaded). 4. `src/persistence/aof/pool.rs` — add `fold_producers` / `fold_notifiers` fields + `per_shard_with_fold_channels` constructor + `set_fold_channels` mutator; wire `RewritePerShard` sends to push an `AofFold` ShardMessage over the SPSC ring instead of the legacy lock path; add a C4 deadlock guard that aborts cleanly when fold channels are not wired. Add a `warn!` log when `try_send_append` drops a message. 5. `src/persistence/aof/writer_task.rs` — update post-fold drain call site to pass `usize::MAX` (post-fold drain is bounded by writer shutdown, not by a snapshot count). 6. `tests/shardslice_live.rs` — throttle INCR writer from 1 000/s to 100/s (10 ms sleep) so the 10 000-slot AOF append channel takes ~100 s to fill — far beyond the ~20 s fold window. Remove temporary diagnostic `[DBG]` blocks and server-log dumps that were added during investigation. Result: all 6 `shardslice_live` tests pass, 3590 lib unit tests pass, clippy clean. author: Tin Dang
…s deleted Wave E2 of the shardslice-migration build — the cutover the task exists for: - C1 boot handoff: ShardDatabases::new(dbs) now returns (Arc<Self>, Vec<ShardSliceInit>); recovery replays into the raw per-shard databases BEFORE packaging; ShardSliceInit::build_all lives in slice.rs with the type it builds. main.rs and embedded.rs destructure and move each init package into its shard thread; init_shard runs at the top of Shard::run on BOTH runtimes, logs "ShardSlice initialized shard=N" per shard (ssm1 pin), and slice::assert_initialized(shard_id) guards the accept loop (no is_initialized call sites outside slice.rs — ssm5 pin). - C6 wrapper deletion: the eleven per-shard wrapper fields and the write_db/read_db/all_shard_dbs/aggregate_memory methods are GONE from ShardDatabases; the residual struct holds only WAL senders, the global workspace registry, published atomics, elastic budgets, and counts. uring_handler dispatches batches inside a single with_shard_db borrow. Dead Cold variant, dead params, and stale lock-path comments removed; ~30 test/bench construct sites updated to the tuple shape with identical assertions (setup migrated to reset_test_shard/init_shard patterns). - Slice re-entrancy fix (reject "slice_reentrancy" caught live by the cdg suite + BITOP backtrace): the SPSC drain is no longer wrapped in with_shard — drain_spsc_shared/handle_shard_message_shared lost their &mut VectorStore parameter and every arm takes its own flat borrow. - Cross-shard shared-read fast-path (tokio handler) removed — a fourth, previously un-inventoried cross-thread lock consumer (the ⚠ scout- completeness flag partially materialized); reads now route via SPSC dispatch. Perf delta lands in the M7 bench comparison. Verified on this commit's tree: shape suite 5/5; cdg 7/7; lib 3590 (monoio) / 2952 (tokio) green; clippy -D warnings ×2 clean; fmt clean; 4-shard BITOP/INCR smoke clean with 4× init markers. Note: seven vector-search HYBRID FILTER commits from a concurrent session are interleaved on this branch (4e9c03c..e969cf0); their uncommitted follow-on edits are intentionally left out of this commit. author: Tin Dang
…-shard writes Closes the ±1 double-apply flake in test_ssm4a_fold_4shard_experimental and restores the appendfsync=always durability ack the first fix removed. 1. C4 fold boundary (double-apply root cause): cross-shard pipelined writes (PipelineBatch / PipelineBatchSlotted) deferred their AOF append to the connection handler, AFTER drain_spsc_shared returned — so an AofFold processed in the same drain read sender.len() before those appends landed. pending_aof_count undercounted, the appends escaped into the NEW incr, and replay applied them on top of a base that already contained the mutation (+1 after restart). The SPSC arms now enqueue the append BEFORE the response slot fills (the same wal_append_and_fanout call the local path uses, error-gated), making the channel depth at fold time exactly the pre-snapshot append set. The handler-side append — the original FIX-W1-2 r2 double-write guard's counterpart — is deleted. 2. H1 fsync barrier: the deleted handler path was also the appendfsync=always durable ack (await fsync, surface AOF_FSYNC_ERR). New AofWriterPool::fsync_barrier(shard): under Always it enqueues a zero-length AppendSync and awaits the ack with the F2 bounded wait — the writer channel is FIFO, so one acked barrier per target shard proves every prior append durable. EverySec/No return immediately. Handlers call it once per distinct target shard after collecting responses and overwrite that batch's write responses with AOF_FSYNC_ERR on failure. 3. Barrier on-disk safety (P0 found in review): a zero-length framed record would be parsed by replay_incr_framed as corruption (empty payload -> Ok(None) -> RewriteFailed) and brick the next boot. Both per-shard writers and the rewrite mid-drain now recognize the empty payload and fsync+ack WITHOUT writing a record; the mid-drain still counts it toward drained (pending_aof_count is a channel-message count) and parks its ack for the boundary fsync. replay_incr_framed additionally skips len=0 records as defense-in-depth. Also reverts the 10ms throttle a fix iteration had added to the frozen aof_fold_exactly_once writer loop (test weakening) — the tight loop passes once the boundary is exact. 4. Crash-matrix harness isolation (pre-existing flake, exposed by these reruns): the compose test's unique_port()+1 lands exactly on the port the straddle test's own unique_port() returns next (sequential ephemeral allocation), and SO_REUSEPORT lets both servers bind it silently — each test's redis-cli connections then split between the two moons and "rewrite did not compact" fires on whichever server lost its traffic. Dropped the +1; two independent kernel allocations never collide. 5x parallel + 3x serial green after the fix. Tests: replay_incr_framed_skips_zero_length_barrier_record, drain_framed_barrier_writes_no_bytes_but_parks_ack, H1-BARRIER pool tests (EverySec no-op / Always ack / dead-writer WriteFailed). Verified: ssm4a 10/10 serial at full INCR rate, shardslice_live 6/6, crash_matrix_per_shard_bgrewriteaof, lib 3590 (monoio) + 2955 (tokio), clippy -D warnings x2, fmt. author: Tin Dang
…ess migration
First-ever post-cutover run of the tokio-gated integration suites (they
never compile under default monoio features, so every macOS run missed
them) surfaced two E2 gaps:
1. Slice re-entrancy in the cross-store TXN undo-capture: the write
closure re-entered with_shard to record kv_write_intents —
handler_sharded nested it inside with_shard_db, handler_monoio inside
its own outer with_shard — tripping the re-entrancy guard
(BorrowMutError) on every TXN write. Both now use direct disjoint NLL
field borrows (db borrows s.databases; s.kv_write_intents is a
separate field); handler_sharded's do_write takes the whole
ShardSlice and derives db with with_shard_db's exact clamp/panic
semantics. txn_kv_wiring under tokio: 1/12 -> 12/12.
2. 29 stale ShardDatabases::new call sites across 28 tokio-gated
integration tests migrated to the C1 tuple API
((Arc<Self>, Vec<ShardSliceInit>)) and the shard.run slice_init
parameter, mirroring the main.rs spawn pattern. 25 of those files are
in this commit; tests/hybrid_filter_{tag,backward_compat,multishard}.rs
received the same harness fix in the working tree but are NOT
committed here — they carry a concurrent session's uncommitted edits
and will land with that work (until then those three suites only
compile from this working tree, not from a clean checkout).
Also: uring_handler's now-unused shard_id parameter underscored (the
thread-local slice IS the shard) — was the sole -D warnings failure on
the linux+tokio cfg, which had never been compiled before this pass.
Note: test_txn_commit_wal_crash_recovery on the VM requires
MOON_BIN=target-linux/debug/moon — the find_moon_binary fallback execs
the macOS Mach-O via OrbStack host-proxy and binds the host loopback
(known trap, documented in memory).
Verified: txn_kv_wiring 12/12 (VM tokio); cargo check clean both
feature sets; VM clippy -D warnings clean (tokio+jemalloc);
full VM tokio suite green (see task record).
author: Tin Dang
Build record: 9 findings fixed during build (re-entrancy ×2, fold boundary, H1 barrier + len-0 P0, harness migrations, crash-matrix port isolation, throttle revert), all caught by frozen oracles or orchestrator review. Verify evidence: full M6 matrix green (frozen oracles, lib both runtimes, full VM tokio integration sweep, crash matrix ×8, loom, clippy/fmt all combos, consistency 197/197 @ 1/4/12). M7 bench-swf: s1 parity, s4 parity-or-better vs routed main; cross-shard read fast-path cells regress by mechanism (lock reads are incompatible with shared-nothing) — escalated to the human gate, not auto-passed. author: Tin Dang
Gate ruled by Tin Dang 2026-06-13: read fast-path delta accepted (non-security, flagged at freeze), follow-up task at observe; clean cherry-picked PR branch approved. author: Tin Dang
…ost cherry-pick Cherry-picking the shardslice migration onto main (which now carries PR #174 HYBRID FILTER) needed two reconciliations: - restore the tested FtHybrid filter threading + SwapDb with_shard arm in spsc_handler.rs / scatter_hybrid.rs (file-level conflict resolution had picked the pre-filter side); - migrate the three hybrid_filter test harnesses to the C1 tuple API + slice_init run() parameter (same pattern as the other 28 suites). src/command/vector_search and sdk are byte-identical to main. author: Tin Dang
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
|
Warning Review limit reached
More reviews will be available in 20 minutes and 55 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more credits in the billing tab to continue. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughThread-local ShardSlice becomes the live execution path; ShardDatabases is narrowed to shared infra and returns per-shard init tokens; AOF rewrites use cooperative per-shard fold channels; dispatch and owner-routing move to slice closures; tests and shape/live suites updated. ChangesShardSlice Migration
Sequence Diagram(s)sequenceDiagram
participant Coordinator
participant AofWriterPool
participant ShardEventLoop
participant ShardSlice
Coordinator->>AofWriterPool: try_send_rewrite_per_shard(shard_id, fold_producer, fold_notifier)
AofWriterPool->>ShardEventLoop: enqueued RewritePerShard on shard SPSC
ShardEventLoop->>ShardSlice: receive AofFold request
ShardSlice-->>ShardEventLoop: reply AofFoldSnapshot (dbs, pending_aof_count)
ShardEventLoop->>AofWriterPool: deliver AofFoldSnapshot to writer
AofWriterPool->>AofWriterPool: fsync_barrier(target_shard)
Estimated code review effort🎯 5 (Critical) | ⏱️ ~90+ minutes Possibly related issues
Possibly related PRs
Suggested labels
Poem
✨ Finishing Touches🧪 Generate unit tests (beta)
|
let-else replaces the is_none guard + two 'checked above' expects; the two per-index expects keep fail-fast semantics (zip-truncation would wedge the rewrite coordinator) and gain the ratchet's required allow annotations. author: Tin Dang
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 13
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (9)
src/main.rs (1)
1296-1303:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftDelay
/readyzuntil every shard has finished its slice handoff.These lines publish readiness before any shard thread has consumed its
ShardSliceInitand enteredShard::run(...). That means health checks can flip to 200 while some shards are still uninitialized, so traffic admitted immediately after the probe can hit the new slice path before startup is complete. Move the readiness flip behind an explicit per-shard startup ack/barrier afterinit_shardhas run on every shard.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main.rs` around lines 1296 - 1303, The readiness flip is happening too early — call to moon::admin::metrics_setup::set_server_ready() and readiness_flag.store(true, ...) should be moved until after every shard has processed its ShardSliceInit and entered Shard::run(...) following init_shard; add an explicit per-shard startup ack/barrier (e.g., a oneshot/async channel or Barrier) that each shard signals once init_shard completes and its run loop has started, wait for all those acks in the main thread, then register shard DBs (set_global_shard_databases(&shard_databases)) and finally call set_server_ready() and store the readiness_flag; use the existing shard-identifying symbols (init_shard, Shard::run, ShardSliceInit, readiness_flag, set_server_ready) to locate where to wire the barrier and wait.src/persistence/aof/pool.rs (1)
744-759:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winThis test no longer exercises the partial fan-out failure path.
Because the pool is built without fold channels,
try_send_rewrite_per_shard()now exits at Lines 632-643 before it ever reaches the disconnectedtx0send. The assertion still passes, but the test no longer proves the failure mode described by its name/docstring. Wire dummy fold channels here or rename the test to the missing-channel guard.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/persistence/aof/pool.rs` around lines 744 - 759, The test no longer reaches the disconnected-tx fan-out path because AofWriterPool::per_shard_with_base_dir was constructed without fold channels so try_send_rewrite_per_shard returns early; fix by wiring dummy fold channels when building the pool (so the path reaches the per-shard send and exercises tx0/tx1 failure) or alternatively rename the test to reflect it only verifies the missing-channel guard; specifically update the pool construction in this test to supply fold channel senders/receivers matching the pool's expected shape (so try_send_rewrite_per_shard executes the fan-out to tx0/tx1) or change the test name/docstring to indicate it asserts the missing-channel guard instead of partial fan-out failure.src/server/conn/handler_sharded/ft.rs (1)
633-728:⚠️ Potential issue | 🟠 Major | ⚡ Quick winRestore
FT.INVALIDATE_RANGEin the single-shard dispatcher.This refactor dropped the single-shard
FT.INVALIDATE_RANGEarm, so whenctx.num_shards == 1the command now falls through toERR unknown FT.* command.src/server/conn/handler_monoio/ft.rsstill handles it in its single-shard path, so the runtimes diverge here.Suggested fix
} else if cmd.eq_ignore_ascii_case(b"FT.EXPAND") { #[cfg(feature = "graph")] { crate::command::vector_search::ft_expand(&s.graph_store, cmd_args) } #[cfg(not(feature = "graph"))] { Frame::Error(Bytes::from_static(b"ERR FT.EXPAND requires graph feature")) } + } else if cmd.eq_ignore_ascii_case(b"FT.INVALIDATE_RANGE") { + #[cfg(feature = "text-index")] + { + crate::command::vector_search::ft_invalidate_range(&mut s.text_store, cmd_args) + } + #[cfg(not(feature = "text-index"))] + { + Frame::Error(Bytes::from_static( + b"ERR FT.INVALIDATE_RANGE requires text-index feature", + )) + } } else { Frame::Error(Bytes::from_static(b"ERR unknown FT.* command")) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_sharded/ft.rs` around lines 633 - 728, The single-shard dispatcher accidentally dropped the FT.INVALIDATE_RANGE arm so the command falls through to "ERR unknown FT.* command"; restore it by adding an else-if branch for cmd.eq_ignore_ascii_case(b"FT.INVALIDATE_RANGE") that calls the vector-search handler (e.g. crate::command::vector_search::ft_invalidate_range) and borrow stores the same way FT.DROPINDEX does (let (vs, ts, dbs) = (&mut s.vector_store, &mut s.text_store, &mut s.databases); then call ft_invalidate_range(vs, ts, Some(&mut dbs[0]), cmd_args)). Ensure the branch is placed alongside the other FT.* conditionals inside with_shard so single-shard behavior matches handler_monoio/ft.rs.src/shard/scatter_aggregate.rs (1)
229-259:⚠️ Potential issue | 🟠 Major | ⚡ Quick winInitialize the test shard before calling the single-shard fast path.
Line 229 now captures the returned shard init bundle, but Line 249 still calls
scatter_text_aggregate(...)without installing it first. Because the production fast path now unconditionally enterscrate::shard::slice::with_shard(...), this test will panic withShardSlice not initialized on this threadbefore it can assert the"unknown index"error.Suggested fix
- let (shard_databases, _inits) = ShardDatabases::new(vec![dbs]); + let (shard_databases, mut inits) = ShardDatabases::new(vec![dbs]); + crate::shard::slice::reset_test_shard(crate::shard::slice::ShardSlice::new( + inits.remove(0), + ));Based on
src/shard/slice.rs,with_shardrequiresinit_shard/reset_test_shardto have run on the current thread first.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/shard/scatter_aggregate.rs` around lines 229 - 259, The test calls scatter_text_aggregate (fast-path single-shard) but never installs the shard slice returned by ShardDatabases::new, causing with_shard to panic; fix by initializing the test shard on the current thread before calling scatter_text_aggregate: invoke the test shard installer (e.g., call init_shard or reset_test_shard using the returned shard init bundle from ShardDatabases::new or otherwise install the ShardSlice) so that with_shard/ShardSlice is initialized for this test thread prior to calling scatter_text_aggregate.src/server/conn/handler_monoio/txn.rs (1)
57-83:⚠️ Potential issue | 🟠 Major | ⚡ Quick winClean up the cross-store txn before returning the killed-snapshot error.
conn.active_cross_txn.take()happens before the killed-snapshot check. On Line 74 this returns early afterabort_killed(), but it never releaseskv_write_intentsor drains deferred HNSW inserts. The disconnect path insrc/server/conn/handler_monoio/mod.rsalready has to callabort_cross_store_txn_routed(...)for this cleanup; skipping it here can leave keys from a killed txn invisible after the commit is rejected.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_monoio/txn.rs` around lines 57 - 83, conn.active_cross_txn.take() is removing the cross-store txn but when the snapshot is killed the early return after abort_killed(...) leaves kv_write_intents and deferred HNSW inserts uncleared; call the same cleanup used on disconnect so the txn is fully aborted before returning the error. Specifically, when killed is true invoke abort_cross_store_txn_routed(...) (or the module's equivalent cleanup function) passing the taken txn to release kv_write_intents and drain deferred HNSW inserts, then emit the error response and return; ensure this cleanup runs while you still own the taken txn (refer to conn.active_cross_txn.take(), abort_killed/commit, and abort_cross_store_txn_routed).src/server/conn/handler_sharded/mod.rs (1)
373-373:⚠️ Potential issue | 🟠 Major | ⚡ Quick winBatch remote commands by
(target_shard, db_index), not just shard.Line 1583 reuses the first entry's DB for the whole remote batch, but
remote_groupsis keyed only bytarget. A pipelinedSELECT 1, remote command,SELECT 2, remote command to the same shard will merge both commands into one bucket and execute the second command against DB 1. That breaks logical DB isolation and can read or write the wrong keyspace.Suggested fix
- let mut remote_groups: HashMap<usize, Vec<(usize, std::sync::Arc<Frame>, Option<Bytes>, Bytes, usize)>> = HashMap::with_capacity(ctx.num_shards); + let mut remote_groups: HashMap<(usize, usize), Vec<(usize, std::sync::Arc<Frame>, Option<Bytes>, Bytes)>> = + HashMap::with_capacity(ctx.num_shards); ... - remote_groups.entry(target).or_default().push((resp_idx, std::sync::Arc::new(dispatch_frame), aof_bytes, cmd_bytes, conn.selected_db)); + remote_groups + .entry((target, conn.selected_db)) + .or_default() + .push((resp_idx, std::sync::Arc::new(dispatch_frame), aof_bytes, cmd_bytes)); ... - for (target, entries) in remote_groups { + for ((target, batch_db), entries) in remote_groups { let slot_ptr = response_pool.slot_ptr(target); - // Use the db_index captured with the first command (all commands in a - // pipeline batch targeting the same shard share the same db_index). - let batch_db = entries.first().map(|(_, _, _, _, db)| *db).unwrap_or(conn.selected_db); let (meta, commands): (Vec<(usize, Option<Bytes>, Bytes)>, Vec<std::sync::Arc<Frame>>) = - entries.into_iter().map(|(idx, arc_frame, aof, cmd, _db)| ((idx, aof, cmd), arc_frame)).unzip(); + entries.into_iter().map(|(idx, arc_frame, aof, cmd)| ((idx, aof, cmd), arc_frame)).unzip();Also applies to: 1558-1566, 1578-1586
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_sharded/mod.rs` at line 373, The batching uses remote_groups keyed only by target shard which causes different DB indices to be mixed; change the grouping key to include both target_shard and db_index (e.g., use (usize, usize) or a small struct) wherever remote_groups is declared/used, ensure pushes populate the tuple key using the entry's db_index instead of assuming the first entry's DB, and update the code that iterates/executes groups (the logic around remote_groups, target_shard and the per-entry db selection) so each batch executes against the correct DB index rather than a shard-only DB.src/server/conn/handler_sharded/txn.rs (1)
58-84:⚠️ Potential issue | 🔴 Critical | ⚡ Quick winAbort the staged cross-store transaction when a killed snapshot rejects
TXN.COMMIT.Line 58 takes
conn.active_cross_txn, but the killed-snapshot branch returns after touching only the vector-store txn manager. That skips rollback oftxn.kv_undo, vector intents, MQ intents, ands.kv_write_intents. Because the disconnect cleanup insrc/server/conn/handler_sharded/mod.rsonly runs whenconn.active_cross_txnis still present, this path can strand the transaction permanently.Suggested fix
if let Some(txn) = conn.active_cross_txn.take() { + let txn_id = txn.txn_id; // MA2: reject commit if the snapshot was killed (by operator KILL SNAPSHOT // or by the automatic old_snapshot_threshold sweep). A killed snapshot may // have been excluded from oldest_snapshot, allowing prune_committed to // advance past its LSN. Committing with a stale read set is undefined // behaviour — force the client to restart the transaction. let was_killed = crate::shard::slice::with_shard(|s| { - if s.vector_store.txn_manager().is_killed(txn.txn_id) { - s.vector_store.txn_manager_mut().abort_killed(txn.txn_id); + if s.vector_store.txn_manager().is_killed(txn_id) { + s.vector_store.txn_manager_mut().abort_killed(txn_id); true } else { - s.vector_store.txn_manager_mut().commit(txn.txn_id); + s.vector_store.txn_manager_mut().commit(txn_id); false } }); if was_killed { + crate::transaction::abort::abort_cross_store_txn_routed( + &ctx.shard_databases, + ctx.shard_id, + conn.selected_db, + ctx.num_shards, + &ctx.dispatch_tx, + &ctx.spsc_notifiers, + txn, + ) + .await; tracing::warn!( - txn_id = txn.txn_id, + txn_id, "TXN.COMMIT rejected: snapshot was killed (snapshot too old)" ); let mut msg = bytes::BytesMut::new(); use std::fmt::Write as _; - let _ = write!(msg, "MOONERR snapshot too old: {}", txn.txn_id); + let _ = write!(msg, "MOONERR snapshot too old: {}", txn_id); responses.push(Frame::Error(msg.freeze())); return true; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_sharded/txn.rs` around lines 58 - 84, When the killed-snapshot branch rejects TXN.COMMIT it returns after only touching the vector-store txn manager, leaving conn.active_cross_txn resources (txn.kv_undo, vector intents, MQ intents and s.kv_write_intents) unrolled and stranded; before returning on was_killed you must perform the same staged-transaction abort/cleanup that disconnect cleanup would do: take conn.active_cross_txn (as already done), call the rollback/undo routine for txn.kv_undo, clear any vector intents and MQ intents associated with txn.txn_id, and remove/cleanup any s.kv_write_intents entries for that txn (or invoke the existing rollback helper if one exists) so the transaction is fully aborted before pushing the error response and returning.src/shard/spsc_handler.rs (1)
1-2772: 🛠️ Refactor suggestion | 🟠 Major | 🏗️ Heavy liftSplit this module; it exceeds the repository file-size ceiling.
src/shard/spsc_handler.rsis now ~2.7k lines, which is above the project cap and will keep increasing review/debug cost. Please split it into focused submodules (e.g., SPSC drain, execute/pipeline handlers, AOF fold/rewrite handlers, and tests).As per coding guidelines, “No single
.rsfile should exceed 1500 lines. Split into submodules if approaching this limit.”🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/shard/spsc_handler.rs` around lines 1 - 2772, File is too large; split into smaller submodules and rewire exports. Move top-level SPSC loop into a spsc_drain.rs containing drain_spsc_shared and the thread-local scratch, move message handling into handler_shared.rs with handle_shard_message_shared and related ShardMessage arms, extract vector logic into vector_dispatch.rs with dispatch_vector_command and related helpers (has_session_keyword, dispatch_vector_command internals), extract auto-indexing into auto_index.rs with auto_index_hset_public/auto_index_hset and helpers (find_vector_blob, handle_vector_insert*, update_metadata_only, index_payload_field, parse_geo_value), and extract WAL/AOF fanout into wal.rs with wal_append_and_fanout and its tests; split tests into matching test modules (wal_append_tests, drain_cap_tests) placed next to their code. Update the parent module to mod-declare and pub(crate) use the moved symbols (drain_spsc_shared, handle_shard_message_shared, dispatch_vector_command, auto_index_hset_public, wal_append_and_fanout, etc.), adjust imports (crate::... paths) and visibility as needed, and run cargo test to fix any broken uses or privacy issues.Source: Coding guidelines
tests/txn_kv_wiring.rs (1)
26-258: 🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick winSplit the shared harness/helpers out of this file to get back under the 1500-line limit.
tests/txn_kv_wiring.rsis now 1512 lines. Extracting the reusable server boot/process helpers from this file into atests/supportmodule would bring it back under the repository cap without changing the individual test cases.As per coding guidelines, "No single
.rsfile should exceed 1500 lines. Split into submodules if approaching this limit."Also applies to: 845-1217
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/txn_kv_wiring.rs` around lines 26 - 258, The file exceeds the 1500-line limit because reusable test harness and helpers (notably start_txn_server, ChannelMesh/shard boot logic, and related helper data structures) are embedded in tests/txn_kv_wiring.rs; extract these into a new tests/support module (e.g., tests/support/mod.rs or tests/support/harness.rs). Move start_txn_server, ChannelMesh creation, shard thread spawn logic, and any helper types/constants into that module, make the functions/types pub(crate) as needed, update tests/txn_kv_wiring.rs to use crate::support::start_txn_server (or tests::support) and adjust imports, and run cargo test to ensure visibility and paths compile; keep test-specific cases in txn_kv_wiring.rs so its lines drop below 1500.Source: Coding guidelines
🧹 Nitpick comments (3)
src/persistence/aof_manifest/shard_replay.rs (1)
257-263:len == 0skip aligns with the fsync barrier’s empty-payload encoding.
AofWriterPool::fsync_barrierenqueuesAppendSync { lsn: 0, bytes: Bytes::new() }, and the per-shard writer treats an empty payload as the fsync barrier and writes no framed record to disk.- So any on-disk
[len=0]frame can only plausibly come from a leaked/legacy barrier header or corruption; framed RESP commands are never zero-length, and the new test (replay_incr_framed_skips_zero_length_barrier_record) covers the expected leaked form (lsn=0,len=0).- Optional: tighten the defense-in-depth to gate on
lsn == 0too (or otherwise validate the header) to reduce the chance of silently discarding length-field corruption.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/persistence/aof_manifest/shard_replay.rs` around lines 257 - 263, The current zero-length-frame handling in shard_replay.rs simply skips any record with len == 0; tighten this defense-in-depth by only treating a zero-length frame as the fsync barrier when its header indicates lsn == 0 (the form enqueued by AofWriterPool::fsync_barrier / AppendSync { lsn: 0, bytes: Bytes::new() }); change the check around the len==0 branch to require header.lsn == 0 && len == 0 and, if len == 0 but header.lsn != 0, treat it as a header corruption/error (log/return an error instead of silently advancing offset), and update/ensure the replay_incr_framed_skips_zero_length_barrier_record test still passes.tests/workspace_integration.rs (1)
155-156: ShardSliceInit pairing is already index-stable, but remove the front-pop and assert intent.
ShardDatabases::newbuildsVec<ShardSliceInit>viaShardSliceInit::build_all(...).enumerate(), settingShardSliceInit { shard_id: <index>, ... }.tests/workspace_integration.rsbuildsall_dbsfromshardscreated with IDs0..num_shards, so positional pairing already corresponds to shard-id order.- Replace
slice_inits.remove(0)withzip(..., slice_inits.into_iter())(and optionally assertshard_slice_init.shard_id == id) to make the contract explicit.♻️ Proposed change
- let (shard_databases, mut slice_inits) = + let (shard_databases, slice_inits) = moon::shard::shared_databases::ShardDatabases::new(all_dbs); + assert_eq!( + slice_inits.len(), + num_shards, + "expected one ShardSliceInit per shard" + ); let mut shard_handles = Vec::with_capacity(num_shards); - for (id, mut shard) in shards.into_iter().enumerate() { + for ((id, mut shard), shard_slice_init) in + shards.into_iter().enumerate().zip(slice_inits.into_iter()) + { let producers = mesh.take_producers(id); let consumers = mesh.take_consumers(id); let conn_rx = mesh.take_conn_rx(id); @@ - let shard_slice_init = slice_inits.remove(0); - let handle = std::thread::Builder::new()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/workspace_integration.rs` around lines 155 - 156, The test currently pops the first element from slice_inits (slice_inits.remove(0)) to pair shard databases, but ShardSliceInit::build_all already assigns shard_id by enumeration so the positional order is stable; change the pairing to use zip(shard_databases, slice_inits.into_iter()) instead of removing the front, and add an assert like assert_eq!(shard_slice_init.shard_id, id) inside the loop to make the positional contract explicit; this touches the call site that invokes ShardDatabases::new and iterates over slice_inits and uses ShardSliceInit values.tests/txn_ft_search_snapshot.rs (1)
171-230: ⚡ Quick winMake the C1 shard/init handoff explicit in both test harnesses.
Both
tests/txn_ft_search_snapshot.rsandtests/txn_kv_wiring.rsvalidate the new boot path by pullingShardSliceInitvalues withslice_inits.remove(0). That keeps the happy path working, but ifShardDatabases::newever stops returning exactly one init per shard, both harnesses will fail with a generic index panic instead of a targeted contract failure. Assertslice_inits.len() == num_shardsonce, then zipshards.into_iter().enumerate()withslice_inits.into_iter()so the shard/init pairing is explicit in both files.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/txn_ft_search_snapshot.rs` around lines 171 - 230, The test currently relies on slice_inits.remove(0) which causes an index panic if ShardDatabases::new no longer returns exactly one init per shard; instead assert that slice_inits.len() == num_shards and replace the for (id, mut shard) in shards.into_iter().enumerate() { ... let shard_slice_init = slice_inits.remove(0); ... } pattern by iterating shards and slice_inits together (e.g., zip shards.into_iter().enumerate() with slice_inits.into_iter()) so each shard gets its corresponding ShardSliceInit explicitly; update both tests/txn_ft_search_snapshot.rs and tests/txn_kv_wiring.rs to perform the length assertion and use the zipped iteration to bind shard_slice_init rather than remove(0).
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/persistence/aof/pool.rs`:
- Around line 404-418: The try_send_append function currently swallows
flume::TrySendError and only logs a warning, which lets callers (e.g. the
EverySec/No ack paths) treat an append as successful even when it was dropped;
change try_send_append to return a Result<(), Error> (or Result<bool, Error>)
and propagate the flume error to the caller instead of just logging: update the
signature of try_send_append, return Err when
sender(shard_id).try_send(AofMessage::Append { lsn, bytes }) fails (map
flume::TrySendError to your crate error type), and adjust callers in the
EverySec/No code paths to handle the failure (abort/return error to the client)
rather than proceeding as if the append succeeded. Ensure references to
try_send_append, sender, and AofMessage::Append are updated accordingly.
- Around line 138-170: per_shard_with_fold_channels (and set_fold_channels)
currently only use debug_assert_eq! for vector parity, letting a miswire slip in
release builds and causing try_send_rewrite_per_shard to later index with
expect(...) and panic; change these APIs to perform runtime validation of
lengths/contents and refuse (return a Result::Err or log+no-op) instead of
storing bad wiring: replace debug_assert_eq! checks in
per_shard_with_fold_channels and set_fold_channels with real checks that return
an error (or Result<Arc<Self>, Error>) when senders.len() !=
fold_producers.len() or != fold_notifiers.len(), and update
try_send_rewrite_per_shard to not rely on expect() — handle the invalid state
gracefully (return Err or skip rewrite) and log a clear error; update
callers/tests to handle the new Result/option behavior.
In `@src/persistence/aof/rewrite.rs`:
- Around line 871-886: AofMessage::RewriteSharded can still be sent/handled but
do_rewrite_sharded/rewrite_aof_sharded_sync are now hard-fail stubs, so either
prevent that message from being produced or route it to the PerShard
implementation; update the producer at try_send_rewrite (where
AofMessage::RewriteSharded is created in src/command/persistence.rs) to emit the
PerShard variant (or include layout check and choose per-shard), or change the
consumer in writer_task.rs that matches on AofMessage::RewriteSharded to call
the working per-shard entry (e.g., rewrite_aof_per_shard_sync /
do_rewrite_per_shard) or return a guarded error before dispatch; additionally
ensure any required fold_producers/fold_notifiers plumbing is provided when
routing to the per-shard path so the call signatures match.
In `@src/persistence/aof/writer_task.rs`:
- Around line 1032-1051: The Ok-only handling of drain_pending_appends_framed
and the subsequent sync_and_fulfill_drain error path can leave write_error unset
and continue appending into a potentially torn framed stream; update the branch
that calls drain_pending_appends_framed and the sync_and_fulfill_drain Err case
to set the shared write_error (or similar failure flag used in this function) to
the I/O error returned, stop further appends for this shard, and ensure any
resources are cleaned or the loop exits; specifically modify the code around
drain_pending_appends_framed(...) and sync_and_fulfill_drain(...) to propagate
errors into write_error (and not just log them) so later logic sees the failure
for shard_id (and avoid backdating last_fsync on error).
In `@src/server/conn/handler_monoio/txn.rs`:
- Around line 143-159: The commit path currently awaits reply_rx.recv().await
with no timeout after sending ShardMessage::MqTxnMaterialize via spsc_send,
which can hang forever if the owner shard wedges; update the code that calls
spsc_send(&ctx.dispatch_tx, ctx.shard_id, owner, msg, &ctx.spsc_notifiers).await
to await the reply with a bounded timeout/cancellation (e.g., wrap
reply_rx.recv() in the runtime timeout or cancellation API used elsewhere in
handler_monoio/mod.rs) and treat a timeout or cancelled wait as an explicit
failure that returns an error response to the client instead of hanging; ensure
the failure path cleans up any state and logs context (db_index, shard_id,
owner) so the caller of ShardMessage::MqTxnMaterialize observes a timely error.
In `@src/shard/event_loop.rs`:
- Around line 1581-1595: The monoio shutdown path is missing the same
graph-store persistence performed in the tokio branch; ensure the monoio
shutdown block calls the same logic that saves
crate::shard::slice::with_shard(|s| { if s.graph_store.graph_count() > 0 {
crate::graph::recovery::save_graph_store(...) } }) so the graph_store is written
to disk on monoio runtime shutdown as well (use the same dir and shard_id and
log warnings/errors similarly via tracing::warn! / info!).
In `@src/shard/mq_exec.rs`:
- Around line 238-248: The MqCreate WAL record must include the current db_index
so replays can target the correct DB: modify crate::mq::wal::encode_mq_create to
accept and serialize db_index (in addition to eff_key and max_delivery_count),
update the call site in handle_create (the block that constructs payload and
calls wal_append_on_slice / WalRecordType::MqCreate) to pass the current
db_index, and update the MqCreate replay/decoder path to read that db_index and
switch to/apply against that DB before restoring the queue state; reference
encode_mq_create, handle_create (the WAL write block), wal_append_on_slice, and
WalRecordType::MqCreate when making these changes.
- Around line 435-463: The current logic WAL-logs every requested id even when
stream.xack only acknowledges a subset; change the flow to compute the actual
acked IDs and persist only those: before calling stream.xack (or by changing
stream.xack to return the acked id list), determine the ackable subset from the
stream (e.g. check PEL membership for each id from msg_ids/ids or have
stream.xack return Vec<MsgId>), call xack with that subset (or keep xack as-is
but use its returned acked ids), and then iterate over that acked_ids when
calling crate::mq::wal::encode_mq_ack and wal_append_on_slice; update references
to with_shard_db, db.get_stream_mut, stream.xack, msg_ids/ids, encode_mq_ack,
and wal_append_on_slice accordingly so only actually-acked message ids are
written to WAL.
In `@src/shard/scatter_aggregate.rs`:
- Around line 56-63: Replace the panicking expect() calls in the with_shard
closure with a non-panicking error return: check s.databases.first() (and the
analogous check at the other site around lines 87-93) and if it is None return a
protocol error by returning Frame::Error(...) with an appropriate error Bytes
payload instead of calling expect/unwrap; update the closure in
crate::shard::slice::with_shard so the variable db is obtained via match/if-let
and any missing-db path returns Frame::Error(Bytes::from(...)) rather than
panicking.
In `@src/shard/shared_databases.rs`:
- Around line 397-437: The replay functions replay_mq_wal and
replay_temporal_wal are scanning shard-{id}/ instead of the nested
shard-{id}/wal-v3/ directory, so replay_wal_v3_file never sees the v3 files;
update the wal_dir construction in both functions (the code that builds wal_dir
from persistence_dir and shard_id) to point to the wal-v3 subdirectory (e.g.,
persistence_dir.join(format!("shard-{}")).join("wal-v3") or equivalent) before
listing files and call replay_wal_v3_file as before so the WAL v3 files are
discovered and replayed.
In `@src/shard/slice.rs`:
- Around line 150-179: build_all currently hard-codes
ShardSliceInit::wal_append_tx to None which leaves each ShardSlice with a
permanently stale WAL sender and causes slice-based writes to skip WAL; fix by
wiring the real mpsc_bounded sender into each slice after the sender is created
in startup (where Shard::run installs it into ShardDatabases) — locate build_all
and Shard::run and, after creating the mpsc_bounded sender, iterate the
Vec<ShardSliceInit> (or set the field when converting to ShardSlice) to set
ShardSlice::wal_append_tx = Some(sender.clone()) for the matching shard_id, or
alternatively remove the wal_append_tx field from ShardSlice and always route
WAL enqueue via the ShardDatabases path consistently.
- Around line 363-374: assert_initialized currently only checks is_initialized()
and therefore allows a shard thread to run with the wrong ShardSlice; update it
to also validate the initialized shard identity by comparing the stored init
shard id (from the global/once-initialized ShardSlice/ShardSliceInit instance)
against the shard_id parameter passed to assert_initialized, and panic with a
clear message if they differ. Locate assert_initialized and the global/once cell
that holds ShardSliceInit (or the accessor that returns the installed
ShardSlice) and add an explicit check like "installed_shard.id == shard_id" (or
equivalent accessor) so a mismatch aborts startup rather than silently using the
wrong slice. Ensure the panic message references both expected and actual shard
ids to aid debugging.
In `@tests/shardslice_shape.rs`:
- Around line 78-93: split_off_test_module currently records the last
#[cfg(test)] location unconditionally which can cut off production code if the
following non-blank line is not a test module; update split_off_test_module to,
upon seeing a trimmed == "#[cfg(test)]", scan subsequent lines (skipping blank
lines) and only set cfg_test_pos when the first non-blank trimmed line after it
is a test module declaration (e.g., starts_with "mod " and typically "mod tests"
or "mod tests {"/"mod tests;"); otherwise ignore that #[cfg(test)] and continue
scanning. Keep the same byte_offset calculation and return behavior (use the
recorded pos to slice &text[..pos] when set).
---
Outside diff comments:
In `@src/main.rs`:
- Around line 1296-1303: The readiness flip is happening too early — call to
moon::admin::metrics_setup::set_server_ready() and readiness_flag.store(true,
...) should be moved until after every shard has processed its ShardSliceInit
and entered Shard::run(...) following init_shard; add an explicit per-shard
startup ack/barrier (e.g., a oneshot/async channel or Barrier) that each shard
signals once init_shard completes and its run loop has started, wait for all
those acks in the main thread, then register shard DBs
(set_global_shard_databases(&shard_databases)) and finally call
set_server_ready() and store the readiness_flag; use the existing
shard-identifying symbols (init_shard, Shard::run, ShardSliceInit,
readiness_flag, set_server_ready) to locate where to wire the barrier and wait.
In `@src/persistence/aof/pool.rs`:
- Around line 744-759: The test no longer reaches the disconnected-tx fan-out
path because AofWriterPool::per_shard_with_base_dir was constructed without fold
channels so try_send_rewrite_per_shard returns early; fix by wiring dummy fold
channels when building the pool (so the path reaches the per-shard send and
exercises tx0/tx1 failure) or alternatively rename the test to reflect it only
verifies the missing-channel guard; specifically update the pool construction in
this test to supply fold channel senders/receivers matching the pool's expected
shape (so try_send_rewrite_per_shard executes the fan-out to tx0/tx1) or change
the test name/docstring to indicate it asserts the missing-channel guard instead
of partial fan-out failure.
In `@src/server/conn/handler_monoio/txn.rs`:
- Around line 57-83: conn.active_cross_txn.take() is removing the cross-store
txn but when the snapshot is killed the early return after abort_killed(...)
leaves kv_write_intents and deferred HNSW inserts uncleared; call the same
cleanup used on disconnect so the txn is fully aborted before returning the
error. Specifically, when killed is true invoke
abort_cross_store_txn_routed(...) (or the module's equivalent cleanup function)
passing the taken txn to release kv_write_intents and drain deferred HNSW
inserts, then emit the error response and return; ensure this cleanup runs while
you still own the taken txn (refer to conn.active_cross_txn.take(),
abort_killed/commit, and abort_cross_store_txn_routed).
In `@src/server/conn/handler_sharded/ft.rs`:
- Around line 633-728: The single-shard dispatcher accidentally dropped the
FT.INVALIDATE_RANGE arm so the command falls through to "ERR unknown FT.*
command"; restore it by adding an else-if branch for
cmd.eq_ignore_ascii_case(b"FT.INVALIDATE_RANGE") that calls the vector-search
handler (e.g. crate::command::vector_search::ft_invalidate_range) and borrow
stores the same way FT.DROPINDEX does (let (vs, ts, dbs) = (&mut s.vector_store,
&mut s.text_store, &mut s.databases); then call ft_invalidate_range(vs, ts,
Some(&mut dbs[0]), cmd_args)). Ensure the branch is placed alongside the other
FT.* conditionals inside with_shard so single-shard behavior matches
handler_monoio/ft.rs.
In `@src/server/conn/handler_sharded/mod.rs`:
- Line 373: The batching uses remote_groups keyed only by target shard which
causes different DB indices to be mixed; change the grouping key to include both
target_shard and db_index (e.g., use (usize, usize) or a small struct) wherever
remote_groups is declared/used, ensure pushes populate the tuple key using the
entry's db_index instead of assuming the first entry's DB, and update the code
that iterates/executes groups (the logic around remote_groups, target_shard and
the per-entry db selection) so each batch executes against the correct DB index
rather than a shard-only DB.
In `@src/server/conn/handler_sharded/txn.rs`:
- Around line 58-84: When the killed-snapshot branch rejects TXN.COMMIT it
returns after only touching the vector-store txn manager, leaving
conn.active_cross_txn resources (txn.kv_undo, vector intents, MQ intents and
s.kv_write_intents) unrolled and stranded; before returning on was_killed you
must perform the same staged-transaction abort/cleanup that disconnect cleanup
would do: take conn.active_cross_txn (as already done), call the rollback/undo
routine for txn.kv_undo, clear any vector intents and MQ intents associated with
txn.txn_id, and remove/cleanup any s.kv_write_intents entries for that txn (or
invoke the existing rollback helper if one exists) so the transaction is fully
aborted before pushing the error response and returning.
In `@src/shard/scatter_aggregate.rs`:
- Around line 229-259: The test calls scatter_text_aggregate (fast-path
single-shard) but never installs the shard slice returned by
ShardDatabases::new, causing with_shard to panic; fix by initializing the test
shard on the current thread before calling scatter_text_aggregate: invoke the
test shard installer (e.g., call init_shard or reset_test_shard using the
returned shard init bundle from ShardDatabases::new or otherwise install the
ShardSlice) so that with_shard/ShardSlice is initialized for this test thread
prior to calling scatter_text_aggregate.
In `@src/shard/spsc_handler.rs`:
- Around line 1-2772: File is too large; split into smaller submodules and
rewire exports. Move top-level SPSC loop into a spsc_drain.rs containing
drain_spsc_shared and the thread-local scratch, move message handling into
handler_shared.rs with handle_shard_message_shared and related ShardMessage
arms, extract vector logic into vector_dispatch.rs with dispatch_vector_command
and related helpers (has_session_keyword, dispatch_vector_command internals),
extract auto-indexing into auto_index.rs with
auto_index_hset_public/auto_index_hset and helpers (find_vector_blob,
handle_vector_insert*, update_metadata_only, index_payload_field,
parse_geo_value), and extract WAL/AOF fanout into wal.rs with
wal_append_and_fanout and its tests; split tests into matching test modules
(wal_append_tests, drain_cap_tests) placed next to their code. Update the parent
module to mod-declare and pub(crate) use the moved symbols (drain_spsc_shared,
handle_shard_message_shared, dispatch_vector_command, auto_index_hset_public,
wal_append_and_fanout, etc.), adjust imports (crate::... paths) and visibility
as needed, and run cargo test to fix any broken uses or privacy issues.
In `@tests/txn_kv_wiring.rs`:
- Around line 26-258: The file exceeds the 1500-line limit because reusable test
harness and helpers (notably start_txn_server, ChannelMesh/shard boot logic, and
related helper data structures) are embedded in tests/txn_kv_wiring.rs; extract
these into a new tests/support module (e.g., tests/support/mod.rs or
tests/support/harness.rs). Move start_txn_server, ChannelMesh creation, shard
thread spawn logic, and any helper types/constants into that module, make the
functions/types pub(crate) as needed, update tests/txn_kv_wiring.rs to use
crate::support::start_txn_server (or tests::support) and adjust imports, and run
cargo test to ensure visibility and paths compile; keep test-specific cases in
txn_kv_wiring.rs so its lines drop below 1500.
---
Nitpick comments:
In `@src/persistence/aof_manifest/shard_replay.rs`:
- Around line 257-263: The current zero-length-frame handling in shard_replay.rs
simply skips any record with len == 0; tighten this defense-in-depth by only
treating a zero-length frame as the fsync barrier when its header indicates lsn
== 0 (the form enqueued by AofWriterPool::fsync_barrier / AppendSync { lsn: 0,
bytes: Bytes::new() }); change the check around the len==0 branch to require
header.lsn == 0 && len == 0 and, if len == 0 but header.lsn != 0, treat it as a
header corruption/error (log/return an error instead of silently advancing
offset), and update/ensure the
replay_incr_framed_skips_zero_length_barrier_record test still passes.
In `@tests/txn_ft_search_snapshot.rs`:
- Around line 171-230: The test currently relies on slice_inits.remove(0) which
causes an index panic if ShardDatabases::new no longer returns exactly one init
per shard; instead assert that slice_inits.len() == num_shards and replace the
for (id, mut shard) in shards.into_iter().enumerate() { ... let shard_slice_init
= slice_inits.remove(0); ... } pattern by iterating shards and slice_inits
together (e.g., zip shards.into_iter().enumerate() with slice_inits.into_iter())
so each shard gets its corresponding ShardSliceInit explicitly; update both
tests/txn_ft_search_snapshot.rs and tests/txn_kv_wiring.rs to perform the length
assertion and use the zipped iteration to bind shard_slice_init rather than
remove(0).
In `@tests/workspace_integration.rs`:
- Around line 155-156: The test currently pops the first element from
slice_inits (slice_inits.remove(0)) to pair shard databases, but
ShardSliceInit::build_all already assigns shard_id by enumeration so the
positional order is stable; change the pairing to use zip(shard_databases,
slice_inits.into_iter()) instead of removing the front, and add an assert like
assert_eq!(shard_slice_init.shard_id, id) inside the loop to make the positional
contract explicit; this touches the call site that invokes ShardDatabases::new
and iterates over slice_inits and uses ShardSliceInit values.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 9b51ea18-dbdf-467e-8398-a2a279a9c175
📒 Files selected for processing (73)
.add/state.json.add/tasks/shardslice-migration/TASK.mdsrc/admin/metrics_setup.rssrc/command/persistence.rssrc/command/server_admin.rssrc/main.rssrc/persistence/aof/mod.rssrc/persistence/aof/pool.rssrc/persistence/aof/rewrite.rssrc/persistence/aof/writer_task.rssrc/persistence/aof_manifest/shard_replay.rssrc/replication/master.rssrc/server/conn/blocking.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/ft.rssrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_monoio/txn.rssrc/server/conn/handler_monoio/write.rssrc/server/conn/handler_sharded/dispatch.rssrc/server/conn/handler_sharded/ft.rssrc/server/conn/handler_sharded/mod.rssrc/server/conn/handler_sharded/txn.rssrc/server/conn/handler_sharded/write.rssrc/server/conn/handler_single.rssrc/server/conn/shared.rssrc/server/conn/tests.rssrc/server/embedded.rssrc/shard/coordinator.rssrc/shard/dispatch.rssrc/shard/event_loop.rssrc/shard/mesh.rssrc/shard/mod.rssrc/shard/mq_exec.rssrc/shard/persistence_tick.rssrc/shard/scatter_aggregate.rssrc/shard/scatter_hybrid.rssrc/shard/shared_databases.rssrc/shard/slice.rssrc/shard/spsc_handler.rssrc/shard/timers.rssrc/shard/uring_handler.rssrc/transaction/abort.rstests/adversarial_v0110_fix01_set_delete_rollback.rstests/adversarial_v0110_fix02_err_path_intent.rstests/adversarial_v0110_fix03_simplestring_graph.rstests/adversarial_v0110_fix04_shortest_path_call_parity.rstests/adversarial_v0110_fix06_shortest_path_min_hops.rstests/adversarial_v0110_fix07_multihop_edge_var_reject.rstests/crash_matrix_per_shard_bgrewriteaof.rstests/ft_search_as_of_boundary.rstests/ft_search_as_of_filter.rstests/ft_search_concurrent_readers.rstests/ft_search_multi_shard_as_of.rstests/ft_search_temporal_parity.rstests/hybrid_filter_backward_compat.rstests/hybrid_filter_multishard.rstests/hybrid_filter_tag.rstests/integration.rstests/kill_snapshot.rstests/lunaris_cypher_shortest_path.rstests/lunaris_cypher_temporal.rstests/lunaris_hybrid_ft_search.rstests/mq_integration.rstests/pipeline_auto_index.rstests/quickwins_red.rstests/shardslice_live.rstests/shardslice_shape.rstests/txn_completeness_edge_cases.rstests/txn_cypher_write_rollback.rstests/txn_ft_search_snapshot.rstests/txn_graph_wiring.rstests/txn_kv_wiring.rstests/workspace_integration.rs
Comment on lines
+138
to
170
| pub fn per_shard_with_fold_channels( | ||
| senders: Vec<channel::MpscSender<AofMessage>>, | ||
| fsync_policy: FsyncPolicy, | ||
| fsync_timeout: Duration, | ||
| base_dir: PathBuf, | ||
| fold_producers: Vec< | ||
| Arc<parking_lot::Mutex<ringbuf::HeapProd<crate::shard::dispatch::ShardMessage>>>, | ||
| >, | ||
| fold_notifiers: Vec<Arc<crate::runtime::channel::Notify>>, | ||
| ) -> Arc<Self> { | ||
| debug_assert!( | ||
| senders.len() >= 2, | ||
| "per_shard pool needs >=2 writers; use top_level for single-writer" | ||
| ); | ||
| debug_assert_eq!( | ||
| senders.len(), | ||
| fold_producers.len(), | ||
| "fold_producers must have one entry per shard" | ||
| ); | ||
| debug_assert_eq!( | ||
| senders.len(), | ||
| fold_notifiers.len(), | ||
| "fold_notifiers must have one entry per shard" | ||
| ); | ||
| Arc::new(Self { | ||
| senders, | ||
| layout: crate::persistence::aof_manifest::AofLayout::PerShard, | ||
| fsync_policy, | ||
| fsync_timeout, | ||
| base_dir: Some(base_dir), | ||
| fold_producers: Some(fold_producers), | ||
| fold_notifiers: Some(fold_notifiers), | ||
| }) |
There was a problem hiding this comment.
Reject mis-sized fold-channel wiring in release builds.
per_shard_with_fold_channels and set_fold_channels only validate vector parity with debug_assert_eq!, but try_send_rewrite_per_shard() later indexes those vectors with expect(...). In release builds, a startup miswire turns BGREWRITEAOF into a panic instead of a clean rewrite refusal. Please add runtime validation here and refuse/log bad wiring before storing it.
As per coding guidelines, "No unwrap() or expect() in library code outside tests."
Also applies to: 189-220, 645-657
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/persistence/aof/pool.rs` around lines 138 - 170,
per_shard_with_fold_channels (and set_fold_channels) currently only use
debug_assert_eq! for vector parity, letting a miswire slip in release builds and
causing try_send_rewrite_per_shard to later index with expect(...) and panic;
change these APIs to perform runtime validation of lengths/contents and refuse
(return a Result::Err or log+no-op) instead of storing bad wiring: replace
debug_assert_eq! checks in per_shard_with_fold_channels and set_fold_channels
with real checks that return an error (or Result<Arc<Self>, Error>) when
senders.len() != fold_producers.len() or != fold_notifiers.len(), and update
try_send_rewrite_per_shard to not rely on expect() — handle the invalid state
gracefully (return Err or skip rewrite) and log a clear error; update
callers/tests to handle the new Result/option behavior.
Source: Coding guidelines
Comment on lines
404
to
+418
| pub fn try_send_append(&self, shard_id: usize, lsn: u64, bytes: Bytes) { | ||
| let _ = self | ||
| if let Err(e) = self | ||
| .sender(shard_id) | ||
| .try_send(AofMessage::Append { lsn, bytes }); | ||
| .try_send(AofMessage::Append { lsn, bytes }) | ||
| { | ||
| warn!( | ||
| "AOF append dropped for shard {} (lsn {}): channel {}", | ||
| shard_id, | ||
| lsn, | ||
| match e { | ||
| flume::TrySendError::Full(_) => "full", | ||
| flume::TrySendError::Disconnected(_) => "disconnected", | ||
| } | ||
| ); | ||
| } |
There was a problem hiding this comment.
Don't turn dropped AOF appends into warning-only success.
This still drops the append on Full/Disconnected without surfacing failure to the caller. On the EverySec/No path that means the command can still be acknowledged even though the AOF entry never reached the writer, which is persisted-data loss rather than a logging concern.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/persistence/aof/pool.rs` around lines 404 - 418, The try_send_append
function currently swallows flume::TrySendError and only logs a warning, which
lets callers (e.g. the EverySec/No ack paths) treat an append as successful even
when it was dropped; change try_send_append to return a Result<(), Error> (or
Result<bool, Error>) and propagate the flume error to the caller instead of just
logging: update the signature of try_send_append, return Err when
sender(shard_id).try_send(AofMessage::Append { lsn, bytes }) fails (map
flume::TrySendError to your crate error type), and adjust callers in the
EverySec/No code paths to handle the failure (abort/return error to the client)
rather than proceeding as if the append succeeded. Ensure references to
try_send_append, sender, and AofMessage::Append are updated accordingly.
Comment on lines
+143
to
+159
| // Foreign: send MqTxnMaterialize hop and await ack. | ||
| let (reply_tx, reply_rx) = crate::runtime::channel::oneshot(); | ||
| let msg = crate::shard::dispatch::ShardMessage::MqTxnMaterialize { | ||
| db_index: conn.selected_db as usize, | ||
| intents, | ||
| reply_tx, | ||
| }; | ||
| crate::shard::coordinator::spsc_send( | ||
| &ctx.dispatch_tx, | ||
| ctx.shard_id, | ||
| owner, | ||
| msg, | ||
| &ctx.spsc_notifiers, | ||
| ) | ||
| .await; | ||
| // Await the ack before replying OK to the client. | ||
| let _ = reply_rx.recv().await; |
There was a problem hiding this comment.
Don't let TXN.COMMIT wait forever on foreign MQ materialization.
After spsc_send(...), Line 159 does an unbounded reply_rx.recv().await. If the owner shard accepts the hop and then wedges, this monoio task never finishes the commit and the client connection stays parked indefinitely. src/server/conn/handler_monoio/mod.rs already added bounded cross-shard waits for batched dispatch for the same reason; this path needs the same timeout/cancel handling and an explicit failure instead of a permanent hang.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/server/conn/handler_monoio/txn.rs` around lines 143 - 159, The commit
path currently awaits reply_rx.recv().await with no timeout after sending
ShardMessage::MqTxnMaterialize via spsc_send, which can hang forever if the
owner shard wedges; update the code that calls spsc_send(&ctx.dispatch_tx,
ctx.shard_id, owner, msg, &ctx.spsc_notifiers).await to await the reply with a
bounded timeout/cancellation (e.g., wrap reply_rx.recv() in the runtime timeout
or cancellation API used elsewhere in handler_monoio/mod.rs) and treat a timeout
or cancelled wait as an explicit failure that returns an error response to the
client instead of hanging; ensure the failure path cleans up any state and logs
context (db_index, shard_id, owner) so the caller of
ShardMessage::MqTxnMaterialize observes a timely error.
Comment on lines
+56
to
+63
| let result = crate::shard::slice::with_shard(|s| { | ||
| // Every shard slice is constructed with at least one database | ||
| // (db 0); see Shard::new in src/shard/mod.rs. | ||
| #[allow(clippy::expect_used)] | ||
| let db = s | ||
| .databases | ||
| .first() | ||
| .expect("shard slice must have at least db 0"); |
There was a problem hiding this comment.
Return Frame::Error here instead of panicking.
These new expect() calls turn a bad shard setup into a process panic in library code. This path should surface a protocol error, not crash the server.
Suggested fix
- #[allow(clippy::expect_used)]
- let db = s
- .databases
- .first()
- .expect("shard slice must have at least db 0");
+ let Some(db) = s.databases.first() else {
+ return Frame::Error(Bytes::from_static(b"ERR shard has no databases"));
+ };Apply the same change in the local-partial branch.
As per coding guidelines, "All command errors return Frame::Error(Bytes)" and "No unwrap() or expect() in library code outside tests."
Also applies to: 87-93
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/shard/scatter_aggregate.rs` around lines 56 - 63, Replace the panicking
expect() calls in the with_shard closure with a non-panicking error return:
check s.databases.first() (and the analogous check at the other site around
lines 87-93) and if it is None return a protocol error by returning
Frame::Error(...) with an appropriate error Bytes payload instead of calling
expect/unwrap; update the closure in crate::shard::slice::with_shard so the
variable db is obtained via match/if-let and any missing-db path returns
Frame::Error(Bytes::from(...)) rather than panicking.
Source: Coding guidelines
Comment on lines
+397
to
+437
| for init in inits.iter_mut() { | ||
| let shard_id = init.shard_id; | ||
| let wal_dir = persistence_dir.join(format!("shard-{}", shard_id)); | ||
| if !wal_dir.exists() { | ||
| continue; | ||
| } | ||
|
|
||
| let on_command = &mut |record: &WalRecord| match record.record_type { | ||
| WalRecordType::MqCreate => { | ||
| if let Some((queue_key, max_delivery_count)) = | ||
| crate::mq::wal::decode_mq_create(&record.payload) | ||
| { | ||
| durable_configs.insert(queue_key, max_delivery_count); | ||
| } | ||
| } | ||
| WalRecordType::MqAck => { | ||
| if crate::mq::wal::decode_mq_ack(&record.payload).is_some() { | ||
| ack_count += 1; | ||
| } | ||
| let mut durable_configs: HashMap<Vec<u8>, u32> = HashMap::new(); | ||
| let mut ack_count = 0u64; | ||
|
|
||
| let on_command = &mut |record: &WalRecord| match record.record_type { | ||
| WalRecordType::MqCreate => { | ||
| if let Some((queue_key, max_delivery_count)) = | ||
| crate::mq::wal::decode_mq_create(&record.payload) | ||
| { | ||
| durable_configs.insert(queue_key, max_delivery_count); | ||
| } | ||
| _ => {} | ||
| }; | ||
| let on_fpi = &mut |_: &WalRecord| {}; | ||
|
|
||
| // Scan WAL files in the shard directory | ||
| if let Ok(entries) = std::fs::read_dir(&wal_dir) { | ||
| let mut wal_files: Vec<_> = entries | ||
| .filter_map(|e| e.ok()) | ||
| .filter(|e| e.file_name().to_str().is_some_and(|n| n.ends_with(".wal"))) | ||
| .map(|e| e.path()) | ||
| .collect(); | ||
| wal_files.sort(); | ||
|
|
||
| for wal_file in &wal_files { | ||
| let _ = crate::persistence::wal_v3::replay::replay_wal_v3_file( | ||
| wal_file, 0, on_command, on_fpi, | ||
| ); | ||
| } | ||
| WalRecordType::MqAck => { | ||
| if crate::mq::wal::decode_mq_ack(&record.payload).is_some() { | ||
| ack_count += 1; | ||
| } | ||
| } | ||
| _ => {} | ||
| }; | ||
| let on_fpi = &mut |_: &WalRecord| {}; | ||
|
|
||
| if let Ok(entries) = std::fs::read_dir(&wal_dir) { | ||
| let mut wal_files: Vec<_> = entries | ||
| .filter_map(|e| e.ok()) | ||
| .filter(|e| e.file_name().to_str().is_some_and(|n| n.ends_with(".wal"))) | ||
| .map(|e| e.path()) | ||
| .collect(); | ||
| wal_files.sort(); | ||
|
|
||
| for wal_file in &wal_files { | ||
| let _ = crate::persistence::wal_v3::replay::replay_wal_v3_file( | ||
| wal_file, 0, on_command, on_fpi, | ||
| ); | ||
| } | ||
| } |
There was a problem hiding this comment.
wal-v3 replay is scanning the wrong directory.
These two functions call replay_wal_v3_file(...), but they read shard-{id}/ instead of shard-{id}/wal-v3/. replay_workspace_wal in the same module already uses the nested wal-v3 directory, so MQ and temporal replay will quietly miss their persisted records on normal boots. That means durable queue metadata and temporal bindings can disappear after restart.
Suggested fix
- let wal_dir = persistence_dir.join(format!("shard-{}", shard_id));
+ let wal_dir = persistence_dir
+ .join(format!("shard-{}", shard_id))
+ .join("wal-v3");Apply the same change in both replay_mq_wal(...) and replay_temporal_wal(...).
Also applies to: 584-659
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/shard/shared_databases.rs` around lines 397 - 437, The replay functions
replay_mq_wal and replay_temporal_wal are scanning shard-{id}/ instead of the
nested shard-{id}/wal-v3/ directory, so replay_wal_v3_file never sees the v3
files; update the wal_dir construction in both functions (the code that builds
wal_dir from persistence_dir and shard_id) to point to the wal-v3 subdirectory
(e.g., persistence_dir.join(format!("shard-{}")).join("wal-v3") or equivalent)
before listing files and call replay_wal_v3_file as before so the WAL v3 files
are discovered and replayed.
Comment on lines
+150
to
+179
| pub fn build_all( | ||
| shard_databases: Vec<Vec<Database>>, | ||
| memory_per_shard: &[Arc<AtomicUsize>], | ||
| store_memory_per_shard: &[Arc<ShardStoreMemory>], | ||
| ) -> Vec<ShardSliceInit> { | ||
| shard_databases | ||
| .into_iter() | ||
| .enumerate() | ||
| .map(|(shard_id, dbs)| { | ||
| let databases: Box<[Database]> = dbs.into_boxed_slice(); | ||
| ShardSliceInit { | ||
| shard_id, | ||
| databases, | ||
| vector_store: VectorStore::new(), | ||
| text_store: TextStore::new(), | ||
| #[cfg(feature = "graph")] | ||
| graph_store: GraphStore::new(), | ||
| kv_write_intents: KvWriteIntents::new(), | ||
| deferred_hnsw_inserts: DeferredHnswInserts::new(), | ||
| temporal_registry: None, | ||
| temporal_kv_index: None, | ||
| durable_queue_registry: None, | ||
| trigger_registry: None, | ||
| wal_append_tx: None, | ||
| estimated_memory: memory_per_shard[shard_id].clone(), | ||
| store_memory: store_memory_per_shard[shard_id].clone(), | ||
| } | ||
| }) | ||
| .collect() | ||
| } |
There was a problem hiding this comment.
Don't seed a permanently stale wal_append_tx.
build_all() hard-codes wal_append_tx: None, and Shard::run() later installs the real sender only into ShardDatabases. That leaves ShardSlice::wal_append_tx wrong whenever appendonly or disk-offload is enabled, so any slice-based write path will observe None and skip WAL enqueue. Either populate the slice after mpsc_bounded() in startup or remove this field from ShardSlice entirely.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/shard/slice.rs` around lines 150 - 179, build_all currently hard-codes
ShardSliceInit::wal_append_tx to None which leaves each ShardSlice with a
permanently stale WAL sender and causes slice-based writes to skip WAL; fix by
wiring the real mpsc_bounded sender into each slice after the sender is created
in startup (where Shard::run installs it into ShardDatabases) — locate build_all
and Shard::run and, after creating the mpsc_bounded sender, iterate the
Vec<ShardSliceInit> (or set the field when converting to ShardSlice) to set
ShardSlice::wal_append_tx = Some(sender.clone()) for the matching shard_id, or
alternatively remove the wal_append_tx field from ShardSlice and always route
WAL enqueue via the ShardDatabases path consistently.
Comment on lines
+363
to
+374
| pub fn assert_initialized(shard_id: usize) { | ||
| if !is_initialized() { | ||
| // Startup abort: the shard event loop entered without init_shard. | ||
| // Panic here (before the first accept) so the process terminates | ||
| // before any command can be answered on an uninitialized thread. | ||
| panic!( | ||
| "ShardSlice not initialized on shard thread {} — \ | ||
| init_shard must be called before the accept/drain loop. \ | ||
| This is a startup-configuration bug, not a runtime error.", | ||
| shard_id | ||
| ); | ||
| } |
There was a problem hiding this comment.
Validate the shard identity here, not just initialization.
This only proves that some ShardSlice is installed. If shard 2 is accidentally started with shard 3's ShardSliceInit, assert_initialized(2) still passes and the thread will run on the wrong databases/stores.
Suggested fix
pub fn assert_initialized(shard_id: usize) {
- if !is_initialized() {
- // Startup abort: the shard event loop entered without init_shard.
- // Panic here (before the first accept) so the process terminates
- // before any command can be answered on an uninitialized thread.
- panic!(
- "ShardSlice not initialized on shard thread {} — \
- init_shard must be called before the accept/drain loop. \
- This is a startup-configuration bug, not a runtime error.",
- shard_id
- );
- }
+ SHARD.with(|cell| {
+ let guard = cell
+ .try_borrow()
+ .expect("assert_initialized must not run inside with_shard");
+ match guard.as_ref() {
+ Some(slice) if slice.shard_id == shard_id => {}
+ Some(slice) => panic!(
+ "ShardSlice initialized for shard {} but running on shard thread {}",
+ slice.shard_id,
+ shard_id
+ ),
+ None => panic!(
+ "ShardSlice not initialized on shard thread {} — \
+ init_shard must be called before the accept/drain loop. \
+ This is a startup-configuration bug, not a runtime error.",
+ shard_id
+ ),
+ }
+ });
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/shard/slice.rs` around lines 363 - 374, assert_initialized currently only
checks is_initialized() and therefore allows a shard thread to run with the
wrong ShardSlice; update it to also validate the initialized shard identity by
comparing the stored init shard id (from the global/once-initialized
ShardSlice/ShardSliceInit instance) against the shard_id parameter passed to
assert_initialized, and panic with a clear message if they differ. Locate
assert_initialized and the global/once cell that holds ShardSliceInit (or the
accessor that returns the installed ShardSlice) and add an explicit check like
"installed_shard.id == shard_id" (or equivalent accessor) so a mismatch aborts
startup rather than silently using the wrong slice. Ensure the panic message
references both expected and actual shard ids to aid debugging.
Comment on lines
+78
to
+93
| fn split_off_test_module(text: &str) -> &str { | ||
| let mut cfg_test_pos: Option<usize> = None; | ||
| let mut byte_offset = 0usize; | ||
|
|
||
| for line in text.lines() { | ||
| let trimmed = line.trim(); | ||
| if trimmed == "#[cfg(test)]" { | ||
| cfg_test_pos = Some(byte_offset); | ||
| } | ||
| byte_offset += line.len() + 1; // +1 for the '\n' | ||
| } | ||
|
|
||
| match cfg_test_pos { | ||
| Some(pos) => &text[..pos], | ||
| None => text, | ||
| } |
There was a problem hiding this comment.
split_off_test_module doesn’t implement its documented boundary rule.
The function currently truncates at the last #[cfg(test)] line, but never verifies the following non-blank line is mod tests. That can mis-slice files and make these shape-pin tests miss production violations or scan test code as production.
Suggested fix
fn split_off_test_module(text: &str) -> &str {
- let mut cfg_test_pos: Option<usize> = None;
- let mut byte_offset = 0usize;
-
- for line in text.lines() {
- let trimmed = line.trim();
- if trimmed == "#[cfg(test)]" {
- cfg_test_pos = Some(byte_offset);
- }
- byte_offset += line.len() + 1; // +1 for the '\n'
- }
-
- match cfg_test_pos {
- Some(pos) => &text[..pos],
- None => text,
- }
+ let lines: Vec<&str> = text.lines().collect();
+ let mut offsets = Vec::with_capacity(lines.len());
+ let mut off = 0usize;
+ for line in &lines {
+ offsets.push(off);
+ off += line.len() + 1;
+ }
+
+ for (i, line) in lines.iter().enumerate() {
+ if line.trim() == "#[cfg(test)]" {
+ let mut j = i + 1;
+ while j < lines.len() && lines[j].trim().is_empty() {
+ j += 1;
+ }
+ if j < lines.len() && lines[j].contains("mod tests") {
+ return &text[..offsets[i]];
+ }
+ }
+ }
+ text
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/shardslice_shape.rs` around lines 78 - 93, split_off_test_module
currently records the last #[cfg(test)] location unconditionally which can cut
off production code if the following non-blank line is not a test module; update
split_off_test_module to, upon seeing a trimmed == "#[cfg(test)]", scan
subsequent lines (skipping blank lines) and only set cfg_test_pos when the first
non-blank trimmed line after it is a test module declaration (e.g., starts_with
"mod " and typically "mod tests" or "mod tests {"/"mod tests;"); otherwise
ignore that #[cfg(test)] and continue scanning. Keep the same byte_offset
calculation and return behavior (use the recorded pos to slice &text[..pos] when
set).
…F at --shards 1 PR review found BGREWRITEAOF silently inoperative on the TopLevel AOF layout (always used at --shards 1): the writer routed RewriteSharded to a stub that errored "RwLock path removed", while the client still received "Background append only file rewriting started" and the AOF grew unboundedly. The frozen live test missed it because its completion poll broke out of the 30s deadline silently instead of failing. Implement the C4 cooperative fold for the TopLevel layout on both runtimes: - main.rs: wire fold channels (SPSC producer + shard-0 notifier) at pool construction for the TopLevel layout; consumer is merged into shard 0's consumer set. listener/embedded pass None (no TopLevel writer there). - do_rewrite_sharded (monoio): full fold — pre-drain (raw RESP) into the old generation, AofFold snapshot from shard 0, phase-3 mid-drain bounded by pending_aof_count (C4-DRAIN-BOUND: draining beyond it would pull post-snapshot appends into the old generation, which manifest.advance() deletes), base write, manifest advance. Clean-abort if fold channels are unwired — old generation stays authoritative. - rewrite_aof_sharded_sync (tokio): same protocol against the legacy single appendonly.aof — bounded phase-3 mid-drain, atomic tmp+rename; post-snapshot appends remain queued and reach the reopened new file. AppendSync acks in both drains are parked and resolved only after the boundary fsync (Synced/FsyncFailed), never before durability. - writer_task: thread fold channels through both writer loops; the tokio arm converts BufWriter -> std File for the sync fold, then reopens. - shardslice_live test: completion poll now panics on the 30s deadline (no silent proceed) and recognizes all three layouts, including the tokio TopLevel in-place rewrite (MOON RDB magic on appendonly.aof). Verification: VM tokio shardslice_live 6/6 (7.0s, previously 40s+ with the silent timeout), ssm4a x3 deterministic; local monoio 6/6 + crash matrix + lib suite; clippy -D warnings both feature sets; fmt; audit-unsafe + unwrap ratchet clean. Refs: #175 (shardslice-migration, contract C4) author: Tin Dang
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/server/embedded.rs (1)
140-166:⚠️ Potential issue | 🟠 Major | ⚡ Quick winWire the C4 fold path in embedded mode too.
run_embeddednow boots the sharded handler, but this path still launches the TopLevel AOF writer withfold_channels = Noneand never prepends an AOF-fold consumer to the shard loops. The new TopLevel rewrite helpers abort when fold channels are missing, so BGREWRITEAOF in embedded mode will fail by construction instead of folding shard 0 likemain.rsdoes.Also applies to: 303-367
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/persistence/aof/rewrite.rs`:
- Around line 918-920: The TopLevel phase-1 drain currently waits for rx to
empty and can block forever; capture a pre_drain_bound = rx.len() before
starting the tokio drain loop and enforce it by introducing a local drained
counter that you increment each time you handle a ShardMessage::Append or
ShardMessage::AppendSync; break the drain loop once drained >= pre_drain_bound,
then call sync_and_fulfill_drain (same change needed for the second TopLevel
loop around the 1138-1180 region). Ensure you update the loop that calls
drain_pending_appends(rx, file) to use this bounded behavior so
ShardMessage::AofFold can be sent.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 907a79fd-0024-4c9d-837c-e60b51b52bbd
📒 Files selected for processing (6)
src/main.rssrc/persistence/aof/rewrite.rssrc/persistence/aof/writer_task.rssrc/server/embedded.rssrc/server/listener.rstests/shardslice_live.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- tests/shardslice_live.rs
Comment on lines
+918
to
920
| // Phase 1: drain pre-rewrite queued appends into old incr (RAW RESP), fsync. | ||
| let mut pre_drain = drain_pending_appends(rx, file)?; | ||
| sync_and_fulfill_drain(&mut pre_drain, file, manifest.incr_path())?; |
There was a problem hiding this comment.
Bound the TopLevel phase-1 drain before starting the fold.
Both TopLevel paths still wait for the AOF queue to become empty before they send ShardMessage::AofFold. Under sustained writes that queue can stay non-empty forever, so BGREWRITEAOF never reaches the snapshot phase. The per-shard path already fixed this by snapshotting rx.len() first; the TopLevel phase-1 drain needs the same bound.
Suggested fix
- let mut pre_drain = drain_pending_appends(rx, file)?;
+ let pre_drain_bound = rx.len();
+ let mut pre_drain = drain_pending_appends_bounded(rx, file, pre_drain_bound)?;Mirror the same pre_drain_bound = rx.len() cap in the tokio phase-1 loop by incrementing a local drained counter for each Append / AppendSync message processed.
Also applies to: 1138-1180
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/persistence/aof/rewrite.rs` around lines 918 - 920, The TopLevel phase-1
drain currently waits for rx to empty and can block forever; capture a
pre_drain_bound = rx.len() before starting the tokio drain loop and enforce it
by introducing a local drained counter that you increment each time you handle a
ShardMessage::Append or ShardMessage::AppendSync; break the drain loop once
drained >= pre_drain_bound, then call sync_and_fulfill_drain (same change needed
for the second TopLevel loop around the 1138-1180 region). Ensure you update the
loop that calls drain_pending_appends(rx, file) to use this bounded behavior so
ShardMessage::AofFold can be sent.
CI (macOS, tokio) caught test_ssm4a_aof_fold_cooperative losing 266 acked INCRs across a SIGKILL — almost exactly one 8KB tokio BufWriter of ~29-byte entries. The TopLevel tokio writer enforced everysec with an interval.tick() select arm gated on last_fsync >= 1s: the gate stretched the worst-case flush cadence to ~2s (a tick arriving at 0.999s elapsed was skipped, deferring the flush a full period), and a long-lived tick arm is fairness-starvable under sustained writes — the per-shard writer already documents and avoids exactly this. The new TopLevel cooperative fold re-phased the timing enough for slow CI to land a SIGKILL inside the stretched window. Convert the loop to the validated per-shard pattern: 200ms timeout-bounded recv + post-select 1s flush deadline, so the oldest unflushed byte reaches disk at most ~1.2s after it was written regardless of message pressure. last_fsync is back-dated by 900ms after both rewrite reopens so the channel backlog drained during a blocking fold reaches disk within ~100ms + wake floor instead of a full second. monoio is unaffected (unbuffered std File — every append reaches the page cache immediately and survives process kill). Also adds the Unreleased CHANGELOG entry for the shared-nothing migration (CI changelog gate requires it). Verification: macOS tokio ssm4a x5 green; VM tokio shardslice_live 6/6; local monoio 6/6 + lib 3594; clippy -D warnings both feature sets; fmt; audit scripts clean. Refs: #175 (shardslice-migration, contract C4/H1) author: Tin Dang
This was referenced Jul 4, 2026
Merged
Merged
Merged
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.
ShardSlice Phase 4 — thread-local shared-nothing storage becomes the live path
init_shardis wired at shard startup on both runtimes, every command path runs on the thread-localShardSlice, and the RwLock/Mutex wrappers are deleted fromShardDatabases. This completes the milestone's "no per-command global locks" goal: before this PR,init_shardhad zero production call sites and all ~130 dual-branch points took the lock path.Contract (frozen @ v1, approved 2026-06-12)
ShardDatabases::newreturns(Arc<Self>, Vec<ShardSliceInit>); recovery replays before construction;init_shardruns at the top ofShard::runand logsShardSlice initialized shard=Nbefore the first accept;assert_initializedguards the loop.ShardMessagevariants (MqCommand,MqTxnMaterialize,WsDropCleanup,AofFold) route the five formerly-divergent MQ/WS/TXN groups to the owning shard.pending_aof_count, giving an exactly-once boundary (verified 10/10 serial under saturating INCR load; +1 double-apply and a fold deadlock found and fixed on the way). Newfsync_barrierpreserves theappendfsync=alwaysack for cross-shard writes.ShardDatabaseskeeps only WAL senders, the global registry, published atomics, elastic budgets, and counts.Evidence
test_ssm4a10/10 serial repeats.-D warnings+ fmt on all OS×featureset combos (incl. the previously never-compiledcfg(linux, runtime-tokio)uring path).Accepted risk (gate: RISK-ACCEPTED, owner @tindang, expires 2026-08-01)
Main's default
--cross-shard-fast-pathserved cross-shard GETs via a cross-threadRwLockread — definitionally incompatible with shared-nothing. Those cells regress (worst: single-client cross-shard GET 4 µs → ~47 µs SPSC round trip; P1 GET −16 %). Mitigations:{tag}co-location,--shards 1; follow-up task: lock-free cross-shard read acceleration.Notes for reviewers
src/command/vector_searchandsdk/are byte-identical to main. The threehybrid_filter_*test harnesses are migrated to the new boot API in the final commit..add/tasks/shardslice-migration/TASK.md.Summary by CodeRabbit
New Features
Bug Fixes
Performance
Tests