perf(daemon): compact retained read tracking state

.github/workflows/ci.yml

@@ -96,6 +96,8 @@ jobs:
           cargo fmt --all -- --check
           cargo fmt --manifest-path crates/gui/tests/webcodecs/Cargo.toml --all -- --check
           cargo fmt --manifest-path crates/audio/src/web/call_device/tests/Cargo.toml --all -- --check
+      - name: Test memory measurement driver
+        run: python3 -m unittest discover -s scripts -p test_measure_daemon_memory.py
       - name: Check unused dependencies
         run: cargo machete
       - name: Test benchmark cleanup

crates/chat-store/tests/account_scale_paging.rs

@@ -0,0 +1,154 @@
+//! Synthetic account-sized coverage for the reads a front end makes.
+//! This is a correctness fixture, not an RSS or latency benchmark.
+
+mod common;
+
+use common::*;
+use oxidezap_chat_store::ChatArchiveFilter;
+
+const CHATS: i64 = 1_032;
+const CONTACTS: i64 = 1_653;
+const MESSAGES: i64 = 75_703;
+
+async fn seed_account(store: &SqliteStore) {
+    let mut message = wa::Message::default();
+    message.conversation = Some("synthetic fixture message".into());
+    let proto = waproto::codec::message_to_vec(&message);
+    store
+        .shared()
+        .run(move |conn| {
+            diesel::sql_query(
+                "WITH RECURSIVE rows(n) AS (
+                     SELECT 1 UNION ALL SELECT n + 1 FROM rows WHERE n < ?
+                 ) INSERT INTO chats (device_id, jid, name, last_message_ts)
+                 SELECT 1, printf('99%011d@s.whatsapp.net', n),
+                        printf('Synthetic chat %d', n),
+                        1700000000000 + (n + ((? - n) / ?) * ?) * 1000
+                 FROM rows",
+            )
+            .bind::<diesel::sql_types::BigInt, _>(CHATS)
+            .bind::<diesel::sql_types::BigInt, _>(MESSAGES)
+            .bind::<diesel::sql_types::BigInt, _>(CHATS)
+            .bind::<diesel::sql_types::BigInt, _>(CHATS)
+            .execute(conn)
+            .map_err(db_err)?;
+            diesel::sql_query(
+                "WITH RECURSIVE rows(n) AS (
+                     SELECT 1 UNION ALL SELECT n + 1 FROM rows WHERE n < ?
+                 ) INSERT INTO contacts (device_id, jid, full_name)
+                 SELECT 1, printf('99%011d@s.whatsapp.net', n),
+                        printf('Synthetic contact %d', n) FROM rows",
+            )
+            .bind::<diesel::sql_types::BigInt, _>(CONTACTS)
+            .execute(conn)
+            .map_err(db_err)?;
+            diesel::sql_query(
+                "WITH RECURSIVE rows(n) AS (
+                     SELECT 1 UNION ALL SELECT n + 1 FROM rows WHERE n < ?
+                 ) INSERT INTO messages
+                     (device_id, chat_jid, msg_id, sender_jid, timestamp_ms,
+                      kind, text_content, proto)
+                 SELECT 1, printf('99%011d@s.whatsapp.net', 1 + (n - 1) % ?),
+                        printf('synthetic-%d', n),
+                        printf('99%011d@s.whatsapp.net', 1 + (n - 1) % ?),
+                        1700000000000 + n * 1000, 'text',
+                        'synthetic fixture message', ? FROM rows",
+            )
+            .bind::<diesel::sql_types::BigInt, _>(MESSAGES)
+            .bind::<diesel::sql_types::BigInt, _>(CHATS)
+            .bind::<diesel::sql_types::BigInt, _>(CHATS)
+            .bind::<diesel::sql_types::Binary, _>(proto)
+            .execute(conn)
+            .map_err(db_err)?;
+            Ok(())
+        })
+        .await
+        .expect("seed synthetic account");
+}
+
+#[tokio::test]
+async fn account_sized_history_pages_without_skips_or_duplicates() {
+    let (store, history) = test_store().await;
+    seed_account(&store).await;
+
+    #[derive(diesel::QueryableByName)]
+    struct Count {
+        #[diesel(sql_type = diesel::sql_types::BigInt)]
+        messages: i64,
+    }
+    let count = store
+        .shared()
+        .read(|conn| {
+            diesel::sql_query("SELECT COUNT(*) AS messages FROM messages WHERE device_id = 1")
+                .get_result::<Count>(conn)
+                .map_err(db_err)
+        })
+        .await
+        .unwrap();
+    assert_eq!(count.messages, MESSAGES);
+
+    let mut cursor = None;
+    let mut chats = std::collections::HashSet::new();
+    let mut previous_activity = i64::MAX;
+    loop {
+        let page = history
+            .chats_page(ChatArchiveFilter::Unarchived, cursor, 100)
+            .await
+            .unwrap();
+        assert!(page.len() <= 100);
+        for chat in &page {
+            assert!(chats.insert(chat.jid.to_string()), "repeated chat");
+            let activity = chat.last_message_at.unwrap().timestamp_millis();
+            assert!(
+                activity < previous_activity,
+                "chat activity order across pages"
+            );
+            previous_activity = activity;
+        }
+        cursor = page.last().map(ChatCursor::from);
+        if page.len() < 100 {
+            break;
+        }
+    }
+    assert_eq!(chats.len(), CHATS as usize);
+
+    let peer = jid("9900000000001@s.whatsapp.net");
+    let mut cursor = None;
+    let mut ids = std::collections::HashSet::new();
+    let mut previous_timestamp = i64::MAX;
+    loop {
+        let page = history.messages(&peer, cursor, 50).await.unwrap();
+        assert!(page.len() <= 50);
+        for message in &page {
+            assert!(ids.insert(message.id.clone()), "repeated message");
+            assert!(message.timestamp.timestamp_millis() < previous_timestamp);
+            previous_timestamp = message.timestamp.timestamp_millis();
+            assert_eq!(
+                message
+                    .message
+                    .as_deref()
+                    .and_then(|m| m.conversation.as_deref()),
+                Some("synthetic fixture message"),
+                "stored payload decodes to the seeded text"
+            );
+        }
+        cursor = page.last().map(MessageCursor::from);
+        if page.len() < 50 {
+            break;
+        }
+    }
+    assert_eq!(ids.len() as i64, 1 + (MESSAGES - 1) / CHATS);
+    assert_eq!(
+        history.contacts(None, CONTACTS + 1).await.unwrap().len(),
+        CONTACTS as usize
+    );
+
+    #[cfg(feature = "search")]
+    {
+        let hits = history.search_messages("s", 20).await.unwrap();
+        assert_eq!(hits.len(), 20);
+        let unique: std::collections::HashSet<_> = hits.iter().map(|m| &m.id).collect();
+        assert_eq!(unique.len(), hits.len());
+    }
+    history.close().await.unwrap();
+}

