diff --git a/agent_docs/observability.md b/agent_docs/observability.md index 5060660b7..fa0078b7d 100644 --- a/agent_docs/observability.md +++ b/agent_docs/observability.md @@ -403,7 +403,174 @@ 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), *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 | +| 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. +- **`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`**, + **`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 + 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: + +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..3d6617139 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, @@ -462,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, @@ -509,7 +543,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 +557,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 +584,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 ---")?; @@ -610,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] { @@ -627,11 +666,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 +677,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..c723801d8 100644 --- a/src/client/accessors.rs +++ b/src/client/accessors.rs @@ -349,6 +349,17 @@ 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); + 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 +438,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, @@ -443,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 bdba35a25..0637beb09 100644 --- a/src/client/tests.rs +++ b/src/client/tests.rs @@ -4844,6 +4844,62 @@ 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}" + ); + // 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..drain_end].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..4493dfb9f 100644 --- a/src/message/commit_batch.rs +++ b/src/message/commit_batch.rs @@ -137,6 +137,35 @@ 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. 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) + } + /// 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 +987,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..d33e76c4f 100644 --- a/src/msg_secret_buffer.rs +++ b/src/msg_secret_buffer.rs @@ -431,7 +431,16 @@ impl MsgSecretWriteBuffer { } } - #[cfg(test)] + /// 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/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..a69b71c44 --- /dev/null +++ b/tests/e2e/tests/per_client_retention.rs @@ -0,0 +1,304 @@ +//! 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] +} + +/// 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. +/// +/// 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..81e6284fc 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,47 @@ 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", + ), + ( + "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 { @@ -64,10 +118,173 @@ 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); + // `wacore` too: `Client` holds several of its types, and a collection behind + // one of those is no less a per-session cost for living in another crate. + // `AppStateProcessor`'s unbounded key cache is the one that proved it. + crate_sources(&manifest_path("wacore/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 to a fixed point, then applied to every type's + // field idents. + flatten_alias_chains(&mut aliases); + for idents in defs.values_mut() { + *idents = resolve_aliases(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, one substitution deep, by what it stands +/// for. Chains are handled by [`flatten_alias_chains`] having already collapsed +/// the map, so one pass here is enough. +fn resolve_aliases(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() +} + +/// 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. +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 +351,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 +360,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 +399,80 @@ 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 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);