feat: client connection security hardening (P0 + P1)

[TinDang97]

Summary

Production security hardening for Moon's client connection layer. Implements all P0 (ship-blocker) and P1 (production hardening) security features identified in the connection security audit.

P0 — Ship Blockers

• maxclients (--maxclients 10000): Atomic CAS connection limit with -ERR max number of clients reached rejection. Both runtimes.
• Client idle timeout (--timeout 0): Configurable idle disconnect via tokio::time::timeout / monoio::select!. Subscriber mode exempted.
• TCP keepalive (--tcp-keepalive 300): SO_KEEPALIVE + TCP_KEEPIDLE on accepted sockets via socket2. Dead connection detection.
• AUTH rate limiting: Per-IP exponential backoff (100ms base → 10s cap). Auto-reset after 60s. Global parking_lot::Mutex<HashMap>.

P1 — Production Hardening

• CLIENT LIST / INFO / KILL: Global client registry (parking_lot::RwLock<HashMap>) with Drop-guard deregister. Redis-compatible output format. Kill by ID/ADDR/USER with atomic kill flag.
• CLIENT PAUSE / UNPAUSE: Global pause state with ALL/WRITE modes and auto-expiry. Handlers delay command processing while paused.
• CLIENT NO-EVICT / NO-TOUCH: Accepted (no-op stubs for Redis compatibility).
• ACL GENPASS [bits]: Cryptographically secure random password generation (1-4096 bits, hex output).

All features

• Compile clean on both runtime-monoio and runtime-tokio
• Zero clippy warnings on Linux (both runtimes)
• CONFIG GET/SET support for maxclients, timeout, tcp-keepalive
• No allocations on the command hot path
• 1937 tests passing (0 failures)

New modules

Module	LOC	Purpose
src/auth_ratelimit.rs	132	Per-IP AUTH failure rate limiting
src/client_registry.rs	276	Global client connection registry
src/client_pause.rs	106	CLIENT PAUSE/UNPAUSE state machine

New unit tests: 15

• auth_ratelimit: 4 tests (base delay, exponential backoff, cap, success reset)
• client_registry: 6 tests (register/list, info, kill by id, kill by user, update, parse args)
• client_pause: 2 tests (pause/check, write mode allows reads)
• acl::genpass: 3 tests (default bits, custom bits, invalid input)

Test plan

• cargo fmt --check — pass
• cargo clippy -- -D warnings (monoio) — zero warnings
• cargo clippy --no-default-features --features runtime-tokio,jemalloc -- -D warnings — zero warnings
• cargo test --lib — 1937 passed, 0 failed
• All checks run on Linux (OrbStack moon-dev VM)

Summary by CodeRabbit

• New Features

  • Per-IP AUTH rate limiting (exponential backoff); global CLIENT registry with LIST/INFO/KILL/SETNAME; CLIENT PAUSE/UNPAUSE (All/Write) with expiry; ACL GENPASS secure password generation; major GRAPH subsystem (query, Cypher parser/planner, traversal, hybrid graph+vector search, persistence, tooling).

• Configuration

  • New runtime options: maxclients (connection limit), timeout (idle disconnect), tcp-keepalive; keepalive enabled on accepts.

• CLI

  • CONFIG GET/SET for maxclients/timeout/tcp-keepalive; CLIENT NO-EVICT/NO-TOUCH accepted as OK.

• Tests

  • Several integration tests now skip when server unavailable; new unit/bench suites for graph.

[coderabbitai]

Actionable comments posted: 9

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (4)

src/server/conn/handler_sharded.rs (2)

527-547: ⚠️ Potential issue | 🔴 Critical

Pipelined AUTH failures only pay one delay.

auth_delay_ms is overwritten per failure and applied once after the whole batch. A client can pipeline many bad AUTHs and only incur the final sleep once, which materially weakens the brute-force mitigation.

Also applies to: 723-732, 1729-1732

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_sharded.rs` around lines 527 - 547, The code
overwrites auth_delay_ms on each AUTH failure so only the last failure's delay
is applied; change the assignment to accumulate delays instead (e.g.,
auth_delay_ms += crate::auth_ratelimit::record_failure(addr.ip())) inside the
AUTH handling block in the for frame in batch loop (the AUTH gate that checks
conn.authenticated and calls conn_cmd::auth_acl), and make the same change at
the other occurrences referenced (the similar AUTH blocks around the other
locations). Ensure auth_delay_ms remains a u64 and that record_failure return
values are added rather than assigned.

---

562-580: ⚠️ Potential issue | 🟠 Major

Rate-limit HELLO ... AUTH the same way as AUTH.

hello_acl() can authenticate, but these branches never call record_failure() / record_success() or set auth_delay_ms. A client can switch to HELLO 3 AUTH ... and bypass the new backoff entirely.

Also applies to: 738-746

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_sharded.rs` around lines 562 - 580, The HELLO branch
that calls conn_cmd::hello_acl currently updates protocol, client_name, and
current_user but does not call the same rate-limit bookkeeping used by AUTH;
modify the HELLO handling in handler_sharded.rs so that after calling hello_acl
you call the same record_failure(...) when response is Frame::Error and
record_success(...) when authentication succeeds, and ensure auth_delay_ms (and
any backoff state used by AUTH handling) is set/updated on failures just like in
the AUTH branch; make the identical change for the second HELLO handling near
the other occurrence (lines around the existing 738-746 block) so HELLO ... AUTH
cannot bypass the backoff.

src/server/conn/handler_monoio.rs (2)

523-540: ⚠️ Potential issue | 🔴 Critical

Pipelined AUTH failures only pay one delay.

auth_delay_ms is overwritten per failure and applied once after the whole batch. A client can pipeline many bad AUTHs and only incur the final sleep once, which materially weakens the brute-force mitigation.

Also applies to: 747-756, 2147-2150

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 523 - 540, The code currently
overwrites auth_delay_ms on each failed AUTH so a pipelined client only suffers
the last failure delay; change the logic to accumulate delays (e.g.,
auth_delay_ms += ...) when calling crate::auth_ratelimit::record_failure(...) so
each failed AUTH adds its penalty before the final sleep. Update the same
pattern at the other occurrences you noted (around the other AUTH gate blocks)
and keep the surrounding checks (extract_command, conn_cmd::auth_acl,
conn.authenticated, conn.current_user, peer_addr.parse()) unchanged.

---

556-574: ⚠️ Potential issue | 🟠 Major

Rate-limit HELLO ... AUTH the same way as AUTH.

hello_acl() can authenticate, but these branches never call record_failure() / record_success() or set auth_delay_ms. A client can switch to HELLO 3 AUTH ... and bypass the new backoff entirely.

Also applies to: 762-780

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 556 - 574, The HELLO branch
that calls conn_cmd::hello_acl must mirror the AUTH branch’s backoff
bookkeeping: after calling hello_acl, if the response indicates success (not
Frame::Error and opt_user set / conn.authenticated changed), call
record_success(client_id) and assign its returned delay to conn.auth_delay_ms
(or otherwise clear/reset delay as AUTH does); if authentication failed, call
record_failure(client_id) and set conn.auth_delay_ms from that result. In short,
update the HELLO handling around hello_acl to invoke the same
record_failure()/record_success() calls and assign conn.auth_delay_ms exactly
like the AUTH branch so HELLO 3 AUTH cannot bypass the backoff. Ensure you still
update conn.protocol_version, conn.client_name, conn.current_user and push the
response as before.

🧹 Nitpick comments (1)

src/server/listener.rs (1)

214-220: TCP keepalive configuration is sound; adding interval is optional for more robust dead-connection detection.

The socket2::TcpKeepalive::new().with_time() approach correctly sets TCP_KEEPIDLE. To match patterns like Redis, you could optionally add with_interval() to also set TCP_KEEPINTVL:

let ka = socket2::TcpKeepalive::new()
    .with_time(std::time::Duration::from_secs(config.tcp_keepalive))
    .with_interval(std::time::Duration::from_secs(config.tcp_keepalive / 3));

The current implementation is acceptable—OS defaults for probe interval are reasonable, and silently ignoring errors is appropriate for this best-effort feature.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/listener.rs` around lines 214 - 220, The TCP keepalive currently
sets only TCP_KEEPIDLE via socket2::TcpKeepalive::new().with_time(...); update
the keepalive construction in the block that checks config.tcp_keepalive to also
call .with_interval(...) so TCP_KEEPINTVL is set (e.g. compute interval as
config.tcp_keepalive / 3 but ensure it’s at least 1 second to avoid zero), then
continue to apply it with socket2::SockRef::from(&stream) and
sock.set_tcp_keepalive(&ka) as before; keeping the error from set_tcp_keepalive
ignored is fine since this is a best-effort setting.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/client_pause.rs`:
- Around line 66-74: expire_if_needed() currently does a read lock then
separately a write that can clear a newly-set pause (TOCTOU); fix by taking a
write lock for the whole operation (use PAUSE.write()), recompute the condition
inside that write lock (check state.active && Instant::now() >= state.until) and
only then set state.active = false; reference PAUSE and the expire_if_needed()
function and re-check the same predicate before mutating so a concurrent pause()
can't be lost.

In `@src/client_registry.rs`:
- Around line 151-176: The function parse_kill_args currently returns None when
args.len() < 2, making the legacy single-argument form unreachable; update
parse_kill_args (and use KillFilter) to handle the single-argument legacy case
first (or change the initial guard to only return when args.is_empty()), e.g. if
args.len() == 1 parse that as KillFilter::Addr from args[0], otherwise proceed
with the existing key/value loop and then the multi-arg parsing; ensure you keep
the same UTF-8 and parse error handling (ok()?) semantics when creating the Addr
value.

In `@src/server/conn/handler_monoio.rs`:
- Around line 971-993: The registry refresh is currently done only inside the
CLIENT LIST/INFO branches, causing stale client metadata elsewhere; update the
client's registry entry whenever the connection's visible state changes or on
every command dispatch. Concretely, add a shared call to
crate::client_registry::update(client_id, |e| { e.db = conn.selected_db;
e.last_cmd_at = std::time::Instant::now(); e.flags =
crate::client_registry::ClientFlags { subscriber: conn.subscription_count > 0,
in_multi: conn.in_multi, blocked: false }; if applicable set e.user =
conn.auth_user.clone(); }) from the common command-path (or right after handling
AUTH/HELLO/SELECT/SUBSCRIBE/MULTI and on registration) instead of only inside
the sub_bytes.eq_ignore_ascii_case(b"LIST"/"INFO") branches so client_list() and
client_info() always see fresh state.
- Around line 498-502: CLIENT PAUSE is treating all batches as writes because
check_pause(true) is hardcoded; change the call to pass the actual batch
write-ness instead (e.g., compute a boolean like is_write or use the existing
batch.is_write()/is_readonly() flag) so you call
crate::client_pause::check_pause(is_write) instead of check_pause(true), keeping
the expire_if_needed() call in place; update the surrounding logic in
handler_monoio.rs where the batch is available to derive that boolean and await
sleep only when check_pause returns Some for true-write batches.

In `@src/server/conn/handler_sharded.rs`:
- Around line 984-1006: Registry entries are only refreshed when a client
introspects itself (CLIENT LIST/INFO); instead, call
crate::client_registry::update(client_id, |e| { ... }) to refresh e.db,
e.last_cmd_at, e.flags (subscriber/in_multi/blocked) and e.user whenever the
connection's state changes—specifically after AUTH/HELLO (user change), SELECT
(selected_db change), SUBSCRIBE/UNSUBSCRIBE and any place subscription_count is
modified, MULTI/EXEC/DISCARD (in_multi changes), and immediately when any
command that should bump activity is handled; also ensure last_cmd_at is updated
on normal command execution paths so client_registry::client_list and
client_registry::client_info reflect current state rather than only on
self-inspection.
- Around line 513-518: The code unconditionally calls
client_pause::check_pause(true) which treats every batch as write traffic;
change this to compute an is_write boolean from the current request/batch
(inspect the batch/command type or use an existing method like Batch::is_write
or any per-request write flag) and pass that value to
client_pause::check_pause(is_write) after
crate::client_pause::expire_if_needed(); ensure read-only batches pass false so
they do not sleep while preserving the current conservative behavior for ALL
mode when a batch actually contains writes.

In `@src/shard/conn_accept.rs`:
- Around line 213-218: The TLS and monoio early-exit paths in conn_accept.rs
(where try_accept_connection(maxclients) is checked) currently just return on
rejection (seen near the block using shard_id), causing EOF instead of the RESP
error; before returning, write the advertised RESP error string "-ERR max number
of clients reached\r\n" to the connection/socket/stream used by the TLS and
monoio branches so clients receive the documented error, and then return/close;
apply the same change to the other rejection sites in this file that check
try_accept_connection (the other similar branches noted in the review).
- Around line 578-585: The code sets _migrated = true before performing the fd
dup and the SPSC push in handling MonoioHandlerResult::MigrateConnection, which
can cause record_connection_closed() to be skipped on failure and leak
CONNECTED_CLIENTS/maxclients; fix by deferring setting _migrated (and thus
skipping close accounting) until after dup() succeeds and the ChannelMesh
producer.push(...) returns Ok, ensure that if dup or the push fails you call
record_connection_closed() for the original stream (or do not drop it
prematurely) so CONNECTED_CLIENTS is decremented, and apply the same change to
the other migration block (lines ~591-629) that uses ChannelMesh and the SPSC
push.

---

Outside diff comments:
In `@src/server/conn/handler_monoio.rs`:
- Around line 523-540: The code currently overwrites auth_delay_ms on each
failed AUTH so a pipelined client only suffers the last failure delay; change
the logic to accumulate delays (e.g., auth_delay_ms += ...) when calling
crate::auth_ratelimit::record_failure(...) so each failed AUTH adds its penalty
before the final sleep. Update the same pattern at the other occurrences you
noted (around the other AUTH gate blocks) and keep the surrounding checks
(extract_command, conn_cmd::auth_acl, conn.authenticated, conn.current_user,
peer_addr.parse()) unchanged.
- Around line 556-574: The HELLO branch that calls conn_cmd::hello_acl must
mirror the AUTH branch’s backoff bookkeeping: after calling hello_acl, if the
response indicates success (not Frame::Error and opt_user set /
conn.authenticated changed), call record_success(client_id) and assign its
returned delay to conn.auth_delay_ms (or otherwise clear/reset delay as AUTH
does); if authentication failed, call record_failure(client_id) and set
conn.auth_delay_ms from that result. In short, update the HELLO handling around
hello_acl to invoke the same record_failure()/record_success() calls and assign
conn.auth_delay_ms exactly like the AUTH branch so HELLO 3 AUTH cannot bypass
the backoff. Ensure you still update conn.protocol_version, conn.client_name,
conn.current_user and push the response as before.

