refactor: runtime agnostic - #393
Conversation
|
@claude review |
|
Caution Review failedPull request was closed or merged during review Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughIntroduce a runtime abstraction (wacore::runtime::Runtime) with a Tokio-backed implementation and thread it through Bot, Client, NoiseSocket, persistence and blocking helpers; replace many Tokio primitives with runtime + futures/async-channel/async-lock/event-listener equivalents; add multiple wacore modules (time, session, appstate_sync, message_processing, store caches, media/retry, etc.). Changes
Sequence Diagram(s)sequenceDiagram
participant Builder as BotBuilder
participant Bot
participant Runtime as wacore::Runtime
participant Client
participant Backend
Builder->>Bot: build(with runtime)
Bot->>Runtime: spawn(main client loop)
Runtime->>Client: run client task
Client->>Runtime: spawn background tasks / blocking(f)
Runtime->>Backend: perform blocking I/O/DB
Backend-->>Client: return result
Client-->>Runtime: sleep/timeout via runtime
Bot->>Bot: BotHandle.abort() -> Runtime.abort()
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Poem
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
|
|
Claude finished @jlucaso1's task in 1m 53s —— View job PR Review: refactor: runtime agnostic
Overall this is a well-structured refactor that cleanly separates runtime-agnostic logic into Here are the issues I found: Duplicated code not consolidated1.
2. These functions exist in both 3. Keepalive constants duplicated
Incomplete runtime decoupling4. Tokio-specific imports remain in The main client still uses Tokio-specific types:
These should be replaced with runtime-agnostic alternatives (e.g., Two diverging
|
Bencher Report![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
src/client/sessions.rs (1)
29-45:⚠️ Potential issue | 🟠 MajorDon't publish offline-sync completion before the semaphore upgrade.
offline_sync_completedis flipped before the 64-permit semaphore is installed, and the flag is only usingRelaxedordering. The wait paths in this file short-circuit on that flag, so they can proceed while the old 1-permitArcis still visible, and any task that already cloned thatArcstays serialized. Please make the wider semaphore visible before publishing completion, and use release/acquire semantics if this flag is the publication boundary.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/client/sessions.rs` around lines 29 - 45, Move the installation of the 64-permit semaphore to occur before you set offline_sync_completed and make the publish atomic with proper memory ordering: first acquire the lock on message_processing_semaphore and replace the Arc with the new async_lock::Semaphore::new(64), then perform the compare_exchange that flips offline_sync_completed using Release on success (and Acquire for any loads that read it elsewhere); only after a successful flip call offline_sync_notifier.notify(usize::MAX). Ensure references to offline_sync_completed, message_processing_semaphore, compare_exchange, and offline_sync_notifier are updated so the wider semaphore is visible before you publish completion.src/request.rs (1)
159-178:⚠️ Potential issue | 🟠 MajorMove
shutdown_notifier.listen()before theis_runningcheck to prevent missed shutdown notifications.Line 161 registers the shutdown listener only after the
is_runningcheck andsend_node(). Sinceevent_listener::Eventhas snapshot semantics—notifications are lost if no active listeners are present—a disconnect that occurs between theis_runningcheck andlisten()registration will miss the shutdown notification. The IQ will then wait for the full timeout instead of returningNotConnectedpromptly.Minimal fix
pub async fn send_iq(&self, query: InfoQuery<'_>) -> Result<Node, IqError> { + let shutdown = self.shutdown_notifier.listen(); // Fail fast if the client is shutting down if !self.is_running.load(Ordering::Relaxed) { return Err(IqError::NotConnected); } let req_id = query .id .clone() @@ -158,8 +159,6 @@ impl Client { // Race the IQ response against shutdown so we fail fast on disconnect // instead of waiting the full timeout. - let shutdown = self.shutdown_notifier.listen(); let iq_timeout = query.timeout.unwrap_or(default_timeout); futures::select! {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/request.rs` around lines 159 - 178, The shutdown listener is registered too late (shutdown_notifier.listen() is called after the is_running check and send_node()), risking missed disconnect notifications; move the call to shutdown_notifier.listen() to occur before the is_running check and before calling send_node() so the Event listener is active during the window where a shutdown may occur (update the code around the is_running check and send_node() usage in request.rs to create let shutdown = self.shutdown_notifier.listen(); early, then perform the is_running check/send_node(), and leave the existing futures::select! logic unchanged).src/socket/noise_socket.rs (1)
102-155:⚠️ Potential issue | 🔴 CriticalAdvance the Noise counter only after a successful send.
process_send_job()commitswrite_counterbefore the new blocking encryption path runs. If encryption/framing fails there, this socket consumes a nonce without ever producing a frame, and the next successful send uses a counter the peer never saw. That desynchronizes the session immediately.🛠️ Proposed fix
) -> SendResult { let counter = *write_counter; - *write_counter = write_counter.wrapping_add(1); // For small messages, encrypt plaintext_buf in-place then frame into out_buf. // This avoids the previous triple-copy pattern (plaintext→out→plaintext→out). if plaintext_buf.len() <= INLINE_ENCRYPT_THRESHOLD { if let Err(e) = write_key.encrypt_in_place_with_counter(counter, &mut plaintext_buf) { @@ if let Err(e) = transport.send(out_buf).await { return Err(EncryptSendError::transport(e)); } + *write_counter = counter.wrapping_add(1); Ok(()) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/socket/noise_socket.rs` around lines 102 - 155, The code currently increments *write_counter immediately (via counter = *write_counter; *write_counter = write_counter.wrapping_add(1)) which advances the nonce even if encryption/framing or transport.send fails; change this so the counter is reserved locally but only commit the increment after transport.send succeeds: compute let counter = *write_counter; let next = counter.wrapping_add(1); use counter for encrypt_in_place_with_counter / encrypt_with_counter and for framing, and only after if let Err(e) = transport.send(out_buf).await { ... } else { *write_counter = next; } — ensure both the INLINE_ENCRYPT_THRESHOLD (small-path using encrypt_in_place_with_counter) and the blocking-path (encrypt_with_counter + framing) follow this pattern so failed encrypt/framing/transport never advance *write_counter.
🧹 Nitpick comments (4)
wacore/src/runtime.rs (1)
103-113: Panic propagation may cause unexpected behavior.If the blocking closure
fpanics,txis dropped without sending, causingrx.awaitto returnErr(Canceled), which then panics at.expect(). This propagates the panic to the async caller in a potentially unexpected way.Consider returning a
Resultor usingunwrap_or_elsewith an explicit panic message indicating the blocking task failed:♻️ Suggested improvement for panic handling
pub async fn blocking<T: Send + 'static>( rt: &dyn Runtime, f: impl FnOnce() -> T + Send + 'static, -) -> T { +) -> Result<T, BlockingTaskPanicked> { let (tx, rx) = futures::channel::oneshot::channel(); rt.spawn_blocking(Box::new(move || { let _ = tx.send(f()); })) .await; - rx.await.expect("spawn_blocking task completed") + rx.await.map_err(|_| BlockingTaskPanicked) } + +/// Error returned when a blocking task panics. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +#[error("blocking task panicked")] +pub struct BlockingTaskPanicked;Alternatively, if panic propagation is intentional, consider documenting this behavior.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/runtime.rs` around lines 103 - 113, The blocking helper currently panics if the spawned blocking closure `f` panics because `tx` is dropped and `rx.await.expect()` panics; update `blocking` to explicitly propagate panic/errors by wrapping the closure with std::panic::catch_unwind and sending a Result over `tx` (e.g., send Ok(value) on success or Err(panic_payload) on unwind), change the receive side (`rx.await`) to return a Result (or change the function signature to return Result<T, E>) and map the Canceled/Err case to a clear error variant or message; locate the `blocking` function and the use of `rt.spawn_blocking`, `tx`, and `rx` to implement catch_unwind inside the spawned closure and handle the received Result instead of calling `.expect()`.src/socket/error.rs (1)
25-26: Rename Join error text to runtime-neutral wording.
join(...)is now runtime-agnostic, but the kind message still says “tokio join error”, which is misleading in logs.Proposed wording update
pub enum EncryptSendErrorKind { @@ - #[error("tokio join error")] + #[error("task join error")] Join,Also applies to: 61-64
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/socket/error.rs` around lines 25 - 26, The error variant Join in src/socket/error.rs uses the message "tokio join error" which is runtime-specific; update its Display/error string to a runtime-neutral phrase like "task join error" (and apply the same change to the analogous variant mentioned at lines 61-64). Locate the enum variant named Join and replace the literal "tokio join error" with a neutral message (e.g., "task join error" or "join error") so logs no longer reference Tokio specifically while preserving the variant name and semantics.src/handlers/message.rs (1)
48-54: PreferClient::chat_lockshere instead of a second per-chat lock map.Using
message_enqueue_lockscreates another ordering domain for the same chat, so code paths that follow the repo-standardchat_locksdiscipline can still interleave with this queue. ReusingClient::chat_lockskeeps the per-chat serialization boundary consistent.As per coding guidelines, "Use
Client::chat_locksto serialize per-chat operations for concurrency safety".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/handlers/message.rs` around lines 48 - 54, Replace the separate per-chat lock map usage with the repository-standard Client::chat_locks to ensure a single serialization domain: instead of calling client.message_enqueue_locks.get_with_by_ref(&chat_id, ...) and using enqueue_mutex/ _enqueue_guard, obtain the mutex from client.chat_locks (using the same get_with_by_ref and async_lock::Mutex type), await the returned mutex.lock().await, and use the guard to serialize the enqueue operation for chat_id (preserving the await and scope of the guard).src/bot.rs (1)
650-655: Avoidunwrap()in this non-test builder path.Typestate already proves these fields are present, so destructuring
self(or matching theSome(...)cases) keeps the same guarantee without a panic path. As per coding guidelines: "Do not use.unwrap()outside of test code".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/bot.rs` around lines 650 - 655, The build method uses .unwrap() on runtime, backend, transport_factory, and http_client; replace these panicking calls by destructuring self to extract the Some(...) values (e.g. match or let Self { runtime: Some(runtime), backend: Some(backend), transport_factory: Some(transport_factory), http_client: Some(http_client), .. } = self) so the typestate guarantee is preserved without any unwraps and without introducing a panic path in Bot::build.
🤖 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/bot.rs`:
- Around line 115-127: BotHandle::abort() is currently a no-op; implement it to
call the underlying abort handle so callers can explicitly cancel the run task
instead of relying on Drop. In the abort(&self) method, invoke the AbortHandle's
abort method on the _abort_handle field (e.g. self._abort_handle.abort()) so the
task is cancelled immediately; keep the existing Drop behavior intact. Also
apply the same change to the other BotHandle implementation variant referenced
around lines 242-251 so both handle types provide explicit abort semantics.
In `@src/client.rs`:
- Around line 250-252: The Client struct's runtime must be Tokio-backed and we
should not accept arbitrary Runtime; change the runtime field to use
TokioRuntime (e.g., Arc<TokioRuntime>) and update all constructors/factory
functions (e.g., Client::new and any builders that currently accept Arc<dyn
Runtime>) to instantiate and inject a TokioRuntime internally instead of taking
a Runtime parameter; likewise ensure the persistence layer used by the main
client is the Diesel SQLite implementation (replace any runtime-agnostic or
alternative persistence injections with the Diesel SQLite persistence backend
used by whatsapp-rust) and update references to wacore::client::CoreClient
construction to use the TokioRuntime-backed Client.
In `@src/store/persistence_manager.rs`:
- Line 80: The background saver uses save_notify.notify(1) with
event_listener::Event which can drop notifications if no listener exists
(lost-wakeup), so ensure pending writes are always flushed on shutdown: modify
the persistence manager’s background task (the loop that calls save_to_disk) to
either 1) keep a persistent listener outside the select/save boundary and use
the "check dirty flag -> await listener -> re-check" pattern so notifications
aren't missed, or 2) add a graceful shutdown path that awaits the saver task and
forces a final save_to_disk when dirty is true before exiting; reference
save_notify, the atomic dirty flag, save_to_disk, and the background saver task
when implementing the chosen fix.
In `@wacore/src/protocol/retry.rs`:
- Around line 65-79: The function extract_registration_id_from_node currently
treats any bytes.len() >= 4 as valid and truncates longer payloads; change its
validation to only accept 1..=4 bytes and reject (return None) when the
registration payload is empty or longer than 4 bytes. Specifically, in
extract_registration_id_from_node (and using get_bytes_content and the
registration child lookup), replace the bytes.len() >= 4 branch with an exact
check for bytes.len() == 4, keep the 1..=3 variable-length branch for
bytes.len() between 1 and 3, return None for bytes.is_empty(), and also return
None when bytes.len() > 4 so oversized payloads are treated as invalid rather
than truncated.
In `@wacore/src/session.rs`:
- Around line 112-173: The code marks JIDs as in-flight by inserting jid_str
into self.processing before awaiting fetch_and_establish, but cleanup (removing
from self.processing and notifying self.pending) only runs after the await so
cancellation can permanently leave those JIDs marked; fix by introducing a
drop/RAII guard (e.g., a small struct like ProcessingGuard) created when you
insert jid_str into processing (or for each batch) that, on Drop, acquires the
same locks on self.processing and self.pending and removes the jid_str entries
and notifies waiters with an appropriate SessionError (e.g.,
SessionError::FetchFailed("cancelled")), and then disarm the guard (disable its
Drop behavior) when the normal post-await cleanup runs successfully; place the
guard creation right after processing.insert(...) (in the same scope where
to_process/to_wait are handled) and ensure notify logic uses the same
notify_result handling in both normal and Drop paths so waiters are never
stranded.
In `@wacore/src/store/in_memory.rs`:
- Around line 358-363: put_lid_mapping currently overwrites
lid_mappings[entry.lid] but doesn't remove the previous phone->lid reverse
entry, leaving a stale pn_to_lid key; to fix, inside put_lid_mapping check
s.lid_mappings.get(&entry.lid) before inserting, and if an existing entry exists
with a different phone_number, remove that old phone_number from s.pn_to_lid
(e.g., s.pn_to_lid.remove(&old.phone_number)), then proceed to insert the new
pn_to_lid and lid_mappings entries so the reverse index stays consistent;
references: put_lid_mapping, s.lid_mappings, s.pn_to_lid, and the
LidPnMappingEntry.phone_number field.
- Around line 538-540: The in-memory create() currently only increments
next_device_id, but PersistenceManager::new() expects create() to materialize
the device so exists() returns true and load() can read it; modify create() to
allocate a new device record in the in-memory storage (seed state.device) using
the generated id from next_device_id and appropriate default/placeholder fields
so subsequent exists() and load() observe the created device; update references
in create() (and any helper like state or storage map) to store that Device
entry and then return the id.
In `@wacore/src/store/persistence.rs`:
- Around line 124-143: run_background_saver currently takes self: Arc<Self> and
moves that strong Arc into a detached infinite task, preventing
PersistenceManager from ever being dropped; change the implementation to avoid
owning a strong Arc in the detached loop by converting the Arc<Self> to a
Weak<Self> (e.g., let weak = Arc::downgrade(&self)) and attempt to upgrade
inside the loop before calling save_to_disk or listening on save_notify,
aborting/returning when upgrade fails, or alternatively return/store an abort
handle from runtime.spawn so callers can cancel the background task; update
references in run_background_saver, the loop where save_notify.listen() and
save_to_disk() are used, and the task creation to use the Weak upgrade pattern
or to expose the AbortHandle.
- Around line 85-98: The save_to_disk implementation clears self.dirty before
awaiting backend.save, so if save fails the dirty flag is lost; change the
backend.save call so that on error you restore the dirty flag before propagating
the error (e.g. replace the .await.map_err(db_err)? pattern with handling that
on Err runs self.dirty.store(true, Ordering::Release) and then returns the
mapped StoreError). This touches save_to_disk, the dirty AtomicBool field, and
the backend.save call — ensure the restored dirty write happens before returning
the error.
---
Outside diff comments:
In `@src/client/sessions.rs`:
- Around line 29-45: Move the installation of the 64-permit semaphore to occur
before you set offline_sync_completed and make the publish atomic with proper
memory ordering: first acquire the lock on message_processing_semaphore and
replace the Arc with the new async_lock::Semaphore::new(64), then perform the
compare_exchange that flips offline_sync_completed using Release on success (and
Acquire for any loads that read it elsewhere); only after a successful flip call
offline_sync_notifier.notify(usize::MAX). Ensure references to
offline_sync_completed, message_processing_semaphore, compare_exchange, and
offline_sync_notifier are updated so the wider semaphore is visible before you
publish completion.
In `@src/request.rs`:
- Around line 159-178: The shutdown listener is registered too late
(shutdown_notifier.listen() is called after the is_running check and
send_node()), risking missed disconnect notifications; move the call to
shutdown_notifier.listen() to occur before the is_running check and before
calling send_node() so the Event listener is active during the window where a
shutdown may occur (update the code around the is_running check and send_node()
usage in request.rs to create let shutdown = self.shutdown_notifier.listen();
early, then perform the is_running check/send_node(), and leave the existing
futures::select! logic unchanged).
In `@src/socket/noise_socket.rs`:
- Around line 102-155: The code currently increments *write_counter immediately
(via counter = *write_counter; *write_counter = write_counter.wrapping_add(1))
which advances the nonce even if encryption/framing or transport.send fails;
change this so the counter is reserved locally but only commit the increment
after transport.send succeeds: compute let counter = *write_counter; let next =
counter.wrapping_add(1); use counter for encrypt_in_place_with_counter /
encrypt_with_counter and for framing, and only after if let Err(e) =
transport.send(out_buf).await { ... } else { *write_counter = next; } — ensure
both the INLINE_ENCRYPT_THRESHOLD (small-path using
encrypt_in_place_with_counter) and the blocking-path (encrypt_with_counter +
framing) follow this pattern so failed encrypt/framing/transport never advance
*write_counter.
---
Nitpick comments:
In `@src/bot.rs`:
- Around line 650-655: The build method uses .unwrap() on runtime, backend,
transport_factory, and http_client; replace these panicking calls by
destructuring self to extract the Some(...) values (e.g. match or let Self {
runtime: Some(runtime), backend: Some(backend), transport_factory:
Some(transport_factory), http_client: Some(http_client), .. } = self) so the
typestate guarantee is preserved without any unwraps and without introducing a
panic path in Bot::build.
In `@src/handlers/message.rs`:
- Around line 48-54: Replace the separate per-chat lock map usage with the
repository-standard Client::chat_locks to ensure a single serialization domain:
instead of calling client.message_enqueue_locks.get_with_by_ref(&chat_id, ...)
and using enqueue_mutex/ _enqueue_guard, obtain the mutex from client.chat_locks
(using the same get_with_by_ref and async_lock::Mutex type), await the returned
mutex.lock().await, and use the guard to serialize the enqueue operation for
chat_id (preserving the await and scope of the guard).
In `@src/socket/error.rs`:
- Around line 25-26: The error variant Join in src/socket/error.rs uses the
message "tokio join error" which is runtime-specific; update its Display/error
string to a runtime-neutral phrase like "task join error" (and apply the same
change to the analogous variant mentioned at lines 61-64). Locate the enum
variant named Join and replace the literal "tokio join error" with a neutral
message (e.g., "task join error" or "join error") so logs no longer reference
Tokio specifically while preserving the variant name and semantics.
In `@wacore/src/runtime.rs`:
- Around line 103-113: The blocking helper currently panics if the spawned
blocking closure `f` panics because `tx` is dropped and `rx.await.expect()`
panics; update `blocking` to explicitly propagate panic/errors by wrapping the
closure with std::panic::catch_unwind and sending a Result over `tx` (e.g., send
Ok(value) on success or Err(panic_payload) on unwind), change the receive side
(`rx.await`) to return a Result (or change the function signature to return
Result<T, E>) and map the Canceled/Err case to a clear error variant or message;
locate the `blocking` function and the use of `rt.spawn_blocking`, `tx`, and
`rx` to implement catch_unwind inside the spawned closure and handle the
received Result instead of calling `.expect()`.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: b36ab876-507a-41f0-9b5a-e8a85a538af9
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (55)
Cargo.tomlexamples/benchmark.rssrc/appstate_sync.rssrc/bot.rssrc/client.rssrc/client/device_registry.rssrc/client/sender_keys.rssrc/client/sessions.rssrc/download.rssrc/features/presence.rssrc/handlers/ib.rssrc/handlers/message.rssrc/handlers/notification.rssrc/handlers/router.rssrc/handshake.rssrc/history_sync.rssrc/keepalive.rssrc/lib.rssrc/main.rssrc/message.rssrc/pair_code.rssrc/pdo.rssrc/prekeys.rssrc/receipt.rssrc/request.rssrc/retry.rssrc/runtime_impl.rssrc/send.rssrc/session.rssrc/socket/error.rssrc/socket/noise_socket.rssrc/store/persistence_manager.rssrc/store/signal.rssrc/store/signal_adapter.rssrc/store/signal_cache.rssrc/test_utils.rssrc/types/enc_handler.rssrc/unified_session.rssrc/upload.rstests/e2e/src/lib.rstests/e2e/tests/connection.rswacore/Cargo.tomlwacore/src/lib.rswacore/src/protocol/keepalive.rswacore/src/protocol/mod.rswacore/src/protocol/retry.rswacore/src/runtime.rswacore/src/session.rswacore/src/stanza/mod.rswacore/src/stanza/notification.rswacore/src/stanza/receipt.rswacore/src/store/in_memory.rswacore/src/store/mod.rswacore/src/store/persistence.rswacore/src/store/signal_cache.rs
| /// Handle returned by [`Bot::run`] that can be awaited to wait for the | ||
| /// client's run loop to finish. | ||
| pub struct BotHandle { | ||
| done_rx: futures::channel::oneshot::Receiver<()>, | ||
| _abort_handle: wacore::runtime::AbortHandle, | ||
| } | ||
|
|
||
| impl BotHandle { | ||
| /// Abort the bot's run task. | ||
| pub fn abort(&self) { | ||
| // AbortHandle aborts on drop, but we also allow explicit abort | ||
| // by disconnecting. The actual abort happens via _abort_handle's Drop. | ||
| } |
There was a problem hiding this comment.
BotHandle::abort() currently does nothing, while Drop aborts.
Line 124 is a no-op, but the returned handle still owns the AbortHandle, so dropping BotHandle becomes the only working cancellation path. That inverts the old JoinHandle behavior: callers who ignore the handle stop the bot immediately, while callers who keep it still cannot explicitly abort.
Also applies to: 242-251
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/bot.rs` around lines 115 - 127, BotHandle::abort() is currently a no-op;
implement it to call the underlying abort handle so callers can explicitly
cancel the run task instead of relying on Drop. In the abort(&self) method,
invoke the AbortHandle's abort method on the _abort_handle field (e.g.
self._abort_handle.abort()) so the task is cancelled immediately; keep the
existing Drop behavior intact. Also apply the same change to the other BotHandle
implementation variant referenced around lines 242-251 so both handle types
provide explicit abort semantics.
| pub struct Client { | ||
| pub(crate) runtime: Arc<dyn Runtime>, | ||
| pub(crate) core: wacore::client::CoreClient, |
There was a problem hiding this comment.
Keep the main client's public API Tokio-backed.
Lines 251, 497, and 516 now let downstream instantiate whatsapp-rust's main Client with any Runtime. That pushes the runtime-agnostic boundary out of wacore and into whatsapp-rust/src, which is explicitly outside the supported architecture. Please inject TokioRuntime at this boundary instead of accepting an arbitrary runtime here. Based on learnings: "Applies to whatsapp-rust/src/**/*.rs : whatsapp-rust main client must use Tokio runtime and SQLite persistence with Diesel ORM".
Also applies to: 496-522
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/client.rs` around lines 250 - 252, The Client struct's runtime must be
Tokio-backed and we should not accept arbitrary Runtime; change the runtime
field to use TokioRuntime (e.g., Arc<TokioRuntime>) and update all
constructors/factory functions (e.g., Client::new and any builders that
currently accept Arc<dyn Runtime>) to instantiate and inject a TokioRuntime
internally instead of taking a Runtime parameter; likewise ensure the
persistence layer used by the main client is the Diesel SQLite implementation
(replace any runtime-agnostic or alternative persistence injections with the
Diesel SQLite persistence backend used by whatsapp-rust) and update references
to wacore::client::CoreClient construction to use the TokioRuntime-backed
Client.
| let result = process_history_sync( | ||
| compressed_data, | ||
| own_user_ref, | ||
| Some(|raw_bytes: Bytes| { | ||
| // Send Bytes through channel (zero-copy clone) | ||
| let _ = tx.blocking_send(raw_bytes); | ||
| let _ = tx.send_blocking(raw_bytes); | ||
| }), |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cd src && wc -l history_sync.rsRepository: jlucaso1/whatsapp-rust
Length of output: 85
🏁 Script executed:
cat -n src/history_sync.rs | sed -n '140,230p'Repository: jlucaso1/whatsapp-rust
Length of output: 4677
Stop awaiting the full parse after shutdown or propagate a cancellation signal to the parser.
Line 162 ignores send_blocking errors, and line 203 still awaits the full parse completion even after the receive loop breaks for shutdown (line 179-184). This causes process_history_sync() to continue decompressing the entire blob even though results are no longer being processed, defeating the early-exit intent. For large history blobs, this unnecessarily delays shutdown and wastes CPU. Either propagate the closed-channel signal into the parser to halt early, or stop awaiting result_rx once shutdown has broken the dispatch loop.
|
|
||
| self.dirty.store(true, Ordering::Relaxed); | ||
| self.save_notify.notify_one(); | ||
| self.save_notify.notify(1); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n src/store/persistence_manager.rs | sed -n '70,90p'Repository: jlucaso1/whatsapp-rust
Length of output: 889
🏁 Script executed:
cat -n src/store/persistence_manager.rs | sed -n '120,150p'Repository: jlucaso1/whatsapp-rust
Length of output: 1451
🏁 Script executed:
git log --oneline --all --grep="Notify" | head -20Repository: jlucaso1/whatsapp-rust
Length of output: 48
🏁 Script executed:
rg "event_listener::Event" -A 5 -B 5Repository: jlucaso1/whatsapp-rust
Length of output: 8261
🏁 Script executed:
web_search
event_listener::Event Rust crate listen() semantics notification race conditionRepository: jlucaso1/whatsapp-rust
Length of output: 281
🏁 Script executed:
rg "save_notify" -B 5 -A 5 src/store/persistence_manager.rs | head -60Repository: jlucaso1/whatsapp-rust
Length of output: 1439
🏁 Script executed:
rg "pub fn run_background_saver|fn modify_device" -A 15 src/store/persistence_manager.rsRepository: jlucaso1/whatsapp-rust
Length of output: 1362
🏁 Script executed:
rg "dirty" -B 2 -A 2 src/store/persistence_manager.rs | head -40Repository: jlucaso1/whatsapp-rust
Length of output: 826
🌐 Web query:
event_listener Rust crate github documentation listen() semantics
💡 Result:
The event_listener Rust crate (https://crates.io/crates/event-listener, latest v5.4.1 as of 2025-08-03, repo: https://github.com/smol-rs/event-listener) provides a synchronization primitive similar to eventcounts for notifying async tasks or threads. It allows turning non-blocking data structures into async or blocking ones. The relevant method is Event::listen(&self) -> EventListener, which returns a guard (EventListener) listening for a notification on that Event. It emits a SeqCst fence after registering the listener, ensuring proper synchronization. Semantics and usage: - Register listeners with event.listen to get an EventListener. - Await the listener (.await) for async waiting or call listener.wait for blocking. - Notifications via event.notify(n: usize) wake at least n active listeners (or event.notify_additional(n) for previously unnotified ones). Notifications lost if no listeners. - Listeners FIFO for fairness. - If a notified listener drops without consuming, it wakes another (depending on notify type). - Often used in loops: check condition, listen if not met, wait if still not met. - Stack-based alternative: listener!(event => listener) macro for efficiency (no heap alloc). Example from docs: use event_listener::Event; let event = Event::new; let listener = event.listen; event.notify(1); listener.wait; // or await listener {value: "The event_listener Rust crate (https://crates.io/crates/event-listener, latest v5.4.1 as of 2025-08-03, repo: https://github.com/smol-rs/event-listener) provides a synchronization primitive similar to eventcounts for notifying async tasks or threads. It allows turning non-blocking data structures into async or blocking ones.\n\nThe relevant method is Event::listen(&self) -> EventListener, which returns a guard (EventListener) listening for a notification on that Event. It emits a SeqCst fence after registering the listener, ensuring proper synchronization.\n\nSemantics and usage:\n- Register listeners with event.listen to get an EventListener.\n- Await the listener (.await) for async waiting or call listener.wait for blocking.\n- Notifications via event.notify(n: usize) wake at least n active listeners (or event.notify_additional(n) for previously unnotified ones). Notifications lost if no listeners.\n- Listeners FIFO for fairness.\n- If a notified listener drops without consuming, it wakes another (depending on notify type).\n- Often used in loops: check condition, listen if not met, wait if still not met.\n- Stack-based alternative: listener!(event => listener) macro for efficiency (no heap alloc).\n\nExample from docs:\n\nuse event_listener::Event;\nlet event = Event::new;\nlet listener = event.listen;\nevent.notify(1);\nlistener.wait; // or await listener\n", citations: [1,3,9,10,13], confidence: "high"}
Citations:
- 1: https://docs.rs/event-listener/latest/event_listener
- 2: https://crates.io/crates/event-listener
- 3: https://docs.rs/event-listener/latest/event_listener/struct.Event.html
- 4: https://docs.rs/event-listener/latest/event_listener/struct.EventListener.html
- 5: https://rustdocs.bsx.fi/event_listener/struct.Event.html
🏁 Script executed:
rg "Event::new|event_listener" src/store/persistence_manager.rs -B 3 -A 3Repository: jlucaso1/whatsapp-rust
Length of output: 441
🏁 Script executed:
rg "dirty.store|dirty.swap" src/store/persistence_manager.rs -B 5 -A 5Repository: jlucaso1/whatsapp-rust
Length of output: 664
🏁 Script executed:
git log --oneline --all -- src/store/persistence_manager.rs | head -10Repository: jlucaso1/whatsapp-rust
Length of output: 715
🏁 Script executed:
rg "shutdown|detach|drop" src/store/persistence_manager.rs -B 3 -A 3Repository: jlucaso1/whatsapp-rust
Length of output: 522
🏁 Script executed:
cat -n src/store/persistence_manager.rs | sed -n '1,70p'Repository: jlucaso1/whatsapp-rust
Length of output: 3078
🏁 Script executed:
rg "PersistenceManager" src/ -l | head -5Repository: jlucaso1/whatsapp-rust
Length of output: 166
🏁 Script executed:
git show aa3660a:src/store/persistence_manager.rs | head -150 | tail -80Repository: jlucaso1/whatsapp-rust
Length of output: 2892
🏁 Script executed:
rg "run_background_saver" src/ -B 5 -A 2Repository: jlucaso1/whatsapp-rust
Length of output: 1034
🏁 Script executed:
rg "detach|shutdown_notifier" src/client.rs | head -20Repository: jlucaso1/whatsapp-rust
Length of output: 869
🏁 Script executed:
cat -n src/store/persistence_manager.rs | sed -n '85,100p'Repository: jlucaso1/whatsapp-rust
Length of output: 729
🏁 Script executed:
rg "PersistenceManager::new|run_background_saver" src/ -B 3 -A 8 | head -50Repository: jlucaso1/whatsapp-rust
Length of output: 2416
🏁 Script executed:
rg "run_background_saver|shutdown" src/bot.rs -B 5 -A 5Repository: jlucaso1/whatsapp-rust
Length of output: 559
🏁 Script executed:
rg "async fn shutdown|async fn stop|async fn close" src/ -B 2 -A 8Repository: jlucaso1/whatsapp-rust
Length of output: 48
🏁 Script executed:
cat -n src/store/persistence_manager.rs | sed -n '126,148p'Repository: jlucaso1/whatsapp-rust
Length of output: 1172
🏁 Script executed:
rg "Notify::|event_listener::Event" src/store/ -B 2 -A 2Repository: jlucaso1/whatsapp-rust
Length of output: 365
🏁 Script executed:
rg "process.*shutdown|drop.*persistence|stop.*persistence" src/ -B 3 -A 3Repository: jlucaso1/whatsapp-rust
Length of output: 48
🏁 Script executed:
rg "\.notify\(|notify_one\(" src/store/persistence_manager.rs -B 3 -A 3Repository: jlucaso1/whatsapp-rust
Length of output: 225
🏁 Script executed:
cat -n src/store/persistence_manager.rs | sed -n '75,105p'Repository: jlucaso1/whatsapp-rust
Length of output: 1266
event_listener::Event creates a lost-wakeup window that can drop device updates on fast shutdown.
Unlike tokio::sync::Notify, event_listener::Event only wakes listeners that are already registered. In this loop, a listener only exists within the select! scope. If notify(1) fires before the next iteration's listen() call or during save_to_disk() execution, the notification is lost. The atomic dirty flag ensures the write isn't forgotten, but if the background saver task exits before the interval timeout (30s) fires, the pending write is never flushed to disk.
Either add a graceful shutdown handler that flushes pending writes, keep a persistent listener across the save boundary with the flag-check pattern mentioned, or wait for the saver to complete before shutdown.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/store/persistence_manager.rs` at line 80, The background saver uses
save_notify.notify(1) with event_listener::Event which can drop notifications if
no listener exists (lost-wakeup), so ensure pending writes are always flushed on
shutdown: modify the persistence manager’s background task (the loop that calls
save_to_disk) to either 1) keep a persistent listener outside the select/save
boundary and use the "check dirty flag -> await listener -> re-check" pattern so
notifications aren't missed, or 2) add a graceful shutdown path that awaits the
saver task and forces a final save_to_disk when dirty is true before exiting;
reference save_notify, the atomic dirty flag, save_to_disk, and the background
saver task when implementing the chosen fix.
| pub fn device_arc(&self) -> Arc<RwLock<Device>> { | ||
| self.device.clone() | ||
| } |
There was a problem hiding this comment.
Don't publish raw mutable access to Device.
Line 60 exposes the underlying RwLock<Device> directly, so callers can mutate state without setting dirty; line 72 adds a generic mutator that bypasses the DeviceCommand invariants entirely. These should be private/test-only helpers, with process_command() remaining the public mutation path. As per coding guidelines: "Never modify Device state directly; use DeviceCommand + PersistenceManager::process_command() instead".
Also applies to: 72-83
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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/keepalive.rs`:
- Around line 100-101: The loop recreates the shutdown_notifier listener inside
each iteration causing a race where a shutdown can be missed; move the call to
shutdown_notifier.listen() to the top of the loop (before awaiting sleep or
other branches such as self.runtime.sleep(...).fuse()) so the listener is
registered first, then immediately check the is_running flag (call is_running()
or check the shared state) right after registration and break/return if not
running; apply the same change for the second occurrence around the code at
lines referenced (the block containing disconnect(), listen(), and the sleep
select) so disconnect() cannot fire between sleep completion and a new listen()
registration.
In `@wacore/src/runtime.rs`:
- Around line 113-125: The panic message in function blocking (symbols:
blocking, rt.spawn_blocking, tx, rx, f()) incorrectly attributes a canceled
oneshot to "runtime shutting down"; update the failure handling to use a
clearer, more generic message that covers both closure panics and runtime
shutdown (e.g., "blocking task failed to complete (closure panic or runtime
shutdown?)") or alternatively update the function/doc comment to explicitly note
that f() panicking will cause the oneshot to be canceled; modify the
rx.await.unwrap_or_else panic call to use the new generic message and ensure any
comment above blocking reflects the possible causes.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 1a8613ab-2794-4904-a014-0e2ae925d71a
📒 Files selected for processing (4)
src/keepalive.rssrc/message.rswacore/src/runtime.rswacore/src/store/persistence.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- wacore/src/store/persistence.rs
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (3)
wacore/src/store/persistence.rs (1)
68-70:⚠️ Potential issue | 🟠 MajorDon’t expose raw
Devicemutation from the public API.
device_arc()lets callers take a write lock without settingdirty, andmodify_device()lets them bypass theDeviceCommandinvariants entirely. The public surface should stayget_device_snapshot()+process_command(), with these helpers kept internal/test-only.As per coding guidelines, "Never modify Device state directly; use
DeviceCommand+PersistenceManager::process_command()instead" and "Read Device state viaget_device_snapshot()method".Also applies to: 80-91
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/store/persistence.rs` around lines 68 - 70, Publicly exposing device_arc() (and any helpers like modify_device()) allows callers to mutate Device without setting dirty flags or going through DeviceCommand invariants; make these helper methods non-public (privated or cfg(test)) so the public API only exposes get_device_snapshot() and PersistenceManager::process_command(), and update any call sites to use get_device_snapshot() for reads and process_command() for mutations instead.wacore/src/store/in_memory.rs (1)
547-549:⚠️ Potential issue | 🟠 Major
create()still needs to materialize the device row.
PersistenceManager::new()assumesexists() -> create() -> load()makes the backend observable immediately. As written,exists()stays false andload()staysNoneuntil some latersave(), so repeated initialization against the same backend can keep allocating new ids without ever seeing the created device.🛠️ Minimal fix
async fn create(&self) -> Result<i32> { let id = self.next_device_id.fetch_add(1, Ordering::Relaxed); + let mut s = self.state.lock().await; + s.device.get_or_insert_with(Device::new); Ok(id) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/store/in_memory.rs` around lines 547 - 549, create() currently only allocates an id via next_device_id.fetch_add but does not materialize the device row, so PersistenceManager::new()'s assumed exists() -> create() -> load() sequence fails; update create() to insert a new device entry into the in-memory backend (the same data structure used by exists()/load()), initializing any required fields/state for the device so exists() returns true and load() returns the newly created device immediately (ensure you use the generated id and update whatever map or collection holds devices, and keep next_device_id usage as is).src/store/persistence_manager.rs (1)
127-158:⚠️ Potential issue | 🟠 MajorThe detached saver can still lose the last dirty write.
A
modify_device()call that lands duringsave_to_disk()or between listener lifetimes setsdirty = true, but itsnotify(1)is dropped byEvent. If the manager is then dropped before the next interval tick, the weak-exit path returns without ever flushing that snapshot. Please re-checkdirtybefore awaiting a new listener and add a final-save/shutdown path instead of relying on the next timer wakeup.🤖 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 127 - 158, Re-check the PersistenceManager's dirty flag and add a proper shutdown flush: inside run_background_saver, after creating listener = this.save_notify.listen() (and before dropping the strong Arc and awaiting), check this.dirty and call this.save_to_disk().await if true so writes that happen between listener lifetimes aren't lost; additionally implement a final-save/shutdown path by adding a shutdown mechanism (e.g., a new shutdown_notify or a shutdown_and_wait() on PersistenceManager) that calls save_notify.notify(1) and waits for the background task to finish, and wire callers (or Drop) to invoke it so the background saver can perform a last save before the manager is dropped. Ensure references to save_notify.listen(), save_to_disk(), modify_device(), and notify(1) are used to locate and implement these changes.
🤖 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/request.rs`:
- Around line 150-152: After calling self.shutdown_notifier.listen() (i.e.,
after creating the shutdown listener), immediately re-check the shutdown state
by calling self.is_running() (or the appropriate shutdown check used earlier)
and return early if it indicates shutdown; this ensures that any shutdown which
occurred between the earlier guard and the listener registration is handled
promptly so the request doesn't wait the full IQ timeout—update the code around
the send_node()/shutdown_notifier.listen() sequence to perform that second
is_running check and return the same early-fail behavior used for the initial
check.
In `@wacore/src/store/persistence.rs`:
- Around line 133-165: The background saver currently exits immediately when the
manager Arc is dropped, risking lost changes; add a graceful shutdown that
performs a final save: introduce a place to store the spawned task handle (e.g.,
a field like background_task: Mutex<Option<JoinHandle<()>>> on
PersistenceManager), stop calling .detach() in run_background_saver and instead
save the JoinHandle there, and implement an async shutdown(&self) method that
notifies the saver (use save_notify.notify_one()), awaits the background task
handle to finish, and calls save_to_disk(). Alternatively, if you need a
synchronous API, provide a shutdown_blocking(&self, runtime: Arc<dyn Runtime>)
that calls runtime.block_on(self.shutdown()). Ensure run_background_saver,
save_to_disk, save_notify, and the new background_task field are used
consistently so the final flush happens before the manager is dropped.
---
Duplicate comments:
In `@src/store/persistence_manager.rs`:
- Around line 127-158: Re-check the PersistenceManager's dirty flag and add a
proper shutdown flush: inside run_background_saver, after creating listener =
this.save_notify.listen() (and before dropping the strong Arc and awaiting),
check this.dirty and call this.save_to_disk().await if true so writes that
happen between listener lifetimes aren't lost; additionally implement a
final-save/shutdown path by adding a shutdown mechanism (e.g., a new
shutdown_notify or a shutdown_and_wait() on PersistenceManager) that calls
save_notify.notify(1) and waits for the background task to finish, and wire
callers (or Drop) to invoke it so the background saver can perform a last save
before the manager is dropped. Ensure references to save_notify.listen(),
save_to_disk(), modify_device(), and notify(1) are used to locate and implement
these changes.
In `@wacore/src/store/in_memory.rs`:
- Around line 547-549: create() currently only allocates an id via
next_device_id.fetch_add but does not materialize the device row, so
PersistenceManager::new()'s assumed exists() -> create() -> load() sequence
fails; update create() to insert a new device entry into the in-memory backend
(the same data structure used by exists()/load()), initializing any required
fields/state for the device so exists() returns true and load() returns the
newly created device immediately (ensure you use the generated id and update
whatever map or collection holds devices, and keep next_device_id usage as is).
In `@wacore/src/store/persistence.rs`:
- Around line 68-70: Publicly exposing device_arc() (and any helpers like
modify_device()) allows callers to mutate Device without setting dirty flags or
going through DeviceCommand invariants; make these helper methods non-public
(privated or cfg(test)) so the public API only exposes get_device_snapshot() and
PersistenceManager::process_command(), and update any call sites to use
get_device_snapshot() for reads and process_command() for mutations instead.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 7c665c9f-5096-4055-8d76-5c26240cf2c9
📒 Files selected for processing (9)
src/bot.rssrc/client/sessions.rssrc/request.rssrc/socket/error.rssrc/store/persistence_manager.rswacore/src/protocol/retry.rswacore/src/runtime.rswacore/src/store/in_memory.rswacore/src/store/persistence.rs
🚧 Files skipped from review as they are similar to previous changes (3)
- src/socket/error.rs
- wacore/src/protocol/retry.rs
- src/bot.rs
| pub fn run_background_saver(self: Arc<Self>, runtime: Arc<dyn Runtime>, interval: Duration) { | ||
| let rt = runtime.clone(); | ||
| let weak = Arc::downgrade(&self); | ||
| drop(self); // Release the strong reference; the caller's Arc keeps it alive | ||
| runtime | ||
| .spawn(Box::pin(async move { | ||
| loop { | ||
| let Some(this) = weak.upgrade() else { | ||
| debug!("PersistenceManager dropped, exiting background saver."); | ||
| return; | ||
| }; | ||
| let listener = this.save_notify.listen(); | ||
| drop(this); // Don't hold strong ref while sleeping | ||
|
|
||
| futures::select! { | ||
| _ = listener.fuse() => { | ||
| debug!("Save notification received."); | ||
| } | ||
| _ = rt.sleep(interval).fuse() => {} | ||
| } | ||
|
|
||
| let Some(this) = weak.upgrade() else { | ||
| debug!("PersistenceManager dropped, exiting background saver."); | ||
| return; | ||
| }; | ||
| if let Err(e) = this.save_to_disk().await { | ||
| error!("Error saving device state in background: {e}"); | ||
| } | ||
| } | ||
| })) | ||
| .detach(); | ||
| debug!("Background saver task started with interval {interval:?}"); | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
fd persistence.rs -t f | head -20Repository: jlucaso1/whatsapp-rust
Length of output: 97
🏁 Script executed:
cat -n wacore/src/store/persistence.rs | sed -n '120,180p'Repository: jlucaso1/whatsapp-rust
Length of output: 2709
🏁 Script executed:
# Look for save_notify definition and Event usage
rg "save_notify" wacore/src/store/persistence.rs -B 2 -A 2Repository: jlucaso1/whatsapp-rust
Length of output: 599
🏁 Script executed:
# Look for the Event struct/enum definition
rg "enum Event|struct Event" wacore/src/store/persistence.rs -B 2 -A 5Repository: jlucaso1/whatsapp-rust
Length of output: 48
🏁 Script executed:
# Find save_to_disk method
ast-grep --pattern 'fn save_to_disk($_) {
$$$
}'Repository: jlucaso1/whatsapp-rust
Length of output: 48
🏁 Script executed:
rg "use.*Event" wacore/src/store/persistence.rsRepository: jlucaso1/whatsapp-rust
Length of output: 92
🏁 Script executed:
# Check the full struct definition and imports at top of file
head -50 wacore/src/store/persistence.rsRepository: jlucaso1/whatsapp-rust
Length of output: 2106
🏁 Script executed:
# Look for any Drop implementations or cleanup mechanisms
rg "impl Drop" wacore/src/store/persistence.rs -A 10Repository: jlucaso1/whatsapp-rust
Length of output: 48
🏁 Script executed:
# Check how dirty flag is checked/used
rg "dirty" wacore/src/store/persistence.rs -B 1 -A 1Repository: jlucaso1/whatsapp-rust
Length of output: 1046
🏁 Script executed:
# Look for save_to_disk implementation to understand what it does
rg "fn save_to_disk" wacore/src/store/persistence.rs -A 20Repository: jlucaso1/whatsapp-rust
Length of output: 872
🏁 Script executed:
rg "fn drop|impl Drop" wacore/src/store/persistence.rsRepository: jlucaso1/whatsapp-rust
Length of output: 48
🏁 Script executed:
# Check how run_background_saver is called and if there's a shutdown pattern
rg "run_background_saver" wacore/src -r -B 2 -A 2Repository: jlucaso1/whatsapp-rust
Length of output: 376
🏁 Script executed:
# Look for any graceful shutdown or final save mechanism
rg "shutdown|finalize|flush" wacore/src/store/persistence.rs -B 2 -A 2Repository: jlucaso1/whatsapp-rust
Length of output: 285
🏁 Script executed:
# Search for run_background_saver more carefully across the repo
find . -name "*.rs" -type f | xargs grep -l "run_background_saver" 2>/dev/nullRepository: jlucaso1/whatsapp-rust
Length of output: 147
🏁 Script executed:
# Look for any graceful shutdown patterns in the codebase
rg "drop.*Arc|Drop|shutdown" wacore/src --type rs | head -20Repository: jlucaso1/whatsapp-rust
Length of output: 92
🏁 Script executed:
# Check the entire file structure around the background saver to understand context
sed -n '133,200p' wacore/src/store/persistence.rsRepository: jlucaso1/whatsapp-rust
Length of output: 2410
🏁 Script executed:
# Check how run_background_saver is used in bot.rs and persistence_manager.rs
grep -B 5 -A 5 "run_background_saver" ./src/bot.rsRepository: jlucaso1/whatsapp-rust
Length of output: 559
🏁 Script executed:
grep -B 5 -A 5 "run_background_saver" ./src/store/persistence_manager.rsRepository: jlucaso1/whatsapp-rust
Length of output: 505
🏁 Script executed:
# Check if there's a shutdown method or if Arc is held until end of program
grep -B 10 -A 10 "run_background_saver" ./src/store/persistence_manager.rs | head -40Repository: jlucaso1/whatsapp-rust
Length of output: 861
Add a final-flush mechanism on Arc drop to prevent data loss.
The background saver exits immediately when the manager is dropped, without flushing any remaining dirty state. If modify_device() is called and the Arc is dropped before the next save interval, unsaved changes are lost. Implement a graceful shutdown (e.g., via a Drop impl or explicit shutdown() method) that performs a final save_to_disk() before the task exits.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@wacore/src/store/persistence.rs` around lines 133 - 165, The background saver
currently exits immediately when the manager Arc is dropped, risking lost
changes; add a graceful shutdown that performs a final save: introduce a place
to store the spawned task handle (e.g., a field like background_task:
Mutex<Option<JoinHandle<()>>> on PersistenceManager), stop calling .detach() in
run_background_saver and instead save the JoinHandle there, and implement an
async shutdown(&self) method that notifies the saver (use
save_notify.notify_one()), awaits the background task handle to finish, and
calls save_to_disk(). Alternatively, if you need a synchronous API, provide a
shutdown_blocking(&self, runtime: Arc<dyn Runtime>) that calls
runtime.block_on(self.shutdown()). Ensure run_background_saver, save_to_disk,
save_notify, and the new background_task field are used consistently so the
final flush happens before the manager is dropped.
|
@claude review |
|
Claude finished @jlucaso1's task in 2m 15s —— View job PR Review: refactor: runtime agnostic
This is a well-executed refactor. The Issues to address1. Noise counter advanced before successful send —
// Current (line 109-110):
let counter = *write_counter;
*write_counter = write_counter.wrapping_add(1);
// Should be:
let counter = *write_counter;
// ... encrypt + send ...
// Only after transport.send() succeeds:
*write_counter = counter.wrapping_add(1);2. Keepalive loop shutdown listener race — The 3.
4. The shutdown listener is correctly registered before 5. Background saver has no final flush on drop — Both persistence managers use 6. The coding guidelines say "Never modify Device state directly. Use Non-blocking observations
Overall this is solid work — the core abstraction is well-designed, the previous review feedback was addressed, and the codebase is in a good position for future WASM support. The noise counter issue (item 1) is the highest priority fix. |
|
@coderabbitai review |
|
@codex review |
![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: b480b2ce91
ℹ️ 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".
| pub struct BotHandle { | ||
| done_rx: futures::channel::oneshot::Receiver<()>, | ||
| _abort_handle: wacore::runtime::AbortHandle, |
There was a problem hiding this comment.
BotHandle is dropped
Storing AbortHandle inside BotHandle changes Bot::run() from the old JoinHandle semantics to "abort on drop". Any existing caller that does bot.run().await?; without keeping the returned handle alive, or lets the handle fall out of scope, will now cancel client.run() immediately after startup instead of leaving the bot running in the background.
Useful? React with 👍 / 👎.
| pub async fn get_with<F>(&self, key: K, init: F) -> V | ||
| where | ||
| F: std::future::Future<Output = V>, | ||
| { | ||
| if let Some(v) = self.get(&key).await { | ||
| return v; | ||
| } | ||
| let value = init.await; | ||
| self.insert(key, value.clone()).await; | ||
| value | ||
| } | ||
|
|
||
| /// Same as [`get_with`](Self::get_with) but takes the key by reference. | ||
| pub async fn get_with_by_ref<F>(&self, key: &K, init: F) -> V | ||
| where | ||
| F: std::future::Future<Output = V>, | ||
| { | ||
| if let Some(v) = self.get(key).await { | ||
| return v; | ||
| } | ||
| let value = init.await; | ||
| self.insert(key.clone(), value.clone()).await; |
There was a problem hiding this comment.
PortableCache::get_with* single-flight per key
When moka-cache is disabled, these helpers no longer provide the atomic initialization behavior that the call sites rely on. src/handlers/message.rs explicitly uses get_with_by_ref to establish per-chat ordering, and src/send.rs uses get_with to reuse one mutex per session. Two concurrent misses for the same key can both run init, create different Arc<Mutex<_>>/channel instances, and then overwrite each other, which breaks serialization and can process same-chat messages out of order or establish the same session in parallel.
Useful? React with 👍 / 👎.
| pub fn invalidate_all(&self) { | ||
| if let Some(mut guard) = self.inner.try_write() { | ||
| guard.map.clear(); | ||
| guard.insertion_order.clear(); | ||
| } |
There was a problem hiding this comment.
invalidate_all on lock contention
In the non-moka path this clear operation becomes a best-effort no-op whenever another task is reading or writing the cache. Client::mark_disconnected() calls retried_group_messages.invalidate_all() during reconnect (src/client.rs:1007) specifically to reset retry deduplication, so a concurrent cache access can leave stale entries behind for their full TTL and cause later retry receipts after reconnect to be ignored as duplicates.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Actionable comments posted: 16
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/socket/noise_socket.rs (1)
152-158:⚠️ Potential issue | 🔴 CriticalDo not reuse the Noise write counter after a send error.
Advancing
write_counteronly aftertransport.send()succeeds makes the next send reuse the same AEAD nonce whenever the transport reportsErr. That error boundary is not strong enough to guarantee the frame never left the process, so this can desynchronize the Noise state and, in the worst case, reuse a nonce under the same key.🔐 Safer counter ordering
- if let Err(e) = transport.send(out_buf).await { - return Err(EncryptSendError::transport(e)); - } - - // Only advance the counter after the encrypted frame was successfully sent. - // If transport.send() fails, we can retry with the same counter value. *write_counter = write_counter.wrapping_add(1); + if let Err(e) = transport.send(out_buf).await { + return Err(EncryptSendError::transport(e)); + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/socket/noise_socket.rs` around lines 152 - 158, The current code delays advancing write_counter until after transport.send succeeds, which can cause nonce reuse if transport reports Err; change the logic in the send path that builds and sends out_buf so that *write_counter is incremented (using write_counter.wrapping_add(1)) immediately after sealing the frame and before calling transport.send(out_buf). Keep the same error mapping to EncryptSendError::transport for transport.send failures but do not decrement or reuse the counter on error; update the section around transport.send, write_counter, out_buf, and EncryptSendError::transport so the counter is always advanced prior to the send attempt.
♻️ Duplicate comments (3)
src/keepalive.rs (1)
95-105:⚠️ Potential issue | 🟡 MinorAlso check
is_runningimmediately after registeringshutdown.This still leaves a narrow lost-shutdown window: if
disconnect()flips shutdown state just before the new listener is created, the event is gone and the loop can sleep one more full interval. Checkingis_runningimmediately after Line 97 makes the exit deterministic.Suggested patch
let shutdown = self.shutdown_notifier.listen(); + if !self.is_running.load(Ordering::Relaxed) { + debug!(target: "Client/Keepalive", "Shutdown already requested, exiting keepalive loop."); + return; + }Expected result:
disconnect()updates shutdown state independently ofis_connected(), so the guard above is still needed to cover a missed notification.#!/bin/bash set -euo pipefail sed -n '89,185p' src/keepalive.rs rg -n -C2 'shutdown_notifier|is_running|is_connected|disconnect\(' src/client.rs src/keepalive.rs🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/keepalive.rs` around lines 95 - 105, After calling self.shutdown_notifier.listen() in the keepalive loop, immediately re-check the shutdown state by calling is_running() (or equivalent) and break/return if it reports stopped to avoid a lost-shutdown window; i.e., after creating the listener from shutdown_notifier.listen() but before awaiting self.runtime.sleep(interval) (and before the futures::select! sleep branch), call self.is_running() and exit if false so disconnect() cannot be missed between listener creation and the sleep. Use the existing symbols shutdown_notifier.listen(), is_running(), and the surrounding keepalive loop to place this check.src/client.rs (1)
249-250:⚠️ Potential issue | 🟠 MajorKeep
whatsapp-rust::ClientTokio-backed at this boundary.Accepting
Arc<dyn Runtime>in the main client API pushes the runtime-agnostic surface out ofwacoreand intosrc/.Client::new*should construct/useTokioRuntimeinternally so downstream code cannot instantiate the main client on a non-Tokio executor. Based on learnings:Applies to whatsapp-rust/src/**/*.rs : whatsapp-rust main client must use Tokio runtime and SQLite persistence with Diesel ORM.Also applies to: 496-522
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/client.rs` around lines 249 - 250, Client currently exposes pub(crate) runtime: Arc<dyn Runtime> on the Client boundary which allows non-Tokio runtimes to be injected; change Client construction so the public API does not accept or store a generic Runtime: make Client::new / Client::new_* constructors create and embed a TokioRuntime internally (e.g., use TokioRuntime within the Client implementation and remove or make private any APIs that accept Arc<dyn Runtime>), ensure the Client struct no longer exposes a runtime-agnostic field and that wacore-facing code uses the concrete TokioRuntime for execution and SQLite/Diesel persistence initialization to keep the main client Tokio-backed.src/history_sync.rs (1)
157-170:⚠️ Potential issue | 🟠 MajorShutdown still waits for the full history parse.
Closing
rxhere only makessend_blockingfail; it does not stopprocess_history_sync(), because the callback ignores that failure and parsing continues until the blob is fully consumed.result_rx.awaitthen keeps this task alive anyway, so the early-exit path still burns CPU during shutdown. Please propagate cancellation back intoprocess_history_syncfrom the callback, or stop awaitingresult_rxonce the dispatch loop breaks for shutdown.Also applies to: 205-206
🧹 Nitpick comments (2)
wacore/src/protocol/keepalive.rs (1)
63-67: Loosen the near-zero timing assertion to reduce CI flakiness.
Line 66 (elapsed < 100) is fragile on busy runners; consider a wider tolerance or clock injection for deterministic tests.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/protocol/keepalive.rs` around lines 63 - 67, The test helper ms_since_recent currently asserts elapsed < 100 which is flaky on CI; either relax the tolerance (e.g., increase to a few hundred ms such as 500) or refactor ms_since_recent to accept an injected time provider/value so tests can pass a deterministic now_ms; update the call site to use crate::time::now_millis() (or a mock) and adjust the assertion on ms_since(now_ms) accordingly to use the new tolerance or injected clock in the ms_since_recent helper.transports/tokio-transport/src/lib.rs (1)
136-137: Remove unnecessary wasm32 conditional on a Tokio-specific transport crate.The conditional
async_trait(?Send)pattern here is redundant because Tokio does not support wasm32 targets—it requires a real OS runtime. This crate (tokio-transport) will fail to compile for wasm32 during dependency resolution, not due to trait bounds. The conditional attributes suggest cross-platform compatibility that cannot be achieved through trait syntax alone.If wasm32 support is needed, create a separate transport implementation (e.g., using browser WebSocket APIs), rather than relying on a Tokio-based transport with conditional trait attributes.
Remove the conditional
async_traitdeclarations or document why this crate targets wasm32 despite Tokio's platform constraints.Also applies to: 201-202
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@transports/tokio-transport/src/lib.rs` around lines 136 - 137, The async_trait attribute is conditionally applied for wasm32 targets but this crate (tokio-transport) relies on Tokio and cannot target wasm32; remove the redundant conditional attributes. Edit the attribute usages of #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] / #[cfg_attr(not(target_arch = "wasm32"), async_trait)] (and the duplicate occurrences later) and replace them with a single plain #[async_trait] (or remove the cfg_attr wrapper altogether) so the async_trait macro is consistently applied for the Tokio-based transport; alternatively, if you intend to keep wasm32 documentation, add a crate-level comment stating that tokio-transport does not support wasm32 and that a separate transport is required for browser targets.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@http_clients/ureq-client/src/lib.rs`:
- Around line 42-43: The crate currently uses ureq::Agent and
tokio::task::spawn_blocking but only annotates the async_trait with a
conditional ?Send; add a compile-time exclusion for wasm by gating the entire
backend implementation with #[cfg(not(target_arch = "wasm32"))] (e.g. on the
crate root or on the module/impl containing the ureq client and functions that
call tokio::task::spawn_blocking) so the ureq-based code isn't compiled for
wasm32, or alternatively provide a separate wasm-compatible HTTP client
implementation and gate selection by target_arch; ensure the symbols to protect
include the module/impl containing ureq::Agent and any functions invoking
tokio::task::spawn_blocking.
In `@src/bot.rs`:
- Around line 415-425: The public with_runtime<Rt: Runtime> on BotBuilder
exposes arbitrary runtimes; keep the top-level crate Tokio-only by removing or
privatizing the generic runtime hook and defaulting/injecting a Tokio runtime
instead. Replace the public signature of BotBuilder::with_runtime (or make it
crate-private) so it no longer accepts a generic Rt: Runtime; ensure
BotBuilder::runtime is initialized with a TokioRuntime (or
Arc::new(TokioRuntime)) in the public builder path and move any generic Runtime
trait factory/with_runtime functionality into the internal wacore crate or test
helpers (keep references to BotBuilder, with_runtime, Runtime, and TokioRuntime
to locate and update the code).
In `@src/lib.rs`:
- Line 36: Re-exporting wacore::runtime::Runtime at the crate root exposes a
runtime-pluggable API; remove the public re-export (pub use
wacore::runtime::Runtime) so the Runtime trait stays internal, and instead
expose only the Tokio-backed surface from src (create or re-export a
Tokio-specific runtime type or factory in src, e.g., TokioRuntime or functions
that construct the Tokio runtime used by the main client), ensuring the public
API and main client remain Tokio-only and continue to use SQLite/Diesel
persistence as implemented.
In `@src/message.rs`:
- Around line 125-146: The increment_retry_count logic uses get+insert which is
not serialized and can race with detached tasks (e.g., spawn_retry_receipt) or
the sender-count preseed; wrap the per-chat retry updates using the per-chat
mutex in Client::chat_locks so only one operation mutates message_retry_counts
for a given cache_key at a time. Specifically, in increment_retry_count (and in
the sender-count preseed and spawn_retry_receipt code paths referenced), acquire
the chat lock for the chat/queue key from Client::chat_locks before calling
get/insert on message_retry_counts, perform the read/conditional-update while
holding the lock, then release it; ensure the lock key matches the same chat
identifier used by other retry-related routines so updates are properly
serialized.
In `@src/pdo.rs`:
- Around line 84-101: The current get-then-insert on self.pdo_pending_requests
allows a race where two coroutines both see missing entries and both insert;
make the check-and-insert atomic by using the cache's atomic API or by guarding
this section with a lock: for example, replace the get()+insert() sequence
around cache_key with an atomic insert_if_absent/entry or use a dedicated async
Mutex/RwLock keyed by cache_key to perform the "if absent then insert
PendingPdoRequest { message_info: info.clone(), requested_at: Instant::now() }"
in one critical section so only one request is created for the same message;
keep the debug return Ok(()) path when the insert fails because an entry already
existed.
In `@src/portable_cache.rs`:
- Around line 261-266: The invalidate_all() method currently uses
self.inner.try_write() and can silently no-op under contention; change it to
acquire a deterministic write lock (e.g., use self.inner.write() or loop until
try_write succeeds) so the clear is guaranteed to run. Locate invalidate_all and
the inner RwLock/lock field (self.inner, guard.map, guard.insertion_order) and
replace the non-blocking try_write with a blocking write acquisition or a
retrying strategy so the map and insertion_order are always cleared before
returning.
- Around line 209-229: The current insertion path ignores a configured zero
capacity because the eviction loop is skipped when cap == 0 but insertion still
proceeds; update the logic around self.max_capacity to explicitly treat Some(0)
as "no caching" by returning early (or skipping the insertion) when cap == 0.
Concretely, inside the block that reads self.max_capacity, check for cap == 0
and exit before mutating guard.insertion_order or guard.map; keep the eviction
loop for cap > 0 and only push to guard.insertion_order and insert into
guard.map when caching is enabled. Ensure you reference and modify the existing
symbols guard.insertion_order, guard.map, and self.max_capacity so the behavior
for max_capacity == Some(0) is correctly honored.
- Around line 279-301: get_with and get_with_by_ref perform get / init.await /
insert as separate steps which allows concurrent misses for the same key to run
the initializer concurrently and insert different values; change these helpers
to perform an atomic "get-or-init" so only one initializer runs per key (e.g.,
use the cache's entry API or a write lock to check-and-insert in one critical
section, or store a shared in-progress placeholder/future so other callers await
the same initializer), ensuring you call the existing get/insert symbols (get,
insert) only within that protected/atomic path and for get_with_by_ref clone the
key exactly when inserting (insert(key.clone(), ...)) so the semantics remain
correct.
In `@src/retry.rs`:
- Around line 128-137: The cleanup currently uses scopeguard::guard with a
synchronous try_lock on pending_retries which can lose the race and never remove
dedupe_key; change the cleanup to perform an awaited removal by spawning an
async task that acquires self.pending_retries.lock().await and then calls
remove(&key) (or otherwise ensure the lock is awaited) instead of using try_lock
in the scopeguard closure — i.e., replace the try_lock/remove logic in the
scopeguard closure with tokio::spawn(async move { let mut set =
client.pending_retries.lock().await; set.remove(&key); }) (or an equivalent
awaited cleanup) so the dedupe_key is reliably removed.
In `@src/runtime_impl.rs`:
- Around line 11-31: TokioRuntime must hold a tokio::runtime::Handle to avoid
panics when calling runtime APIs from non-entered contexts: add a Handle field
to the TokioRuntime struct and provide a constructor (e.g.,
TokioRuntime::new(handle: tokio::runtime::Handle) and/or
TokioRuntime::try_current() that uses Handle::try_current()). Replace direct
calls to tokio::spawn and tokio::task::spawn_blocking with handle.spawn(...) and
handle.spawn_blocking(...), and ensure any use of tokio::time::sleep remains
compatible (constructing the Sleep future is fine but keep the Handle for
spawn/spawn_blocking); update method implementations for spawn, spawn_blocking,
and sleep to use the stored Handle so the wrapper is self-contained and safe
from runtime-entry panics.
In `@wacore/src/appstate_sync.rs`:
- Around line 72-79: prefetch_keys currently swallows errors from
get_app_state_key causing partial application; change prefetch_keys (the async
fn prefetch_keys(&self, pl: &PatchList)) to propagate failures from
get_app_state_key instead of ignoring them (e.g., use the ? operator or
explicitly return Err when get_app_state_key(&key_id).await fails) so that
missing keys or backend read errors abort before any snapshot/patch is applied;
keep the existing key collection via collect_key_ids_from_patch_list and ensure
the function returns the error Result immediately when any get_app_state_key
call fails.
In `@wacore/src/message_processing.rs`:
- Around line 208-221: DecryptedMessageResult currently only stores
sender_key_distribution_message in its skdm field, so fast-ratchet sender-key
payloads held in fast_ratchet_key_sender_key_distribution_message are dropped
and fast-ratchet-only messages become protocol-only with no key material to
persist; update the code that builds/returns DecryptedMessageResult (the logic
around is_sender_key_distribution_only(),
fast_ratchet_key_sender_key_distribution_message, and
sender_key_distribution_message) so that DecryptedMessageResult.skdm carries the
fast-ratchet sender-key payload as well (e.g., normalize or merge
fast_ratchet_key_sender_key_distribution_message into the skdm field or add and
populate skdm from whichever of those two fields is present) and ensure
is_skdm_only tracks fast-ratchet-only cases accordingly.
- Line 137: The code currently assigns padding_version with silent truncation
via `let padding_version = enc_node.attrs().optional_u64("v").unwrap_or(2) as
u8;`; replace the direct `as u8` cast with `u8::try_from(...)` and explicitly
handle the Err case by rejecting the stanza (returning or propagating an error)
when the value is out of range, while preserving the default of 2 when the
attribute is absent; update the code paths around
`enc_node.attrs().optional_u64("v")` and the `padding_version` variable to use
the `try_from` result and return a clear error (e.g., InvalidPaddingVersion) on
conversion failure.
In `@wacore/src/protocol/keepalive.rs`:
- Around line 46-48: The timeout check in the keepalive logic uses a strict
greater-than which treats a socket as alive when elapsed == DEAD_SOCKET_TIME;
update the comparison in the ms_since(...).map(...) closure to use >= instead of
> so the dead-socket threshold is inclusive (change the expression using
ms_since, last_sent_ms and DEAD_SOCKET_TIME accordingly).
In `@wacore/src/store/in_memory.rs`:
- Around line 351-375: get_pn_mapping and pn_to_lid currently implement a
last-write-wins reverse index that ignores LidPnMappingEntry.updated_at, so
out-of-order replays can return older mappings; fix by making the reverse lookup
respect updated_at: either (A) change get_pn_mapping to ignore pn_to_lid and
instead scan s.lid_mappings for entries where entry.phone_number == phone and
return the entry with the greatest updated_at (use LidPnMappingEntry.updated_at
for ordering), or (B) keep pn_to_lid but update put_lid_mapping to only
overwrite s.pn_to_lid when the incoming entry.updated_at is >= the currently
indexed entry's updated_at (lookup existing via s.pn_to_lid -> lid ->
s.lid_mappings to compare timestamps) and otherwise skip replacing the reverse
index; reference functions/fields: get_pn_mapping, put_lid_mapping, pn_to_lid,
lid_mappings, LidPnMappingEntry.updated_at.
---
Outside diff comments:
In `@src/socket/noise_socket.rs`:
- Around line 152-158: The current code delays advancing write_counter until
after transport.send succeeds, which can cause nonce reuse if transport reports
Err; change the logic in the send path that builds and sends out_buf so that
*write_counter is incremented (using write_counter.wrapping_add(1)) immediately
after sealing the frame and before calling transport.send(out_buf). Keep the
same error mapping to EncryptSendError::transport for transport.send failures
but do not decrement or reuse the counter on error; update the section around
transport.send, write_counter, out_buf, and EncryptSendError::transport so the
counter is always advanced prior to the send attempt.
---
Duplicate comments:
In `@src/client.rs`:
- Around line 249-250: Client currently exposes pub(crate) runtime: Arc<dyn
Runtime> on the Client boundary which allows non-Tokio runtimes to be injected;
change Client construction so the public API does not accept or store a generic
Runtime: make Client::new / Client::new_* constructors create and embed a
TokioRuntime internally (e.g., use TokioRuntime within the Client implementation
and remove or make private any APIs that accept Arc<dyn Runtime>), ensure the
Client struct no longer exposes a runtime-agnostic field and that wacore-facing
code uses the concrete TokioRuntime for execution and SQLite/Diesel persistence
initialization to keep the main client Tokio-backed.
In `@src/keepalive.rs`:
- Around line 95-105: After calling self.shutdown_notifier.listen() in the
keepalive loop, immediately re-check the shutdown state by calling is_running()
(or equivalent) and break/return if it reports stopped to avoid a lost-shutdown
window; i.e., after creating the listener from shutdown_notifier.listen() but
before awaiting self.runtime.sleep(interval) (and before the futures::select!
sleep branch), call self.is_running() and exit if false so disconnect() cannot
be missed between listener creation and the sleep. Use the existing symbols
shutdown_notifier.listen(), is_running(), and the surrounding keepalive loop to
place this check.
---
Nitpick comments:
In `@transports/tokio-transport/src/lib.rs`:
- Around line 136-137: The async_trait attribute is conditionally applied for
wasm32 targets but this crate (tokio-transport) relies on Tokio and cannot
target wasm32; remove the redundant conditional attributes. Edit the attribute
usages of #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] /
#[cfg_attr(not(target_arch = "wasm32"), async_trait)] (and the duplicate
occurrences later) and replace them with a single plain #[async_trait] (or
remove the cfg_attr wrapper altogether) so the async_trait macro is consistently
applied for the Tokio-based transport; alternatively, if you intend to keep
wasm32 documentation, add a crate-level comment stating that tokio-transport
does not support wasm32 and that a separate transport is required for browser
targets.
In `@wacore/src/protocol/keepalive.rs`:
- Around line 63-67: The test helper ms_since_recent currently asserts elapsed <
100 which is flaky on CI; either relax the tolerance (e.g., increase to a few
hundred ms such as 500) or refactor ms_since_recent to accept an injected time
provider/value so tests can pass a deterministic now_ms; update the call site to
use crate::time::now_millis() (or a mock) and adjust the assertion on
ms_since(now_ms) accordingly to use the new tolerance or injected clock in the
ms_since_recent helper.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 1bfbdebc-9653-4ac9-85ce-ee32be8a6580
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (69)
Cargo.tomlhttp_clients/ureq-client/src/lib.rssrc/appstate_sync.rssrc/bot.rssrc/cache.rssrc/cache_config.rssrc/cache_store.rssrc/client.rssrc/client/context_impl.rssrc/client/sender_keys.rssrc/features/chat_actions.rssrc/features/profile.rssrc/features/tctoken.rssrc/handlers/basic.rssrc/handlers/chatstate.rssrc/handlers/ib.rssrc/handlers/iq.rssrc/handlers/message.rssrc/handlers/notification.rssrc/handlers/presence.rssrc/handlers/receipt.rssrc/handlers/traits.rssrc/handlers/unimplemented.rssrc/history_sync.rssrc/keepalive.rssrc/lib.rssrc/message.rssrc/pair.rssrc/pdo.rssrc/portable_cache.rssrc/receipt.rssrc/request.rssrc/retry.rssrc/runtime_impl.rssrc/send.rssrc/socket/noise_socket.rssrc/store/signal.rssrc/store/signal_adapter.rssrc/transport.rssrc/types/enc_handler.rssrc/unified_session.rssrc/usync.rssrc/version.rsstorages/sqlite-storage/src/sqlite_store.rstransports/tokio-transport/src/lib.rswacore/Cargo.tomlwacore/libsignal/benches/libsignal_benchmark.rswacore/libsignal/src/store/mod.rswacore/src/appstate_sync.rswacore/src/client/context.rswacore/src/ib.rswacore/src/iq/tctoken.rswacore/src/lib.rswacore/src/message_processing.rswacore/src/messages.rswacore/src/net.rswacore/src/protocol/keepalive.rswacore/src/request.rswacore/src/runtime.rswacore/src/send.rswacore/src/stanza/notification.rswacore/src/store/cache.rswacore/src/store/commands.rswacore/src/store/in_memory.rswacore/src/store/persistence.rswacore/src/store/traits.rswacore/src/time.rswacore/src/types/lid_pn.rswacore/src/types/message.rs
✅ Files skipped from review due to trivial changes (3)
- wacore/src/types/message.rs
- src/features/profile.rs
- src/store/signal_adapter.rs
🚧 Files skipped from review as they are similar to previous changes (10)
- src/types/enc_handler.rs
- wacore/Cargo.toml
- wacore/src/lib.rs
- src/unified_session.rs
- src/send.rs
- src/request.rs
- src/store/signal.rs
- src/handlers/message.rs
- wacore/src/store/persistence.rs
- Cargo.toml
| impl BotHandle { | ||
| /// Abort the bot's run task. | ||
| pub fn abort(&self) { | ||
| self._abort_handle.abort(); | ||
| } |
There was a problem hiding this comment.
Track every task started by run() under BotHandle.
run() detaches the sync worker and pair-code task, but BotHandle::abort() and done_rx only cover client_for_run.run(). After callers cancel or await the handle, those detached tasks can still sit on receiver.recv() / wait_for_socket() and keep operating on the same Client, so shutdown is only partial.
Also applies to: 167-200, 209-250
| impl<B, T, H> BotBuilder<B, T, H, Missing> { | ||
| /// Set the async runtime implementation to use. | ||
| /// | ||
| /// This is required to build a bot. | ||
| pub fn with_runtime<Rt: Runtime>(self, runtime: Rt) -> BotBuilder<B, T, H, Provided> { | ||
| BotBuilder { | ||
| backend: self.backend, | ||
| transport_factory: self.transport_factory, | ||
| http_client: self.http_client, | ||
| runtime: Some(Arc::new(runtime)), | ||
| event_handler: self.event_handler, |
There was a problem hiding this comment.
Keep src::BotBuilder Tokio-only.
The public with_runtime<Rt: Runtime> hook makes arbitrary runtimes part of the supported API of the top-level crate. The runtime abstraction belongs in wacore, but src/** is still supposed to stay Tokio-backed; I'd default/inject TokioRuntime here and keep custom runtimes internal to wacore or tests.
Based on learnings, whatsapp-rust/src/**/*.rs main client must use Tokio runtime and SQLite persistence with Diesel ORM.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/bot.rs` around lines 415 - 425, The public with_runtime<Rt: Runtime> on
BotBuilder exposes arbitrary runtimes; keep the top-level crate Tokio-only by
removing or privatizing the generic runtime hook and defaulting/injecting a
Tokio runtime instead. Replace the public signature of BotBuilder::with_runtime
(or make it crate-private) so it no longer accepts a generic Rt: Runtime; ensure
BotBuilder::runtime is initialized with a TokioRuntime (or
Arc::new(TokioRuntime)) in the public builder path and move any generic Runtime
trait factory/with_runtime functionality into the internal wacore crate or test
helpers (keep references to BotBuilder, with_runtime, Runtime, and TokioRuntime
to locate and update the code).
| pub mod runtime_impl; | ||
| #[cfg(not(target_arch = "wasm32"))] | ||
| pub use runtime_impl::TokioRuntime; | ||
| pub use wacore::runtime::Runtime; |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Keep the main crate's public runtime API Tokio-only.
Re-exporting wacore::runtime::Runtime from the crate root makes whatsapp-rust look runtime-pluggable, but the repo guidance says the main client should stay on Tokio. I'd keep this trait internal and only expose the Tokio-backed surface from src.
Based on learnings: whatsapp-rust main client must use Tokio runtime and SQLite persistence with Diesel ORM.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib.rs` at line 36, Re-exporting wacore::runtime::Runtime at the crate
root exposes a runtime-pluggable API; remove the public re-export (pub use
wacore::runtime::Runtime) so the Runtime trait stays internal, and instead
expose only the Tokio-backed surface from src (create or re-export a
Tokio-specific runtime type or factory in src, e.g., TokioRuntime or functions
that construct the Tokio runtime used by the main client), ensuring the public
API and main client remain Tokio-only and continue to use SQLite/Diesel
persistence as implemented.
| /// Increments the retry count for a message and returns the new count. | ||
| /// Returns `None` if max retries have been reached. | ||
| /// | ||
| /// Uses moka's `and_compute_with` for truly atomic read-modify-write operations, | ||
| /// preventing race conditions where concurrent calls could exceed MAX_DECRYPT_RETRIES. | ||
| /// Uses get + insert for portability across cache backends. | ||
| async fn increment_retry_count(&self, cache_key: &str) -> Option<u8> { | ||
| use moka::ops::compute::Op; | ||
|
|
||
| let result = self | ||
| .message_retry_counts | ||
| .entry_by_ref(cache_key) | ||
| .and_compute_with(|maybe_entry| { | ||
| let op = if let Some(entry) = maybe_entry { | ||
| let current = entry.into_value(); | ||
| if current >= MAX_DECRYPT_RETRIES { | ||
| // Max retries reached, don't increment | ||
| Op::Nop | ||
| } else { | ||
| Op::Put(current + 1) | ||
| } | ||
| } else { | ||
| Op::Put(1_u8) | ||
| }; | ||
| std::future::ready(op) | ||
| }) | ||
| .await; | ||
|
|
||
| match result { | ||
| moka::ops::compute::CompResult::Inserted(entry) => Some(entry.into_value()), | ||
| moka::ops::compute::CompResult::ReplacedWith(entry) => Some(entry.into_value()), | ||
| moka::ops::compute::CompResult::Unchanged(_) => None, // Max retries reached | ||
| moka::ops::compute::CompResult::StillNone(_) => None, | ||
| moka::ops::compute::CompResult::Removed(_) => None, | ||
| let current = self.message_retry_counts.get(&cache_key.to_string()).await; | ||
| match current { | ||
| Some(count) if count >= MAX_DECRYPT_RETRIES => None, | ||
| Some(count) => { | ||
| let new_count = count + 1; | ||
| self.message_retry_counts | ||
| .insert(cache_key.to_string(), new_count) | ||
| .await; | ||
| Some(new_count) | ||
| } | ||
| None => { | ||
| self.message_retry_counts | ||
| .insert(cache_key.to_string(), 1_u8) | ||
| .await; | ||
| Some(1) | ||
| } | ||
| } |
There was a problem hiding this comment.
Serialize retry-count updates.
increment_retry_count() and the sender-count preseed both switched to get + insert, while spawn_retry_receipt() still runs on detached tasks. Two failures for the same message can now reuse the same retry number or overwrite a higher cached sender count with a lower one, which weakens MAX_DECRYPT_RETRIES and can re-enable retry loops.
As per coding guidelines, "Use Client::chat_locks to serialize per-chat operations for concurrency safety".
Also applies to: 188-243, 483-488
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/message.rs` around lines 125 - 146, The increment_retry_count logic uses
get+insert which is not serialized and can race with detached tasks (e.g.,
spawn_retry_receipt) or the sender-count preseed; wrap the per-chat retry
updates using the per-chat mutex in Client::chat_locks so only one operation
mutates message_retry_counts for a given cache_key at a time. Specifically, in
increment_retry_count (and in the sender-count preseed and spawn_retry_receipt
code paths referenced), acquire the chat lock for the chat/queue key from
Client::chat_locks before calling get/insert on message_retry_counts, perform
the read/conditional-update while holding the lock, then release it; ensure the
lock key matches the same chat identifier used by other retry-related routines
so updates are properly serialized.
| /// Pre-fetch and cache all keys needed for a patch list. | ||
| async fn prefetch_keys(&self, pl: &PatchList) -> Result<()> { | ||
| let key_ids = collect_key_ids_from_patch_list(pl.snapshot.as_ref(), &pl.patches); | ||
| for key_id in key_ids { | ||
| // This will fetch and cache if not already cached | ||
| let _ = self.get_app_state_key(&key_id).await; | ||
| } | ||
| Ok(()) |
There was a problem hiding this comment.
Fail before applying any snapshot/patch state when a key is missing.
prefetch_keys() ignores get_app_state_key() errors, so a missing key or backend read failure can be discovered only after earlier snapshot/patch work has already been processed and persisted. That turns a collection that should fail atomically into a partially-applied one.
💡 Minimal fix
async fn prefetch_keys(&self, pl: &PatchList) -> Result<()> {
let key_ids = collect_key_ids_from_patch_list(pl.snapshot.as_ref(), &pl.patches);
for key_id in key_ids {
- // This will fetch and cache if not already cached
- let _ = self.get_app_state_key(&key_id).await;
+ self.get_app_state_key(&key_id).await?;
}
Ok(())
}📝 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.
| /// Pre-fetch and cache all keys needed for a patch list. | |
| async fn prefetch_keys(&self, pl: &PatchList) -> Result<()> { | |
| let key_ids = collect_key_ids_from_patch_list(pl.snapshot.as_ref(), &pl.patches); | |
| for key_id in key_ids { | |
| // This will fetch and cache if not already cached | |
| let _ = self.get_app_state_key(&key_id).await; | |
| } | |
| Ok(()) | |
| /// Pre-fetch and cache all keys needed for a patch list. | |
| async fn prefetch_keys(&self, pl: &PatchList) -> Result<()> { | |
| let key_ids = collect_key_ids_from_patch_list(pl.snapshot.as_ref(), &pl.patches); | |
| for key_id in key_ids { | |
| self.get_app_state_key(&key_id).await?; | |
| } | |
| Ok(()) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@wacore/src/appstate_sync.rs` around lines 72 - 79, prefetch_keys currently
swallows errors from get_app_state_key causing partial application; change
prefetch_keys (the async fn prefetch_keys(&self, pl: &PatchList)) to propagate
failures from get_app_state_key instead of ignoring them (e.g., use the ?
operator or explicitly return Err when get_app_state_key(&key_id).await fails)
so that missing keys or backend read errors abort before any snapshot/patch is
applied; keep the existing key collection via collect_key_ids_from_patch_list
and ensure the function returns the error Result immediately when any
get_app_state_key call fails.
| } | ||
| }; | ||
|
|
||
| let padding_version = enc_node.attrs().optional_u64("v").unwrap_or(2) as u8; |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n wacore/src/message_processing.rs | sed -n '130,145p'Repository: jlucaso1/whatsapp-rust
Length of output: 784
🏁 Script executed:
cat -n wacore/src/message_processing.rs | sed -n '1,50p'Repository: jlucaso1/whatsapp-rust
Length of output: 2098
🌐 Web query:
Rust u64 as u8 casting truncation behavior untrusted input
💡 Result:
In Rust, casting a u64 to u8 using as performs truncation by retaining only the 8 least significant bits (LSB) of the u64 value, discarding the higher bits. This behavior is well-defined, follows C-like conventions for integer casts, and is safe (no panic or UB). For example: - 1000u64 as u8 == 232 (since 1000 in binary is ...1111101000, LSB 8 bits are 11101000 == 232) - This is equivalent to value % 256 for unsigned values within reasonable ranges. This truncation is deterministic and platform-independent for integer-to-smaller-integer casts. For untrusted input (e.g., potentially large u64 values from external sources), the cast silently truncates higher bits without error. This is expected behavior but can lead to logical errors if higher bits are significant (e.g., index out of bounds if assuming full u64 range fits in u8). Use checked methods like u64::checked_cast_to:: (nightly) or manual bounds checking (e.g., if value > u8::MAX { handle_error } else { value as u8 }) for safety. Official sources confirm: - Rust By Example: "the first 8 least significant bits (LSB) are kept, while the rest ... get truncated." - Rust Reference: Numeric casts truncate bit representation.
Citations:
- 1: https://doc.rust-lang.org/rust-by-example/types/cast.html
- 2: https://doc.rust-lang.org/stable/reference/expressions/operator-expr.html?highlight=overflow
- 3: https://doc.rust-lang.org/beta/rust-by-example/types/cast.html
- 4: https://doc.rust-lang.org/1.39.0/nomicon/casts.html
- 5: https://doc.rust-lang.org/reference/expressions/operator-expr.html?highlight=casting
Use u8::try_from() instead of silent truncation for the padding version attribute.
Line 137 casts an untrusted u64 attribute directly to u8 with as u8, which silently truncates oversized values (e.g., v="258" becomes 2). For malformed stanza data, reject invalid ranges explicitly using u8::try_from() instead of masking the error through truncation.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@wacore/src/message_processing.rs` at line 137, The code currently assigns
padding_version with silent truncation via `let padding_version =
enc_node.attrs().optional_u64("v").unwrap_or(2) as u8;`; replace the direct `as
u8` cast with `u8::try_from(...)` and explicitly handle the Err case by
rejecting the stanza (returning or propagating an error) when the value is out
of range, while preserving the default of 2 when the attribute is absent; update
the code paths around `enc_node.attrs().optional_u64("v")` and the
`padding_version` variable to use the `try_from` result and return a clear error
(e.g., InvalidPaddingVersion) on conversion failure.
| pub struct DecryptedMessageResult { | ||
| /// The user-visible message content (with DeviceSentMessage unwrapped). | ||
| pub message: wa::Message, | ||
| /// The sender key distribution message, if present. | ||
| /// Must be processed to store the sender key for future group decryption. | ||
| pub skdm: Option<wa::message::SenderKeyDistributionMessage>, | ||
| /// Protocol-level messages that require special handling. | ||
| pub protocol_message: Option<ProtocolMessageInfo>, | ||
| /// True if the message contains only SKDM with no user-visible content. | ||
| /// These should not be surfaced as user events. | ||
| pub is_skdm_only: bool, | ||
| /// True if a DeviceSentMessage wrapper was present but the sender was | ||
| /// not "from me" (protocol violation — should be logged as a warning). | ||
| pub has_invalid_dsm: bool, |
There was a problem hiding this comment.
Fast-ratchet sender-key payloads are dropped here.
is_sender_key_distribution_only() already treats fast_ratchet_key_sender_key_distribution_message as SKDM, but DecryptedMessageResult.skdm only carries sender_key_distribution_message. A fast-ratchet-only message will therefore be classified as protocol-only while exposing no key material to persist for later group decrypts.
Also applies to: 249-269
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@wacore/src/message_processing.rs` around lines 208 - 221,
DecryptedMessageResult currently only stores sender_key_distribution_message in
its skdm field, so fast-ratchet sender-key payloads held in
fast_ratchet_key_sender_key_distribution_message are dropped and
fast-ratchet-only messages become protocol-only with no key material to persist;
update the code that builds/returns DecryptedMessageResult (the logic around
is_sender_key_distribution_only(),
fast_ratchet_key_sender_key_distribution_message, and
sender_key_distribution_message) so that DecryptedMessageResult.skdm carries the
fast-ratchet sender-key payload as well (e.g., normalize or merge
fast_ratchet_key_sender_key_distribution_message into the skdm field or add and
populate skdm from whichever of those two fields is present) and ensure
is_skdm_only tracks fast-ratchet-only cases accordingly.
| ms_since(last_sent_ms) | ||
| .map(|elapsed| elapsed > DEAD_SOCKET_TIME.as_millis() as u64) | ||
| .unwrap_or(false) |
There was a problem hiding this comment.
Use inclusive timeout check for dead-socket threshold.
Line 47 uses >; at exactly DEAD_SOCKET_TIME the socket is still treated as alive for one extra tick. Use >= to match the documented timeout boundary.
Proposed fix
- ms_since(last_sent_ms)
- .map(|elapsed| elapsed > DEAD_SOCKET_TIME.as_millis() as u64)
- .unwrap_or(false)
+ ms_since(last_sent_ms)
+ .map(|elapsed| elapsed >= DEAD_SOCKET_TIME.as_millis() as u64)
+ .unwrap_or(false)📝 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.
| ms_since(last_sent_ms) | |
| .map(|elapsed| elapsed > DEAD_SOCKET_TIME.as_millis() as u64) | |
| .unwrap_or(false) | |
| ms_since(last_sent_ms) | |
| .map(|elapsed| elapsed >= DEAD_SOCKET_TIME.as_millis() as u64) | |
| .unwrap_or(false) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@wacore/src/protocol/keepalive.rs` around lines 46 - 48, The timeout check in
the keepalive logic uses a strict greater-than which treats a socket as alive
when elapsed == DEAD_SOCKET_TIME; update the comparison in the
ms_since(...).map(...) closure to use >= instead of > so the dead-socket
threshold is inclusive (change the expression using ms_since, last_sent_ms and
DEAD_SOCKET_TIME accordingly).
| async fn get_pn_mapping(&self, phone: &str) -> Result<Option<LidPnMappingEntry>> { | ||
| let s = self.state.lock().await; | ||
| let entry = s | ||
| .pn_to_lid | ||
| .get(phone) | ||
| .and_then(|lid| s.lid_mappings.get(lid)) | ||
| .cloned(); | ||
| Ok(entry) | ||
| } | ||
|
|
||
| async fn put_lid_mapping(&self, entry: &LidPnMappingEntry) -> Result<()> { | ||
| let mut s = self.state.lock().await; | ||
| // Remove stale reverse entry if the LID was previously mapped to a different phone number | ||
| if let Some(old_phone) = s | ||
| .lid_mappings | ||
| .get(&entry.lid) | ||
| .filter(|old| old.phone_number != entry.phone_number) | ||
| .map(|old| old.phone_number.clone()) | ||
| { | ||
| s.pn_to_lid.remove(&old_phone); | ||
| } | ||
| s.pn_to_lid | ||
| .insert(entry.phone_number.clone(), entry.lid.clone()); | ||
| s.lid_mappings.insert(entry.lid.clone(), entry.clone()); | ||
| Ok(()) |
There was a problem hiding this comment.
PN reverse lookups still ignore updated_at.
pn_to_lid makes the last write win. If mappings are replayed out of order, get_pn_mapping() can return an older LID even though the trait docs say "the most recent" mapping and the SQLite backend resolves this with ORDER BY updated_at DESC. Either pick the max-updated_at entry on lookup or only overwrite the reverse index when the incoming record is newer.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@wacore/src/store/in_memory.rs` around lines 351 - 375, get_pn_mapping and
pn_to_lid currently implement a last-write-wins reverse index that ignores
LidPnMappingEntry.updated_at, so out-of-order replays can return older mappings;
fix by making the reverse lookup respect updated_at: either (A) change
get_pn_mapping to ignore pn_to_lid and instead scan s.lid_mappings for entries
where entry.phone_number == phone and return the entry with the greatest
updated_at (use LidPnMappingEntry.updated_at for ordering), or (B) keep
pn_to_lid but update put_lid_mapping to only overwrite s.pn_to_lid when the
incoming entry.updated_at is >= the currently indexed entry's updated_at (lookup
existing via s.pn_to_lid -> lid -> s.lid_mappings to compare timestamps) and
otherwise skip replacing the reverse index; reference functions/fields:
get_pn_mapping, put_lid_mapping, pn_to_lid, lid_mappings,
LidPnMappingEntry.updated_at.
There was a problem hiding this comment.
♻️ Duplicate comments (1)
src/lib.rs (1)
33-37:⚠️ Potential issue | 🟠 MajorKeep the crate-root runtime surface Tokio-only.
Line 34 and Line 37 currently expose both the Tokio implementation module and the generic
Runtimetrait from the main crate. That makeswhatsapp-rustlook runtime-pluggable even though this crate is supposed to stay Tokio-based. Keep the trait inwacore, makeruntime_implprivate, and only re-exportTokioRuntimehere.Based on learnings: whatsapp-rust main client must use Tokio runtime and SQLite persistence with Diesel ORM.
Proposed API narrowing
#[cfg(feature = "tokio-runtime")] -pub mod runtime_impl; +mod runtime_impl; #[cfg(feature = "tokio-runtime")] pub use runtime_impl::TokioRuntime; -pub use wacore::runtime::Runtime;🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib.rs` around lines 33 - 37, The crate currently exposes the internal runtime module and re-exports the generic Runtime trait; make the runtime implementation private and stop re-exporting the trait so the crate surface is Tokio-only: change the module declaration from `pub mod runtime_impl` to a private `mod runtime_impl` (keep the `#[cfg(feature = "tokio-runtime")]`), keep the `pub use runtime_impl::TokioRuntime;` export, and remove the `pub use wacore::runtime::Runtime;` re-export so only `TokioRuntime` is publicly exposed while the `Runtime` trait stays in `wacore`.
🤖 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/lib.rs`:
- Around line 33-37: The crate currently exposes the internal runtime module and
re-exports the generic Runtime trait; make the runtime implementation private
and stop re-exporting the trait so the crate surface is Tokio-only: change the
module declaration from `pub mod runtime_impl` to a private `mod runtime_impl`
(keep the `#[cfg(feature = "tokio-runtime")]`), keep the `pub use
runtime_impl::TokioRuntime;` export, and remove the `pub use
wacore::runtime::Runtime;` re-export so only `TokioRuntime` is publicly exposed
while the `Runtime` trait stays in `wacore`.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5a287cfb-d567-43c5-b8ce-2299afced263
📒 Files selected for processing (3)
Cargo.tomlsrc/lib.rswacore/Cargo.toml
✅ Files skipped from review due to trivial changes (1)
- wacore/Cargo.toml
|
Working hard on |
|
@claude review |
|
Claude finished @jlucaso1's task in 3m 3s —— View job PR Review: refactor: runtime agnostic
Third review pass. Previous reviews covered the main architecture well — this focuses on what's been fixed, what hasn't, and new issues introduced in recent commits. Previously reported issues — now fixedThese were flagged in reviews 1–2 and are confirmed resolved:
Issues still open from prior reviews1. This is the most critical remaining issue. The
Moka's 2. Uses 3. The 4. The 5.
New observations6. When 7. Noise write counter — current approach is correct — The counter is now advanced after successful What's done well
Priority summary
Issues 1-3 are functional correctness bugs that can cause races or stuck state in production. Issue 4 risks silent data loss. Issues 5-6 are edge cases. |
|
@claude review |
|
Claude finished @jlucaso1's task in 2m 34s —— View job PR Review: refactor: runtime agnostic
Fourth review pass. This focuses on what's changed since the last three reviews and assesses remaining issues. Previously reported issues — status
Remaining issues1. This was flagged in reviews 2 and 3 and remains unfixed. The Consider spawning an async cleanup task instead: let client_clone = self.clone();
let key_clone = dedupe_key.clone();
scopeguard::defer! {
client_clone.runtime.spawn(Box::pin(async move {
client_clone.pending_retries.lock().await.remove(&key_clone);
})).detach();
}2. The 3. Bot detached tasks not tracked by
Design feedback (non-blocking)4. Noise counter: current approach is correct There was disagreement between CodeRabbit reviews about whether to advance the counter before or after 5. The per-key init lock approach at 6. The 64-iteration spin with fallback warning (line 284-298) is a reasonable compromise for a sync method. The critical sections are short HashMap ops, so 64 spins should always suffice in practice. The What's done well
Priority summary
The PR is in very good shape after addressing feedback across multiple review rounds. Issue #1 is the only remaining functional correctness bug. Issues #2 and #3 are documented limitations that can be addressed in follow-up work. |
|
@claude review |
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
|
Claude finished @jlucaso1's task in 2m 31s —— View job PR Review: refactor: runtime agnostic
Fifth review pass. This focuses on what changed in the latest commits ( Previously reported issues — all addressedEvery significant issue from reviews 1–4 has been fixed. Here's the final status:
Remaining observations (non-blocking)1. Bot detached tasks not tracked by
2. The 64-iteration spin with 3. The New code review: media reupload feature
What's done well
VerdictThis PR is in excellent shape. All previously reported functional correctness bugs have been fixed. The remaining observations are minor edge cases and documentation concerns that can be addressed in follow-up work. The runtime abstraction is clean, minimal, and well-tested. |
`#[serde(skip)]` on `Device.account` (added in #393) caused `AdvSignedDeviceIdentity` to be lost on restart for custom stores using serde. Without it, `<device-identity>` is missing from outgoing stanzas and recipients silently drop pkmsg from unverifiable companion devices. Replace with a custom serde module that encodes as protobuf bytes. Shared helpers used by both serde and the SQLite store (DRY).
2 participants
Summary
Major refactor to make
whatsapp-rustruntime-agnostic. The core protocol logic (wacore) no longer depends on Tokio, enabling alternative runtimes (e.g., for WASM/embedded targets). Tokio remains the default runtime via thetokio-runtimefeature flag.What changed
New runtime abstraction (
wacore::runtime)Runtimetrait withspawn,spawn_blocking,sleepmethodsTokioRuntimeimplementation behindtokio-runtimefeature flagAbortHandleabstraction for task cancellationblocking()function for runtime-agnosticspawn_blockingCrate-level changes
Runtimetrait. Addedwacore::time::now_millis()for portable timestamps.Client,Bot, and all subsystems acceptArc<dyn Runtime>. Builders require.with_runtime().Concurrency primitives replaced
tokio::sync::Notify→event_listener::Event(runtime-agnostic)tokio::sync::Mutex/RwLock→async_lock::Mutex/RwLocktokio::time::sleep→Runtime::sleep()tokio::task::spawn→Runtime::spawn()tokio::task::spawn_blocking→Runtime::spawn_blocking()std::time::Instant→wacore::time::now_millis()(WASM-compatible)Portable cache (
src/portable_cache.rs)PortableCache<K, V>that replacesmoka::future::Cachewhenmoka-cachefeature is disabledwacore::time::now_millis()Cachefor drop-in replacementmoka-cachefeature (moka pulls in Tokio internals)In-memory store (
wacore::store::in_memory)InMemoryBackendimplementing theBackendtraitOther changes
dashmapdependency removed (replaced byasync_lock::RwLock<HashMap>)BotHandlenow usesAbortHandlefor explicit cancellationWeak<Self>to avoid Arc leakssave_to_disk()restores dirty flag on save failure (prevents data loss)is_running(closes race window)src/features/media_reupload.rs)flush()method added to persistence managerBreaking Changes
1.
BotBuilderrequires.with_runtime()Before:
After:
2.
Client::new()requiresruntimeparameterBefore:
After:
3.
Runtimetrait re-exported from crate root4.
Bot::run()returnsBotHandleinstead ofJoinHandleBefore:
After:
Important: Dropping
BotHandleaborts the run task. If you want the bot to keep running, you must hold onto the handle.5. Cache type changes (when
moka-cachefeature is disabled)If you disable
moka-cache, caches usePortableCacheinstead ofmoka::future::Cache. The API is compatible but:get_with()/get_with_by_ref()are NOT single-flight (concurrent misses may run init twice)invalidate_all()is best-effort (usestry_write())6.
PersistenceManager::run_background_saver()requiresruntimeparameterBefore:
After:
Known Limitations
PersistenceManagerbackground saver does not perform a final save when dropped. Dirty state that hasn't been flushed within the interval (30s) will be lost. A propershutdown()method is planned separately.ureq-clientHTTP backend uses blocking I/O andtokio::task::spawn_blocking— it won't work on wasm32 targets. A wasm-compatible HTTP client is needed for full wasm support.get_withrace: Whenmoka-cacheis disabled, concurrent cache misses for the same key can run the initializer multiple times. This affects session lock creation — a single-flight mechanism is planned.Test Plan
cargo fmt && cargo clippy --all-targetspassescargo test --allpassesTokioRuntimeinjectedBotHandlecancellation works (abort + drop)Summary by CodeRabbit
Release Notes
New Features
Breaking Changes
Bot::builder()now requires.with_runtime()configurationBot::run()returnsBotHandleinstead ofJoinHandleClient::new()signature updated to accept runtime parameterImprovements