From bb74d009cbd1f2f519639ab7edb17d1975c3e6a6 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 10 Aug 2026 21:48:40 +0000 Subject: [PATCH 1/2] docs(perf): inventory per-client retention and bounds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audits a published per-client retained-memory technique against this library: store encoded bytes instead of object graphs, allocate on first use, and cap every cache a workload or a peer can grow. The first two already hold here; the third holds nearly everywhere, and the audit's value is the written inventory of which bound is which and why the uncapped collections stay uncapped. Measured at the global allocator rather than in RssAnon, because a 2 KiB structure and a 3 KiB one both round to the same page count. Identical under dev and release, median of the tail over 16 clients: Client + HTTP client 24 267 B + empty InMemoryBackend 26 211 B major_sync_task_sender 2 816 B (32 slots, ~11% of a client) transport events 3 840 B (64 slots, per connection) Both queues together are ~1.2% of a connected session's ~530 KiB, so neither is made lazy. A capacity cap costs nothing at construction: PortableCache starts on an empty map, so capacity 1 and capacity 10 000 both retain 248 B. The one gap the audit found is in the instrument, not the memory: memory_report() could not see the transient inbound retention, which is the largest thing a client holds — inbound_commit_batch accumulates to 4 MiB of decoded protos, two orders of magnitude above every cache in the report. It, msg_secret_buffer and pending_device_sync now report, under their own section. No bound is added, changed or exposed. report_coverage.rs missed all three because a collection behind a newtype (PendingDeviceSync) or an alias (type Pending = HashMap<..>) names nothing growable in the field's own type. The scan now resolves crate-local types one level, aliases expanded on both the field and the definitions it reaches. One level, not transitively: self_weak makes the type graph reach every collection from every field. --- agent_docs/observability.md | 156 +++++++++++++- src/client.rs | 51 ++++- src/client/accessors.rs | 8 + src/client/tests.rs | 50 +++++ src/message/commit_batch.rs | 65 ++++++ src/msg_secret_buffer.rs | 5 +- src/pending_device_sync.rs | 13 ++ tests/e2e/tests/per_client_retention.rs | 263 ++++++++++++++++++++++++ tests/report_coverage.rs | 220 +++++++++++++++++++- 9 files changed, 817 insertions(+), 14 deletions(-) create mode 100644 tests/e2e/tests/per_client_retention.rs diff --git a/agent_docs/observability.md b/agent_docs/observability.md index 5060660b7..a92d2eb4a 100644 --- a/agent_docs/observability.md +++ b/agent_docs/observability.md @@ -403,7 +403,161 @@ Not everything is expressible. The noise sender task's batch buffer grows to can read it without a channel built for the purpose; it stays unreported rather than guessed. The safety net for the parts that *are* reachable is `tests/report_coverage.rs`, which parses `Client`'s fields and fails when one -whose type names a collection never reaches `memory_report()`. +that reaches a collection — in its own type, or through one crate-local type it +names, aliases resolved — never reaches `memory_report()`. Resolution stops at +one level on purpose: `self_weak: Weak` makes the type graph reach every +collection from every field, and a guard that flags everything flags nothing. +Fields past that boundary are listed in the file's `EXEMPT` with their reason. + +## Per-client retention and cache bounds + +Two questions a multi-session consumer asks that the sections above do not +answer directly: what does one more `Client` retain before it does anything, and +which of its collections can a workload or a peer grow without limit. Both were +audited in full; this is the result, so nobody re-derives it. + +### What one client retains at construction + +Measured at the global allocator (`tests/e2e/tests/per_client_retention.rs`, +`#[ignore]`d, 16 clients, median of the tail) rather than in `RssAnon`, because a +2 KiB structure and a 3 KiB one both round to the same page count. Identical +under `dev` and `release`: + +| what | retained | +| --- | ---: | +| `Client` + `UreqHttpClient` | 24 267 B | +| + `InMemoryBackend`, empty | 26 211 B | + +Quote the pair, not the two halves: ureq materializes part of its agent lazily, +so a few hundred bytes land on either side of the boundary between building the +HTTP client and building the client that takes it, and the split moves between +runs while the sum holds to within a handful of bytes. The client's own share is +~22.4 KiB and the HTTP client's ~1.9 KiB, which is the right granularity to +reason at and the wrong one to regression-test. + +Against that, the two preallocated bounded queues a session owns: + +| queue | capacity | payload | retained | +| --- | ---: | ---: | ---: | +| `major_sync_task_sender` (`Client::new`) | 32 | 56 B | 2 816 B | +| transport events (`EVENT_CHANNEL_CAPACITY`, per connection) | 64 | 40 B | 3 840 B | + +So the sync queue is ~11% of a constructed client — but a connected session's +marginal cost is ~530 KiB (see the table further up), against which both queues +together are ~1.2%. **Neither is worth making lazy.** The sync queue's receiver +is handed to `Bot::build` to spawn its worker, so deferring the allocation means +an `Option` plus a builder handoff that no longer has a receiver to give; the +transport queue is created by the transport at connect, and every connected +session drains it. Paying an indirection on a per-connection path to defer 0.5% +of a session is the wrong trade. + +**A capacity cap is a bound here, not a reservation.** `PortableCache` starts on +an empty `HashMap`: capacity 1 and capacity 10 000 both retain 248 B until +entries arrive (pinned by `cache_capacity_is_not_preallocated`). Raising a cap +costs nothing at construction, which is why the coordination caches can afford +to be sized generously. + +### Bytes vs. object graphs, and first-use allocation + +Both already hold, so neither is an open question: + +- The retry cache (`recent_messages`) is `Cache>>` — + encoded protobuf, never a `waproto::` graph — and its default capacity is 0, + so the DB is the only copy unless a consumer opts into the L1. +- First-use allocation is already the pattern for everything whose cost is worth + deferring: `group_cache`, `app_state_processor`, `delivery_receipt_queue`, + `transport_ack_queue` and `custom_enc_handlers` are `OnceLock`s built on first + use, not in the constructor. + +The one place decoded protos are retained is `inbound_commit_batch`, and there +the decode is the point: the entries are dispatched to the consumer and handed +to the durability hook as `wa::Message`. Re-encoding them to save bytes would +add a decode per delivery to a path that already holds the batch for +milliseconds. It is bounded at 400 messages / 4 MiB instead — a *flush +threshold*, not a hard ceiling: `maybe_flush_inbound_commits` checks it after the +entry is inserted, so a batch overshoots by its last message, and one very large +message can overshoot substantially on its own. + +### What is bounded, and by what + +| collection | bound | what eviction costs | +| --- | --- | --- | +| `group_cache` | 1h TTL, 250 | re-query on miss | +| `device_registry_cache` | 1h TTL, 5 000 | store stays authoritative | +| `recent_messages` | 5m TTL, 0 (disabled) | DB is authoritative | +| `message_retry_counts` | 1h TTL, 500, FIFO | a forgiven `MAX_DECRYPT_RETRIES` — see below | +| `undecryptable_dispatched` | 5m TTL, 1 000 | a duplicate event | +| `pdo_pending_requests` / `pdo_requested` | 30s TTL, 200 / 24h TTL, 512 | a repeated PDO request | +| `sender_key_devices_cache` | 1h TTI, 500 | a redundant SKDM | +| `session_recreate_history` | 1h TTL, 256 | one un-throttled recreate | +| `session_locks` / `chat_lanes` / `group_distribution_locks` | 10 000 / 5 000 / 512 | nothing: an `evict_guard` refuses to evict a lock a task holds, so the map briefly exceeds capacity instead of minting a second lock for one key | +| `resend_rate_limiter` | 4 096, FIFO | fail-open by design — an evicted bucket is recreated full, so undersizing forgives rate, never over-throttles | +| `group_devices_memo` / `skdm_warm_memo` / `dm_devices_memo` | 64 / 64 / 512 | a recompute | +| `SignalStoreCache` sessions / identities / sender keys | 2 000 each (+1/8 slack before an eviction scan) | nothing: only *clean* entries are evicted, so an unpersisted record is never dropped | +| `SignalStoreCache::sender_key_locks` | 2 000, idle-only | nothing: only locks held solely by the map are dropped | +| `inbound_commit_batch` | 400 messages / 4 MiB, checked after insert | commits early, no loss; overshoots by one message | +| `msg_secret_buffer` | 4 096, except on cancellation | nothing: a producer that would exceed it parks on `capacity_available`, and a cancelled one force-buffers past the mark rather than losing captures | +| device-topology changed-users log | 256 | a memo recompute; overflow can never serve stale data | +| `AbPropsCache` | the compile-time `WATCHED` interest set | server props outside it are discarded at parse | +| `CallRegistry` pre-offer controls / ringing group calls / event queues | 64 entries or 1 MiB each | fail-closed admission | +| `PendingCallLinkJoins` transitions | 32 | fail-closed | +| `major_sync_task_sender` / transport events / noise send jobs | 32 / 64 / 8 | backpressure, no loss | + +### What is unbounded, and why it stays that way + +Every one of these except the last reaches `memory_report()`, which is the point: +the bound is a drain or a lifecycle, so the count is the only warning available. + +- **`lid_pn_cache`** — deliberate, and pinned by a test. Evicting a still-valid + mapping silently downgrades Signal addresses to `@c.us`; WA Web's + `WAWebLidPnCache` is plain `Map`s with no expiry either. +- **`app_state_key_requests`** — swept by deadline (`retain`) on every insert + path, so a busy client holds only key ids whose retry deadline has not passed. + The sweep is lazy, not timed: a client that requests keys and then goes idle + keeps those stamps until the next request or a reconnect clears the map. No + capacity cap, because dropping a stamp either re-asks the phone for a key + already requested or loses the dedup that keeps a stuck sender from re-asking + every few seconds. Growth is self-limiting anyway: each new key id costs a peer + message on the wire. +- **`pending_device_sync`** — one entry per distinct user seen with an unknown + device. Offline entries are drained by `doPendingDeviceSync` at the end of the + backlog; entries the *online* path adds are removed only by that same drain or + by teardown, so a connection that never drains keeps them for its lifetime, + which also suppresses a second immediate refresh for those users. A cap would + skip a device refresh and leave the next send to that user addressed to a stale + device list, so the fix if this ever matters is a removal on the online path, + not a ceiling. +- **`pending_retries`** — held only for the duration of one retry receipt (a + `scopeguard` removes it), so the bound is concurrent receipts. +- **`presence_subscriptions`**, **`response_waiters`**, **`node_waiters`**, + **`sent_node_waiters`**, **`stanza_interceptors`** — one entry per thing the + *application* asked for. Not a peer-driven growth vector. +- **`transport_ack_queue`** / **`delivery_receipt_queue`** — unbounded + `async_channel`s whose depth is a stalled-transport signal; capping them would + drop acks the server is waiting for. +- **`offline_receipt_buffer`** — drained at the end of every offline batch, and + the one entry here the report does *not* count: it is listed in + `report_coverage.rs`'s `EXEMPT` because its `MessageInfo` values are already + attributed where the batch owns them. Its depth during a drain is therefore + invisible; if that ever matters, it needs a field of its own rather than a cap. + +Two design rules the audit confirmed, and one place where the second does not +hold as tightly as the code comments suggest: + +1. **A cap must not evict state something is relying on.** The `evict_guard` on + the coordination caches and the clean-only eviction in `SignalStoreCache` are + the same idea applied twice: exceed capacity rather than break an invariant. +2. **A rate limiter should fail closed, not evict.** `message_retry_counts` is + what enforces `MAX_DECRYPT_RETRIES`, and forgetting a counter forgives the cap + it exists to apply. Its 1h TTL is chosen for exactly that (a 5m TTL expired + between reconnects, so the count never reached the cap) — but its 500-entry + capacity is a plain FIFO eviction with no guard, so **more than 500 distinct + retry keys inside the hour does forgive the ceiling**: the evicted key's next + decrypt failure re-enters `increment_retry_count` on the `None => 1` arm. + Left as is deliberately — the counters are two integers, so the honest fix is + a larger capacity rather than a mechanism, and no workload has been measured + against 500 concurrent retry keys. Recorded here so the next person measures + instead of re-deriving. ## Relation to the `metrics`/`tracing` features diff --git a/src/client.rs b/src/client.rs index 43e9e3883..846e07547 100644 --- a/src/client.rs +++ b/src/client.rs @@ -433,6 +433,35 @@ pub struct MemoryReport { pub history_sync_tasks_peak: u64, /// Lifetime high-water mark of logical compressed-payload bytes. pub history_sync_payload_bytes_peak: u64, + // -- Transient retention (accumulated, not yet handed on) -- + /// Inbound messages accumulated for the next per-batch commit, and the + /// encoded-byte sum compared against the 4 MiB flush threshold. The decoded + /// protos this holds are the largest per-client allocation in this report by + /// two orders of magnitude. + /// + /// "Accumulated", not "resident": a batch already handed to its commit is + /// still in memory but no longer counted here (see + /// `InboundCommitBatcher::pending_stats`). Live traffic commits + /// immediately, so outside an offline drain this is normally zero. + pub inbound_commit_batch: CollectionStats, + /// `messageSecret` captures buffered for write-behind persistence — from + /// live receives and sends as well as an offline drain, so a slow backend + /// can saturate this with no drain in progress. + /// + /// A producer that would exceed the 4096-entry limit waits for an in-flight + /// write rather than the buffer growing. That limit is not a hard ceiling: + /// a queueing future cancelled while backpressured force-buffers what it + /// still holds rather than losing it, so this can read above the limit + /// during teardown. + pub msg_secret_buffer: usize, + /// Users awaiting a device-list refresh, and the dedup that suppresses a + /// second refresh for the same user while one is outstanding. + /// + /// Offline entries are drained by `doPendingDeviceSync` at the end of the + /// backlog. Entries added by the *online* path are removed only by that + /// same drain or by teardown, so on a connection with no offline drain this + /// grows with the distinct users seen with an unknown device. + pub pending_device_sync: usize, // -- Capacity-only caches (coordination, counts only) -- pub session_locks: u64, pub chat_lanes: u64, @@ -509,7 +538,7 @@ pub struct MemoryReport { impl MemoryReport { /// Common byte-carrying collections used by both totals and `Display`. /// Feature-specific collections stay beside their gated report section. - fn collections(&self) -> [(&'static str, &CollectionStats); 12] { + fn collections(&self) -> [(&'static str, &CollectionStats); 13] { [ ("group_cache:", &self.group_cache), ("device_registry_cache:", &self.device_registry_cache), @@ -523,6 +552,7 @@ impl MemoryReport { ("signal_identities:", &self.signal_identities), ("signal_sender_keys:", &self.signal_sender_keys), ("history_sync_tasks:", &self.history_sync_tasks), + ("inbound_commit_batch:", &self.inbound_commit_batch), ] } @@ -549,11 +579,14 @@ impl std::fmt::Display for MemoryReport { writeln!(f, " {name:<22} {:>7} entries {:>10} B", c.entries, c.bytes) } // First TTL_BOUNDED entries of collections() are the TTL-bounded - // caches; the next SIGNAL_CACHES are Signal store caches. The final - // entry is transient history-sync retention. Adding a cache to - // collections() means moving this boundary, or the sections shift. + // caches; the next SIGNAL_CACHES are Signal store caches. The last two + // are transient retention: history sync, then the inbound commit batch. + // Adding a cache to collections() means moving this boundary, or the + // sections shift. const TTL_BOUNDED: usize = 8; const SIGNAL_CACHES: usize = 3; + const HISTORY_SYNC: usize = TTL_BOUNDED + SIGNAL_CACHES; + const COMMIT_BATCH: usize = HISTORY_SYNC + 1; let collections = self.collections(); writeln!(f, "=== Memory Report ===")?; writeln!(f, "--- TTL-bounded caches ---")?; @@ -627,11 +660,7 @@ impl std::fmt::Display for MemoryReport { )?; } writeln!(f, "--- In-flight history sync ---")?; - line( - f, - collections[TTL_BOUNDED + SIGNAL_CACHES].0, - &self.history_sync_tasks, - )?; + line(f, collections[HISTORY_SYNC].0, &self.history_sync_tasks)?; writeln!( f, " peak tasks: {}", @@ -642,6 +671,10 @@ impl std::fmt::Display for MemoryReport { " peak payload storage: {} B", self.history_sync_payload_bytes_peak )?; + writeln!(f, "--- Transient retention ---")?; + line(f, collections[COMMIT_BATCH].0, &self.inbound_commit_batch)?; + writeln!(f, " msg_secret_buffer: {}", self.msg_secret_buffer)?; + writeln!(f, " pending_device_sync: {}", self.pending_device_sync)?; #[cfg(feature = "plugins")] { writeln!(f, "--- Plugins ---")?; diff --git a/src/client/accessors.rs b/src/client/accessors.rs index ba60ea889..b858ffcad 100644 --- a/src/client/accessors.rs +++ b/src/client/accessors.rs @@ -349,6 +349,11 @@ impl Client { .len(); let app_state_key_requests = self.app_state_key_requests.lock().await.len(); let app_state_syncing = self.app_state_syncing.len(); + let (commit_batch_entries, commit_batch_bytes) = self.inbound_commit_batch.pending_stats(); + let inbound_commit_batch = + CollectionStats::new(commit_batch_entries as u64, commit_batch_bytes as u64); + let msg_secret_buffer = self.msg_secret_buffer.pending_len(); + let pending_device_sync = self.pending_device_sync.len(); let chatstate_handlers = self.chatstate_handler_count.load(Ordering::Acquire); let history_sync_activity = self.history_sync_activity.snapshot(); let history_sync_tasks = CollectionStats::new( @@ -427,6 +432,9 @@ impl Client { history_sync_tasks, history_sync_tasks_peak: history_sync_activity.tasks_peak as u64, history_sync_payload_bytes_peak: history_sync_activity.payload_bytes_peak as u64, + inbound_commit_batch, + msg_secret_buffer, + pending_device_sync, session_locks: self.session_locks.entry_count(), chat_lanes: self.chat_lanes.entry_count(), group_distribution_locks: group_distribution_locks.entries, diff --git a/src/client/tests.rs b/src/client/tests.rs index bdba35a25..653765b88 100644 --- a/src/client/tests.rs +++ b/src/client/tests.rs @@ -4844,6 +4844,56 @@ async fn memory_report_display_sections_stay_aligned() { "{name} must render under the Signal heading, got:\n{rendered}" ); } + + // The last two `collections()` entries are transient retention, one section + // each. Their order is what the two boundary constants encode, so a cache + // appended to `collections()` without moving them lands here. + let history_start = rendered + .find("--- In-flight history sync ---") + .expect("history sync section"); + let drain_start = rendered + .find("--- Transient retention ---") + .expect("transient-retention section"); + assert!( + history_start < drain_start, + "sections must render in `collections()` order, got:\n{rendered}" + ); + assert!( + rendered[history_start..drain_start].contains("history_sync_tasks:"), + "history_sync_tasks must render under its own heading, got:\n{rendered}" + ); + for name in [ + "inbound_commit_batch:", + "msg_secret_buffer:", + "pending_device_sync:", + ] { + assert!( + rendered[drain_start..].contains(name), + "{name} must render under the transient-retention heading, got:\n{rendered}" + ); + } +} + +/// The offline unknown-device queue has no capacity cap: its bound is the drain +/// that empties it, since dropping a user would leave sends to them addressed to +/// a stale device list. That makes the count the only warning a consumer gets, +/// so it has to reach the report. +#[tokio::test] +async fn memory_report_counts_the_offline_device_sync_queue() { + let client = + crate::test_utils::create_test_client_with_name("pending_device_sync_report").await; + assert_eq!(client.memory_report().await.pending_device_sync, 0); + + let jid: Jid = "559980000002@s.whatsapp.net".parse().expect("a test jid"); + assert!(client.pending_device_sync.add(&jid)); + assert!( + !client.pending_device_sync.add(&jid), + "the queue dedups per user, so a retry storm from one sender adds one entry" + ); + assert_eq!(client.memory_report().await.pending_device_sync, 1); + + client.pending_device_sync.take_all(); + assert_eq!(client.memory_report().await.pending_device_sync, 0); } #[tokio::test] diff --git a/src/message/commit_batch.rs b/src/message/commit_batch.rs index cbebb2f64..b49878b12 100644 --- a/src/message/commit_batch.rs +++ b/src/message/commit_batch.rs @@ -137,6 +137,32 @@ impl InboundCommitBatcher { !self.lock().entries.is_empty() } + /// Accumulated entries and the encoded bytes they account for, for + /// `memory_report()`. + /// + /// This is the largest thing one client retains: a drain batch holds + /// `MAX_BATCH_BYTES` of decoded protos, two orders of magnitude above every + /// other collection in the report. The byte figure is the same + /// `message_encoded_len` sum the flush threshold is compared against, so it + /// is a proxy for the decoded footprint, not a measurement of it — the + /// decoded `wa::Message` graph is larger than its wire form. + /// + /// Two exclusions, both deliberate: + /// + /// - **A batch inside its commit.** [`Self::take`] empties this state + /// before `commit_inbound_batch` awaits the backend, the Signal flush and + /// the durability hook, all of which still hold the messages. Counting + /// them would mean re-encoding the batch to price it on the commit path — + /// real work on the receive path to sharpen a report — so this reads + /// "waiting to be committed", not "resident". `has_entries` draws the + /// same line, for the same reason. + /// - **The reusable encode arena**, whose lock a drain commit holds across + /// its backend write; a report must not queue behind one. + pub(crate) fn pending_stats(&self) -> (usize, usize) { + let state = self.lock(); + (state.entries.len(), state.bytes) + } + /// Switch to immediate (live) commits. Only the end-of-drain flush (or a /// later flush completing a deferred transition) calls this, while /// holding the processing permit. @@ -958,6 +984,45 @@ mod tests { } } + // The accumulating batch is the largest allocation one client holds (4 MiB + // of decoded protos at the flush threshold, two orders of magnitude above + // every cache in the report), so it has to be visible while it accumulates + // rather than only in aggregate afterwards. The flush assertion pins the + // other half of `pending_stats`'s contract: it counts what is waiting to be + // committed, so handing the batch to a commit clears it. + #[tokio::test] + async fn memory_report_names_the_uncommitted_drain_batch() { + let client = create_test_client_with_failing_http("batch_report").await; + client.inbound_commit_batch.reset(); + + let empty = client.memory_report().await; + assert_eq!(empty.inbound_commit_batch.entries, 0); + assert_eq!(empty.inbound_commit_batch.bytes, 0); + + for id in ["R1", "R2"] { + client.commit_or_batch_inbound(item(id), false).await; + } + + let held = client.memory_report().await; + assert_eq!(held.inbound_commit_batch.entries, 2); + assert!( + held.inbound_commit_batch.bytes > 0, + "accumulated entries must carry their encoded-byte estimate" + ); + assert!( + held.total_estimated_bytes() >= held.inbound_commit_batch.bytes, + "the batch must reach the report's byte total, not only its own field" + ); + + client + .flush_inbound_commits_under_permit(false, None, None) + .await; + + let flushed = client.memory_report().await; + assert_eq!(flushed.inbound_commit_batch.entries, 0); + assert_eq!(flushed.inbound_commit_batch.bytes, 0); + } + // Live traffic commits immediately as a batch of one. #[tokio::test] async fn live_commits_as_batch_of_one() { diff --git a/src/msg_secret_buffer.rs b/src/msg_secret_buffer.rs index fccfe71f1..82c312373 100644 --- a/src/msg_secret_buffer.rs +++ b/src/msg_secret_buffer.rs @@ -431,7 +431,10 @@ impl MsgSecretWriteBuffer { } } - #[cfg(test)] + /// Captures buffered but not yet persisted. Bounded by the pending limit + /// (`MAX_PENDING_MSG_SECRETS` in production): a producer that would exceed + /// it parks on `capacity_available` rather than the buffer growing, so this + /// count is a saturation gauge as much as a memory figure. pub(crate) fn pending_len(&self) -> usize { self.pending.lock().unwrap_or_else(|p| p.into_inner()).len() } diff --git a/src/pending_device_sync.rs b/src/pending_device_sync.rs index 3ebf0476a..596eb7eea 100644 --- a/src/pending_device_sync.rs +++ b/src/pending_device_sync.rs @@ -38,6 +38,19 @@ impl PendingDeviceSync { self.lock().drain().collect() } + /// Users queued for the next `doPendingDeviceSync`, plus the users an + /// online refresh already covered — [`Self::add`] is the dedup for both, and + /// only [`Self::take_all`] and [`Self::clear`] remove anything, so an entry + /// added by the online path stays until the next drain or teardown. + /// + /// The set has no capacity cap, and a cap would be the wrong fix: dropping a + /// user silently skips a device refresh and leaves the next send to them + /// addressed to a stale device list. The bound is the drain, not a number — + /// which is why the count belongs in `memory_report()`. + pub(crate) fn len(&self) -> usize { + self.lock().len() + } + pub(crate) fn clear(&self) { self.lock().clear(); } diff --git a/tests/e2e/tests/per_client_retention.rs b/tests/e2e/tests/per_client_retention.rs new file mode 100644 index 000000000..817d4eba1 --- /dev/null +++ b/tests/e2e/tests/per_client_retention.rs @@ -0,0 +1,263 @@ +//! Per-client retained heap, measured at the allocator. +//! +//! `process_footprint.rs` answers the same question in `RssAnon`, which is +//! page-quantised: a 2 KiB structure and a 3 KiB one both read as "one page, or +//! none". This file counts bytes as the global allocator hands them out, so a +//! single preallocated queue is separable from the client that owns it. That is +//! the resolution an audit of *which* per-client structure costs what needs; +//! `RssAnon` remains the right instrument for what a consumer actually pays. +//! +//! Both tests are `#[ignore]`d: they are measuring tools, not guards. A number +//! worth guarding gets a deterministic observable instead (see +//! `agent_docs/observability.md`, "Should a residency probe be permanent?"). +//! +//! Run: `cargo nextest run --profile e2e -p e2e-tests --run-ignored all -E +//! 'binary(per_client_retention)' --no-capture` + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::Arc; +use std::sync::atomic::{AtomicIsize, Ordering}; + +use log::info; +use wacore::net::TransportEvent; +use wacore::store::InMemoryBackend; +use whatsapp_rust::bot::Bot; +use whatsapp_rust::portable_cache::PortableCache; +use whatsapp_rust::sync_task::MajorSyncTask; +use whatsapp_rust_ureq_http_client::UreqHttpClient; + +/// Live bytes handed out by the global allocator and not yet returned. +/// +/// Signed: a measurement window can free more than it allocates (a builder that +/// drops its inputs), and clamping that to zero would silently turn a net +/// release into "no change". +static LIVE: AtomicIsize = AtomicIsize::new(0); + +struct Counting; + +// SAFETY: every method forwards to `System` with the same layout it received +// and only adds a relaxed counter update, so the allocator contract is exactly +// `System`'s. +unsafe impl GlobalAlloc for Counting { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc(layout) }; + if !ptr.is_null() { + LIVE.fetch_add(layout.size() as isize, Ordering::Relaxed); + } + ptr + } + + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + LIVE.fetch_sub(layout.size() as isize, Ordering::Relaxed); + unsafe { System.dealloc(ptr, layout) } + } + + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + let new_ptr = unsafe { System.realloc(ptr, layout, new_size) }; + if !new_ptr.is_null() { + LIVE.fetch_add( + new_size as isize - layout.size() as isize, + Ordering::Relaxed, + ); + } + new_ptr + } +} + +#[global_allocator] +static ALLOC: Counting = Counting; + +fn live() -> isize { + LIVE.load(Ordering::Relaxed) +} + +/// Live-heap growth across `build`, with the built value kept alive. +/// +/// Returns the value so the caller decides when it dies; measuring a value that +/// has already been dropped would report zero for every structure. +fn retained(build: impl FnOnce() -> T) -> (T, isize) { + let before = live(); + let value = build(); + (value, live() - before) +} + +fn median(mut v: Vec) -> isize { + v.sort_unstable(); + v[v.len() / 2] +} + +/// What the two per-client bounded queues cost, and what fraction of a client +/// that is. +/// +/// The capacities are the constants the client and the Tokio transport build +/// with (`Client::new` and `EVENT_CHANNEL_CAPACITY`); they are restated here +/// because a channel's preallocation is not reachable from outside it, so the +/// only way to price one is to build an identical channel. +#[test] +#[ignore] +fn preallocated_queue_slots() { + let _ = env_logger::builder().is_test(true).try_init(); + + let (major_sync, major_sync_bytes) = retained(|| async_channel::bounded::(32)); + let (transport, transport_bytes) = retained(|| async_channel::bounded::(64)); + + info!( + "major_sync_task_sender capacity=32 payload={:>4} B retained={:>6} B", + size_of::(), + major_sync_bytes + ); + info!( + "transport events capacity=64 payload={:>4} B retained={:>6} B", + size_of::(), + transport_bytes + ); + + drop(major_sync); + drop(transport); + + // Both channels are preallocated in full at construction: a slot array, not + // a growing queue. If either ever reported ~0 the measurement would be + // reading a lazily-allocated buffer instead of the reservation. + assert!( + major_sync_bytes > 0 && transport_bytes > 0, + "a bounded channel must retain its slot array on construction" + ); +} + +/// A configured capacity is a bound, not a reservation. +/// +/// Worth measuring rather than assuming, because the sibling test above shows +/// the opposite case: a bounded channel *does* size its allocation by capacity. +/// `PortableCache` starts on an empty `HashMap`, so a client carrying +/// `session_locks_capacity = 10_000` pays for none of it until entries arrive — +/// raising a cap costs nothing at construction. +#[test] +#[ignore] +fn cache_capacity_is_not_preallocated() { + let _ = env_logger::builder().is_test(true).try_init(); + + let (small, small_bytes) = retained(|| { + PortableCache::::builder() + .max_capacity(1) + .build() + }); + let (large, large_bytes) = retained(|| { + PortableCache::::builder() + .max_capacity(10_000) + .build() + }); + + info!("cache capacity=1 retained={small_bytes:>6} B"); + info!("cache capacity=10000 retained={large_bytes:>6} B"); + + drop(small); + drop(large); + + assert_eq!( + small_bytes, large_bytes, + "a capacity cap must not size any allocation, or every cap becomes a per-client cost" + ); +} + +/// Retained heap per constructed `Client`, at allocator resolution, with the +/// pieces a consumer supplies priced separately. +/// +/// Needs no mock server: `build()` performs no I/O, so the transport factory +/// points at a port nothing listens on (same recipe as +/// `process_footprint.rs::client_construction_footprint`). +/// +/// Report the median of the tail, not the mean over all: the first clients pay +/// for process-wide lazily-built state (codec tables, TLS roots) that no later +/// client repeats, which is exactly the asymmetry a per-client figure must not +/// absorb. +/// +/// The counter is process-wide, so in principle a build window can absorb +/// allocation from an earlier client's background tasks — the clients stay alive +/// and the runtime is multi-threaded. The median over the tail is the guard +/// against that, and the check that it worked is in the output: the +/// `http+client` column comes out flat to within a few bytes across clients +/// 2..N, and the same under `dev` and `release`. Read that column and read the +/// table, not only the summary — a run whose tail is not flat is a run where +/// something else was allocating, and its median means less. +#[tokio::test(flavor = "multi_thread")] +#[ignore] +async fn constructed_client_retained_heap() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let n: usize = std::env::var("RETENTION_CLIENTS") + .ok() + .map(|raw| raw.parse().expect("RETENTION_CLIENTS is not a number")) + .unwrap_or(16); + assert!(n >= 3, "need a steady-state tail to take a marginal from"); + + let mut bots = Vec::with_capacity(n); + let mut backend_deltas = Vec::with_capacity(n); + let mut http_deltas = Vec::with_capacity(n); + let mut client_deltas = Vec::with_capacity(n); + + for i in 0..n { + // Built outside the client window and priced on their own: both are + // consumer-supplied and shareable (`with_http_client` takes a clone of + // one agent), so folding them into the client figure would report a + // cost a multi-session process may not pay per session. + let (backend, backend_bytes) = retained(|| Arc::new(InMemoryBackend::new())); + let (http, http_bytes) = retained(UreqHttpClient::new); + + let before = live(); + let bot = Bot::builder() + .with_backend_arc(backend) + .with_transport_factory( + whatsapp_rust_tokio_transport::TokioWebSocketTransportFactory::new() + .with_url("wss://127.0.0.1:1/ws/chat"), + ) + .with_http_client(http) + .with_runtime(whatsapp_rust::TokioRuntime) + .with_version((2, 3000, 0)) + .with_push_name(format!("retention_{i}")) + .build() + .await?; + let client_bytes = live() - before; + + bots.push(bot); + backend_deltas.push(backend_bytes); + http_deltas.push(http_bytes); + client_deltas.push(client_bytes); + info!( + "client {:>3} backend={:>7} B http={:>6} B client={:>8} B http+client={:>8} B", + i + 1, + backend_bytes, + http_bytes, + client_bytes, + http_bytes + client_bytes + ); + } + + let tail = |v: &[isize]| median(v.iter().copied().skip(2).collect()); + // `http` and `client` are reported together as well as apart: ureq + // materializes part of its agent lazily, so which of the two windows a + // given allocation lands in drifts by a few hundred bytes between runs + // while their sum is stable to the byte. Quote the sum. + info!( + "marginal (median 3..N) backend={:>7} B http={:>6} B client={:>8} B http+client={:>8} B", + tail(&backend_deltas), + tail(&http_deltas), + tail(&client_deltas), + tail( + &http_deltas + .iter() + .zip(&client_deltas) + .map(|(h, c)| h + c) + .collect::>() + ) + ); + info!( + "first client backend={:>7} B http={:>6} B client={:>8} B http+client={:>8} B", + backend_deltas[0], + http_deltas[0], + client_deltas[0], + http_deltas[0] + client_deltas[0] + ); + + assert_eq!(bots.len(), n); + Ok(()) +} diff --git a/tests/report_coverage.rs b/tests/report_coverage.rs index 68b273181..d585da336 100644 --- a/tests/report_coverage.rs +++ b/tests/report_coverage.rs @@ -6,7 +6,20 @@ //! alone, since they cost `size_of` and cannot drift. Collections can, so every //! field whose type names one is either walked by the report or listed below //! with the reason it is not. +//! +//! A field's own type expression is not the whole answer: a collection wrapped +//! in a newtype (`pending_device_sync: PendingDeviceSync`) names nothing +//! growable and used to pass unseen, which is how the biggest per-client +//! retention in the library — the inbound commit batch, which accumulates to +//! 4 MiB of decoded protos — stayed out of the report. So the scan also resolves +//! each field's crate-local types one level, aliases expanded, and looks at +//! *their* fields. +//! +//! One level, not transitively: `self_weak: Weak` makes the graph +//! reach every collection from every field, and a guard that flags everything +//! flags nothing. +use std::collections::HashMap; use std::path::{Path, PathBuf}; /// Type constructors that can retain an unbounded number of entries. A field @@ -33,6 +46,36 @@ const EXEMPT: &[(&str, &str)] = &[ "message_retry_counts", "counter-only map bounded by the retry ceiling; entries are two integers", ), + ( + "self_weak", + "a `Weak` back to this same client, so its collections are the \ + ones every other field already accounts for", + ), + ( + "plugin_host", + "walked by the feature-gated plugin section of the report (installed \ + plugins, tasks, subscriptions, endpoint queues), not by the common list", + ), + ( + "lifecycle", + "callbacks and connection scopes registered once at build by the \ + extension host; fixed for the client's lifetime, not workload-driven", + ), + ( + "stanza_router", + "one handler per protocol tag, populated by `create_stanza_router` at \ + construction and never written again", + ), + ( + "device_topology", + "the changed-users log is a fixed-length ring (TOPOLOGY_LOG_CAPACITY); \ + overflow degrades memos to a recompute rather than retaining more", + ), + ( + "media_conn", + "the host list from the most recent `` IQ — one server \ + response, replaced wholesale on refresh", + ), ]; fn manifest_path(relative: &str) -> PathBuf { @@ -64,10 +107,142 @@ fn type_idents(ty: &syn::Type, out: &mut Vec) { syn::Type::Paren(p) => type_idents(&p.elem, out), syn::Type::Group(g) => type_idents(&g.elem, out), syn::Type::Tuple(t) => t.elems.iter().for_each(|e| type_idents(e, out)), + syn::Type::Array(a) => type_idents(&a.elem, out), _ => {} } } +/// Every `.rs` file under `src/`, so a type can be looked up wherever it is +/// defined rather than only where it is used. +fn crate_sources(dir: &Path, out: &mut Vec) { + let entries = std::fs::read_dir(dir).unwrap_or_else(|e| panic!("read_dir {dir:?}: {e}")); + for entry in entries { + let path = entry.expect("a directory entry").path(); + if path.is_dir() { + crate_sources(&path, out); + } else if path.extension().is_some_and(|ext| ext == "rs") { + out.push(path); + } + } +} + +/// Type name -> the idents appearing in that type's own field types. +/// +/// Structs and enums both, since either can hold a collection. Names are not +/// qualified by module: two same-named types would merge their fields, which +/// can only over-report (a candidate that then has to be reported or exempted), +/// never hide one. +fn crate_type_fields() -> (HashMap>, HashMap>) { + fn collect( + items: &[syn::Item], + defs: &mut HashMap>, + aliases: &mut HashMap>, + ) { + for item in items { + let (name, field_types) = match item { + syn::Item::Struct(item) => ( + item.ident.to_string(), + item.fields.iter().map(|f| f.ty.clone()).collect::>(), + ), + syn::Item::Enum(item) => ( + item.ident.to_string(), + item.variants + .iter() + .flat_map(|v| v.fields.iter().map(|f| f.ty.clone())) + .collect(), + ), + syn::Item::Type(item) => { + // An alias is a spelling, not a type: `type Pending = + // HashMap<..>` must not let a field hide a map behind one. + let entry = aliases.entry(item.ident.to_string()).or_default(); + type_idents(&item.ty, entry); + continue; + } + syn::Item::Mod(item) => { + if let Some((_, inner)) = &item.content { + collect(inner, defs, aliases); + } + continue; + } + _ => continue, + }; + let entry = defs.entry(name).or_default(); + for ty in &field_types { + type_idents(ty, entry); + } + } + } + + let mut files = Vec::new(); + crate_sources(&manifest_path("src"), &mut files); + let mut defs = HashMap::new(); + let mut aliases = HashMap::new(); + for file in &files { + let text = std::fs::read_to_string(file).unwrap_or_else(|e| panic!("read {file:?}: {e}")); + let parsed = syn::parse_file(&text).unwrap_or_else(|e| panic!("parse {file:?}: {e}")); + collect(&parsed.items, &mut defs, &mut aliases); + } + + // Substitute aliases so the lookup below stays one level deep. Aliases are + // resolved among themselves first (two rounds cover an alias of an alias, + // the deepest chain here), then applied to every type's field idents. + fn expand(idents: &[String], aliases: &HashMap>) -> Vec { + idents + .iter() + .flat_map(|ident| match aliases.get(ident) { + // A self-referential spelling (`type Cache = Cache<..>`) would + // otherwise expand forever without adding anything. + Some(target) if target != std::slice::from_ref(ident) => target.clone(), + _ => vec![ident.clone()], + }) + .collect() + } + + for _ in 0..2 { + let snapshot = aliases.clone(); + for idents in aliases.values_mut() { + *idents = expand(idents, &snapshot); + } + } + for idents in defs.values_mut() { + *idents = expand(idents, &aliases); + } + // The alias map is returned too: a `Client` field can name one directly + // (`group_cache: OnceLock>` over `type GroupCache = + // TypedCache<..>`), and expanding only inside type definitions would leave + // that field invisible — removing it from the report would keep this green. + (defs, aliases) +} + +/// `idents` with every alias replaced by what it stands for. +fn resolve_aliases(idents: &[String], aliases: &HashMap>) -> Vec { + idents + .iter() + .flat_map(|ident| match aliases.get(ident) { + Some(target) if target != std::slice::from_ref(ident) => target.clone(), + _ => vec![ident.clone()], + }) + .collect() +} + +/// How a field reaches a growable: directly in its own type, or through one +/// crate-local type it names. Returned for the failure message, so a new +/// candidate says which collection put it on the list. +fn growable_path(idents: &[String], defs: &HashMap>) -> Option { + if let Some(direct) = idents.iter().find(|i| GROWABLE.contains(&i.as_str())) { + return Some(direct.clone()); + } + for ident in idents { + let Some(fields) = defs.get(ident) else { + continue; + }; + if let Some(inner) = fields.iter().find(|i| GROWABLE.contains(&i.as_str())) { + return Some(format!("{ident}::{inner}")); + } + } + None +} + /// Whether `name` occurs in `text` as a whole identifier. A plain substring /// search would let a field called `cache` pass on any mention of `group_cache`; /// requiring the boundaries keeps that from counting. @@ -134,6 +309,7 @@ fn client_struct(text: &str) -> syn::ItemStruct { fn every_growable_client_field_reaches_the_memory_report() { let client = client_struct(&read("src/client.rs")); let report = memory_report_body(&read("src/client/accessors.rs")); + let (defs, aliases) = crate_type_fields(); let mut missing = Vec::new(); for field in &client.fields { @@ -142,14 +318,15 @@ fn every_growable_client_field_reaches_the_memory_report() { }; let mut idents = Vec::new(); type_idents(&field.ty, &mut idents); - if !idents.iter().any(|i| GROWABLE.contains(&i.as_str())) { + let idents = resolve_aliases(&idents, &aliases); + let Some(via) = growable_path(&idents, &defs) else { continue; - } + }; if EXEMPT.iter().any(|(exempt, _)| *exempt == name) { continue; } if !mentions(&report, &name) { - missing.push(format!(" {name}: {}", idents.join("<"))); + missing.push(format!(" {name}: {} (via {via})", idents.join("<"))); } } @@ -180,6 +357,43 @@ fn a_mention_is_a_whole_identifier() { assert!(!mentions("self.group_devices_memo", "devices_memo")); } +/// The capability the scan gained: a collection wrapped in a crate-local type, +/// or spelled through an alias, still puts its field on the candidate list. +#[test] +fn a_newtype_or_alias_does_not_hide_its_collection() { + let (defs, aliases) = crate_type_fields(); + let path = |ident: &str| growable_path(&resolve_aliases(&[ident.to_string()], &aliases), &defs); + + // A field that names an alias directly — `group_cache: OnceLock>` — resolves through it, so dropping that cache from the + // report fails the guard rather than passing silently. + assert_eq!( + path("GroupCache").as_deref(), + Some("TypedCache"), + "`type GroupCache = TypedCache<..>` must resolve on a Client field type" + ); + assert_eq!( + path("PendingDeviceSync").as_deref(), + Some("PendingDeviceSync::HashSet"), + "the offline unknown-device queue is a HashSet behind a newtype" + ); + assert_eq!( + path("InboundCommitBatcher").as_deref(), + Some("InboundCommitBatcher::Vec"), + "the offline-drain commit batch is a Vec behind a newtype" + ); + assert_eq!( + path("MsgSecretWriteBuffer").as_deref(), + Some("MsgSecretWriteBuffer::HashMap"), + "`type Pending = HashMap<..>` must not read as an opaque type" + ); + + // A type that holds no collection stays off the list, so the resolution is + // selective rather than flagging every crate-local field type. + assert_eq!(path("SentFrameTap"), None); + assert_eq!(path("NotAClientFieldTypeAnywhere"), None); +} + /// An exemption must name a field that still exists, so the list cannot rot /// into permission for a field nobody checks any more. #[test] From e6e8a3375618d614cc0267faa2918430bb96db2a Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 10 Aug 2026 23:16:14 +0000 Subject: [PATCH 2/2] docs(perf): widen the retention audit to wacore, correct four bounds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second review round. The substantive find is that the coverage guard only parsed this crate's `src/`, so a collection behind a `wacore` type was invisible to it — and one was: `AppStateProcessor::key_cache` gains an expanded key per distinct key id the server's patches reference, has no cap and no TTL, and is emptied only by `clear_key_cache` on reconnect. It now reports a count, and the guard parses `wacore/src` too. The two other candidates that widening surfaced are exempted with their reasons: `ab_props` is bounded by the compile-time WATCHED set, and `pair_code_state`'s `Vec` is one server pairing ref, not a collection. Four bound claims in the inventory were wrong or too strong, each checked against the code before rewriting: - The Signal store caches are bounded at 2 000 *while flushes succeed*. `evict_clean_entries` skips dirty and deleted keys — correct, since dropping unpersisted state is worse — so a backend that stops accepting writes leaves everything dirty and the maps grow past the cap. - `msg_secret_buffer`'s 4 096 is not a hard ceiling on the reported count: a queueing future cancelled while backpressured force-buffers what it holds rather than losing captures. Said on the accessor as well as on the report field. - The drain commit's encode arena is cleared but not shrunk, so one oversized message leaves its capacity resident for the session. It stays unreported (sampling it means taking a lock a commit holds across its backend write), but it is named in the inventory now rather than implied to be transient. - The transient-retention display assertion searched to the end of the report, so a field moving into a later section would still have passed. Alias resolution in the guard now iterates to a fixed point instead of stopping after two rounds, so a chain longer than that cannot read as "no collection here". Each round dedups: without that, an alias naming another one twice doubles its ident list per round, which with wacore in the map is an OOM rather than a slow test. Multiplicity carries no information for the question being asked, so dropping it is free. Adds `the_allocator_counts_what_it_hands_out`, the one non-ignored test in the measurement harness: alloc, realloc and dealloc accounting, in bounds rather than equalities because the counter is process-wide. --- agent_docs/observability.md | 17 +++- src/client.rs | 6 ++ src/client/accessors.rs | 7 ++ src/client/tests.rs | 8 +- src/message/commit_batch.rs | 5 +- src/msg_secret_buffer.rs | 14 ++- tests/e2e/tests/per_client_retention.rs | 41 ++++++++ tests/report_coverage.rs | 125 +++++++++++++++++++----- wacore/src/appstate_sync.rs | 12 +++ 9 files changed, 204 insertions(+), 31 deletions(-) diff --git a/agent_docs/observability.md b/agent_docs/observability.md index a92d2eb4a..fa0078b7d 100644 --- a/agent_docs/observability.md +++ b/agent_docs/observability.md @@ -493,7 +493,7 @@ message can overshoot substantially on its own. | `session_locks` / `chat_lanes` / `group_distribution_locks` | 10 000 / 5 000 / 512 | nothing: an `evict_guard` refuses to evict a lock a task holds, so the map briefly exceeds capacity instead of minting a second lock for one key | | `resend_rate_limiter` | 4 096, FIFO | fail-open by design — an evicted bucket is recreated full, so undersizing forgives rate, never over-throttles | | `group_devices_memo` / `skdm_warm_memo` / `dm_devices_memo` | 64 / 64 / 512 | a recompute | -| `SignalStoreCache` sessions / identities / sender keys | 2 000 each (+1/8 slack before an eviction scan) | nothing: only *clean* entries are evicted, so an unpersisted record is never dropped | +| `SignalStoreCache` sessions / identities / sender keys | 2 000 each (+1/8 slack before an eviction scan), *while flushes succeed* | nothing: only *clean* entries are evicted, so an unpersisted record is never dropped — which also means a backend that stops accepting writes leaves everything dirty and the maps grow past the cap. Correct, and the reason to watch the counts rather than trust the number | | `SignalStoreCache::sender_key_locks` | 2 000, idle-only | nothing: only locks held solely by the map are dropped | | `inbound_commit_batch` | 400 messages / 4 MiB, checked after insert | commits early, no loss; overshoots by one message | | `msg_secret_buffer` | 4 096, except on cancellation | nothing: a producer that would exceed it parks on `capacity_available`, and a cancelled one force-buffers past the mark rather than losing captures | @@ -527,6 +527,13 @@ the bound is a drain or a lifecycle, so the count is the only warning available. skip a device refresh and leave the next send to that user addressed to a stale device list, so the fix if this ever matters is a removal on the online path, not a ceiling. +- **`AppStateProcessor::key_cache`** — expanded app-state keys, one entry per + distinct key id the server's patches reference, with no cap and no TTL; + emptied only by `clear_key_cache` on reconnect. The backend stays + authoritative, so unlike the three above a cap here would be *safe* — nothing + has measured how many distinct keys a real account accumulates, which is why + it reports a count instead. It lives in `wacore`, which is why the coverage + guard now parses that crate too. - **`pending_retries`** — held only for the duration of one retry receipt (a `scopeguard` removes it), so the bound is concurrent receipts. - **`presence_subscriptions`**, **`response_waiters`**, **`node_waiters`**, @@ -536,10 +543,16 @@ the bound is a drain or a lifecycle, so the count is the only warning available. `async_channel`s whose depth is a stalled-transport signal; capping them would drop acks the server is waiting for. - **`offline_receipt_buffer`** — drained at the end of every offline batch, and - the one entry here the report does *not* count: it is listed in + one of two things here the report does *not* count: it is listed in `report_coverage.rs`'s `EXEMPT` because its `MessageInfo` values are already attributed where the batch owns them. Its depth during a drain is therefore invisible; if that ever matters, it needs a field of its own rather than a cap. +- **The drain commit's encode arena** — the other unreported one, and the only + entry on this list that is not a collection: a `Vec` reused across drain + commits. `commit_inbound_batch` clears it but keeps its capacity, so one + oversized message leaves that capacity resident for the rest of the session. + Unreported because sampling it means taking a lock a commit holds across its + backend write. Sizing it is a shrink-after-use question, not a cap question. Two design rules the audit confirmed, and one place where the second does not hold as tightly as the code comments suggest: diff --git a/src/client.rs b/src/client.rs index 846e07547..3d6617139 100644 --- a/src/client.rs +++ b/src/client.rs @@ -491,6 +491,11 @@ pub struct MemoryReport { pub pending_retries: usize, pub presence_subscriptions: usize, pub app_state_key_requests: usize, + /// Expanded app-state keys the processor holds in memory. No capacity cap + /// and no TTL — one entry per distinct key id the server's patches + /// reference, emptied only on reconnect. Zero until the first app-state + /// sync builds the processor. + pub app_state_key_cache: usize, pub app_state_syncing: usize, pub signal_sessions: CollectionStats, pub signal_identities: CollectionStats, @@ -643,6 +648,7 @@ impl std::fmt::Display for MemoryReport { " app_state_key_requests: {}", self.app_state_key_requests )?; + writeln!(f, " app_state_key_cache: {}", self.app_state_key_cache)?; writeln!(f, " app_state_syncing: {}", self.app_state_syncing)?; writeln!(f, "--- Signal store caches ---")?; for (name, c) in &collections[TTL_BOUNDED..TTL_BOUNDED + SIGNAL_CACHES] { diff --git a/src/client/accessors.rs b/src/client/accessors.rs index b858ffcad..c723801d8 100644 --- a/src/client/accessors.rs +++ b/src/client/accessors.rs @@ -349,6 +349,12 @@ impl Client { .len(); let app_state_key_requests = self.app_state_key_requests.lock().await.len(); let app_state_syncing = self.app_state_syncing.len(); + // `get()`, not the builder: a report must not be what constructs the + // processor, so an un-synced client still reports zero. + let app_state_key_cache = match self.app_state_processor.get() { + Some(processor) => processor.cached_key_count().await, + None => 0, + }; let (commit_batch_entries, commit_batch_bytes) = self.inbound_commit_batch.pending_stats(); let inbound_commit_batch = CollectionStats::new(commit_batch_entries as u64, commit_batch_bytes as u64); @@ -451,6 +457,7 @@ impl Client { pending_retries: pending_retries_count, presence_subscriptions, app_state_key_requests, + app_state_key_cache, app_state_syncing, signal_sessions, signal_identities, diff --git a/src/client/tests.rs b/src/client/tests.rs index 653765b88..0637beb09 100644 --- a/src/client/tests.rs +++ b/src/client/tests.rs @@ -4862,13 +4862,19 @@ async fn memory_report_display_sections_stay_aligned() { rendered[history_start..drain_start].contains("history_sync_tasks:"), "history_sync_tasks must render under its own heading, got:\n{rendered}" ); + // Bounded to the section, not "somewhere after its heading": an unbounded + // slice would keep passing if one of these moved into `Plugins` or `Misc`, + // which is exactly the drift this test exists to catch. + let drain_end = rendered[drain_start + 1..] + .find("\n--- ") + .map_or(rendered.len(), |at| drain_start + 1 + at); for name in [ "inbound_commit_batch:", "msg_secret_buffer:", "pending_device_sync:", ] { assert!( - rendered[drain_start..].contains(name), + rendered[drain_start..drain_end].contains(name), "{name} must render under the transient-retention heading, got:\n{rendered}" ); } diff --git a/src/message/commit_batch.rs b/src/message/commit_batch.rs index b49878b12..4493dfb9f 100644 --- a/src/message/commit_batch.rs +++ b/src/message/commit_batch.rs @@ -157,7 +157,10 @@ impl InboundCommitBatcher { /// "waiting to be committed", not "resident". `has_entries` draws the /// same line, for the same reason. /// - **The reusable encode arena**, whose lock a drain commit holds across - /// its backend write; a report must not queue behind one. + /// its backend write; a report must not queue behind one. Worth knowing + /// that this hides real memory rather than a transient: the arena is + /// cleared but not shrunk, so one oversized message leaves its capacity + /// resident for the session. pub(crate) fn pending_stats(&self) -> (usize, usize) { let state = self.lock(); (state.entries.len(), state.bytes) diff --git a/src/msg_secret_buffer.rs b/src/msg_secret_buffer.rs index 82c312373..d33e76c4f 100644 --- a/src/msg_secret_buffer.rs +++ b/src/msg_secret_buffer.rs @@ -431,10 +431,16 @@ impl MsgSecretWriteBuffer { } } - /// Captures buffered but not yet persisted. Bounded by the pending limit - /// (`MAX_PENDING_MSG_SECRETS` in production): a producer that would exceed - /// it parks on `capacity_available` rather than the buffer growing, so this - /// count is a saturation gauge as much as a memory figure. + /// Captures buffered but not yet persisted. Normally bounded by the pending + /// limit (`MAX_PENDING_MSG_SECRETS` in production): a producer that would + /// exceed it parks on `capacity_available` rather than the buffer growing, + /// so this count is a saturation gauge as much as a memory figure. + /// + /// The limit is not a hard ceiling on this number. A queueing future + /// cancelled while backpressured force-buffers what it still holds (see + /// `QueuedEntries::drop`, which inserts with no limit) rather than dropping + /// captures, so a reading above the limit during teardown is expected + /// rather than impossible. pub(crate) fn pending_len(&self) -> usize { self.pending.lock().unwrap_or_else(|p| p.into_inner()).len() } diff --git a/tests/e2e/tests/per_client_retention.rs b/tests/e2e/tests/per_client_retention.rs index 817d4eba1..a69b71c44 100644 --- a/tests/e2e/tests/per_client_retention.rs +++ b/tests/e2e/tests/per_client_retention.rs @@ -86,6 +86,47 @@ fn median(mut v: Vec) -> isize { v[v.len() / 2] } +/// The counter itself, since every figure below is only as good as it is. +/// +/// Not `#[ignore]`d, unlike the measurements: this one is fast, deterministic +/// and is the thing a reader has to trust. Bounds rather than equalities, +/// because `LIVE` is process-wide and the harness threads allocate too — one +/// large allocation makes that noise irrelevant to the assertions. +#[test] +fn the_allocator_counts_what_it_hands_out() { + const SIZE: usize = 4 * 1024 * 1024; + + let (buffer, alloc_bytes) = retained(|| vec![0u8; SIZE]); + assert!( + alloc_bytes >= SIZE as isize, + "an allocation must raise the live count by at least its size, got {alloc_bytes}" + ); + + // realloc: growing in place or by copy must be counted as the delta, not as + // a second whole allocation. + let (grown, realloc_bytes) = retained(|| { + let mut buffer = buffer; + buffer.resize(SIZE * 2, 0); + buffer + }); + assert!( + realloc_bytes >= SIZE as isize, + "growing by SIZE must raise the count by about SIZE, got {realloc_bytes}" + ); + assert!( + realloc_bytes < (2 * SIZE) as isize, + "growing by SIZE must not be counted as a fresh 2*SIZE allocation, got {realloc_bytes}" + ); + + let before_drop = live(); + drop(grown); + let freed = before_drop - live(); + assert!( + freed >= (2 * SIZE) as isize, + "dropping must return the whole allocation to the count, got {freed}" + ); +} + /// What the two per-client bounded queues cost, and what fraction of a client /// that is. /// diff --git a/tests/report_coverage.rs b/tests/report_coverage.rs index d585da336..81e6284fc 100644 --- a/tests/report_coverage.rs +++ b/tests/report_coverage.rs @@ -76,6 +76,17 @@ const EXEMPT: &[(&str, &str)] = &[ "the host list from the most recent `` IQ — one server \ response, replaced wholesale on refresh", ), + ( + "ab_props", + "server props are filtered against the compile-time `WATCHED` interest \ + set at parse time, so the map is bounded by that list and not by what \ + the server sends", + ), + ( + "pair_code_state", + "one in-flight pairing attempt; its `Vec` is the server's pairing ref, \ + a handful of bytes, not a collection of entries", + ), ]; fn manifest_path(relative: &str) -> PathBuf { @@ -175,6 +186,10 @@ fn crate_type_fields() -> (HashMap>, HashMap (HashMap>, HashMap>) -> Vec { - idents - .iter() - .flat_map(|ident| match aliases.get(ident) { - // A self-referential spelling (`type Cache = Cache<..>`) would - // otherwise expand forever without adding anything. - Some(target) if target != std::slice::from_ref(ident) => target.clone(), - _ => vec![ident.clone()], - }) - .collect() - } - - for _ in 0..2 { - let snapshot = aliases.clone(); - for idents in aliases.values_mut() { - *idents = expand(idents, &snapshot); - } - } + // Substitute aliases so the lookup below stays one level deep: aliases are + // resolved among themselves to a fixed point, then applied to every type's + // field idents. + flatten_alias_chains(&mut aliases); for idents in defs.values_mut() { - *idents = expand(idents, &aliases); + *idents = resolve_aliases(idents, &aliases); } // The alias map is returned too: a `Client` field can name one directly // (`group_cache: OnceLock>` over `type GroupCache = @@ -214,17 +212,61 @@ fn crate_type_fields() -> (HashMap>, HashMap>) -> Vec { idents .iter() .flat_map(|ident| match aliases.get(ident) { + // A self-referential spelling (`type Cache = Cache<..>`) would + // otherwise expand forever without adding anything. Some(target) if target != std::slice::from_ref(ident) => target.clone(), _ => vec![ident.clone()], }) .collect() } +/// Collapse `A -> B -> … -> Vec` so every alias maps directly to what it +/// bottoms out in. +/// +/// Iterates to a fixed point rather than a fixed number of rounds: a chain +/// longer than the loop would otherwise leave its tail unresolved, and the +/// resulting miss looks exactly like "this field holds no collection". +/// +/// Each round dedups, which is what makes the iteration terminate rather than +/// blow up. Without it an alias naming another one twice (`type Pair = (Foo, +/// Foo)`) doubles its ident list every round — with the whole of `wacore` in the +/// map that is an OOM, not a slow test. Since the question asked of the result +/// is only *which* idents are reachable, multiplicity carries no information, +/// so dropping it costs nothing and bounds each list by the number of distinct +/// idents in the tree. +fn flatten_alias_chains(aliases: &mut HashMap>) { + fn dedup(mut idents: Vec) -> Vec { + let mut seen = std::collections::HashSet::new(); + idents.retain(|ident| seen.insert(ident.clone())); + idents + } + + for idents in aliases.values_mut() { + *idents = dedup(std::mem::take(idents)); + } + loop { + let snapshot = aliases.clone(); + let mut changed = false; + for idents in aliases.values_mut() { + let expanded = dedup(resolve_aliases(idents, &snapshot)); + if *idents != expanded { + *idents = expanded; + changed = true; + } + } + if !changed { + return; + } + } +} + /// How a field reaches a growable: directly in its own type, or through one /// crate-local type it names. Returned for the failure message, so a new /// candidate says which collection put it on the list. @@ -394,6 +436,43 @@ fn a_newtype_or_alias_does_not_hide_its_collection() { assert_eq!(path("NotAClientFieldTypeAnywhere"), None); } +/// An alias chain longer than any fixed number of rounds still bottoms out. +/// +/// Synthetic rather than drawn from the tree: the real chains are one link, and +/// the failure this guards against — a longer chain silently reading as "no +/// collection here" — would arrive with the code that introduced it. +#[test] +fn alias_chains_resolve_to_a_fixed_point() { + let mut aliases: HashMap> = [ + ("A", vec!["B"]), + ("B", vec!["C"]), + ("C", vec!["D"]), + ("D", vec!["E"]), + ("E", vec!["Arc", "Vec", "u8"]), + // Self-shadowing: `type Cache = Cache<..>` re-exports a foreign name and + // must not expand forever. + ("Cache", vec!["Cache"]), + ] + .into_iter() + .map(|(name, target)| { + ( + name.to_string(), + target.into_iter().map(str::to_string).collect(), + ) + }) + .collect(); + + flatten_alias_chains(&mut aliases); + + assert_eq!(aliases["A"], ["Arc", "Vec", "u8"]); + assert_eq!(aliases["Cache"], ["Cache"]); + assert_eq!( + resolve_aliases(&["A".to_string()], &aliases), + ["Arc", "Vec", "u8"], + "a field naming the head of the chain must reach the Vec at its end" + ); +} + /// An exemption must name a field that still exists, so the list cannot rot /// into permission for a field nobody checks any more. #[test] diff --git a/wacore/src/appstate_sync.rs b/wacore/src/appstate_sync.rs index 4e7d5cfb8..f3b769511 100644 --- a/wacore/src/appstate_sync.rs +++ b/wacore/src/appstate_sync.rs @@ -176,6 +176,18 @@ impl AppStateProcessor { *self.key_cache.lock().await = HashMap::new(); } + /// Expanded app-state keys held in memory, for `Client::memory_report()`. + /// + /// The cache has neither a capacity cap nor a TTL: it gains an entry per + /// distinct key id the server's patches reference and is emptied only by + /// [`Self::clear_key_cache`] on reconnect, so a long-lived connection is its + /// only bound. That is why the count is worth reporting — the backend stays + /// authoritative, so a cap would be safe here, but nothing has measured how + /// many distinct keys a real account accumulates. + pub async fn cached_key_count(&self) -> usize { + self.key_cache.lock().await.len() + } + /// Pre-fetch and cache all keys needed for a patch list. async fn prefetch_keys(&self, pl: &PatchList) -> Result<()> { let key_ids = collect_key_id_refs_from_patch_list(pl.snapshot.as_ref(), &pl.patches);