refactor: architecture audit — cache bounds, circuit breaker, correctness fixes - #478
Conversation
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (6)
📝 WalkthroughWalkthroughHandshake failures now explicitly disconnect the transport; node-waiter synchronization uses stronger atomic ordering. Retry paths propagate signal-cache flush errors in some resend flows. Persistence saver halts after repeated failures with an observable flag. Message retry keys use a pre-sized Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client
participant Handshake as Handshake
participant Transport as Transport
Client->>Handshake: do_handshake(...).await
alt Handshake succeeds
Handshake-->>Client: Ok(session)
Client->>Transport: mark connected / proceed
else Handshake fails
Handshake-->>Client: Err(e)
Client->>Transport: disconnect().await
Transport-->>Client: disconnected
Client-->>Client: return Err(e.into())
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@claude review |
|
Claude finished @jlucaso1's task in 1m 16s —— View job Code Review: Architecture Audit Fixes (P0-P3)
Overall this is a solid set of hardening fixes. The changes are well-scoped and each addresses a real concern. A few items worth flagging: Bug: Eviction does not actually prioritize negative cache entriesIn // Negative entries first (is_none = true sorts after false with reverse)But To actually prioritize negative entries, you'd need to sort: .collect::<Vec<_>>();
candidates.sort_by_key(|(_, is_none)| std::cmp::Reverse(*is_none));
candidates.into_iter()
.take(...)Severity: Medium — The eviction still works (removes clean entries), it just doesn't prefer negative entries as intended. P0: Signal cache eviction — consider code dedupThe three P0: Background saver circuit breaker — looks goodThe circuit breaker in P1: Transport cleanup on handshake failure — good catch
P1: Flush error propagation in retry.rs — careful with the change
P2: Pre-sized retry cache key
P3: Atomic ordering fix
P3: Clear app_state_key_requests on reconnect
P3: Lock ordering comment in send.rs
Summary: Solid hardening PR. The main actionable issue is the eviction priority bug where |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 310a429d56
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self.flush_signal_cache().await.unwrap_or_else(|e| { | ||
| log::warn!("Failed to flush signal cache after session delete: {e}"); | ||
| }); | ||
| self.flush_signal_cache().await?; |
There was a problem hiding this comment.
In the DM retry path, self.flush_signal_cache().await? now aborts handle_retry_receipt before the resend happens. This function inserts the dedupe key near the top and does not remove it on error, so a temporary backend flush failure can mark the retry as handled while sending nothing, and subsequent retry receipts for the same message are ignored until the dedupe TTL expires. That can leave retry recovery stalled after transient storage errors; either keep this path warn-and-continue or clear dedupe state when this flush fails.
Useful? React with 👍 / 👎.
Bencher Report
310a429 to
6751ada
Compare
jlucaso1
changed the title
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 6751ada298
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| error!( | ||
| "Background saver: {consecutive_failures} consecutive flush failures, \ | ||
| halting to prevent silent data loss. Last error: {e}" | ||
| ); | ||
| return; |
There was a problem hiding this comment.
Returning from the background saver after 10 consecutive save_to_disk() errors permanently disables periodic persistence for the rest of the process lifetime, so a temporary backend outage can turn into ongoing unsaved device state even after storage recovers. Because this loop is the only automatic save path, the return here creates a durability regression compared to the previous behavior that kept retrying.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🤖 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.rs`:
- Around line 1606-1608: The atomic counters node_waiter_count and
sent_node_waiter_count are incremented (fetch_add) before the waiter is actually
pushed into the mutex-protected vectors (node_waiters / sent_node_waiters),
allowing resolve_node_waiters to observe a positive count while the vector is
still empty; to fix, move each fetch_add() so it happens after acquiring the
respective mutex and after the waiter is inserted (i.e. inside wait_for_node()
and wait_for_sent_node(), acquire the lock, push the waiter into
node_waiters/sent_node_waiters, then perform the fetch_add on
node_waiter_count/sent_node_waiter_count), leaving resolve_node_waiters()
unchanged.
In `@src/message.rs`:
- Around line 168-170: The buffer is being reserved from chat.user.len() and
sender.user.len(), which underestimates size because the write! uses full
formatted JIDs (including `@server` and optional :device). Before allocating key,
compute the full string forms for chat and sender (e.g., chat_jid =
chat.to_string() / formatted_jid, sender_jid = sender.to_string() /
formatted_jid), then reserve capacity using chat_jid.len() + msg_id.len() +
sender_jid.len() + 2 (for the two ':' separators); finally write the precomputed
strings into key (write!(key, "{chat_jid}:{msg_id}:{sender_jid}")). This avoids
reallocations when JIDs include domain/device parts.
In `@src/retry.rs`:
- Line 445: handle_retry_receipt() currently uses fallible awaits like
self.flush_signal_cache().await? and returns the error into a detached caller
which just logs it, causing lost retries; change these to explicit
error-handling paths that either (A) roll back any consumed retry state before
returning (undo the dedupe key insertion and re-insert the message into
recent_messages) when flush_signal_cache() or send_message_impl() fail, or (B)
propagate the error to the detached caller by returning a clear enum/error that
the caller in receipt.rs can use to requeue/escalate; update the code paths
around flush_signal_cache(), insert_dedupe_key/remove from recent_messages, and
send_message_impl() in handle_retry_receipt() to perform rollback on failure or
return a requeueable error instead of using `?`.
In `@src/store/persistence_manager.rs`:
- Around line 159-164: The background saver currently just logs and returns when
consecutive_failures >= MAX_CONSECUTIVE_FAILURES, leaving the process running
without persistence; change that branch to surface the tripped circuit-breaker
to the owner by updating a monitored health/fatal state instead of only logging.
Specifically, in persistence_manager.rs where you check consecutive_failures and
call error!(...), set a shared, observable flag or send a one-shot message
(e.g., set an Arc<AtomicBool> like persistence_failed, call a
PersistenceManager::mark_unrecoverable(), or send on a provided mpsc/oneshot
channel) so src/bot.rs (which fire-and-forgets the task) can detect the failure
and take action (alert, reject writes, or terminate); ensure the
PersistenceManager exposes a getter or channel to read this health signal so
modify_device() and the owner can observe the non-durable state.
In `@wacore/src/store/signal_cache.rs`:
- Around line 70-92: The three nearly-identical evict_if_needed implementations
(in SessionStoreState, SenderKeyStoreState, and ByteStoreState) should be
consolidated into a single helper function to remove duplication and fix the
sorting bug: extract the eviction logic into a function (e.g.,
evict_clean_entries) that accepts &mut HashMap<Arc<str>, Option<V>>,
&HashSet<Arc<str>> for dirty, Option<&HashSet<Arc<str>>> for deleted (or
&HashSet when always present), and max_entries, perform candidate collection,
sort to prioritize None (negative) entries first, and remove the excess keys;
then call this helper from each evict_if_needed to replace the duplicated code
paths.
- Around line 77-88: The eviction currently reverses iteration but doesn't
prioritize negative entries; change the logic that builds to_remove from
iterating HashMap into collecting pairs (key, is_none) (e.g., Vec<(Arc<str>,
bool)>), then sort that Vec so entries with is_none == true come first (use
sort_by_key or sort_by comparing the boolean), then take the excess entries
using take(self.cache.len().saturating_sub(max_entries)) and map to the keys;
update references to self.cache, to_remove, max_entries, and the is_none boolean
accordingly so negative (None) entries are evicted before positives.
- Around line 219-229: The current to_remove construction in evict_if_needed
reverses the collected iterator before taking entries, causing the wrong
(most-recent) items to be evicted; fix it by collecting the filtered (k.clone(),
v.is_none()) pairs into a Vec, then take the excess count first and only then
reverse (or map keys) so that the oldest entries are removed. In practice,
change the chain around the collect() on self.cache to perform
.into_iter().take(self.cache.len().saturating_sub(max_entries)).rev().map(|(k,
_)| k).collect() (mirroring the fixes applied in the other two evict_if_needed
implementations) so to_remove contains the oldest keys not in dirty/deleted.
- Around line 138-148: The to_remove computation in signal_cache.rs reverses the
iteration with .rev() which breaks the eviction priority; apply the same change
as in SessionStoreState::evict_if_needed: stop reversing the iterator and
instead take the first N candidates to evict (compute n =
self.cache.len().saturating_sub(max_entries)) from the filtered list of (k,
v.is_none()) entries, then map to k; update the chain that builds to_remove (the
let to_remove: Vec<Arc<str>> = ... expression) to remove .rev() and take the
correct number of entries in forward order so eviction prioritization works as
intended.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 0d5d6919-06ce-4f2e-9fdb-43f81c4d3b50
📒 Files selected for processing (6)
src/client.rssrc/message.rssrc/retry.rssrc/send.rssrc/store/persistence_manager.rswacore/src/store/signal_cache.rs
| if self.node_waiter_count.load(Ordering::Acquire) > 0 { | ||
| self.resolve_node_waiters(&node); | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Verify the publication order for incoming and outgoing node waiters.
nl -ba src/client.rs | sed -n '1603,1608p;3196,3207p;3215,3226p'Repository: jlucaso1/whatsapp-rust
Length of output: 108
🏁 Script executed:
cat -n src/client.rs | sed -n '1603,1610p;3196,3230p'Repository: jlucaso1/whatsapp-rust
Length of output: 1933
Race condition in waiter registration: count published before insertion into protected vector.
Lines 3201 and 3220 increment node_waiter_count and sent_node_waiter_count respectively, before acquiring the mutex to insert the waiter into the vector. A stanza arriving on another thread can observe node_waiter_count > 0 at line 1606 (via Acquire load), lock node_waiters in resolve_node_waiters(), and find the vector still empty since waiters.push() hasn't executed yet (lines 3206, 3225). The Acquire/Release ordering synchronizes only those two atomic operations, not the subsequent mutex-protected insertion.
Move fetch_add() into the mutex-protected section in both wait_for_node() and wait_for_sent_node() to ensure the count is incremented after the waiter is enqueued.
Suggested fix
pub fn wait_for_node(
&self,
filter: NodeFilter,
) -> futures::channel::oneshot::Receiver<Arc<Node>> {
let (tx, rx) = futures::channel::oneshot::channel();
- self.node_waiter_count.fetch_add(1, Ordering::Release);
let mut waiters = self
.node_waiters
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
+ self.node_waiter_count.fetch_add(1, Ordering::Release);
waiters.push(NodeWaiter { filter, tx });
rx
}
pub fn wait_for_sent_node(
&self,
filter: NodeFilter,
) -> futures::channel::oneshot::Receiver<Arc<Node>> {
let (tx, rx) = futures::channel::oneshot::channel();
- self.sent_node_waiter_count.fetch_add(1, Ordering::Release);
let mut waiters = self
.sent_node_waiters
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
+ self.sent_node_waiter_count.fetch_add(1, Ordering::Release);
waiters.push(SentNodeWaiter { filter, tx });
rx
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/client.rs` around lines 1606 - 1608, The atomic counters
node_waiter_count and sent_node_waiter_count are incremented (fetch_add) before
the waiter is actually pushed into the mutex-protected vectors (node_waiters /
sent_node_waiters), allowing resolve_node_waiters to observe a positive count
while the vector is still empty; to fix, move each fetch_add() so it happens
after acquiring the respective mutex and after the waiter is inserted (i.e.
inside wait_for_node() and wait_for_sent_node(), acquire the lock, push the
waiter into node_waiters/sent_node_waiters, then perform the fetch_add on
node_waiter_count/sent_node_waiter_count), leaving resolve_node_waiters()
unchanged.
| let mut key = | ||
| String::with_capacity(chat.user.len() + msg_id.len() + sender.user.len() + 20); | ||
| let _ = write!(key, "{chat}:{msg_id}:{sender}"); |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Reserve space from the full formatted JIDs, not just user.len().
Line 169 sizes the buffer from user.len() plus a fixed 20, but the key writes full JIDs (@server and optional :device). Two @s.whatsapp.net JIDs alone add 30 bytes before the separators, so many 1:1 retry keys still reallocate and lose most of the intended hot-path win.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/message.rs` around lines 168 - 170, The buffer is being reserved from
chat.user.len() and sender.user.len(), which underestimates size because the
write! uses full formatted JIDs (including `@server` and optional :device). Before
allocating key, compute the full string forms for chat and sender (e.g.,
chat_jid = chat.to_string() / formatted_jid, sender_jid = sender.to_string() /
formatted_jid), then reserve capacity using chat_jid.len() + msg_id.len() +
sender_jid.len() + 2 (for the two ':' separators); finally write the precomputed
strings into key (write!(key, "{chat_jid}:{msg_id}:{sender_jid}")). This avoids
reallocations when JIDs include domain/device parts.
| self.flush_signal_cache().await.unwrap_or_else(|e| { | ||
| log::warn!("Failed to flush signal cache after session delete: {e}"); | ||
| }); | ||
| self.flush_signal_cache().await?; |
There was a problem hiding this comment.
These ?s still drop the retry on the floor.
handle_retry_receipt() is only awaited from the detached task in src/receipt.rs (Lines 66-83), and that caller just logs Err. On the DM branch, Line 445 can now return after the dedupe key is inserted and the message is removed from recent_messages, but before send_message_impl() runs, so a transient flush_signal_cache() failure becomes a terminal lost resend instead of triggering recovery. Line 492 has the same detached-caller problem, just after the send has already happened. Either make the caller requeue/escalate these failures, or roll back the consumed retry state before returning them.
Also applies to: 492-492
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/retry.rs` at line 445, handle_retry_receipt() currently uses fallible
awaits like self.flush_signal_cache().await? and returns the error into a
detached caller which just logs it, causing lost retries; change these to
explicit error-handling paths that either (A) roll back any consumed retry state
before returning (undo the dedupe key insertion and re-insert the message into
recent_messages) when flush_signal_cache() or send_message_impl() fail, or (B)
propagate the error to the detached caller by returning a clear enum/error that
the caller in receipt.rs can use to requeue/escalate; update the code paths
around flush_signal_cache(), insert_dedupe_key/remove from recent_messages, and
send_message_impl() in handle_retry_receipt() to perform rollback on failure or
return a requeueable error instead of using `?`.
6751ada to
89735c7
Compare
|
@claude review |
|
Claude finished @jlucaso1's task in 1m 36s —— View job Code Review: Architecture Audit Fixes (P0-P3)
Solid hardening PR. The eviction sorting bug from the first review round is now properly fixed using P0: Signal cache eviction — looks correct nowThe Minor note: The three implementations are near-identical (the only difference is P0: Background saver circuit breaker — good, but consider follow-up
Worth noting for follow-up: After halting, the process continues running with no persistence path. P1: Transport cleanup on handshake failure — correct
P1: Flush error propagation in retry.rs — needs attention
The caller in
The reg-ID mismatch path keeping warn-and-continue (
Severity: Medium — transient persistence failures could cause lost retries. P3: Atomic ordering fix — correct
Note on CodeRabbit's race condition comment: CodeRabbit flagged that P2: Pre-sized retry cache key — fine
P3: Clear stale app_state_key_requests — correct
P3: Lock ordering comment — improved
Summary: Good hardening PR. The eviction bug is properly fixed. The main actionable concern is the flush error propagation in retry.rs (lines 445/492) where |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 89735c7efb
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if self.dirty.contains(k.as_ref()) || self.deleted.contains(k.as_ref()) { | ||
| continue; |
There was a problem hiding this comment.
This eviction logic skips dirty/deleted keys, so during write-heavy periods overflow can remain unresolved because almost every entry is dirty. After flush() clears dirty/deleted sets, no eviction is triggered, which means the cache can stay well above max_entries indefinitely until some later get/put happens. In long-lived processes this defeats the new capacity bound and can retain peak memory long after traffic drops.
Useful? React with 👍 / 👎.
89735c7 to
96681f4
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (4)
src/message.rs (1)
168-170:⚠️ Potential issue | 🟡 MinorPreallocation still underestimates key length.
chat.user.len()/sender.user.len()exclude@serverand optional:device, so this still reallocates frequently in the hot path.Suggested fix
- let mut key = - String::with_capacity(chat.user.len() + msg_id.len() + sender.user.len() + 20); - let _ = write!(key, "{chat}:{msg_id}:{sender}"); + let chat_jid = chat.to_string(); + let sender_jid = sender.to_string(); + let mut key = String::with_capacity(chat_jid.len() + msg_id.len() + sender_jid.len() + 2); + let _ = write!(key, "{chat_jid}:{msg_id}:{sender_jid}"); key🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/message.rs` around lines 168 - 170, The preallocation uses chat.user.len() and sender.user.len(), which omit the "@server" and optional ":device" parts and thus still underestimates capacity; change the capacity calculation for the key (the variable named key used with write!(key, "{chat}:{msg_id}:{sender}")) to use the full slice lengths (e.g. chat.len() and sender.len()) plus msg_id.len() and the number of separator chars (':' occurrences) — e.g. String::with_capacity(chat.len() + msg_id.len() + sender.len() + 2) — so the buffer is sized correctly before calling write!.src/client.rs (1)
1606-1607:⚠️ Potential issue | 🔴 CriticalAcquire load here does not fix waiter publication race.
Line 1606 now uses
Ordering::Acquire, but the root race remains: waiter count is incremented before the waiter is inserted (Line 3201beforeLine 3206). The resolver can observecount > 0and still see no waiter, dropping a matching stanza.Suggested fix
pub fn wait_for_node( &self, filter: NodeFilter, ) -> futures::channel::oneshot::Receiver<Arc<Node>> { let (tx, rx) = futures::channel::oneshot::channel(); - self.node_waiter_count.fetch_add(1, Ordering::Release); let mut waiters = self .node_waiters .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); waiters.push(NodeWaiter { filter, tx }); + self.node_waiter_count.fetch_add(1, Ordering::Release); rx } pub fn wait_for_sent_node( &self, filter: NodeFilter, ) -> futures::channel::oneshot::Receiver<Arc<Node>> { let (tx, rx) = futures::channel::oneshot::channel(); - self.sent_node_waiter_count.fetch_add(1, Ordering::Release); let mut waiters = self .sent_node_waiters .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); waiters.push(SentNodeWaiter { filter, tx }); + self.sent_node_waiter_count.fetch_add(1, Ordering::Release); rx }#!/bin/bash set -euo pipefail # Verify publication order around waiter counters and insertion points. nl -ba src/client.rs | sed -n '1602,1610p;3196,3208p;3215,3227p'🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/client.rs` around lines 1606 - 1607, The resolver sees node_waiter_count > 0 before a waiter is actually published because the registration path increments node_waiter_count (the fetch_add on node_waiter_count in the waiter registration code) before inserting the waiter into the waiter list; fix by publishing the waiter first and then updating the counter (or by making the counter update a Release store after the waiter insertion and keeping the resolver's load as Acquire). Concretely: move the node waiter insertion (the code that pushes/inserts the waiter into the list at the registration site) to occur before calling node_waiter_count.fetch_add, or change that fetch_add to be the Release/paired ordering so resolve_node_waiters' load(Ordering::Acquire) reliably synchronizes with the published waiter; update the registration routine and verify resolve_node_waiters still uses load(Ordering::Acquire).src/retry.rs (1)
445-445:⚠️ Potential issue | 🟠 MajorDon't fail-fast here unless the caller requeues the retry.
By Line 445 the dedupe key is already inserted and the DM payload has been removed from
recent_messages. If the receipt worker still just logshandle_retry_receipt()errors, a transientflush_signal_cache()failure becomes a permanently dropped resend. Either roll back the consumed retry state before returning, or surface a requeueable outcome that the caller actually handles.Verify the caller contract and the absence/presence of rollback paths with a read-only scan:
#!/bin/bash set -euo pipefail echo "== handle_retry_receipt call sites ==" rg -n -C4 '\bhandle_retry_receipt\s*\(' src echo echo "== receipt worker error handling around retry processing ==" rg -n -C6 'handle_retry_receipt|detach\(|spawn\(|warn!\(|error!\(' src/receipt.rs src echo echo "== retry state consumed before the DM flush ==" rg -n -C3 '\bretried_group_messages\b|\btake_recent_message\b|\badd_recent_message\b' src/retry.rs🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/retry.rs` at line 445, The call to self.flush_signal_cache().await? can cause a transient failure after the dedupe key was inserted and the DM payload removed from recent_messages, leading to permanently dropped retries; modify handle_retry_receipt (and the retry path that calls it) to either roll back consumed retry state on flush_signal_cache() failure (e.g., reinsert the DM into recent_messages and remove the dedupe key) or change its return type to surface a requeueable error so the caller can requeue the retry; locate symbols flush_signal_cache(), handle_retry_receipt(), recent_messages, take_recent_message/add_recent_message and the dedupe-key insertion site in retry.rs to implement the rollback or propagate a specific RequeueableError and update callers to handle that outcome.src/store/persistence_manager.rs (1)
157-164:⚠️ Potential issue | 🟠 MajorExpose the tripped saver circuit breaker to the owner.
Returning here only emits a log.
src/bot.rs:650-656still detaches this task, so after the 10th failure the process can keep acceptingmodify_device()writes with no background persistence path and lose them on crash. Please surface an observable fatal/health state or otherwise make writers/the owner react when this loop halts.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/store/persistence_manager.rs` around lines 157 - 164, The background saver currently returns on MAX_CONSECUTIVE_FAILURES without notifying the owner; add an observable circuit-breaker flag to PersistenceManager (e.g., a new field like saver_tripped: Arc<AtomicBool> or an Arc<Notify>/watch::Sender) and set it to true right before the early return in the background flush loop where save_to_disk().await errors are counted (the block that compares consecutive_failures >= MAX_CONSECUTIVE_FAILURES); also provide an accessor or expose the watch/channel so bot.rs (or callers of modify_device) can observe the tripped state and react (e.g., stop accepting writes, mark health unhealthy, or restart the task). Ensure the field is initialized when PersistenceManager is constructed and updated atomically inside the loop where the error is logged.
🤖 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.rs`:
- Around line 1226-1227: The cleanup path redundantly clears
app_state_key_requests via self.app_state_key_requests.lock().await.clear() and
then later replaces it with a new HashMap in cleanup_connection_state(); remove
the earlier clear to avoid duplicate locking and work. Locate the first call to
self.app_state_key_requests.lock().await.clear() in the connection cleanup
sequence and delete that statement, keeping the single reset performed in
cleanup_connection_state() (the assignment that replaces the map with a new
HashMap) so only one lock and reset occurs.
In `@src/retry.rs`:
- Around line 244-247: Wrap the delete+flush sequence that calls
self.signal_cache.delete_session(&signal_address).await and
self.flush_signal_cache().await inside the same per-sender session lock used by
session_lock_for(...) so the delete and subsequent flush are serialized with
other Signal operations; mirror the same change for the other occurrence that
mutates the same session (the block referenced alongside
process_retry_key_bundle()), ensuring you acquire the session_locks entry for
signal_address before calling delete_session and hold it until after
flush_signal_cache completes.
In `@wacore/src/store/signal_cache.rs`:
- Around line 280-287: The cache can remain oversized after flush() clears
dirty/deleted flags because eviction is only run on put/update; to fix, invoke
state.evict_if_needed(self.max_entries) after any flush() that may make entries
evictable so the size bound is enforced immediately. Locate the flush()
implementations (the ones manipulating dirty/deleted flags and calling
sessions.lock()/state.flush()) and add a call to
state.evict_if_needed(self.max_entries) after the flush completes (and while
still holding the same mutex), similarly update any other flush paths referenced
(the other flush occurrences around the same logic) so they all re-run
evict_if_needed using the same max_entries field. Ensure you reference the
existing methods flush(), evict_if_needed(), put_session(), and the
self.max_entries field when making the changes.
---
Duplicate comments:
In `@src/client.rs`:
- Around line 1606-1607: The resolver sees node_waiter_count > 0 before a waiter
is actually published because the registration path increments node_waiter_count
(the fetch_add on node_waiter_count in the waiter registration code) before
inserting the waiter into the waiter list; fix by publishing the waiter first
and then updating the counter (or by making the counter update a Release store
after the waiter insertion and keeping the resolver's load as Acquire).
Concretely: move the node waiter insertion (the code that pushes/inserts the
waiter into the list at the registration site) to occur before calling
node_waiter_count.fetch_add, or change that fetch_add to be the Release/paired
ordering so resolve_node_waiters' load(Ordering::Acquire) reliably synchronizes
with the published waiter; update the registration routine and verify
resolve_node_waiters still uses load(Ordering::Acquire).
In `@src/message.rs`:
- Around line 168-170: The preallocation uses chat.user.len() and
sender.user.len(), which omit the "@server" and optional ":device" parts and
thus still underestimates capacity; change the capacity calculation for the key
(the variable named key used with write!(key, "{chat}:{msg_id}:{sender}")) to
use the full slice lengths (e.g. chat.len() and sender.len()) plus msg_id.len()
and the number of separator chars (':' occurrences) — e.g.
String::with_capacity(chat.len() + msg_id.len() + sender.len() + 2) — so the
buffer is sized correctly before calling write!.
In `@src/retry.rs`:
- Line 445: The call to self.flush_signal_cache().await? can cause a transient
failure after the dedupe key was inserted and the DM payload removed from
recent_messages, leading to permanently dropped retries; modify
handle_retry_receipt (and the retry path that calls it) to either roll back
consumed retry state on flush_signal_cache() failure (e.g., reinsert the DM into
recent_messages and remove the dedupe key) or change its return type to surface
a requeueable error so the caller can requeue the retry; locate symbols
flush_signal_cache(), handle_retry_receipt(), recent_messages,
take_recent_message/add_recent_message and the dedupe-key insertion site in
retry.rs to implement the rollback or propagate a specific RequeueableError and
update callers to handle that outcome.
In `@src/store/persistence_manager.rs`:
- Around line 157-164: The background saver currently returns on
MAX_CONSECUTIVE_FAILURES without notifying the owner; add an observable
circuit-breaker flag to PersistenceManager (e.g., a new field like
saver_tripped: Arc<AtomicBool> or an Arc<Notify>/watch::Sender) and set it to
true right before the early return in the background flush loop where
save_to_disk().await errors are counted (the block that compares
consecutive_failures >= MAX_CONSECUTIVE_FAILURES); also provide an accessor or
expose the watch/channel so bot.rs (or callers of modify_device) can observe the
tripped state and react (e.g., stop accepting writes, mark health unhealthy, or
restart the task). Ensure the field is initialized when PersistenceManager is
constructed and updated atomically inside the loop where the error is logged.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 97ae0430-e7ef-4cef-96cb-9a52dd42c252
📒 Files selected for processing (6)
src/client.rssrc/message.rssrc/retry.rssrc/send.rssrc/store/persistence_manager.rswacore/src/store/signal_cache.rs
| // Clear stale app state requests from previous connection | ||
| self.app_state_key_requests.lock().await.clear(); |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Remove redundant app-state map clearing in cleanup path.
Line 1227 clears app_state_key_requests, but Line 1270 immediately replaces the same map with a new HashMap. Keep one reset path to avoid duplicate lock/work in cleanup_connection_state().
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/client.rs` around lines 1226 - 1227, The cleanup path redundantly clears
app_state_key_requests via self.app_state_key_requests.lock().await.clear() and
then later replaces it with a new HashMap in cleanup_connection_state(); remove
the earlier clear to avoid duplicate locking and work. Locate the first call to
self.app_state_key_requests.lock().await.clear() in the connection cleanup
sequence and delete that statement, keeping the single reset performed in
cleanup_connection_state() (the assignment that replaces the map with a new
HashMap) so only one lock and reset occurs.
| self.signal_cache.delete_session(&signal_address).await; | ||
| self.flush_signal_cache().await.unwrap_or_else(|e| { | ||
| if let Err(e) = self.flush_signal_cache().await { | ||
| log::warn!("Failed to flush session deletion for reg ID mismatch: {e}"); | ||
| }); | ||
| } |
There was a problem hiding this comment.
Take the per-sender session lock around both delete+flush paths.
Both branches mutate signal_address's Signal session outside session_lock_for(...), so concurrent encrypt/decrypt work for the same sender can race the delete and repopulate stale state. process_retry_key_bundle() already uses that lock later in this file; these paths need the same guard.
As per coding guidelines, "Use session_locks to serialize per-sender Signal encrypt/decrypt operations and message_enqueue_locks to serialize per-chat incoming message processing".
Also applies to: 442-445
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/retry.rs` around lines 244 - 247, Wrap the delete+flush sequence that
calls self.signal_cache.delete_session(&signal_address).await and
self.flush_signal_cache().await inside the same per-sender session lock used by
session_lock_for(...) so the delete and subsequent flush are serialized with
other Signal operations; mirror the same change for the other occurrence that
mutates the same session (the block referenced alongside
process_retry_key_bundle()), ensuring you acquire the session_locks entry for
signal_address before calling delete_session and hold it until after
flush_signal_cache completes.
96681f4 to
8d83174
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (2)
src/client.rs (1)
1606-1608:⚠️ Potential issue | 🔴 Critical
Acquirehere only orders the counter, not the waiter publication.
src/client.rs::wait_for_node()still incrementsnode_waiter_countbefore it acquires the mutex and pushes intonode_waiters(andsrc/client.rs::wait_for_sent_node()has the same pattern). That meanssrc/client.rs::resolve_node_waiters()can observe> 0, lock the vector, and still find it empty, which loses the stanza for a just-registered waiter. Move thefetch_add()under the same mutex hold as thepush()so the count is never published independently of the vector update.Suggested fix
pub fn wait_for_node( &self, filter: NodeFilter, ) -> futures::channel::oneshot::Receiver<Arc<Node>> { let (tx, rx) = futures::channel::oneshot::channel(); - self.node_waiter_count.fetch_add(1, Ordering::Release); let mut waiters = self .node_waiters .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); + self.node_waiter_count.fetch_add(1, Ordering::Release); waiters.push(NodeWaiter { filter, tx }); rx } pub fn wait_for_sent_node( &self, filter: NodeFilter, ) -> futures::channel::oneshot::Receiver<Arc<Node>> { let (tx, rx) = futures::channel::oneshot::channel(); - self.sent_node_waiter_count.fetch_add(1, Ordering::Release); let mut waiters = self .sent_node_waiters .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); + self.sent_node_waiter_count.fetch_add(1, Ordering::Release); waiters.push(SentNodeWaiter { filter, tx }); rx }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/client.rs` around lines 1606 - 1608, The counter node_waiter_count is incremented (fetch_add) before the waiter is actually pushed into the node_waiters vector, allowing resolve_node_waiters to see a positive count but find an empty vector; change wait_for_node() (and wait_for_sent_node()) to acquire the same mutex that guards node_waiters before performing the push and then increment node_waiter_count while still holding that mutex so the publication of the count cannot race ahead of the vector update; ensure resolve_node_waiters still reads the count with Ordering::Acquire and locks the same mutex around inspecting/popping node_waiters.src/store/persistence_manager.rs (1)
157-166:⚠️ Potential issue | 🟠 MajorSurface the tripped saver breaker instead of just returning.
After this branch returns, the detached background saver is gone, but
src/store/persistence_manager.rs::modify_device()still accepts writes, setsdirty = true, and notifies an event with no listener behind it. That silently degrades durability until an explicitflush()happens. Expose a health/fatal flag or similar monitored signal so the owner can alert or reject writes once persistence is no longer running.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/store/persistence_manager.rs` around lines 157 - 166, The background saver currently returns on repeated save_to_disk() errors but leaves persistence running and modify_device() still accepting writes (setting dirty = true and notifying events), so add a monitored fatal/health flag on the PersistenceManager (e.g., a AtomicBool or an enum field like fatal_error) and set it when consecutive_failures >= MAX_CONSECUTIVE_FAILURES inside the save loop (where save_to_disk() errors are handled). Update modify_device() to check that flag and either reject new writes (return an error) or route them to a safe fallback, and ensure any event notifications include the health state so the owner can observe the broken saver; also expose a public is_healthy()/has_fatal_error() accessor and ensure flush() and any callers respect that flag.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@wacore/src/store/signal_cache.rs`:
- Around line 284-287: delete_session, delete_identity, and delete_sender_key
currently only insert a negative entry and do not run the eviction hook, leaving
evictable entries in memory until the next flush; update each of these delete_*
methods to call the same eviction routine used by put_* and flush (i.e., after
acquiring the lock and inserting the tombstone/negative entry call
state.evict_if_needed(self.max_entries)) so the cache cap is enforced
immediately; reference the existing put_session pattern and the
state.evict_if_needed(self.max_entries) call as the model to follow.
---
Duplicate comments:
In `@src/client.rs`:
- Around line 1606-1608: The counter node_waiter_count is incremented
(fetch_add) before the waiter is actually pushed into the node_waiters vector,
allowing resolve_node_waiters to see a positive count but find an empty vector;
change wait_for_node() (and wait_for_sent_node()) to acquire the same mutex that
guards node_waiters before performing the push and then increment
node_waiter_count while still holding that mutex so the publication of the count
cannot race ahead of the vector update; ensure resolve_node_waiters still reads
the count with Ordering::Acquire and locks the same mutex around
inspecting/popping node_waiters.
In `@src/store/persistence_manager.rs`:
- Around line 157-166: The background saver currently returns on repeated
save_to_disk() errors but leaves persistence running and modify_device() still
accepting writes (setting dirty = true and notifying events), so add a monitored
fatal/health flag on the PersistenceManager (e.g., a AtomicBool or an enum field
like fatal_error) and set it when consecutive_failures >=
MAX_CONSECUTIVE_FAILURES inside the save loop (where save_to_disk() errors are
handled). Update modify_device() to check that flag and either reject new writes
(return an error) or route them to a safe fallback, and ensure any event
notifications include the health state so the owner can observe the broken
saver; also expose a public is_healthy()/has_fatal_error() accessor and ensure
flush() and any callers respect that flag.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 0c9deac6-895c-4d7b-831d-e5f488072ba7
📒 Files selected for processing (6)
src/client.rssrc/message.rssrc/retry.rssrc/send.rssrc/store/persistence_manager.rswacore/src/store/signal_cache.rs
8d83174 to
c036184
Compare
- Signal cache: add capacity bound (10K entries per store), evict clean/negative entries when exceeded (runtime-agnostic, no moka) - Background saver: circuit breaker after 10 consecutive flush failures - Transport: explicit disconnect on handshake failure - Retry: propagate flush_signal_cache errors instead of swallowing - Cache key: pre-sized String builder for retry cache key hot path - Reconnect: clear stale app_state_key_requests - Atomics: Relaxed→Acquire for node_waiter_count - Docs: lock ordering invariant in send.rs
c036184 to
4215867
Compare
There was a problem hiding this comment.
♻️ Duplicate comments (4)
src/retry.rs (2)
245-247:⚠️ Potential issue | 🟠 MajorTake the per-sender session lock around both delete+flush sequences.
Both branches mutate
signal_addressoutsidesession_lock_for(...), so concurrent encrypt/decrypt work can race the deletion and repopulate stale session state.process_retry_key_bundle()already serializes the same address.🔒 Suggested pattern
+let session_mutex = self.session_lock_for(signal_address.as_str()).await; +let _session_guard = session_mutex.lock().await; self.signal_cache.delete_session(&signal_address).await; self.flush_signal_cache().await?;As per coding guidelines, "Use
session_locksto serialize per-sender Signal encrypt/decrypt operations andmessage_enqueue_locksto serialize per-chat incoming message processing".Also applies to: 445-445
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/retry.rs` around lines 245 - 247, The deletion+flush sequence around signal_address must be executed while holding the per-sender session lock to prevent races with concurrent encrypt/decrypt; wrap the existing delete and the subsequent call to flush_signal_cache() inside the same session_lock_for(...) used elsewhere (e.g., where process_retry_key_bundle() serializes that address) so both branches acquire the session lock before mutating signal_address and only release it after the delete and flush complete, ensuring you call session_lock_for(signal_address).await (or the equivalent lock guard) around the delete and flush calls.
445-445:⚠️ Potential issue | 🟠 MajorDon't propagate this DM flush error without restoring retry state.
By Line 445 the dedupe entry is already inserted and the DM
recent_messagehas already been consumed. Ifflush_signal_cache()fails here, the resend never reachessend_message_impl(), but later retry receipts are skipped because the message is already marked handled. Either move the dedupe/take step after this flush succeeds, or roll both structures back on error.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/retry.rs` at line 445, The DM dedupe entry and the DM's recent_message are being consumed before calling flush_signal_cache(), so if flush_signal_cache() fails the retry state remains mutated and future retries are skipped; update the logic around where the dedupe entry is inserted and recent_message is taken (the code that prepares the resend before calling send_message_impl()) so that either (a) you move the dedupe insertion and recent_message take to after flush_signal_cache() completes successfully, or (b) on any error returned from flush_signal_cache() you roll back both the dedupe map entry and restore the DM recent_message to its prior state before returning the error. Ensure you reference and adjust the code paths that call flush_signal_cache(), the dedupe insertion site, and the recent_message consumption so state is consistent on error.wacore/src/store/signal_cache.rs (1)
284-287:⚠️ Potential issue | 🟡 MinorApply the same eviction hook to the
delete_*paths.These new hooks enforce the cap on read/write/flush, but an uncached
delete_session,delete_identity, ordelete_sender_keystill leaves the old clean working set resident until some later access or flush. That makes the new bound inconsistent on delete-heavy paths.♻️ Proposed fix
pub async fn delete_session(&self, address: &ProtocolAddress) { - self.sessions.lock().await.delete(address.as_str()); + let mut state = self.sessions.lock().await; + state.delete(address.as_str()); + state.evict_if_needed(self.max_entries); } pub async fn delete_identity(&self, address: &ProtocolAddress) { - self.identities.lock().await.delete(address.as_str()); + let mut state = self.identities.lock().await; + state.delete(address.as_str()); + state.evict_if_needed(self.max_entries); } pub async fn delete_sender_key(&self, cache_key: &str) { - self.sender_keys.lock().await.delete(cache_key); + let mut state = self.sender_keys.lock().await; + state.delete(cache_key); + state.evict_if_needed(self.max_entries); }Also applies to: 333-336, 364-367
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/store/signal_cache.rs` around lines 284 - 287, The delete paths currently remove entries but don't call the eviction hook, leaving the in-memory working set larger than the enforced cap; update the delete methods to mirror put_session by calling state.evict_if_needed(self.max_entries) after performing the deletion. Specifically, in the methods delete_session, delete_identity, and delete_sender_key (the same locked state used in put_session), invoke state.evict_if_needed(self.max_entries) immediately after state.delete(...) while holding the lock so the cap is consistently enforced on delete-heavy paths.src/client.rs (1)
1604-1605:⚠️ Potential issue | 🔴 Critical
Acquirehere still does not fix the waiter-registration race.The counters are still published before the waiter is enqueued, so
process_node()/send_node()can observe> 0, take the mutex, and miss the just-registered waiter for the very stanza being awaited. Move thefetch_add()calls under the same mutex and beforepush(...); strengthening the load alone does not close that gap.Suggested fix
pub fn wait_for_node( &self, filter: NodeFilter, ) -> futures::channel::oneshot::Receiver<Arc<Node>> { let (tx, rx) = futures::channel::oneshot::channel(); - self.node_waiter_count.fetch_add(1, Ordering::Release); let mut waiters = self .node_waiters .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); + self.node_waiter_count.fetch_add(1, Ordering::Release); waiters.push(NodeWaiter { filter, tx }); rx } pub fn wait_for_sent_node( &self, filter: NodeFilter, ) -> futures::channel::oneshot::Receiver<Arc<Node>> { let (tx, rx) = futures::channel::oneshot::channel(); - self.sent_node_waiter_count.fetch_add(1, Ordering::Release); let mut waiters = self .sent_node_waiters .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); + self.sent_node_waiter_count.fetch_add(1, Ordering::Release); waiters.push(SentNodeWaiter { filter, tx }); rx }Also applies to: 3198-3204, 3217-3223
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/client.rs` around lines 1604 - 1605, The atomic counter increments (node_waiter_count.fetch_add) are happening before the waiter is actually enqueued, so a racing path in process_node()/send_node() can see the counter > 0, take the mutex and miss the newly-registered waiter; fix by moving the fetch_add() calls to occur while holding the same mutex used to protect the waiter queue and place them immediately before the push(...) that enqueues the waiter (so the counter is published only when the waiter is guaranteed enqueued), then update all similar sites (including the other occurrences around resolve_node_waiters, process_node, send_node) to use the same pattern.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@src/client.rs`:
- Around line 1604-1605: The atomic counter increments
(node_waiter_count.fetch_add) are happening before the waiter is actually
enqueued, so a racing path in process_node()/send_node() can see the counter >
0, take the mutex and miss the newly-registered waiter; fix by moving the
fetch_add() calls to occur while holding the same mutex used to protect the
waiter queue and place them immediately before the push(...) that enqueues the
waiter (so the counter is published only when the waiter is guaranteed
enqueued), then update all similar sites (including the other occurrences around
resolve_node_waiters, process_node, send_node) to use the same pattern.
In `@src/retry.rs`:
- Around line 245-247: The deletion+flush sequence around signal_address must be
executed while holding the per-sender session lock to prevent races with
concurrent encrypt/decrypt; wrap the existing delete and the subsequent call to
flush_signal_cache() inside the same session_lock_for(...) used elsewhere (e.g.,
where process_retry_key_bundle() serializes that address) so both branches
acquire the session lock before mutating signal_address and only release it
after the delete and flush complete, ensuring you call
session_lock_for(signal_address).await (or the equivalent lock guard) around the
delete and flush calls.
- Line 445: The DM dedupe entry and the DM's recent_message are being consumed
before calling flush_signal_cache(), so if flush_signal_cache() fails the retry
state remains mutated and future retries are skipped; update the logic around
where the dedupe entry is inserted and recent_message is taken (the code that
prepares the resend before calling send_message_impl()) so that either (a) you
move the dedupe insertion and recent_message take to after flush_signal_cache()
completes successfully, or (b) on any error returned from flush_signal_cache()
you roll back both the dedupe map entry and restore the DM recent_message to its
prior state before returning the error. Ensure you reference and adjust the code
paths that call flush_signal_cache(), the dedupe insertion site, and the
recent_message consumption so state is consistent on error.
In `@wacore/src/store/signal_cache.rs`:
- Around line 284-287: The delete paths currently remove entries but don't call
the eviction hook, leaving the in-memory working set larger than the enforced
cap; update the delete methods to mirror put_session by calling
state.evict_if_needed(self.max_entries) after performing the deletion.
Specifically, in the methods delete_session, delete_identity, and
delete_sender_key (the same locked state used in put_session), invoke
state.evict_if_needed(self.max_entries) immediately after state.delete(...)
while holding the lock so the cap is consistently enforced on delete-heavy
paths.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 1ac5beff-29c0-4d87-bb88-13049a7264b9
📒 Files selected for processing (6)
src/client.rssrc/message.rssrc/retry.rssrc/send.rssrc/store/persistence_manager.rswacore/src/store/signal_cache.rs
1 participant
Summary
Addresses findings from a senior-level architecture audit of concurrency, error handling, memory, and correctness patterns.
Changes
P0: Signal cache capacity bound (
wacore/src/store/signal_cache.rs)max_entries(default 10K per store) toSignalStoreCachewith_max_entries()constructor for custom limitsevict_if_needed()uses O(n) two-vec partition (negative entries evicted first)P0: Background saver circuit breaker (
src/store/persistence_manager.rs)P1: Transport cleanup on handshake failure (
src/client.rs)transport.disconnect()whendo_handshake()failsP1: Propagate flush errors in retry path (
src/retry.rs)?instead ofunwrap_or_else(warn)on critical pathsP2: Pre-sized retry cache key (
src/message.rs)P3: Clear stale
app_state_key_requestson reconnect (src/client.rs)P3:
Relaxed→Acquirefornode_waiter_count(src/client.rs)P3: Document lock ordering invariant (
src/send.rs)Evaluated and deferred
HttpRequest.body: Vec<u8>. Needs new HttpClient trait method.Bytes::slice()zero-copy.Test plan
-D warnings)Summary by CodeRabbit
Bug Fixes
Performance & Reliability
New