In `@src/server/conn/handler_sharded.rs`:
- Around line 527-547: The code overwrites auth_delay_ms on each AUTH failure so
only the last failure's delay is applied; change the assignment to accumulate
delays instead (e.g., auth_delay_ms +=
crate::auth_ratelimit::record_failure(addr.ip())) inside the AUTH handling block
in the for frame in batch loop (the AUTH gate that checks conn.authenticated and
calls conn_cmd::auth_acl), and make the same change at the other occurrences
referenced (the similar AUTH blocks around the other locations). Ensure
auth_delay_ms remains a u64 and that record_failure return values are added
rather than assigned.
- Around line 562-580: The HELLO branch that calls conn_cmd::hello_acl currently
updates protocol, client_name, and current_user but does not call the same
rate-limit bookkeeping used by AUTH; modify the HELLO handling in
handler_sharded.rs so that after calling hello_acl you call the same
record_failure(...) when response is Frame::Error and record_success(...) when
authentication succeeds, and ensure auth_delay_ms (and any backoff state used by
AUTH handling) is set/updated on failures just like in the AUTH branch; make the
identical change for the second HELLO handling near the other occurrence (lines
around the existing 738-746 block) so HELLO ... AUTH cannot bypass the backoff.

---

Nitpick comments:
In `@src/server/listener.rs`:
- Around line 214-220: The TCP keepalive currently sets only TCP_KEEPIDLE via
socket2::TcpKeepalive::new().with_time(...); update the keepalive construction
in the block that checks config.tcp_keepalive to also call .with_interval(...)
so TCP_KEEPINTVL is set (e.g. compute interval as config.tcp_keepalive / 3 but
ensure it’s at least 1 second to avoid zero), then continue to apply it with
socket2::SockRef::from(&stream) and sock.set_tcp_keepalive(&ka) as before;
keeping the error from set_tcp_keepalive ignored is fine since this is a
best-effort setting.

🪄 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: cbe24da0-21f9-40e5-9130-ac600d4770f7

📥 Commits

Reviewing files that changed from the base of the PR and between 2029eae and 320edc4.

📒 Files selected for processing (13)

• src/admin/metrics_setup.rs
• src/auth_ratelimit.rs
• src/client_pause.rs
• src/client_registry.rs
• src/command/acl.rs
• src/command/config.rs
• src/config.rs
• src/lib.rs
• src/server/conn/handler_monoio.rs
• src/server/conn/handler_sharded.rs
• src/server/listener.rs
• src/shard/conn_accept.rs
• src/storage/eviction.rs

[coderabbitai]

⚠️ Potential issue | 🟠 Major

CLIENT LIST / CLIENT INFO are still emitting incorrect state.

format_client_line() hard-codes sub=0 ... multi=-1 and ignores the tracked flags except for the single-character flags= field. Even with correct registry updates, subscribed or MULTI clients will still be reported incorrectly.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

CLIENT PAUSE WRITE is acting like CLIENT PAUSE ALL.

check_pause(true) marks every batch as write traffic, so read-only batches sleep too. That breaks the WRITE-mode contract.

Proposed fix

