macltok

public

Star0

Fork 0

Watch0

81 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `82d37c369846`

tarball

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

12 files changed+705-266

mentions: keep the sync record small and gaps honest

A failed command write left the in-memory inbox_revision ahead of the inbox, so the next sync-only commit made the following start reset everything. inbox_revision is now taken from the inbox at write time, observed rooms move to their own record and drop out on leave, and whether a response or a background completion changed anything is decided by comparing only the rooms it touched instead of cloning the whole inbox. HookTimedOut now means a response was dropped on a full queue, StoreFailed clears once a write lands, and a busy store or one from a newer schema runs in memory and is retried or left alone rather than wiped.

Modified file: Cargo.toml+1−1

@@ -33,7 +33,7 @@ matrix-sdk-store-encryption = { git = "ssh://brewery@pls.lobotomise.me:33/atagen

3333

 matrix-sdk-ui = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "b04108dd83394442231b7418d017ec30af791676", default-features = false }

3434 ruma = { version = "0.16.0", default-features = false, features = ["unstable-msc2545", "unstable-msc4095"] }

3535 serde = { version = "1.0.219", features = ["derive", "rc"], optional = true }

36-serde_json = "1.0.143"

36+serde_json = { version = "1.0.143", features = ["raw_value"] }

3737 thiserror = "2.0.16"

3838 mime = "0.3.17"

3939 sha2 = "0.10.9"

Modified file: docs/mentions-discovery.md+28 additions−18 deletions@@ -102,9 +102,11 @@ records a `SyncDiscontinuity` gap.
 
 The hook never fails sync. A failed write keeps the inbox in memory, records a
 `StoreFailed` gap, acks, and retries the write a minute later, so a restart
-before a write lands shows up as a discontinuity. The hook also stops waiting
-after ten seconds and records `HookTimedOut`, so a stuck actor cannot hold the
-position lock.
+before a write lands shows up as a discontinuity. The gap goes again once a
+later write lands, since nothing was lost. The hook gives up after ten seconds
+waiting for room in the actor's queue, drops the response and records
+`HookTimedOut`. Once queued, it waits another ten seconds for the commit and
+then acks regardless, because a queued response still commits in order.
 
 Sources are `crates/matrix-sdk/src/event_handler/mod.rs`,
 `crates/matrix-sdk/src/event_handler/static_events.rs`,
@@ -198,9 +200,11 @@ mentions in muted rooms and marks notification delivery as suppressed. Room-wide
 reasons require the relevant push conditions and sender power level. Missing
 authorisation context remains unresolved for reevaluation. An event that
 notifies and highlights with no other reason, such as a loud keyword, enters the
-inbox as `Keyword`. `add_notification_keyword` gives keyword rules a highlight
-tweak, and upgrades an existing rule for the pattern that lacks one, since the
-SDK's keyword rules only notify and play a sound. The classifier excludes the signed-in
+inbox as `Keyword`. The SDK's keyword rules only notify and play a sound, so
+`add_notification_keyword` reads the content rule the SDK names after the
+keyword and adds a highlight tweak when it lacks one, leaving its other actions
+alone. A rule the user made quiet stays quiet apart from the highlight. A
+change of reason alone keeps an entry's saved context. The classifier excludes the signed-in
 user's own events even when they contain explicit mention metadata.
 
 The existing `NotificationClient` combines notification evaluation with
@@ -260,10 +264,13 @@ inbox commits and discovery checkpoints.
 
 The passphrase wraps a random SDK `StoreCipher`. Logical keys are hashed, and
 authenticated encrypted records include schema, account identity, and logical
-key. Restore treats mismatched accounts, unsupported schemas, corrupt records,
-and inconsistent record counts or byte totals as unreadable, empties the store
-and records a `StoreReset` gap rather than failing `Account::build`. A store
-that cannot be opened at all leaves the inbox in memory behind `StoreFailed`.
+key. Restore treats mismatched accounts, older schemas, corrupt records, and
+inconsistent record counts or byte totals as unreadable, empties the store and
+records a `StoreReset` gap rather than failing `Account::build`. A busy or
+failing store and one written by a newer macltok are left alone, and the inbox
+runs in memory behind `StoreFailed`. The busy or failing case is retried every
+minute, and when it opens, what it held and what was built in memory since are
+merged.
 Atomic transactions compare the previous encrypted header before publishing a
 new revision, so stale writers cannot overwrite a newer checkpoint.
 
@@ -282,15 +289,18 @@ writes, replay from the previous checkpoint, reset of an unreadable inbox,
 notification recovery, live context updates, and capture persistence across
 session resume.
 
-The store holds two records. `inbox` carries entries, saved events, context
+The store holds three records. `inbox` carries entries, saved events, context
 windows, candidates, edits, redactions, coverage and recovery cursors. `sync`
-carries the connection checkpoints, observed rooms and the inbox
-revision it belongs to. Every response writes `sync`, and
-`inbox` only when the response changed it, both in one commit. A quiet
-response therefore costs one small record, while a changed inbox is still
-reserialised and re-encrypted whole. Background completions are applied in
-memory and written together at most once a second. When the budget is short,
-eviction runs against a size estimate and the inbox is written once.
+carries the connection checkpoints and the inbox revision it belongs to.
+`rooms` carries the rooms already observed, which drop out when left. Every
+response writes `sync`, and `inbox` or `rooms` only when the response changed
+them, all in one commit. Whether the inbox changed is decided by comparing the
+rooms the response touched, so a quiet response costs those rooms and one small
+record, while a changed inbox is still reserialised and re-encrypted whole, in
+one serialisation pass. Background completions are applied in memory and
+written together at most once a second, and published on that tick even while
+writes back off. When the budget is short, eviction runs against a size
+estimate and the inbox is written once.
 Decrypted previews, saved context, raw candidates, pending capture state,
 media source registrations, read state, coverage, and sync checkpoints stay
 inside the authenticated encrypted payloads. SDK SQLite custom values remain plaintext and are not used

Modified file: src/account.rs+14 additions−3 deletions@@ -854,12 +854,23 @@ impl Session {
                 .ok_or(AccountError::MissingSession)?
                 .to_owned(),
         };
