Bound chat-list and media retention query work (#1749)

* Bound media reference retention work

Index cached secrets by group and epoch so retaining a media message
does not scan unrelated exporter history. Mark shared epoch rows once
per message and avoid rewriting secrets already managed by retention.

Cover reference replay, history scaling, and populated migration
upgrades while preserving epoch retirement behavior.

* Bound chat readiness and pinned-list query work

Index each group's receipt timestamps so checking a complete chat
projection does not scan retained message history. Compute pin ranks
once for full-list reads instead of counting earlier pins per row.
Keep keyed lookups and ranks across gaps or missing projections intact.

Cover query scaling, stale receipt detection, archive filtering, and
populated index upgrades.

* Benchmark file-backed chat-list readiness

Measure release-mode readiness and row fetches on encrypted files
with 10,000 and 100,000 retained messages. Alternate the receipt
index off and on, exclude setup, and verify identical chat rows.

Author: dannym-arx

crates/storage-sqlite/src/chat_list.rs

@@ -2316,20 +2316,25 @@ pub(crate) fn chat_list_rows_tx(
     tx: &Connection,
     query: ChatListQuery,
 ) -> StorageResult<Vec<ChatListRow>> {
-    let sql = if query.include_archived {
-        format!(
-            "{CHAT_LIST_ROW_SELECT_AND_JOINS}
-             ORDER BY pin.ordinal IS NULL, pin.ordinal ASC,
-                      row.activity_sort_at DESC, row.group_id_hex"
-        )
+    // Rank the pin table once, including pins whose projection is absent.
+    // Keyed reads retain their single ordinal-count lookup.
+    let archived_filter = if query.include_archived {
+        ""
     } else {
-        format!(
-            "{CHAT_LIST_ROW_SELECT_AND_JOINS}
-             WHERE row.archived = 0
-             ORDER BY pin.ordinal IS NULL, pin.ordinal ASC,
-                      row.activity_sort_at DESC, row.group_id_hex"
-        )
+        "WHERE row.archived = 0"
     };
+    let sql = format!(
+        "{CHAT_LIST_ROW_SELECT_LIST} pin.position
+         {CHAT_LIST_ROW_JOINS}
+         LEFT JOIN (
+             SELECT group_id_hex, ordinal,
+                    ROW_NUMBER() OVER (ORDER BY ordinal) - 1 AS position
+             FROM chat_pin_positions
+         ) AS pin ON pin.group_id_hex = row.group_id_hex
+         {archived_filter}
+         ORDER BY pin.ordinal IS NULL, pin.ordinal ASC,
+                  row.activity_sort_at DESC, row.group_id_hex"
+    );
     let now_ms = unix_now_ms();
     let mut stmt = tx.prepare_cached(&sql).storage()?;
     let mut rows = stmt
@@ -2358,7 +2363,7 @@ fn direct_conversation_candidate_sql() -> String {
     // Drive from the peer index, then join the matching chat-list row.
     // Durable activity order, not pin-first chat-list order.
     format!(
-        "{CHAT_LIST_ROW_SELECT_LIST}
+        "{CHAT_LIST_ROW_SELECT_LIST} {CHAT_PIN_POSITION_SQL}
          FROM direct_conversation_members AS dcm
          JOIN chat_list_rows AS row ON row.group_id_hex = dcm.group_id_hex
          LEFT JOIN account_groups AS ag ON ag.group_id_hex = row.group_id_hex
@@ -2412,7 +2417,8 @@ pub(crate) fn chat_list_row_tx(
 ) -> StorageResult<Option<ChatListRow>> {
     let now_ms = unix_now_ms();
     let sql = format!(
-        "{CHAT_LIST_ROW_SELECT_AND_JOINS}
+        "{CHAT_LIST_ROW_SELECT_LIST} {CHAT_PIN_POSITION_SQL}
+         {CHAT_LIST_ROW_JOINS} {CHAT_PIN_JOIN}
          WHERE row.group_id_hex = ?1"
     );
     tx.query_row_cached(&sql, params![group_id_hex], |row| {
@@ -2434,11 +2440,7 @@ pub(crate) fn chat_list_row_tx(
     .transpose()
 }
 
-// Keep this projection in one place: `chat_list_row_from_row` decodes it by
-// index, so list and single-row queries must never drift in column order.
-// `CHAT_LIST_ROW_SELECT_LIST` must stay column-identical to
-// `CHAT_LIST_ROW_SELECT_AND_JOINS` so the peer-driven candidate query
-// decodes the same way.
+// All chat reads share these columns; only pin-rank computation differs.
 const CHAT_LIST_ROW_SELECT_LIST: &str =
     "SELECT row.group_id_hex, row.archived, row.pending_confirmation,
             row.title, row.group_name, row.avatar_url,
@@ -2460,45 +2462,20 @@ const CHAT_LIST_ROW_SELECT_LIST: &str =
                 SELECT 1 FROM cgka_disband_tombstones AS tomb
                 WHERE lower(hex(tomb.group_id)) = lower(row.group_id_hex)
             ),
-            pin.group_id_hex IS NOT NULL,
-            CASE WHEN pin.ordinal IS NULL THEN NULL ELSE (
+            pin.group_id_hex IS NOT NULL,";
+
+const CHAT_PIN_POSITION_SQL: &str = "CASE WHEN pin.ordinal IS NULL THEN NULL ELSE (
                 SELECT COUNT(*)
                 FROM chat_pin_positions AS earlier_pin
                 WHERE earlier_pin.ordinal < pin.ordinal
             ) END";
 
-const CHAT_LIST_ROW_SELECT_AND_JOINS: &str =
-    "SELECT row.group_id_hex, row.archived, row.pending_confirmation,
-            row.title, row.group_name, row.avatar_url,
-            row.avatar_image_hash_hex, row.avatar_image_key_hex,
-            row.avatar_image_nonce_hex, row.avatar_image_upload_key_hex,
-            row.avatar_media_type, row.last_message_id_hex,
-            row.last_message_sender, row.last_message_preview,
-            row.last_message_kind, row.last_message_timeline_at,
-            row.last_message_deleted, row.last_message_media_json,
-            row.last_message_delivery_state, row.unread_count,
-            row.manually_marked_unread, row.unread_mention_count,
-            row.first_unread_message_id_hex, row.last_read_message_id_hex,
-            row.last_read_timeline_at, row.conversation_created_at,
-            row.activity_sort_at, row.updated_at, row.self_membership,
-            ag.member_count,
-            mute.group_id_hex IS NOT NULL,
-            mute.muted_until_ms,
-            EXISTS (
-                SELECT 1 FROM cgka_disband_tombstones AS tomb
-                WHERE lower(hex(tomb.group_id)) = lower(row.group_id_hex)
-            ),
-            pin.group_id_hex IS NOT NULL,
-            CASE WHEN pin.ordinal IS NULL THEN NULL ELSE (
-                SELECT COUNT(*)
-                FROM chat_pin_positions AS earlier_pin
-                WHERE earlier_pin.ordinal < pin.ordinal
-            ) END
-     FROM chat_list_rows AS row
+const CHAT_LIST_ROW_JOINS: &str = "FROM chat_list_rows AS row
      LEFT JOIN account_groups AS ag ON ag.group_id_hex = row.group_id_hex
      LEFT JOIN chat_notification_settings AS mute
-        ON mute.group_id_hex = row.group_id_hex
-     LEFT JOIN chat_pin_positions AS pin
+        ON mute.group_id_hex = row.group_id_hex";
+
+const CHAT_PIN_JOIN: &str = "LEFT JOIN chat_pin_positions AS pin
         ON pin.group_id_hex = row.group_id_hex";
 
 fn chat_list_row_from_row(row: &rusqlite::Row<'_>, now_ms: i64) -> rusqlite::Result<ChatListRow> {

crates/storage-sqlite/src/chat_list/tests.rs

@@ -158,6 +158,84 @@ fn setup_store() -> SqliteAccountStorage {
     setup_store_with_group(group())
 }
 
+// Compare the existing readiness API with migration 0066’s index on synthetic history.
+// Setup, index construction, and the connection/KDF are outside the timed region.
+// Run: cargo test -p storage-sqlite --release --lib chat_startup_benchmark -- --ignored --nocapture
+#[test]
+#[ignore = "file-backed chat startup performance investigation"]
+fn chat_startup_benchmark() {
+    let dir = tempfile::tempdir().unwrap();
+    let key = SqlCipherKey::new("synthetic chat benchmark").unwrap();
+    let store =
+        SqliteAccountStorage::open_encrypted(dir.path().join("account.sqlite3"), &key).unwrap();
+    store
+        .save_account_projection_state(
+            &StoredAccountState {
+                label: "benchmark".to_owned(),
+                groups: vec![group()],
+                ..StoredAccountState::default()
+            },
+            256,
+            MAX_FUTURE_SKEW_SECS,
+        )
+        .unwrap();
+    store
+        .lock()
+        .unwrap()
+        .execute_batch("DROP INDEX idx_message_timeline_group_received")
+        .unwrap();
+    let mut seeded = 0;
+    let body = "x".repeat(512);
+    for count in [10_000, 100_000] {
+        cgka_traits::StorageProvider::with_transaction(&store, |store| {
+            for index in seeded..count {
+                let mut event = chat(&format!("{index:064x}"), REMOTE, index, &body);
+                event.source_epoch = Some(1);
+                store.record_app_event(&event)?;
+            }
+            Ok::<_, cgka_traits::StorageError>(())
+        })
+        .unwrap();
+        seeded = count;
+        let id = format!("{:064x}", count - 1);
+        store
+            .mark_timeline_message_read(LOCAL, GROUP, &id, &no_mentions)
+            .unwrap();
+        store.ensure_chat_list_rows(LOCAL, &no_mentions).unwrap();
+        let expected = store.chat_list_rows(ChatListQuery::default()).unwrap();
+        assert_eq!(expected.len(), 1);
+        assert_eq!(
+            expected[0].last_message.as_ref().unwrap().message_id_hex,
+            id
+        );
+        assert_eq!(expected[0].unread_count, 0);
+        // Repeat A/B after dropping the index to expose cache/order effects.
+        for mode in ["baseline", "indexed", "baseline", "indexed"] {
+            if mode == "indexed" {
+                store.lock().unwrap().execute_batch("CREATE INDEX idx_message_timeline_group_received ON message_timeline(group_id_hex, received_at)").unwrap();
+            }
+            store.ensure_chat_list_rows(LOCAL, &no_mentions).unwrap();
+            let mut samples = Vec::new();
+            for _ in 0..7 {
+                let start = std::time::Instant::now();
+                store.ensure_chat_list_rows(LOCAL, &no_mentions).unwrap();
+                let rows = store.chat_list_rows(ChatListQuery::default()).unwrap();
+                samples.push(start.elapsed().as_micros());
+                assert_eq!(rows, expected);
+            }
+            samples.sort_unstable();
+            eprintln!("count={count} mode={mode} median_us={}", samples[3]);
+            if mode == "indexed" {
+                store
+                    .lock()
+                    .unwrap()
+                    .execute_batch("DROP INDEX idx_message_timeline_group_received")
+                    .unwrap();
+            }
+        }
+    }
+}
+
 /// Preview work stays bounded for accepted history and displaced pending sends.
 #[test]
 fn preview_query_work() {

crates/storage-sqlite/src/encrypted_media_secrets.rs

@@ -245,24 +245,29 @@ pub(crate) fn replace_encrypted_media_secret_references_for_parts_tx(
         && let Some(source_epoch) = source_epoch
     {
         let source_epoch = u64_to_i64(source_epoch)?;
-        for component_id in encrypted_media_component_ids(tags) {
+        let component_ids = encrypted_media_component_ids(tags);
+        for component_id in &component_ids {
             tx.execute_cached(
                 "INSERT INTO encrypted_media_epoch_secret_references (
                      group_id_hex, message_id_hex, component_id, source_epoch
                  ) VALUES (?1, ?2, ?3, ?4)",
                 params![
                     group_id_hex,
                     message_id_hex,
-                    i64::from(component_id),
+                    i64::from(*component_id),
                     source_epoch,
                 ],
             )
             .storage()?;
+        }
+        // All media formats share the exporter; mark its cached rows once.
+        if !component_ids.is_empty() {
             tx.execute_cached(
                 "UPDATE encrypted_media_epoch_secrets
                  SET retention_managed = 1
                  WHERE group_id_hex = ?1
-                   AND source_epoch = ?2",
+                   AND source_epoch = ?2
+                   AND retention_managed = 0",
                 params![group_id_hex, source_epoch],
             )
             .storage()?;

crates/storage-sqlite/src/migrations.rs

@@ -132,6 +132,10 @@ mod migration_0065_chat_presentation;
 mod migration_0066_chat_presentation_maintenance;
 #[path = "migrations/0067_invitation_recovery.rs"]
 mod migration_0067_invitation_recovery;
+#[path = "migrations/0068_media_epoch_index.rs"]
+mod migration_0068_media_epoch_index;
+#[path = "migrations/0069_chat_readiness_index.rs"]
+mod migration_0069_chat_readiness_index;
 #[cfg(test)]
 #[path = "migrations/query_work_tests.rs"]
 mod query_work_tests;
@@ -485,6 +489,16 @@ const MIGRATIONS: &[Migration] = &[
         name: "0067_invitation_recovery",
         apply: migration_0067_invitation_recovery::apply,
     },
+    Migration {
+        version: 68,
+        name: "0068_media_epoch_index",
+        apply: migration_0068_media_epoch_index::apply,
+    },
+    Migration {
+        version: 69,
+        name: "0069_chat_readiness_index",
+        apply: migration_0069_chat_readiness_index::apply,
+    },
 ];
 
 pub(crate) fn run_all(connection: &mut Connection) -> StorageResult<usize> {
@@ -1255,7 +1269,7 @@ mod tests {
         assert!(matches!(
             error,
             StorageError::UnsupportedSchemaVersion {
-                found: 67,
+                found: 69,
                 latest_supported: 46,
             }
         ));
@@ -1311,7 +1325,7 @@ mod tests {
         assert!(matches!(
             error,
             StorageError::UnsupportedSchemaVersion {
-                found: 67,
+                found: 69,
                 latest_supported: 46,
             }
         ));
@@ -1615,7 +1629,7 @@ mod tests {
         assert!(matches!(
             error,
             StorageError::UnsupportedSchemaVersion {
-                found: 67,
+                found: 69,
                 latest_supported: 46,
             }
         ));

crates/storage-sqlite/src/migrations/0068_media_epoch_index.rs

@@ -0,0 +1,13 @@
+use crate::SqliteResultExt;
+use cgka_traits::storage::StorageResult;
+use rusqlite::Transaction;
+
+pub(super) fn apply(tx: &Transaction<'_>) -> StorageResult<()> {
+    // Both first-reference retention and retirement address an entire epoch.
+    tx.execute_batch(
+        "DROP INDEX IF EXISTS idx_media_secrets_retirement;
+         CREATE INDEX IF NOT EXISTS idx_media_secrets_epoch
+             ON encrypted_media_epoch_secrets(group_id_hex, source_epoch);",
+    )
+    .storage()
+}

crates/storage-sqlite/src/migrations/0069_chat_readiness_index.rs

@@ -0,0 +1,12 @@
+use crate::SqliteResultExt;
+use cgka_traits::storage::StorageResult;
+use rusqlite::Transaction;
+
+pub(super) fn apply(tx: &Transaction<'_>) -> StorageResult<()> {
+    // Readiness probes need the latest receipt, regardless of preview eligibility.
+    tx.execute_batch(
+        "CREATE INDEX IF NOT EXISTS idx_message_timeline_group_received
+             ON message_timeline(group_id_hex, received_at);",
+    )
+    .storage()
+}