crates/daemon/Cargo.toml

@@ -14,6 +14,8 @@ path = "src/main.rs"
 
 [features]
 test-support = ["oxidezap-session/test-support"]
+# Instrument only native test binaries; production allocators are unchanged.
+memory-profiling = []
 
 [dependencies]
 chrono = { workspace = true, features = ["std"] }

crates/daemon/src/lib.rs

@@ -32,6 +32,9 @@ pub mod server;
 pub mod session_bridge;
 pub mod state;
 
+#[cfg(all(test, feature = "memory-profiling", not(target_family = "wasm")))]
+mod memory_metrics;
+
 /// The ways in, which are a platform each.
 #[cfg(not(target_family = "wasm"))]
 pub mod listener;

crates/daemon/src/memory_metrics.rs

@@ -0,0 +1,141 @@
+//! Requested Rust heap bytes, separate from the process's resident pages.
+//! Opt-in test instrumentation, never linked into a production allocator.
+
+// Machine-readable output is the test harness's measurement interface.
+#![allow(clippy::print_stdout)]
+
+use portable_atomic::{AtomicUsize, Ordering};
+use std::alloc::{GlobalAlloc, Layout, System};
+
+struct Counting;
+
+static LIVE: AtomicUsize = AtomicUsize::new(0);
+static PEAK: AtomicUsize = AtomicUsize::new(0);
+static ALLOCATIONS: AtomicUsize = AtomicUsize::new(0);
+static REQUESTED: AtomicUsize = AtomicUsize::new(0);
+
+fn added(size: usize) {
+    ALLOCATIONS.fetch_add(1, Ordering::Relaxed);
+    REQUESTED.fetch_add(size, Ordering::Relaxed);
+    let live = LIVE.fetch_add(size, Ordering::Relaxed) + size;
+    PEAK.fetch_max(live, Ordering::Relaxed);
+}
+
+unsafe impl GlobalAlloc for Counting {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        // SAFETY: pass the allocation contract through to System unchanged.
+        let ptr = unsafe { System.alloc(layout) };
+        if !ptr.is_null() {
+            added(layout.size());
+        }
+        ptr
+    }
+
+    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
+        // SAFETY: pass the allocation contract through to System unchanged.
+        let ptr = unsafe { System.alloc_zeroed(layout) };
+        if !ptr.is_null() {
+            added(layout.size());
+        }
+        ptr
+    }
+
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        LIVE.fetch_sub(layout.size(), Ordering::Relaxed);
+        // SAFETY: the caller provides the original pointer and layout.
+        unsafe { System.dealloc(ptr, layout) };
+    }
+
+    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 {
+        // SAFETY: the caller provides a live pointer/layout and a valid size.
+        let next = unsafe { System.realloc(ptr, layout, size) };
+        if !next.is_null() {
+            ALLOCATIONS.fetch_add(1, Ordering::Relaxed);
+            REQUESTED.fetch_add(size, Ordering::Relaxed);
+            let live = if size >= layout.size() {
+                LIVE.fetch_add(size - layout.size(), Ordering::Relaxed) + size - layout.size()
+            } else {
+                LIVE.fetch_sub(layout.size() - size, Ordering::Relaxed) - (layout.size() - size)
+            };
+            PEAK.fetch_max(live, Ordering::Relaxed);
+        }
+        next
+    }
+}
+
+#[global_allocator]
+static ALLOCATOR: Counting = Counting;
+
+pub(crate) fn begin() {
+    PEAK.store(LIVE.load(Ordering::Relaxed), Ordering::Relaxed);
+    println!("OXIDEZAP_ALLOCATOR counting");
+}
+
+pub(crate) fn checkpoint(stage: &str) {
+    if std::env::var_os("OXIDEZAP_MEMORY_REPORT").is_none() {
+        return;
+    }
+    // Read counters before the sampling/formatting allocations. Samples
+    // themselves are counted in subsequent checkpoints on both revisions.
+    let live = LIVE.load(Ordering::Relaxed);
+    let peak = PEAK.load(Ordering::Relaxed);
+    let allocations = ALLOCATIONS.load(Ordering::Relaxed);
+    let requested = REQUESTED.load(Ordering::Relaxed);
+    let process = process::snapshot();
+    println!(
+        "OXIDEZAP_MEMORY {}",
+        serde_json::json!({
+            "stage": stage, "live_bytes": live, "peak_bytes": peak,
+            "allocations": allocations, "requested_bytes": requested,
+            "process_kib": process,
+        })
+    );
+}
+
+#[cfg(target_os = "linux")]
+mod process {
+    pub(super) fn snapshot() -> serde_json::Value {
+        let mut fields = serde_json::Map::new();
+        for path in ["/proc/self/smaps_rollup", "/proc/self/status"] {
+            if let Ok(contents) = std::fs::read_to_string(path) {
+                for line in contents.lines() {
+                    let Some((name, value)) = line.split_once(':') else {
+                        continue;
+                    };
+                    if matches!(
+                        name,
+                        "Rss"
+                            | "Pss"
+                            | "Pss_Anon"
+                            | "Pss_File"
+                            | "Pss_Shmem"
+                            | "Anonymous"
+                            | "Private_Clean"
+                            | "Private_Dirty"
+                            | "Shared_Clean"
+                            | "Shared_Dirty"
+                            | "Swap"
+                            | "VmHWM"
+                            | "RssAnon"
+                            | "RssFile"
+                            | "RssShmem"
+                    ) && let Some(number) = value
+                        .split_whitespace()
+                        .next()
+                        .and_then(|v| v.parse::<u64>().ok())
+                    {
+                        fields.insert(name.into(), number.into());
+                    }
+                }
+            }
+        }
+        serde_json::Value::Object(fields)
+    }
+}
+
+#[cfg(not(target_os = "linux"))]
+mod process {
+    pub(super) fn snapshot() -> serde_json::Value {
+        serde_json::Value::Null
+    }
+}