macltok

public

Star0

Fork 0

Watch0

64 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `7cdaa86b9489`

tarball

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

12 files changed+1269-565

mentions: write the sync checkpoint apart from the inbox and never fail sync

Every response used to serialise, encrypt and commit the whole inbox inside the sync hook, and any failure there stopped room list sync or Account::build. The checkpoint, observed rooms and repair coverage now live in a small sync record written with every response, the inbox body only when it changed, both in one head compare-and-swap commit. Write failures, hook timeouts and unreadable stores become typed MentionGapReason gaps while the inbox carries on in memory, and a replay from the previous checkpoint is a superset rather than a discontinuity. Background completions are batched per tick, candidates expire after a week, decryption reports retry only their room, and mentions in a room left within the same response still classify.

Modified file: src/account.rs+1 additions−2 deletions@@ -859,8 +859,7 @@ impl Session {
             account,
             config.max_mentions_bytes,
         )
-        .await
-        .map_err(|error| account_error("opening the mentions store", error))?;
+        .await;
         Engine::with_persistence(
             self.client,
             config,

Modified file: src/engine/mod.rs+1−1

@@ -346,7 +346,7 @@ impl Engine {

346346 session_changes: broadcast::Receiver<SessionChange>,

347347 persistence: Option<Persistence>,

348348 store_lock: Option<StoreLock>,

349- mentions: Option<MentionsStore>,

349+ mentions: Option<Result<MentionsStore, crate::Error>>,

350350 ) -> Result<(Self, ClientHandle, EventStream), EngineError> {

351351 if client.user_id().is_none() {

352352 return Err(EngineError::Unauthenticated);

Modified file: src/lib.rs+3 additions−2 deletions@@ -45,8 +45,9 @@ pub use engine::{Engine, EngineConfig, EngineError};
 pub use error::{Error, ErrorKind, ServerError};
 pub use imbl;
 pub use mentions_model::{
-    MentionCaptureState, MentionCoverage, MentionEntry, MentionGap, MentionKey, MentionReason,
-    MentionsBoundary, MentionsCommand, MentionsPage, MentionsUpdate, SavedMentionContext,
+    MentionCaptureState, MentionCoverage, MentionEntry, MentionGap, MentionGapReason, MentionKey,
+    MentionReason, MentionsBoundary, MentionsCommand, MentionsPage, MentionsUpdate,
+    SavedMentionContext,
 };
 #[cfg(feature = "mirror")]
 pub use mirror::{

Modified file: src/mentions/service.rs+584 additions−309 deletions@@ -1,26 +1,35 @@
 use std::{
     collections::{BTreeMap, BTreeSet},
-    sync::Arc,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
     time::Duration,
 };
 
 use futures_util::{StreamExt, stream::FuturesUnordered};
 use matrix_sdk::{
-    Client, RoomState,
+    Client, Room, RoomState,
+    event_cache::RedecryptorReport,
     sliding_sync::{ResponseCommitBatch, ResponseCommitHook},
 };
 use ruma::{
-    OwnedEventId, OwnedRoomId, OwnedUserId,
-    events::{AnySyncTimelineEvent, room::encrypted::OriginalSyncRoomEncryptedEvent},
+    OwnedEventId, OwnedRoomId, OwnedUserId, RoomId,
+    events::{
+        AnySyncTimelineEvent, ignored_user_list::IgnoredUserListEventContent,
+        room::encrypted::OriginalSyncRoomEncryptedEvent,
+    },
     serde::Raw,
 };
 use serde_json::Value;
 use tokio::sync::{mpsc, oneshot, watch};
 
-use super::state::{Candidate, Data, RepairCoverage, SavedEvent, Window, id, invalid, now};
+use super::state::{
+    Candidate, Checkpoint, Data, RepairCoverage, SavedEvent, SyncState, Window, id, invalid, now,
+};
 use crate::{
     CommandOutput, EngineConfig, Error, ErrorKind, Event, HistoryRepairPolicy, MentionCaptureState,
-    MentionEntry, MentionGap, MentionKey, MentionReason, MentionsCommand, MentionsPage,
+    MentionEntry, MentionGapReason, MentionKey, MentionReason, MentionsCommand, MentionsPage,
     MentionsUpdate, TimelineContent, TimelineItem,
     engine::{MediaService, MentionsStore, context, executor::EngineBoxFuture},
     normalise::{MediaRegistration, normalise_observed},
@@ -31,11 +40,17 @@ use crate::{
 #[path = "service_tests.rs"]
 mod tests;
 
+const HOOK_TIMEOUT: Duration = Duration::from_secs(10);
+const FLUSH_RETRY_MS: u64 = 60_000;
+
 #[derive(Clone)]
-pub(crate) struct Handle(mpsc::Sender<Request>);
+pub(crate) struct Handle {
+    requests: mpsc::Sender<Request>,
+    timed_out: Arc<AtomicBool>,
+}
 
 enum Request {
-    Batch(ResponseCommitBatch, oneshot::Sender<Result<(), Error>>),
+    Batch(ResponseCommitBatch, oneshot::Sender<()>),
     Command(
         MentionsCommand,
         oneshot::Sender<Result<CommandOutput, Error>>,
@@ -53,6 +68,12 @@ pub(crate) struct Service {
     media: Arc<MediaService>,
     store: Option<MentionsStore>,
     data: Data,
+    sync: SyncState,
+    dirty: bool,
+    flush_after: u64,
+    ignored: BTreeSet<OwnedUserId>,
+    leaving: BTreeSet<OwnedRoomId>,
+    timed_out: Arc<AtomicBool>,
     config: EngineConfig,
     requests: mpsc::Receiver<Request>,
     updates: watch::Sender<MentionsPage>,
@@ -76,28 +97,37 @@ enum Work {
 }
 
 impl Handle {
+    /// never fails, a mentions failure becomes a coverage gap rather than a stalled sync
     pub(crate) fn hook(&self) -> ResponseCommitHook {
         let handle = self.clone();
         ResponseCommitHook::new(move |batch| {
             let handle = handle.clone();
             async move {
-                let (reply, response) = oneshot::channel();
-                handle
-                    .0
-                    .send(Request::Batch(batch, reply))
-                    .await
-                    .map_err(|_| "mentions service stopped".to_owned())?;
-                response
-                    .await
-                    .map_err(|_| "mentions commit acknowledgement dropped".to_owned())?
-                    .map_err(|error| error.to_string())
+                let committed = async {
+                    let (reply, response) = oneshot::channel();
+                    handle
+                        .requests
+                        .send(Request::Batch(batch, reply))
+                        .await
+                        .ok()?;
+                    response.await.ok()
+                };
+                match matrix_sdk::timeout::timeout(committed, HOOK_TIMEOUT).await {
+                    Ok(Some(())) => {}
+                    Ok(None) => tracing::warn!("mentions service stopped, sync carries on"),
+                    Err(_) => {
+                        handle.timed_out.store(true, Ordering::Release);
+                        tracing::warn!("mentions commit timed out, sync carries on");
+                    }
+                }
+                Ok(())
             }
         })
     }
 
     pub(crate) async fn execute(&self, command: MentionsCommand) -> Result<CommandOutput, Error> {
         let (reply, response) = oneshot::channel();
-        self.0
+        self.requests
             .send(Request::Command(command, reply))
             .await
             .map_err(|_| Error::interrupted())?;
@@ -106,7 +136,7 @@ impl Handle {
 
     pub(crate) async fn repair_policy(&self) -> Result<HistoryRepairPolicy, Error> {
         let (reply, response) = oneshot::channel();
-        self.0
+        self.requests
             .send(Request::GetRepairPolicy(reply))
             .await
             .map_err(|_| Error::interrupted())?;
@@ -115,7 +145,7 @@ impl Handle {
 
     pub(crate) async fn set_repair_policy(&self, policy: HistoryRepairPolicy) -> Result<(), Error> {
         let (reply, response) = oneshot::channel();
-        self.0
+        self.requests
             .send(Request::SetRepairPolicy(policy, reply))
             .await
             .map_err(|_| Error::interrupted())?;
@@ -127,7 +157,7 @@ impl Handle {
         room_id: OwnedRoomId,
     ) -> Result<RepairCoverage, Error> {
         let (reply, response) = oneshot::channel();
-        self.0
+        self.requests
             .send(Request::GetRepairCoverage(room_id, reply))
             .await
             .map_err(|_| Error::interrupted())?;
@@ -136,7 +166,7 @@ impl Handle {
 
     pub(crate) async fn repair_resolved(&self, room_id: OwnedRoomId) -> Result<(), Error> {
         let (reply, response) = oneshot::channel();
-        self.0
+        self.requests
             .send(Request::RepairResolved(room_id, reply))
             .await
             .map_err(|_| Error::interrupted())?;
@@ -145,17 +175,91 @@ impl Handle {
 
     pub(crate) async fn stop(&self) {
         let (reply, response) = oneshot::channel();
-        if self.0.send(Request::Stop(reply)).await.is_ok() {
+        if self.requests.send(Request::Stop(reply)).await.is_ok() {
             let _ = response.await;
         }
     }
 }
 
+#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+async fn open_store(
+    store: Option<Result<MentionsStore, Error>>,
+) -> (
+    Option<MentionsStore>,
+    Data,
+    SyncState,
+    Vec<MentionGapReason>,
+) {
+    let mut store = match store {
+        None => return (None, Data::default(), SyncState::default(), Vec::new()),
+        Some(Err(error)) => {
+            tracing::error!(%error, "mentions store unavailable, the inbox runs in memory");
+            return (
+                None,
+                Data::default(),
+                SyncState::default(),
+                vec![MentionGapReason::StoreFailed],
+            );
+        }
+        Some(Ok(store)) => store,
+    };
+    let loaded = (|| {
+        let data: Data = store.read("inbox")?.unwrap_or_default();
+        let sync: SyncState = store.read("sync")?.unwrap_or_default();
+        if sync.inbox_revision != data.revision {
+            return Err(Error::new(
+                ErrorKind::Store,
+                "mentions checkpoint does not match its inbox",
+            ));
+        }
+        Ok((data, sync))
+    })();
+    let (data, sync) = match loaded {
+        Ok(loaded) => loaded,
+        Err(error) => {
+            tracing::warn!(%error, "resetting an unreadable mentions inbox");
+            if let Err(error) = store.reset().await {
+                tracing::error!(%error, "mentions store unavailable, the inbox runs in memory");
+                return (
+                    None,
+                    Data::default(),
+                    SyncState::default(),
+                    vec![MentionGapReason::StoreFailed],
+                );
+            }
+            (Data::default(), SyncState::default())
+        }
+    };
+    let gaps = if store.was_reset() {
+        vec![MentionGapReason::StoreReset]
+    } else {
+        Vec::new()
+    };
+    (Some(store), data, sync, gaps)
+}
+
+#[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
+async fn open_store(
+    store: Option<Result<MentionsStore, Error>>,
+) -> (
+    Option<MentionsStore>,
+    Data,
+    SyncState,
+    Vec<MentionGapReason>,
+) {
+    let gaps = match store {
+        None => Vec::new(),
+        Some(Err(_)) => vec![MentionGapReason::StoreFailed],
+        Some(Ok(store)) => match store {},
+    };
+    (None, Data::default(), SyncState::default(), gaps)
+}
+
 impl Service {
     pub(crate) async fn new(
         client: Client,
         media: Arc<MediaService>,
-        store: Option<MentionsStore>,
+        store: Option<Result<MentionsStore, Error>>,
         config: EngineConfig,
     ) -> Result<(Handle, Self, watch::Receiver<MentionsPage>), Error> {
         let account = client
@@ -167,24 +271,9 @@ impl Service {
                 )
             })?
             .to_owned();
-        #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
-        let data: Data = store
-            .as_ref()
-            .map(|store| store.read("inbox"))
-            .transpose()?
-            .flatten()
-            .unwrap_or_default();
-        #[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
-        let data = Data::default();
-        #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
-        if store
-            .as_ref()
-            .is_some_and(|store| store.revision() != data.revision)
-        {
-            return Err(Error::new(
-                ErrorKind::Store,
-                "mentions revision does not match its checkpoint",
-            ));
+        let (store, mut data, sync, gaps) = open_store(store).await;
+        for reason in &gaps {
+            data.gap(None, *reason, None);
         }
         let page = data.page(&account, None, 50)?;
         media
@@ -198,63 +287,53 @@ impl Service {
             .await;
         let (updates, receiver) = watch::channel(page);
         let (sender, requests) = mpsc::channel(config.command_capacity.max(1));
+        let timed_out = Arc::new(AtomicBool::new(false));
+        let mut service = Self {
+            client,
+            account,
+            media,
+            store,
+            data,
+            sync,
+            dirty: !gaps.is_empty(),
+            flush_after: 0,
+            ignored: BTreeSet::new(),
+            leaving: BTreeSet::new(),
+            timed_out: timed_out.clone(),
+            config,
+            requests,
+            updates,
+            recovery_budget: 20,
+            latest_scan: true,
+            recovery_retry: 0,
+        };
+        service.ignored = service.ignored_users().await;
         Ok((
-            Handle(sender),
-            Self {
-                client,
-                account,
-                media,
-                store,
-                data,
-                config,
-                requests,
-                updates,
-                recovery_budget: 20,
-                latest_scan: true,
-                recovery_retry: 0,
+            Handle {
+                requests: sender,
+                timed_out,
             },
+            service,
             receiver,
         ))
     }
 
-    async fn save(&mut self, mut next: Data) -> Result<(), Error> {
-        next.revision = self
-            .data
-            .revision
-            .checked_add(1)
-            .ok_or_else(|| invalid("mentions revision exhausted"))?;
-        next.bound(
-            self.config.max_mentions_entries,
-            self.config.max_mention_context_events,
-            self.config.max_mention_candidates,
-        );
-        #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
-        if let Some(store) = &mut self.store {
-            loop {
-                let value = serde_json::to_value(&next).map_err(|error| {
-                    Error::with_source(
-                        ErrorKind::Store,
-                        "encoding the mentions inbox failed",
-                        error,
-                    )
-                })?;
-                match store
-                    .commit(
-                        self.data.revision,
-                        vec![super::store::Write::Set("inbox".to_owned(), value)],
-                    )
-                    .await
-                {
-                    Ok(_) => break,
-                    Err(error)
-                        if error.message == "mentions storage limit exceeded"
-                            && next.evict_one() => {}
-                    Err(error) => return Err(error),
-                }
-            }
-        }
-        self.data = next;
-        let page = self.data.page(&self.account, None, 50)?;
+    async fn ignored_users(&self) -> BTreeSet<OwnedUserId> {
+        self.client
+            .account()
+            .account_data::<IgnoredUserListEventContent>()
+            .await
+            .ok()
+            .flatten()
+            .and_then(|raw| raw.deserialize().ok())
+            .map(|content| content.ignored_users.into_keys().collect())
+            .unwrap_or_default()
+    }
+
+    async fn publish(&mut self) {
+        let Ok(page) = self.data.page(&self.account, None, 50) else {
+            return;
+        };
         self.media
             .register(
                 page.entries
@@ -265,24 +344,110 @@ impl Service {
             )
             .await;
         self.updates.send_replace(page);
+    }
+
+    /// writes the sync record, and the inbox with it when `dirty`, in one commit so a
+    /// checkpoint never lands without the inbox changes it covers
+    async fn flush(&mut self) -> Result<(), Error> {
+        if self.dirty {
+            self.advance()?;
+        }
+        #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+        if let Some(store) = &mut self.store {
+            use super::state::json_len;
+            use super::store::Write;
+
+            let encode = |error| {
+                Error::with_source(
+                    ErrorKind::Store,
+                    "encoding the mentions inbox failed",
+                    error,
+                )
+            };
+            let sync = serde_json::to_value(&self.sync).map_err(encode)?;
+            let mut writes = Vec::new();
+            if self.dirty {
+                let available = store
+                    .available(&["inbox", "sync"])
+                    .saturating_sub(store.sealed_bound("sync", json_len(&sync)));
+                let mut size = store.sealed_bound("inbox", json_len(&self.data));
+                while size > available {
+                    let Some(freed) = self.data.evict_one() else {
+                        break;
+                    };
+                    size = size.saturating_sub(freed * 3);
+                }
+                writes.push(Write::Set(
+                    "inbox".to_owned(),
+                    serde_json::to_value(&self.data).map_err(encode)?,
+                ));
+            }
+            writes.push(Write::Set("sync".to_owned(), sync));
+            store.commit(store.revision(), writes).await?;
+        }
+        if self.dirty {
+            self.dirty = false;
+            self.publish().await;
+        }
         Ok(())
     }
 
+    fn advance(&mut self) -> Result<(), Error> {
+        self.data.revision = self
+            .data
+            .revision
+            .checked_add(1)
+            .ok_or_else(|| invalid("mentions revision exhausted"))?;
+        self.data.bound(
+            self.config.max_mentions_entries,
+            self.config.max_mention_context_events,
+            self.config.max_mention_candidates,
+        );
+        self.sync.inbox_revision = self.data.revision;
+        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();
+        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.dirty = dirty;
+        }
+        result
+    }
+
+    async fn store_failed(&mut self, error: Error) {
+        tracing::error!(%error, "saving the mentions inbox failed");
+        self.data.gap(None, MentionGapReason::StoreFailed, None);
+        self.dirty = true;
+        self.flush_after = now().saturating_add(FLUSH_RETRY_MS);
+        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.save(next).await
+        self.apply(Some(next), None).await
     }
 
     async fn repair_resolved(&mut self, room_id: OwnedRoomId) -> Result<(), Error> {
-        let mut next = self.data.clone();
-        let coverage = next.repair_coverage.entry(room_id.clone()).or_default();
+        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;
-        next.observed_rooms.insert(room_id);
-        self.save(next).await
+        sync.observed_rooms.insert(room_id);
+        self.apply(None, Some(sync)).await
     }
 
     async fn command(&mut self, command: MentionsCommand) -> Result<CommandOutput, Error> {
@@ -331,7 +496,7 @@ impl Service {
                     start: None,
                     end: None,
                 });
-                self.save(next).await?;
+                self.apply(Some(next), None).await?;
                 let (context, media) = self.data.context(&key)?;
                 self.media.register(media).await;
                 Ok(CommandOutput::MentionContext(context))
@@ -344,7 +509,7 @@ impl Service {
                         entry.revision = self.data.revision + 1;
                     }
                 }
-                self.save(next).await?;
+                self.apply(Some(next), None).await?;
                 Ok(CommandOutput::Done)
             }
             MentionsCommand::ReadThrough(boundary) => {
@@ -360,7 +525,7 @@ impl Service {
                     entry.read = true;
                     entry.revision = self.data.revision + 1;
                 }
-                self.save(next).await?;
+                self.apply(Some(next), None).await?;
                 Ok(CommandOutput::Done)
             }
             MentionsCommand::Recover => {
@@ -370,61 +535,55 @@ impl Service {
                 let mut next = self.data.clone();
                 next.coverage.recovering = true;
                 next.coverage.historical_supported = None;
-                self.save(next).await?;
+                self.apply(Some(next), None).await?;
                 Ok(CommandOutput::Done)
             }
         }
     }
 
-    fn gap(data: &mut Data, room_id: Option<OwnedRoomId>, reason: &str, token: Option<String>) {
-        if !data
-            .coverage
-            .gaps
-            .iter()
-            .any(|gap| gap.room_id == room_id && gap.reason == reason && gap.token == token)
-        {
-            data.coverage.gaps.push(MentionGap {
-                room_id,
-                reason: reason.to_owned(),
-                token,
-            });
-            if data.coverage.gaps.len() > 128 {
-                data.coverage.gaps.remove(0);
-                data.coverage.truncated = true;
-            }
-        }
+    fn observable(&self, room_id: &RoomId) -> Option<Room> {
+        self.client
+            .get_room(room_id)
+            .filter(|room| room.state() == RoomState::Joined || self.leaving.contains(room_id))
     }
 
-    async fn batch(&mut self, batch: ResponseCommitBatch) -> Result<(), Error> {
+    async fn batch(&mut self, batch: ResponseCommitBatch) {
+        let checkpoint = self.sync.checkpoints.get(&batch.connection_id).cloned();
+        if checkpoint
+            .as_ref()
+            .is_some_and(|checkpoint| checkpoint.pos == batch.next_pos)
+        {
+            return;
+        }
+        self.ignored = self.ignored_users().await;
+        self.leaving = batch.response.rooms.left.keys().cloned().collect();
         let mut next = self.data.clone();
-        if next.checkpoints.get(&batch.connection_id) == Some(&batch.next_pos) {
-            return Ok(());
+        let mut sync = self.sync.clone();
+        if self.timed_out.swap(false, Ordering::AcqRel) {
+            next.gap(None, MentionGapReason::HookTimedOut, None);
         }
         let observed_at_ms = now();
-        let checkpoint_discontinuity =
-            next.checkpoints.get(&batch.connection_id) != batch.previous_pos.as_ref();
-        if checkpoint_discontinuity {
-            Self::gap(
-                &mut next,
-                None,
-                "sync checkpoint continuity is unknown",
-                None,
-            );
-            let observed_rooms: Vec<_> = next.observed_rooms.iter().cloned().collect();
+        // 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 {
+            None => batch.previous_pos.is_none(),
+            Some(checkpoint) => {
+                batch.previous_pos.as_ref() == Some(&checkpoint.pos)
+                    || batch.previous_pos == checkpoint.previous
+            }
+        };
+        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 next, &room_id, observed_at_ms);
+                Self::mark_unresolved(&mut sync, &room_id, observed_at_ms);
             }
         }
         for (room_id, raw) in &batch.raw_rooms {
-            if raw.limited || !next.observed_rooms.contains(room_id) {
-                Self::gap(
-                    &mut next,
+            if raw.limited && sync.observed_rooms.contains(room_id) {
+                next.gap(
                     Some(room_id.clone()),
-                    if raw.limited {
-                        "limited sync timeline"
-                    } else {
-                        "initial room history is incomplete"
-                    },
+                    MentionGapReason::LimitedTimeline,
                     raw.prev_batch.clone(),
                 );
             }
@@ -433,36 +592,32 @@ impl Service {
             .raw_rooms
             .keys()
             .chain(batch.response.rooms.joined.keys())
+            .chain(batch.response.rooms.left.keys())
             .cloned()
             .collect();
         for room_id in &room_ids {
-            if !self
-                .client
-                .get_room(room_id)
-                .is_some_and(|room| room.state() == RoomState::Joined)
-            {
+            if self.observable(room_id).is_none() {
                 continue;
             }
-            let previously_observed = next.observed_rooms.contains(room_id);
+            let previously_observed = sync.observed_rooms.contains(room_id);
             let limited = batch.raw_rooms.get(room_id).is_some_and(|raw| raw.limited);
-            if checkpoint_discontinuity || limited || !previously_observed {
-                Self::mark_unresolved(&mut next, room_id, observed_at_ms);
+            if !continuous || limited || !previously_observed {
+                Self::mark_unresolved(&mut sync, room_id, observed_at_ms);
             } else {
-                let coverage = next.repair_coverage.entry(room_id.clone()).or_default();
+                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;
             }
-            next.observed_rooms.insert(room_id.clone());
+            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 value: Value =
-                        serde_json::from_str(event.json().get()).map_err(|error| {
-                            Error::with_source(ErrorKind::Invalid, "invalid observed event", error)
-                        })?;
+                    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);
@@ -470,44 +625,51 @@ impl Service {
                 }
             }
             if let Some(latest_anchor) = latest_anchor {
-                next.repair_coverage
+                sync.repair_coverage
                     .entry(room_id.clone())
                     .or_default()
                     .latest_anchor = Some(latest_anchor);
             }
-            if let Some(update) = batch.response.rooms.joined.get(room_id) {
-                for event in &update.timeline.events {
-                    let value: Value =
-                        serde_json::from_str(event.raw().json().get()).map_err(|error| {
-                            Error::with_source(ErrorKind::Invalid, "invalid observed event", error)
-                        })?;
-                    if let Some(event_id) = value["event_id"].as_str() {
-                        observed.insert(event_id.to_owned(), value);
-                    }
+            let processed = batch
+                .response
+                .rooms
+                .joined
+                .get(room_id)
+                .map(|update| &update.timeline.events)
+                .into_iter()
+                .chain(
+                    batch
+                        .response
+                        .rooms
+                        .left
+                        .get(room_id)
+                        .map(|update| &update.timeline.events),
+                )
+                .flatten();
+            for event in processed {
+                let Ok(value) = serde_json::from_str::<Value>(event.raw().json().get()) else {
+                    continue;
+                };
+                if let Some(event_id) = value["event_id"].as_str() {
+                    observed.insert(event_id.to_owned(), value);
                 }
             }
             let mut observed: Vec<_> = observed.into_values().collect();
             observed.sort_by_key(|value| value["origin_server_ts"].as_u64().unwrap_or(0));
             for value in observed {
-                self.observe(&mut next, room_id, value).await?;
+                self.observe(&mut next, room_id, value).await;
             }
         }
         for room_id in batch.response.rooms.left.keys() {
-            Self::gap(
-                &mut next,
-                Some(room_id.clone()),
-                "room left, retained context is read only",
-                None,
-            );
+            next.gap(Some(room_id.clone()), MentionGapReason::RoomLeft, None);
             next.candidates
                 .retain(|_, candidate| &candidate.room_id != room_id);
-            for entry in next
-                .entries
-                .values_mut()
-                .filter(|entry| &entry.key.room_id == room_id)
-            {
+            let revision = next.revision + 1;
+            for entry in next.entries.values_mut().filter(|entry| {
+                &entry.key.room_id == room_id && entry.capture != MentionCaptureState::Unavailable
+            }) {
                 entry.capture = MentionCaptureState::Unavailable;
-                entry.revision = next.revision + 1;
+                entry.revision = revision;
             }
         }
         let reevaluate: Vec<_> = next
@@ -522,15 +684,34 @@ impl Service {
             })
             .collect();
         for (room_id, raw) in reevaluate {
-            self.observe(&mut next, &room_id, raw).await?;
+            self.observe(&mut next, &room_id, raw).await;
+        }
+        self.filter_ignored(&mut next);
+        self.leaving.clear();
+        sync.checkpoints.insert(
+            batch.connection_id,
+            Checkpoint {
+                pos: batch.next_pos,
+                previous: batch.previous_pos,
+            },
+        );
+        let changed = next != self.data;
+        if changed {
+            self.data = next;
+            self.dirty = true;
+        }
+        self.sync = sync;
+        if now() < self.flush_after {
+            if changed && self.advance().is_ok() {
+                self.publish().await;
+            }
+        } else if let Err(error) = self.flush().await {
+            self.store_failed(error).await;
         }
-        self.filter_ignored(&mut next).await;
-        next.checkpoints.insert(batch.connection_id, batch.next_pos);
-        self.save(next).await
     }
 
-    fn mark_unresolved(data: &mut Data, room_id: &OwnedRoomId, observed_at_ms: u64) {
-        let coverage = data.repair_coverage.entry(room_id.clone()).or_default();
+    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;
@@ -538,33 +719,29 @@ impl Service {
         coverage.reliable = false;
     }
 
-    async fn filter_ignored(&self, next: &mut Data) {
-        let mut removed = Vec::new();
-        for (key, entry) in &next.entries {
-            if let TimelineItem::Event(event) = &entry.event {
-                if self.client.is_user_ignored(&event.sender.user_id).await {
-                    removed.push(key.clone());
-                }
-            }
-        }
-        for key in removed {
-            next.entries.remove(&key);
-            next.windows.remove(&key);
+    fn filter_ignored(&self, next: &mut Data) -> bool {
+        let removed: Vec<_> = next
+            .entries
+            .iter()
+            .filter(|(_, entry)| {
+                matches!(&entry.event, TimelineItem::Event(event) if self.ignored.contains(&event.sender.user_id))
+            })
+            .map(|(key, _)| key.clone())
+            .collect();
+        for key in &removed {
+            next.entries.remove(key);
+            next.windows.remove(key);
         }
         next.collect_events();
+        !removed.is_empty()
     }
 
-    async fn observe(
-        &self,
-        data: &mut Data,
-        room_id: &OwnedRoomId,
-        mut value: Value,
-    ) -> Result<(), Error> {
+    async fn observe(&self, data: &mut Data, room_id: &OwnedRoomId, mut value: Value) {
         let Some(event_id) = value["event_id"]
             .as_str()
             .and_then(|value| value.parse().ok())
         else {
-            return Ok(());
+            return;
         };
         let key = MentionKey {
             room_id: room_id.clone(),
@@ -572,7 +749,7 @@ impl Service {
         };
         let identity = id(&key);
         let Some(timestamp) = value["origin_server_ts"].as_u64() else {
-            return Ok(());
+            return;
         };
         data.coverage.earliest_timestamp_ms = Some(
             data.coverage
@@ -604,7 +781,7 @@ impl Service {
                 }
                 data.touch_contexts(&target);
             }
-            return Ok(());
+            return;
         }
         if data.redacted.contains_key(&identity)
             || value["unsigned"].get("redacted_because").is_some()
@@ -617,7 +794,7 @@ impl Service {
                 redact(saved);
             }
             data.touch_contexts(&identity);
-            return Ok(());
+            return;
         }
         if value["content"]["m.relates_to"]["rel_type"] == "m.replace" {
             if let Some(target) = value["content"]["m.relates_to"]["event_id"]
@@ -629,7 +806,7 @@ impl Service {
                     event_id: target,
                 });
                 if data.redacted.contains_key(&target) {
-                    return Ok(());
+                    return;
                 }
                 let newer = data.edits.get(&target).is_none_or(|old| {
                     (old["origin_server_ts"].as_u64(), old["event_id"].as_str())
@@ -640,24 +817,19 @@ impl Service {
                         data.events.get(&target).and_then(|saved| saved.raw.clone())
                     {
                         if original["sender"] != value["sender"] {
-                            return Ok(());
+                            return;
                         }
                         data.edits.insert(target.clone(), value.clone());
                         value = edited(original, &value);
                         return Box::pin(self.observe(data, room_id, value)).await;
                     }
-                    data.candidates.insert(
-                        identity,
-                        Candidate {
-                            room_id: room_id.clone(),
-                            raw: value,
-                            attempts: 0,
-                            retry_at_ms: 0,
-                        },
-                    );
+                    data.candidates
+                        .entry(identity)
+                        .and_modify(|candidate| candidate.raw = value.clone())
+                        .or_insert_with(|| Candidate::new(room_id.clone(), value, 0));
                 }
             }
-            return Ok(());
+            return;
         }
         if let Some(edit) = value["unsigned"]["m.relations"].get("m.replace").cloned() {
             if edit["sender"] == value["sender"]
@@ -683,28 +855,20 @@ impl Service {
             data.edits.insert(identity.clone(), edit);
         }
         if value["type"] == "m.room.encrypted" {
-            data.candidates.entry(identity).or_insert(Candidate {
-                room_id: room_id.clone(),
-                raw: value,
-                attempts: 0,
-                retry_at_ms: 0,
-            });
-            return Ok(());
+            data.candidates
+                .entry(identity)
+                .or_insert_with(|| Candidate::new(room_id.clone(), value, 0));
+            return;
         }
         let Some((item, media)) = normalise_observed(&value, Some(&self.account), &identity) else {
-            return Ok(());
+            return;
         };
         self.supplement(data, &key, &item, &media, &value);
         let (reasons, suppressed, unresolved) = self.classify(data, room_id, &value).await;
         if unresolved {
             data.candidates
                 .entry(identity.clone())
-                .or_insert(Candidate {
-                    room_id: room_id.clone(),
-                    raw: value.clone(),
-                    attempts: 0,
-                    retry_at_ms: now() + 30_000,
-                });
+                .or_insert_with(|| Candidate::new(room_id.clone(), value.clone(), now() + 30_000));
         } else {
             data.candidates.remove(&identity);
         }
@@ -721,7 +885,7 @@ impl Service {
                     },
                 );
             }
-            return Ok(());
+            return;
         }
         let previous = data.entries.get(&identity);
         let changed = previous.is_none_or(|entry| {
@@ -769,7 +933,6 @@ impl Service {
                 raw: Some(value),
             },
         );
-        Ok(())
     }
 
     fn supplement(
@@ -860,16 +1023,12 @@ impl Service {
             return (vec![], false, false);
         };
         if sender == self.account
-            || self.client.is_user_ignored(&sender).await
+            || self.ignored.contains(&sender)
             || value["type"] != "m.room.message"
         {
             return (vec![], false, false);
         }
-        let Some(room) = self
-            .client
-            .get_room(room_id)
-            .filter(|room| room.state() == RoomState::Joined)
-        else {
+        let Some(room) = self.observable(room_id) else {
             return (vec![], false, false);
         };
         let raw: Raw<AnySyncTimelineEvent> =
@@ -939,100 +1098,216 @@ impl Service {
         let mut retries = BTreeSet::new();
         let mut history = false;
         let mut next_scan = now() + 60_000;
+        let mut next_expiry = 0;
         let mut tick = Box::pin(matrix_sdk::sleep::sleep(Duration::from_secs(1)));
         loop {
             self.schedule(&mut work, &mut captures, &mut retries, &mut history);
             tokio::select! {
                 biased;
                 request = self.requests.recv() => match request {
-                    Some(Request::Stop(reply)) => { let _ = reply.send(()); return; }
-                    Some(Request::Batch(batch, reply)) => { let result = self.batch(batch).await; let _ = reply.send(result); }
+                    Some(Request::Stop(reply)) => {
+                        if self.dirty && let Err(error) = self.flush().await {
+                            tracing::error!(%error, "saving the mentions inbox at shutdown failed");
+                        }
+                        let _ = reply.send(());
+                        return;
+                    }
+                    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.data.repair_coverage.get(&room_id).cloned().unwrap_or_default(); let _ = reply.send(Ok(coverage)); }
+                    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() => {
-                    let mut next = self.data.clone();
-                    let recovery = (self.latest_scan, self.recovery_budget);
+                    let mut next = std::mem::take(&mut self.data);
+                    // a candidate that failed again only moved its retry time, which needs no write
+                    let changed = self.complete(&mut next, result, &mut captures, &mut retries, &mut history).await;
+                    self.data = next;
+                    self.dirty |= changed;
+                }
+                Some(report) = reports.next() => {
+                    let room = match report { Ok(RedecryptorReport::ResolvedUtds { room_id, .. }) => Some(room_id), _ => None };
+                    for candidate in self.data.candidates.values_mut().filter(|candidate| candidate.encrypted() && room.as_ref().is_none_or(|room| *room == candidate.room_id)) { candidate.retry_at_ms = 0; }
+                }
+                Some(_) = ignores.next() => {
+                    self.ignored = self.ignored_users().await;
+                    let mut next = std::mem::take(&mut self.data);
+                    self.dirty |= self.filter_ignored(&mut next);
+                    self.data = next;
+                }
+                () = &mut tick => {
+                    tick = Box::pin(matrix_sdk::sleep::sleep(Duration::from_secs(1)));
+                    if now() >= next_scan && !history { self.recovery_budget = 20; self.latest_scan = true; next_scan = now() + 60_000; }
+                    if now() >= next_expiry { self.dirty |= self.data.expire_candidates(now()); next_expiry = now() + 60_000; }
+                    if self.dirty && now() >= self.flush_after && let Err(error) = self.flush().await { self.store_failed(error).await; }
+                }
+            }
+        }
+    }
+
+    async fn complete(
+        &mut self,
+        next: &mut Data,
+        result: Work,
+        captures: &mut BTreeSet<String>,
+        retries: &mut BTreeSet<String>,
+        history: &mut bool,
+    ) -> bool {
+        match result {
+            Work::Capture(key, revision, result) => {
+                captures.remove(&key);
+                let observed = result
+                    .as_ref()
+                    .ok()
+                    .map(|(_, _, observed)| observed.clone())
+                    .unwrap_or_default();
+                let room_id = next
+                    .entries
+                    .get(&key)
+                    .filter(|entry| entry.revision == revision)
+                    .map(|entry| entry.key.room_id.clone())
+                    .filter(|room_id| {
+                        self.client
+                            .get_room(room_id)
+                            .is_some_and(|room| room.state() == RoomState::Joined)
+                    });
+                self.captured(next, key, revision, result);
+                if let Some(room_id) = room_id {
+                    for value in observed {
+                        let identity = value["event_id"]
+                            .as_str()
+                            .map(|event_id| format!("{room_id}\n{event_id}"));
+                        if identity.as_ref().is_some_and(|key| {
+                            next.events
+                                .get(key)
+                                .is_some_and(|event| event.raw.is_some())
+                        }) {
+                            continue;
+                        }
+                        self.observe(next, &room_id, value).await;
+                    }
+                }
+            }
+            Work::Candidate(key, result) => {
+                retries.remove(&key);
+                if let Some(mut candidate) = next.candidates.remove(&key) {
                     match result {
-                        Work::Capture(key, revision, result) => {
-                            captures.remove(&key);
-                            let observed = result.as_ref().ok().map(|(_, _, observed)| observed.clone()).unwrap_or_default();
-                            let room_id = next.entries.get(&key).filter(|entry| entry.revision == revision)
-                                .map(|entry| entry.key.room_id.clone())
-                                .filter(|room_id| self.client.get_room(room_id).is_some_and(|room| room.state() == RoomState::Joined));
-                            self.captured(&mut next, key, revision, result);
-                            if let Some(room_id) = room_id {
-                                for value in observed {
-                                    let identity = value["event_id"].as_str().map(|event_id| format!("{room_id}\n{event_id}"));
-                                    if identity.as_ref().is_some_and(|key| next.events.get(key).is_some_and(|event| event.raw.is_some())) { continue; }
-                                    self.observe(&mut next, &room_id, value).await.ok();
-                                }
+                        Ok(value) => {
+                            self.observe(next, &candidate.room_id, value).await;
+                            if let Some(unresolved) = next.candidates.get_mut(&key) {
+                                unresolved.attempts = candidate.attempts.saturating_add(1);
+                                unresolved.first_seen_ms = candidate.first_seen_ms;
+                                unresolved.retry_at_ms =
+                                    now() + (1000u64 << unresolved.attempts.min(8));
                             }
                         }
-                        Work::Candidate(key, result) => {
-                            retries.remove(&key);
-                            if let Some(mut candidate) = next.candidates.remove(&key) {
-                                match result {
-                                    Ok(value) => {
-                                        if let Err(error) = self.observe(&mut next, &candidate.room_id, value).await { tracing::warn!(%error, "mention retry failed"); }
-                                        if let Some(unresolved) = next.candidates.get_mut(&key) { unresolved.attempts = candidate.attempts.saturating_add(1); unresolved.retry_at_ms = now() + (1000u64 << unresolved.attempts.min(8)); }
-                                    }
-                                    Err(error) => {
-                                        if candidate.raw["type"] != "m.room.encrypted" && (matches!(error.kind, ErrorKind::EventNotFound | ErrorKind::Forbidden) || error.server.as_ref().is_some_and(|server| matches!(server.status, 403 | 404))) {
-                                            Self::gap(&mut next, Some(candidate.room_id), "mention notification context is unavailable", None);
-                                        } else { candidate.attempts = candidate.attempts.saturating_add(1); candidate.retry_at_ms = retry_at(&error, 1000u64 << candidate.attempts.min(8)); next.candidates.insert(key, candidate); }
-                                    }
-                                }
+                        Err(error) => {
+                            if !candidate.encrypted()
+                                && (matches!(
+                                    error.kind,
+                                    ErrorKind::EventNotFound | ErrorKind::Forbidden
+                                ) || error
+                                    .server
+                                    .as_ref()
+                                    .is_some_and(|server| matches!(server.status, 403 | 404)))
+                            {
+                                next.gap(
+                                    Some(candidate.room_id),
+                                    MentionGapReason::ContextUnavailable,
+                                    None,
+                                );
+                            } else {
+                                candidate.attempts = candidate.attempts.saturating_add(1);
+                                candidate.retry_at_ms =
+                                    retry_at(&error, 1000u64 << candidate.attempts.min(8));
+                                next.candidates.insert(key, candidate);
+                                return false;
                             }
                         }
-                        Work::History(latest, from, result) => {
-                            history = false; self.recovery_budget = self.recovery_budget.saturating_sub(1);
-                            match result {
-                                Ok(page) => {
-                                    next.coverage.historical_supported = Some(true);
-                                    let mut reached_anchor = false;
-                                    for notification in page.notifications {
-                                        if let Some(event_id) = notification.event.get_field::<String>("event_id").ok().flatten() {
-                                            let identity = format!("{}\n{}", notification.room_id, event_id);
-                                            if latest && from.is_none() && next.latest_head.is_none() { next.latest_head = Some(identity.clone()); }
-                                            if latest && next.latest_anchor.as_ref() == Some(&identity) { reached_anchor = true; break; }
-                                        }
-                                        if self.client.get_room(&notification.room_id).is_some_and(|room| room.state() == RoomState::Joined) {
-                                        if let Ok(value) = serde_json::from_str(notification.event.json().get()) { let _ = self.observe(&mut next, &notification.room_id, value).await; }
-                                    }}
-                                    let token = page.next_token.filter(|token| Some(token) != from.as_ref() && !reached_anchor);
-                                    if latest { next.latest_cursor = token.clone(); if token.is_none() { self.latest_scan = false; next.latest_anchor = next.latest_head.take().or(next.latest_anchor); } }
-                                    else { next.history_cursor = token.clone(); next.history_exhausted = token.is_none(); }
+                    }
+                }
+            }
+            Work::History(latest, from, result) => {
+                *history = false;
+                self.recovery_budget = self.recovery_budget.saturating_sub(1);
+                match result {
+                    Ok(page) => {
+                        next.coverage.historical_supported = Some(true);
+                        let mut reached_anchor = false;
+                        for notification in page.notifications {
+                            if let Some(event_id) = notification
+                                .event
+                                .get_field::<String>("event_id")
+                                .ok()
+                                .flatten()
+                            {
+                                let identity = format!("{}\n{}", notification.room_id, event_id);
+                                if latest && from.is_none() && next.latest_head.is_none() {
+                                    next.latest_head = Some(identity.clone());
                                 }
-                                Err(error) => {
-                                    if error.server.as_ref().is_some_and(|server| matches!(server.status, 404 | 405 | 501)) { next.coverage.historical_supported = Some(false); self.recovery_budget = 0; }
-                                    if error.server.as_ref().is_some_and(|server| server.status == 400) {
-                                        if latest { next.latest_cursor = None; next.latest_head = None; } else { next.history_cursor = None; }
-                                    }
-                                    Self::gap(&mut next, None, "notification history could not be recovered", from);
-                                    self.recovery_retry = retry_at(&error, 60_000);
+                                if latest && next.latest_anchor.as_ref() == Some(&identity) {
+                                    reached_anchor = true;
+                                    break;
                                 }
                             }
-                            next.coverage.recovering = self.recovery_budget > 0 && (self.latest_scan || !next.history_exhausted) && next.coverage.historical_supported != Some(false);
+                            if self
+                                .client
+                                .get_room(&notification.room_id)
+                                .is_some_and(|room| room.state() == RoomState::Joined)
+                                && let Ok(value) =
+                                    serde_json::from_str(notification.event.json().get())
+                            {
+                                self.observe(next, &notification.room_id, value).await;
+                            }
+                        }
+                        let token = page
+                            .next_token
+                            .filter(|token| Some(token) != from.as_ref() && !reached_anchor);
+                        if latest {
+                            next.latest_cursor = token.clone();
+                            if token.is_none() {
+                                self.latest_scan = false;
+                                next.latest_anchor =
+                                    next.latest_head.take().or(next.latest_anchor.take());
+                            }
+                        } else {
+                            next.history_cursor = token.clone();
+                            next.history_exhausted = token.is_none();
                         }
                     }
-                    if let Err(error) = self.save(next).await { tracing::error!(%error, "saving mention background work failed"); self.latest_scan = recovery.0; self.recovery_budget = recovery.1; self.recovery_retry = now() + 60_000; }
-                }
-                Some(_) = reports.next() => { for candidate in self.data.candidates.values_mut() { candidate.retry_at_ms = 0; } }
-                Some(_) = ignores.next() => {
-                    let mut next = self.data.clone(); self.filter_ignored(&mut next).await;
-                    if let Err(error) = self.save(next).await { tracing::error!(%error, "saving ignored mention removals failed"); }
-                }
-                () = &mut tick => {
-                    tick = Box::pin(matrix_sdk::sleep::sleep(Duration::from_secs(1)));
-                    if now() >= next_scan && !history { self.recovery_budget = 20; self.latest_scan = true; next_scan = now() + 60_000; }
+                    Err(error) => {
+                        if error
+                            .server
+                            .as_ref()
+                            .is_some_and(|server| matches!(server.status, 404 | 405 | 501))
+                        {
+                            next.coverage.historical_supported = Some(false);
+                            self.recovery_budget = 0;
+                        }
+                        if error
+                            .server
+                            .as_ref()
+                            .is_some_and(|server| server.status == 400)
+                        {
+                            if latest {
+                                next.latest_cursor = None;
+                                next.latest_head = None;
+                            } else {
+                                next.history_cursor = None;
+                            }
+                        }
+                        next.gap(None, MentionGapReason::NotificationHistoryUnavailable, from);
+                        self.recovery_retry = retry_at(&error, 60_000);
+                    }
                 }
+                next.coverage.recovering = self.recovery_budget > 0
+                    && (self.latest_scan || !next.history_exhausted)
+                    && next.coverage.historical_supported != Some(false);
             }
         }
+        true
     }
 
     fn schedule(

Modified file: src/mentions/service_tests.rs+264 additions−87 deletions@@ -72,43 +72,88 @@ async fn joined_client(
     (client, own_user)
 }
 
-#[tokio::test]
-async fn failed_batch_keeps_inbox_and_checkpoint_at_the_same_commit_boundary() {
-    let server = MatrixMockServer::new().await;
-    let room_id = room_id!("!mentions:localhost");
-    let (client, own_user) = joined_client(&server, room_id).await;
+fn mention_batch(
+    room_id: &ruma::RoomId,
+    own_user: &ruma::UserId,
+    event: &str,
+    previous_pos: Option<&str>,
+    next_pos: &str,
+) -> ResponseCommitBatch {
     let event = EventFactory::new()
         .room(room_id)
-        .text_msg("commit boundary")
-        .mentions(Mentions::with_user_ids([own_user.clone()]))
-        .event_id(event_id!("$commit-boundary:localhost"))
+        .text_msg(event)
+        .mentions(Mentions::with_user_ids([own_user.to_owned()]))
+        .event_id(&ruma::EventId::parse(format!("${event}:localhost")).unwrap())
         .sender(user_id!("@bob:localhost"))
         .into_raw_sync();
     let raw_room = serde_json::from_value(json!({
-        "initial": true,
         "timeline": [serde_json::to_value(event).unwrap()],
     }))
     .unwrap();
-    let batch = ResponseCommitBatch {
+    ResponseCommitBatch {
         connection_id: "mentions-connection".to_owned(),
-        previous_pos: None,
-        next_pos: "mentions-position-1".to_owned(),
+        previous_pos: previous_pos.map(str::to_owned),
+        next_pos: next_pos.to_owned(),
         response: Default::default(),
         raw_rooms: [(room_id.to_owned(), raw_room)].into(),
-    };
-    let config = config();
-    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
-        .await
-        .unwrap();
+    }
+}
+
+fn quiet_batch(previous_pos: &str, next_pos: &str) -> ResponseCommitBatch {
+    ResponseCommitBatch {
+        connection_id: "mentions-connection".to_owned(),
+        previous_pos: Some(previous_pos.to_owned()),
+        next_pos: next_pos.to_owned(),
+        response: Default::default(),
+        raw_rooms: Default::default(),
+    }
+}
+
+async fn service(client: &Client, store: Option<Result<MentionsStore, Error>>) -> Service {
     let media = Arc::new(MediaService::new(
         client.clone(),
         EngineConfig::default().max_media_bytes,
         EngineConfig::default().max_media_sources,
     ));
-    let (_handle, mut service, updates) =
-        Service::new(client, media, Some(store), EngineConfig::default())
-            .await
-            .unwrap();
+    let (_, mut service, _) = Service::new(client.clone(), media, store, EngineConfig::default())
+        .await
+        .unwrap();
+    service.recovery_budget = 0;
+    service.latest_scan = false;
+    service
+}
+
+fn rows(connection: &Connection) -> BTreeMap<String, Vec<u8>> {
+    let mut statement = connection
+        .prepare("SELECT key, value FROM records WHERE key NOT IN ('!cipher', '!head')")
+        .unwrap();
+    statement
+        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
+        .unwrap()
+        .map(Result::unwrap)
+        .collect()
+}
+
+fn discontinuities(service: &Service) -> usize {
+    service
+        .data
+        .coverage
+        .gaps
+        .iter()
+        .filter(|gap| gap.reason == MentionGapReason::SyncDiscontinuity)
+        .count()
+}
+
+#[tokio::test]
+async fn failed_write_acks_and_keeps_the_inbox_in_memory_until_a_later_write() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!mentions:localhost");
+    let (client, own_user) = joined_client(&server, room_id).await;
+    let config = config();
+    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
+        .await
+        .unwrap();
+    let mut service = service(&client, Some(Ok(store))).await;
 
     let connection = Connection::open(config.mentions_location()).unwrap();
     connection
@@ -118,19 +163,27 @@ async fn failed_batch_keeps_inbox_and_checkpoint_at_the_same_commit_boundary() {
         )
         .unwrap();
 
-    assert!(service.batch(batch.clone()).await.is_err());
-    assert_eq!(service.data.revision, 0);
-    assert!(service.data.entries.is_empty());
-    assert!(service.data.checkpoints.is_empty());
-    assert_eq!(updates.borrow().revision, 0);
-    assert!(updates.borrow().entries.is_empty());
-
-    let reopened = MentionsStore::open(&config, account(&service.client), MAX_BYTES)
-        .await
-        .unwrap();
-    assert_eq!(reopened.revision(), 0);
-    assert!(reopened.read::<Data>("inbox").unwrap().is_none());
-    drop(reopened);
+    service
+        .batch(mention_batch(
+            room_id,
+            &own_user,
+            "first",
+            None,
+            "position-1",
+        ))
+        .await;
+    assert!(service.dirty);
+    assert_eq!(service.data.entries.len(), 1);
+    assert!(
+        service
+            .data
+            .coverage
+            .gaps
+            .iter()
+            .any(|gap| gap.reason == MentionGapReason::StoreFailed)
+    );
+    assert_eq!(service.updates.borrow().entries.len(), 1);
+    assert!(rows(&connection).is_empty());
 
     connection
         .execute_batch(
@@ -138,48 +191,165 @@ async fn failed_batch_keeps_inbox_and_checkpoint_at_the_same_commit_boundary() {
              DROP TRIGGER fail_mentions_inbox_update;",
         )
         .unwrap();
+    service.flush_after = 0;
+    service.batch(quiet_batch("position-1", "position-2")).await;
+    assert!(!service.dirty);
+    drop(service);
+
+    let reopened = MentionsStore::open(&config, account(&client), MAX_BYTES)
+        .await
+        .unwrap();
+    let persisted: Data = reopened.read("inbox").unwrap().unwrap();
+    let sync: SyncState = reopened.read("sync").unwrap().unwrap();
+    assert_eq!(persisted.entries.len(), 1);
+    assert_eq!(sync.inbox_revision, persisted.revision);
+    assert_eq!(sync.checkpoints["mentions-connection"].pos, "position-2");
+    drop(reopened);
     drop(connection);
+    config.delete().await.unwrap();
+}
+
+#[tokio::test]
+async fn a_quiet_response_writes_only_the_checkpoint() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!mentions:localhost");
+    let (client, own_user) = joined_client(&server, room_id).await;
+    let config = config();
+    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
+        .await
+        .unwrap();
+    let mut service = service(&client, Some(Ok(store))).await;
+    service
+        .batch(mention_batch(
+            room_id,
+            &own_user,
+            "first",
+            None,
+            "position-1",
+        ))
+        .await;
+    let connection = Connection::open(config.mentions_location()).unwrap();
+    let before = rows(&connection);
+    assert_eq!(before.len(), 2);
 
-    service.batch(batch.clone()).await.unwrap();
+    service.batch(quiet_batch("position-1", "position-2")).await;
+    let after = rows(&connection);
+    assert_eq!(after.len(), 2);
+    assert_eq!(
+        before
+            .iter()
+            .filter(|(key, value)| after.get(*key) != Some(value))
+            .count(),
+        1
+    );
     assert_eq!(service.data.revision, 1);
-    assert_eq!(service.data.entries.len(), 1);
     assert_eq!(
-        service.data.checkpoints.get("mentions-connection"),
-        Some(&batch.next_pos)
+        service.sync.checkpoints["mentions-connection"].pos,
+        "position-2"
     );
-    assert_eq!(updates.borrow().revision, 1);
-    assert_eq!(updates.borrow().entries.len(), 1);
+    drop(service);
+    drop(connection);
+    config.delete().await.unwrap();
+}
+
+#[tokio::test]
+async fn a_replay_from_the_previous_checkpoint_is_not_a_discontinuity() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!mentions:localhost");
+    let (client, own_user) = joined_client(&server, room_id).await;
+    let config = config();
+    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
+        .await
+        .unwrap();
+    let mut first = service(&client, Some(Ok(store))).await;
+    first
+        .batch(mention_batch(room_id, &own_user, "one", None, "position-1"))
+        .await;
+    first
+        .batch(mention_batch(
+            room_id,
+            &own_user,
+            "two",
+            Some("position-1"),
+            "position-2",
+        ))
+        .await;
+    drop(first);
 
-    let reopened = MentionsStore::open(&config, account(&service.client), MAX_BYTES)
+    // the sdk never saved position-2, so it asks again from position-1
+    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
         .await
         .unwrap();
-    let persisted: Data = reopened.read("inbox").unwrap().unwrap();
-    assert_eq!(reopened.revision(), 1);
-    assert_eq!(persisted.revision, 1);
-    assert_eq!(persisted.entries.len(), 1);
+    let mut restarted = service(&client, Some(Ok(store))).await;
+    restarted
+        .batch(mention_batch(
+            room_id,
+            &own_user,
+            "three",
+            Some("position-1"),
+            "position-3",
+        ))
+        .await;
+    assert_eq!(discontinuities(&restarted), 0);
+    assert_eq!(restarted.data.entries.len(), 3);
     assert_eq!(
-        persisted.checkpoints.get("mentions-connection"),
-        Some(&batch.next_pos)
+        restarted.sync.checkpoints["mentions-connection"],
+        Checkpoint {
+            pos: "position-3".to_owned(),
+            previous: Some("position-1".to_owned()),
+        }
     );
-    drop(reopened);
 
-    service.batch(batch).await.unwrap();
-    assert_eq!(service.data.revision, 1);
-    assert_eq!(service.store.as_ref().unwrap().revision(), 1);
-
-    drop(service);
+    restarted
+        .batch(quiet_batch("position-3", "position-4"))
+        .await;
+    assert_eq!(discontinuities(&restarted), 0);
+    restarted
+        .batch(quiet_batch("position-1", "position-5"))
+        .await;
+    assert_eq!(discontinuities(&restarted), 1);
+    drop(restarted);
     config.delete().await.unwrap();
 }
 
-#[test]
-fn repair_fields_default_when_loading_an_older_inbox() {
-    let mut inbox = serde_json::to_value(Data::default()).unwrap();
-    inbox.as_object_mut().unwrap().remove("repair_policy");
-    inbox.as_object_mut().unwrap().remove("repair_coverage");
+#[tokio::test]
+async fn an_unreadable_store_resets_behind_a_gap() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!mentions:localhost");
+    let (client, own_user) = joined_client(&server, room_id).await;
+    let config = config();
+    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
+        .await
+        .unwrap();
+    let mut first = service(&client, Some(Ok(store))).await;
+    first
+        .batch(mention_batch(room_id, &own_user, "one", None, "position-1"))
+        .await;
+    drop(first);
+    let connection = Connection::open(config.mentions_location()).unwrap();
+    connection
+        .execute(
+            "UPDATE records SET value = x'00' WHERE key NOT IN ('!cipher', '!head')",
+            [],
+        )
+        .unwrap();
+    drop(connection);
 
-    let restored: Data = serde_json::from_value(inbox).unwrap();
-    assert_eq!(restored.repair_policy, HistoryRepairPolicy::default());
-    assert!(restored.repair_coverage.is_empty());
+    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
+        .await
+        .unwrap();
+    let restarted = service(&client, Some(Ok(store))).await;
+    assert!(restarted.data.entries.is_empty());
+    assert!(
+        restarted
+            .data
+            .coverage
+            .gaps
+            .iter()
+            .any(|gap| gap.reason == MentionGapReason::StoreReset)
+    );
+    drop(restarted);
+    config.delete().await.unwrap();
 }
 
 #[tokio::test]
@@ -196,10 +366,14 @@ async fn repair_policy_and_coverage_roundtrip_through_the_inbox() {
         EngineConfig::default().max_media_bytes,
         EngineConfig::default().max_media_sources,
     ));
-    let (handle, mut service, _) =
-        Service::new(client.clone(), media, Some(store), EngineConfig::default())
-            .await
-            .unwrap();
+    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());
@@ -226,7 +400,8 @@ async fn repair_policy_and_coverage_roundtrip_through_the_inbox() {
         .unwrap();
     let persisted: Data = reopened.read("inbox").unwrap().unwrap();
     assert_eq!(persisted.repair_policy, stored_policy);
-    let coverage = persisted.repair_coverage.get(room_id).unwrap();
+    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);
@@ -248,13 +423,16 @@ async fn repair_limited_sync_freezes_the_last_room_checkpoint() {
     let (_handle, mut service, _) = Service::new(client, media, None, EngineConfig::default())
         .await
         .unwrap();
-    service.data.checkpoints.insert(
+    service.sync.checkpoints.insert(
         "mentions-connection".to_owned(),
-        "position-before".to_owned(),
+        Checkpoint {
+            pos: "position-before".to_owned(),
+            previous: None,
+        },
     );
-    service.data.observed_rooms.insert(room_id.to_owned());
+    service.sync.observed_rooms.insert(room_id.to_owned());
     service
-        .data
+        .sync
         .repair_coverage
         .entry(room_id.to_owned())
         .or_default()
@@ -292,9 +470,9 @@ async fn repair_limited_sync_freezes_the_last_room_checkpoint() {
         raw_rooms: [(room_id.to_owned(), raw_room)].into(),
     };
 
-    service.batch(batch).await.unwrap();
+    service.batch(batch).await;
 
-    let coverage = service.data.repair_coverage.get(room_id).unwrap();
+    let coverage = service.sync.repair_coverage.get(room_id).unwrap();
     assert_eq!(coverage.last_confirmed_ms, Some(1_000));
     assert!(
         coverage
@@ -372,10 +550,14 @@ async fn service_recovers_notification_history_into_durable_unread_inbox() {
         EngineConfig::default().max_media_bytes,
         EngineConfig::default().max_media_sources,
     ));
-    let (handle, service, mut updates) =
-        Service::new(client.clone(), media, Some(store), EngineConfig::default())
-            .await
-            .unwrap();
+    let (handle, service, mut updates) = Service::new(
+        client.clone(),
+        media,
+        Some(Ok(store)),
+        EngineConfig::default(),
+    )
+    .await
+    .unwrap();
     let service_task = tokio::spawn(service.run());
 
     let discovered = tokio::time::timeout(Duration::from_secs(10), async {
@@ -398,7 +580,8 @@ async fn service_recovers_notification_history_into_durable_unread_inbox() {
         .await
         .unwrap();
     let persisted: Data = reopened.read("inbox").unwrap().unwrap();
-    assert_eq!(reopened.revision(), persisted.revision);
+    let sync: SyncState = reopened.read("sync").unwrap().unwrap();
+    assert_eq!(sync.inbox_revision, persisted.revision);
     assert_eq!(persisted.entries.len(), 2);
     assert!(persisted.entries.values().all(|entry| !entry.read));
     assert!(persisted.latest_cursor.is_none());
@@ -471,11 +654,8 @@ async fn live_context_survives_capture_then_edit_and_redaction() {
     let following_id = format!("{room_id}\n$live-following:localhost");
     let mut data = Data::default();
 
-    service.observe(&mut data, &room_id, mention).await.unwrap();
-    service
-        .observe(&mut data, &room_id, following)
-        .await
-        .unwrap();
+    service.observe(&mut data, &room_id, mention).await;
+    service.observe(&mut data, &room_id, following).await;
 
     let mention_identity = id(&mention_key);
     let entry = data.entries.get(&mention_identity).unwrap();
@@ -521,7 +701,7 @@ async fn live_context_survives_capture_then_edit_and_redaction() {
         "sender": "@bob:localhost",
         "type": "m.room.message"
     });
-    service.observe(&mut data, &room_id, edit).await.unwrap();
+    service.observe(&mut data, &room_id, edit).await;
     let (saved, _) = data.context(&mention_key).unwrap();
     let context = saved.context.unwrap();
     assert_eq!(
@@ -542,10 +722,7 @@ async fn live_context_survives_capture_then_edit_and_redaction() {
         "sender": "@bob:localhost",
         "type": "m.room.redaction"
     });
-    service
-        .observe(&mut data, &room_id, redaction)
-        .await
-        .unwrap();
+    service.observe(&mut data, &room_id, redaction).await;
 
     let saved_following = data.events.get(&following_id).unwrap();
     assert!(matches!(

Modified file: src/mentions/state.rs+166 additions−34 deletions@@ -4,11 +4,13 @@ use ruma::{OwnedEventId, OwnedUserId};
 use serde_json::Value;
 
 use crate::{
-    Error, ErrorKind, EventContext, HistoryRepairPolicy, MentionCaptureState, MentionCoverage,
-    MentionEntry, MentionKey, MentionsBoundary, MentionsPage, SavedMentionContext, TimelineItem,
-    normalise::MediaRegistration,
+    Error, ErrorKind, EventContext, MentionCaptureState, MentionCoverage, MentionEntry, MentionGap,
+    MentionGapReason, MentionKey, MentionsBoundary, MentionsPage, SavedMentionContext,
+    TimelineItem, normalise::MediaRegistration,
 };
 
+const CANDIDATE_LIFETIME_MS: u64 = 7 * 24 * 60 * 60 * 1000;
+
 #[derive(Clone)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub(super) struct SavedEvent {
@@ -17,16 +19,45 @@ pub(super) struct SavedEvent {
     pub raw: Option<Value>,
 }
 
-#[derive(Clone)]
+impl PartialEq for SavedEvent {
+    fn eq(&self, other: &Self) -> bool {
+        self.item == other.item
+            && self.raw == other.raw
+            && self
+                .media
+                .iter()
+                .map(|media| &media.id)
+                .eq(other.media.iter().map(|media| &media.id))
+    }
+}
+
+#[derive(Clone, PartialEq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub(super) struct Candidate {
     pub room_id: ruma::OwnedRoomId,
     pub raw: Value,
     pub attempts: u32,
     pub retry_at_ms: u64,
+    pub first_seen_ms: u64,
 }
 
-#[derive(Clone)]
+impl Candidate {
+    pub fn new(room_id: ruma::OwnedRoomId, raw: Value, retry_at_ms: u64) -> Self {
+        Self {
+            room_id,
+            raw,
+            attempts: 0,
+            retry_at_ms,
+            first_seen_ms: now(),
+        }
+    }
+
+    pub fn encrypted(&self) -> bool {
+        self.raw["type"] == "m.room.encrypted"
+    }
+}
+
+#[derive(Clone, PartialEq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub(super) struct Window {
     pub before: Vec<String>,
@@ -36,7 +67,7 @@ pub(super) struct Window {
     pub end: Option<String>,
 }
 
-#[derive(Clone, Debug, Default)]
+#[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>,
@@ -55,7 +86,7 @@ impl RepairCoverage {
     }
 }
 
-#[derive(Clone, Default)]
+#[derive(Clone, Default, PartialEq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub(super) struct Data {
     pub revision: u64,
@@ -67,19 +98,28 @@ pub(super) struct Data {
     pub redacted: BTreeMap<String, u64>,
     pub edits: BTreeMap<String, Value>,
     pub coverage: MentionCoverage,
-    pub checkpoints: BTreeMap<String, String>,
-    pub observed_rooms: BTreeSet<ruma::OwnedRoomId>,
-    #[cfg_attr(feature = "serde", serde(default))]
-    pub repair_policy: HistoryRepairPolicy,
-    #[cfg_attr(feature = "serde", serde(default))]
-    pub repair_coverage: BTreeMap<ruma::OwnedRoomId, RepairCoverage>,
     pub history_cursor: Option<String>,
     pub history_exhausted: bool,
     pub latest_cursor: Option<String>,
-    #[cfg_attr(feature = "serde", serde(default))]
     pub latest_anchor: Option<String>,
-    #[cfg_attr(feature = "serde", serde(default))]
     pub latest_head: Option<String>,
+    pub repair_policy: crate::HistoryRepairPolicy,
+}
+
+#[derive(Clone, Default, PartialEq)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+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)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+pub(super) struct Checkpoint {
+    pub pos: String,
+    pub previous: Option<String>,
 }
 
 pub(super) fn id(key: &MentionKey) -> String {
@@ -96,7 +136,78 @@ pub(super) fn invalid(message: &'static str) -> Error {
     Error::new(ErrorKind::Invalid, message)
 }
 
+#[cfg(feature = "serde")]
+pub(super) fn json_len(value: &impl serde::Serialize) -> usize {
+    struct Count(usize);
+    impl std::io::Write for Count {
+        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+            self.0 += bytes.len();
+            Ok(bytes.len())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+    let mut count = Count(0);
+    let _ = serde_json::to_writer(&mut count, value);
+    count.0
+}
+
+#[cfg(not(feature = "serde"))]
+pub(super) fn json_len<T>(_: &T) -> usize {
+    0
+}
+
+fn timestamp(value: &Value) -> u64 {
+    value["origin_server_ts"].as_u64().unwrap_or(0)
+}
+
 impl Data {
+    pub fn gap(
+        &mut self,
+        room_id: Option<ruma::OwnedRoomId>,
+        reason: MentionGapReason,
+        token: Option<String>,
+    ) {
+        if self
+            .coverage
+            .gaps
+            .iter()
+            .any(|gap| gap.room_id == room_id && gap.reason == reason && gap.token == token)
+        {
+            return;
+        }
+        self.coverage.gaps.push(MentionGap {
+            room_id,
+            reason,
+            token,
+        });
+        if self.coverage.gaps.len() > 128 {
+            self.coverage.gaps.remove(0);
+            self.coverage.truncated = true;
+        }
+    }
+
+    pub fn expire_candidates(&mut self, now_ms: u64) -> bool {
+        let expired: Vec<_> = self
+            .candidates
+            .iter()
+            .filter(|(_, candidate)| {
+                now_ms.saturating_sub(candidate.first_seen_ms) > CANDIDATE_LIFETIME_MS
+            })
+            .map(|(key, candidate)| (key.clone(), candidate.room_id.clone()))
+            .collect();
+        for (key, room_id) in &expired {
+            self.candidates.remove(key);
+            self.gap(
+                Some(room_id.clone()),
+                MentionGapReason::CandidateExpired,
+                None,
+            );
+        }
+        !expired.is_empty()
+    }
+
     pub fn touch_contexts(&mut self, identity: &str) {
         for (key, window) in &self.windows {
             if window.target.as_deref() == Some(identity)
@@ -213,23 +324,32 @@ impl Data {
         ))
     }
 
-    pub fn collect_events(&mut self) {
+    pub fn collect_events(&mut self) -> usize {
         let mut used: BTreeSet<String> = self.entries.keys().cloned().collect();
         for window in self.windows.values() {
             used.extend(window.before.iter().chain(&window.after).cloned());
             used.extend(window.target.iter().cloned());
         }
-        self.events.retain(|key, _| used.contains(key));
+        let mut freed = 0;
+        self.events.retain(|key, event| {
+            let keep = used.contains(key);
+            if !keep {
+                freed += key.len() + json_len(event);
+            }
+            keep
+        });
         self.capture_attempts
             .retain(|key, _| self.entries.contains_key(key));
+        freed
     }
 
-    pub fn evict_one(&mut self) -> bool {
+    /// returns roughly how many plaintext JSON bytes the eviction freed
+    pub fn evict_one(&mut self) -> Option<usize> {
         let oldest = |keys: Vec<String>, entries: &BTreeMap<String, MentionEntry>| {
             keys.into_iter()
                 .min_by_key(|key| entries.get(key).map_or(0, |entry| entry.timestamp_ms))
         };
-        if let Some(key) = oldest(
+        let freed = if let Some(key) = oldest(
             self.windows
                 .keys()
                 .filter(|key| self.entries.get(*key).is_some_and(|entry| entry.read))
@@ -239,30 +359,35 @@ impl Data {
         )
         .or_else(|| oldest(self.windows.keys().cloned().collect(), &self.entries))
         {
-            self.windows.remove(&key);
+            let window = self.windows.remove(&key);
             if let Some(entry) = self.entries.get_mut(&key) {
                 entry.capture = MentionCaptureState::Unavailable;
                 entry.revision = self.revision;
             }
+            key.len() + json_len(&window)
         } else if let Some(key) = self
             .entries
             .iter()
             .min_by_key(|(_, entry)| (!entry.read, entry.timestamp_ms))
             .map(|(key, _)| key.clone())
         {
-            self.entries.remove(&key);
-        } else if let Some(key) = self.candidates.keys().next().cloned() {
-            self.candidates.remove(&key);
+            key.len() + json_len(&self.entries.remove(&key))
+        } else if let Some(key) = self
+            .candidates
+            .iter()
+            .min_by_key(|(_, candidate)| timestamp(&candidate.raw))
+            .map(|(key, _)| key.clone())
+        {
+            key.len() + json_len(&self.candidates.remove(&key))
         } else if let Some(key) = self.edits.keys().next().cloned() {
-            self.edits.remove(&key);
-        } else if let Some(key) = self.redacted.keys().next().cloned() {
-            self.redacted.remove(&key);
+            key.len() + json_len(&self.edits.remove(&key))
         } else {
-            return false;
-        }
+            let key = self.redacted.keys().next().cloned()?;
+            self.redacted.remove(&key);
+            key.len() + 24
+        };
         self.coverage.truncated = true;
-        self.collect_events();
-        true
+        Some(freed + self.collect_events())
     }
 
     pub fn bound(&mut self, entries: usize, events: usize, candidates: usize) {
@@ -279,10 +404,17 @@ impl Data {
             }
         }
         self.collect_events();
-        while self.events.len() > events && self.evict_one() {}
-        while self.candidates.len() > candidates {
-            let key = self.candidates.keys().next().cloned().unwrap();
-            self.candidates.remove(&key);
+        while self.events.len() > events && self.evict_one().is_some() {}
+        if self.candidates.len() > candidates {
+            let mut oldest: Vec<_> = self
+                .candidates
+                .iter()
+                .map(|(key, candidate)| (timestamp(&candidate.raw), key.clone()))
+                .collect();
+            oldest.sort();
+            for (_, key) in oldest.into_iter().take(self.candidates.len() - candidates) {
+                self.candidates.remove(&key);
+            }
             self.coverage.truncated = true;
         }
         for map in [&mut self.redacted] {

Modified file: src/mentions/state_tests.rs+3 additions−22 deletions@@ -196,7 +196,7 @@ async fn retention_discards_read_context_first_and_keeps_shared_events_until_unu
         },
     );
 
-    assert!(data.evict_one());
+    assert!(data.evict_one().is_some());
     assert!(!data.windows.contains_key(&read_id));
     assert!(data.windows.contains_key(&unread_id));
     assert_eq!(
@@ -206,7 +206,7 @@ async fn retention_discards_read_context_first_and_keeps_shared_events_until_unu
     assert!(!data.entries[&unread_id].read);
     assert!(data.events.contains_key(&shared_id));
 
-    assert!(data.evict_one());
+    assert!(data.evict_one().is_some());
     assert!(data.windows.is_empty());
     assert!(data.entries.contains_key(&read_id));
     assert!(data.entries.contains_key(&unread_id));
@@ -365,8 +365,6 @@ mod persistence {
                 end: Some("end-token".to_owned()),
             },
         );
-        data.checkpoints
-            .insert("sync-connection".to_owned(), "position-1".to_owned());
         data.history_cursor = Some("history-position".to_owned());
         data
     }
@@ -379,7 +377,7 @@ mod persistence {
 
     #[cfg_attr(not(target_family = "wasm"), tokio::test)]
     #[cfg_attr(target_family = "wasm", wasm_bindgen_test::wasm_bindgen_test)]
-    async fn encrypted_store_roundtrips_inbox_context_read_state_and_checkpoints() {
+    async fn encrypted_store_roundtrips_inbox_context_and_read_state() {
         let config = config();
         let data = saved_data();
         let mut store = open(&config).await;
@@ -402,13 +400,6 @@ mod persistence {
         let mut restored_data: Data = restored.read("inbox").unwrap().unwrap();
         assert_eq!(restored.revision(), restored_data.revision);
         assert!(restored_data.entries.values().next().unwrap().read);
-        assert_eq!(
-            restored_data
-                .checkpoints
-                .get("sync-connection")
-                .map(String::as_str),
-            Some("position-1")
-        );
         let mention = restored_data.entries.values().next().unwrap().key.clone();
         let (context, media) = restored_data.context(&mention).unwrap();
         assert_eq!(
@@ -425,9 +416,6 @@ mod persistence {
         let updated_entry = restored_data.entries.values_mut().next().unwrap();
         updated_entry.read = false;
         updated_entry.revision = 2;
-        restored_data
-            .checkpoints
-            .insert("sync-connection".to_owned(), "position-2".to_owned());
         restored
             .commit(
                 1,
@@ -445,13 +433,6 @@ mod persistence {
         assert_eq!(restored.revision(), 2);
         assert_eq!(restored_data.revision, 2);
         assert!(!restored_data.entries.values().next().unwrap().read);
-        assert_eq!(
-            restored_data
-                .checkpoints
-                .get("sync-connection")
-                .map(String::as_str),
-            Some("position-2")
-        );
         drop(restored);
         config.delete().await.unwrap();
     }

Modified file: src/mentions/store/indexeddb.rs+19 additions@@ -62,6 +62,15 @@ impl Backend {
         }
     }
 
+    pub(super) async fn reset(&self) -> Result<(), crate::Error> {
+        let database = open_database(&self.name)
+            .await
+            .map_err(|error| super::storage_error("opening the mentions database", error))?;
+        clear(&database)
+            .await
+            .map_err(|error| super::storage_error("clearing the mentions database", error))
+    }
+
     pub(super) async fn commit(
         &self,
         expected: Option<Vec<u8>>,
@@ -100,6 +109,16 @@ async fn open_database(name: &str) -> Result<DatabaseHandle, OpenDbError> {
         .map(DatabaseHandle)
 }
 
+async fn clear(database: &Database) -> Result<(), IndexedDbError> {
+    let transaction = database
+        .transaction(RECORDS)
+        .with_mode(TransactionMode::Readwrite)
+        .build()?;
+    transaction.object_store(RECORDS)?.clear()?.await?;
+    transaction.commit().await?;
+    Ok(())
+}
+
 enum ReadError {
     Invalid(&'static str),
     Storage(IndexedDbError),

Modified file: src/mentions/store/mod.rs+148 additions−68 deletions@@ -17,7 +17,7 @@ mod indexeddb;
 #[cfg(feature = "indexeddb")]
 use indexeddb::Backend;
 
-const SCHEMA: u32 = 1;
+const SCHEMA: u32 = 2;
 const MAX_RECORDS: usize = 35_128;
 
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -61,6 +61,7 @@ pub(crate) struct MentionsStore {
     snapshot: Snapshot,
     revision: u64,
     max_bytes: usize,
+    reset: bool,
 }
 
 impl fmt::Debug for MentionsStore {
@@ -74,71 +75,30 @@ impl fmt::Debug for MentionsStore {
 }
 
 impl MentionsStore {
+    /// an unreadable, corrupt or foreign store is emptied rather than refused
     pub(crate) async fn open(
         config: &StoreConfig,
         account: Account,
         max_bytes: usize,
     ) -> Result<Self, Error> {
         let backend = Backend::open(config.mentions_location()).await?;
-        let (cipher, snapshot, revision) = match backend.load().await? {
-            Some(snapshot) => {
-                validate_size(&snapshot, max_bytes)?;
-                let cipher = StoreCipher::import(config.passphrase(), &snapshot.exported)
-                    .map_err(|error| storage_error("opening the mentions cipher", error))?;
-                let head: Head = cipher
-                    .decrypt_value(&snapshot.head)
-                    .map_err(|error| storage_error("opening the mentions header", error))?;
-                validate_account(head.schema, &head.account, &account)?;
-                if head.records != snapshot.rows.len() || head.bytes != row_bytes(&snapshot.rows)? {
-                    return Err(invalid(
-                        "mentions records do not match their committed header",
-                    ));
-                }
-                for (key, encrypted) in &snapshot.rows {
-                    let record: Record = cipher
-                        .decrypt_value(encrypted)
-                        .map_err(|error| storage_error("opening a mentions record", error))?;
-                    validate_account(record.schema, &record.account, &account)?;
-                    if storage_key(&cipher, &record.key) != *key {
-                        return Err(invalid("a mentions record belongs to a different key"));
-                    }
-                }
-                let revision = head.revision;
-                (cipher, snapshot, revision)
-            }
-            None => {
-                let cipher = StoreCipher::new()
-                    .map_err(|error| storage_error("creating the mentions cipher", error))?;
-                let exported = cipher
-                    .export(config.passphrase())
-                    .map_err(|error| storage_error("sealing the mentions cipher", error))?;
-                let head = cipher
-                    .encrypt_value(&Head {
-                        schema: SCHEMA,
-                        account: account.clone(),
-                        revision: 0,
-                        records: 0,
-                        bytes: 0,
-                    })
-                    .map_err(|error| storage_error("sealing the mentions header", error))?;
-                let snapshot = Snapshot {
-                    exported,
-                    head,
-                    rows: BTreeMap::new(),
-                };
-                validate_size(&snapshot, max_bytes)?;
-                backend
-                    .commit(
-                        None,
-                        snapshot.exported.clone(),
-                        snapshot.head.clone(),
-                        BTreeMap::new(),
-                    )
-                    .await?;
-                (cipher, snapshot, 0)
+        let loaded = match backend.load().await {
+            Ok(Some(snapshot)) => restore(config, &account, snapshot, max_bytes).map(Some),
+            Ok(None) => Ok(None),
+            Err(error) => Err(error),
+        };
+        let (reset, restored) = match loaded {
+            Ok(restored) => (false, restored),
+            Err(error) => {
+                tracing::warn!(%error, "resetting an unreadable mentions store");
+                backend.reset().await?;
+                (true, None)
             }
         };
-        validate_size(&snapshot, max_bytes)?;
+        let (cipher, snapshot, revision) = match restored {
+            Some(restored) => restored,
+            None => create(&backend, config, &account, max_bytes).await?,
+        };
         Ok(Self {
             backend,
             cipher,
@@ -146,13 +106,65 @@ impl MentionsStore {
             snapshot,
             revision,
             max_bytes,
+            reset,
         })
     }
 
+    pub(crate) async fn reset(&mut self) -> Result<(), Error> {
+        self.backend.reset().await?;
+        let head = seal_head(&self.cipher, &self.account, 0, 0, 0)?;
+        self.backend
+            .commit(
+                None,
+                self.snapshot.exported.clone(),
+                head.clone(),
+                BTreeMap::new(),
+            )
+            .await?;
+        self.snapshot.head = head;
+        self.snapshot.rows.clear();
+        self.revision = 0;
+        self.reset = true;
+        Ok(())
+    }
+
+    pub(crate) fn was_reset(&self) -> bool {
+        self.reset
+    }
+
     pub(crate) fn revision(&self) -> u64 {
         self.revision
     }
 
+    pub(crate) fn available(&self, keys: &[&str]) -> usize {
+        let replaced: Vec<_> = keys
+            .iter()
+            .map(|key| storage_key(&self.cipher, key))
+            .collect();
+        let others = self
+            .snapshot
+            .rows
+            .iter()
+            .filter(|(key, _)| !replaced.contains(key))
+            .map(|(key, value)| key.len() + value.len())
+            .sum::<usize>();
+        let keys = replaced.iter().map(String::len).sum::<usize>();
+        self.max_bytes
+            .saturating_sub(others + keys + self.snapshot.exported.len() + self.snapshot.head.len())
+    }
+
+    /// an upper bound on the sealed size of a record whose value encodes to `value_len`
+    /// bytes, `StoreCipher` writes its ciphertext as a JSON array of decimal bytes, at
+    /// most four characters each
+    pub(crate) fn sealed_bound(&self, key: &str, value_len: usize) -> usize {
+        let record = value_len
+            + key.len()
+            + self.account.homeserver.len()
+            + self.account.user_id.as_str().len()
+            + 128;
+        (record + 16 + 24) * 4 + 64
+    }
+
     pub(crate) fn read<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, Error> {
         let Some(encrypted) = self.snapshot.rows.get(&storage_key(&self.cipher, key)) else {
             return Ok(None);
@@ -222,16 +234,7 @@ impl MentionsStore {
                 records += 1;
             }
         }
-        let head = self
-            .cipher
-            .encrypt_value(&Head {
-                schema: SCHEMA,
-                account: self.account.clone(),
-                revision,
-                records,
-                bytes,
-            })
-            .map_err(|error| storage_error("sealing the mentions header", error))?;
+        let head = seal_head(&self.cipher, &self.account, revision, records, bytes)?;
         validate_budget(
             bytes,
             records,
@@ -263,6 +266,83 @@ impl MentionsStore {
     }
 }
 
+fn restore(
+    config: &StoreConfig,
+    account: &Account,
+    snapshot: Snapshot,
+    max_bytes: usize,
+) -> Result<(StoreCipher, Snapshot, u64), Error> {
+    validate_size(&snapshot, max_bytes)?;
+    let cipher = StoreCipher::import(config.passphrase(), &snapshot.exported)
+        .map_err(|error| storage_error("opening the mentions cipher", error))?;
+    let head: Head = cipher
+        .decrypt_value(&snapshot.head)
+        .map_err(|error| storage_error("opening the mentions header", error))?;
+    validate_account(head.schema, &head.account, account)?;
+    if head.records != snapshot.rows.len() || head.bytes != row_bytes(&snapshot.rows)? {
+        return Err(invalid(
+            "mentions records do not match their committed header",
+        ));
+    }
+    for (key, encrypted) in &snapshot.rows {
+        let record: Record = cipher
+            .decrypt_value(encrypted)
+            .map_err(|error| storage_error("opening a mentions record", error))?;
+        validate_account(record.schema, &record.account, account)?;
+        if storage_key(&cipher, &record.key) != *key {
+            return Err(invalid("a mentions record belongs to a different key"));
+        }
+    }
+    Ok((cipher, snapshot, head.revision))
+}
+
+async fn create(
+    backend: &Backend,
+    config: &StoreConfig,
+    account: &Account,
+    max_bytes: usize,
+) -> Result<(StoreCipher, Snapshot, u64), Error> {
+    let cipher =
+        StoreCipher::new().map_err(|error| storage_error("creating the mentions cipher", error))?;
+    let exported = cipher
+        .export(config.passphrase())
+        .map_err(|error| storage_error("sealing the mentions cipher", error))?;
+    let head = seal_head(&cipher, account, 0, 0, 0)?;
+    let snapshot = Snapshot {
+        exported,
+        head,
+        rows: BTreeMap::new(),
+    };
+    validate_size(&snapshot, max_bytes)?;
+    backend
+        .commit(
+            None,
+            snapshot.exported.clone(),
+            snapshot.head.clone(),
+            BTreeMap::new(),
+        )
+        .await?;
+    Ok((cipher, snapshot, 0))
+}
+
+fn seal_head(
+    cipher: &StoreCipher,
+    account: &Account,
+    revision: u64,
+    records: usize,
+    bytes: usize,
+) -> Result<Vec<u8>, Error> {
+    cipher
+        .encrypt_value(&Head {
+            schema: SCHEMA,
+            account: account.clone(),
+            revision,
+            records,
+            bytes,
+        })
+        .map_err(|error| storage_error("sealing the mentions header", error))
+}
+
 fn validate_account(schema: u32, actual: &Account, expected: &Account) -> Result<(), Error> {
     if schema != SCHEMA {
         return Err(invalid("unsupported mentions schema"));

Modified file: src/mentions/store/sqlite.rs+26 additions−3 deletions@@ -21,7 +21,7 @@ pub(super) struct Backend {
 impl Backend {
     pub(super) async fn open(path: PathBuf) -> Result<Self, Error> {
         let setup_path = path.clone();
-        tokio::task::spawn_blocking(move || initialize(&setup_path))
+        tokio::task::spawn_blocking(move || initialise(&setup_path))
             .await
             .map_err(|error| storage_error("opening the mentions store", error))??;
         Ok(Self { path })
@@ -34,6 +34,23 @@ impl Backend {
             .map_err(|error| storage_error("loading the mentions store", error))?
     }
 
+    pub(super) async fn reset(&self) -> Result<(), Error> {
+        let path = self.path.clone();
+        tokio::task::spawn_blocking(move || {
+            for file in [path.clone(), journal(&path)] {
+                match fs::remove_file(&file) {
+                    Err(error) if error.kind() != io::ErrorKind::NotFound => {
+                        return Err(storage_error("removing the mentions database", error));
+                    }
+                    _ => {}
+                }
+            }
+            initialise(&path)
+        })
+        .await
+        .map_err(|error| storage_error("resetting the mentions store", error))?
+    }
+
     pub(super) async fn commit(
         &self,
         expected: Option<Vec<u8>>,
@@ -50,7 +67,7 @@ impl Backend {
     }
 }
 
-fn initialize(path: &Path) -> Result<(), Error> {
+fn initialise(path: &Path) -> Result<(), Error> {
     let parent = path
         .parent()
         .filter(|parent| !parent.as_os_str().is_empty())
@@ -65,7 +82,7 @@ fn initialize(path: &Path) -> Result<(), Error> {
             "CREATE TABLE IF NOT EXISTS records (key TEXT PRIMARY KEY, value BLOB NOT NULL)",
             [],
         )
-        .map_err(|error| storage_error("initializing the mentions database", error))?;
+        .map_err(|error| storage_error("initialising the mentions database", error))?;
     Ok(())
 }
 
@@ -212,3 +229,9 @@ fn create_private_file(path: &Path) -> io::Result<()> {
     }
     Ok(())
 }
+
+fn journal(path: &Path) -> PathBuf {
+    let mut journal = path.as_os_str().to_owned();
+    journal.push("-journal");
+    PathBuf::from(journal)
+}

Modified file: src/mentions/store/tests.rs+34 additions−20 deletions@@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use ruma::user_id;
 use serde_json::{Value, json};
 
-use super::{Account, Head, MentionsStore, Write, row_bytes, storage_key};
+use super::{Account, Head, MentionsStore, SCHEMA, Write, row_bytes, storage_key};
 use crate::StoreConfig;
 
 #[cfg(target_family = "wasm")]
@@ -244,7 +244,7 @@ async fn records_hide_content_keys_and_account_metadata() {
 
 #[cfg_attr(not(target_family = "wasm"), tokio::test)]
 #[cfg_attr(target_family = "wasm", wasm_bindgen_test::wasm_bindgen_test)]
-async fn wrong_account_server_and_passphrase_cannot_open_the_store() {
+async fn wrong_account_server_and_passphrase_reset_the_store() {
     let config = config();
     let store = open(&config).await;
     drop(store);
@@ -255,7 +255,8 @@ async fn wrong_account_server_and_passphrase_cannot_open_the_store() {
     assert!(
         MentionsStore::open(&config, other, MAX_BYTES)
             .await
-            .is_err()
+            .unwrap()
+            .was_reset()
     );
     let other_server = Account {
         homeserver: "https://different.example/".to_owned(),
@@ -264,7 +265,8 @@ async fn wrong_account_server_and_passphrase_cannot_open_the_store() {
     assert!(
         MentionsStore::open(&config, other_server, MAX_BYTES)
             .await
-            .is_err()
+            .unwrap()
+            .was_reset()
     );
     #[cfg(feature = "sqlite")]
     let wrong_passphrase =
@@ -274,7 +276,8 @@ async fn wrong_account_server_and_passphrase_cannot_open_the_store() {
     assert!(
         MentionsStore::open(&wrong_passphrase, account(), MAX_BYTES)
             .await
-            .is_err()
+            .unwrap()
+            .was_reset()
     );
     drop(open(&config).await);
     config.delete().await.unwrap();
@@ -282,16 +285,20 @@ async fn wrong_account_server_and_passphrase_cannot_open_the_store() {
 
 #[cfg_attr(not(target_family = "wasm"), tokio::test)]
 #[cfg_attr(target_family = "wasm", wasm_bindgen_test::wasm_bindgen_test)]
-async fn ciphertext_corruption_and_record_swaps_are_rejected() {
+async fn ciphertext_corruption_and_record_swaps_reset_the_store() {
     let config = config();
-    let mut store = open(&config).await;
-    store
-        .commit(
-            0,
-            vec![set("one", json!("first")), set("two", json!("other"))],
-        )
-        .await
-        .unwrap();
+    let seeded = async || {
+        let mut store = open(&config).await;
+        store
+            .commit(
+                0,
+                vec![set("one", json!("first")), set("two", json!("other"))],
+            )
+            .await
+            .unwrap();
+        store
+    };
+    let store = seeded().await;
     let one = storage_key(&store.cipher, "one");
     let two = storage_key(&store.cipher, "two");
     let mut swapped = store.snapshot.rows.clone();
@@ -315,8 +322,13 @@ async fn ciphertext_corruption_and_record_swaps_are_rejected() {
     assert!(
         MentionsStore::open(&config, account(), MAX_BYTES)
             .await
-            .is_err()
+            .unwrap()
+            .was_reset()
     );
+    drop(store);
+    let store = seeded().await;
+    assert!(!store.was_reset());
+    let one = storage_key(&store.cipher, "one");
     let mut corrupt = store.snapshot.rows[&one].clone();
     let middle = corrupt.len() / 2;
     corrupt[middle] ^= 1;
@@ -339,7 +351,8 @@ async fn ciphertext_corruption_and_record_swaps_are_rejected() {
     assert!(
         MentionsStore::open(&config, account(), MAX_BYTES)
             .await
-            .is_err()
+            .unwrap()
+            .was_reset()
     );
     drop(store);
     config.delete().await.unwrap();
@@ -347,7 +360,7 @@ async fn ciphertext_corruption_and_record_swaps_are_rejected() {
 
 #[cfg_attr(not(target_family = "wasm"), tokio::test)]
 #[cfg_attr(target_family = "wasm", wasm_bindgen_test::wasm_bindgen_test)]
-async fn unsupported_schema_fails_without_resetting_data() {
+async fn unsupported_schema_resets_the_store() {
     let config = config();
     let mut store = open(&config).await;
     store
@@ -357,7 +370,7 @@ async fn unsupported_schema_fails_without_resetting_data() {
     let header = store
         .cipher
         .encrypt_value(&Head {
-            schema: 2,
+            schema: SCHEMA + 1,
             account: account(),
             revision: 1,
             records: store.snapshot.rows.len(),
@@ -377,9 +390,10 @@ async fn unsupported_schema_fails_without_resetting_data() {
     assert!(
         MentionsStore::open(&config, account(), MAX_BYTES)
             .await
-            .is_err()
+            .unwrap()
+            .was_reset()
     );
-    assert_eq!(store.backend.load().await.unwrap().unwrap().head, header);
+    assert_ne!(store.backend.load().await.unwrap().unwrap().head, header);
     drop(store);
     config.delete().await.unwrap();
 }

Modified file: src/mentions_model.rs+20 additions−17 deletions@@ -45,40 +45,42 @@ pub struct MentionEntry {
     pub notification_suppressed: bool,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub struct MentionCoverage {
     pub earliest_timestamp_ms: Option<u64>,
     pub historical_supported: Option<bool>,
     pub recovering: bool,
-    pub incomplete_history: bool,
     pub gaps: Vec<MentionGap>,
     pub truncated: bool,
     pub unresolved_candidates: usize,
 }
 
-impl Default for MentionCoverage {
-    fn default() -> Self {
-        Self {
-            earliest_timestamp_ms: None,
-            historical_supported: None,
-            recovering: false,
-            incomplete_history: true,
-            gaps: Vec::new(),
-            truncated: false,
-            unresolved_candidates: 0,
-        }
-    }
-}
-
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub struct MentionGap {
     pub room_id: Option<OwnedRoomId>,
-    pub reason: String,
+    pub reason: MentionGapReason,
     pub token: Option<String>,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum MentionGapReason {
+    SyncDiscontinuity,
+    /// an already observed room came back limited, `token` is its `prev_batch`
+    LimitedTimeline,
+    RoomLeft,
+    ContextUnavailable,
+    NotificationHistoryUnavailable,
+    /// a candidate stayed undecryptable or unresolved for a week and was dropped
+    CandidateExpired,
+    StoreReset,
+    /// a write failed, the inbox carries on in memory and a restart loses the unwritten part
+    StoreFailed,
+    HookTimedOut,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub struct MentionsBoundary {
@@ -117,6 +119,7 @@ pub struct MentionsUpdate {
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
 pub enum MentionsCommand {
     Page { cursor: Option<String>, limit: u16 },
     Context(MentionKey),

macltok

Commit 7cdaa86b9489

Commit `7cdaa86b9489`