macltok

public

Star0

Fork 0

Watch0

66 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `2c13c8db6174`

tarball

atagen <boss@atagen.co>2026-10-25 09:15 UTC

4 files changed+17-280

mentions: drop the stored history repair policy and repair coverage

The policy was echoed into the inbox and never read, and RepairCoverage, GetRepairCoverage and RepairResolved had no reader outside the mentions actor and no public output. A sync discontinuity now shows only as its gap. Only compiles together with the change that removes HistoryRepairCommand::GetPolicy and SetPolicy from the engine.

Modified file: docs/mentions-discovery.md+3 additions−3 deletions@@ -98,7 +98,7 @@ The checkpoint keeps the position the committed response started from as well
 as the one it ended at. A crash after the commit but before the SDK saves `pos`
 replays from that earlier position, and the replay is a superset of what was
 committed, so it is not treated as a discontinuity. Any other starting position
-records a `SyncDiscontinuity` gap and marks every observed room unresolved.
+records a `SyncDiscontinuity` gap.
 
 The hook never fails sync. A failed write keeps the inbox in memory, records a
 `StoreFailed` gap, acks, and retries the write a minute later, so a restart
@@ -281,8 +281,8 @@ session resume.
 
 The store holds two records. `inbox` carries entries, saved events, context
 windows, candidates, edits, redactions, coverage and recovery cursors. `sync`
-carries the connection checkpoints, observed rooms, per-room repair coverage
-and the inbox revision it belongs to. Every response writes `sync`, and
+carries the connection checkpoints, observed rooms and the inbox
+revision it belongs to. Every response writes `sync`, and
 `inbox` only when the response changed it, both in one commit. A quiet
 response therefore costs one small record, while a changed inbox is still
 reserialised and re-encrypted whole. Background completions are applied in