-        let mentions = crate::mentions::store::MentionsStore::open(
+        let opener = crate::mentions::store::Opener::new(
             self.tracking.store.store(),
             account,
             config.max_mentions_bytes,
-        )
-        .await;
+        );
+        let mentions = match opener.open().await {
+            Ok(store) => Ok(store),
+            Err(crate::mentions::store::OpenError::Unavailable(error)) => {
+                Err(crate::mentions::service::Unopened {
+                    error,
+                    retry: Some(opener),
+                })
+            }
+            Err(crate::mentions::store::OpenError::Newer(error)) => {
+                Err(crate::mentions::service::Unopened { error, retry: None })
+            }
+        };
         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<Result<MentionsStore, crate::Error>>,

349+ mentions: Option<Result<MentionsStore, crate::mentions::service::Unopened>>,

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

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

352352 return Err(EngineError::Unauthenticated);

Modified file: src/mentions/service.rs+270 additions−130 deletions@@ -25,6 +25,8 @@ use serde_json::Value;
 use tokio::sync::{mpsc, oneshot, watch};
 
 use super::state::{Candidate, Checkpoint, Data, SavedEvent, SyncState, Window, id, invalid, now};
+#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+use super::store::Opener;
 use crate::{
     CommandOutput, EngineConfig, Error, ErrorKind, Event, MentionCaptureState, MentionEntry,
     MentionGapReason, MentionKey, MentionReason, MentionsCommand, MentionsPage, MentionsUpdate,
@@ -41,6 +43,38 @@ mod tests;
 const HOOK_TIMEOUT: Duration = Duration::from_secs(10);
 const FLUSH_RETRY_MS: u64 = 60_000;
 
+#[cfg_attr(not(any(feature = "sqlite", feature = "indexeddb")), allow(dead_code))]
+pub(crate) struct Unopened {
+    pub(crate) error: Error,
+    /// set when another attempt may succeed
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    pub(crate) retry: Option<Opener>,
+}
+
+struct Loaded {
+    store: Option<MentionsStore>,
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    reopen: Option<Opener>,
+    data: Data,
+    sync: SyncState,
+    rooms: BTreeSet<OwnedRoomId>,
+    gaps: Vec<MentionGapReason>,
+}
+
+impl Loaded {
+    fn empty(gaps: Vec<MentionGapReason>) -> Self {
+        Self {
+            store: None,
+            #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+            reopen: None,
+            data: Data::default(),
+            sync: SyncState::default(),
+            rooms: BTreeSet::new(),
+            gaps,
+        }
+    }
+}
+
 #[derive(Clone)]
 pub(crate) struct Handle {
     requests: mpsc::Sender<Request>,
@@ -63,9 +97,14 @@ pub(crate) struct Service {
     /// as many media as the inbox keeps events
     media_scope: MediaScope,
     store: Option<MentionsStore>,
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    reopen: Option<Opener>,
     data: Data,
     sync: SyncState,
+    rooms: BTreeSet<OwnedRoomId>,
     dirty: bool,
+    rooms_dirty: bool,
+    unpublished: bool,
     flush_after: u64,
     ignored: BTreeSet<OwnedUserId>,
     leaving: BTreeSet<OwnedRoomId>,
@@ -92,6 +131,24 @@ enum Work {
     ),
 }
 
+impl Work {
+    fn rooms(&self) -> BTreeSet<OwnedRoomId> {
+        let room = |key: &str| {
+            key.split_once('\n')
+                .and_then(|(room, _)| OwnedRoomId::try_from(room).ok())
+        };
+        match self {
+            Self::Capture(key, ..) | Self::Candidate(key, _) => room(key).into_iter().collect(),
+            Self::History(_, _, Ok(page)) => page
+                .notifications
+                .iter()
+                .map(|notification| notification.room_id.clone())
+                .collect(),
+            Self::History(..) => BTreeSet::new(),
+        }
+    }
+}
+
 impl Handle {
     /// never fails, a mentions failure becomes a coverage gap rather than a stalled sync
     pub(crate) fn hook(&self) -> ResponseCommitHook {
@@ -99,23 +156,29 @@ impl Handle {
         ResponseCommitHook::new(move |batch| {
             let handle = handle.clone();
             async move {
-                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"),
+                let (reply, response) = oneshot::channel();
+                let queued = handle.requests.send(Request::Batch(batch, reply));
+                match matrix_sdk::timeout::timeout(queued, HOOK_TIMEOUT).await {
+                    Ok(Ok(())) => {}
+                    Ok(Err(_)) => {
+                        tracing::warn!("mentions service stopped, sync carries on");
+                        return Ok(());
+                    }
                     Err(_) => {
                         handle.timed_out.store(true, Ordering::Release);
-                        tracing::warn!("mentions commit timed out, sync carries on");
+                        tracing::warn!("mentions queue stayed full, a response was dropped");
+                        return Ok(());
                     }
                 }
+                // a queued batch still commits in order, so giving up here loses nothing
+                // unless the process dies first, which the next start sees as a
+                // discontinuity
+                if matrix_sdk::timeout::timeout(response, HOOK_TIMEOUT)
+                    .await
+                    .is_err()
+                {
+                    tracing::warn!("mentions commit is slow, sync carries on");
+                }
                 Ok(())
             }
         })
@@ -139,52 +202,40 @@ impl Handle {
 }
 
 #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
-async fn open_store(
-    store: Option<Result<MentionsStore, Error>>,
-) -> (
-    Option<MentionsStore>,
-    Data,
-    SyncState,
-    Vec<MentionGapReason>,
-) {
+fn read_store(store: &MentionsStore) -> Result<(Data, SyncState, BTreeSet<OwnedRoomId>), Error> {
+    let data: Data = store.read("inbox")?.unwrap_or_default();
+    let sync: SyncState = store.read("sync")?.unwrap_or_default();
+    let rooms = store.read("rooms")?.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, rooms))
+}
+
+#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+async fn open_store(store: Option<Result<MentionsStore, Unopened>>) -> Loaded {
     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],
-            );
+        None => return Loaded::empty(Vec::new()),
+        Some(Err(unopened)) => {
+            tracing::error!(error = %unopened.error, "mentions store unavailable, the inbox runs in memory");
+            let mut loaded = Loaded::empty(vec![MentionGapReason::StoreFailed]);
+            loaded.reopen = unopened.retry;
+            return loaded;
         }
         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 {
+    let (data, sync, rooms) = match read_store(&store) {
         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],
-                );
+                return Loaded::empty(vec![MentionGapReason::StoreFailed]);
             }
-            (Data::default(), SyncState::default())
+            Default::default()
         }
     };
     let gaps = if store.was_reset() {
@@ -192,31 +243,30 @@ async fn open_store(
     } else {
         Vec::new()
     };
-    (Some(store), data, sync, gaps)
+    Loaded {
+        store: Some(store),
+        reopen: None,
+        data,
+        sync,
+        rooms,
+        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],
+async fn open_store(store: Option<Result<MentionsStore, Unopened>>) -> Loaded {
+    match store {
+        None => Loaded::empty(Vec::new()),
+        Some(Err(_)) => Loaded::empty(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<Result<MentionsStore, Error>>,
+        store: Option<Result<MentionsStore, Unopened>>,
         config: EngineConfig,
     ) -> Result<(Handle, Self, watch::Receiver<MentionsPage>), Error> {
         let account = client
@@ -228,18 +278,18 @@ impl Service {
                 )
             })?
             .to_owned();
-        let (store, mut data, sync, gaps) = open_store(store).await;
-        for reason in &gaps {
-            data.gap(None, *reason, None);
+        let mut loaded = open_store(store).await;
+        for reason in &loaded.gaps {
+            loaded.data.gap(None, *reason, None);
         }
-        let page = data.page(&account, None, 50)?;
+        let page = loaded.data.page(&account, None, 50)?;
         let media_scope = MediaScope::bounded(config.max_mention_context_events);
         media
             .register(
                 &media_scope,
                 page.entries
                     .iter()
-                    .filter_map(|entry| data.events.get(&id(&entry.key)))
+                    .filter_map(|entry| loaded.data.events.get(&id(&entry.key)))
                     .flat_map(|event| event.media.clone())
                     .collect(),
             )
@@ -252,10 +302,15 @@ impl Service {
             account,
             media,
             media_scope,
-            store,
-            data,
-            sync,
-            dirty: !gaps.is_empty(),
+            store: loaded.store,
+            #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+            reopen: loaded.reopen,
+            data: loaded.data,
+            sync: loaded.sync,
+            rooms: loaded.rooms,
+            dirty: !loaded.gaps.is_empty(),
+            rooms_dirty: false,
+            unpublished: false,
             flush_after: 0,
             ignored: BTreeSet::new(),
             leaving: BTreeSet::new(),
@@ -291,6 +346,7 @@ impl Service {
     }
 
     async fn publish(&mut self) {
+        self.unpublished = false;
         let Ok(page) = self.data.page(&self.account, None, 50) else {
             return;
         };
@@ -307,15 +363,15 @@ impl Service {
         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
+    /// writes the sync record, with the inbox when `dirty` and observed rooms when
+    /// `rooms_dirty`, in one commit so a checkpoint never lands without the 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| {
@@ -325,27 +381,52 @@ impl Service {
                     error,
                 )
             };
-            let sync = serde_json::to_value(&self.sync).map_err(encode)?;
+            self.sync.inbox_revision = self.data.revision;
+            let sync = serde_json::value::to_raw_value(&self.sync).map_err(encode)?;
+            let rooms = if self.rooms_dirty {
+                Some(serde_json::value::to_raw_value(&self.rooms).map_err(encode)?)
+            } else {
+                None
+            };
             let mut writes = Vec::new();
             if self.dirty {
+                self.data
+                    .coverage
+                    .gaps
+                    .retain(|gap| gap.reason != MentionGapReason::StoreFailed);
+                let replaced: &[&str] = if rooms.is_some() {
+                    &["inbox", "sync", "rooms"]
+                } else {
+                    &["inbox", "sync"]
+                };
                 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);
+                    .available(replaced)
+                    .saturating_sub(store.sealed_bound("sync", sync.get().len()))
+                    .saturating_sub(
+                        rooms
+                            .as_ref()
+                            .map_or(0, |rooms| store.sealed_bound("rooms", rooms.get().len())),
+                    );
+                let mut inbox = serde_json::value::to_raw_value(&self.data).map_err(encode)?;
+                let mut size = store.sealed_bound("inbox", inbox.get().len());
+                if size > available {
+                    while size > available {
+                        let Some(freed) = self.data.evict_one() else {
+                            break;
+                        };
+                        size = size.saturating_sub(freed * 3);
+                    }
+                    inbox = serde_json::value::to_raw_value(&self.data).map_err(encode)?;
                 }
-                writes.push(Write::Set(
-                    "inbox".to_owned(),
-                    serde_json::to_value(&self.data).map_err(encode)?,
-                ));
+                writes.push(Write::Set("inbox".to_owned(), inbox));
             }
             writes.push(Write::Set("sync".to_owned(), sync));
+            if let Some(rooms) = rooms {
+                writes.push(Write::Set("rooms".to_owned(), rooms));
+            }
             store.commit(store.revision(), writes).await?;
         }
+        self.rooms_dirty = false;
         if self.dirty {
             self.dirty = false;
             self.publish().await;
@@ -364,7 +445,6 @@ impl Service {
             self.config.max_mention_context_events,
             self.config.max_mention_candidates,
         );
-        self.sync.inbox_revision = self.data.revision;
         Ok(())
     }
 
@@ -387,6 +467,54 @@ impl Service {
         self.publish().await;
     }
 
+    /// the store is back after failing at startup, what it held and what was built in
+    /// memory since are merged so nothing is lost
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    async fn reopen(&mut self) {
+        let Some(opener) = &self.reopen else {
+            return;
+        };
+        let store = match opener.open().await {
+            Ok(store) => store,
+            Err(super::store::OpenError::Unavailable(error)) => {
+                tracing::warn!(%error, "the mentions store is still unavailable");
+                self.flush_after = now().saturating_add(FLUSH_RETRY_MS);
+                return;
+            }
+            Err(super::store::OpenError::Newer(error)) => {
+                tracing::error!(%error, "the mentions store belongs to a newer macltok");
+                self.reopen = None;
+                return;
+            }
+        };
+        let mut store = store;
+        let (mut data, mut sync, mut rooms) = match read_store(&store) {
+            Ok(loaded) => loaded,
+            Err(error) => {
+                tracing::warn!(%error, "resetting an unreadable mentions inbox");
+                if let Err(error) = store.reset().await {
+                    tracing::warn!(%error, "the mentions store is still unavailable");
+                    self.flush_after = now().saturating_add(FLUSH_RETRY_MS);
+                    return;
+                }
+                self.data.gap(None, MentionGapReason::StoreReset, None);
+                Default::default()
+            }
+        };
+        data.absorb(std::mem::take(&mut self.data));
+        sync.checkpoints
+            .extend(std::mem::take(&mut self.sync.checkpoints));
+        rooms.extend(std::mem::take(&mut self.rooms));
+        self.data = data;
+        self.sync = sync;
+        self.rooms = rooms;
+        self.store = Some(store);
+        self.reopen = None;
+        self.dirty = true;
+        self.rooms_dirty = true;
+        self.flush_after = 0;
+    }
+
     async fn command(&mut self, command: MentionsCommand) -> Result<CommandOutput, Error> {
         match command {
             MentionsCommand::Page { cursor, limit } => {
@@ -495,8 +623,16 @@ impl Service {
         }
         self.ignored = self.ignored_users().await;
         self.leaving = batch.response.rooms.left.keys().cloned().collect();
-        let mut next = self.data.clone();
-        let mut sync = self.sync.clone();
+        let room_ids: BTreeSet<_> = batch
+            .raw_rooms
+            .keys()
+            .chain(batch.response.rooms.joined.keys())
+            .chain(batch.response.rooms.left.keys())
+            .cloned()
+            .collect();
+        let before = self.data.scope(&room_ids);
+        let mut next = std::mem::take(&mut self.data);
+        let mut changed = false;
         if self.timed_out.swap(false, Ordering::AcqRel) {
             next.gap(None, MentionGapReason::HookTimedOut, None);
         }
@@ -513,7 +649,7 @@ impl Service {
             next.gap(None, MentionGapReason::SyncDiscontinuity, None);
         }
         for (room_id, raw) in &batch.raw_rooms {
-            if raw.limited && sync.observed_rooms.contains(room_id) {
+            if raw.limited && self.rooms.contains(room_id) {
                 next.gap(
                     Some(room_id.clone()),
                     MentionGapReason::LimitedTimeline,
@@ -521,18 +657,11 @@ impl Service {
                 );
             }
         }
-        let room_ids: BTreeSet<_> = batch
-            .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.observable(room_id).is_none() {
                 continue;
             }
-            sync.observed_rooms.insert(room_id.clone());
+            self.rooms_dirty |= self.rooms.insert(room_id.clone());
             let mut observed = BTreeMap::new();
             if let Some(raw) = batch.raw_rooms.get(room_id) {
                 for event in &raw.timeline {
@@ -575,6 +704,7 @@ impl Service {
             }
         }
         for room_id in batch.response.rooms.left.keys() {
+            self.rooms_dirty |= self.rooms.remove(room_id);
             next.gap(Some(room_id.clone()), MentionGapReason::RoomLeft, None);
             next.candidates
                 .retain(|_, candidate| &candidate.room_id != room_id);
@@ -586,35 +716,30 @@ impl Service {
                 entry.revision = revision;
             }
         }
-        let reevaluate: Vec<_> = next
-            .entries
-            .values()
-            .filter(|entry| room_ids.contains(&entry.key.room_id))
-            .filter_map(|entry| {
-                next.events
-                    .get(&id(&entry.key))
-                    .and_then(|event| event.raw.clone())
-                    .map(|raw| (entry.key.room_id.clone(), raw))
+        let reevaluate: Vec<_> = room_ids
+            .iter()
+            .flat_map(|room_id| {
+                next.entries
+                    .range(format!("{room_id}\n")..format!("{room_id}\u{b}"))
+                    .filter_map(|(key, _)| next.events.get(key).and_then(|event| event.raw.clone()))
+                    .map(move |raw| (room_id.clone(), raw))
             })
             .collect();
         for (room_id, raw) in reevaluate {
             self.observe(&mut next, &room_id, raw).await;
         }
-        self.filter_ignored(&mut next);
+        changed |= self.filter_ignored(&mut next);
         self.leaving.clear();
-        sync.checkpoints.insert(
+        self.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;
+        changed |= next.scope(&room_ids) != before;
+        self.data = next;
+        self.dirty |= changed;
         if now() < self.flush_after {
             if changed && self.advance().is_ok() {
                 self.publish().await;
@@ -633,12 +758,15 @@ impl Service {
             })
             .map(|(key, _)| key.clone())
             .collect();
+        if removed.is_empty() {
+            return false;
+        }
         for key in &removed {
             next.entries.remove(key);
             next.windows.remove(key);
         }
         next.collect_events();
-        !removed.is_empty()
+        true
     }
 
     async fn observe(&self, data: &mut Data, room_id: &OwnedRoomId, mut value: Value) {
@@ -793,11 +921,11 @@ impl Service {
             return;
         }
         let previous = data.entries.get(&identity);
-        let changed = previous.is_none_or(|entry| {
-            entry.event != item
-                || entry.reasons != reasons
-                || entry.notification_suppressed != suppressed
-        });
+        let event_changed = previous.is_none_or(|entry| entry.event != item);
+        let changed = event_changed
+            || previous.is_some_and(|entry| {
+                entry.reasons != reasons || entry.notification_suppressed != suppressed
+            });
         let entry = MentionEntry {
             key,
             timestamp_ms: timestamp,
@@ -813,7 +941,7 @@ impl Service {
             reasons,
             read: previous.is_some_and(|entry| entry.read),
             capture: previous.map_or(MentionCaptureState::Pending, |entry| {
-                if changed {
+                if event_changed {
                     MentionCaptureState::Pending
                 } else {
                     entry.capture.clone()
@@ -937,7 +1065,7 @@ impl Service {
             return (vec![], false, false);
         };
         let raw: Raw<AnySyncTimelineEvent> =
-            Raw::from_json_string(value.to_string()).expect("serialized event");
+            Raw::from_json_string(value.to_string()).expect("serialised event");
         let actions = room.event_push_actions(&raw).await.ok().flatten();
         let notify = actions.as_ref().is_some_and(|actions| {
             actions
@@ -1026,11 +1154,15 @@ impl Service {
                     None => return,
                 },
                 Some(result) = work.next(), if !work.is_empty() => {
+                    let rooms = result.rooms();
+                    let before = self.data.scope(&rooms);
                     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;
+                    let changed = self.complete(&mut next, result, &mut captures, &mut retries, &mut history).await
+                        && next.scope(&rooms) != before;
                     self.data = next;
                     self.dirty |= changed;
+                    self.unpublished |= changed;
                 }
                 Some(report) = reports.next() => {
                     let room = match report { Ok(RedecryptorReport::ResolvedUtds { room_id, .. }) => Some(room_id), _ => None };
@@ -1039,14 +1171,22 @@ impl Service {
                 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);
+                    let changed = self.filter_ignored(&mut next);
                     self.data = next;
+                    self.dirty |= changed;
+                    self.unpublished |= changed;
                 }
                 () = &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; }
+                    if now() >= next_expiry { let expired = self.data.expire_candidates(now()); self.dirty |= expired; self.unpublished |= expired; next_expiry = now() + 60_000; }
+                    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+                    if self.store.is_none() && self.reopen.is_some() && now() >= self.flush_after { self.reopen().await; }
+                    if self.dirty && now() >= self.flush_after {
+                        if let Err(error) = self.flush().await { self.store_failed(error).await; }
+                    } else if self.unpublished && self.advance().is_ok() {
+                        self.publish().await;
+                    }
                 }
             }
         }

Modified file: src/mentions/service_tests.rs+67 additions−3 deletions@@ -109,7 +109,7 @@ fn quiet_batch(previous_pos: &str, next_pos: &str) -> ResponseCommitBatch {
     }
 }
 
-async fn service(client: &Client, store: Option<Result<MentionsStore, Error>>) -> Service {
+async fn service(client: &Client, store: Option<Result<MentionsStore, Unopened>>) -> Service {
     let media = Arc::new(MediaService::new(
         client.clone(),
         EngineConfig::default().max_media_bytes,
@@ -230,11 +230,11 @@ async fn a_quiet_response_writes_only_the_checkpoint() {
         .await;
     let connection = Connection::open(config.mentions_location()).unwrap();
     let before = rows(&connection);
-    assert_eq!(before.len(), 2);
+    assert_eq!(before.len(), 3);
 
     service.batch(quiet_batch("position-1", "position-2")).await;
     let after = rows(&connection);
-    assert_eq!(after.len(), 2);
+    assert_eq!(after.len(), 3);
     assert_eq!(
         before
             .iter()
@@ -253,6 +253,70 @@ async fn a_quiet_response_writes_only_the_checkpoint() {
 }
 
 #[tokio::test]
+async fn a_failed_command_write_leaves_a_store_that_still_opens() {
+    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,
+            "first",
+            None,
+            "position-1",
+        ))
+        .await;
+    let key = first.data.entries.values().next().unwrap().key.clone();
+
+    let connection = Connection::open(config.mentions_location()).unwrap();
+    connection
+        .execute_batch(
+            "CREATE TRIGGER fail_mentions_update BEFORE UPDATE ON records WHEN OLD.key NOT IN ('!cipher', '!head') BEGIN SELECT RAISE(ABORT, 'injected write failure'); END;",
+        )
+        .unwrap();
+    assert!(
+        first
+            .command(MentionsCommand::SetRead {
+                keys: vec![key],
+                read: true,
+            })
+            .await
+            .is_err()
+    );
+    connection
+        .execute_batch("DROP TRIGGER fail_mentions_update;")
+        .unwrap();
+    drop(connection);
+    first.batch(quiet_batch("position-1", "position-2")).await;
+    drop(first);
+
+    let store = MentionsStore::open(&config, account(&client), MAX_BYTES)
+        .await
+        .unwrap();
+    let restarted = service(&client, Some(Ok(store))).await;
+    assert_eq!(restarted.data.entries.len(), 1);
+    assert!(
+        restarted
+            .data
+            .coverage
+            .gaps
+            .iter()
+            .all(|gap| gap.reason != MentionGapReason::StoreReset)
+    );
+    assert_eq!(
+        restarted.sync.checkpoints["mentions-connection"].pos,
+        "position-2"
+    );
+    drop(restarted);
+    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");

Modified file: src/mentions/state.rs+97 additions−1 deletions@@ -91,7 +91,47 @@ pub(super) struct Data {
 pub(super) struct SyncState {
     pub inbox_revision: u64,
     pub checkpoints: BTreeMap<String, Checkpoint>,
-    pub observed_rooms: BTreeSet<ruma::OwnedRoomId>,
+}
+
+/// what a set of rooms holds in the inbox, compared before and after a change so an
+/// unchanged inbox costs only the rooms it touched
+#[derive(PartialEq)]
+pub(super) struct Scope {
+    rooms: Vec<RoomScope>,
+    coverage: MentionCoverage,
+    cursors: (
+        Option<String>,
+        bool,
+        Option<String>,
+        Option<String>,
+        Option<String>,
+    ),
+}
+
+#[derive(PartialEq)]
+struct RoomScope {
+    entries: Vec<MentionEntry>,
+    events: Vec<SavedEvent>,
+    windows: Vec<Window>,
+    capture_attempts: Vec<u32>,
+    candidates: Vec<Candidate>,
+    redacted: Vec<u64>,
+    edits: Vec<Value>,
+    keys: Vec<String>,
+}
+
+fn in_room<V: Clone>(
+    map: &BTreeMap<String, V>,
+    room: &ruma::RoomId,
+    keys: &mut Vec<String>,
+) -> Vec<V> {
+    // no room id holds a newline, so a room's "room\nevent" keys sit below "room\x0b"
+    map.range(format!("{room}\n")..format!("{room}\u{b}"))
+        .map(|(key, value)| {
+            keys.push(key.clone());
+            value.clone()
+        })
+        .collect()
 }
 
 #[derive(Clone, Debug, PartialEq)]
@@ -142,6 +182,62 @@ fn timestamp(value: &Value) -> u64 {
 }
 
 impl Data {
+    pub fn scope<'a>(&self, rooms: impl IntoIterator<Item = &'a ruma::OwnedRoomId>) -> Scope {
+        Scope {
+            rooms: rooms
+                .into_iter()
+                .map(|room| {
+                    let mut keys = Vec::new();
+                    RoomScope {
+                        entries: in_room(&self.entries, room, &mut keys),
+                        events: in_room(&self.events, room, &mut keys),
+                        windows: in_room(&self.windows, room, &mut keys),
+                        capture_attempts: in_room(&self.capture_attempts, room, &mut keys),
+                        candidates: in_room(&self.candidates, room, &mut keys),
+                        redacted: in_room(&self.redacted, room, &mut keys),
+                        edits: in_room(&self.edits, room, &mut keys),
+                        keys,
+                    }
+                })
+                .collect(),
+            coverage: self.coverage.clone(),
+            cursors: (
+                self.history_cursor.clone(),
+                self.history_exhausted,
+                self.latest_cursor.clone(),
+                self.latest_anchor.clone(),
+                self.latest_head.clone(),
+            ),
+        }
+    }
+
+    /// folds an inbox built in memory while the store was unreachable into the one the
+    /// store held, keeping the newer side where both have a key
+    pub fn absorb(&mut self, newer: Data) {
+        self.revision = self.revision.max(newer.revision);
+        self.entries.extend(newer.entries);
+        self.events.extend(newer.events);
+        self.windows.extend(newer.windows);
+        self.capture_attempts.extend(newer.capture_attempts);
+        self.candidates.extend(newer.candidates);
+        self.redacted.extend(newer.redacted);
+        self.edits.extend(newer.edits);
+        for gap in newer.coverage.gaps {
+            self.gap(gap.room_id, gap.reason, gap.token);
+        }
+        self.coverage.earliest_timestamp_ms = self
+            .coverage
+            .earliest_timestamp_ms
+            .into_iter()
+            .chain(newer.coverage.earliest_timestamp_ms)
+            .min();
+        self.coverage.truncated |= newer.coverage.truncated;
+        self.history_cursor = newer.history_cursor.or(self.history_cursor.take());
+        self.latest_cursor = newer.latest_cursor.or(self.latest_cursor.take());
+        self.latest_anchor = newer.latest_anchor.or(self.latest_anchor.take());
+        self.latest_head = newer.latest_head.or(self.latest_head.take());
+    }
+
     pub fn gap(
         &mut self,
         room_id: Option<ruma::OwnedRoomId>,

Modified file: src/mentions/state_tests.rs+2 additions−2 deletions@@ -388,7 +388,7 @@ mod persistence {
                     0,
                     vec![Write::Set(
                         "inbox".to_owned(),
-                        serde_json::to_value(&data).unwrap(),
+                        serde_json::value::to_raw_value(&data).unwrap(),
                     )],
                 )
                 .await
@@ -422,7 +422,7 @@ mod persistence {
                 1,
                 vec![Write::Set(
                     "inbox".to_owned(),
-                    serde_json::to_value(&restored_data).unwrap(),
+                    serde_json::value::to_raw_value(&restored_data).unwrap(),
                 )],
             )
             .await

Modified file: src/mentions/store/indexeddb.rs+15 additions−13 deletions@@ -9,7 +9,7 @@ use indexed_db_futures::{
     typed_array::{Uint8Array, Uint8ArraySlice},
 };
 
-use super::Snapshot;
+use super::{Failure, Snapshot};
 
 const RECORDS: &str = "records";
 const CIPHER: &str = "!cipher";
@@ -39,26 +39,28 @@ pub(super) struct Backend {
 }
 
 impl Backend {
-    pub(super) async fn open(name: String) -> Result<Self, crate::Error> {
+    pub(super) async fn open(name: String) -> Result<(Self, bool), crate::Error> {
         let _database = open_database(&name)
             .await
             .map_err(|error| super::storage_error("opening the mentions database", error))?;
-        Ok(Self { name })
+        Ok((Self { name }, false))
     }
 
-    pub(super) async fn load(&self) -> Result<Option<Snapshot>, crate::Error> {
-        let database = open_database(&self.name)
-            .await
-            .map_err(|error| super::storage_error("opening the mentions database", error))?;
+    pub(super) async fn load(&self) -> Result<Option<Snapshot>, Failure> {
+        let database = open_database(&self.name).await.map_err(|error| {
+            Failure::Unavailable(super::storage_error("opening the mentions database", error))
+        })?;
         let snapshot = read_snapshot(&database).await;
         match snapshot {
             Ok(snapshot) => Ok(snapshot),
-            Err(ReadError::Invalid(message)) => {
-                Err(crate::Error::new(crate::ErrorKind::Store, message))
-            }
-            Err(ReadError::Storage(error)) => {
-                Err(super::storage_error("reading the mentions database", error))
-            }
+            Err(ReadError::Invalid(message)) => Err(Failure::Corrupt(crate::Error::new(
+                crate::ErrorKind::Store,
+                message,
+            ))),
+            Err(ReadError::Storage(error)) => Err(Failure::Unavailable(super::storage_error(
+                "reading the mentions database",
+                error,
+            ))),
         }
     }
 

Modified file: src/mentions/store/mod.rs+140 additions−47 deletions@@ -4,7 +4,7 @@ use matrix_sdk::{SendOutsideWasm, SyncOutsideWasm};
 use matrix_sdk_store_encryption::StoreCipher;
 use ruma::OwnedUserId;
 use serde::{Deserialize, Serialize, de::DeserializeOwned};
-use serde_json::Value;
+use serde_json::{Value, value::RawValue};
 
 use crate::{Error, ErrorKind, StoreConfig};
 
@@ -35,7 +35,7 @@ struct Head {
     bytes: usize,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Deserialize)]
 struct Record {
     schema: u32,
     account: Account,
@@ -43,6 +43,14 @@ struct Record {
     value: Value,
 }
 
+#[derive(Serialize)]
+struct SealedRecord<'a> {
+    schema: u32,
+    account: &'a Account,
+    key: &'a str,
+    value: &'a RawValue,
+}
+
 pub(super) struct Snapshot {
     pub(super) exported: Vec<u8>,
     pub(super) head: Vec<u8>,
@@ -50,10 +58,67 @@ pub(super) struct Snapshot {
 }
 
 pub(crate) enum Write {
-    Set(String, Value),
+    Set(String, Box<RawValue>),
     Remove(String),
 }
 
+/// what a backend could not read, content can be wiped and anything else is left alone
+#[derive(Debug)]
+pub(super) enum Failure {
+    Corrupt(Error),
+    Unavailable(Error),
+}
+
+impl Failure {
+    fn into_error(self) -> Error {
+        match self {
+            Self::Corrupt(error) | Self::Unavailable(error) => error,
+        }
+    }
+}
+
+impl From<Error> for Failure {
+    fn from(error: Error) -> Self {
+        Self::Unavailable(error)
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum OpenError {
+    /// worth another try later, nothing was wiped
+    Unavailable(Error),
+    /// written by a newer macltok, left as it is
+    Newer(Error),
+}
+
+/// everything needed to open the store again after a failed attempt
+pub(crate) struct Opener {
+    location: Location,
+    passphrase: String,
+    account: Account,
+    max_bytes: usize,
+}
+
+#[cfg(feature = "sqlite")]
+type Location = std::path::PathBuf;
+#[cfg(feature = "indexeddb")]
+type Location = String;
+
+impl Opener {
+    pub(crate) fn new(config: &StoreConfig, account: Account, max_bytes: usize) -> Self {
+        Self {
+            location: config.mentions_location(),
+            passphrase: config.passphrase().to_owned(),
+            account,
+            max_bytes,
+        }
+    }
+
+    pub(crate) async fn open(&self) -> Result<MentionsStore, OpenError> {
+        MentionsStore::open_with(self).await
+    }
+}
+
 pub(crate) struct MentionsStore {
     backend: Backend,
     cipher: StoreCipher,
@@ -75,37 +140,53 @@ 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 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)
+    ) -> Result<Self, OpenError> {
+        Opener::new(config, account, max_bytes).open().await
+    }
+
+    /// a corrupt or foreign store is emptied, one that is busy, failing or newer is
+    /// left alone
+    async fn open_with(opener: &Opener) -> Result<Self, OpenError> {
+        let (backend, mut reset) = Backend::open(opener.location.clone())
+            .await
+            .map_err(OpenError::Unavailable)?;
+        let restored = match backend.load().await {
+            Ok(Some(snapshot)) => match restore(opener, snapshot) {
+                Ok(restored) => Some(restored),
+                Err(Rejected::Newer(error)) => return Err(OpenError::Newer(error)),
+                Err(Rejected::Content(error)) => {
+                    tracing::warn!(%error, "resetting an unreadable mentions store");
+                    backend.reset().await.map_err(OpenError::Unavailable)?;
+                    reset = true;
+                    None
+                }
+            },
+            Ok(None) => None,
+            Err(Failure::Corrupt(error)) => {
+                tracing::warn!(%error, "resetting a corrupt mentions store");
+                backend.reset().await.map_err(OpenError::Unavailable)?;
+                reset = true;
+                None
             }
+            Err(Failure::Unavailable(error)) => return Err(OpenError::Unavailable(error)),
         };
         let (cipher, snapshot, revision) = match restored {
             Some(restored) => restored,
-            None => create(&backend, config, &account, max_bytes).await?,
+            None => create(&backend, opener)
+                .await
+                .map_err(OpenError::Unavailable)?,
         };
         Ok(Self {
             backend,
             cipher,
-            account,
+            account: opener.account.clone(),
             snapshot,
             revision,
-            max_bytes,
+            max_bytes: opener.max_bytes,
             reset,
         })
     }
@@ -202,11 +283,11 @@ impl MentionsStore {
             match write {
                 Write::Set(key, value) => {
                     let hashed = storage_key(&self.cipher, &key);
-                    let record = Record {
+                    let record = SealedRecord {
                         schema: SCHEMA,
-                        account: self.account.clone(),
-                        key,
-                        value,
+                        account: &self.account,
+                        key: &key,
+                        value: &value,
                     };
                     let encrypted = self
                         .cipher
@@ -266,54 +347,66 @@ 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)
+enum Rejected {
+    Content(Error),
+    Newer(Error),
+}
+
+impl From<Error> for Rejected {
+    fn from(error: Error) -> Self {
+        Self::Content(error)
+    }
+}
+
+#[derive(Deserialize)]
+struct Schema {
+    schema: u32,
+}
+
+fn restore(opener: &Opener, snapshot: Snapshot) -> Result<(StoreCipher, Snapshot, u64), Rejected> {
+    let cipher = StoreCipher::import(&opener.passphrase, &snapshot.exported)
         .map_err(|error| storage_error("opening the mentions cipher", error))?;
+    let schema: Schema = cipher
+        .decrypt_value(&snapshot.head)
+        .map_err(|error| storage_error("opening the mentions header", error))?;
+    if schema.schema > SCHEMA {
+        return Err(Rejected::Newer(invalid(
+            "mentions store was written by a newer macltok",
+        )));
+    }
+    validate_size(&snapshot, opener.max_bytes)?;
     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)?;
+    validate_account(head.schema, &head.account, &opener.account)?;
     if head.records != snapshot.rows.len() || head.bytes != row_bytes(&snapshot.rows)? {
-        return Err(invalid(
-            "mentions records do not match their committed header",
-        ));
+        return Err(invalid("mentions records do not match their committed header").into());
     }
     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)?;
+        validate_account(record.schema, &record.account, &opener.account)?;
         if storage_key(&cipher, &record.key) != *key {
-            return Err(invalid("a mentions record belongs to a different key"));
+            return Err(invalid("a mentions record belongs to a different key").into());
         }
     }
     Ok((cipher, snapshot, head.revision))
 }
 
-async fn create(
-    backend: &Backend,
-    config: &StoreConfig,
-    account: &Account,
-    max_bytes: usize,
-) -> Result<(StoreCipher, Snapshot, u64), Error> {
+async fn create(backend: &Backend, opener: &Opener) -> 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())
+        .export(&opener.passphrase)
         .map_err(|error| storage_error("sealing the mentions cipher", error))?;
-    let head = seal_head(&cipher, account, 0, 0, 0)?;
+    let head = seal_head(&cipher, &opener.account, 0, 0, 0)?;
     let snapshot = Snapshot {
         exported,
         head,
         rows: BTreeMap::new(),
     };
-    validate_size(&snapshot, max_bytes)?;
+    validate_size(&snapshot, opener.max_bytes)?;
     backend
         .commit(
             None,

Modified file: src/mentions/store/sqlite.rs+59 additions−37 deletions@@ -8,7 +8,7 @@ use std::{
 
 use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
 
-use super::{Snapshot, conflict, storage_error};
+use super::{Failure, Snapshot, conflict, storage_error};
 use crate::{Error, ErrorKind};
 
 const CIPHER_KEY: &str = "!cipher";
@@ -19,36 +19,36 @@ pub(super) struct Backend {
 }
 
 impl Backend {
-    pub(super) async fn open(path: PathBuf) -> Result<Self, Error> {
+    /// a file that is not a database is replaced, `true` when that happened
+    pub(super) async fn open(path: PathBuf) -> Result<(Self, bool), Error> {
         let setup_path = path.clone();
-        tokio::task::spawn_blocking(move || initialise(&setup_path))
-            .await
-            .map_err(|error| storage_error("opening the mentions store", error))??;
-        Ok(Self { path })
+        let reset = tokio::task::spawn_blocking(move || match initialise(&setup_path) {
+            Ok(()) => Ok(false),
+            Err(Failure::Corrupt(error)) => {
+                tracing::warn!(%error, "replacing a corrupt mentions database");
+                replace(&setup_path).map(|()| true)
+            }
+            Err(Failure::Unavailable(error)) => Err(error),
+        })
+        .await
+        .map_err(|error| storage_error("opening the mentions store", error))??;
+        Ok((Self { path }, reset))
     }
 
-    pub(super) async fn load(&self) -> Result<Option<Snapshot>, Error> {
+    pub(super) async fn load(&self) -> Result<Option<Snapshot>, Failure> {
         let path = self.path.clone();
         tokio::task::spawn_blocking(move || load_snapshot(&path))
             .await
-            .map_err(|error| storage_error("loading the mentions store", error))?
+            .map_err(|error| {
+                Failure::Unavailable(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))?
+        tokio::task::spawn_blocking(move || replace(&path))
+            .await
+            .map_err(|error| storage_error("resetting the mentions store", error))?
     }
 
     pub(super) async fn commit(
@@ -67,7 +67,32 @@ impl Backend {
     }
 }
 
-fn initialise(path: &Path) -> Result<(), Error> {
+fn replace(path: &Path) -> Result<(), Error> {
+    for file in [path.to_owned(), 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).map_err(Failure::into_error)
+}
+
+fn sql(context: &'static str, error: rusqlite::Error) -> Failure {
+    let corrupt = matches!(
+        error.sqlite_error_code(),
+        Some(rusqlite::ErrorCode::NotADatabase | rusqlite::ErrorCode::DatabaseCorrupt)
+    );
+    let error = storage_error(context, error);
+    if corrupt {
+        Failure::Corrupt(error)
+    } else {
+        Failure::Unavailable(error)
+    }
+}
+
+fn initialise(path: &Path) -> Result<(), Failure> {
     let parent = path
         .parent()
         .filter(|parent| !parent.as_os_str().is_empty())
@@ -82,7 +107,7 @@ fn initialise(path: &Path) -> Result<(), Error> {
             "CREATE TABLE IF NOT EXISTS records (key TEXT PRIMARY KEY, value BLOB NOT NULL)",
             [],
         )
-        .map_err(|error| storage_error("initialising the mentions database", error))?;
+        .map_err(|error| sql("initialising the mentions database", error))?;
     Ok(())
 }
 
@@ -98,42 +123,39 @@ fn connect(path: &Path) -> Result<Connection, Error> {
     Ok(connection)
 }
 
-fn load_snapshot(path: &Path) -> Result<Option<Snapshot>, Error> {
+fn load_snapshot(path: &Path) -> Result<Option<Snapshot>, Failure> {
     let mut connection = connect(path)?;
     let transaction = connection
         .transaction_with_behavior(TransactionBehavior::Deferred)
-        .map_err(|error| storage_error("starting a mentions snapshot", error))?;
+        .map_err(|error| sql("starting a mentions snapshot", error))?;
     let mut rows = BTreeMap::new();
     {
         let mut statement = transaction
             .prepare("SELECT key, value FROM records")
-            .map_err(|error| storage_error("reading the mentions snapshot", error))?;
+            .map_err(|error| sql("reading the mentions snapshot", error))?;
         let result = statement
             .query_map([], |row| {
                 Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
             })
-            .map_err(|error| storage_error("reading the mentions snapshot", error))?;
+            .map_err(|error| sql("reading the mentions snapshot", error))?;
         for row in result {
-            let (key, value) =
-                row.map_err(|error| storage_error("reading the mentions snapshot", error))?;
+            let (key, value) = row.map_err(|error| sql("reading the mentions snapshot", error))?;
             rows.insert(key, value);
         }
     }
     transaction
         .commit()
-        .map_err(|error| storage_error("finishing the mentions snapshot", error))?;
+        .map_err(|error| sql("finishing the mentions snapshot", error))?;
     if rows.is_empty() {
         return Ok(None);
     }
-    let exported = rows.remove(CIPHER_KEY).ok_or_else(|| {
-        Error::new(
-            ErrorKind::Store,
-            "mentions store is missing cipher metadata",
-        )
-    })?;
+    let missing = |message| Failure::Corrupt(Error::new(ErrorKind::Store, message));
+    let exported = rows
+        .remove(CIPHER_KEY)
+        .ok_or_else(|| missing("mentions store is missing cipher metadata"))?;
     let head = rows
         .remove(HEAD_KEY)
-        .ok_or_else(|| Error::new(ErrorKind::Store, "mentions store is missing head metadata"))?;
+        .ok_or_else(|| missing("mentions store is missing head metadata"))?;
     Ok(Some(Snapshot {
         exported,
         head,

Modified file: src/mentions/store/tests.rs+11 additions−10 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, SCHEMA, Write, row_bytes, storage_key};
+use super::{Account, Head, MentionsStore, OpenError, SCHEMA, Write, row_bytes, storage_key};
 use crate::StoreConfig;
 
 #[cfg(target_family = "wasm")]
@@ -34,7 +34,10 @@ fn config() -> StoreConfig {
 }
 
 fn set(key: &str, value: Value) -> Write {
-    Write::Set(key.to_owned(), value)
+    Write::Set(
+        key.to_owned(),
+        serde_json::value::to_raw_value(&value).unwrap(),
+    )
 }
 
 async fn open(config: &StoreConfig) -> MentionsStore {
@@ -360,7 +363,7 @@ async fn ciphertext_corruption_and_record_swaps_reset_the_store() {
 
 #[cfg_attr(not(target_family = "wasm"), tokio::test)]
 #[cfg_attr(target_family = "wasm", wasm_bindgen_test::wasm_bindgen_test)]
-async fn unsupported_schema_resets_the_store() {
+async fn a_newer_schema_is_left_alone() {
     let config = config();
     let mut store = open(&config).await;
     store
@@ -387,13 +390,11 @@ async fn unsupported_schema_resets_the_store() {
         )
         .await
         .unwrap();
-    assert!(
-        MentionsStore::open(&config, account(), MAX_BYTES)
-            .await
-            .unwrap()
-            .was_reset()
-    );
-    assert_ne!(store.backend.load().await.unwrap().unwrap().head, header);
+    assert!(matches!(
+        MentionsStore::open(&config, account(), MAX_BYTES).await,
+        Err(OpenError::Newer(_))
+    ));
+    assert_eq!(store.backend.load().await.unwrap().unwrap().head, header);
     drop(store);
     config.delete().await.unwrap();
 }

macltok

Commit 82d37c369846

Commit `82d37c369846`