-        crate::client_pause::expire_if_needed();
-        if let Some(remaining) = crate::client_pause::check_pause(true) {
+        let batch_has_write = frames.iter().any(|frame| {
+            extract_command(frame)
+                .is_some_and(|(cmd, _)| metadata::is_write(cmd))
+        });
+        crate::client_pause::expire_if_needed();
+        if let Some(remaining) = crate::client_pause::check_pause(batch_has_write) {
             monoio::time::sleep(remaining).await;
         }

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	// CLIENT PAUSE: delay processing if server is paused
	crate::client_pause::expire_if_needed();
	if let Some(remaining) = crate::client_pause::check_pause(true) {
	monoio::time::sleep(remaining).await;
	}
	// CLIENT PAUSE: delay processing if server is paused
	let batch_has_write = frames.iter().any(|frame| {
	extract_command(frame)
	.is_some_and(|(cmd, _)| metadata::is_write(cmd))
	});
	crate::client_pause::expire_if_needed();
	if let Some(remaining) = crate::client_pause::check_pause(batch_has_write) {
	monoio::time::sleep(remaining).await;
	}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 498 - 502, CLIENT PAUSE is
treating all batches as writes because check_pause(true) is hardcoded; change
the call to pass the actual batch write-ness instead (e.g., compute a boolean
like is_write or use the existing batch.is_write()/is_readonly() flag) so you
call crate::client_pause::check_pause(is_write) instead of check_pause(true),
keeping the expire_if_needed() call in place; update the surrounding logic in
handler_monoio.rs where the batch is available to derive that boolean and await
sleep only when check_pause returns Some for true-write batches.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

CLIENT PAUSE WRITE is acting like CLIENT PAUSE ALL.

check_pause(true) marks every batch as write traffic, so read-only batches sleep too. That breaks the WRITE-mode contract.

Proposed fix

-                crate::client_pause::expire_if_needed();
-                if let Some(remaining) = crate::client_pause::check_pause(true) {
+                let batch_has_write = batch.iter().any(|frame| {
+                    extract_command(frame)
+                        .is_some_and(|(cmd, _)| metadata::is_write(cmd))
+                });
+                crate::client_pause::expire_if_needed();
+                if let Some(remaining) = crate::client_pause::check_pause(batch_has_write) {
                     tokio::time::sleep(remaining).await;
                 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_sharded.rs` around lines 513 - 518, The code
unconditionally calls client_pause::check_pause(true) which treats every batch
as write traffic; change this to compute an is_write boolean from the current
request/batch (inspect the batch/command type or use an existing method like
Batch::is_write or any per-request write flag) and pass that value to
client_pause::check_pause(is_write) after
crate::client_pause::expire_if_needed(); ensure read-only batches pass false so
they do not sleep while preserving the current conservative behavior for ALL
mode when a batch actually contains writes.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Return the advertised maxclients error on these rejection paths.

The Tokio plain path writes -ERR max number of clients reached, but these TLS/monoio branches just return. Clients on those paths will see EOF instead of the documented RESP error.

Also applies to: 515-521, 555-561

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/shard/conn_accept.rs` around lines 213 - 218, The TLS and monoio
early-exit paths in conn_accept.rs (where try_accept_connection(maxclients) is
checked) currently just return on rejection (seen near the block using
shard_id), causing EOF instead of the RESP error; before returning, write the
advertised RESP error string "-ERR max number of clients reached\r\n" to the
connection/socket/stream used by the TLS and monoio branches so clients receive
the documented error, and then return/close; apply the same change to the other
rejection sites in this file that check try_accept_connection (the other similar
branches noted in the review).

[coderabbitai]

Actionable comments posted: 3

🧹 Nitpick comments (1)

.github/workflows/ci.yml (1)

111-111: Track these redis-compat skips with explicit expiry/issue links.

Line 111 adds permanent skips for xadd, xrange, and xtrim. Please attach issue IDs (or a TODO with removal criteria) so these don’t silently become long-term blind spots.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In @.github/workflows/ci.yml at line 111, The CI step that runs cargo test with
permanent skips for tests "xadd", "xrange", and "xtrim" (the MOON_TEST_PORT=6399
cargo test line) must be annotated so the skips are tracked: update that
workflow line by adding inline comments or adjacent TODOs linking to an issue ID
(e.g., GH-XXXX) or a removal criteria (e.g., "remove when redis compatibility
implemented / test unskips by YYYY-MM-DD") for each skipped test name so these
skips don’t become permanent blind spots; ensure the comments reference the
exact test names "xadd", "xrange", and "xtrim" so reviewers can find and resolve
them later.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In @.github/workflows/ci.yml:
- Line 44: The current CI step uses the command string "cargo test --lib
--no-default-features --features runtime-tokio,jemalloc" which excludes
integration tests; restore CI coverage by removing the "--lib" flag (or add an
additional job/step that runs "cargo test" without "--lib") so integration tests
run under the same feature set (e.g., "cargo test --no-default-features
--features runtime-tokio,jemalloc"); update the workflow step that contains the
exact run line to either run the full test suite or add a follow-up step to
execute integration tests.

In `@tests/blocking_list_timeout.rs`:
- Around line 35-36: Remove the blanket #[ignore] on the blocking-list
integration tests (e.g., the async test blpop_timeout_returns_nil and the other
tests in this file currently marked ignored) and replace it with a conditional
runtime skip so CI can run them by default while still allowing local opt-in;
implement a small helper (e.g., should_run_integration_tests()) that checks an
environment variable or feature (like RUN_INTEGRATION_TESTS) and at the start of
each test (blpop_timeout_returns_nil and the other ignored tests) return early
with a clear skip message when the helper is false, or alternatively gate the
tests behind a cfg(feature = "integration_tests") so automated runs include them
unless explicitly disabled.

In `@tests/functions_fcall.rs`:
- Around line 42-43: The test file functions_fcall.rs was marked entirely
ignored (e.g., the #[ignore] on function_load_and_fcall and other test
functions), which disables CI coverage; remove the blanket #[ignore] attributes
from those test functions (e.g., function_load_and_fcall and the other tests in
this file) so they run in CI, or alternatively replace the attribute with a
runtime skip: in each async test (e.g., function_load_and_fcall) check an
environment variable or feature flag (std::env::var("MOON_SERVER_ADDR") or a cfg
feature) at the start and return early when the external moon server is not
available, leaving the test unignored for CI while still avoiding failures when
the dependency is absent.

---

Nitpick comments:
In @.github/workflows/ci.yml:
- Line 111: The CI step that runs cargo test with permanent skips for tests
"xadd", "xrange", and "xtrim" (the MOON_TEST_PORT=6399 cargo test line) must be
annotated so the skips are tracked: update that workflow line by adding inline
comments or adjacent TODOs linking to an issue ID (e.g., GH-XXXX) or a removal
criteria (e.g., "remove when redis compatibility implemented / test unskips by
YYYY-MM-DD") for each skipped test name so these skips don’t become permanent
blind spots; ensure the comments reference the exact test names "xadd",
"xrange", and "xtrim" so reviewers can find and resolve them later.

🪄 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: 936cb625-298f-4064-b950-0fe39f772e27

📥 Commits

Reviewing files that changed from the base of the PR and between e5e4e97 and c2c32f5.

📒 Files selected for processing (7)

• .github/workflows/ci.yml
• Cargo.toml
• deny.toml
• tests/blocking_list_timeout.rs
• tests/functions_fcall.rs
• tests/integration.rs
• tests/replication_test.rs

✅ Files skipped from review due to trivial changes (1)

• tests/integration.rs

[coderabbitai]

Actionable comments posted: 13

Note

Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.

♻️ Duplicate comments (5)

tests/functions_fcall.rs (1)

10-10: ⚠️ Potential issue | 🟠 Major

Avoid blanket #[ignore] on the FUNCTION integration suite.

Line 42 through Line 200 now mark every runnable test ignored, and Line 10 documents running only --ignored; this effectively removes default CI coverage for these behaviors.

Suggested pattern (keep tests enabled, skip at runtime when env/server is absent)

+fn should_run_function_it() -> bool {
+    std::env::var("MOON_RUN_FUNCTION_IT").as_deref() == Ok("1")
+}

 #[tokio::test]
-#[ignore] // Requires running moon server on port 16479
 async fn function_load_and_fcall() {
+    if !should_run_function_it() {
+        return;
+    }
     let mut con = get_conn().await;
     flush_functions(&mut con).await;
     // ...
 }

Apply the same guard to the other tests currently marked with #[ignore].

Also applies to: 42-43, 61-62, 77-78, 101-102, 129-130, 150-151, 178-179, 199-200

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/functions_fcall.rs` at line 10, Remove the blanket #[ignore] attributes
from the FUNCTION integration tests in tests/functions_fcall.rs and instead add
a runtime guard at the start of each test (the same pattern used in the
suggested snippet) that checks for the required env/server (e.g., std::env::var
or a shared helper like ensure_functions_server_available()) and calls
early-return or test::ignore at runtime when the server/env is absent; update
each affected test function (the ones previously annotated with #[ignore]) to
perform this guard (or call the shared helper) so tests run by default but skip
safely when prerequisites are missing.

.github/workflows/ci.yml (1)

44-44: ⚠️ Potential issue | 🟠 Major

Restore integration-test coverage in the main test lane.

Line 44 uses cargo test --lib, which drops integration tests from this job and weakens regression detection. Please either remove --lib or add a dedicated integration-test step with the same feature set.

Suggested fix

-      - run: cargo test --lib --no-default-features --features runtime-tokio,jemalloc
+      - run: cargo test --no-default-features --features runtime-tokio,jemalloc

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In @.github/workflows/ci.yml at line 44, The CI job currently runs `cargo test
--lib --no-default-features --features runtime-tokio,jemalloc`, which excludes
integration tests; either remove the `--lib` flag from that command or add a
separate step that runs integration tests with the same feature set (e.g., a new
step that runs `cargo test --tests --no-default-features --features
runtime-tokio,jemalloc` or simply `cargo test --no-default-features --features
runtime-tokio,jemalloc`) so integration tests are executed with the same
features; update the workflow step that contains the `run: cargo test --lib
--no-default-features --features runtime-tokio,jemalloc` command accordingly.

src/client_registry.rs (2)

40-52: ⚠️ Potential issue | 🟠 Major

CLIENT LIST/INFO still flatten multi-state client flags.

to_flag_str() picks only one flag character, and format_client_line() still hardcodes sub=0 and multi=-1. Subscribed or MULTI clients will keep reporting incorrect state even when the registry entry itself is fresh.

Also applies to: 180-193

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/client_registry.rs` around lines 40 - 52, The current
ClientFlags::to_flag_str only returns a single flag char and format_client_line
hardcodes sub=0 and multi=-1, causing flattened/misreported client state; update
ClientFlags::to_flag_str to return a combined flag string containing all
applicable characters (e.g., include "S" if subscriber, "x" if in_multi, "b" if
blocked, fallback "N" only when none set) and modify format_client_line to
compute and emit the correct sub and multi values from the ClientFlags (set
sub=1 when subscriber, set multi to 1/0 or appropriate value when in_multi is
true/false instead of hardcoding) so the registry reports accurate multi-state
flags and fields.

---

151-178: ⚠️ Potential issue | 🟡 Minor

Preserve the legacy single-argument CLIENT KILL addr:port form.

args.len() < 2 returns before the single-argument fallback, so that compatibility path is still unreachable.

Proposed fix

 pub fn parse_kill_args(args: &[&[u8]]) -> Option<KillFilter> {
-    if args.len() < 2 {
-        return None;
-    }
+    if args.len() == 1 {
+        let addr = std::str::from_utf8(args[0]).ok()?;
+        return Some(KillFilter::Addr(addr.to_string()));
+    }
+    if args.len() < 2 {
+        return None;
+    }
@@
-    // Legacy single-arg form: CLIENT KILL addr:port
-    if args.len() == 1 {
-        let addr = std::str::from_utf8(args[0]).ok()?;
-        return Some(KillFilter::Addr(addr.to_string()));
-    }
     None
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/client_registry.rs` around lines 151 - 178, The function parse_kill_args
incorrectly returns None for single-argument calls because the initial guard
checks if args.len() < 2; change the guard and control flow so the legacy
single-arg form is reachable: replace the initial early-return with a check for
args.len() == 0 (return None), keep the existing paired-argument while loop
handling when args.len() >= 2 (using the same key/value iteration over args and
checking key.eq_ignore_ascii_case for "ID"/"ADDR"/"USER"), and ensure that after
that loop you still handle the legacy single-argument case (when args.len() ==
1) returning KillFilter::Addr; update only parse_kill_args and reference the
KillFilter variants (Id, Addr, User).

src/server/conn/handler_monoio.rs (1)

500-504: ⚠️ Potential issue | 🟠 Major

Pause enforcement runs too late and still treats every batch as a write.

This was already flagged on the earlier revision, and the hardcoded check_pause(true) is still here. That still makes CLIENT PAUSE WRITE behave like ALL, and because the gate sits after the inline-dispatch fast path, inlined commands can execute during a pause. It also means a CLIENT UNPAUSE batch will sleep instead of taking effect immediately.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 500 - 504, The pause logic is
executed too late and always uses check_pause(true) (treating every batch as
WRITE), so move the pause enforcement earlier in the request handling path
(before the inline-dispatch/fast-path) and call
crate::client_pause::check_pause(false) when evaluating non-write commands so
that CLIENT PAUSE WRITE only affects write commands; also ensure
crate::client_pause::expire_if_needed() and the sleep only run after confirming
the batch actually needs pausing (so CLIENT UNPAUSE/UNPAUSE batches are not
slept) and make the pause check discriminate per-command/batch type rather than
hardcoding true in handler_monoio.rs.

🟡 Minor comments (9)

src/storage/eviction.rs-535-537 (1)

535-537: ⚠️ Potential issue | 🟡 Minor

Record eviction metrics only when deletion actually succeeds.

record_eviction() is currently called even if db.remove(...) removes nothing (racey disappearance), which can inflate eviction metrics.

💡 Minimal fix

-    db.remove(key.as_bytes());
-    crate::admin::metrics_setup::record_eviction();
+    if db.remove(key.as_bytes()).is_some() {
+        crate::admin::metrics_setup::record_eviction();
+    }
     true

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/storage/eviction.rs` around lines 535 - 537, The eviction metric is
recorded unconditionally after calling db.remove(key.as_bytes()), which can
overcount when nothing was actually deleted; change the code to inspect
db.remove's result (e.g., match or if let on the Result<Option<_>, _> returned
by db.remove in eviction.rs), call
crate::admin::metrics_setup::record_eviction() only when the removal succeeded
and returned Some(_) (i.e., an entry was removed), and return true in that
branch; in the error or None branch return false (or propagate the error per
existing function behavior).

scripts/bench-graph.sh-118-119 (1)

118-119: ⚠️ Potential issue | 🟡 Minor

Report throughput using successful operations.

Using configured totals overstates performance when insertions fail. Update lines 118-119 and 153-154 to report actual operation counts:

Metric correction

-OPS_PER_SEC=$(( NODES * 1000 / (ELAPSED_MS + 1) ))
-log "  $NODES nodes inserted in ${ELAPSED_MS}ms (${OPS_PER_SEC} ops/s)"
+OPS_PER_SEC=$(( ACTUAL_NODES * 1000 / (ELAPSED_MS + 1) ))
+log "  $ACTUAL_NODES nodes inserted in ${ELAPSED_MS}ms (${OPS_PER_SEC} ops/s)"

-OPS_PER_SEC=$(( EDGES * 1000 / (ELAPSED_MS + 1) ))
-log "  $EDGES edges inserted in ${ELAPSED_MS}ms (${OPS_PER_SEC} ops/s)"
+OPS_PER_SEC=$(( EDGE_OK * 1000 / (ELAPSED_MS + 1) ))
+log "  $EDGE_OK edges inserted in ${ELAPSED_MS}ms (${OPS_PER_SEC} ops/s)"

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@scripts/bench-graph.sh` around lines 118 - 119, Replace the current
throughput calculation and log that use the configured NODES with the actual
successful-operations count so failed inserts don't inflate throughput: compute
OPS_PER_SEC using the real successful count variable (e.g., INSERTED or
SUCCESS_COUNT) instead of NODES (OPS_PER_SEC=$(( SUCCESS_COUNT * 1000 /
(ELAPSED_MS + 1) ))), and update the log line to print SUCCESS_COUNT and
ELAPSED_MS rather than NODES; apply the same change to the second occurrence
later in the script (the analogous calculation/log around lines where
OPS_PER_SEC and the node log are repeated).

scripts/bench-graph.sh-60-60 (1)

60-60: ⚠️ Potential issue | 🟡 Minor

Remove --protected-mode no and quote shell variables.

The server already binds to 127.0.0.1 by default. Keeping protected mode enabled (the default) provides an extra layer of safety by rejecting non-loopback connections when no password is set. Additionally, quote variables to protect against word splitting.

-MOON_NO_URING=1 $BINARY --port $PORT --shards $SHARDS --protected-mode no > /tmp/moon_graph_bench.log 2>&1 &
+MOON_NO_URING=1 "$BINARY" --port "$PORT" --shards "$SHARDS" > /tmp/moon_graph_bench.log 2>&1 &

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@scripts/bench-graph.sh` at line 60, Remove the explicit "--protected-mode no"
flag from the server invocation and quote the shell variables to prevent
word-splitting: update the line that uses MOON_NO_URING and $BINARY (the command
that currently includes "--port $PORT --shards $SHARDS --protected-mode no >
/tmp/moon_graph_bench.log 2>&1 &") to omit "--protected-mode no" and wrap
variables like "$BINARY", "$PORT", "$SHARDS" and the logfile path in quotes so
the command runs with default protected mode and is safe against spaces/special
characters.

src/graph/traversal_guard.rs-72-80 (1)

72-80: ⚠️ Potential issue | 🟡 Minor

Make zero-duration budgets expire immediately.

Line 74 uses >, so a Duration::ZERO guard can still return Ok(()) on the first hop. That doesn't match the documented immediate-timeout behavior.

🩹 Proposed fix

-        if elapsed > self.timeout {
+        if elapsed >= self.timeout {

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/traversal_guard.rs` around lines 72 - 80, In check_timeout
(TraversalGuard::check_timeout) the comparison uses `>` so a zero-duration
timeout can still pass on the first check; change the comparison to use `>=` (or
otherwise treat elapsed >= timeout as expired) so a Duration::ZERO budget will
return Err(TraversalTimeout { elapsed, timeout: self.timeout, snapshot_lsn:
self.snapshot_lsn }) immediately; update only the comparison logic in
check_timeout to ensure immediate expiry semantics are enforced.

src/graph/segment.rs-53-61 (1)

53-61: ⚠️ Potential issue | 🟡 Minor

Potential race in non-atomic read-modify-write.

add_immutable performs a load, modifies the list, then stores. If two concurrent calls happen, one update could be lost. Consider using ArcSwap::rcu for atomic read-copy-update:

🔒 Suggested fix using RCU

     pub fn add_immutable(&self, csr: CsrSegment) {
-        let current = self.segments.load();
-        let mut new_immutable = current.immutable.clone();
-        new_immutable.insert(0, Arc::new(csr)); // newest first
-        self.segments.store(Arc::new(GraphSegmentList {
-            mutable: current.mutable.clone(),
-            immutable: new_immutable,
-        }));
+        self.segments.rcu(|current| {
+            let mut new_immutable = current.immutable.clone();
+            new_immutable.insert(0, Arc::new(csr.clone())); // requires Clone
+            Arc::new(GraphSegmentList {
+                mutable: current.mutable.clone(),
+                immutable: new_immutable,
+            })
+        });
     }

Note: This requires CsrSegment: Clone or restructuring to pass Arc<CsrSegment> directly.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/segment.rs` around lines 53 - 61, add_immutable currently does a
non-atomic load->modify->store on self.segments (load/store of GraphSegmentList)
which can lost updates under concurrency; change it to use an atomic RCU update
(ArcSwap::rcu or equivalent) so the insertion of the new Arc<CsrSegment> is
applied atomically: build the new GraphSegmentList from the provided
Arc<CsrSegment> (or make CsrSegment Clone/accept Arc<CsrSegment>), then call
segments.rcu(|current| { let mut new = current.clone(); new.immutable.insert(0,
new_arc); Arc::new(new) }) (or the library's equivalent) to guarantee no races
instead of plain load/clone/store in add_immutable.

src/graph/replay.rs-404-414 (1)

404-414: ⚠️ Potential issue | 🟡 Minor

Silent data loss risk when Arc::try_unwrap fails.

If Arc::try_unwrap fails (extra owners exist), the code logs a warning and creates a fresh MemGraph, discarding any data in the existing one. During WAL replay at startup this should not happen, but if it does, previously replayed nodes/edges would be lost.

Consider making this a hard error or at least returning an Option to signal the failure to the caller, rather than silently proceeding with an empty graph.

src/graph/cross_shard.rs-103-120 (1)

103-120: ⚠️ Potential issue | 🟡 Minor

Potential integer overflow when casting external ID to i64.

neighbor_ext_id is u64, but it's cast to i64 at line 120 for Frame::Integer. External IDs greater than i64::MAX would wrap to negative values, potentially causing incorrect behavior on the coordinator side.

Consider using Frame::BulkString with the decimal representation, or validate that external IDs stay within i64 range:

// Option 1: Use saturating conversion and document the limit
discovered_ids.push(Frame::Integer(neighbor_ext_id.min(i64::MAX as u64) as i64));

// Option 2: Use BulkString for large IDs
discovered_ids.push(Frame::BulkString(Bytes::from(neighbor_ext_id.to_string())));

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/cross_shard.rs` around lines 103 - 120, neighbor_ext_id is a u64
that’s being cast to i64 when calling discovered_ids.push(Frame::Integer(...)),
which can overflow for IDs > i64::MAX; update the handling in this block (the
code around neighbor_ext_id, result_frames, and discovered_ids.push) to avoid
truncation by either validating/limiting the value before converting (e.g.,
clamp/saturating to i64::MAX) or emit the ID as a string/binary
(Frame::BulkString with neighbor_ext_id.to_string()) so large external IDs are
preserved; ensure the change is applied where
discovered_ids.push(Frame::Integer(...)) is currently invoked and keep existing
edge_to_traverse_frame and node_to_traverse_frame calls unchanged.

src/graph/cypher/parser.rs-321-345 (1)

321-345: ⚠️ Potential issue | 🟡 Minor

Reject reversed variable-length ranges.

*3..1 currently parses as (3, 1) and is treated as valid. That should fail during parsing instead of passing an impossible hop range downstream.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/cypher/parser.rs` around lines 321 - 345, In parse_var_length,
detect and reject reversed ranges by validating that the computed min is not
greater than max before returning; after computing min and max in fn
parse_var_length (the logic around parse_u32_lit, Token::DotDot and the min/max
assignments), add a check if min > max and return an Err(CypherError::new(...))
(or the project’s standard parse error) with a clear message like "invalid
variable-length range: min > max" so inputs like `*3..1` fail during parsing
rather than being accepted.

src/command/graph/graph_read.rs-617-623 (1)

617-623: ⚠️ Potential issue | 🟡 Minor

Docs advertise EXPAND, but the dispatcher rejects it.

The command docs list EXPAND as a supported GRAPH.HYBRID mode, but the implementation only accepts FILTER and WALK and returns ERR unknown ... otherwise.

Also applies to: 713-717

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/command/graph/graph_read.rs` around lines 617 - 623, The docs list an
EXPAND mode for GRAPH.HYBRID but graph_hybrid rejects it; update graph_hybrid to
recognize the "EXPAND" mode (in the same dispatch that handles "FILTER" and
"WALK") and either call the existing expansion implementation (e.g.,
graph_hybrid_expand or the function used by other expansion code) or implement
the expansion logic inline: parse args as "<k> <expansion_hops> <vector>",
validate them the same way FILTER/WALK do, and return the appropriate Frame
result instead of falling through to the "ERR unknown" branch; if no expansion
handler exists, add a new function (graph_hybrid_expand) and invoke it from
graph_hybrid.

🧹 Nitpick comments (28)

src/persistence/wal_v3/segment.rs (1)

218-220: Emit WAL rotation metric only after successful segment open.

record_wal_rotation() is called before open_new_segment(). If opening the segment fails, rotations are overcounted.

Proposed fix

         self.current_sequence += 1;
-        crate::admin::metrics_setup::record_wal_rotation();
-        self.open_new_segment()
+        self.open_new_segment()?;
+        crate::admin::metrics_setup::record_wal_rotation();
+        Ok(())

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/persistence/wal_v3/segment.rs` around lines 218 - 220, Move the WAL
rotation metric emission so it only runs after a new segment is successfully
opened: call record_wal_rotation() after open_new_segment() returns Ok, and only
on the successful path (handle Err without recording). Update the sequence of
operations in the method that increments self.current_sequence and opens a new
segment (refer to self.current_sequence, open_new_segment(), and
crate::admin::metrics_setup::record_wal_rotation()) so that the metric call
follows a successful open_new_segment() result; do not call
record_wal_rotation() on the error path.

scripts/bench-graph.sh (1)

46-46: Use an argv array for redis-cli instead of a command string.

The current pattern depends on word-splitting and is brittle with argument values. Use an array and invoke with "${CLI[@]}" everywhere.

Proposed refactor

-CLI="redis-cli -p $PORT"
+CLI=(redis-cli -p "$PORT")
...
-    if $CLI PING > /dev/null 2>&1; then
+    if "${CLI[@]}" PING > /dev/null 2>&1; then
...
-if ! $CLI PING > /dev/null 2>&1; then
+if ! "${CLI[@]}" PING > /dev/null 2>&1; then

Also applies to: 65-65, 71-71, 157-157, 226-226, 231-231, 238-238, 249-250

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@scripts/bench-graph.sh` at line 46, Replace the CLI command string with an
argv array and use array expansion everywhere: change the variable declaration
CLI="redis-cli -p $PORT" to CLI=(redis-cli -p "$PORT") and then update all
invocations to use "${CLI[@]}" instead of unquoted expansions or plain strings
(this applies to the usages at lines referenced in the review such as the calls
around the earlier CLI declaration and at locations noted: 65, 71, 157, 226,
231, 238, 249-250); ensure any additional arguments appended to CLI are added as
separate array elements or passed after "${CLI[@]}" so word-splitting and
quoting are preserved.

.github/workflows/compat.yml (1)

365-367: Replace fixed sleep 2 with an active readiness probe.

Using a fixed delay is flaky under variable runner load; poll for port/readiness with timeout before launching client tests.

Also applies to: 423-425

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In @.github/workflows/compat.yml around lines 365 - 367, The workflow uses a
fixed "sleep 2" after starting the moon server ("./target/release/moon --port
6399 --shards 1 &"), which is flaky; replace it with an active readiness probe
that polls the server port or health endpoint with a timeout (e.g., loop until
port 6399 is accepting connections or the server's /health responds, with a
per-try sleep and overall timeout) and fail the job if the probe times out;
apply the same change to the other occurrence noted (the block at lines 423-425)
so tests only start after a successful readiness check.

src/auth_ratelimit.rs (1)

45-48: Cleanup latency is bounded but worth noting.

The retain() call iterates all entries when the map reaches 10k IPs. In the worst case (10k entries, all stale), this is a brief O(n) scan. Since AUTH failures are rare in normal operation and the cleanup only triggers at the 10k threshold, this is acceptable. If AUTH-heavy workloads become common, consider amortizing cleanup across multiple calls.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/auth_ratelimit.rs` around lines 45 - 48, The current full-map cleanup
triggered when map.len() >= MAX_ENTRIES (using map.retain with last_failure and
STALE_THRESHOLD_SECS) can cause an O(n) scan; change this to an amortized
cleanup: when MAX_ENTRIES is reached, remove only a bounded batch of stale
entries per call (e.g., iterate a fixed number of keys/entries and remove those
whose r.last_failure <= cutoff) or probabilistically sample a subset to check,
using the same cutoff logic and the existing map, MAX_ENTRIES, last_failure and
STALE_THRESHOLD_SECS symbols; ensure the batch size is a constant so each auth
path remains O(1) amortized and fallback to full retain only if necessary.

src/command/graph/graph_write.rs (2)

332-342: Consider documenting collision behavior.

label_to_id truncates FNV-1a to 16 bits, meaning collisions are possible with >65536 distinct labels/types/properties. This is acceptable for space efficiency, but consider documenting that label lookups by name may not be unique (e.g., in GRAPH.INFO output or debugging scenarios).

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/command/graph/graph_write.rs` around lines 332 - 342, Document in the
comment for the function label_to_id that it uses FNV-1a truncated to 16 bits
and therefore collisions are possible when there are more than 65,536 distinct
labels; mention that label_to_id(name: &[u8]) returns a non-unique u16 index in
that case and callers (e.g., GRAPH.INFO or debugging routines) must handle
potential collisions or perform full-name comparisons when uniqueness is
required.

---

127-140: Scoped borrow pattern is correct but clones are expensive.

The scoped borrow releasing the graph before WAL push is necessary. However, labels.clone(), properties.clone(), and embedding.clone() on line 134 duplicate data that could potentially be moved. Consider restructuring to avoid clones:

♻️ Potential optimization to avoid clones

-    let external_id = {
-        let graph = match store.get_graph_mut(graph_name) {
-            Some(g) => g,
-            None => return Frame::Error(Bytes::from_static(b"ERR graph not found")),
-        };
-
-        let node_key = graph.write_buf.add_node(labels.clone(), properties.clone(), embedding.clone(), lsn);
-
-        // Update graph stats incrementally.
-        graph.stats.on_node_insert(&labels);
-
-        node_key.data().as_ffi()
-    };
+    let graph = match store.get_graph_mut(graph_name) {
+        Some(g) => g,
+        None => return Frame::Error(Bytes::from_static(b"ERR graph not found")),
+    };
+
+    let node_key = graph.write_buf.add_node(labels.clone(), properties.clone(), embedding.clone(), lsn);
+    graph.stats.on_node_insert(&labels);
+    let external_id = node_key.data().as_ffi();
+
+    // Drop mutable borrow before WAL access
+    drop(graph);

This still requires clones since add_node takes ownership, but makes the intent clearer.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/command/graph/graph_write.rs` around lines 127 - 140, Move ownership of
labels/properties/embedding into local owned variables before taking the mutable
borrow so you don't clone while the graph is borrowed; e.g., create owned values
from the originals (using std::mem::take(&mut labels) / take(&mut properties) /
take(&mut embedding) or otherwise convert the outer references into owned
containers) and then call store.get_graph_mut(graph_name) and
graph.write_buf.add_node(owned_labels, owned_properties, owned_embedding, lsn);
keep the graph.stats.on_node_insert(&owned_labels) call and return
node_key.data().as_ffi() as before so the graph borrow is released before WAL
push without performing .clone().

src/graph/wal.rs (1)

289-303: Consider asserting total element count.

Line 295 comments "= 7" but doesn't assert elems.len() == 7. Adding the assertion would catch format changes:

💚 Suggested fix

         let elems = parse_resp_array(&data);
-        // cmd + graph + node_id + num_labels(1) + 2 labels + num_props(1) = 7
+        // cmd + graph + node_id + num_labels(1) + 2 labels + num_props(1) = 7
+        assert_eq!(elems.len(), 7);
         assert_eq!(elems[0], b"GRAPH.ADDNODE");

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/wal.rs` around lines 289 - 303, The test
test_serialize_add_node_no_props_no_embedding should explicitly assert the total
RESP element count to catch future format changes: add an assertion like
assert_eq!(elems.len(), 7) (or the appropriate constant) immediately after
parsing with parse_resp_array(&data) in that test, and update the comment if you
change the expected count; this references the elems vector produced by
parse_resp_array and the test function name to locate where to add the
assertion.

src/graph/segment.rs (2)

65-78: Same race condition applies; also consider HashSet for larger LSN sets.

The non-atomic read-modify-write pattern has the same lost-update risk. Additionally, old_lsns.contains() is O(m) per segment. If compacting many segments, consider using a HashSet:

♻️ Suggested optimization

+    use std::collections::HashSet;
+
     pub fn replace_immutable(&self, old_lsns: &[u64], new_csr: CsrSegment) {
+        let old_set: HashSet<u64> = old_lsns.iter().copied().collect();
         let current = self.segments.load();
         let mut new_immutable: Vec<Arc<CsrSegment>> = current
             .immutable
             .iter()
-            .filter(|seg| !old_lsns.contains(&seg.created_lsn))
+            .filter(|seg| !old_set.contains(&seg.created_lsn))
             .cloned()
             .collect();

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/segment.rs` around lines 65 - 78, The replace_immutable function
does a non-atomic read-modify-write on self.segments (using segments.load then
segments.store) which can lose concurrent updates; change it to a CAS loop that
loads current = self.segments.load(), builds the new immutable list (filtering
using a HashSet constructed from old_lsns for O(1) lookups), create the new
Arc<GraphSegmentList> with current.mutable.clone() and the new immutable vec,
then attempt to swap via compare_and_swap / compare_exchange (retrying on
failure) so updates are atomic and no concurrent update is lost; keep the
insertion of the new CsrSegment as Arc::new(new_csr) at index 0 as before.

---

104-128: Test doesn't catch the race condition.

The test spawns a thread and joins it before checking the old snapshot. This verifies Arc lifetime semantics but doesn't expose the race in add_immutable/replace_immutable. Consider adding a loom test for concurrent modifications.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/segment.rs` around lines 104 - 128, The test
test_concurrent_readers_see_consistent_snapshot only verifies Arc lifetime
because it joins the spawned thread before asserting; instead add a concurrent
loom-based test that exercises GraphSegmentHolder's mutation paths (call sites
like add_immutable and replace_immutable) without serializing with join so
interleavings are explored: write a loom::model test that spawns multiple
loom::thread::spawn closures which concurrently call
GraphSegmentHolder::add_immutable / GraphSegmentHolder::replace_immutable while
other threads retain a loaded snapshot (holder.load()) and assert snapshots
remain consistent across interleavings, using GraphSegmentHolder,
GraphSegmentList, add_immutable and replace_immutable identifiers to locate the
logic to stress.

src/shard/shared_databases.rs (1)

121-127: Consider propagating recovery errors instead of only logging.

Recovery failures are logged but don't prevent the server from starting. If graph persistence was enabled and recovery fails, users might unknowingly lose data. Consider returning a Result from recover_graph_stores so the caller can decide whether to fail startup.

♻️ Suggested approach

-    pub fn recover_graph_stores(&self, persistence_dir: &std::path::Path) {
+    pub fn recover_graph_stores(&self, persistence_dir: &std::path::Path) -> Result<(), String> {
         for shard_id in 0..self.num_shards {
             match crate::graph::recovery::recover_graph_store(persistence_dir, shard_id) {
                 // ... existing Ok cases ...
                 Err(e) => {
-                    tracing::error!(
-                        "Shard {}: graph recovery failed: {}",
-                        shard_id,
-                        e
-                    );
+                    return Err(format!("Shard {}: graph recovery failed: {}", shard_id, e));
                 }
             }
         }
+        Ok(())
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/shard/shared_databases.rs` around lines 121 - 127, recover_graph_stores
currently swallows recovery errors (logs via tracing::error with shard_id) which
allows startup to proceed; change recover_graph_stores to return Result<(),
YourErrorType> (or anyhow::Result) and propagate errors to callers instead of
only logging, update call sites to handle/propagate the Result (use ? where
appropriate) so that when graph persistence is enabled a failed recovery can
cause startup to fail; remove or convert the tracing::error block in
recover_graph_stores to return Err(e) (or attach context and return Err) so
callers can decide failure behavior.

src/graph/cypher/lexer.rs (1)

219-223: Silent skip of unrecognized bytes may mask issues.

The lexer silently skips Err(()) (unrecognized bytes) and continues. While this provides resilience, it could mask malformed input that the parser should reject. Consider tracking skipped bytes or returning a specific error token for the parser to handle.

♻️ Alternative: Add an Error token

+    /// Unrecognized input that couldn't be lexed.
+    Error(&'a [u8]),
+
                 Err(()) => {
-                    // Skip unrecognized bytes and continue.
-                    // The parser will handle unexpected tokens.
-                    continue;
+                    // Return error token so parser can produce proper error.
+                    return Some(SpannedToken {
+                        token: Token::Error(self.inner.slice()),
+                        span,
+                    });
                 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/cypher/lexer.rs` around lines 219 - 223, The lexer currently
swallows Err(()) and continues, which can hide malformed input; update the lexer
in lexer.rs (the match arm handling Err(()) inside the tokenization function) to
emit a specific error token instead of silently skipping: add or use a
Token::Error (or Token::InvalidByte) variant and return/push it with the
offending byte and current position (use the existing cursor/pos/index
variable), or alternatively increment a skipped-bytes counter and attach that
info to the token stream so the parser can detect and handle it. Ensure the
change updates any consumers to handle Token::Error and preserves existing
behavior when callers explicitly choose to ignore errors.

benches/graph_bench.rs (1)

203-221: Minor: Duration scaling may lose precision.

Line 218 scales the measured duration: total * (iters as u32) / (rounds as u32). For large iteration counts, this could lose precision due to integer operations. Consider using floating-point for the ratio:

♻️ Suggested fix

-            total * (iters as u32) / (rounds as u32)
+            total.mul_f64(iters as f64 / rounds as f64)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@benches/graph_bench.rs` around lines 203 - 221, The duration scaling uses
integer arithmetic which can lose precision; in bench_csr_freeze replace the
integer scale expression that uses total, iters and rounds with a floating-point
scale (e.g., use Duration::mul_f64 or otherwise compute the ratio as a f64) so
you compute total.mul_f64(iters as f64 / rounds as f64) (refer to the
bench_csr_freeze function and variables total, iters, rounds and
CsrSegment::from_frozen).

src/graph/compaction.rs (2)

249-253: Consider using RoaringBitmap::from_iter for bulk insertion.

Inserting edges one-by-one in a loop is less efficient than bulk construction:

-    let mut validity = RoaringBitmap::new();
-    for i in 0..edge_count as u32 {
-        validity.insert(i);
-    }
+    let validity = RoaringBitmap::from_iter(0..edge_count as u32);

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/compaction.rs` around lines 249 - 253, Replace the manual loop that
populates the RoaringBitmap named validity with a bulk constructor: use
RoaringBitmap::from_iter over the range 0..edge_count as u32 (or
0u32..edge_count as u32) to build the bitmap in one step; update the
initialization (the validity variable) to use from_iter instead of creating an
empty bitmap and calling insert in a loop.

---

378-386: Quadratic complexity in community degree computation.

For each node and each candidate community, sigma_tot is computed by iterating all nodes to sum degrees. This is O(n² × avg_neighbors) in the worst case. For large graphs, this could make compaction very slow.

Consider precomputing and maintaining community degree sums incrementally:

♻️ Suggested optimization

+    // Precompute community degree sums
+    let mut comm_degree: Vec<f64> = degree.clone(); // Initially each node is its own community
+
     for u in 0..node_count {
         // ...
         for (&target_comm, &edge_weight_to_comm) in &comm_weights {
-            let sigma_tot: f64 = community
-                .iter()
-                .enumerate()
-                .filter(|&(_, c)| *c == target_comm)
-                .map(|(i, _)| degree[i])
-                .sum();
+            let sigma_tot = comm_degree[target_comm as usize];
             // ...
         }
         if best_comm != u_comm {
+            // Update community degree sums
+            comm_degree[u_comm as usize] -= degree[u];
+            comm_degree[best_comm as usize] += degree[u];
             community[u] = best_comm;
         }
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/compaction.rs` around lines 378 - 386, The current loop over for
(&target_comm, &edge_weight_to_comm) recalculates sigma_tot by scanning
community and summing degree for each candidate, which is quadratic; instead add
a precomputed map/array (e.g., community_deg_sum: HashMap<CommId, f64> or
Vec<f64>) computed once before the outer node loop and use
community_deg_sum[&target_comm] to get sigma_tot, and update community_deg_sum
incrementally whenever you change a node's community (the same places that
modify community[] or move nodes in the compaction routine). Replace the
sigma_tot computation in the for loop with a lookup and ensure all node-move
code paths update community_deg_sum and any related totals to keep values
consistent.

src/command/graph/mod.rs (1)

35-95: Duplicate dispatch logic between the two entry points.

dispatch_graph_command and dispatch_graph_cmd_args share identical if-else chains for routing commands. Consider extracting the routing logic to a single function that both call:

♻️ Suggested refactor

+fn dispatch_graph_impl(store: &mut GraphStore, cmd: &[u8], args: &[Frame]) -> Frame {
+    if cmd.eq_ignore_ascii_case(b"GRAPH.CREATE") {
+        graph_create(store, args)
+    } else if cmd.eq_ignore_ascii_case(b"GRAPH.ADDNODE") {
+        graph_addnode(store, args)
+    // ... rest of routing
+    } else {
+        Frame::Error(Bytes::from_static(b"ERR unknown GRAPH.* command"))
+    }
+}
+
 pub fn dispatch_graph_command(store: &mut GraphStore, command: &Frame) -> Frame {
     let (cmd, args) = match extract_command(command) {
         Some(pair) => pair,
         None => {
             return Frame::Error(Bytes::from_static(
                 b"ERR invalid command format for GRAPH.*",
             ));
         }
     };
-
-    if cmd.eq_ignore_ascii_case(b"GRAPH.CREATE") {
-        graph_create(store, args)
-    // ... duplicated routing
+    dispatch_graph_impl(store, cmd, args)
 }

 pub fn dispatch_graph_cmd_args(store: &mut GraphStore, cmd: &[u8], args: &[Frame]) -> Frame {
-    if cmd.eq_ignore_ascii_case(b"GRAPH.CREATE") {
-    // ... duplicated routing
+    dispatch_graph_impl(store, cmd, args)
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/command/graph/mod.rs` around lines 35 - 95, Both dispatch_graph_command
and dispatch_graph_cmd_args duplicate the same if-else routing chain; extract
that routing into a single helper (e.g., route_graph_cmd) that accepts the
GraphStore, the cmd bytes, and an Option<&[Frame]> (or two overloads) and
returns Frame; replace both dispatch_graph_command and dispatch_graph_cmd_args
to call this helper and forward args (None for the caller that only has raw
Frame list), moving the existing comparisons (cmd.eq_ignore_ascii_case(...)) and
corresponding calls (graph_create, graph_addnode, etc.) into the new helper so
there is a single authoritative dispatch point.

src/graph/cross_shard.rs (1)

176-179: to_vec() clones all frames from the response.

items[1].to_vec() allocates and clones the entire frame array. If the traversal returns many frames, this could be expensive. Consider returning a reference or using Arc to avoid the clone if the parsed result is short-lived.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/cross_shard.rs` around lines 176 - 179, The current match in the
parsing code clones the entire frame array via Frame::Array(f) => f.to_vec(),
which is expensive; change it to borrow or share instead: bind frames to a
reference (e.g., Frame::Array(f) => f or f.as_slice()) or wrap the parsed frame
vector in Arc (e.g., return Arc<Vec<Frame>>) so you avoid the clone; update the
surrounding function signature and any consumers of frames (the variable named
frames and the match on items[1]) to accept a &[Frame] or Arc<Vec<Frame>>
accordingly to prevent the unnecessary allocation and clone.

src/server/conn/handler_sharded.rs (1)

1762-1771: Registry update after batch partially addresses drift, but state changes mid-batch are still missed.

The batch-end update is a good addition. However, state changes from AUTH, HELLO, SELECT, SUBSCRIBE, MULTI/EXEC/DISCARD during command processing still won't be reflected until after the batch completes. For long-running batches or blocking commands, CLIENT LIST from another connection may show stale state.

Consider updating the registry immediately after critical state transitions (user change, db change, subscription count change) in addition to the batch-end update.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_sharded.rs` around lines 1762 - 1771, The registry
update at batch end misses mid-batch state transitions; after each critical
state change (AUTH/HELLO when conn.user is set, SELECT when conn.selected_db is
changed, SUBSCRIBE/UNSUBSCRIBE when conn.subscription_count changes, and
MULTI/EXEC/DISCARD when conn.in_multi toggles) call
crate::client_registry::update(client_id, |e| { e.db = conn.selected_db;
e.last_cmd_at = std::time::Instant::now(); e.flags =
crate::client_registry::ClientFlags { subscriber: conn.subscription_count > 0,
in_multi: conn.in_multi, blocked: false, }; }); so the registry reflects changes
immediately (keep the existing batch-end update as well); locate these points in
the command handling code paths that process AUTH/HELLO, SELECT,
SUBSCRIBE/UNSUBSCRIBE and MULTI/EXEC/DISCARD and insert the update call right
after the state mutation.

src/graph/replay.rs (2)

359-365: GRAPH.REMOVEEDGE during replay silently skips operations.

While the warning is logged, the command is counted as "not replayed" but processing continues. This could lead to inconsistent graph state if edge removals were recorded in the WAL. Consider documenting this limitation more prominently or tracking skipped commands in the return value.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/replay.rs` around lines 359 - 365, The current
GraphCommand::RemoveEdge arm only logs a warning and silently skips the command;
update the replay path that processes GraphCommand (the function handling WAL
replay) to record skipped removals in the replay result/metrics instead of just
continuing: add/increment a counter (e.g., skipped_remove_edge_count or
increment the existing not_replayed/not_applied counter on the
ReplayResult/ReplayStats structure returned by the replay function) when
encountering GraphCommand::RemoveEdge, keep the tracing::warn, and ensure the
ReplayResult type and all call sites are updated to carry and surface that
counter so skipped edge-removes are visible to callers/metrics.

---

76-78: Avoid allocation on each collect_command call.

cmd.to_ascii_uppercase() allocates a new Vec<u8> on every invocation. Since collect_command is called for every WAL entry during replay, consider matching case-insensitively directly:

♻️ Suggested refactor

     pub fn collect_command(&mut self, cmd: &[u8], args: &[&[u8]]) -> bool {
-        let upper = cmd.to_ascii_uppercase();
-        match upper.as_slice() {
-            b"GRAPH.CREATE" => {
+        if cmd.eq_ignore_ascii_case(b"GRAPH.CREATE") {
+            if args.is_empty() {
+                return false;
+            }
+            self.commands.push(GraphCommand::Create {
+                name: Bytes::copy_from_slice(args[0]),
+            });
+            true
+        } else if cmd.eq_ignore_ascii_case(b"GRAPH.DROP") {
+            // ... similar pattern for other commands

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/replay.rs` around lines 76 - 78, collect_command currently calls
cmd.to_ascii_uppercase() which allocates per WAL entry; replace this with a
non-allocating case-insensitive comparison. In collect_command, remove the
to_ascii_uppercase() call and either (A) try to convert cmd to &str with
std::str::from_utf8 and use str::eq_ignore_ascii_case against literal commands,
or (B) implement a small helper (e.g., fn eq_ignore_ascii_case_bytes(a: &[u8],
b: &[u8]) -> bool) that compares lengths and checks a.iter().zip(b).all(|(x,y)|
x.to_ascii_uppercase() == y.to_ascii_uppercase()), then use that helper to match
against byte literals (e.g., b"PING", b"SET") so no Vec allocation occurs in
collect_command.

src/graph/store.rs (1)

182-186: Silently ignoring duplicate graph errors during metadata load could hide corruption.

If graph_meta.json contains duplicate entries (possibly due to corruption or a bug during save), the duplicates are silently ignored. Consider logging a warning:

         for entry in &meta.graphs {
             let name = Bytes::from(entry.name.clone());
-            // Ignore duplicate errors (shouldn't happen from valid metadata).
-            let _ = store.create_graph(name, entry.edge_threshold, entry.created_lsn);
+            if store.create_graph(name.clone(), entry.edge_threshold, entry.created_lsn).is_err() {
+                tracing::warn!(
+                    graph = %String::from_utf8_lossy(&name),
+                    "duplicate graph entry in metadata file, skipping"
+                );
+            }
         }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/store.rs` around lines 182 - 186, During metadata load, don't
silently ignore errors from store.create_graph called in the loop over
meta.graphs; instead match the Result from store.create_graph(name,
entry.edge_threshold, entry.created_lsn) and handle errors: if the error
represents a duplicate-graph condition (match the concrete error variant or
check its kind/message) emit a warning including entry.name and the error
details (using the project's logger e.g., warn! or tracing::warn!), otherwise
propagate or return the error so non-duplicate failures are not swallowed; keep
the loop over meta.graphs and use the existing Bytes::from(entry.name.clone())
to identify the offending entry.

src/command/graph/graph_read.rs (1)

315-335: GRAPH.RO_QUERY reparses and replans the same query.

This already parsed the Cypher and checked is_read_only(), then calls graph_query(), which repeats the full parse+compile path. Reuse the validated query/plan here instead of paying that second pass on every RO request.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/command/graph/graph_read.rs` around lines 315 - 335, The RO handler
currently reparses and replans by calling graph_query(store, args) after already
producing a parsed query via cypher::parse_cypher and validating
query.is_read_only(); to fix, refactor graph_query to accept a pre-parsed
query/plan (or add a new helper like graph_query_with_plan) so you can pass the
existing parsed Query/Plan object instead of raw args, and update this handler
to call that new API (use the same validation/error paths such as is_read_only
and existing Frame::Error returns) so the second parse/compile is skipped;
reference the symbols extract_bulk, cypher::parse_cypher, Query::is_read_only
(or is_read_only), and graph_query/graph_query_with_plan to locate the change
points.

src/graph/csr.rs (2)

214-218: Neighbor query functions assume valid row index.

Functions like neighbors_out, neighbors_by_label, edges_by_type, and neighbor_edges will panic if row >= node_count. Since lookup_node returns Option<u32>, this is acceptable for internal use, but consider documenting the precondition or adding debug assertions.

📝 Optional: Add debug assertion

     /// Returns the slice of outgoing neighbor row indices for the given CSR row.
     pub fn neighbors_out(&self, row: u32) -> &[u32] {
+        debug_assert!(
+            (row as usize) < self.row_offsets.len().saturating_sub(1),
+            "row index out of bounds"
+        );
         let start = self.row_offsets[row as usize] as usize;
         let end = self.row_offsets[row as usize + 1] as usize;
         &self.col_indices[start..end]
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/csr.rs` around lines 214 - 218, Add a precondition check to avoid
panics when callers pass an out-of-range row: insert a debug assertion (or
explicit bounds check if you prefer runtime safety) at the start of
neighbors_out, neighbors_by_label, edges_by_type, and neighbor_edges that
validates row < self.node_count() (or row as usize < self.row_offsets.len() -
1); this documents the requirement and prevents silent panics during debug
builds while preserving performance in release builds.

---

162-166: Consider using add_range for validity bitmap initialization.

Inserting elements one-by-one in a loop is less efficient than using Roaring's range insertion. The same pattern appears in from_bytes at lines 491-494.

♻️ Suggested optimization

         // Initialize validity bitmap: all edges valid.
-        let mut validity = RoaringBitmap::new();
-        for i in 0..edge_count as u32 {
-            validity.insert(i);
-        }
+        let mut validity = RoaringBitmap::new();
+        if edge_count > 0 {
+            validity.insert_range(0..edge_count as u32);
+        }

Apply the same change at lines 491-494 in from_bytes.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/csr.rs` around lines 162 - 166, Replace the one-by-one insertion
loop that fills the RoaringBitmap named `validity` with a single range insertion
call (use `validity.add_range(0..edge_count as u64)` or the appropriate integer
width for your RoaringBitmap version) to initialize all edges as valid; also
make the identical replacement in the `from_bytes` implementation where the same
per-element loop appears to build the validity bitmap. Ensure the range end is
exclusive and matches the original loop semantics (0 up to edge_count).

src/graph/cypher/ast.rs (1)

14-28: Consider adding REMOVE clause for completeness.

Standard Cypher includes REMOVE for deleting properties and labels (e.g., REMOVE n.prop, REMOVE n:Label). If you plan to support full Cypher compatibility, this should be added alongside the write clause check in is_read_only().

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/cypher/ast.rs` around lines 14 - 28, The Clause enum is missing a
REMOVE variant; add a Remove(RemoveClause) variant to the Clause enum and define
a corresponding RemoveClause struct/type where other clause types are declared
(e.g., alongside CreateClause, DeleteClause, SetClause). Update the
is_read_only() logic (the function/method that checks read-only status) to treat
Clause::Remove as a write-modifying clause (i.e., return false / not read-only)
alongside Create/Delete/Merge/Set, and update any pattern matches or exhaustive
matches over Clause to handle the new Remove variant.

src/graph/hybrid.rs (2)

391-406: Prefer destructuring pattern to discard unused binding.

Using let _ = edge_key; works but is less idiomatic than using _ in the pattern directly.

♻️ Cleaner pattern

-                for (edge_key, neighbor_key) in memgraph.neighbors(current, Direction::Both, lsn) {
+                for (_, neighbor_key) in memgraph.neighbors(current, Direction::Both, lsn) {
                     if visited.contains(&neighbor_key) {
                         continue;
                     }

                     let sim = memgraph
                         .get_node(neighbor_key)
                         .and_then(|n| n.embedding.as_ref())
                         .map(|emb| cosine_similarity(emb, &self.query_vector))
                         .unwrap_or(0.0);

-                    // Use edge_key to avoid unused variable warning.
-                    let _ = edge_key;
-
                     candidates.push((neighbor_key, sim));
                 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/hybrid.rs` around lines 391 - 406, The loop currently binds an
unused variable edge_key and then discards it with let _ = edge_key;—replace the
binding with a discard in the pattern itself by changing the loop header from
for (edge_key, neighbor_key) in memgraph.neighbors(...) to for (_, neighbor_key)
in memgraph.neighbors(...), and remove the redundant let _ = edge_key; line so
only neighbor_key is bound and used in the body (affecting the loop that
constructs candidates and the use of memgraph.get_node and cosine_similarity).

---

198-203: Strategy selection computed but unused.

The _strategy variable is computed but never used. The comment explains this is intentional since HNSW pre-filter isn't implemented for MemGraph embeddings yet. Consider either removing the dead code or adding a TODO for tracking.

📝 Option: Add tracking comment

         // Step 2: Select strategy.
-        let _strategy = select_strategy(candidates.len(), self.threshold);
-        // Both strategies score the same way for MemGraph embeddings.
-        // HNSW pre-filter would be used when an external VectorStore index exists.
-        // For MemGraph-embedded vectors, brute-force is always used.
+        // TODO(perf): Implement HNSW pre-filter strategy for large candidate sets.
+        // Currently brute-force is used for all MemGraph-embedded vectors since
+        // they aren't indexed in HNSW. When VectorStore index exists, use:
+        //   let strategy = select_strategy(candidates.len(), self.threshold);

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/hybrid.rs` around lines 198 - 203, Remove the dead computation of
_strategy (the call select_strategy(candidates.len(), self.threshold)) since its
result is never used; alternatively, if you want to retain the call for future
telemetry, remove the assignment and add a single-line TODO comment above
explaining why the strategy selection is intentionally omitted for MemGraph
embeddings and reference select_strategy/candidates.len()/self.threshold so it's
clear this is deliberate and tracked.

src/graph/traversal.rs (2)

183-191: CSR edges use hardcoded weight of 1.0.

Edge weights aren't stored in EdgeMeta, so CSR edges default to weight 1.0. This affects weighted traversals like Dijkstra across CSR segments. Consider documenting this limitation or extending EdgeMeta to include weight if weighted CSR traversal is needed.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/traversal.rs` around lines 183 - 191, CSR edges are currently
hardcoded to weight 1.0 when constructing MergedNeighbor, which breaks weighted
algorithms; update EdgeMeta to carry an optional weight (e.g., Option<f64>),
ensure any CSR segment writing/reading code sets that weight when available, and
change the creation logic (the result.push that builds MergedNeighbor using
csr.created_lsn) to use meta.weight.unwrap_or(1.0) for MergedNeighbor.weight;
alternatively, if you don't want to change storage, add a comment/doc near
EdgeMeta and the result.push call documenting that CSR edges are unweighted and
default to 1.0.

---

399-421: PartialEq implementation compares f64 with == which doesn't handle NaN.

The DijkstraEntry PartialEq uses self.cost == other.cost, which returns false for NaN comparisons. While costs shouldn't be NaN in practice, this could cause subtle bugs if they ever are. The Ord impl handles this with unwrap_or(Ordering::Equal).

♻️ Consider total ordering wrapper

You could use ordered_float::OrderedFloat<f64> for total ordering, or document that NaN costs are undefined behavior. The current implementation is acceptable if costs are guaranteed non-NaN.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/traversal.rs` around lines 399 - 421, The PartialEq impl for
DijkstraEntry currently uses self.cost == other.cost which treats NaN specially;
change the equality check to mirror the Ord behavior by comparing via
self.cost.partial_cmp(&other.cost).unwrap_or(Ordering::Equal) == Ordering::Equal
(i.e., update the PartialEq impl for DijkstraEntry to use
partial_cmp+unwrap_or(Ordering::Equal) instead of ==), or alternatively replace
the f64 cost with ordered_float::OrderedFloat<f64> and derive PartialEq/Eq/Ord;
update the PartialEq impl (or struct field) accordingly so DijkstraEntry, its
PartialEq impl and Ord/PartialOrd remain consistent.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Avoid hard-coded test skips in the redis-compat gate without replacement coverage.

Line 111 now skips xadd, xrange, and xtrim, which reduces compatibility signal in CI. If these are flaky, move them to a quarantined job instead of silently removing them from the primary compat run.

Suggested fix

-          MOON_TEST_PORT=6399 cargo test --release --no-default-features --features runtime-tokio,jemalloc --test redis_compat -- --ignored --skip xadd --skip xrange --skip xtrim
+          MOON_TEST_PORT=6399 cargo test --release --no-default-features --features runtime-tokio,jemalloc --test redis_compat -- --ignored

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	MOON_TEST_PORT=6399 cargo test --release --no-default-features --features runtime-tokio,jemalloc --test redis_compat -- --ignored --skip xadd --skip xrange --skip xtrim
	MOON_TEST_PORT=6399 cargo test --release --no-default-features --features runtime-tokio,jemalloc --test redis_compat -- --ignored

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In @.github/workflows/ci.yml at line 111, The CI step invoking
"MOON_TEST_PORT=6399 cargo test ... --test redis_compat -- --ignored --skip xadd
--skip xrange --skip xtrim" is silently removing coverage for xadd/xrange/xtrim;
instead remove the --skip flags from the redis_compat invocation so the compat
gate runs them, and if they are flaky extract those three tests into a dedicated
quarantined job/step (e.g., a separate CI job that runs only xadd, xrange, xtrim
with retries or marked flaky) and update the workflow to run that quarantined
job in parallel while keeping the primary redis_compat run unskipped. Ensure the
new job name and step reference the same cargo test invocation but targeting
only the three tests or using --skip inversely so primary coverage is preserved.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

cat -n .github/workflows/compat.yml | sed -n '375,395p'

Repository: pilotspace/moon

Length of output: 1767

---

Harden JAR downloads: fail fast on HTTP errors and add checksum verification.

Lines 382-385 use curl -sL, which won't fail if an HTTP error occurs (missing -f flag). Additionally, there is no validation of artifact integrity via checksums. This can allow corrupted or error responses to be silently downloaded and used.

Replace curl -sL with curl -fSL --retry 3 --retry-delay 2 to fail immediately on HTTP errors and retry transient failures. Add checksum verification by downloading the .sha256 files from Maven Central and validating each JAR before use.

Example hardening approach

-          curl -sL "https://repo1.maven.org/maven2/io/lettuce/lettuce-core/${LETTUCE_VERSION}/lettuce-core-${LETTUCE_VERSION}.jar" -o /tmp/lettuce-test/lettuce-core.jar
+          curl -fSL --retry 3 --retry-delay 2 "https://repo1.maven.org/maven2/io/lettuce/lettuce-core/${LETTUCE_VERSION}/lettuce-core-${LETTUCE_VERSION}.jar" -o /tmp/lettuce-test/lettuce-core.jar
+          curl -fSL --retry 3 --retry-delay 2 "https://repo1.maven.org/maven2/io/lettuce/lettuce-core/${LETTUCE_VERSION}/lettuce-core-${LETTUCE_VERSION}.jar.sha256" -o /tmp/lettuce-test/lettuce-core.jar.sha256
+          (cd /tmp/lettuce-test && sha256sum -c lettuce-core.jar.sha256)

Apply similar changes for netty-all, reactor-core, and reactive-streams JARs.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In @.github/workflows/compat.yml around lines 382 - 385, Replace the fragile
curl invocations that use "curl -sL" for the
lettuce/netty/reactor/reactive-streams downloads with hardened commands that
fail fast and retry (use "curl -fSL --retry 3 --retry-delay 2" and keep the
existing -o target) and add checksum verification: for each artifact referenced
by LETTUCE_VERSION, NETTY_VERSION, REACTOR_VERSION, and REACTIVE_STREAMS_VERSION
download the corresponding .sha256 (or .sha256.txt) from Maven Central, then
verify the downloaded JAR using a SHA‑256 check (e.g., via "sha256sum -c" or
equivalent) and abort the workflow if the checksum fails.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

# First, let's find the compat.yml file and examine its contents
find . -name "compat.yml" -type f

Repository: pilotspace/moon

Length of output: 89

---

🏁 Script executed:

# Also check the file structure to understand the workflow layout
fd -t f "\.github/workflows" | head -20

Repository: pilotspace/moon

Length of output: 41

---

🏁 Script executed:

# Let's examine the compat.yml file to see the context around line 437
wc -l .github/workflows/compat.yml

Repository: pilotspace/moon

Length of output: 91

---

🏁 Script executed:

# Read the relevant section of the workflow file around line 437
sed -n '420,450p' .github/workflows/compat.yml

Repository: pilotspace/moon

Length of output: 1053

---

🏁 Script executed:

# Let's see more context about the test and check the entire file structure
sed -n '430,468p' .github/workflows/compat.yml

Repository: pilotspace/moon

Length of output: 1545

---

🏁 Script executed:

# Check if there are any version constraints documented elsewhere in the project
rg "StackExchange\.Redis" --type-list all -i

Repository: pilotspace/moon

Length of output: 448

---

🏁 Script executed:

# Search for StackExchange.Redis references in the codebase
rg "StackExchange\.Redis" -i

Repository: pilotspace/moon

Length of output: 195

---

🏁 Script executed:

# Let's also check the git history of this workflow to see when it was added
git log --oneline .github/workflows/compat.yml | head -10

Repository: pilotspace/moon

Length of output: 111

---

🏁 Script executed:

# Let's understand the purpose of this workflow better by looking at the beginning
head -50 .github/workflows/compat.yml

Repository: pilotspace/moon

Length of output: 1532

---

🏁 Script executed:

# Check when this specific test was added or modified
git log -p .github/workflows/compat.yml | grep -A 10 -B 5 "StackExchange.Redis" | head -40

Repository: pilotspace/moon

Length of output: 41

---

🏁 Script executed:

# Let's check the complete workflow to see all compatibility tests and version pinning patterns
cat .github/workflows/compat.yml | grep -A 5 -B 5 "add package\|pip install\|npm install" | head -60

Repository: pilotspace/moon

Length of output: 1098

---

🏁 Script executed:

# Let's also look for any other version pinning patterns in the workflow
grep -n "version" .github/workflows/compat.yml

Repository: pilotspace/moon

Length of output: 339

---

🏁 Script executed:

# Let's check if there are any documentation or comments about version strategy
grep -r "version\|deterministic\|pinning" docs/ README.md 2>/dev/null | head -20

Repository: pilotspace/moon

Length of output: 2124

---

🏁 Script executed:

# Check if there are any recent issues or PRs related to CI stability or version pinning
git log --all --oneline --grep="version\|deterministic\|pin" | head -10

Repository: pilotspace/moon

Length of output: 742

---

🏁 Script executed:

# Let's also check if there are any other workflows that pin versions
grep -r "version\|@\|--version" .github/workflows/ | grep -E "pip|npm|dotnet|add package" | head -20

Repository: pilotspace/moon

Length of output: 188

---

🏁 Script executed:

# Let's check if there are any recent issues, failures, or discussions about this
git log --all --oneline | grep -i "stackexchange\|dotnet\|compat" | head -20

Repository: pilotspace/moon

Length of output: 436

---

🏁 Script executed:

# Check the full compat test section for all clients to see if there's a pattern
sed -n '1,100p' .github/workflows/compat.yml | grep -E "pip install|npm install|add package|Install"

Repository: pilotspace/moon

Length of output: 120

---

🏁 Script executed:

# Let's look at the end of the file to see all test jobs
tail -100 .github/workflows/compat.yml

Repository: pilotspace/moon

Length of output: 4660

---

🏁 Script executed:

# Let's check if there's any documentation about version requirements in the project
cat README.md | grep -A 5 -B 5 "version\|client" | head -50

Repository: pilotspace/moon

Length of output: 2363

---

🏁 Script executed:

# Let's also check if there are any issues or recent discussions about compatibility test failures
git log --all --oneline .github/workflows/compat.yml | head -20

Repository: pilotspace/moon

Length of output: 183

---

🏁 Script executed:

# Check what the actual latest version of StackExchange.Redis is
# Let's see if there's any package.lock or similar file that might pin this
fd -t f "package.lock\|packages.lock\|\.csproj\|global.json" 2>/dev/null

Repository: pilotspace/moon

Length of output: 41

---

🏁 Script executed:

# Let's look at the commit that added the compat tests to understand the intent
git show 794be8d --stat | head -20

Repository: pilotspace/moon

Length of output: 640

---

🏁 Script executed:

# Check if there are any recent CI failures or issues related to this
git log --all --oneline --grep="compat\|stack\|dotnet" | head -10

Repository: pilotspace/moon

Length of output: 778

---

🏁 Script executed:

# Let's see the pattern: are version pins used for other tests?
cat .github/workflows/compat.yml | grep -E "VERSION|version" | grep -v "python-version\|dotnet-version\|java-version\|node-version\|go-version"

Repository: pilotspace/moon

Length of output: 1904

---

🏁 Script executed:

# Let's check the history of when this compat test was introduced
git show 43b0462 --stat

Repository: pilotspace/moon

Length of output: 1641

---

🏁 Script executed:

# Check if there are any CI failures or issues related to StackExchange.Redis
git log --all --oneline --grep="stackexchange\|dotnet.*compat\|Redis.*test" -i | head -10

Repository: pilotspace/moon

Length of output: 743

---

🏁 Script executed:

# Let's look more carefully at how the Java tests handle versioning and compare
sed -n '300,400p' .github/workflows/compat.yml | grep -A 15 "jedis-test"

Repository: pilotspace/moon

Length of output: 2346

---

🏁 Script executed:

# Search for any existing StackExchange.Redis version references or constraints in the codebase
rg "2\.8\.|StackExchange" -i

Repository: pilotspace/moon

Length of output: 224

---

🌐 Web query:

StackExchange.Redis version 2.8.31 release history

💡 Result:

StackExchange.Redis version 2.8.31 was released on March 7, 2025, according to NuGet records. The key changes in this release are: - Fix: Respect IReconnectRetryPolicy timing in the case that a node that was present disconnects indefinitely (pull requests #2853 & #2856 by NickCraver). Special thanks to sampdei for tracking this down and working a fix. - Changes max default retry policy backoff to 60 seconds (#2853 by NickCraver). - Fix #2652: Track client-initiated shutdown for any pipe type (#2814 by bgrainger). This information is from the official release notes.

Citations:

• 1: https://github.com/StackExchange/StackExchange.Redis/blob/328b4b5a/docs/ReleaseNotes.md
• 2: https://www.nuget.org/packages/StackExchange.Redis/2.8.31
• 3: https://stackexchange.github.io/StackExchange.Redis/ReleaseNotes.html

---

🌐 Web query:

StackExchange.Redis latest version 2025 2026

💡 Result:

The latest version of StackExchange.Redis is 2.12.14, released on 2026-04-02. This is confirmed as the most recent release on the official GitHub repository. Previous recent releases include 2.12.8 (2026-03-25, listed on NuGet with 66,675 downloads), 2.12.4, 2.12.1 (2026-03-13), 2.11.8 (2026-02-25), and 2.11.3 (2026-02-19). The NuGet package page shows 2.12.8 as the latest stable with recent downloads, but GitHub releases confirm 2.12.14 as newer, addressing critical issues like high-integrity-mode in cluster scenarios.

Citations:

• 1: https://github.com/StackExchange/StackExchange.Redis/releases
• 2: https://www.github.com/StackExchange/StackExchange.Redis
• 3: https://stackexchange.github.io/StackExchange.Redis/ReleaseNotes.html
• 4: https://www.nuget.org/packages/stackexchange.redis/
• 5: https://github.com/StackExchange/StackExchange.Redis/releases/tag/2.12.14
• 6: https://github.com/StackExchange/StackExchange.Redis/blob/main/docs/ReleaseNotes.md
• 7: https://github.com/StackExchange/StackExchange.Redis?wt.mc_id=DT-MVP-5004638

---

Pin StackExchange.Redis version for deterministic CI.

Line 437 installs the latest package version at runtime, which can break compatibility unexpectedly. This is inconsistent with the Java tests in the same workflow (Jedis, Lettuce), which explicitly pin dependency versions. Update to use version 2.12.14 (latest stable as of April 2026).

Suggested fix

+        env:
+          STACKEXCHANGE_REDIS_VERSION: "2.12.14"
         run: |
           mkdir -p /tmp/se-redis-test
           cd /tmp/se-redis-test
           dotnet new console -n CompatTest --force
           cd CompatTest
-          dotnet add package StackExchange.Redis
+          dotnet add package StackExchange.Redis --version ${STACKEXCHANGE_REDIS_VERSION}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In @.github/workflows/compat.yml at line 437, Update the workflow step that
currently runs the command "dotnet add package StackExchange.Redis" to pin the
package to version 2.12.14; change the invoked package install command so it
explicitly specifies "StackExchange.Redis" version "2.12.14" to ensure
deterministic CI and match the other pinned Java dependencies.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

cat scripts/bench-graph.sh

Repository: pilotspace/moon

Length of output: 8479

---

Gate success on return code, not just error text matching.

The code captures RC=$? but only checks if $RESULT contains "ERR". A redis-cli exit failure (connection/transport error, timeout, etc.) may produce no output, causing silent failures. Check the exit code first.

This pattern affects multiple phases. Update error checks from if [[ "$RESULT" == *"ERR"* ]] to if [[ $RC -ne 0 || "$RESULT" == *"ERR"* ]] wherever RC is captured, and add RC capture where missing.

Note: The proposed diff's array expansion change ("${CLI[@]}") is incorrect since CLI is a string, not an array; leave $CLI as-is or redefine CLI as an array if array syntax is required.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@scripts/bench-graph.sh` around lines 84 - 92, The script currently captures
the redis-cli exit status into RC but only inspects RESULT for "ERR", which can
miss transport/exit failures; modify each block that runs commands (e.g., the
GRAPH.CREATE invocation using RESULT and RC) to test the exit code first and
fail when RC != 0 by changing the conditional to check if [[ $RC -ne 0 ||
"$RESULT" == *"ERR"* ]], and ensure RC is always assigned immediately after
running the command (e.g., RC=$? right after RESULT=$($CLI ...)); do this
consistently for all phases that capture RC/RESULT, and do not change $CLI to
"${CLI[@]}" unless you convert CLI into an array.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Update degree bins on edge removal too.

Line 232 only decrements edge counts. The src/dst nodes never move from d to d-1, so p50/p99/max stay inflated even after recompute_percentiles(). This hook needs the same degree-transition inputs that on_edge_insert() already takes.

🩹 Proposed fix

-    pub fn on_edge_delete(&mut self, edge_type: u16) {
+    pub fn on_edge_delete(
+        &mut self,
+        edge_type: u16,
+        src_old_degree: u32,
+        dst_old_degree: u32,
+    ) {
         self.total_edges = self.total_edges.saturating_sub(1);
         if let Some(count) = self.edges_by_type.get_mut(&edge_type) {
             *count = count.saturating_sub(1);
             if *count == 0 {
                 self.edges_by_type.remove(&edge_type);
             }
         }
+        self.degree_stats.record_degree_change(
+            src_old_degree,
+            src_old_degree.saturating_sub(1),
+        );
+        self.degree_stats.record_degree_change(
+            dst_old_degree,
+            dst_old_degree.saturating_sub(1),
+        );
         self.maybe_recompute();
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/graph/stats.rs` around lines 231 - 240, on_edge_delete currently only
decrements total_edges and edges_by_type but does not update degree bins, so
percentiles remain stale; modify on_edge_delete to take the same
degree-transition inputs as on_edge_insert (e.g., previous/old degree values or
node ids used by on_edge_insert) and perform the corresponding degree-bin
updates (decrement the bin for degree d and increment for d-1 for both source
and destination nodes using the same helper logic used by on_edge_insert), keep
the existing total_edges and edges_by_type adjustments, and then call
maybe_recompute() so recompute_percentiles()/percentile state is kept in sync.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Guard connection-open accounting with Drop.

record_connection_closed() only runs on the fallthrough path, but this handler still has early returns and a migration return. That leaves the open/close accounting unbalanced on error paths and during shard migration.

Proposed fix

     crate::admin::metrics_setup::record_connection_opened();
+    struct ConnectionMetricGuard;
+    impl Drop for ConnectionMetricGuard {
+        fn drop(&mut self) {
+            crate::admin::metrics_setup::record_connection_closed();
+        }
+    }
+    let _connection_metric_guard = ConnectionMetricGuard;
@@
-    crate::admin::metrics_setup::record_connection_closed();
     (MonoioHandlerResult::Done, None)

Also applies to: 2211-2217, 2286-2286

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 84 - 85, Wrap the open/close
accounting in a RAII guard: after calling record_connection_opened() create a
small struct (e.g., ConnectionOpenGuard) that implements Drop and calls
record_connection_closed(), then instantiate it (let _guard =
ConnectionOpenGuard;) in the handler so the Drop runs for all exit paths (early
returns and migration returns). Modify code paths that currently call
record_connection_closed() explicitly to rely on the guard (remove duplicate
explicit calls) so the close accounting is always balanced regardless of how the
function exits.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

CLIENT KILL does not wake idle monoio handlers.

The kill flag is only polled before entering the next blocking wait. Once this task is parked in stream.read(...) or the subscriber select!, it will not observe the kill until more socket/pubsub activity arrives, so quiet clients with timeout=0 can survive CLIENT KILL indefinitely.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 151 - 154, The handler
currently only checks crate::client_registry::is_killed(client_id) before
entering blocking operations (stream.read(...) and the subscriber select!), so a
CLIENT KILL won't wake tasks already parked; fix by adding a kill-notification
future and include it in the same select/read await so the task can break
immediately: add or use a client_registry subscription API (e.g.,
crate::client_registry::subscribe_kill(client_id) or watch_kill) that returns a
cancel/recv/Notify future, wire that into the existing select! alongside
stream.read(...) and the subscriber branch, and on receiving the kill
notification call the same break/cleanup path; update any code paths using
client_registry::is_killed to rely on the subscribe/notify where the task may be
parked.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Keep oldest_snapshot in sync with allocate_lsn().

Line 157 advances next_lsn, but when there are no active transactions the watermark stays at the old value. That leaves GC/zombie cleanup behind the true head after out-of-transaction graph writes.

🩹 Proposed fix

     pub fn allocate_lsn(&mut self) -> u64 {
         let lsn = self.next_lsn;
         self.next_lsn += 1;
+        if self.active.is_empty() {
+            self.oldest_snapshot = self.next_lsn;
+        }
         lsn
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/vector/mvcc/manager.rs` around lines 152 - 158, allocate_lsn() advances
next_lsn but does not update the GC watermark oldest_snapshot, so
out-of-transaction writes can move the head beyond the oldest snapshot; after
incrementing next_lsn in allocate_lsn(), if there are no active transactions
(check the manager's active transaction set/map), set oldest_snapshot to the new
next_lsn (or otherwise advance oldest_snapshot to the head) so the watermark
stays in sync with next_lsn and prevents GC/zombie cleanup from falling behind;
update the method allocate_lsn(), referencing next_lsn and oldest_snapshot, to
perform this conditional assignment.

[coderabbitai]

Actionable comments posted: 2

♻️ Duplicate comments (8)

src/shard/conn_accept.rs (1)

213-219: ⚠️ Potential issue | 🟠 Major

These maxclients rejection branches still drop the accepted socket silently.

Each branch logs and returns, so TLS/monoio clients never get the advertised -ERR max number of clients reached payload that the plain Tokio path sends. That still breaks the documented rejection contract and makes listener behavior inconsistent.

Also applies to: 516-522, 556-562

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/shard/conn_accept.rs` around lines 213 - 219, The TLS/monoio accept
branches currently just log and return, silently dropping sockets instead of
sending the Redis error; modify the rejection paths where
try_accept_connection(maxclients) is checked (the branches shown around the call
in src/shard/conn_accept.rs and the similar branches at the other occurrences)
to first write the Redis error payload "-ERR max number of clients reached\r\n"
to the newly accepted connection using the appropriate write/send API for that
socket type (use the TLS stream write method for TLS paths and the monoio/tokio
write for monoio paths), flush if needed, and then close the socket/connection
before returning; ensure you handle and ignore write errors so the rejection
still proceeds.

src/server/conn/handler_sharded.rs (2)

1001-1025: ⚠️ Potential issue | 🟠 Major

Registry state can still drift after AUTH/HELLO and subscriber-mode commands.

Both the self-refresh here and the common post-batch refresh still omit current_user / HELLO-set client_name, and the subscriber loop never reaches the post-batch block after PING, UNSUBSCRIBE, or RESET. CLIENT LIST / CLIENT INFO can therefore keep stale user/flag/activity data, and CLIENT KILL USER can miss authenticated clients.

Also applies to: 1779-1788

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_sharded.rs` around lines 1001 - 1025, The client
registry updates in the CLIENT LIST/INFO paths and the subscriber loop only
refresh db/last_cmd_at/flags but omit the HELLO/AUTH-set fields (current_user
and client_name), causing stale info and missed authenticated state; update the
registry::update closures used in these branches (the ones around
client_registry::update called in the CLIENT LIST branch and the INFO branch,
and the equivalent update in the subscriber loop handling
PING/UNSUBSCRIBE/RESET) to also set e.current_user and e.client_name from
conn.current_user and conn.client_name (or the HELLO fields), and refactor the
subscriber loop handlers for PING/UNSUBSCRIBE/RESET to invoke the same
post-batch refresh routine used elsewhere so they don't bypass the shared
refresh logic.

---

513-518: ⚠️ Potential issue | 🟠 Major

This pause gate still makes CLIENT UNPAUSE ineffective and WRITE mode too broad.

The whole batch sleeps before any frame is dispatched, so a batch containing CLIENT UNPAUSE cannot clear the pause until the original timeout has already elapsed. check_pause(true) also turns CLIENT PAUSE WRITE into CLIENT PAUSE ALL for pure-read batches.

Also applies to: 1080-1083

src/server/conn/handler_monoio.rs (4)

151-155: ⚠️ Potential issue | 🟠 Major

CLIENT KILL still won't wake an idle monoio handler.

The flag is polled before the blocking read/subscriber waits, so a quiet connection with timeout=0 can remain parked after being killed. Put a kill notification into the same wait path so idle clients exit promptly.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 151 - 155, The loop currently
polls crate::client_registry::is_killed(client_id) before entering the blocking
read/subscriber wait, which allows an idle handler to stay parked; instead
subscribe to kill notifications and include that receiver in the same
wait/select as the read/subscriber so a kill will wake the task. Replace or
augment the pre-wait is_killed check with a call like
client_registry::subscribe_kill(client_id) (or the existing notifier API) and
select between the socket read/subscriber future and the kill-notification
future inside the loop; on receiving the kill notification, break/clean up
immediately. Ensure you also handle the race by checking the notification once
before/after entering the wait if your subscribe API requires it.

---

84-85: ⚠️ Potential issue | 🔴 Critical

Close accounting is still split between this handler and conn_accept.rs.

src/shard/conn_accept.rs already decrements non-migrated accepted sockets after this future returns, so the fallthrough at Line 2318 double-closes those paths. But migrated targets and the many early return (MonoioHandlerResult::Done, None) exits above rely on Line 2318 and skip it entirely, so other exits leak instead. This still corrupts CONNECTED_CLIENTS / maxclients.

Also applies to: 2318-2318

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 84 - 85, The handler
currently splits connection-close accounting between this module and
conn_accept.rs which causes double-decrements for non-migrated sockets and leaks
for migrated or early-return paths (e.g., returns of MonoioHandlerResult::Done),
corrupting CONNECTED_CLIENTS/maxclients; fix by centralizing responsibility:
remove the decrement/close fallthrough in the handler (the code currently
executed at the end of the handler that decrements CONNECTED_CLIENTS/maxclients)
and ensure the caller try_accept_connection()/conn_accept.rs remains the sole
place to decrement non-migrated accepted sockets, or alternatively add a single
helper (e.g., decrement_connected_clients()) and call it from every early-return
path including migrated-target branches and places returning
MonoioHandlerResult::Done so every exit does exactly one decrement; update
handler_monoio functions to stop double-closing and to call the unified helper
on all exit paths if you choose to keep handler-side accounting.

---

1311-1333: ⚠️ Potential issue | 🟠 Major

Registry state can still drift after AUTH/HELLO and subscriber-mode commands.

This self-refresh/post-batch refresh pair still omits current_user / HELLO-set client_name, and the subscriber loop never reaches the post-batch block after PING, UNSUBSCRIBE, or RESET. CLIENT LIST / CLIENT INFO can therefore keep stale user/flag/activity data, and CLIENT KILL USER can miss authenticated clients.

Also applies to: 2217-2226

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 1311 - 1333, The client
registry updates in the "LIST" and "INFO" branches only refresh
db/flags/last_cmd_at and thus miss current_user and HELLO-set client_name, and
the subscriber-mode path never triggers the post-batch refresh after commands
like PING/UNSUBSCRIBE/RESET; update the client_registry::update call (the
closures used in the LIST/INFO branches and inside the subscriber loop) to also
set e.current_user and e.client_name from conn (the HELLO/auth state) and ensure
the subscriber loop invokes the same post-batch refresh path after handling
PING, UNSUBSCRIBE, RESET so client_registry::client_list / client_info and
CLIENT KILL USER see fresh auth/name/flag/activity info (references:
conn.selected_db, conn.in_multi, conn.subscription_count,
client_registry::update, client_registry::client_list,
client_registry::client_info, and the subscriber loop handling).

---

501-505: ⚠️ Potential issue | 🟠 Major

This pause gate still makes CLIENT UNPAUSE ineffective and WRITE mode too broad.

The batch sleeps before dispatch, so a queued CLIENT UNPAUSE cannot clear the pause until the original timeout has already elapsed. check_pause(true) also makes CLIENT PAUSE WRITE stall pure-read batches.

Also applies to: 1398-1401

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_monoio.rs` around lines 501 - 505, The current
pre-dispatch sleep uses crate::client_pause::check_pause(true) and
monoio::time::sleep(remaining), which both blocks pure-read batches and cannot
be interrupted by a later CLIENT UNPAUSE; change the logic to call
crate::client_pause::check_pause(false) so pure-read batches are not stalled,
and replace the single long monoio::time::sleep(remaining).await with an
interruptible loop that repeatedly calls expire_if_needed() and
check_pause(false) and awaits a short interval (or awaits a pause-notify future
if one exists) so a subsequent CLIENT UNPAUSE can clear the pause immediately
instead of waiting the entire original timeout. Ensure the same fix is applied
to the other occurrence around the same logic.

src/client_registry.rs (1)

184-196: ⚠️ Potential issue | 🟠 Major

CLIENT LIST / CLIENT INFO still serialize incorrect state.

format_client_line() hard-codes sub=0 ... multi=-1, and flags= can only represent one tracked boolean at a time. Even with fresh registry updates, subscribed or transactional clients will still be reported as if they were normal idle clients.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/client_registry.rs` around lines 184 - 196, format_client_line currently
hard-codes many client metrics (sub=0 psub=0 ssub=0 multi=-1 etc.) and uses
flags.to_flag_str which only emits one boolean flag, causing CLIENT LIST/INFO to
report stale/incorrect state; update format_client_line to read actual fields
from ClientEntry (e.g., subscription counts, transactional state/multi, watch
count, qbuf/qbuf_free/argv_mem/tot_mem, net_i/net_o, connected fd if tracked,
etc.) and build a flags string that concatenates all applicable single-letter
flags from the ClientEntry booleans (or change flags.to_flag_str to return a
combined string) so multiple flags appear together; ensure you reference
ClientEntry's fields (connected_at, last_cmd_at, id, addr, name, db, user, and
the subscription/transaction/memory/network/flag fields) when formatting the
output.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/server/conn/handler_sharded.rs`:
- Around line 265-268: The current check using
crate::client_registry::is_killed(client_id) runs only once per outer loop so
idle connections can remain blocked; move this kill-check into the same
tokio::select! that awaits I/O/subscriber events by adding a kill-notification
branch (e.g., subscribe to a kill-notify future or use a
registry.notify_kill(client_id) channel) alongside the existing read/subscribe
arms so that when client_registry signals the client is killed the select wakes,
you break/return immediately; update the code around client_id and the existing
tokio::select! in handler_sharded.rs to remove the standalone is_killed(...)
break and instead handle termination inside the select arm.
- Around line 69-82: The branch that returns after try_accept_connection
currently never decrements the global CONNECTED_CLIENTS, causing leaks; modify
handle_connection_sharded so that any early return (after the "-ERR max number
of clients reached" write or after a failed into_std()/SPSC handoff) explicitly
releases/decrements CONNECTED_CLIENTS, and ensure the successful handoff path
(the code in src/shard/conn_accept.rs that ultimately drives
handle_connection_sharded_inner) also decrements/releases the same counter on
the final disconnect; ideally change try_accept_connection to return a RAII
guard that decrements CONNECTED_CLIENTS on Drop and replace current usage in
handle_connection_sharded, into_std/handoff error paths, and the migrated
handler (handle_connection_sharded_inner / conn_accept.rs) with that guard so
all exit paths automatically release the slot.

---

Duplicate comments:
In `@src/client_registry.rs`:
- Around line 184-196: format_client_line currently hard-codes many client
metrics (sub=0 psub=0 ssub=0 multi=-1 etc.) and uses flags.to_flag_str which
only emits one boolean flag, causing CLIENT LIST/INFO to report stale/incorrect
state; update format_client_line to read actual fields from ClientEntry (e.g.,
subscription counts, transactional state/multi, watch count,
qbuf/qbuf_free/argv_mem/tot_mem, net_i/net_o, connected fd if tracked, etc.) and
build a flags string that concatenates all applicable single-letter flags from
the ClientEntry booleans (or change flags.to_flag_str to return a combined
string) so multiple flags appear together; ensure you reference ClientEntry's
fields (connected_at, last_cmd_at, id, addr, name, db, user, and the
subscription/transaction/memory/network/flag fields) when formatting the output.

In `@src/server/conn/handler_monoio.rs`:
- Around line 151-155: The loop currently polls
crate::client_registry::is_killed(client_id) before entering the blocking
read/subscriber wait, which allows an idle handler to stay parked; instead
subscribe to kill notifications and include that receiver in the same
wait/select as the read/subscriber so a kill will wake the task. Replace or
augment the pre-wait is_killed check with a call like
client_registry::subscribe_kill(client_id) (or the existing notifier API) and
select between the socket read/subscriber future and the kill-notification
future inside the loop; on receiving the kill notification, break/clean up
immediately. Ensure you also handle the race by checking the notification once
before/after entering the wait if your subscribe API requires it.
- Around line 84-85: The handler currently splits connection-close accounting
between this module and conn_accept.rs which causes double-decrements for
non-migrated sockets and leaks for migrated or early-return paths (e.g., returns
of MonoioHandlerResult::Done), corrupting CONNECTED_CLIENTS/maxclients; fix by
centralizing responsibility: remove the decrement/close fallthrough in the
handler (the code currently executed at the end of the handler that decrements
CONNECTED_CLIENTS/maxclients) and ensure the caller
try_accept_connection()/conn_accept.rs remains the sole place to decrement
non-migrated accepted sockets, or alternatively add a single helper (e.g.,
decrement_connected_clients()) and call it from every early-return path
including migrated-target branches and places returning
MonoioHandlerResult::Done so every exit does exactly one decrement; update
handler_monoio functions to stop double-closing and to call the unified helper
on all exit paths if you choose to keep handler-side accounting.
- Around line 1311-1333: The client registry updates in the "LIST" and "INFO"
branches only refresh db/flags/last_cmd_at and thus miss current_user and
HELLO-set client_name, and the subscriber-mode path never triggers the
post-batch refresh after commands like PING/UNSUBSCRIBE/RESET; update the
client_registry::update call (the closures used in the LIST/INFO branches and
inside the subscriber loop) to also set e.current_user and e.client_name from
conn (the HELLO/auth state) and ensure the subscriber loop invokes the same
post-batch refresh path after handling PING, UNSUBSCRIBE, RESET so
client_registry::client_list / client_info and CLIENT KILL USER see fresh
auth/name/flag/activity info (references: conn.selected_db, conn.in_multi,
conn.subscription_count, client_registry::update, client_registry::client_list,
client_registry::client_info, and the subscriber loop handling).
- Around line 501-505: The current pre-dispatch sleep uses
crate::client_pause::check_pause(true) and monoio::time::sleep(remaining), which
both blocks pure-read batches and cannot be interrupted by a later CLIENT
UNPAUSE; change the logic to call crate::client_pause::check_pause(false) so
pure-read batches are not stalled, and replace the single long
monoio::time::sleep(remaining).await with an interruptible loop that repeatedly
calls expire_if_needed() and check_pause(false) and awaits a short interval (or
awaits a pause-notify future if one exists) so a subsequent CLIENT UNPAUSE can
clear the pause immediately instead of waiting the entire original timeout.
Ensure the same fix is applied to the other occurrence around the same logic.

In `@src/server/conn/handler_sharded.rs`:
- Around line 1001-1025: The client registry updates in the CLIENT LIST/INFO
paths and the subscriber loop only refresh db/last_cmd_at/flags but omit the
HELLO/AUTH-set fields (current_user and client_name), causing stale info and
missed authenticated state; update the registry::update closures used in these
branches (the ones around client_registry::update called in the CLIENT LIST
branch and the INFO branch, and the equivalent update in the subscriber loop
handling PING/UNSUBSCRIBE/RESET) to also set e.current_user and e.client_name
from conn.current_user and conn.client_name (or the HELLO fields), and refactor
the subscriber loop handlers for PING/UNSUBSCRIBE/RESET to invoke the same
post-batch refresh routine used elsewhere so they don't bypass the shared
refresh logic.

In `@src/shard/conn_accept.rs`:
- Around line 213-219: The TLS/monoio accept branches currently just log and
return, silently dropping sockets instead of sending the Redis error; modify the
rejection paths where try_accept_connection(maxclients) is checked (the branches
shown around the call in src/shard/conn_accept.rs and the similar branches at
the other occurrences) to first write the Redis error payload "-ERR max number
of clients reached\r\n" to the newly accepted connection using the appropriate
write/send API for that socket type (use the TLS stream write method for TLS
paths and the monoio/tokio write for monoio paths), flush if needed, and then
close the socket/connection before returning; ensure you handle and ignore write
errors so the rejection still proceeds.

🪄 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: d0630b24-a628-4a38-bb70-00991f6662d2

📥 Commits

Reviewing files that changed from the base of the PR and between 27056ec and 4e79568.

📒 Files selected for processing (9)

• .github/workflows/ci.yml
• src/client_pause.rs
• src/client_registry.rs
• src/server/conn/handler_monoio.rs
• src/server/conn/handler_sharded.rs
• src/server/listener.rs
• src/shard/conn_accept.rs
• tests/blocking_list_timeout.rs
• tests/functions_fcall.rs

✅ Files skipped from review due to trivial changes (2)

• tests/functions_fcall.rs
• tests/blocking_list_timeout.rs

🚧 Files skipped from review as they are similar to previous changes (2)

• src/server/listener.rs
• .github/workflows/ci.yml

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

Tokio migration still leaks CONNECTED_CLIENTS.

Once this branch is taken, the source side never records a close. If into_std() or the SPSC handoff fails, the connection is already lost locally and the slot leaks immediately; if the handoff succeeds, the migrated target path in src/shard/conn_accept.rs drives handle_connection_sharded_inner() directly and still never closes the original slot on final disconnect. maxclients will ratchet upward over time.

Also applies to: 100-177

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_sharded.rs` around lines 69 - 82, The branch that
returns after try_accept_connection currently never decrements the global
CONNECTED_CLIENTS, causing leaks; modify handle_connection_sharded so that any
early return (after the "-ERR max number of clients reached" write or after a
failed into_std()/SPSC handoff) explicitly releases/decrements
CONNECTED_CLIENTS, and ensure the successful handoff path (the code in
src/shard/conn_accept.rs that ultimately drives handle_connection_sharded_inner)
also decrements/releases the same counter on the final disconnect; ideally
change try_accept_connection to return a RAII guard that decrements
CONNECTED_CLIENTS on Drop and replace current usage in
handle_connection_sharded, into_std/handoff error paths, and the migrated
handler (handle_connection_sharded_inner / conn_accept.rs) with that guard so
all exit paths automatically release the slot.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

CLIENT KILL only takes effect between blocking waits.

The flag is checked once per outer loop, before the read/subscriber waits below. A quiet connection with timeout=0 can stay parked indefinitely after being killed. Add a kill notification to the same tokio::select! so idle clients wake and exit immediately.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/server/conn/handler_sharded.rs` around lines 265 - 268, The current check
using crate::client_registry::is_killed(client_id) runs only once per outer loop
so idle connections can remain blocked; move this kill-check into the same
tokio::select! that awaits I/O/subscriber events by adding a kill-notification
branch (e.g., subscribe to a kill-notify future or use a
registry.notify_kill(client_id) channel) alongside the existing read/subscribe
arms so that when client_registry signals the client is killed the select wakes,
you break/return immediately; update the code around client_id and the existing
tokio::select! in handler_sharded.rs to remove the standalone is_killed(...)
break and instead handle termination inside the select arm.