feat: add awaited pre-ACK message hook - #919
Conversation
📝 WalkthroughSummary by CodeRabbit
WalkthroughThe PR adds a parsed-message pre-ACK hook, pending replay state, and updated inbound ACK/dispatch handling. Client initialization, message classification, session/group processing, newsletter handling, and tests are updated to use the new hook-driven flow. ChangesParsed-message pre-ACK pipeline
Sequence Diagram(s)sequenceDiagram
participant Client
participant ParsedMessagePreAckHook
participant PendingParsedMessagePreAck
participant NodeIO
participant EventMessage as Event::Message
Client->>ParsedMessagePreAckHook: ParsedMessagePreAckContext
alt hook succeeds
ParsedMessagePreAckHook-->>Client: Ok(())
Client->>NodeIO: ack_parsed_message_after_pre_ack
Client->>EventMessage: dispatch Event::Message
else hook fails
ParsedMessagePreAckHook-->>Client: Err(...)
Client->>PendingParsedMessagePreAck: store_pending_pre_ack_message
end
loop duplicate redelivery
Client->>PendingParsedMessagePreAck: load pending pre-ACK message
Client->>ParsedMessagePreAckHook: ParsedMessagePreAckContext
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~90+ minutes Possibly related PRs
Suggested labels
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/client/adapters.rs`:
- Around line 68-70: The deferred flush path in flush_signal_cache_logged is
treating expected backpressure as an error, which causes noisy error-level
logging. Update the signal-cache flush flow so the pending pre-ACK commit case
is represented as a distinct deferred outcome from the core flush_signal_cache
logic, then have the _logged wrapper downgrade that case to debug or trace
instead of error, or make callers special-case it using the existing
anyhow::anyhow return point as the location to distinguish the status.
- Around line 62-71: The global pending-commit gate in flush_signal_cache is too
strict because a single stuck parsed-message pre-ACK commit can block all Signal
persistence forever, including cleanup_connection_state. Update the logic around
pending_parsed_message_pre_ack_count and parsed_message_pre_ack_hook so there is
an escape hatch such as a max pending age or retry limit; once exceeded, abandon
the pending entry, allow the cache flush to proceed, and leave the message for
redelivery. Keep the fix localized to the flush_signal_cache path and any helper
state that tracks pending pre-ACK commits.
In `@src/client/lifecycle.rs`:
- Around line 195-201: The pending_parsed_message_pre_ack cache in lifecycle
state is unbounded and never drained, so add a lifecycle path that removes stuck
entries or caps growth. Review pending_parsed_message_pre_ack,
pending_parsed_message_pre_ack_count, cleanup_connection_state, and
flush_signal_cache to ensure failed pre-ACK items are eventually evicted or
accounted for without relying on TTL, and make sure reconnect/cleanup logic
clears or bounds this cache like the other per-connection caches.
In `@src/client/node_io.rs`:
- Around line 24-32: The receive path in NodeIO is forced to a single in-flight
message whenever a parsed-message pre-ACK hook is installed, which serializes
the whole pipeline for the client’s lifetime. Update the public documentation
for set_parsed_message_pre_ack_hook to clearly call out this throughput cost and
that a heavy hook will back-pressure all inbound messages. Mention the behavior
change explicitly near the NodeIo / parsed_message_pre_ack_hook API so consumers
understand the tradeoff before enabling it.
In `@src/message/receive.rs`:
- Around line 690-695: `process_classified_message` and the payload-batch loops
are continuing after `pre_ack_hook_failed` becomes true, which can allow
additional ACK/nack/event emission in the same stanza. Update the branches that
merge `plaintext_outcome`/batch outcomes so they immediately return the
accumulated `outcome` as soon as `pre_ack_hook_failed` is set, and apply the
same early-exit pattern in the other affected batch-processing blocks mentioned
in the review.
- Around line 493-497: The fallback path in dispatch_pending_pre_ack_message
handling is using ack_received_message() for status messages, but when the
pre-ACK hook is installed that does not send the required transport ACK and can
leave status stanzas queued for redelivery. Update the None branch in the
receive flow to use the status post-pre-ACK helper instead of
ack_received_message(), and apply the same change in the other matching status
fallback site, keeping the logic centered around
dispatch_pending_pre_ack_message and the status ACK helper.
In `@src/message/tests.rs`:
- Around line 5388-5392: The test fixtures in Message tests are using phone-like
JIDs that could look real; replace these identifiers in the affected
create_test_message_info cases with clearly fictitious placeholder JIDs so no
plausible user numbers remain in test code. Update all new/related cases in
src/message/tests.rs, especially the create_test_message_info calls around the
listed PRE_ACK/related scenarios, to use obviously fake values consistently.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 3a418d95-3c7d-4f4b-9101-ef21368b250c
📒 Files selected for processing (12)
src/bot.rssrc/client.rssrc/client/accessors.rssrc/client/adapters.rssrc/client/lifecycle.rssrc/client/node_io.rssrc/lib.rssrc/message.rssrc/message/dispatch.rssrc/message/receive.rssrc/message/special.rssrc/message/tests.rs
| if self.parsed_message_pre_ack_hook.get().is_some() | ||
| && self | ||
| .pending_parsed_message_pre_ack_count | ||
| .load(Ordering::Acquire) | ||
| > 0 | ||
| { | ||
| return Err(anyhow::anyhow!( | ||
| "Signal cache flush deferred while parsed-message pre-ACK commit is pending" | ||
| )); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
This is the part that has to work right: one stuck commit can wedge Signal persistence for the entire client, forever.
The gate is global, not per-message: as long as pending_parsed_message_pre_ack_count > 0, every flush_signal_cache call returns Err — including the one in cleanup_connection_state, which then keeps (never clears) the signal cache. So if a consumer's durable commit fails permanently for a single message (disk full, DB down, a bug in their hook), the result is:
- Signal cache flushes are blocked for all chats, not just the stuck one.
- The in-memory signal cache grows unbounded and its advanced session/sender-key state is never persisted, surviving reconnects only in memory.
- Combined with the serialized receive lane, the whole pipeline degrades behind one poisoned entry.
There's no bound, timeout, or escape hatch here. Please add one — e.g. a max pending age/count after which the entry is abandoned (and the message left for redelivery) so a single bad commit can't brick Signal persistence indefinitely.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/client/adapters.rs` around lines 62 - 71, The global pending-commit gate
in flush_signal_cache is too strict because a single stuck parsed-message
pre-ACK commit can block all Signal persistence forever, including
cleanup_connection_state. Update the logic around
pending_parsed_message_pre_ack_count and parsed_message_pre_ack_hook so there is
an escape hatch such as a max pending age or retry limit; once exceeded, abandon
the pending entry, allow the cache flush to proceed, and leave the message for
redelivery. Keep the fix localized to the flush_signal_cache path and any helper
state that tracks pending pre-ACK commits.
| return Err(anyhow::anyhow!( | ||
| "Signal cache flush deferred while parsed-message pre-ACK commit is pending" | ||
| )); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Don't scream error! for what is normal, expected backpressure.
flush_signal_cache_logged logs this deferral at error!. But a deferred flush while a pre-ACK commit is pending is expected, routine operation — it'll fire on every deferred call and drown the logs with errors that aren't errors. Either return a distinguishable "deferred" outcome that the _logged wrapper downgrades to debug!/trace!, or have callers special-case it. As-is, operators will think something's broken when it's working as designed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/client/adapters.rs` around lines 68 - 70, The deferred flush path in
flush_signal_cache_logged is treating expected backpressure as an error, which
causes noisy error-level logging. Update the signal-cache flush flow so the
pending pre-ACK commit case is represented as a distinct deferred outcome from
the core flush_signal_cache logic, then have the _logged wrapper downgrade that
case to debug or trace instead of error, or make callers special-case it using
the existing anyhow::anyhow return point as the location to distinguish the
status.
| // Intentionally unbounded/no-TTL: an entry means Signal decrypt may | ||
| // have advanced volatile state for a message the application has | ||
| // not durably committed. Expiring it would allow a later Signal | ||
| // flush to make that advance durable and turn server redelivery | ||
| // into an ACKed duplicate without rerunning the hook. | ||
| pending_parsed_message_pre_ack: Cache::builder().build(), | ||
| pending_parsed_message_pre_ack_count: AtomicUsize::new(0), |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
The unbounded, never-cleared pending cache is the thing I'm worried about — let's make sure it can't grow forever.
Look, I get the design: an entry means Signal state advanced for an uncommitted message, so expiring it would be wrong. But this cache has no capacity bound and is never cleared in cleanup_connection_state (unlike chat_lanes, signal_cache, etc.). If a consumer's pre-ACK hook keeps failing for a message that never gets a successful redelivery, that entry — and the pending_parsed_message_pre_ack_count it bumps — stays around indefinitely. Across reconnects it just accumulates. The real-world fallout shows up in flush_signal_cache; I've left the detailed concern there. Please confirm there's a path that bounds this or drains stuck entries.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/client/lifecycle.rs` around lines 195 - 201, The
pending_parsed_message_pre_ack cache in lifecycle state is unbounded and never
drained, so add a lifecycle path that removes stuck entries or caps growth.
Review pending_parsed_message_pre_ack, pending_parsed_message_pre_ack_count,
cleanup_connection_state, and flush_signal_cache to ensure failed pre-ACK items
are eventually evicted or accounted for without relying on TTL, and make sure
reconnect/cleanup logic clears or bounds this cache like the other
per-connection caches.
| let permits = if self.parsed_message_pre_ack_hook.get().is_some() { | ||
| // A pre-ACK hook can intentionally fail after Signal decrypt has | ||
| // advanced volatile state. Keep message processing serial while the | ||
| // hook is installed so pending retry/flush deferral has one inbound | ||
| // message to reason about at a time. | ||
| 1 | ||
| } else { | ||
| permits | ||
| }; |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
This pins the whole receive pipeline to single-threaded for the client's entire lifetime — make sure consumers understand the cost.
Forcing permits = 1 whenever the hook is installed means every inbound message is processed strictly serially forever, not just during the pending-retry window. The reasoning is sound (you need one message to reason about at a time for the flush-deferral contract), but a slow durable-commit hook now serializes and back-pressures the entire receive lane. That's a real throughput ceiling. Worth calling out loudly in the public set_parsed_message_pre_ack_hook docs so nobody ships a heavy hook and wonders why their message throughput tanked.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/client/node_io.rs` around lines 24 - 32, The receive path in NodeIO is
forced to a single in-flight message whenever a parsed-message pre-ACK hook is
installed, which serializes the whole pipeline for the client’s lifetime. Update
the public documentation for set_parsed_message_pre_ack_hook to clearly call out
this throughput cost and that a heavy hook will back-pressure all inbound
messages. Mention the behavior change explicitly near the NodeIo /
parsed_message_pre_ack_hook API so consumers understand the tradeoff before
enabling it.
| match self.dispatch_pending_pre_ack_message(&info).await { | ||
| Some(true) => {} | ||
| Some(false) => return, | ||
| None => self.ack_received_message(&info), | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Send the post-pre-ACK status ACK when no pending replay exists.
These None fallbacks are reachable for status messages when the pre-ACK hook is installed, but ack_received_message() intentionally does not send the status transport ACK. Since the deferred status gate is disabled under the hook, this can leave duplicate status stanzas queued for redelivery. Use the status post-pre-ACK helper in the fallback path.
Proposed fix
- None => self.ack_received_message(&info),
+ None => {
+ self.ack_received_message(&info);
+ self.ack_status_drop_after_pre_ack_hook(&info);
+ } None if !info.source.chat.is_status_broadcast()
|| self.parsed_message_pre_ack_hook.get().is_some() =>
{
self.ack_received_message(info);
+ self.ack_status_drop_after_pre_ack_hook(info);
}Also applies to: 1211-1219
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/message/receive.rs` around lines 493 - 497, The fallback path in
dispatch_pending_pre_ack_message handling is using ack_received_message() for
status messages, but when the pre-ACK hook is installed that does not send the
required transport ACK and can leave status stanzas queued for redelivery.
Update the None branch in the receive flow to use the status post-pre-ACK helper
instead of ack_received_message(), and apply the same change in the other
matching status fallback site, keeping the logic centered around
dispatch_pending_pre_ack_message and the status ACK helper.
| Ok(plaintext_outcome) => { | ||
| outcome.decrypted = true; | ||
| outcome.dispatched |= plaintext_outcome.dispatched; | ||
| outcome.pre_ack_hook_failed |= plaintext_outcome.pre_ack_hook_failed; | ||
| outcome.had_failure |= plaintext_outcome.pre_ack_hook_failed; | ||
| outcome.skdm_only |= plaintext_outcome.skdm_only; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Stop processing the batch immediately after pre-ACK failure.
These branches set pre_ack_hook_failed, but the enclosing payload loops can continue processing later payloads before process_classified_message returns. That can still emit ACKs, nacks, or another message event in the same stanza after a failed commit. Return the accumulated outcome immediately when this flag is set.
Pattern to apply
outcome.pre_ack_hook_failed |= plaintext_outcome.pre_ack_hook_failed;
outcome.had_failure |= plaintext_outcome.pre_ack_hook_failed;
+if plaintext_outcome.pre_ack_hook_failed {
+ return outcome;
+}For group batches:
outcome.pre_ack_hook_failed |= plaintext_outcome.pre_ack_hook_failed;
+if plaintext_outcome.pre_ack_hook_failed {
+ return Ok(outcome);
+}Also applies to: 810-816, 954-966, 1005-1017, 1065-1077, 1192-1194
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/message/receive.rs` around lines 690 - 695, `process_classified_message`
and the payload-batch loops are continuing after `pre_ack_hook_failed` becomes
true, which can allow additional ACK/nack/event emission in the same stanza.
Update the branches that merge `plaintext_outcome`/batch outcomes so they
immediately return the accumulated `outcome` as soon as `pre_ack_hook_failed` is
set, and apply the same early-exit pattern in the other affected
batch-processing blocks mentioned in the review.
| let info = Arc::new(create_test_message_info( | ||
| "5511999998888@s.whatsapp.net", | ||
| "PRE_ACK_SUCCESS", | ||
| "5511777776666@s.whatsapp.net", | ||
| )); |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Use obviously fake test JIDs here.
These phone-like JIDs look plausibly real, and we need this airtight in test fixtures too. Replace them with clearly fictitious placeholders across all new cases so we do not ship possible real-user identifiers. As per coding guidelines, "Use fictitious phone numbers and JIDs in test code; never commit real user numbers (no real PII in tests)".
Also applies to: 5434-5438, 5482-5486, 5579-5583, 5609-5612
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/message/tests.rs` around lines 5388 - 5392, The test fixtures in Message
tests are using phone-like JIDs that could look real; replace these identifiers
in the affected create_test_message_info cases with clearly fictitious
placeholder JIDs so no plausible user numbers remain in test code. Update all
new/related cases in src/message/tests.rs, especially the
create_test_message_info calls around the listed PRE_ACK/related scenarios, to
use obviously fake values consistently.
Source: Coding guidelines
|
Thanks @Davincible. Can you provide a use case for this feature? I want to evaluate the usage in a real scenario |
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
5 issues found across 12 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="src/message/tests.rs">
<violation number="1" location="src/message/tests.rs:5389">
P2: These phone-like JIDs (`5511999998888@s.whatsapp.net`, `5511777776666@s.whatsapp.net`) look plausibly real. Use obviously fictitious placeholders (e.g., `10000000001@s.whatsapp.net` or `test_sender@s.whatsapp.net`) across all new test cases to ensure no real-user identifiers are shipped in test fixtures.</violation>
</file>
<file name="src/message/receive.rs">
<violation number="1" location="src/message/receive.rs:496">
P2: When the pre-ACK hook is installed, the deferred status ACK gate is disabled. In this `None` fallback for session duplicates, `ack_received_message` is called but `ack_status_drop_after_pre_ack_hook` is not — status messages in this path won't get their transport ACK sent, leaving duplicate status stanzas queued for redelivery. Add the post-pre-ACK status ACK call in the `None` fallback.</violation>
<violation number="2" location="src/message/receive.rs:693">
P1: After a pre-ACK hook failure, the batch loop continues processing subsequent payloads in the same stanza. This can emit ACKs, nacks, or additional `Event::Message` events for later payloads after a failed commit — violating the invariant that pre-ACK failure suppresses all further processing. Return the accumulated outcome immediately when `pre_ack_hook_failed` is set.</violation>
</file>
<file name="src/client/lifecycle.rs">
<violation number="1" location="src/client/lifecycle.rs:200">
P2: This cache is intentionally unbounded with no TTL, and it is never cleared in `cleanup_connection_state` (unlike `chat_lanes`, `signal_cache`, etc.). If a hook permanently fails for messages that never get successfully redelivered, entries and the associated `pending_parsed_message_pre_ack_count` accumulate indefinitely across reconnects. Consider adding at minimum a `max_capacity` or draining stale entries during connection cleanup to prevent unbounded growth.</violation>
</file>
<file name="src/client/adapters.rs">
<violation number="1" location="src/client/adapters.rs:68">
P2: This returns an `Err` for what is expected, routine backpressure (a deferred flush while a pre-ACK commit is pending). The `flush_signal_cache_logged` caller will log it at `error!` level on every deferred call, creating noisy false-alarm entries during normal operation. Either return a distinguishable "deferred" variant that the logged wrapper can downgrade to `debug!`/`trace!`, or use a dedicated error type that callers can match on to adjust log severity.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| Ok(plaintext_outcome) => { | ||
| outcome.decrypted = true; | ||
| outcome.dispatched |= plaintext_outcome.dispatched; | ||
| outcome.pre_ack_hook_failed |= plaintext_outcome.pre_ack_hook_failed; |
There was a problem hiding this comment.
P1: After a pre-ACK hook failure, the batch loop continues processing subsequent payloads in the same stanza. This can emit ACKs, nacks, or additional Event::Message events for later payloads after a failed commit — violating the invariant that pre-ACK failure suppresses all further processing. Return the accumulated outcome immediately when pre_ack_hook_failed is set.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/message/receive.rs, line 693:
<comment>After a pre-ACK hook failure, the batch loop continues processing subsequent payloads in the same stanza. This can emit ACKs, nacks, or additional `Event::Message` events for later payloads after a failed commit — violating the invariant that pre-ACK failure suppresses all further processing. Return the accumulated outcome immediately when `pre_ack_hook_failed` is set.</comment>
<file context>
@@ -673,6 +690,8 @@ impl Client {
Ok(plaintext_outcome) => {
outcome.decrypted = true;
outcome.dispatched |= plaintext_outcome.dispatched;
+ outcome.pre_ack_hook_failed |= plaintext_outcome.pre_ack_hook_failed;
+ outcome.had_failure |= plaintext_outcome.pre_ack_hook_failed;
outcome.skdm_only |= plaintext_outcome.skdm_only;
</file context>
| .expect("hook should register once"); | ||
|
|
||
| let info = Arc::new(create_test_message_info( | ||
| "5511999998888@s.whatsapp.net", |
There was a problem hiding this comment.
P2: These phone-like JIDs (5511999998888@s.whatsapp.net, 5511777776666@s.whatsapp.net) look plausibly real. Use obviously fictitious placeholders (e.g., 10000000001@s.whatsapp.net or test_sender@s.whatsapp.net) across all new test cases to ensure no real-user identifiers are shipped in test fixtures.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/message/tests.rs, line 5389:
<comment>These phone-like JIDs (`5511999998888@s.whatsapp.net`, `5511777776666@s.whatsapp.net`) look plausibly real. Use obviously fictitious placeholders (e.g., `10000000001@s.whatsapp.net` or `test_sender@s.whatsapp.net`) across all new test cases to ensure no real-user identifiers are shipped in test fixtures.</comment>
<file context>
@@ -5339,6 +5339,301 @@ async fn capturing_client(
+ .expect("hook should register once");
+
+ let info = Arc::new(create_test_message_info(
+ "5511999998888@s.whatsapp.net",
+ "PRE_ACK_SUCCESS",
+ "5511777776666@s.whatsapp.net",
</file context>
| "5511999998888@s.whatsapp.net", | |
| "10000000001@s.whatsapp.net", |
| match self.dispatch_pending_pre_ack_message(&info).await { | ||
| Some(true) => {} | ||
| Some(false) => return, | ||
| None => self.ack_received_message(&info), |
There was a problem hiding this comment.
P2: When the pre-ACK hook is installed, the deferred status ACK gate is disabled. In this None fallback for session duplicates, ack_received_message is called but ack_status_drop_after_pre_ack_hook is not — status messages in this path won't get their transport ACK sent, leaving duplicate status stanzas queued for redelivery. Add the post-pre-ACK status ACK call in the None fallback.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/message/receive.rs, line 496:
<comment>When the pre-ACK hook is installed, the deferred status ACK gate is disabled. In this `None` fallback for session duplicates, `ack_received_message` is called but `ack_status_drop_after_pre_ack_hook` is not — status messages in this path won't get their transport ACK sent, leaving duplicate status stanzas queued for redelivery. Add the post-pre-ACK status ACK call in the `None` fallback.</comment>
<file context>
@@ -474,14 +482,19 @@ impl Client {
+ match self.dispatch_pending_pre_ack_message(&info).await {
+ Some(true) => {}
+ Some(false) => return,
+ None => self.ack_received_message(&info),
+ }
} else if should_ack_skdm_only_session_fallback(session_outcome, bot_payloads.is_empty()) {
</file context>
| None => self.ack_received_message(&info), | |
| None => { | |
| self.ack_received_message(&info); | |
| self.ack_status_drop_after_pre_ack_hook(&info); | |
| } |
| // not durably committed. Expiring it would allow a later Signal | ||
| // flush to make that advance durable and turn server redelivery | ||
| // into an ACKed duplicate without rerunning the hook. | ||
| pending_parsed_message_pre_ack: Cache::builder().build(), |
There was a problem hiding this comment.
P2: This cache is intentionally unbounded with no TTL, and it is never cleared in cleanup_connection_state (unlike chat_lanes, signal_cache, etc.). If a hook permanently fails for messages that never get successfully redelivered, entries and the associated pending_parsed_message_pre_ack_count accumulate indefinitely across reconnects. Consider adding at minimum a max_capacity or draining stale entries during connection cleanup to prevent unbounded growth.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/client/lifecycle.rs, line 200:
<comment>This cache is intentionally unbounded with no TTL, and it is never cleared in `cleanup_connection_state` (unlike `chat_lanes`, `signal_cache`, etc.). If a hook permanently fails for messages that never get successfully redelivered, entries and the associated `pending_parsed_message_pre_ack_count` accumulate indefinitely across reconnects. Consider adding at minimum a `max_capacity` or draining stale entries during connection cleanup to prevent unbounded growth.</comment>
<file context>
@@ -192,6 +192,13 @@ impl Client {
+ // not durably committed. Expiring it would allow a later Signal
+ // flush to make that advance durable and turn server redelivery
+ // into an ACKed duplicate without rerunning the hook.
+ pending_parsed_message_pre_ack: Cache::builder().build(),
+ pending_parsed_message_pre_ack_count: AtomicUsize::new(0),
</file context>
| .load(Ordering::Acquire) | ||
| > 0 | ||
| { | ||
| return Err(anyhow::anyhow!( |
There was a problem hiding this comment.
P2: This returns an Err for what is expected, routine backpressure (a deferred flush while a pre-ACK commit is pending). The flush_signal_cache_logged caller will log it at error! level on every deferred call, creating noisy false-alarm entries during normal operation. Either return a distinguishable "deferred" variant that the logged wrapper can downgrade to debug!/trace!, or use a dedicated error type that callers can match on to adjust log severity.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/client/adapters.rs, line 68:
<comment>This returns an `Err` for what is expected, routine backpressure (a deferred flush while a pre-ACK commit is pending). The `flush_signal_cache_logged` caller will log it at `error!` level on every deferred call, creating noisy false-alarm entries during normal operation. Either return a distinguishable "deferred" variant that the logged wrapper can downgrade to `debug!`/`trace!`, or use a dedicated error type that callers can match on to adjust log severity.</comment>
<file context>
@@ -59,6 +59,17 @@ impl Client {
+ .load(Ordering::Acquire)
+ > 0
+ {
+ return Err(anyhow::anyhow!(
+ "Signal cache flush deferred while parsed-message pre-ACK commit is pending"
+ ));
</file context>
Merging this PR will not alter performance
Comparing 
Footnotes
|
|
@Davincible take a look in #920 |
2 participants
Summary
Tests
Note: the plain cargo command in this environment uses stable rustc and fails on the existing default simd feature; the commands above force the repository nightly toolchain.