Modified file: src/mentions/service.rs+13 additions−119 deletions@@ -14,7 +14,7 @@ use matrix_sdk::{
     sliding_sync::{ResponseCommitBatch, ResponseCommitHook},
 };
 use ruma::{
-    OwnedEventId, OwnedRoomId, OwnedUserId, RoomId,
+    OwnedRoomId, OwnedUserId, RoomId,
     events::{
         AnySyncTimelineEvent, ignored_user_list::IgnoredUserListEventContent,
         room::encrypted::OriginalSyncRoomEncryptedEvent,
@@ -24,13 +24,11 @@ use ruma::{
 use serde_json::Value;
 use tokio::sync::{mpsc, oneshot, watch};
 
-use super::state::{
-    Candidate, Checkpoint, Data, RepairCoverage, SavedEvent, SyncState, Window, id, invalid, now,
-};
+use super::state::{Candidate, Checkpoint, Data, SavedEvent, SyncState, Window, id, invalid, now};
 use crate::{
-    CommandOutput, EngineConfig, Error, ErrorKind, Event, HistoryRepairPolicy, MentionCaptureState,
-    MentionEntry, MentionGapReason, MentionKey, MentionReason, MentionsCommand, MentionsPage,
-    MentionsUpdate, TimelineContent, TimelineItem,
+    CommandOutput, EngineConfig, Error, ErrorKind, Event, MentionCaptureState, MentionEntry,
+    MentionGapReason, MentionKey, MentionReason, MentionsCommand, MentionsPage, MentionsUpdate,
+    TimelineContent, TimelineItem,
     engine::{MediaService, MentionsStore, context, executor::EngineBoxFuture},
     normalise::{MediaRegistration, normalise_observed},
     transport::EventSender,
@@ -55,10 +53,6 @@ enum Request {
         MentionsCommand,
         oneshot::Sender<Result<CommandOutput, Error>>,
     ),
-    GetRepairPolicy(oneshot::Sender<Result<HistoryRepairPolicy, Error>>),
-    SetRepairPolicy(HistoryRepairPolicy, oneshot::Sender<Result<(), Error>>),
-    GetRepairCoverage(OwnedRoomId, oneshot::Sender<Result<RepairCoverage, Error>>),
-    RepairResolved(OwnedRoomId, oneshot::Sender<Result<(), Error>>),
     Stop(oneshot::Sender<()>),
 }
 
@@ -134,45 +128,6 @@ impl Handle {
         response.await.map_err(|_| Error::interrupted())?
     }
 
-    pub(crate) async fn repair_policy(&self) -> Result<HistoryRepairPolicy, Error> {
-        let (reply, response) = oneshot::channel();
-        self.requests
-            .send(Request::GetRepairPolicy(reply))
-            .await
-            .map_err(|_| Error::interrupted())?;
-        response.await.map_err(|_| Error::interrupted())?
-    }
-
-    pub(crate) async fn set_repair_policy(&self, policy: HistoryRepairPolicy) -> Result<(), Error> {
-        let (reply, response) = oneshot::channel();
-        self.requests
-            .send(Request::SetRepairPolicy(policy, reply))
-            .await
-            .map_err(|_| Error::interrupted())?;
-        response.await.map_err(|_| Error::interrupted())?
-    }
-
-    pub(crate) async fn repair_coverage(
-        &self,
-        room_id: OwnedRoomId,
-    ) -> Result<RepairCoverage, Error> {
-        let (reply, response) = oneshot::channel();
-        self.requests
-            .send(Request::GetRepairCoverage(room_id, reply))
-            .await
-            .map_err(|_| Error::interrupted())?;
-        response.await.map_err(|_| Error::interrupted())?
-    }
-
-    pub(crate) async fn repair_resolved(&self, room_id: OwnedRoomId) -> Result<(), Error> {
-        let (reply, response) = oneshot::channel();
-        self.requests
-            .send(Request::RepairResolved(room_id, reply))
-            .await
-            .map_err(|_| Error::interrupted())?;
-        response.await.map_err(|_| Error::interrupted())?
-    }
-
     pub(crate) async fn stop(&self) {
         let (reply, response) = oneshot::channel();
         if self.requests.send(Request::Stop(reply)).await.is_ok() {
@@ -407,19 +362,12 @@ impl Service {
         Ok(())
     }
 
-    async fn apply(&mut self, data: Option<Data>, sync: Option<SyncState>) -> Result<(), Error> {
-        let dirty = self.dirty;
-        let previous_data = data.map(|data| std::mem::replace(&mut self.data, data));
-        let previous_sync = sync.map(|sync| std::mem::replace(&mut self.sync, sync));
-        self.dirty |= previous_data.is_some();
+    async fn apply(&mut self, data: Data) -> Result<(), Error> {
+        let dirty = std::mem::replace(&mut self.dirty, true);
+        let previous = std::mem::replace(&mut self.data, data);
         let result = self.flush().await;
         if result.is_err() {
-            if let Some(data) = previous_data {
-                self.data = data;
-            }
-            if let Some(sync) = previous_sync {
-                self.sync = sync;
-            }
+            self.data = previous;
             self.dirty = dirty;
         }
         result
@@ -433,23 +381,6 @@ impl Service {
         self.publish().await;
     }
 
-    async fn set_repair_policy(&mut self, policy: HistoryRepairPolicy) -> Result<(), Error> {
-        let mut next = self.data.clone();
-        next.repair_policy = policy;
-        self.apply(Some(next), None).await
-    }
-
-    async fn repair_resolved(&mut self, room_id: OwnedRoomId) -> Result<(), Error> {
-        let mut sync = self.sync.clone();
-        let coverage = sync.repair_coverage.entry(room_id.clone()).or_default();
-        coverage.boundary_ms = None;
-        coverage.last_confirmed_ms = Some(now());
-        coverage.reliable = false;
-        coverage.unresolved = false;
-        sync.observed_rooms.insert(room_id);
-        self.apply(None, Some(sync)).await
-    }
-
     async fn command(&mut self, command: MentionsCommand) -> Result<CommandOutput, Error> {
         match command {
             MentionsCommand::Page { cursor, limit } => {
@@ -496,7 +427,7 @@ impl Service {
                     start: None,
                     end: None,
                 });
-                self.apply(Some(next), None).await?;
+                self.apply(next).await?;
                 let (context, media) = self.data.context(&key)?;
                 self.media.register(media).await;
                 Ok(CommandOutput::MentionContext(context))
@@ -509,7 +440,7 @@ impl Service {
                         entry.revision = self.data.revision + 1;
                     }
                 }
-                self.apply(Some(next), None).await?;
+                self.apply(next).await?;
                 Ok(CommandOutput::Done)
             }
             MentionsCommand::ReadThrough(boundary) => {
@@ -525,7 +456,7 @@ impl Service {
                     entry.read = true;
                     entry.revision = self.data.revision + 1;
                 }
-                self.apply(Some(next), None).await?;
+                self.apply(next).await?;
                 Ok(CommandOutput::Done)
             }
             MentionsCommand::Recover => {
@@ -535,7 +466,7 @@ impl Service {
                 let mut next = self.data.clone();
                 next.coverage.recovering = true;
                 next.coverage.historical_supported = None;
-                self.apply(Some(next), None).await?;
+                self.apply(next).await?;
                 Ok(CommandOutput::Done)
             }
         }
@@ -562,7 +493,6 @@ impl Service {
         if self.timed_out.swap(false, Ordering::AcqRel) {
             next.gap(None, MentionGapReason::HookTimedOut, None);
         }
-        let observed_at_ms = now();
         // a crash after the commit but before the sdk saved `pos` replays from the
         // previous position, and that response is a superset of the one committed
         let continuous = match &checkpoint {
@@ -574,10 +504,6 @@ impl Service {
         };
         if !continuous {
             next.gap(None, MentionGapReason::SyncDiscontinuity, None);
-            let observed_rooms: Vec<_> = sync.observed_rooms.iter().cloned().collect();
-            for room_id in observed_rooms {
-                Self::mark_unresolved(&mut sync, &room_id, observed_at_ms);
-            }
         }
         for (room_id, raw) in &batch.raw_rooms {
             if raw.limited && sync.observed_rooms.contains(room_id) {
@@ -599,37 +525,18 @@ impl Service {
             if self.observable(room_id).is_none() {
                 continue;
             }
-            let previously_observed = sync.observed_rooms.contains(room_id);
-            let limited = batch.raw_rooms.get(room_id).is_some_and(|raw| raw.limited);
-            if !continuous || limited || !previously_observed {
-                Self::mark_unresolved(&mut sync, room_id, observed_at_ms);
-            } else {
-                let coverage = sync.repair_coverage.entry(room_id.clone()).or_default();
-                if !coverage.unresolved {
-                    coverage.last_confirmed_ms = Some(observed_at_ms);
-                }
-                coverage.reliable = false;
-            }
             sync.observed_rooms.insert(room_id.clone());
             let mut observed = BTreeMap::new();
-            let mut latest_anchor = None;
             if let Some(raw) = batch.raw_rooms.get(room_id) {
                 for event in &raw.timeline {
                     let Ok(value) = serde_json::from_str::<Value>(event.json().get()) else {
                         continue;
                     };
                     if let Some(event_id) = value["event_id"].as_str() {
-                        latest_anchor = event_id.parse::<OwnedEventId>().ok();
                         observed.insert(event_id.to_owned(), value);
                     }
                 }
             }
-            if let Some(latest_anchor) = latest_anchor {
-                sync.repair_coverage
-                    .entry(room_id.clone())
-                    .or_default()
-                    .latest_anchor = Some(latest_anchor);
-            }
             let processed = batch
                 .response
                 .rooms
@@ -710,15 +617,6 @@ impl Service {
         }
     }
 
-    fn mark_unresolved(sync: &mut SyncState, room_id: &OwnedRoomId, observed_at_ms: u64) {
-        let coverage = sync.repair_coverage.entry(room_id.clone()).or_default();
-        if !coverage.unresolved {
-            coverage.boundary_ms = Some(observed_at_ms);
-            coverage.unresolved = true;
-        }
-        coverage.reliable = false;
-    }
-
     fn filter_ignored(&self, next: &mut Data) -> bool {
         let removed: Vec<_> = next
             .entries
@@ -1114,10 +1012,6 @@ impl Service {
                     }
                     Some(Request::Batch(batch, reply)) => { self.batch(batch).await; let _ = reply.send(()); }
                     Some(Request::Command(command, reply)) => { let result = self.command(command).await; let _ = reply.send(result); }
-                    Some(Request::GetRepairPolicy(reply)) => { let _ = reply.send(Ok(self.data.repair_policy.clone())); }
-                    Some(Request::SetRepairPolicy(policy, reply)) => { let result = self.set_repair_policy(policy).await; let _ = reply.send(result); }
-                    Some(Request::GetRepairCoverage(room_id, reply)) => { let coverage = self.sync.repair_coverage.get(&room_id).cloned().unwrap_or_default(); let _ = reply.send(Ok(coverage)); }
-                    Some(Request::RepairResolved(room_id, reply)) => { let result = self.repair_resolved(room_id).await; let _ = reply.send(result); }
                     None => return,
                 },
                 Some(result) = work.next(), if !work.is_empty() => {

Modified file: src/mentions/service_tests.rs−136 deletions@@ -353,142 +353,6 @@ async fn an_unreadable_store_resets_behind_a_gap() {
 }
 
 #[tokio::test]
-async fn repair_policy_and_coverage_roundtrip_through_the_inbox() {
-    let server = MatrixMockServer::new().await;
-    let room_id = room_id!("!repair-coverage:localhost");
-    let (client, _) = joined_client(&server, room_id).await;
-    let config = config();
-    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
-        .await
-        .unwrap();
-    let media = Arc::new(MediaService::new(
-        client.clone(),
-        EngineConfig::default().max_media_bytes,
-        EngineConfig::default().max_media_sources,
-    ));
-    let (handle, mut service, _) = Service::new(
-        client.clone(),
-        media,
-        Some(Ok(store)),
-        EngineConfig::default(),
-    )
-    .await
-    .unwrap();
-    service.recovery_budget = 0;
-    service.latest_scan = false;
-    let service_task = tokio::spawn(service.run());
-
-    let default_policy = HistoryRepairPolicy::default();
-    assert_eq!(handle.repair_policy().await.unwrap(), default_policy);
-    let stored_policy = HistoryRepairPolicy::Bridge;
-    handle
-        .set_repair_policy(stored_policy.clone())
-        .await
-        .unwrap();
-    handle.repair_resolved(room_id.to_owned()).await.unwrap();
-    let coverage = handle.repair_coverage(room_id.to_owned()).await.unwrap();
-    assert!(coverage.last_confirmed_ms.is_some());
-    assert!(!coverage.reliable);
-    assert_eq!(coverage.duration_ms(), None);
-
-    handle.stop().await;
-    service_task.await.unwrap();
-    drop(handle);
-
-    let reopened = MentionsStore::open(&config, account(&client), MAX_BYTES)
-        .await
-        .unwrap();
-    let persisted: Data = reopened.read("inbox").unwrap().unwrap();
-    assert_eq!(persisted.repair_policy, stored_policy);
-    let sync: SyncState = reopened.read("sync").unwrap().unwrap();
-    let coverage = sync.repair_coverage.get(room_id).unwrap();
-    assert!(coverage.last_confirmed_ms.is_some());
-    assert!(coverage.boundary_ms.is_none());
-    assert!(!coverage.unresolved);
-    assert!(!coverage.reliable);
-    drop(reopened);
-    config.delete().await.unwrap();
-}
-
-#[tokio::test]
-async fn repair_limited_sync_freezes_the_last_room_checkpoint() {
-    let server = MatrixMockServer::new().await;
-    let room_id = room_id!("!repair-boundary:localhost");
-    let (client, _) = joined_client(&server, room_id).await;
-    let media = Arc::new(MediaService::new(
-        client.clone(),
-        EngineConfig::default().max_media_bytes,
-        EngineConfig::default().max_media_sources,
-    ));
-    let (_handle, mut service, _) = Service::new(client, media, None, EngineConfig::default())
-        .await
-        .unwrap();
-    service.sync.checkpoints.insert(
-        "mentions-connection".to_owned(),
-        Checkpoint {
-            pos: "position-before".to_owned(),
-            previous: None,
-        },
-    );
-    service.sync.observed_rooms.insert(room_id.to_owned());
-    service
-        .sync
-        .repair_coverage
-        .entry(room_id.to_owned())
-        .or_default()
-        .last_confirmed_ms = Some(1_000);
-
-    let events = [
-        EventFactory::new()
-            .room(room_id)
-            .text_msg("older event")
-            .sender(user_id!("@sender:localhost"))
-            .event_id(event_id!("$repair-older:localhost"))
-            .into_raw_sync(),
-        EventFactory::new()
-            .room(room_id)
-            .text_msg("latest event")
-            .sender(user_id!("@sender:localhost"))
-            .event_id(event_id!("$repair-latest:localhost"))
-            .into_raw_sync(),
-    ];
-    let timeline: Vec<_> = events
-        .iter()
-        .map(|event| serde_json::to_value(event).unwrap())
-        .collect();
-    let raw_room = serde_json::from_value(json!({
-        "limited": true,
-        "prev_batch": "repair-gap-token",
-        "timeline": timeline,
-    }))
-    .unwrap();
-    let batch = ResponseCommitBatch {
-        connection_id: "mentions-connection".to_owned(),
-        previous_pos: Some("position-before".to_owned()),
-        next_pos: "position-after".to_owned(),
-        response: Default::default(),
-        raw_rooms: [(room_id.to_owned(), raw_room)].into(),
-    };
-
-    service.batch(batch).await;
-
-    let coverage = service.sync.repair_coverage.get(room_id).unwrap();
-    assert_eq!(coverage.last_confirmed_ms, Some(1_000));
-    assert!(
-        coverage
-            .boundary_ms
-            .is_some_and(|boundary| boundary > 1_000)
-    );
-    assert!(coverage.unresolved);
-    assert!(!coverage.reliable);
-    assert_eq!(coverage.duration_ms(), None);
-    assert_eq!(
-        coverage.latest_anchor.as_deref(),
-        Some(event_id!("$repair-latest:localhost"))
-    );
-}
-
-#[tokio::test]
 async fn service_recovers_notification_history_into_durable_unread_inbox() {
     let server = MatrixMockServer::new().await;
     let room_id = room_id!("!mentions:localhost");

Modified file: src/mentions/state.rs+1 additions−22 deletions@@ -1,6 +1,6 @@
 use std::collections::{BTreeMap, BTreeSet};
 
-use ruma::{OwnedEventId, OwnedUserId};
+use ruma::OwnedUserId;
 use serde_json::Value;
 
 use crate::{
@@ -67,25 +67,6 @@ pub(super) struct Window {
     pub end: Option<String>,
 }
 
-#[derive(Clone, Debug, Default, PartialEq)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
-pub(crate) struct RepairCoverage {
-    pub(crate) last_confirmed_ms: Option<u64>,
-    pub(crate) boundary_ms: Option<u64>,
-    pub(crate) reliable: bool,
-    pub(crate) unresolved: bool,
-    pub(crate) latest_anchor: Option<OwnedEventId>,
-}
-
-impl RepairCoverage {
-    pub(crate) fn duration_ms(&self) -> Option<u64> {
-        if !self.reliable {
-            return None;
-        }
-        self.boundary_ms?.checked_sub(self.last_confirmed_ms?)
-    }
-}
-
 #[derive(Clone, Default, PartialEq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub(super) struct Data {
@@ -103,7 +84,6 @@ pub(super) struct Data {
     pub latest_cursor: Option<String>,
     pub latest_anchor: Option<String>,
     pub latest_head: Option<String>,
-    pub repair_policy: crate::HistoryRepairPolicy,
 }
 
 #[derive(Clone, Default, PartialEq)]
@@ -112,7 +92,6 @@ pub(super) struct SyncState {
     pub inbox_revision: u64,
     pub checkpoints: BTreeMap<String, Checkpoint>,
     pub observed_rooms: BTreeSet<ruma::OwnedRoomId>,
-    pub repair_coverage: BTreeMap<ruma::OwnedRoomId, RepairCoverage>,
 }
 
 #[derive(Clone, Debug, PartialEq)]

macltok

Commit 2c13c8db6174

Commit `2c13c8db6174`