feat(recv)!: batch the inbound commit pipeline during the offline drain - #961
Merged
Conversation
Production traces showed the offline drain paying every durability cost
per message while globally serialized (1-permit semaphore): one hook
commit, two pending-buffer transactions, one Signal-cache flush and one
spawned event per message — ~2.8ms each on a Postgres-backed hook, so a
421-message backlog spent ~1.2s in per-message round-trips.
WA Web batches all of this. Its MessageProcessorCache accumulates
decrypted messages (size cap + timeout + end-of-drain triggers) and
createSnapshot commits per batch, in order: bulk message-table write,
bulk signal-store commit under lock, THEN the aggregated receipts for
the batch. Our per-message hook was the parity divergence, not the
batch.
This introduces an InboundCommitBatcher with the same triggers (200
msgs / 4 MiB / 3s / end-of-drain) and the same commit ordering: pending
buffer (one txn, single reused encode arena) -> Signal flush -> hook ->
buffer clear (one txn) -> event -> acks. Nothing is acked or observable
before it is durable; a failed batch stays unacked whole and the server
redelivers it. Live traffic commits immediately as a batch of one (also
WA Web behavior). Every flush acquires the global processing permit so
the Signal flush can never interleave with a half-processed stanza,
which would otherwise persist a ratchet advance for an uncommitted
message and turn its redelivery into an unrecoverable duplicate. The
end-of-drain flush runs before the semaphore widens, so the batcher is
provably empty on the live path.
Breaking (pre-1.0):
- InboundDurabilityHook::on_message(info, message) is replaced by
on_messages(&[InboundMessage]) — all-or-nothing, slice order, batch
of one on live traffic.
- Event::Message(Arc<Message>, Arc<MessageInfo>) is replaced by
Event::Messages(MessageBatch { messages: Arc<[InboundMessage]>,
origin: Live | OfflineDrain }) — Baileys' messages.upsert shape. The
slice is the same allocation handed to the hook: what was committed
is exactly what event consumers observe, in order.
- EventKind::Message -> EventKind::Messages; Event::as_message() ->
message_batch()/messages(); MessageContext::from_event ->
from_inbound. Bot::on_message keeps its per-message signature by
fanning out the batch in order.
- ProtocolStore gains store/delete_pending_inbound_batch (defaults
iterate the single-row methods, so third-party backends keep
working); SqliteStore overrides both with one transaction per batch.
Per-message costs for a 421-message drain: 421 hook commits -> ~3, 842
pending-buffer txns -> ~6, 421 Signal flushes -> ~3, 421 event spawns
per handler -> ~3, 421 encode Vecs -> ~3 reused arenas. Delivery
receipts and message secrets were already batched (offline receipt
aggregation, MsgSecretWriteBuffer); bare <ack> stanzas stay 1:1, which
matches WA Web (its pre-ack batcher batches persistence, not sends).
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI (base), Organization UI (inherited) Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughSummary by CodeRabbit
WalkthroughThis PR replaces single-message inbound delivery with batched ChangesBatched inbound messages and durability flow
Estimated code review effort: 4 (Complex) | ~60 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 |
📦 Binary size report
.text per crate
Top movers (cargo-bloat attribution)
🚨 Per-PR size budget exceeded (Δ stripped ≤ 64.00 KiB, Δ .text ≤ 32.00 KiB):
The Baseline: |
MessageContext is not Send on wasm32 (the Client's trait objects carry no Send/Sync bounds there), so the batch fan-out must not hold contexts across an await. Build every handler future before the async block — only the futures, which the Fut: Send bound covers, cross the awaits. Same shape the pre-batch code used for its single future.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
4 issues found and verified against the latest diff
Confidence score: 2/5
- In
src/message/receive.rs, drain messages can still take the live flush path during offline→live transition, reopening a window where Signal ratchet state is persisted before the batch is durably committed; merging as-is risks crashes that leave protocol state ahead of stored messages. Gate drain handling to the offline-safe commit path (or equivalent durability ordering) before merging. - In
examples/durability_hook.rs, duplicate keys within the same batch are not deduplicated before archive write, so one fsync can append the same key twice and create duplicate durable records. Check both previously seen keys and keys already queued in the current batch before writing. - In
src/test_utils.rs, helpers mark tests as live while still using a 1-permit semaphore, so true concurrent live-path behavior is not exercised and race/ordering regressions can slip through. Update the helper to use realistic live-path concurrency before relying on these tests as merge confidence. - In
src/message/commit_batch.rs, stale-permit re-acquire logic is duplicated in multiple paths, increasing the chance of future semaphore-generation fixes being applied inconsistently and reintroducing subtle concurrency bugs. Consolidate permit acquisition into a shared helper to de-risk follow-up changes.
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="examples/durability_hook.rs">
<violation number="1" location="examples/durability_hook.rs:123">
P2: Duplicate messages inside a single batch are not deduplicated before archive write, so the same key can be appended twice in one fsync. Checking both already-seen keys and keys queued earlier in this batch would keep archive output idempotent.</violation>
</file>
<file name="src/test_utils.rs">
<violation number="1" location="src/test_utils.rs:117">
P2: Tests created via this helper are marked as "live" but still run with a 1-permit message semaphore, so concurrent live-path behavior is not exercised and race/ordering bugs can be masked. Consider updating the helper to mirror offline completion state by widening permits when setting `offline_sync_completed=true`.</violation>
</file>
<file name="src/message/receive.rs">
<violation number="1" location="src/message/receive.rs:536">
P1: A drain message can take the live flush path during the offline->live transition, which reintroduces a crash window where Signal ratchet state is persisted before that message batch is durably committed. This comes from gating on `offline_sync_completed` even though `complete_offline_sync()` sets that flag before flushing the tail batch; consider keeping batch-owned flush behavior until the pending inbound batch is fully drained.</violation>
</file>
<file name="src/message/commit_batch.rs">
<violation number="1" location="src/message/commit_batch.rs:145">
P3: Stale-permit re-acquire logic now exists in two places, which makes future fixes to semaphore-generation handling easy to apply in one path and miss in the other. A shared helper for acquiring the current message-processing permit would reduce drift risk in this safety-critical flow.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
![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: e0dc4a4700
ℹ️ 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".
|
| Filename | Overview |
|---|---|
| src/message/commit_batch.rs | New 1279-line core of the PR: InboundCommitBatcher with ReinsertGuard, deferred transition retry, drain/live commit paths, and a comprehensive test suite covering all failure modes. |
| src/client/sessions.rs | complete_offline_sync made async; drain finisher spawned off the read loop to avoid deadlocking slow hooks against pong/IQ responses; publish_offline_sync_live_state handles durable/deferred/upgrade-failure cases. |
| src/client/lifecycle.rs | connection_generation bumped first in cleanup_connection_state; teardown_inbound_commits_bounded replaces raw flush+clear; disconnect/reconnect paths gate offline receipts on durable batch commit. |
| src/client/adapters.rs | flush_signal_cache_batch_safe added: during drain routes through flush_inbound_commits_under_permit to commit rows before persisting ratchet advances; outside drain, plain flush. |
| src/message/receive.rs | Lane-generation guard added at classification and post-permit; live path calls flush_signal_cache_logged, drain path calls maybe_flush_inbound_commits; identity-change path commits drain batch before raw cache flush. |
| wacore/src/types/events.rs | Event::Message replaced by Event::Messages(MessageBatch); InboundMessage, MessageBatch, BatchOrigin added as first-class types with IntoIterator, iter, len, first. |
| storages/sqlite-storage/src/sqlite_store.rs | store_pending_inbound_batch and delete_pending_inbound_batch overrides added, each wrapping per-row calls in a single transaction; shared helpers prevent single/batch path divergence. |
| src/bot.rs | on_message fans out Event::Messages batch in arrival order; all handler closures called up-front then awaited sequentially to avoid holding MessageContext across an await on wasm32. |
| wacore/src/store/traits.rs | store_pending_inbound_batch / delete_pending_inbound_batch added to ProtocolStore with defaults iterating single-row methods; PendingInboundRow / PendingInboundKey zero-copy borrowed types added. |
| src/receipt.rs | try_buffer_offline_receipt extended with inbound_commit_batch.is_active() guard so receipts continue buffering during the deferred drain-to-live window. |
Sequence Diagram
%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant RL as ReadLoop
participant PCM as process_classified_message
participant ICB as InboundCommitBatcher
participant BE as Backend
participant SC as SignalCache
participant Hook as DurabilityHook
participant Bus as EventBus
Note over RL,Bus: Offline drain (single permit, batching active)
RL->>PCM: acquire permit, decrypt
PCM->>ICB: commit_or_batch_inbound(entry)
ICB->>ICB: enqueue + arm 3s timer
PCM->>ICB: maybe_flush_inbound_commits if over limit
Note over RL,Bus: End-of-drain finisher spawned off read loop
RL->>ICB: finish_inbound_commit_drain(generation)
ICB->>ICB: take batch, bump epoch
ICB->>BE: store_pending_inbound_batch one txn
alt store succeeds
ICB->>SC: flush_signal_cache bulk
ICB->>ICB: disarm ReinsertGuard
ICB->>Hook: on_messages(batch)
ICB->>BE: delete_pending_inbound_batch
ICB->>PCM: ack each message
ICB->>Bus: Event::Messages(OfflineDrain)
ICB->>ICB: deactivate + swap semaphore to 64
ICB->>ICB: flush_offline_receipts
else store fails
ICB->>ICB: ReinsertGuard restores entries
ICB->>ICB: defer_live_transition + arm retry loop
end
Note over RL,Bus: Live path (64 permits, immediate batch-of-one)
PCM->>ICB: "commit_or_batch_inbound is_active=false"
ICB->>BE: store 1 row
ICB->>Hook: on_messages batch-of-one
ICB->>BE: delete 1 row
ICB->>PCM: ack
ICB->>Bus: Event::Messages(Live)
PCM->>SC: flush_signal_cache_logged per-stanza
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
participant RL as ReadLoop
participant PCM as process_classified_message
participant ICB as InboundCommitBatcher
participant BE as Backend
participant SC as SignalCache
participant Hook as DurabilityHook
participant Bus as EventBus
Note over RL,Bus: Offline drain (single permit, batching active)
RL->>PCM: acquire permit, decrypt
PCM->>ICB: commit_or_batch_inbound(entry)
ICB->>ICB: enqueue + arm 3s timer
PCM->>ICB: maybe_flush_inbound_commits if over limit
Note over RL,Bus: End-of-drain finisher spawned off read loop
RL->>ICB: finish_inbound_commit_drain(generation)
ICB->>ICB: take batch, bump epoch
ICB->>BE: store_pending_inbound_batch one txn
alt store succeeds
ICB->>SC: flush_signal_cache bulk
ICB->>ICB: disarm ReinsertGuard
ICB->>Hook: on_messages(batch)
ICB->>BE: delete_pending_inbound_batch
ICB->>PCM: ack each message
ICB->>Bus: Event::Messages(OfflineDrain)
ICB->>ICB: deactivate + swap semaphore to 64
ICB->>ICB: flush_offline_receipts
else store fails
ICB->>ICB: ReinsertGuard restores entries
ICB->>ICB: defer_live_transition + arm retry loop
end
Note over RL,Bus: Live path (64 permits, immediate batch-of-one)
PCM->>ICB: "commit_or_batch_inbound is_active=false"
ICB->>BE: store 1 row
ICB->>Hook: on_messages batch-of-one
ICB->>BE: delete 1 row
ICB->>PCM: ack
ICB->>Bus: Event::Messages(Live)
PCM->>SC: flush_signal_cache_logged per-stanza
Reviews (14): Last reviewed commit: "fix(recv): fail closed on active drain w..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 1 file (changes from recent commits).
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/bot.rs">
<violation number="1" location="src/bot.rs:693">
P2: Batch fan-out now calls `handler(...)` for the entire batch before awaiting, so any synchronous side effect or panic during future construction changes per-message sequencing and can drop already-built earlier futures without running them. Preserving call+await per message keeps the documented in-order semantics and avoids this eager-construction edge case.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
![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
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
tests/e2e/tests/session_reuse.rs (1)
9-33: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winFlatten the nested
if letinto a let-chain.
recordis only used to reachstate; this collapses into one condition per our lint policy on nestedif let.♻️ Proposed let-chain refactor
- if let Some(data) = backend.get_session(&addr).await? { - let record = SessionRecord::deserialize(&data)?; - if let Some(state) = record.session_state() { - let has_pending = state - .unacknowledged_pre_key_message_items() - .map_err(|e| anyhow::anyhow!("invalid session state: {e}"))? - .is_some(); - results.push((addr, has_pending)); - } - } + if let Some(data) = backend.get_session(&addr).await? + && let Some(state) = SessionRecord::deserialize(&data)?.session_state() + { + let has_pending = state + .unacknowledged_pre_key_message_items() + .map_err(|e| anyhow::anyhow!("invalid session state: {e}"))? + .is_some(); + results.push((addr, has_pending)); + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/e2e/tests/session_reuse.rs` around lines 9 - 33, The nested conditional in scan_sessions should be collapsed into a single let-chain to satisfy the lint policy. Refactor the backend.get_session, SessionRecord::deserialize, and record.session_state checks into one combined condition, then keep the has_pending calculation and results.push inside that branch.Source: Coding guidelines
src/client/lifecycle.rs (1)
580-591: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftBound the inbound drain wait
flush_inbound_commits_acquiring_permit()has no internal timeout: it waits onSemaphore::acquire_arc().awaitand thencommit_inbound_batch().await. If either stalls,disconnect(),reconnect(), andreconnect_immediately()can block forever. Apply the same timeout pattern used foroutbound_flush.flush()here too.🤖 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 580 - 591, The inbound drain path in `disconnect()` can block forever because `flush_inbound_commits_acquiring_permit()` waits indefinitely on permit acquisition and commit completion. Add the same bounded-wait timeout pattern used by `outbound_flush.flush()` around this call in `lifecycle.rs`, and make sure `disconnect()`, `reconnect()`, and `reconnect_immediately()` continue after the timeout instead of hanging.
🤖 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/sessions.rs`:
- Around line 39-48: The `self_weak.get().and_then(|w| w.upgrade())` check in
the session commit path can silently skip
`flush_inbound_commits_acquiring_permit()` when the weak reference is gone,
which breaks the invariant described in this block. Update this branch in
`src/client/sessions.rs` to log a clear warning or error when `upgrade()`
returns `None`, using the surrounding session/commit context so the failure is
visible. Keep the existing `swap_message_semaphore(64)` and
`flush_offline_receipts()` flow intact, but ensure the `None` path is explicitly
reported rather than failing silently.
In `@src/message/commit_batch.rs`:
- Around line 144-156: The generation-check semaphore acquire loop is duplicated
between flush_inbound_commits_acquiring_permit and process_classified_message,
so extract it into a shared helper on Client such as acquire_message_permit.
Move the read_message_semaphore / acquire_arc / generation-compare retry logic
into that method, then update both call sites to await the helper and keep any
call-site-specific logging separate. This keeps the offline→online semaphore
swap guard consistent in one place and prevents the two code paths from
drifting.
- Around line 194-236: The `arena` mutex guard in `commit_batch` is held longer
than necessary, blocking unrelated inbound commits on the shared `Client` hot
path. After `store_pending_inbound_batch` succeeds and `rows` is dropped,
explicitly release the `arena` lock before calling `flush_signal_cache_logged`,
`hook.on_messages`, and any later cleanup so the encode arena is not held during
slow durability/hook work.
In `@src/message/receive.rs`:
- Around line 531-544: The SKDM-only fallback currently bypasses the
offline-drain flush order by calling ack logic before the Signal cache is
durably written. Update should_ack_skdm_only_session_fallback so it is handled
through commit_or_batch_inbound, or otherwise gate ack_received_message on the
same flush path used in receive.rs to ensure ratchet advancement only happens
after the cache flush completes.
In `@tests/e2e/tests/app_state.rs`:
- Around line 213-228: The message lookup in the app state test is scanning the
same event batch twice, once in the `wait_for_event` predicate and again with
`event.messages().find(...)` to extract the id. Update the logic around
`client_b.wait_for_event` and the subsequent `msg_id` assignment to derive the
matching message and its `info.id` in a single pass, ideally using `find_map`
over `event.messages()` so the code only traverses the batch once.
In `@tests/e2e/tests/media.rs`:
- Around line 413-421: The media E2E tests repeat the same two-step pattern of
waiting for an event and then searching the messages again with the same
predicate. Add a reusable helper such as wait_for_message on TestClient (or a
free function) that wraps wait_for_event plus the follow-up find, returns the
matched InboundMessage directly, and update the repeated call sites in media.rs
to use it so the matching logic lives in one place.
In `@wacore/src/types/events.rs`:
- Around line 846-849: The message_text() helper on Event::Messages is only
returning the first conversation value from messages(), which can silently drop
additional texts in batched offline-drain events. Update the Events API so
callers use an iterator-style accessor over all messages, or make message_text()
explicitly only succeed for single-message batches and avoid partial processing.
Use the existing messages() method and the Event::Messages path to locate the
change.
---
Outside diff comments:
In `@src/client/lifecycle.rs`:
- Around line 580-591: The inbound drain path in `disconnect()` can block
forever because `flush_inbound_commits_acquiring_permit()` waits indefinitely on
permit acquisition and commit completion. Add the same bounded-wait timeout
pattern used by `outbound_flush.flush()` around this call in `lifecycle.rs`, and
make sure `disconnect()`, `reconnect()`, and `reconnect_immediately()` continue
after the timeout instead of hanging.
In `@tests/e2e/tests/session_reuse.rs`:
- Around line 9-33: The nested conditional in scan_sessions should be collapsed
into a single let-chain to satisfy the lint policy. Refactor the
backend.get_session, SessionRecord::deserialize, and record.session_state checks
into one combined condition, then keep the has_pending calculation and
results.push inside that branch.
🪄 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: 7d810c67-69bb-4f58-b2a4-f062133aefa4
📒 Files selected for processing (35)
agent_docs/e2e_testing.mdexamples/benchmark.rsexamples/durability_hook.rssrc/bot.rssrc/client.rssrc/client/lifecycle.rssrc/client/sessions.rssrc/handlers/ib.rssrc/message.rssrc/message/commit_batch.rssrc/message/dispatch.rssrc/message/durability.rssrc/message/receive.rssrc/message/tests.rssrc/pdo.rssrc/reexports_test.rssrc/send/mod.rssrc/test_utils.rssrc/types/durability_hook.rsstorages/sqlite-storage/src/sqlite_store.rstests/e2e/src/lib.rstests/e2e/tests/app_state.rstests/e2e/tests/concurrent_disconnect.rstests/e2e/tests/groups.rstests/e2e/tests/media.rstests/e2e/tests/memory_soak.rstests/e2e/tests/messaging.rstests/e2e/tests/offline_groups.rstests/e2e/tests/offline_messages.rstests/e2e/tests/prekey_sessions.rstests/e2e/tests/privacy_tokens.rstests/e2e/tests/receipts.rstests/e2e/tests/session_reuse.rswacore/src/store/traits.rswacore/src/types/events.rs
…/codex/greptile/coderabbit The drain/live gate moved off offline_sync_completed onto a dedicated batcher-owned `active` state that only the end-of-drain flush flips, while holding the single processing permit. Gating on the flag was racy in two confirmed ways: an in-flight stanza could enqueue pre-flip and then take the live per-stanza Signal flush at stanza end, persisting ratchet state for an uncommitted batch entry (crash => unrecoverable duplicate, cubic P1); and stanzas queued behind the permit could read the flipped flag at dispatch and commit as Live ahead of the still accumulated batch, inverting arrival order at the drain tail (codex P1). finish_inbound_commit_drain commits the tail and deactivates under the permit, so no stanza straddles the transition and queued stanzas provably commit after the tail. Regression test locks the sequence. Also from review: - Acks are sent before the event dispatch (all durability done by then): handle_event runs synchronously, so a panicking or blocking handler must not suppress acks for messages the consumer already owns — matches the old at-most-once ordering (codex). - The encode buffer is scoped to the durable write only and the live path uses a local buffer instead of the shared arena, so concurrent live commits never serialize behind a slow hook (greptile, coderabbit). - store_pending_inbound_batch uses multi-row VALUES inserts (chunked at 100 rows); deletes stay per-row inside the one transaction because Diesel's DSL cannot express a composite-key tuple IN (greptile). - The stale-permit re-acquire loop is one shared helper, acquire_message_processing_permit (cubic, coderabbit). - Teardown flushes are bounded (5s disconnect / 2s reconnect, like the outbound flush): a hung hook cannot wedge disconnect; entries stay unacked for redelivery (coderabbit). - complete_offline_sync logs loudly if the self_weak upgrade ever fails instead of silently skipping the tail commit (coderabbit). - Event::message_texts() iterator; message_text() documented as first-text convenience (coderabbit). - Test clients mirror full live state (flag, 64 permits, batcher live); batch example dedups within the batch; on_message documents the eager future construction wasm32 requires (cubic).
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 66ac89ae54
ℹ️ 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".
There was a problem hiding this comment.
1 issue found across 11 files (changes from recent commits).
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/bot.rs">
<violation number="1" location="src/bot.rs:693">
P2: Batch fan-out now calls `handler(...)` for the entire batch before awaiting, so any synchronous side effect or panic during future construction changes per-message sequencing and can drop already-built earlier futures without running them. Preserving call+await per message keeps the documented in-order semantics and avoids this eager-construction edge case.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
…atch replayed messages Two Codex findings on the review round, both real: - cleanup_connection_state flushed the Signal cache and then reset the batcher, so an unexpected read-loop exit mid-drain persisted ratchet advances for entries it was about to drop — their redelivery would decrypt as a duplicate with no buffered copy and get acked without delivery. The cleanup now commits the batch (bounded, 5s) before that flush; acks/events are best-effort with the socket gone, but the durable hook commit is what prevents the loss. - A message whose hook only succeeds on redelivery replay never emitted Event::Messages (its original batch aborted before dispatch). The replay success path now dispatches the single-message batch after the ack, so consumers see it exactly once; test extended to lock this. Also: live batches build their Arc<[InboundMessage]> slice directly instead of round-tripping through a Vec — measured 2.0 -> 1.0 allocations and ~18ns (-24%) for that step per live message; commit_inbound_batch now takes the Arc slice, which the event reuses without conversion.
66ac89a's message claimed this reorder but the patch hunk had silently failed to apply; the code still dispatched Event::Messages before sending acks (and with a redundant Arc conversion, caught by clippy's useless_conversion). Apply it for real: everything is durable by the time either happens, and acking first means a panicking or blocking synchronous handler cannot suppress acks for messages the consumer already owns — the ordering the pre-batch at-most-once path had.
|
✅ Action performedReview finished.
|
There was a problem hiding this comment.
2 issues found across 3 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 9c9bee768c
ℹ️ 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".
…p .text in budget - An SKDM-only drain stanza mutates Signal state without enqueueing a message, and its buffered receipt flushes at drain end / teardown — right after the batcher flush call sites. The empty-batch early return I added with the Arc refactor silently dropped the "flush Signal even when there is nothing to commit" guarantee, so a crash after the receipt flush could lose an acked sender key. Empty drain flushes now persist the Signal cache before returning (codex). - commit_inbound_batch's doc said event -> acks while the code acks first; the doc now states the ack -> event contract and why (a crash between them trades exactly like the old at-most-once path: the consumer's durable copy is the hook commit, not the event) (cubic). - A replayed message now always dispatches with BatchOrigin::Live, and BatchOrigin is documented as the delivery shape (immediate vs accumulated drain batch) — a stanza's provenance is info.is_offline. Inferring origin from the batcher's current state could mislabel replays (cubic). - complete_offline_sync documents why flag-gated waiters may unblock during the tail commit (same window in-flight stanzas always had; OfflineSyncCompleted still dispatches only after the commit). - Revert the multi-row VALUES insert to per-row statements inside the single batch transaction: the WAL commit is the amortized cost either way, and the multi-row form added ~4 KiB of monomorphized .text against a 32 KiB per-PR budget currently at 31.9 KiB.
There was a problem hiding this comment.
1 issue found across 4 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: dcdc148c39
ℹ️ 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".
During a drain, a buffered redelivery used to commit and dispatch immediately while earlier freshly-decrypted stanzas were still accumulating in the batcher, so consumers could observe the replayed message ahead of its predecessors (codex). The replay's Ok branch now just calls commit_or_batch_inbound: in drain mode it joins the accumulating batch (arrival order preserved, its pending row rewritten and cleared by the batch commit), live it commits as a batch of one — which is store -> hook -> delete -> ack -> event, exactly what the hand-rolled branch did, so 30 lines fold away. This also supersedes dcdc148's origin change for replays: that hunk had silently failed to apply (same failure mode as the one confessed in 9c9bee7 — patch scripts now assert), and the batcher routing makes it moot: a drain replay is genuinely part of an OfflineDrain batch and a live replay genuinely a Live one.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d1d0019190
ℹ️ 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".
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Requires human review: Auto-approval blocked by 2 unresolved issues from previous reviews.
Re-trigger cubic
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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/lifecycle.rs`:
- Around line 755-759: The self_weak.upgrade() failure in the inbound commit
flush path is currently silent, unlike the matching handling in
complete_offline_sync. Update the lifecycle logic around the
self_weak.get().and_then(|w| w.upgrade()) check in this flush section to emit a
log::error! when the upgrade returns None, and keep the same descriptive context
used for the other failure path so both code paths report the same class of bug
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: 1ba51884-9f0e-4ddf-88ca-d8585123a9a0
📒 Files selected for processing (6)
src/client/lifecycle.rssrc/client/sessions.rssrc/message/commit_batch.rssrc/message/durability.rsstorages/sqlite-storage/src/sqlite_store.rswacore/src/types/events.rs
… identity path single-permit Two stale-state windows around awaited drain commits: - The finisher's generation check ran only before the take. A teardown that times out around the awaited commit (slow hook) resets the batcher, and the resuming finisher then deactivated the NEW connection's drain. The generation is now re-checked after the await, before any mode mutation; the commit itself stays sound (entries taken pre-reset, rows durable). - commit_inbound_batch_holding_permit no longer completes a pending deferred transition: the caller (UntrustedIdentity recovery) holds a permit from the old single-permit semaphore and follows up with a raw whole-cache flush, which is only safe while that permit excludes every other stanza — widening to 64 first would let new workers be mid-decrypt under the flush. The deferred-retry loop completes the transition moments later, outside any raw-flush window.
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Requires human review: This is a high-impact refactor of the message processing pipeline that changes core business logic, including breaking API changes to events, hooks, and data structures. Even with no linting issues, such changes need human review to evaluate architectural correctness and potential regressions.
Re-trigger cubic
The comment still described an inline generation-reacquire loop, but that logic moved into acquire_message_processing_permit; trim it to reference the helper while keeping the SKDM-loss rationale for why the 1→N re-acquire matters.
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Requires human review: This PR makes a substantial architectural change to inbound message batching during offline drain, affecting multiple core components, public API, and concurrency logic. The high risk of breakage and complexity necessitates human review.
Re-trigger cubic
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 68358c4fc6
ℹ️ 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".
The timeout branch deferred the cache decision to the batcher reset, but reset only clears when it drops entries — so a worker hung mid-decrypt (holding the permit the settle timed out waiting for) that advanced a ratchet WITHOUT enqueueing an entry (SKDM-only) left rowless state the reset could not see, and a later flush would persist it into an acked-duplicate loss. Clear unconditionally on timeout instead: it does not sample has_entries() (which races that permit holder — the concern from the earlier round), and it is the only rowless-safe action when a stuck worker may hold unenqueued advances. The common hook-timeout case is already safe (the ReinsertGuard restored the entries), so this only changes the rare hung-worker corner. The sole thing it can drop is committed state a prior failed flush retained across a total storage outage — a degraded corner where redeliver-on-reauth beats a silent rowless duplicate.
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Requires human review: Major refactor of inbound message processing pipeline with breaking API changes, affecting core business logic, durability hooks, and event handling. High risk of regressions despite automated reviews.
Re-trigger cubic
No production consumers yet, so shape the new batch API properly instead of carrying the accreted form: - Rename Event::message_batch() → Event::as_messages(): the idiomatic as_ prefix for a &self -> Option<&T> accessor, and it now tiers clearly against messages() (the flattening iterator) instead of two near-synonym names. - Make MessageBatch a first-class collection: iter/len/is_empty/first and IntoIterator for &MessageBatch, so `for m in &batch`, batch.iter(), batch.len() work without reaching through the .messages field. origin stays alongside as the delivery-shape metadata. - messages() is reimplemented on top of as_messages(); it stays because `event.messages().any(...)` / `for m in event.messages()` is the dominant consumer verb and reads better than the derived form. Callers migrated; the batcher tests and benchmark example dogfood the collection methods.
There was a problem hiding this comment.
0 issues found across 4 files (changes from recent commits).
Requires human review: Major refactor of inbound message pipeline: batches commits, changes event types, modifies durability hook API, and updates concurrency model. High risk of regressions in message processing, Signal state, and durability.
Re-trigger cubic
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 19bef0e35b
ℹ️ 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".
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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/lifecycle.rs`:
- Around line 815-825: The teardown path in lifecycle handling silently drops
uncommitted inbound commit entries when `inbound_commit_batch.reset()` returns
true, unlike the `connect_graph` path that already logs a warning for the same
condition. Update this branch to emit a warning or error with clear context
before calling `signal_cache.clear().await`, using the same symbols
(`inbound_commit_batch.reset`, `signal_cache.clear`, and the surrounding
teardown logic) so unexpected redelivery-triggering drops are visible in logs.
🪄 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: 019ae790-28f5-4205-8249-7eb43ec5426c
📒 Files selected for processing (30)
examples/benchmark.rssrc/client.rssrc/client/adapters.rssrc/client/device_registry.rssrc/client/lifecycle.rssrc/client/sender_keys.rssrc/client/sessions.rssrc/client/tests.rssrc/features/message_edit.rssrc/features/signal.rssrc/handlers/message.rssrc/handlers/notification/device.rssrc/lib.rssrc/message/commit_batch.rssrc/message/dispatch.rssrc/message/durability.rssrc/message/receive.rssrc/message/tests.rssrc/msg_secret_buffer.rssrc/receipt.rssrc/retry.rssrc/send/mod.rssrc/test_utils.rssrc/types/durability_hook.rssrc/voip/facade.rsstorages/sqlite-storage/src/sqlite_store.rstests/e2e/src/lib.rstests/e2e/tests/offline_groups.rstests/e2e/tests/receipts.rswacore/src/types/events.rs
flush_signal_cache_batch_safe entered on an is_active() check that the drain finisher can invalidate before the wrapper acquired the permit: the under-permit commit then took an empty batch in live mode and reported durable WITHOUT flushing the caller's out-of-band Signal advance, leaving it dirty until the next flush. Re-check is_active() after the commit and fall through to the raw flush when the batcher deactivated mid-wait — the same path a caller that found it already inactive takes. Covered by a deferred-transition test driven through the wrapper. Also log the teardown branch that drops uncommitted drain entries and clears the Signal cache, matching the connect-side twin and the upgrade-failure path so redelivery-triggering drops are visible.
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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 124-135: The active-empty path in
`flush_inbound_commits_under_permit`/`commit` can return durable without
actually flushing Signal state, so `is_active()` alone is not a safe success
check. Update `src/client/adapters.rs` so the logic around
`inbound_commit_batch.is_active()` also reflects whether Signal was really
flushed: either have `flush_inbound_commits_under_permit(false, None, None)`
return a flag indicating an actual Signal flush, or explicitly perform the
under-permit Signal flush before returning `Ok(())` when the batcher is still
active.
- Around line 114-136: The nested `if` plus `if let` in `src/client/adapters.rs`
should be collapsed into a single let-chain to satisfy the repo’s Clippy rule.
Refactor the `inbound_commit_batch.is_active()` branch in the adapter method so
the `self.self_weak.get().and_then(|w| w.upgrade())` check is combined with the
outer condition using a let-chain, while preserving the existing
`flush_inbound_commits_under_permit` flow and the `has_entries()` fallback path.
In `@src/message/commit_batch.rs`:
- Around line 1204-1210: The test around flush_signal_cache_batch_safe currently
only verifies the deferred transition completes and can still pass if the final
raw flush is skipped. Update the test to make the raw flush observable by using
an injected raw-flush failure or a flush counter on client/flush_signal_cache so
that flush_signal_cache_batch_safe must actually reach flush_signal_cache().
Keep the assertion tied to the existing batch-safe path in commit_batch.rs so
the regression is caught if the raw flush is omitted.
🪄 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: ae399dd5-f0e3-4aa7-a3eb-746989db384d
📒 Files selected for processing (3)
src/client/adapters.rssrc/client/lifecycle.rssrc/message/commit_batch.rs
There was a problem hiding this comment.
1 issue found across 3 files (changes from recent commits).
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/commit_batch.rs">
<violation number="1" location="src/message/commit_batch.rs:1179">
P3: This test validates deferred-transition completion, but it does not verify that the final raw `flush_signal_cache()` path executed. A regression that commits `B1` and returns before that raw flush would still pass these assertions, so adding an observable flush assertion would close the gap.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
… recheck The batch-safe flush proved durability with a post-commit is_active() recheck, which is not a reliable proxy for whether the Signal cache was actually flushed. Instead make flush_inbound_commits_under_permit's empty path flush the cache unconditionally (it already did while draining; now also when the batcher went live between the caller's is_active() check and the permit). That makes "cache flushed under the permit" an unconditional post-condition, so the wrapper drops the racy recheck and collapses to a let-chain. Add a deterministic test (fail_flushes injection) proving the empty commit still flushes — the pre-fix empty+live path returned durable without flushing, so it now guards that regression observably.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 119-134: Fail closed whenever drain is active and the weak self
reference cannot upgrade, instead of falling through to the plain Signal flush
path. Update the logic in the adapter’s drain-handling branch around
`drain_active`, `self_weak.get()`, and `flush_inbound_commits_under_permit` so
that any active drain without a live client returns an error immediately,
regardless of `inbound_commit_batch.has_entries()`, preventing
`flush_signal_cache()` from running in that case.
🪄 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: 53970f28-2154-4a8f-bab8-5506a5d26203
📒 Files selected for processing (2)
src/client/adapters.rssrc/message/commit_batch.rs
There was a problem hiding this comment.
1 issue found across 2 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
…le doc The batch-safe flush's upgrade-failure fallback gated on has_entries() before falling through to a raw flush. But an empty drain can still hold dirty SKDM-only Signal advances with no rows, so that raw flush could persist them rowless — the loss this path exists to prevent. Fail closed for any active drain when the weak self ref can't upgrade (practically unreachable; the run loop holds a strong Arc) so the cache stays unflushed and the server redelivers. Also refresh the flush_inbound_commits_under_permit doc: an empty batch is no longer a no-op — it still flushes the Signal cache under the permit (idempotent when clean).
There was a problem hiding this comment.
0 issues found across 2 files (changes from recent commits).
Requires human review: Major refactor of inbound commit pipeline with new batching module, breaking changes to event types and the durability hook trait, and complex concurrency logic across the drain lifecycle. Requires human review of correctness and safety invariants.
Re-trigger cubic
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 4b672d6904
ℹ️ 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".
|
|
||
| client | ||
| .flush_signal_cache_logged("identity change", None) | ||
| .flush_signal_cache_batch_safe_logged("identity change", None) |
There was a problem hiding this comment.
When an identity-change notification arrives while the offline drain is active, this call can route into flush_inbound_commits_under_permit and commit the accumulated inbound batch, including awaiting the user durability hook, while the stanza/router read loop is still inside this handler. If that hook waits on any client request that needs a server response (or is just slow), the read loop is parked and cannot process the response/pongs; this is the same deadlock class the offline finisher avoids by running off-loop. Defer this batch-safe flush to a spawned task or avoid committing drain batches from read-loop handlers.
Useful? React with 👍 / 👎.
2 tasks
This was referenced Jul 11, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Production tracing showed the offline drain paying every durability cost per message while globally serialized (1-permit semaphore): one durability-hook commit (~2.8 ms on a Postgres-backed hook), two pending-buffer transactions, one Signal-cache flush and one spawned event per message — a 421-message backlog spent ~1.2 s in per-message round-trips.
WA Web batches all of this, verified in the captured JS:
WAWebMessageProcessorCacheaccumulates decrypted messages with three flush triggers: size (web_message_processing_cache_size), timeout (web_offline_message_processor_timeout_seconds, ShiftTimer) and forced flush at end of drain.createSnapshotcommits per batch, in strict order: bulk message-table write → bulk signal-store commit under lock (bulkPutSession/bulkPutSenderKey/…) → only thensendAggregateOfflineReceipts(batch).Our per-message hook was the parity divergence — not the batch. This PR closes it end-to-end: everything per-message in the drain pipeline now amortizes over the batch, with the same ordering guarantee (nothing acked or observable before it is durable).
Design
New
InboundCommitBatcher(src/message/commit_batch.rs), active during the drain:delayMs: 3000), forced flush at end of drain and on disconnect/reconnect.createSnapshotordering): pending-inbound buffer (one SQLite txn, rows encoded into a single reused arena) → Signal-cache flush →hook.on_messages(batch)→ buffer clear (one txn) → acks →Event::Messages. Any failure leaves the whole batch unacked for redelivery.activestate that only the end-of-drain flush flips, while holding the single processing permit — no stanza can straddle the transition, and stanzas queued behind the permit commit strictly after the tail batch, preserving arrival order across the boundary. The end-of-drain finisher runs off the read loop (a slow hook must not starve pongs/IQs), guarded by the connection generation.Failure handling (hardened through review)
The core invariant, enforced end-to-end: a Signal ratchet advance is never persisted without a durable buffered row for its message — otherwise a crash turns the server's redelivery into an acked duplicate (silent loss for hook consumers). The machinery, added across an adversarial self-review plus several bot-review rounds:
ReinsertGuardrestores taken entries when a commit is cancelled (bounded teardown) or fails before its durable point; re-storing rows is idempotent, so a transient Signal-flush failure after the row write keeps the batch retryable in-session.pending_live): the batcher stays in drain mode at one permit, a generation-scoped retry task reflushes every 3 s, and the first durable flush completes the transition (deactivate + widen + buffered-receipt flush together). Completion is still published immediately, so startup waiters never hang.flush_signal_cache_batch_safe, which commits the pending batch under the permit first; session/chain locks are released before it to avoid lock inversion with the pipeline.cleanup_connection_statebumps the connection generation first (lane workers stop draining; a post-permit re-check inprocess_classified_messagecatches stragglers before any ratchet mutation), then commits + settles the Signal cache in one permit-held section. The cache drop is coupled to the batcher resets that drop entries — rowless advances never outlive their entries, while flush-failure-retained committed state survives for the next successful flush.Newsletters bypass the batcher/hook entirely (their plaintext stanza is acked at enqueue and never redelivered — gating them on the hook could permanently drop them) and dispatch as event-only
Event::Messages, which the hook docs state.Breaking changes (pre-1.0)
InboundDurabilityHook::on_message(client, info, message)on_messages(client, &[InboundMessage])— all-or-nothing, slice order; batch of one on live trafficEvent::Message(Arc, Arc)Event::Messages(MessageBatch { messages: Arc<[InboundMessage]>, origin: Live | OfflineDrain })— Baileys'messages.upsertshapeEventKind::MessageEventKind::MessagesEvent::as_message()Event::as_messages()(whole batch) /Event::messages()(message iterator)MessageContext::from_eventMessageContext::from_inboundInboundMessage,MessageBatchandBatchOriginare exported from the prelude.MessageBatchis a first-class collection: it derefs to nothing but exposesiter()/len()/is_empty()/first()andIntoIterator for &MessageBatch, sofor msg in &batchandbatch.len()work directly;messages: Arc<[InboundMessage]>stays public for zero-copy sharing. The event slice is the same allocation handed to the hook: what the hook committed is exactly what event consumers observe, in the same order — a consumer never sees a message whose commit failed (a hook that only succeeds on redelivery replay dispatches its event there).Bot::on_messagekeeps its per-message closure signature by fanning out the batch in order, so simple bots migrate with zero changes.ProtocolStoregainsstore_pending_inbound_batch/delete_pending_inbound_batchwith defaults that iterate the single-row methods (third-party backends keep working);SqliteStoreoverrides both with one transaction per batch. The single-row methods stay: the redelivery replay and the BotBuilder capability probe are inherently single-row.What this buys (421-message drain, from production numbers)
Vecallocationssynchronous_commit=on)Live-path cost is unchanged vs main (single commit, immediate; the batch slice is built in one allocation — measured 2.0→1.0 allocs, −18ns for that step). Delivery receipts and message secrets were already batched (offline receipt aggregation,
MsgSecretWriteBuffer) — this closes the remaining four. Bare<receipt>stanzas stay 1:1, matching WA Web (its pre-ack batcher batches persistence, not socket sends).Validation
cargo clippy -D warningsclean on all crates and feature combinations, including--all-featuresand the wasm32 release build.text+~45 KiB over the 32 KiB per-PR budget — accepted feature cost (size-increase-oklabel); llvm-lines show it is new code, not monomorphization (+1.2% copies)wa.recv.commit_batch(field:count) replaces per-message hook time in dashboards