macltok

public

Star0

Fork 0

Watch0

94 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `74befbc72baf`

tarball

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

20 files changed+1650-539

search: fix the review findings in the index, store and feed

The manifest was read before committed() and so saved one commit stale, losing the last flush and merge on restart and failing the whole load once a delete generation moved, so Changes now carries the manifest as it stands after the commit and a load purges blobs nothing points at. QueryParser read any word before a colon as a field and let a bare exclusion match everything, so queries are built from analysed words, phrases and exclusions with no field syntax. SimpleTokenizer made a CJK or Thai run one token and RemoveLongFilter counted bytes, so unspaced scripts now index singles and neighbouring pairs and the word limit is 64 chars. The head pass counted events indexed live this run as overlap and stopped on its first page, so it now only stops at what was indexed when the engine started.

Modified file: src/engine/context.rs+2 additions@@ -418,6 +418,7 @@ pub(crate) async fn preview(
     current_user_id: Option<&ruma::UserId>,
     values: &[Value],
     highlight: bool,
+    media_prefix: &str,
 ) -> Option<(TimelineEvent, Vec<MediaRegistration>)> {
     let mut page = Assembly::new(
         room,
@@ -425,6 +426,7 @@ pub(crate) async fn preview(
         values,
         &vec![highlight; values.len()],
         Some(0),
+        media_prefix,
     )
     .await;
     match page.take(0..1).pop()? {

Modified file: src/engine/mod.rs+7 additions−2 deletions@@ -422,8 +422,6 @@ impl Engine {
             .await
             // a missing user id is all it still refuses, and that was turned away above
             .map_err(|_| EngineError::Unauthenticated)?;
-        let (search, search_service) =
-            crate::search::Service::new(client.clone(), media.clone(), search, &config);
         let sync = Arc::new(
             SyncService::builder(client.clone())
                 .with_offline_mode()
@@ -444,6 +442,13 @@ impl Engine {
             config.command_capacity,
             config.event_capacity,
         );
+        let (search, search_service) = crate::search::Service::new(
+            client.clone(),
+            media.clone(),
+            search,
+            port.event_sender(),
+            &config,
+        );
         client.send_queue().enable_upload_progress(true);
         client.send_queue().enable_global_upload_progress(true);
         let (note_sender, notes) = mpsc::unbounded_channel();

Modified file: src/lib.rs+2 additions−2 deletions@@ -82,8 +82,8 @@ pub use repair_model::{
 };
 pub use ruma;
 pub use search_model::{
-    RoomSearchCoverage, SearchBackfill, SearchCommand, SearchCoverage, SearchHit, SearchMatch,
-    SearchPage, SearchScope, SearchSession,
+    RoomSearchCoverage, SearchBackfill, SearchCloseReason, SearchCommand, SearchCoverage,
+    SearchHit, SearchInMemory, SearchMatch, SearchPage, SearchScope, SearchSession,
 };
 #[doc(hidden)]
 pub use transport::PublicationAck;

Modified file: src/mirror.rs+7 additions@@ -162,6 +162,10 @@ pub enum StateChange {
         context_id: String,
         reason: EventContextCloseReason,
     },
+    SearchClosed {
+        search_id: String,
+        reason: crate::SearchCloseReason,
+    },
     Upload {
         transaction_id: OwnedTransactionId,
         upload: UploadState,
@@ -442,6 +446,9 @@ impl Mirror {
             Event::EventContextClosed { context_id, reason } => {
                 self.publish(|_| Some(StateChange::EventContextClosed { context_id, reason }));
             }
+            Event::SearchClosed { search_id, reason } => {
+                self.publish(|_| Some(StateChange::SearchClosed { search_id, reason }));
+            }
             Event::ThreadListReset { room_id, entries } => self.publish(|state| {
                 let history = state
                     .thread_lists

Modified file: src/model.rs+6 additions@@ -983,6 +983,12 @@ pub enum Event {
         room_id: OwnedRoomId,
         state: HistoryState,
     },
+    /// a search closed by something other than its own `Close`. its id answers
+    /// `SearchNotOpen` from now on
+    SearchClosed {
+        search_id: String,
+        reason: crate::SearchCloseReason,
+    },
     /// nothing more comes for it until another `ThreadListReset`
     ThreadListClosed {
         room_id: OwnedRoomId,

Modified file: src/search/feed.rs+58 additions−53 deletions@@ -3,7 +3,10 @@ use std::collections::{HashSet, VecDeque};
 use ruma::events::room::message::sanitize::remove_plain_reply_fallback;
 use serde_json::Value;
 
-use super::{MessageIndex, index::Doc};
+use super::{
+    MessageIndex,
+    index::{Doc, provisional_key},
+};
 use crate::normalise::poll_question;
 
 const REMEMBERED_REDACTIONS: usize = 10_000;
@@ -45,6 +48,8 @@ pub(crate) enum Observed {
 pub(crate) enum Effect {
     Indexed(u64),
     Removed,
+    /// written, but nothing searchable changed
+    Held,
     /// an edit was redacted out from under its original, which has to be read again
     Refetch(String),
     Nothing,
@@ -110,7 +115,7 @@ fn edit_of(value: &Value, event_type: &str) -> Option<Edit> {
             .to_owned(),
         sender: value.get("sender")?.as_str()?.to_owned(),
         ts: value.get("origin_server_ts")?.as_u64()?,
-        text: text_of(event_type, content.get("m.new_content")?)?,
+        text: text_of(event_type, content.get("m.new_content")?).unwrap_or_default(),
     })
 }
 
@@ -169,8 +174,8 @@ fn newer(ts: u64, id: &str, than: &Doc) -> bool {
 }
 
 /// what `observed` does to `room`'s part of the index. edits land on their original under
-/// its key, the latest by its own sender wins, and one arriving first holds the key
-/// provisionally until the original turns up
+/// its key, the latest by its own sender wins, and one that arrives first waits under
+/// `provisional_key` until the original turns up
 pub(crate) fn apply(
     index: &mut dyn MessageIndex,
     redactions: &mut Redactions,
@@ -205,21 +210,18 @@ pub(crate) fn apply(
                 doc.content_ts = edit.ts;
                 doc.text = edit.text;
             }
-            match index.get(&key) {
-                Some(existing) if !existing.provisional => {
-                    if !newer(doc.content_ts, &doc.content, &existing) {
-                        return Effect::Nothing;
-                    }
-                }
-                Some(existing) => {
-                    if existing.sender == sender && !newer(doc.content_ts, &doc.content, &existing)
-                    {
-                        doc.content = existing.content;
-                        doc.content_ts = existing.content_ts;
-                        doc.text = existing.text;
-                    }
-                }
-                None => {}
+            if let Some(waiting) = index.get(&provisional_key(&key, &sender))
+                && newer(waiting.content_ts, &waiting.content, &doc)
+            {
+                doc.content = waiting.content;
+                doc.content_ts = waiting.content_ts;
+                doc.text = waiting.text;
+            }
+            let held = index.remove_provisional(&key);
+            if let Some(existing) = index.get(&key)
+                && !newer(doc.content_ts, &doc.content, &existing)
+            {
+                return if held { Effect::Held } else { Effect::Nothing };
             }
             index.put(doc);
             Effect::Indexed(ts)
@@ -228,40 +230,51 @@ pub(crate) fn apply(
             if redactions.contains(&edit.target) || redactions.contains(&edit.id) {
                 return Effect::Nothing;
             }
-            let doc = match index.get(&edit.target) {
-                Some(existing) => {
-                    if existing.sender != edit.sender || !newer(edit.ts, &edit.id, &existing) {
-                        return Effect::Nothing;
-                    }
-                    Doc {
-                        content: edit.id,
-                        content_ts: edit.ts,
-                        text: edit.text,
-                        ..existing
-                    }
+            if index
+                .by_content(&edit.target)
+                .is_some_and(|doc| doc.key != edit.target)
+            {
+                return Effect::Nothing;
+            }
+            if let Some(existing) = index.get(&edit.target) {
+                if existing.sender != edit.sender || !newer(edit.ts, &edit.id, &existing) {
+                    return Effect::Nothing;
                 }
-                None => Doc {
-                    key: edit.target,
-                    room: room.to_owned(),
-                    sender: edit.sender,
-                    ts: edit.ts,
+                let ts = existing.ts;
+                index.put(Doc {
                     content: edit.id,
                     content_ts: edit.ts,
                     text: edit.text,
-                    provisional: true,
-                },
-            };
-            let ts = doc.ts;
-            index.put(doc);
-            Effect::Indexed(ts)
+                    ..existing
+                });
+                return Effect::Indexed(ts);
+            }
+            let key = provisional_key(&edit.target, &edit.sender);
+            if let Some(waiting) = index.get(&key)
+                && !newer(edit.ts, &edit.id, &waiting)
+            {
+                return Effect::Nothing;
+            }
+            index.put(Doc {
+                key,
+                room: room.to_owned(),
+                sender: edit.sender,
+                ts: edit.ts,
+                content: edit.id,
+                content_ts: edit.ts,
+                text: edit.text,
+                provisional: true,
+            });
+            Effect::Held
         }
-        Observed::Redaction { target } => {
+        Observed::Redaction { target } | Observed::Redacted { key: target } => {
             redactions.insert(target.clone());
-            if index.remove(&target) {
+            let removed = index.remove(&target);
+            if index.remove_provisional(&target) || removed {
                 return Effect::Removed;
             }
             match index.by_content(&target) {
-                Some(doc) => {
+                Some(doc) if doc.key != target => {
                     index.remove(&doc.key);
                     if doc.provisional {
                         Effect::Removed
@@ -269,15 +282,7 @@ pub(crate) fn apply(
                         Effect::Refetch(doc.key)
                     }
                 }
-                None => Effect::Nothing,
-            }
-        }
-        Observed::Redacted { key } => {
-            redactions.insert(key.clone());
-            if index.remove(&key) {
-                Effect::Removed
-            } else {
-                Effect::Nothing
+                _ => Effect::Nothing,
             }
         }
         Observed::Encrypted { .. } | Observed::Other => Effect::Nothing,

Modified file: src/search/feed_tests.rs+67 additions−2 deletions@@ -1,7 +1,7 @@
 use serde_json::json;
 
 use super::*;
-use crate::search::index::KeywordIndex;
+use crate::search::index::{KeywordIndex, provisional_key};
 
 const ROOM: &str = "!room:x";
 
@@ -60,9 +60,10 @@ fn an_edit_before_its_original_keeps_the_edit_and_takes_the_original_timestamp()
         &mut redactions,
         edit("$e", "@a:x", 20, "$o", "fixed words"),
     );
-    let provisional = index.get("$o").unwrap();
+    let provisional = index.get(&provisional_key("$o", "@a:x")).unwrap();
     assert!(provisional.provisional);
     assert_eq!(provisional.ts, 20);
+    assert!(keys(&mut index, "fixed").is_empty());
     feed(
         &mut index,
         &mut redactions,
@@ -185,3 +186,67 @@ fn reply_fallbacks_are_not_indexed() {
     assert_eq!(keys(&mut index, "answer"), vec!["$r"]);
     assert!(keys(&mut index, "quoted").is_empty());
 }
+
+#[test]
+fn a_strangers_early_edit_cannot_shut_out_the_author() {
+    let mut index = KeywordIndex::new(false);
+    let mut redactions = Redactions::default();
+    feed(
+        &mut index,
+        &mut redactions,
+        edit("$h", "@b:x", 30, "$o", "hijack"),
+    );
+    feed(
+        &mut index,
+        &mut redactions,
+        edit("$e", "@a:x", 20, "$o", "genuine edit"),
+    );
+    feed(
+        &mut index,
+        &mut redactions,
+        message("$o", "@a:x", 10, "original"),
+    );
+    assert_eq!(keys(&mut index, "genuine"), vec!["$o"]);
+    assert!(keys(&mut index, "hijack").is_empty());
+    assert!(index.get(&provisional_key("$o", "@b:x")).is_none());
+}
+
+#[test]
+fn an_edit_of_an_edit_is_ignored() {
+    let mut index = KeywordIndex::new(false);
+    let mut redactions = Redactions::default();
+    feed(
+        &mut index,
+        &mut redactions,
+        message("$o", "@a:x", 10, "original"),
+    );
+    feed(
+        &mut index,
+        &mut redactions,
+        edit("$e", "@a:x", 20, "$o", "first"),
+    );
+    assert_eq!(
+        feed(
+            &mut index,
+            &mut redactions,
+            edit("$ee", "@a:x", 30, "$e", "nested")
+        ),
+        Effect::Nothing
+    );
+    assert!(index.get(&provisional_key("$e", "@a:x")).is_none());
+}
+
+#[test]
+fn an_edit_to_unreadable_content_clears_the_old_text() {
+    let mut index = KeywordIndex::new(false);
+    let mut redactions = Redactions::default();
+    feed(
+        &mut index,
+        &mut redactions,
+        message("$o", "@a:x", 10, "secret plans"),
+    );
+    let mut blank = edit("$e", "@a:x", 20, "$o", "x");
+    blank["content"]["m.new_content"] = json!({ "msgtype": "m.unknown" });
+    feed(&mut index, &mut redactions, blank);
+    assert!(keys(&mut index, "secret").is_empty());
+}

Modified file: src/search/index.rs+127 additions−121 deletions@@ -8,25 +8,25 @@ use tantivy::{
     directory::{Directory, OwnedBytes, RamDirectory},
     fastfield::{AliveBitSet, write_alive_bitset},
     indexer::{SingleSegmentIndexWriter, merge_filtered_segments},
-    query::{BooleanQuery, Occur, Query, QueryParser, TermSetQuery},
+    query::{BooleanQuery, Occur, Query, TermSetQuery},
     schema::{
         FAST, Field, IndexRecordOption, STORED, STRING, Schema, TextFieldIndexing, TextOptions,
         Value,
     },
-    tokenizer::{
-        AsciiFoldingFilter, LowerCaser, RemoveLongFilter, SimpleTokenizer, TextAnalyzer,
-        TokenStream,
-    },
 };
 use tantivy_common::BitSet;
 
-use super::MessageIndex;
+use super::{
+    MessageIndex,
+    text::{TOKENIZER, analyzer, build_query},
+};
 use crate::{Error, ErrorKind};
 
-const TOKENIZER: &str = "macltok";
 const WRITER_BUDGET: usize = 4_000_000;
 /// segments of a size class merge once there are this many of them
 const MERGE_FACTOR: usize = 6;
+/// decompressed docstore blocks a searcher keeps per segment, tantivy's default is 100
+const DOCSTORE_BLOCKS: usize = 2;
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(crate) struct Doc {
@@ -40,10 +40,16 @@ pub(crate) struct Doc {
     pub(crate) content_ts: u64,
     /// only kept for provisional docs, read back empty otherwise
     pub(crate) text: String,
-    /// an edit seen before its original
+    /// an edit seen before its original, under `provisional_key`. it's kept out of results,
+    /// since its timestamp moves to the original's once that turns up
     pub(crate) provisional: bool,
 }
 
+/// one per target and edit sender, so an edit from anyone else can't crowd out the author's
+pub(crate) fn provisional_key(target: &str, sender: &str) -> String {
+    format!("{target}\n{sender}")
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(crate) struct Hit {
     pub(crate) key: String,
@@ -63,6 +69,8 @@ struct SegmentRecord {
 pub(crate) struct Changes {
     pub(crate) puts: Vec<(String, Vec<u8>)>,
     pub(crate) deletes: Vec<String>,
+    /// the index as it stands once these land, which is what has to be saved with them
+    pub(crate) manifest: serde_json::Value,
     files: Vec<u64>,
     dels: Vec<(u64, u64)>,
     removed: Vec<u64>,
@@ -79,6 +87,7 @@ struct Fields {
     tie: Field,
     content_ts: Field,
     pending: Field,
+    target: Field,
 }
 
 struct Segment {
@@ -103,7 +112,6 @@ pub(crate) struct KeywordIndex {
     schema: Schema,
     fields: Fields,
     settings: IndexSettings,
-    parser: QueryParser,
     segments: Vec<Segment>,
     buffer: HashMap<String, Doc>,
     next_id: u64,
@@ -112,31 +120,6 @@ pub(crate) struct KeywordIndex {
     removed: Vec<(u64, bool, u64)>,
 }
 
-pub(crate) fn analyzer() -> TextAnalyzer {
-    TextAnalyzer::builder(SimpleTokenizer::default())
-        .filter(RemoveLongFilter::limit(40))
-        .filter(LowerCaser)
-        .filter(AsciiFoldingFilter)
-        .build()
-}
-
-/// the byte ranges in `text` whose analysed token is one of `terms`
-pub(crate) fn match_ranges(text: &str, terms: &[String]) -> Vec<(usize, usize)> {
-    if terms.is_empty() {
-        return Vec::new();
-    }
-    let mut analyzer = analyzer();
-    let mut stream = analyzer.token_stream(text);
-    let mut ranges = Vec::new();
-    while stream.advance() {
-        let token = stream.token();
-        if terms.iter().any(|term| *term == token.text) {
-            ranges.push((token.offset_from, token.offset_to));
-        }
-    }
-    ranges
-}
-
 fn schema() -> (Schema, Fields) {
     let mut builder = Schema::builder();
     let text_options = TextOptions::default().set_indexing_options(
@@ -154,6 +137,7 @@ fn schema() -> (Schema, Fields) {
         tie: builder.add_u64_field("tie", FAST),
         content_ts: builder.add_u64_field("content_ts", STORED),
         pending: builder.add_text_field("pending", STORED),
+        target: builder.add_text_field("target", STRING),
     };
     (builder.build(), fields)
 }
@@ -185,6 +169,7 @@ fn open_segment(id: u64, raw: RamDirectory, max_doc_hint: Option<u32>) -> Result
     let reader = index
         .reader_builder()
         .reload_policy(ReloadPolicy::Manual)
+        .doc_store_cache_num_blocks(DOCSTORE_BLOCKS)
         .try_into()
         .map_err(|error| tantivy_error("reading a search segment", error))?;
     let searcher = reader.searcher();
@@ -212,24 +197,25 @@ fn open_segment(id: u64, raw: RamDirectory, max_doc_hint: Option<u32>) -> Result
     })
 }
 
-fn pack(files: &[(String, Vec<u8>)]) -> Vec<u8> {
+fn pack(files: &[(String, OwnedBytes)]) -> Vec<u8> {
     let mut packed = Vec::with_capacity(
         files
             .iter()
             .map(|(name, bytes)| name.len() + bytes.len() + 12)
-            .sum(),
+            .sum::<usize>()
+            + 68,
     );
     packed.extend_from_slice(&(files.len() as u32).to_le_bytes());
     for (name, bytes) in files {
         packed.extend_from_slice(&(name.len() as u32).to_le_bytes());
         packed.extend_from_slice(name.as_bytes());
         packed.extend_from_slice(&(bytes.len() as u64).to_le_bytes());
-        packed.extend_from_slice(bytes);
+        packed.extend_from_slice(bytes.as_slice());
     }
     packed
 }
 
-fn unpack(mut bytes: &[u8]) -> Option<Vec<(String, Vec<u8>)>> {
+fn unpack(mut bytes: &[u8]) -> Option<Vec<(String, &[u8])>> {
     fn take<'a>(bytes: &mut &'a [u8], len: usize) -> Option<&'a [u8]> {
         if bytes.len() < len {
             return None;
@@ -247,7 +233,7 @@ fn unpack(mut bytes: &[u8]) -> Option<Vec<(String, Vec<u8>)>> {
             .to_owned();
         let len =
             usize::try_from(u64::from_le_bytes(take(&mut bytes, 8)?.try_into().ok()?)).ok()?;
-        files.push((name, take(&mut bytes, len)?.to_vec()));
+        files.push((name, take(&mut bytes, len)?));
     }
     bytes.is_empty().then_some(files)
 }
@@ -260,18 +246,20 @@ pub(crate) fn deletes_key(id: u64, generation: u64) -> String {
     format!("del/{id}/{generation}")
 }
 
+fn manifest(records: impl Iterator<Item = SegmentRecord>, next_id: u64) -> serde_json::Value {
+    let segments: Vec<_> = records
+        .map(|record| serde_json::json!([record.id, record.del_gen]))
+        .collect();
+    serde_json::json!({ "segments": segments, "next": next_id })
+}
+
 impl KeywordIndex {
     pub(crate) fn new(persistent: bool) -> Self {
         let (schema, fields) = schema();
-        let parser_index = Index::create_in_ram(schema.clone());
-        parser_index.tokenizers().register(TOKENIZER, analyzer());
-        let mut parser = QueryParser::for_index(&parser_index, vec![fields.text]);
-        parser.set_conjunction_by_default();
         Self {
             schema,
             fields,
             settings: settings(),
-            parser,
             segments: Vec::new(),
             buffer: HashMap::new(),
             next_id: 1,
@@ -280,7 +268,8 @@ impl KeywordIndex {
         }
     }
 
-    /// `blob` hands back what an earlier commit put, any of it missing or unreadable fails the load
+    /// `blob` hands over what an earlier commit put, once each, and any of it missing or
+    /// unreadable fails the load
     pub(crate) fn load(
         manifest: &serde_json::Value,
         mut blob: impl FnMut(&str) -> Option<Vec<u8>>,
@@ -307,14 +296,13 @@ impl KeywordIndex {
             )
         };
         for record in &records {
-            let files = blob(&segment_key(record.id))
-                .and_then(|bytes| unpack(&bytes))
-                .ok_or_else(missing)?;
+            let packed = blob(&segment_key(record.id)).ok_or_else(missing)?;
             let raw = RamDirectory::create();
-            for (name, bytes) in files {
-                raw.atomic_write(&PathBuf::from(name), &bytes)
+            for (name, bytes) in unpack(&packed).ok_or_else(missing)? {
+                raw.atomic_write(&PathBuf::from(name), bytes)
                     .map_err(|error| tantivy_error("loading a search segment", error))?;
             }
+            drop(packed);
             let mut segment = open_segment(record.id, raw, None)?;
             if segment.index.schema() != index.schema {
                 return Err(Error::new(
@@ -345,15 +333,17 @@ impl KeywordIndex {
         Ok(index)
     }
 
-    fn records(&self) -> Vec<SegmentRecord> {
-        self.segments
-            .iter()
-            .filter(|segment| segment.persisted)
-            .map(|segment| SegmentRecord {
-                id: segment.id,
-                del_gen: segment.del_gen,
-            })
-            .collect()
+    /// the blob names what is persisted points at
+    #[cfg(test)]
+    pub(crate) fn referenced(&self) -> Vec<String> {
+        let mut keys = Vec::new();
+        for segment in self.segments.iter().filter(|segment| segment.persisted) {
+            keys.push(segment_key(segment.id));
+            if segment.del_gen > 0 {
+                keys.push(deletes_key(segment.id, segment.del_gen));
+            }
+        }
+        keys
     }
 
     fn document(&self, doc: &Doc) -> TantivyDocument {
@@ -363,12 +353,18 @@ impl KeywordIndex {
         document.add_text(fields.room, &doc.room);
         document.add_text(fields.sender, &doc.sender);
         document.add_text(fields.content, &doc.content);
-        document.add_text(fields.text, &doc.text);
         document.add_u64(fields.ts, doc.ts);
         document.add_u64(fields.tie, tie(&doc.key));
         document.add_u64(fields.content_ts, doc.content_ts);
         if doc.provisional {
             document.add_text(fields.pending, &doc.text);
+            let target = doc
+                .key
+                .split_once('\n')
+                .map_or(doc.key.as_str(), |(target, _)| target);
+            document.add_text(fields.target, target);
+        } else {
+            document.add_text(fields.text, &doc.text);
         }
         document
     }
@@ -426,6 +422,25 @@ impl KeywordIndex {
         })
     }
 
+    /// tombstones every live doc in a segment whose `field` holds `value`
+    fn remove_where(&mut self, field: Field, value: &str) -> bool {
+        let mut removed = false;
+        for position in 0..self.segments.len() {
+            let docs = self.find(&self.segments[position], field, value);
+            if docs.is_empty() {
+                continue;
+            }
+            let segment = &mut self.segments[position];
+            let alive = Arc::make_mut(&mut segment.alive);
+            for doc in docs {
+                alive.remove(doc);
+            }
+            segment.deletes_dirty = true;
+            removed = true;
+        }
+        removed
+    }
+
     fn flush_buffer(&mut self) -> Result<(), Error> {
         if self.buffer.is_empty() {
             return Ok(());
@@ -457,16 +472,22 @@ impl KeywordIndex {
         live.max(1).ilog(MERGE_FACTOR)
     }
 
+    fn forget(&mut self, segment: &Segment) {
+        if self.persistent {
+            self.removed
+                .push((segment.id, segment.persisted, segment.del_gen));
+        }
+    }
+
     /// merges a size class once it fills up, and rewrites a segment that's mostly deleted
     fn compact(&mut self) -> Result<(), Error> {
-        self.segments.retain(|segment| {
-            let empty = segment.live() == 0;
-            if empty && self.persistent {
-                self.removed
-                    .push((segment.id, segment.persisted, segment.del_gen));
-            }
-            !empty
-        });
+        let (empty, kept): (Vec<_>, Vec<_>) = std::mem::take(&mut self.segments)
+            .into_iter()
+            .partition(|segment| segment.live() == 0);
+        self.segments = kept;
+        for segment in &empty {
+            self.forget(segment);
+        }
         loop {
             let mut classes: HashMap<u32, Vec<usize>> = HashMap::new();
             for (position, segment) in self.segments.iter().enumerate() {
@@ -504,26 +525,26 @@ impl KeywordIndex {
             let mut bytes = Vec::new();
             write_alive_bitset(&segment.alive, &mut bytes)
                 .map_err(|error| tantivy_error("merging search segments", error))?;
+            let bytes = OwnedBytes::new(bytes);
             for tantivy_segment in searchable {
                 inputs.push(tantivy_segment);
-                alive.push(Some(AliveBitSet::open(OwnedBytes::new(bytes.clone()))));
+                alive.push(Some(AliveBitSet::open(bytes.clone())));
             }
         }
         let raw = RamDirectory::create();
         merge_filtered_segments(&inputs, self.settings.clone(), alive, raw.clone())
             .map_err(|error| tantivy_error("merging search segments", error))?;
+        drop(inputs);
         let merged = open_segment(self.next_id, raw, None)?;
         self.next_id += 1;
-        let mut position = 0;
-        self.segments.retain(|segment| {
-            let merging = positions.contains(&position);
-            position += 1;
-            if merging && self.persistent {
-                self.removed
-                    .push((segment.id, segment.persisted, segment.del_gen));
-            }
-            !merging
-        });
+        let (merging, kept): (Vec<_>, Vec<_>) = std::mem::take(&mut self.segments)
+            .into_iter()
+            .enumerate()
+            .partition(|(position, _)| positions.contains(position));
+        self.segments = kept.into_iter().map(|(_, segment)| segment).collect();
+        for (_, segment) in &merging {
+            self.forget(segment);
+        }
         self.segments.push(merged);
         Ok(())
     }
@@ -548,10 +569,7 @@ impl KeywordIndex {
                 .map_err(|error| tantivy_error("reading a search segment", error))?
                 .read_bytes()
                 .map_err(|error| tantivy_error("reading a search segment", error))?;
-            files.push((
-                name.to_string_lossy().into_owned(),
-                bytes.as_slice().to_vec(),
-            ));
+            files.push((name.to_string_lossy().into_owned(), bytes));
         }
         Ok(pack(&files))
     }
@@ -578,22 +596,19 @@ impl MessageIndex for KeywordIndex {
     }
 
     fn remove(&mut self, key: &str) -> bool {
-        let mut removed = self.buffer.remove(key).is_some();
+        let buffered = self.buffer.remove(key).is_some();
         let key_field = self.fields.key;
-        for position in 0..self.segments.len() {
-            let docs = self.find(&self.segments[position], key_field, key);
-            if docs.is_empty() {
-                continue;
-            }
-            let segment = &mut self.segments[position];
-            let alive = Arc::make_mut(&mut segment.alive);
-            for doc in docs {
-                alive.remove(doc);
-            }
-            segment.deletes_dirty = true;
-            removed = true;
-        }
-        removed
+        self.remove_where(key_field, key) || buffered
+    }
+
+    fn remove_provisional(&mut self, target: &str) -> bool {
+        let prefix = format!("{target}\n");
+        let before = self.buffer.len();
+        self.buffer
+            .retain(|key, doc| !(doc.provisional && key.starts_with(&prefix)));
+        let buffered = self.buffer.len() != before;
+        let target_field = self.fields.target;
+        self.remove_where(target_field, target) || buffered
     }
 
     fn buffered(&self) -> usize {
@@ -607,16 +622,8 @@ impl MessageIndex for KeywordIndex {
         before: Option<(u64, u64)>,
         limit: usize,
     ) -> Result<(Vec<Hit>, Vec<String>), Error> {
+        let (text_query, terms) = build_query(self.fields.text, query)?;
         self.flush_buffer()?;
-        let (text_query, _) = self.parser.parse_query_lenient(query);
-        let mut terms = Vec::new();
-        text_query.query_terms(&mut |term, _| {
-            if term.field() == self.fields.text
-                && let Some(text) = term.value().as_str()
-            {
-                terms.push(text.to_owned());
-            }
-        });
         let query: Box<dyn Query> = match rooms {
             None => text_query,
             Some(rooms) => Box::new(BooleanQuery::new(vec![
@@ -671,8 +678,6 @@ impl MessageIndex for KeywordIndex {
                 tie,
             });
         }
-        terms.sort();
-        terms.dedup();
         Ok((hits, terms))
     }
 
@@ -693,6 +698,7 @@ impl MessageIndex for KeywordIndex {
             }
             changes.removed.push(*id);
         }
+        let mut records = Vec::with_capacity(self.segments.len());
         for segment in &self.segments {
             if !segment.persisted {
                 changes
@@ -700,22 +706,26 @@ impl MessageIndex for KeywordIndex {
                     .push((segment_key(segment.id), Self::segment_files(segment)?));
                 changes.files.push(segment.id);
             }
+            let mut del_gen = segment.del_gen;
             if segment.deletes_dirty && segment.live() < segment.max_doc as usize {
-                let generation = segment.del_gen + 1;
+                del_gen = segment.del_gen + 1;
                 let mut bytes = Vec::new();
                 write_alive_bitset(&segment.alive, &mut bytes)
                     .map_err(|error| tantivy_error("saving search deletes", error))?;
-                changes
-                    .puts
-                    .push((deletes_key(segment.id, generation), bytes));
+                changes.puts.push((deletes_key(segment.id, del_gen), bytes));
                 if segment.del_gen > 0 {
                     changes
                         .deletes
                         .push(deletes_key(segment.id, segment.del_gen));
                 }
-                changes.dels.push((segment.id, generation));
+                changes.dels.push((segment.id, del_gen));
             }
+            records.push(SegmentRecord {
+                id: segment.id,
+                del_gen,
+            });
         }
+        changes.manifest = manifest(records.into_iter(), self.next_id);
         Ok(changes)
     }
 
@@ -733,13 +743,9 @@ impl MessageIndex for KeywordIndex {
         }
     }
 
-    fn manifest(&self) -> serde_json::Value {
-        let segments: Vec<_> = self
-            .records()
-            .into_iter()
-            .map(|record| serde_json::json!([record.id, record.del_gen]))
-            .collect();
-        serde_json::json!({ "segments": segments, "next": self.next_id })
+    fn detach(&mut self) {
+        self.persistent = false;
+        self.removed.clear();
     }
 }
 

Modified file: src/search/index_tests.rs+59 additions−41 deletions@@ -19,16 +19,29 @@ fn keys(hits: &[Hit]) -> Vec<&str> {
     hits.iter().map(|hit| hit.key.as_str()).collect()
 }
 
-fn commit(index: &mut KeywordIndex, blobs: &mut BTreeMap<String, Vec<u8>>) -> serde_json::Value {
+/// the order `Service::commit` uses, the manifest saved with the blobs before `committed`
+fn commit(index: &mut KeywordIndex, store: &mut BTreeMap<String, Vec<u8>>) {
     let changes = index.changes().unwrap();
     for key in &changes.deletes {
-        blobs.remove(key);
+        store.remove(key);
     }
     for (key, value) in &changes.puts {
-        blobs.insert(key.clone(), value.clone());
+        store.insert(key.clone(), value.clone());
     }
+    store.insert("meta".to_owned(), changes.manifest.to_string().into_bytes());
     index.committed(&changes);
-    index.manifest()
+}
+
+fn reload(store: &BTreeMap<String, Vec<u8>>) -> KeywordIndex {
+    let manifest: serde_json::Value = serde_json::from_slice(&store["meta"]).unwrap();
+    KeywordIndex::load(&manifest, |key| store.get(key).cloned()).unwrap()
+}
+
+fn found(index: &mut KeywordIndex, query: &str) -> Vec<String> {
+    let (hits, _) = index.search(query, None, None, 100).unwrap();
+    let mut found: Vec<_> = keys(&hits).into_iter().map(str::to_owned).collect();
+    found.sort();
+    found
 }
 
 #[test]
@@ -57,11 +70,8 @@ fn a_replaced_doc_is_found_once_by_its_new_text() {
     index.put(doc("$a", "!r:x", 1, "before"));
     index.maintain().unwrap();
     index.put(doc("$a", "!r:x", 1, "after"));
-    assert!(index.search("before", None, None, 10).unwrap().0.is_empty());
-    assert_eq!(
-        keys(&index.search("after", None, None, 10).unwrap().0),
-        vec!["$a"]
-    );
+    assert!(found(&mut index, "before").is_empty());
+    assert_eq!(found(&mut index, "after"), vec!["$a"]);
 }
 
 #[test]
@@ -83,50 +93,58 @@ fn rooms_filter_the_results() {
     );
 }
 
-/// a lost delete brings a redacted message back, and a lost segment loses messages
 #[test]
-fn deletes_and_merges_survive_a_reload() {
-    let mut blobs = BTreeMap::new();
+fn provisional_docs_stay_out_of_results() {
+    let mut index = KeywordIndex::new(false);
+    let mut pending = doc(&provisional_key("$o", "@a:x"), "!r:x", 5, "early edit");
+    pending.provisional = true;
+    index.put(pending);
+    assert!(found(&mut index, "early").is_empty());
+    assert!(index.remove_provisional("$o"));
+    assert!(index.get(&provisional_key("$o", "@a:x")).is_none());
+}
+
+/// the manifest used to be read before `committed`, one commit stale, so a restart lost
+/// the last flush and merge, and a bumped delete generation failed the whole load
+#[test]
+fn every_commit_reloads_to_what_was_committed() {
+    let mut store = BTreeMap::new();
     let mut index = KeywordIndex::new(true);
-    for ts in 0..(MERGE_FACTOR as u64 + 2) {
+    let count = MERGE_FACTOR as u64 + 2;
+    for ts in 0..count {
         index.put(doc(&format!("$e{ts}"), "!r:x", ts, "kept"));
-        commit(&mut index, &mut blobs);
+        commit(&mut index, &mut store);
+        assert_eq!(found(&mut reload(&store), "kept").len(), ts as usize + 1);
     }
-    index.put(doc("$gone", "!r:x", 100, "kept"));
-    commit(&mut index, &mut blobs);
-    index.remove("$gone");
     index.remove("$e0");
-    let manifest = commit(&mut index, &mut blobs);
-    let mut reloaded = KeywordIndex::load(&manifest, |key| blobs.get(key).cloned()).unwrap();
-    let (hits, _) = reloaded.search("kept", None, None, 50).unwrap();
-    let mut found = keys(&hits);
-    found.sort();
-    let mut expected: Vec<String> = (1..(MERGE_FACTOR as u64 + 2))
-        .map(|ts| format!("$e{ts}"))
-        .collect();
-    expected.sort();
-    assert_eq!(
-        found,
-        expected.iter().map(String::as_str).collect::<Vec<_>>()
-    );
-    let referenced: Vec<_> = blobs.keys().filter(|key| key.starts_with("seg/")).collect();
-    assert!(
-        referenced.len() < MERGE_FACTOR,
-        "merged segments were not dropped: {referenced:?}"
-    );
+    commit(&mut index, &mut store);
+    index.remove("$e1");
+    commit(&mut index, &mut store);
+    index.put(doc("$e2", "!r:x", 2, "edited"));
+    commit(&mut index, &mut store);
+    let mut reloaded = reload(&store);
+    let expected: Vec<String> = (3..count).map(|ts| format!("$e{ts}")).collect();
+    assert_eq!(found(&mut reloaded, "kept"), expected);
+    assert_eq!(found(&mut reloaded, "edited"), vec!["$e2"]);
+    let mut referenced = reloaded.referenced();
+    referenced.push("meta".to_owned());
+    referenced.sort();
+    let stored: Vec<_> = store.keys().cloned().collect();
+    assert_eq!(stored, referenced, "a commit left blobs nothing points at");
 }
 
 #[test]
-fn match_ranges_follow_the_analyser() {
-    assert_eq!(
-        match_ranges("Café, CAFE and cafés", &["cafe".to_owned()]),
-        vec![(0, 5), (7, 11)]
-    );
+fn unspaced_text_matches_inside_a_run() {
+    let mut index = KeywordIndex::new(false);
+    index.put(doc("$a", "!r:x", 1, "明日東京都庁に行きます"));
+    assert_eq!(found(&mut index, "東京"), vec!["$a"]);
+    assert_eq!(found(&mut index, "\"都庁\""), vec!["$a"]);
+    assert!(found(&mut index, "大阪").is_empty());
 }
 
 #[test]
 fn packed_segments_reject_truncation() {
-    let packed = pack(&[("a".to_owned(), vec![1, 2, 3])]);
+    let packed = pack(&[("a".to_owned(), OwnedBytes::new(vec![1, 2, 3]))]);
     assert!(unpack(&packed).is_some());
     assert!(unpack(&packed[..packed.len() - 1]).is_none());
 }

Modified file: src/search/mod.rs+546 additions−221 deletions@@ -3,35 +3,38 @@ mod feed;
 mod index;
 #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
 pub(crate) mod store;
+mod text;
 
 use std::{
-    collections::{BTreeMap, HashMap, HashSet, VecDeque},
+    collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque},
     sync::Arc,
     time::Duration,
 };
 
 use futures_util::{
     StreamExt,
-    future::join_all,
+    future::Abortable,
     stream::{AbortHandle, FuturesUnordered, SelectAll, abortable},
 };
 use matrix_sdk::{
     Client, Room, RoomState, SendOutsideWasm,
     deserialized_responses::TimelineEvent as SdkTimelineEvent,
     event_cache::{RoomEventCacheSubscriber, RoomEventCacheUpdate},
-    room::MessagesOptions,
+    room::{Messages, MessagesOptions},
 };
 use matrix_sdk_base::RoomInfoNotableUpdateReasons;
 use matrix_sdk_ui::eyeball_im::VectorDiff;
 use ruma::{
     OwnedRoomId, OwnedUserId, UInt,
+    api::client::filter::RoomEventFilter,
     events::{
-        ignored_user_list::IgnoredUserListEventContent,
+        ignored_user_list::IgnoredUserListEventContent, relation::RelationType,
         room::encrypted::OriginalSyncRoomEncryptedEvent,
     },
     serde::Raw,
 };
 use serde_json::Value;
+use sha2::{Digest, Sha256};
 use tokio::sync::{Mutex, broadcast, mpsc, oneshot};
 
 use self::{
@@ -39,24 +42,41 @@ use self::{
     index::{Changes, Doc, Hit, KeywordIndex},
 };
 use crate::{
-    CommandOutput, EngineConfig, Error, ErrorKind, RoomSearchCoverage, SearchBackfill,
-    SearchCommand, SearchCoverage, SearchHit, SearchMatch, SearchPage, SearchScope, SearchSession,
+    CommandOutput, EngineConfig, Error, ErrorKind, Event, RoomSearchCoverage, SearchBackfill,
+    SearchCloseReason, SearchCommand, SearchCoverage, SearchHit, SearchInMemory, SearchMatch,
+    SearchPage, SearchScope, SearchSession,
     engine::{MediaService, SearchStore, context, executor::EngineBoxFuture},
     normalise::{MediaRegistration, MediaScope},
+    transport::EventSender,
 };
 
 const MAX_LIMIT: u16 = 50;
 /// a consumer that never closes its searches still holds at most this many
 const MAX_SEARCHES: usize = 8;
+/// hits read back at once while a page is put together
+const PREVIEWS_AT_ONCE: usize = 8;
+const DECRYPTS_AT_ONCE: usize = 8;
 const COMMIT_AFTER_MS: u64 = 5_000;
 const COMMIT_BUFFERED: usize = 2_000;
 #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
 const COMMIT_RETRY_MS: u64 = 60_000;
+#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+const COMMIT_ATTEMPTS: u32 = 5;
 const IDLE_SUBSCRIPTION_MS: u64 = 5 * 60_000;
 const BACKFILL_RETRY_MS: u64 = 60_000;
 const BACKFILL_RETRY_MAX_MS: u64 = 15 * 60_000;
 #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
-const SCHEMA: u64 = 1;
+const SCHEMA: u64 = 2;
+/// waiting undecryptable events are saved in this many blobs, so one arriving rewrites a sixteenth
+const UTD_SHARDS: u8 = 16;
+const BACKFILL_TYPES: [&str; 6] = [
+    "m.room.message",
+    "m.room.encrypted",
+    "m.room.redaction",
+    "m.sticker",
+    "m.poll.start",
+    "org.matrix.msc3381.poll.start",
+];
 
 #[cfg(not(target_family = "wasm"))]
 type BoxStream<T> = futures_util::stream::BoxStream<'static, T>;
@@ -72,6 +92,8 @@ pub(crate) trait MessageIndex: SendOutsideWasm {
     /// replaces whatever `doc.key` held
     fn put(&mut self, doc: Doc);
     fn remove(&mut self, key: &str) -> bool;
+    /// every provisional doc waiting on `target`, whoever sent its edit
+    fn remove_provisional(&mut self, target: &str) -> bool;
     fn buffered(&self) -> usize;
     /// newest first, strictly older than `before`. also hands back the analysed query terms
     fn search(
@@ -84,7 +106,8 @@ pub(crate) trait MessageIndex: SendOutsideWasm {
     fn maintain(&mut self) -> Result<(), Error>;
     fn changes(&mut self) -> Result<Changes, Error>;
     fn committed(&mut self, changes: &Changes);
-    fn manifest(&self) -> Value;
+    /// stops tracking what a commit would need, once nothing will be committed again
+    fn detach(&mut self);
 }
 
 #[derive(Clone, Debug, Default)]
@@ -160,46 +183,70 @@ enum Work {
     Backfill {
         room: OwnedRoomId,
         head: bool,
-        result: Result<matrix_sdk::room::Messages, Error>,
+        from: Option<String>,
+        result: Result<Messages, Error>,
     },
+    /// the room was left while its page was in flight
+    BackfillCancelled,
     Decrypted {
         key: String,
         room: OwnedRoomId,
         value: Option<Value>,
     },
+    /// the original first, then whatever edits of it the cache or server still has
     Refetched {
         room: OwnedRoomId,
-        value: Option<Value>,
+        values: Vec<Value>,
     },
 }
 
+enum RoomUpdate {
+    Events(Vec<SdkTimelineEvent>),
+    Lagged,
+    Closed,
+}
+
 struct Failure {
     message: String,
     retry_at_ms: u64,
+    delay_ms: u64,
+}
+
+enum Persistence {
+    Store(SearchStore),
+    Memory(SearchInMemory),
 }
 
 pub(crate) struct Service {
     client: Client,
     index: Box<dyn MessageIndex>,
-    store: Option<Result<SearchStore, Error>>,
+    persistence: Persistence,
+    #[cfg_attr(not(any(feature = "sqlite", feature = "indexeddb")), allow(dead_code))]
+    commit_failures: u32,
+    /// storage keys a load found nothing pointing at, dropped with the next commit
+    #[cfg_attr(not(any(feature = "sqlite", feature = "indexeddb")), allow(dead_code))]
+    orphans: Vec<String>,
     rooms: BTreeMap<String, RoomProgress>,
+    /// each room's newest indexed message when this engine started, what a head pass looks for
+    baselines: HashMap<String, u64>,
     pending: BTreeMap<String, Pending>,
-    pending_dirty: bool,
+    dirty_shards: BTreeSet<u8>,
+    decrypt_queue: VecDeque<String>,
+    decrypting: usize,
     redactions: Redactions,
     dirty_since: Option<u64>,
     commit_after: u64,
     budget: Budget,
     requests: mpsc::Receiver<Request>,
     subscriptions: HashMap<OwnedRoomId, (AbortHandle, u64)>,
-    updates: SelectAll<BoxStream<(OwnedRoomId, Option<Vec<SdkTimelineEvent>>)>>,
+    updates: SelectAll<BoxStream<(OwnedRoomId, RoomUpdate)>>,
     work: FuturesUnordered<EngineBoxFuture<'static, Work>>,
     pages: HashMap<OwnedRoomId, u16>,
     heads: VecDeque<OwnedRoomId>,
     headed: HashSet<OwnedRoomId>,
     head_tokens: HashMap<OwnedRoomId, String>,
-    backfilling: Option<OwnedRoomId>,
+    backfilling: Option<(OwnedRoomId, AbortHandle)>,
     next_backfill: u64,
-    retry_delay: u64,
     failures: HashMap<OwnedRoomId, Failure>,
     turn: usize,
 }
@@ -224,6 +271,18 @@ fn decode_cursor(cursor: &str) -> Option<(u64, u64)> {
     ))
 }
 
+fn shard(key: &str) -> u8 {
+    Sha256::digest(key.as_bytes())[0] % UTD_SHARDS
+}
+
+/// a 400 to a page with a saved token means the server no longer takes the token
+fn token_rejected(error: &Error) -> bool {
+    error
+        .server
+        .as_ref()
+        .is_some_and(|server| server.status == 400 || server.errcode == "M_INVALID_PARAM")
+}
+
 impl Handle {
     pub(crate) async fn execute(&self, command: SearchCommand) -> Result<CommandOutput, Error> {
         match command {
@@ -232,18 +291,30 @@ impl Handle {
                 scope,
                 limit,
             } => {
-                let (page, media) = self.search(&query, &scope, None, limit).await?;
-                let search_id = self.sessions.open(&self.media, query, scope, media).await;
-                Ok(CommandOutput::SearchSession(SearchSession { search_id, page }))
+                let media_scope = self.media.context_scope();
+                let prefix = media_prefix(&media_scope);
+                let (page, media) = self.search(&query, &scope, None, limit, &prefix).await?;
+                let search_id = self
+                    .sessions
+                    .open(&self.media, query, scope, media_scope, media)
+                    .await;
+                Ok(CommandOutput::SearchSession(SearchSession {
+                    search_id,
+                    page,
+                }))
             }
             SearchCommand::More {
                 search_id,
                 cursor,
                 limit,
             } => {
-                let (query, scope) = self.sessions.get(&search_id).await?;
-                let (page, media) = self.search(&query, &scope, Some(cursor.as_str()), limit).await?;
-                self.sessions.register(&self.media, &search_id, media).await?;
+                let (query, scope, prefix) = self.sessions.get(&search_id).await?;
+                let (page, media) = self
+                    .search(&query, &scope, Some(cursor.as_str()), limit, &prefix)
+                    .await?;
+                self.sessions
+                    .register(&self.media, &search_id, media)
+                    .await?;
                 Ok(CommandOutput::SearchPage(page))
             }
             SearchCommand::Close { search_id } => {
@@ -291,6 +362,7 @@ impl Handle {
         scope: &SearchScope,
         cursor: Option<&str>,
         limit: u16,
+        media_prefix: &str,
     ) -> Result<(SearchPage, Vec<MediaRegistration>), Error> {
         if limit == 0 || limit > MAX_LIMIT {
             return Err(Error::new(
@@ -298,9 +370,6 @@ impl Handle {
                 "search limit must be between 1 and 50",
             ));
         }
-        if query.trim().is_empty() {
-            return Err(Error::new(ErrorKind::Invalid, "search query is empty"));
-        }
         let before = match cursor {
             Some(cursor) => Some(
                 decode_cursor(cursor)
@@ -329,12 +398,15 @@ impl Handle {
             .flatten();
         let ignored = self.ignored_users().await;
         let current_user_id = self.client.user_id().map(ToOwned::to_owned);
-        let previews = join_all(
+        let terms = &raw.terms;
+        let previews: Vec<_> = futures_util::stream::iter(
             raw.hits
                 .into_iter()
-                .filter(|hit| !ignored.iter().any(|user| user.as_str() == hit.sender))
-                .map(|hit| self.preview(hit, current_user_id.clone(), &raw.terms)),
+                .filter(|hit| !ignored.iter().any(|user| user.as_str() == hit.sender)),
         )
+        .map(|hit| self.preview(hit, current_user_id.clone(), terms, media_prefix))
+        .buffered(PREVIEWS_AT_ONCE)
+        .collect()
         .await;
         let mut hits = Vec::new();
         let mut media = Vec::new();
@@ -375,6 +447,7 @@ impl Handle {
         hit: Hit,
         current_user_id: Option<OwnedUserId>,
         terms: &[String],
+        media_prefix: &str,
     ) -> Preview {
         let (Ok(room_id), Ok(event_id)) = (
             OwnedRoomId::try_from(hit.room.as_str()),
@@ -385,20 +458,33 @@ impl Handle {
         let Some(room) = self.client.get_room(&room_id) else {
             return Preview::Unavailable;
         };
-        let Some((original, highlight)) = load(&room, &hit.key).await else {
+        let Some((mut original, highlight)) = load(&room, &hit.key).await else {
             return Preview::Unavailable;
         };
         if original.pointer("/unsigned/redacted_because").is_some() {
             return Preview::Redacted(hit.key);
         }
+        // the index already settled which edit is newest, and a bundled one can be older
+        if let Some(relations) = original
+            .pointer_mut("/unsigned/m.relations")
+            .and_then(Value::as_object_mut)
+        {
+            relations.remove("m.replace");
+        }
         let mut values = vec![original];
         if hit.content != hit.key
             && let Some((edit, _)) = load(&room, &hit.content).await
         {
             values.push(edit);
         }
-        let Some((event, registrations)) =
-            context::preview(&room, current_user_id.as_deref(), &values, highlight).await
+        let Some((event, registrations)) = context::preview(
+            &room,
+            current_user_id.as_deref(),
+            &values,
+            highlight,
+            media_prefix,
+        )
+        .await
         else {
             return Preview::Unavailable;
         };
@@ -409,7 +495,7 @@ impl Handle {
             _ => None,
         };
         let matches = text
-            .map(|text| index::match_ranges(text, terms))
+            .map(|text| text::match_ranges(text, terms))
             .unwrap_or_default()
             .into_iter()
             .filter_map(|(start, end)| {
@@ -439,9 +525,14 @@ enum Preview {
 }
 
 /// open searches, oldest first, each owning the media of the hits it handed out
-#[derive(Default)]
 struct Sessions {
     open: Mutex<VecDeque<Session>>,
+    events: EventSender,
+}
+
+/// media ids carry the search, so two owners of one event never share an entry
+fn media_prefix(scope: &MediaScope) -> String {
+    format!("{}#", scope.nonce())
 }
 
 struct Session {
@@ -457,30 +548,51 @@ impl Sessions {
         media: &MediaService,
         query: String,
         scope: SearchScope,
+        media_scope: MediaScope,
         registrations: Vec<MediaRegistration>,
     ) -> String {
-        let media_scope = media.context_scope();
         let id = format!("search-{}", media_scope.nonce());
-        let mut open = self.open.lock().await;
-        media.register(&media_scope, registrations).await;
-        open.push_back(Session {
-            id: id.clone(),
-            query,
-            scope,
-            media: media_scope,
-        });
-        while open.len() > MAX_SEARCHES {
-            if let Some(oldest) = open.pop_front() {
-                media.release(&oldest.media).await;
+        let mut evicted = Vec::new();
+        {
+            let mut open = self.open.lock().await;
+            media.register(&media_scope, registrations).await;
+            open.push_back(Session {
+                id: id.clone(),
+                query,
+                scope,
+                media: media_scope,
+            });
+            while open.len() > MAX_SEARCHES {
+                if let Some(oldest) = open.pop_front() {
+                    media.release(&oldest.media).await;
+                    evicted.push(oldest.id);
+                }
+            }
+        }
+        for search_id in evicted {
+            let event = Event::SearchClosed {
+                search_id,
+                reason: SearchCloseReason::Evicted,
+            };
+            if self.events.send(event).await.is_err() {
+                break;
             }
         }
         id
     }
 
-    async fn get(&self, id: &str) -> Result<(String, SearchScope), Error> {
+    /// the query, scope and media prefix the search's pages use
+    async fn get(&self, id: &str) -> Result<(String, SearchScope, String), Error> {
         let open = self.open.lock().await;
-        let session = open.iter().find(|session| session.id == id).ok_or_else(not_open)?;
-        Ok((session.query.clone(), session.scope.clone()))
+        let session = open
+            .iter()
+            .find(|session| session.id == id)
+            .ok_or_else(not_open)?;
+        Ok((
+            session.query.clone(),
+            session.scope.clone(),
+            media_prefix(&session.media),
+        ))
     }
 
     /// under the lock, so a page that lands after its close registers nothing
@@ -491,7 +603,10 @@ impl Sessions {
         registrations: Vec<MediaRegistration>,
     ) -> Result<(), Error> {
         let open = self.open.lock().await;
-        let session = open.iter().find(|session| session.id == id).ok_or_else(not_open)?;
+        let session = open
+            .iter()
+            .find(|session| session.id == id)
+            .ok_or_else(not_open)?;
         media.register(&session.media, registrations).await;
         Ok(())
     }
@@ -510,7 +625,10 @@ impl Sessions {
 }
 
 fn not_open() -> Error {
-    Error::new(ErrorKind::SearchNotOpen, "search is closed or was never opened")
+    Error::new(
+        ErrorKind::SearchNotOpen,
+        "search is closed or was never opened",
+    )
 }
 
 async fn load(room: &Room, event_id: &str) -> Option<(Value, bool)> {
@@ -526,25 +644,25 @@ async fn load(room: &Room, event_id: &str) -> Option<(Value, bool)> {
     Some((value_of(&event)?, highlight))
 }
 
-#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
-fn decode_meta(
-    store: &SearchStore,
-    blobs: &BTreeMap<String, Vec<u8>>,
-) -> Option<(
-    Value,
-    BTreeMap<String, RoomProgress>,
-    BTreeMap<String, Pending>,
-)> {
-    let meta: Value = serde_json::from_slice(&store.open_blob("meta", blobs.get("meta")?)?).ok()?;
-    if meta["schema"].as_u64() != Some(SCHEMA) {
-        return None;
-    }
-    let rooms = serde_json::from_value(meta["rooms"].clone()).ok()?;
-    let pending = match blobs.get("utd") {
-        Some(sealed) => serde_json::from_slice(&store.open_blob("utd", sealed)?).ok()?,
-        None => BTreeMap::new(),
+/// the original and every edit of it still around, so a redacted latest edit falls back to
+/// the one before it rather than to the original
+async fn load_with_edits(room: &Room, event_id: &str) -> Vec<Value> {
+    let Ok(event_id) = ruma::EventId::parse(event_id) else {
+        return Vec::new();
     };
-    Some((meta["index"].clone(), rooms, pending))
+    match room
+        .load_or_fetch_event_with_relations(&event_id, Some(vec![RelationType::Replacement]), None)
+        .await
+    {
+        Ok((original, edits)) => std::iter::once(original)
+            .chain(edits)
+            .filter_map(|event| value_of(&event))
+            .collect(),
+        Err(error) => {
+            tracing::debug!(%event_id, %error, "reading an original again failed");
+            Vec::new()
+        }
+    }
 }
 
 impl Service {
@@ -552,16 +670,33 @@ impl Service {
         client: Client,
         media: Arc<MediaService>,
         store: Option<Result<SearchStore, Error>>,
+        events: EventSender,
         config: &EngineConfig,
     ) -> (Handle, Self) {
         let (sender, requests) = mpsc::channel(config.command_capacity.max(1));
+        let persistence = match store {
+            Some(Ok(store)) => Persistence::Store(store),
+            Some(Err(error)) => {
+                tracing::error!(%error, "search store unavailable, the index lives in memory");
+                Persistence::Memory(SearchInMemory::OpenFailed)
+            }
+            None => Persistence::Memory(SearchInMemory::NoStore),
+        };
         let service = Self {
             client: client.clone(),
-            index: Box::new(KeywordIndex::new(store.as_ref().is_some_and(Result::is_ok))),
-            store,
+            index: Box::new(KeywordIndex::new(matches!(
+                persistence,
+                Persistence::Store(_)
+            ))),
+            persistence,
+            commit_failures: 0,
+            orphans: Vec::new(),
             rooms: BTreeMap::new(),
+            baselines: HashMap::new(),
             pending: BTreeMap::new(),
-            pending_dirty: false,
+            dirty_shards: BTreeSet::new(),
+            decrypt_queue: VecDeque::new(),
+            decrypting: 0,
             redactions: Redactions::default(),
             dirty_since: None,
             commit_after: 0,
@@ -576,7 +711,6 @@ impl Service {
             head_tokens: HashMap::new(),
             backfilling: None,
             next_backfill: 0,
-            retry_delay: BACKFILL_RETRY_MS,
             failures: HashMap::new(),
             turn: 0,
         };
@@ -585,129 +719,181 @@ impl Service {
                 requests: sender,
                 client,
                 media,
-                sessions: Arc::default(),
+                sessions: Arc::new(Sessions {
+                    open: Mutex::default(),
+                    events,
+                }),
             },
             service,
         )
     }
 
-    fn persistent(&self) -> bool {
-        matches!(self.store, Some(Ok(_)))
+    fn in_memory(&self) -> Option<SearchInMemory> {
+        match self.persistence {
+            Persistence::Store(_) => None,
+            Persistence::Memory(reason) => Some(reason),
+        }
+    }
+
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    fn fall_back(&mut self, reason: SearchInMemory) {
+        self.persistence = Persistence::Memory(reason);
+        self.index.detach();
+        self.orphans.clear();
     }
 
     #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
     async fn load(&mut self) {
-        let Some(Ok(store)) = &self.store else {
-            if let Some(Err(error)) = &self.store {
-                tracing::error!(%error, "search store unavailable, the index lives in memory");
-            }
+        let Persistence::Store(store) = &self.persistence else {
             return;
         };
-        let blobs = match store.load().await {
+        let mut blobs = match store.load().await {
             Ok(blobs) => blobs,
             Err(error) => {
                 tracing::error!(%error, "reading the search store failed, the index lives in memory");
-                self.store = Some(Err(error));
-                self.index = Box::new(KeywordIndex::new(false));
+                self.fall_back(SearchInMemory::ReadFailed);
                 return;
             }
         };
         if blobs.is_empty() {
             return;
         }
-        let restored = decode_meta(store, &blobs).and_then(|(manifest, rooms, pending)| {
-            let index = KeywordIndex::load(&manifest, |key| {
-                blobs
-                    .get(key)
-                    .and_then(|sealed| store.open_blob(key, sealed))
-            })
-            .inspect_err(|error| tracing::warn!(%error, "search segments are unreadable"))
-            .ok()?;
+        let restored = (|| {
+            let meta: Value = serde_json::from_slice(&store.take(&mut blobs, "meta")?).ok()?;
+            if meta["schema"].as_u64() != Some(SCHEMA) {
+                return None;
+            }
+            let rooms: BTreeMap<String, RoomProgress> =
+                serde_json::from_value(meta["rooms"].clone()).ok()?;
+            let mut pending = BTreeMap::new();
+            for shard in 0..UTD_SHARDS {
+                if let Some(bytes) = store.take(&mut blobs, &format!("utd/{shard}")) {
+                    let part: BTreeMap<String, Pending> = serde_json::from_slice(&bytes).ok()?;
+                    pending.extend(part);
+                }
+            }
+            let index = KeywordIndex::load(&meta["index"], |name| store.take(&mut blobs, name))
+                .inspect_err(|error| tracing::warn!(%error, "search segments are unreadable"))
+                .ok()?;
             Some((index, rooms, pending))
-        });
-        drop(blobs);
+        })();
         match restored {
             Some((index, rooms, pending)) => {
                 self.index = Box::new(index);
+                self.baselines = rooms
+                    .iter()
+                    .filter_map(|(room, progress)| Some((room.clone(), progress.newest?)))
+                    .collect();
                 self.rooms = rooms;
                 self.pending = pending;
+                self.orphans = blobs.leftover();
+                if !self.orphans.is_empty() {
+                    self.touch();
+                }
             }
             None => {
                 tracing::warn!("resetting an unreadable search index");
                 if let Err(error) = store.reset().await {
                     tracing::error!(%error, "search store unavailable, the index lives in memory");
-                    self.store = Some(Err(error));
-                    self.index = Box::new(KeywordIndex::new(false));
+                    self.fall_back(SearchInMemory::ReadFailed);
                 }
             }
         }
     }
 
     #[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
-    async fn load(&mut self) {
-        if let Some(Err(error)) = &self.store {
-            tracing::error!(%error, "search store unavailable, the index lives in memory");
-        }
-    }
+    async fn load(&mut self) {}
 
     #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
     async fn commit(&mut self) {
-        if !self.persistent() {
+        if !matches!(self.persistence, Persistence::Store(_)) {
             if let Err(error) = self.index.maintain() {
                 tracing::warn!(%error, "maintaining the search index failed");
             }
             self.dirty_since = None;
+            self.dirty_shards.clear();
             return;
         }
-        let changes = match self.index.changes() {
+        let mut changes = match self.index.changes() {
             Ok(changes) => changes,
             Err(error) => {
                 tracing::warn!(%error, "preparing a search commit failed");
-                self.commit_after = now() + COMMIT_RETRY_MS;
+                self.commit_failed();
                 return;
             }
         };
-        let mut puts = changes.puts.clone();
+        let mut puts = std::mem::take(&mut changes.puts);
+        let mut deletes = changes.deletes.clone();
         let meta = serde_json::json!({
             "schema": SCHEMA,
-            "index": self.index.manifest(),
+            "index": changes.manifest,
             "rooms": self.rooms,
         });
         puts.push(("meta".to_owned(), meta.to_string().into_bytes()));
-        if self.pending_dirty {
-            puts.push((
-                "utd".to_owned(),
-                serde_json::to_vec(&self.pending).unwrap_or_default(),
-            ));
+        for shard in &self.dirty_shards {
+            let part: BTreeMap<&String, &Pending> = self
+                .pending
+                .iter()
+                .filter(|(key, _)| self::shard(key) == *shard)
+                .collect();
+            let name = format!("utd/{shard}");
+            if part.is_empty() {
+                deletes.push(name);
+            } else {
+                puts.push((name, serde_json::to_vec(&part).unwrap_or_default()));
+            }
         }
-        let Some(Ok(store)) = &self.store else {
+        let Persistence::Store(store) = &self.persistence else {
             return;
         };
-        match store.commit(puts, changes.deletes.clone()).await {
+        match store.commit(puts, deletes, self.orphans.clone()).await {
             Ok(()) => {
                 self.index.committed(&changes);
-                self.pending_dirty = false;
+                self.dirty_shards.clear();
+                self.orphans.clear();
                 self.dirty_since = None;
+                self.commit_failures = 0;
             }
             Err(error) => {
-                tracing::warn!(%error, "saving the search index failed, retrying later");
-                self.commit_after = now() + COMMIT_RETRY_MS;
+                tracing::warn!(%error, "saving the search index failed");
+                self.commit_failed();
             }
         }
     }
 
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    fn commit_failed(&mut self) {
+        self.commit_failures += 1;
+        if self.commit_failures >= COMMIT_ATTEMPTS {
+            tracing::error!("search commits keep failing, the index lives in memory from here");
+            self.fall_back(SearchInMemory::CommitFailed);
+            self.dirty_since = None;
+            self.dirty_shards.clear();
+        } else {
+            self.commit_after = now() + COMMIT_RETRY_MS;
+        }
+    }
+
     #[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
     async fn commit(&mut self) {
         if let Err(error) = self.index.maintain() {
             tracing::warn!(%error, "maintaining the search index failed");
         }
         self.dirty_since = None;
+        self.dirty_shards.clear();
     }
 
     fn touch(&mut self) {
         self.dirty_since.get_or_insert_with(now);
     }
 
+    fn unpend(&mut self, key: &str) {
+        if self.pending.remove(key).is_some() {
+            self.dirty_shards.insert(shard(key));
+            self.touch();
+        }
+    }
+
     fn observe(&mut self, room: &OwnedRoomId, value: &Value) {
         let observed = feed::observe(value);
         let key = match &observed {
@@ -733,6 +919,7 @@ impl Service {
                         raw: value.to_string(),
                     },
                 );
+                self.dirty_shards.insert(shard(key));
                 while self.pending.len() > self.budget.max_pending {
                     let Some(oldest) = self
                         .pending
@@ -742,18 +929,14 @@ impl Service {
                     else {
                         break;
                     };
-                    self.pending.remove(&oldest);
+                    self.unpend(&oldest);
                 }
-                self.pending_dirty = true;
                 self.touch();
             }
             return;
         }
-        if let Some(key) = key
-            && self.pending.remove(&key).is_some()
-        {
-            self.pending_dirty = true;
-            self.touch();
+        if let Some(key) = key {
+            self.unpend(&key);
         }
         match feed::apply(
             self.index.as_mut(),
@@ -767,19 +950,19 @@ impl Service {
                 progress.newest = Some(progress.newest.map_or(ts, |newest| newest.max(ts)));
                 self.touch();
             }
-            Effect::Removed => self.touch(),
+            Effect::Removed | Effect::Held => self.touch(),
             Effect::Refetch(key) => {
                 self.touch();
                 let room_id = room.clone();
                 let client = self.client.clone();
                 self.work.push(Box::pin(async move {
-                    let value = match client.get_room(&room_id) {
-                        Some(room) => load(&room, &key).await.map(|(value, _)| value),
-                        None => None,
+                    let values = match client.get_room(&room_id) {
+                        Some(room) => load_with_edits(&room, &key).await,
+                        None => Vec::new(),
                     };
                     Work::Refetched {
                         room: room_id,
-                        value,
+                        values,
                     }
                 }));
             }
@@ -817,11 +1000,66 @@ impl Service {
         let (stream, abort) = abortable(stream);
         self.updates.push(Box::pin(stream));
         self.subscriptions.insert(room_id.clone(), (abort, now()));
-        if self.headed.insert(room_id.clone()) {
+        let started = self
+            .rooms
+            .get(room_id.as_str())
+            .is_some_and(|progress| progress.started);
+        if started
+            && self.baselines.contains_key(room_id.as_str())
+            && self.headed.insert(room_id.clone())
+        {
             self.heads.push_back(room_id);
         }
     }
 
+    /// the generic channel holds 128 updates, and a missed one could be a room we never
+    /// follow, so a lag follows every joined room
+    async fn generic(&mut self, update: Result<OwnedRoomId, broadcast::error::RecvError>) -> bool {
+        match update {
+            Ok(room_id) => self.subscribe(room_id).await,
+            Err(broadcast::error::RecvError::Lagged(_)) => {
+                for room in self.client.joined_rooms() {
+                    self.subscribe(room.room_id().to_owned()).await;
+                }
+            }
+            Err(broadcast::error::RecvError::Closed) => return false,
+        }
+        true
+    }
+
+    async fn room_update(&mut self, room_id: OwnedRoomId, update: RoomUpdate) {
+        match update {
+            RoomUpdate::Events(events) => {
+                if let Some((_, seen)) = self.subscriptions.get_mut(&room_id) {
+                    *seen = now();
+                }
+                self.observe_all(&room_id, &events);
+            }
+            RoomUpdate::Lagged => {
+                if let Some((abort, _)) = self.subscriptions.remove(&room_id) {
+                    abort.abort();
+                }
+                self.subscribe(room_id).await;
+            }
+            RoomUpdate::Closed => {
+                self.subscriptions.remove(&room_id);
+            }
+        }
+    }
+
+    fn left(&mut self, room_id: &OwnedRoomId) {
+        if let Some((abort, _)) = self.subscriptions.remove(room_id) {
+            abort.abort();
+        }
+        self.heads.retain(|head| head != room_id);
+        self.head_tokens.remove(room_id);
+        if let Some((backfilling, abort)) = &self.backfilling
+            && backfilling == room_id
+        {
+            abort.abort();
+        }
+    }
+
     fn drop_idle(&mut self) {
         let cutoff = now().saturating_sub(IDLE_SUBSCRIPTION_MS);
         self.subscriptions.retain(|_, (abort, seen)| {
@@ -834,6 +1072,12 @@ impl Service {
     }
 
     fn coverage(&self, rooms: &[OwnedRoomId]) -> SearchCoverage {
+        let scoped: HashSet<&str> = rooms.iter().map(|room| room.as_str()).collect();
+        let undecrypted = self
+            .pending
+            .values()
+            .filter(|pending| scoped.contains(pending.room.as_str()))
+            .count();
         let rooms = rooms
             .iter()
             .map(|room_id| {
@@ -846,16 +1090,16 @@ impl Service {
                     SearchBackfill::Complete
                 } else if self.budget.pages_per_room == 0 {
                     SearchBackfill::Disabled
-                } else if self.backfilling.as_ref() == Some(room_id) {
-                    SearchBackfill::Running
-                } else if let Some(Failure {
-                    message,
-                    retry_at_ms,
-                }) = self.failures.get(room_id)
+                } else if self
+                    .backfilling
+                    .as_ref()
+                    .is_some_and(|(backfilling, _)| backfilling == room_id)
                 {
+                    SearchBackfill::Running
+                } else if let Some(failure) = self.failures.get(room_id) {
                     SearchBackfill::Failed {
-                        message: message.clone(),
-                        retry_at_ms: *retry_at_ms,
+                        message: failure.message.clone(),
+                        retry_at_ms: failure.retry_at_ms,
                     }
                 } else if self.pages.get(room_id).copied().unwrap_or(0)
                     >= self.budget.pages_per_room
@@ -873,40 +1117,47 @@ impl Service {
                 }
             })
             .collect();
+        let in_memory = self.in_memory();
         SearchCoverage {
             rooms,
-            persistent: self.persistent(),
-            undecrypted: self.pending.len(),
+            persistent: in_memory.is_none(),
+            in_memory,
+            undecrypted,
         }
     }
 
+    fn spent(&self, room_id: &OwnedRoomId) -> bool {
+        self.pages.get(room_id).copied().unwrap_or(0) >= self.budget.pages_per_room
+    }
+
+    fn blocked(&self, room_id: &OwnedRoomId, at: u64) -> bool {
+        self.failures
+            .get(room_id)
+            .is_some_and(|failure| failure.retry_at_ms > at)
+    }
+
     /// head passes first, so what arrived while we were away is found before anything older
     fn schedule_backfill(&mut self) {
-        if self.backfilling.is_some()
-            || self.budget.pages_per_room == 0
-            || now() < self.next_backfill
+        let at = now();
+        if self.backfilling.is_some() || self.budget.pages_per_room == 0 || at < self.next_backfill
         {
             return;
         }
-        let budget = self.budget.pages_per_room;
-        let spent = |pages: &HashMap<OwnedRoomId, u16>, room: &OwnedRoomId| {
-            pages.get(room).copied().unwrap_or(0) >= budget
-        };
-        let blocked = |failures: &HashMap<OwnedRoomId, Failure>, room: &OwnedRoomId| {
-            failures
-                .get(room)
-                .is_some_and(|failure| failure.retry_at_ms > now())
-        };
         let mut chosen = None;
-        while let Some(room_id) = self.heads.pop_front() {
-            let started = self
-                .rooms
-                .get(room_id.as_str())
-                .is_some_and(|progress| progress.started);
-            if started && !spent(&self.pages, &room_id) && !blocked(&self.failures, &room_id) {
-                chosen = Some((room_id, true));
+        for _ in 0..self.heads.len() {
+            let Some(room_id) = self.heads.pop_front() else {
                 break;
+            };
+            if self.spent(&room_id) {
+                self.head_tokens.remove(&room_id);
+                continue;
             }
+            if self.blocked(&room_id, at) {
+                self.heads.push_back(room_id);
+                continue;
+            }
+            chosen = Some((room_id, true));
+            break;
         }
         if chosen.is_none() {
             let mut joined: Vec<_> = self
@@ -923,7 +1174,7 @@ impl Service {
                     .rooms
                     .get(room_id.as_str())
                     .is_some_and(|progress| progress.start_reached);
-                if !done && !spent(&self.pages, room_id) && !blocked(&self.failures, room_id) {
+                if !done && !self.spent(room_id) && !self.blocked(room_id, at) {
                     self.turn = (self.turn + offset + 1) % count;
                     chosen = Some((room_id.clone(), false));
                     break;
@@ -944,63 +1195,99 @@ impl Service {
                 .and_then(|progress| progress.tail.clone())
         };
         *self.pages.entry(room_id.clone()).or_default() += 1;
-        self.backfilling = Some(room_id.clone());
         let mut options = MessagesOptions::backward();
-        options.from = from;
+        options.from = from.clone();
         options.limit = UInt::from(self.budget.page_size);
+        let mut filter = RoomEventFilter::default();
+        filter.types = Some(
+            BACKFILL_TYPES
+                .iter()
+                .map(|kind| (*kind).to_owned())
+                .collect(),
+        );
+        options.filter = filter;
+        let (abort, registration) = AbortHandle::new_pair();
+        self.backfilling = Some((room_id.clone(), abort));
+        let page = Abortable::new(
+            async move {
+                room.messages(options)
+                    .await
+                    .map_err(|error| Error::from_sdk("backfilling search history failed", error))
+            },
+            registration,
+        );
         self.work.push(Box::pin(async move {
-            let result = room
-                .messages(options)
-                .await
-                .map_err(|error| Error::from_sdk("backfilling search history failed", error));
-            Work::Backfill {
-                room: room_id,
-                head,
-                result,
+            match page.await {
+                Ok(result) => Work::Backfill {
+                    room: room_id,
+                    head,
+                    from,
+                    result,
+                },
+                Err(_) => Work::BackfillCancelled,
             }
         }));
     }
 
+    fn fail_backfill(&mut self, room_id: OwnedRoomId, message: String, delay_ms: u64) {
+        self.failures.insert(
+            room_id,
+            Failure {
+                message,
+                retry_at_ms: now() + delay_ms,
+                delay_ms,
+            },
+        );
+    }
+
     fn backfilled(
         &mut self,
         room_id: OwnedRoomId,
         head: bool,
-        result: Result<matrix_sdk::room::Messages, Error>,
+        from: Option<String>,
+        result: Result<Messages, Error>,
     ) {
         self.backfilling = None;
+        self.next_backfill = now() + self.budget.interval_ms;
         let messages = match result {
             Ok(messages) => messages,
+            Err(error) if from.is_some() && token_rejected(&error) => {
+                if head {
+                    self.head_tokens.remove(&room_id);
+                } else {
+                    let progress = self.rooms.entry(room_id.to_string()).or_default();
+                    progress.tail = None;
+                    progress.started = false;
+                    self.touch();
+                }
+                let message = format!("{}, starting again from the newest", error.message);
+                self.fail_backfill(room_id, message, self.budget.interval_ms);
+                return;
+            }
             Err(error) => {
-                let retry_at_ms = now() + self.retry_delay;
-                self.next_backfill = retry_at_ms;
-                self.retry_delay = (self.retry_delay * 2).min(BACKFILL_RETRY_MAX_MS);
-                self.failures.insert(
-                    room_id,
-                    Failure {
-                        message: error.message.clone(),
-                        retry_at_ms,
-                    },
-                );
+                let delay_ms = self
+                    .failures
+                    .get(&room_id)
+                    .map_or(BACKFILL_RETRY_MS, |failure| {
+                        (failure.delay_ms * 2).min(BACKFILL_RETRY_MAX_MS)
+                    });
+                self.fail_backfill(room_id, error.message.clone(), delay_ms);
                 return;
             }
         };
-        self.retry_delay = BACKFILL_RETRY_MS;
         self.failures.remove(&room_id);
-        self.next_backfill = now() + self.budget.interval_ms;
+        let baseline = self.baselines.get(room_id.as_str()).copied();
         let mut overlapped = false;
         for event in &messages.chunk {
             let Some(value) = value_of(event) else {
                 continue;
             };
-            if head
-                && let Some(key) = value.get("event_id").and_then(Value::as_str)
-                && self.index.get(key).is_some_and(|doc| !doc.provisional)
-            {
+            if head && head_overlap(self.index.as_ref(), baseline, &value) {
                 overlapped = true;
             }
             self.observe(&room_id, &value);
         }
-        let ended = messages.end.is_none() || messages.chunk.is_empty();
+        let ended = messages.end.is_none() || (messages.chunk.is_empty() && messages.end == from);
         if head {
             match messages.end {
                 Some(end) if !overlapped && !ended => {
@@ -1024,22 +1311,36 @@ impl Service {
     }
 
     fn retry_pending(&mut self, room_id: Option<&OwnedRoomId>, session_id: Option<&str>) {
+        let queued: HashSet<String> = self.decrypt_queue.iter().cloned().collect();
         let due: Vec<_> = self
             .pending
             .iter()
-            .filter(|(_, pending)| {
-                room_id.is_none_or(|room_id| pending.room == room_id.as_str())
+            .filter(|(key, pending)| {
+                !queued.contains(*key)
+                    && room_id.is_none_or(|room_id| pending.room == room_id.as_str())
                     && session_id.is_none_or(|session_id| pending.session_id == session_id)
             })
-            .map(|(key, pending)| (key.clone(), pending.clone()))
+            .map(|(key, _)| key.clone())
             .collect();
-        for (key, pending) in due {
+        self.decrypt_queue.extend(due);
+        self.pump_decrypts();
+    }
+
+    fn pump_decrypts(&mut self) {
+        while self.decrypting < DECRYPTS_AT_ONCE {
+            let Some(key) = self.decrypt_queue.pop_front() else {
+                return;
+            };
+            let Some(pending) = self.pending.get(&key).cloned() else {
+                continue;
+            };
             let Ok(room_id) = OwnedRoomId::try_from(pending.room.as_str()) else {
                 continue;
             };
             let Some(room) = self.client.get_room(&room_id) else {
                 continue;
             };
+            self.decrypting += 1;
             self.work.push(Box::pin(async move {
                 let value = async {
                     let raw: Raw<OriginalSyncRoomEncryptedEvent> =
@@ -1059,17 +1360,26 @@ impl Service {
 
     fn complete(&mut self, work: Work) {
         match work {
-            Work::Backfill { room, head, result } => self.backfilled(room, head, result),
+            Work::Backfill {
+                room,
+                head,
+                from,
+                result,
+            } => self.backfilled(room, head, from, result),
+            Work::BackfillCancelled => {
+                self.backfilling = None;
+                self.next_backfill = now() + self.budget.interval_ms;
+            }
             Work::Decrypted { key, room, value } => {
+                self.decrypting -= 1;
                 if let Some(value) = value {
-                    self.pending.remove(&key);
-                    self.pending_dirty = true;
+                    self.unpend(&key);
                     self.observe(&room, &value);
-                    self.touch();
                 }
+                self.pump_decrypts();
             }
-            Work::Refetched { room, value } => {
-                if let Some(value) = value {
+            Work::Refetched { room, values } => {
+                for value in values {
                     self.observe(&room, &value);
                 }
             }
@@ -1096,15 +1406,15 @@ impl Service {
     }
 
     pub(crate) async fn run(mut self) {
-        self.load().await;
         let client = self.client.clone();
         let mut generic = client.event_cache().subscribe_to_room_generic_updates();
+        let mut room_info = client.room_info_notable_update_receiver();
+        self.load().await;
         let mut keys = client
             .encryption()
             .room_keys_received_stream()
             .await
             .map(Box::pin);
-        let mut room_info = client.room_info_notable_update_receiver();
         self.retry_pending(None, None);
         let mut tick = Box::pin(matrix_sdk::sleep::sleep(Duration::from_secs(1)));
         loop {
@@ -1137,25 +1447,14 @@ impl Service {
                     None => return,
                 },
                 Some(work) = self.work.next(), if !self.work.is_empty() => self.complete(work),
-                Some((room_id, events)) = self.updates.next(), if !self.updates.is_empty() => {
-                    if let Some((_, seen)) = self.subscriptions.get_mut(&room_id) {
-                        *seen = now();
-                    }
-                    match events {
-                        Some(events) => self.observe_all(&room_id, &events),
-                        None => {
-                            if let Some((abort, _)) = self.subscriptions.remove(&room_id) {
-                                abort.abort();
-                            }
-                            self.subscribe(room_id).await;
-                        }
+                Some((room_id, update)) = self.updates.next(), if !self.updates.is_empty() => {
+                    self.room_update(room_id, update).await;
+                }
+                update = generic.recv() => {
+                    if !self.generic(update.map(|update| update.room_id)).await {
+                        return;
                     }
                 }
-                update = generic.recv() => match update {
-                    Ok(update) => self.subscribe(update.room_id).await,
-                    Err(broadcast::error::RecvError::Lagged(_)) => {}
-                    Err(broadcast::error::RecvError::Closed) => return,
-                },
                 Some(update) = async { keys.as_mut()?.next().await }, if keys.is_some() => match update {
                     Ok(infos) => {
                         for info in infos {
@@ -1167,8 +1466,8 @@ impl Service {
                 update = room_info.recv() => match update {
                     Ok(update) if update.reasons.contains(RoomInfoNotableUpdateReasons::MEMBERSHIP) => {
                         let left = self.client.get_room(&update.room_id).is_none_or(|room| room.state() != RoomState::Joined);
-                        if left && let Some((abort, _)) = self.subscriptions.remove(&update.room_id) {
-                            abort.abort();
+                        if left {
+                            self.left(&update.room_id);
                         }
                     }
                     Ok(_) | Err(broadcast::error::RecvError::Lagged(_)) => {}
@@ -1188,27 +1487,49 @@ impl Service {
     }
 }
 
+/// a head pass stops at what this engine found already indexed, and live events indexed
+/// since it started don't count, or the first page would always stop it
+fn head_overlap(index: &dyn MessageIndex, baseline: Option<u64>, value: &Value) -> bool {
+    let (Some(baseline), Some(key), Some(ts)) = (
+        baseline,
+        value.get("event_id").and_then(Value::as_str),
+        value.get("origin_server_ts").and_then(Value::as_u64),
+    ) else {
+        return false;
+    };
+    ts <= baseline && index.get(key).is_some_and(|doc| !doc.provisional)
+}
+
 fn updates(
     room_id: OwnedRoomId,
     subscriber: RoomEventCacheSubscriber,
     drop_handles: Arc<matrix_sdk::event_cache::EventCacheDropHandles>,
-) -> BoxStream<(OwnedRoomId, Option<Vec<SdkTimelineEvent>>)> {
+) -> BoxStream<(OwnedRoomId, RoomUpdate)> {
     Box::pin(futures_util::stream::unfold(
-        (subscriber, drop_handles),
-        move |(mut subscriber, drop_handles)| {
+        Some((subscriber, drop_handles)),
+        move |state| {
             let room_id = room_id.clone();
             async move {
+                let (mut subscriber, drop_handles) = state?;
                 loop {
                     match subscriber.recv().await {
                         Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) => {
                             let events = update.diffs.into_iter().flat_map(added).collect();
-                            return Some(((room_id, Some(events)), (subscriber, drop_handles)));
+                            return Some((
+                                (room_id, RoomUpdate::Events(events)),
+                                Some((subscriber, drop_handles)),
+                            ));
                         }
                         Ok(_) => {}
                         Err(broadcast::error::RecvError::Lagged(_)) => {
-                            return Some(((room_id, None), (subscriber, drop_handles)));
+                            return Some((
+                                (room_id, RoomUpdate::Lagged),
+                                Some((subscriber, drop_handles)),
+                            ));
+                        }
+                        Err(broadcast::error::RecvError::Closed) => {
+                            return Some(((room_id, RoomUpdate::Closed), None));
                         }
-                        Err(broadcast::error::RecvError::Closed) => return None,
                     }
                 }
             }
@@ -1232,3 +1553,7 @@ fn added(diff: VectorDiff<SdkTimelineEvent>) -> Vec<SdkTimelineEvent> {
         | VectorDiff::Truncate { .. } => Vec::new(),
     }
 }
+
+#[cfg(all(test, feature = "sqlite"))]
+#[path = "service_tests.rs"]
+mod tests;

Added file: src/search/service_tests.rs+141 additions@@ -0,0 +1,141 @@
+use matrix_sdk::{
+    deserialized_responses::TimelineEvent as SdkTimelineEvent,
+    ruma::{events::AnySyncTimelineEvent, room_id},
+    test_utils::mocks::MatrixMockServer,
+};
+use matrix_sdk_test::{JoinedRoomBuilder, event_factory::EventFactory};
+use serde_json::json;
+
+use super::*;
+
+async fn joined(server: &MatrixMockServer, room_id: &ruma::RoomId) -> Client {
+    let client = server.client_builder().build().await;
+    client.event_cache().subscribe().unwrap();
+    let own_user = client.user_id().unwrap().to_owned();
+    let room_id = room_id.to_owned();
+    server
+        .mock_sync()
+        .ok_and_run(&client, move |builder| {
+            let factory = EventFactory::new().room(&room_id);
+            builder.add_joined_room(
+                JoinedRoomBuilder::new(&room_id).add_state_event(factory.member(&own_user)),
+            );
+        })
+        .await;
+    client
+}
+
+fn service(client: &Client) -> Service {
+    let media = Arc::new(MediaService::new(client.clone(), 1 << 20, 64));
+    let (events, _) = EventSender::detached(16);
+    Service::new(
+        client.clone(),
+        media,
+        None,
+        events,
+        &EngineConfig::default(),
+    )
+    .1
+}
+
+fn message(id: &str, ts: u64, body: &str) -> Value {
+    json!({
+        "type": "m.room.message",
+        "event_id": id,
+        "sender": "@a:x",
+        "origin_server_ts": ts,
+        "content": { "msgtype": "m.text", "body": body },
+    })
+}
+
+fn page(events: &[Value], end: Option<&str>) -> Result<Messages, Error> {
+    Ok(Messages {
+        start: "start".to_owned(),
+        end: end.map(ToOwned::to_owned),
+        chunk: events
+            .iter()
+            .map(|value| {
+                SdkTimelineEvent::from_plaintext(
+                    Raw::<AnySyncTimelineEvent>::from_json_string(value.to_string()).unwrap(),
+                )
+            })
+            .collect(),
+        state: Vec::new(),
+    })
+}
+
+/// live events indexed this run used to count as overlap, so the first head page always
+/// ended the pass and an offline gap past one page was never filled
+#[tokio::test]
+async fn a_head_pass_walks_past_what_this_run_indexed_live() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!head:x");
+    let client = joined(&server, room_id).await;
+    let mut service = service(&client);
+    let room = room_id.to_owned();
+    service.rooms.insert(
+        room.to_string(),
+        RoomProgress {
+            newest: Some(100),
+            started: true,
+            ..RoomProgress::default()
+        },
+    );
+    service.baselines.insert(room.to_string(), 100);
+    service.observe(&room, &message("$old", 100, "old"));
+    service.observe(&room, &message("$live", 1_000, "live"));
+    service.backfilled(
+        room.clone(),
+        true,
+        None,
+        page(
+            &[message("$live", 1_000, "live"), message("$gap", 500, "gap")],
+            Some("t1"),
+        ),
+    );
+    assert_eq!(
+        service.head_tokens.get(&room).map(String::as_str),
+        Some("t1")
+    );
+    assert_eq!(service.heads.front(), Some(&room));
+    service.backfilled(
+        room.clone(),
+        true,
+        Some("t1".to_owned()),
+        page(&[message("$old", 100, "old")], Some("t2")),
+    );
+    assert!(!service.head_tokens.contains_key(&room));
+    assert!(service.index.get("$gap").is_some());
+}
+
+/// the generic channel is bounded, and a room whose only update was lost was never followed
+#[tokio::test]
+async fn a_lagged_generic_channel_follows_every_joined_room() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!lag:x");
+    let client = joined(&server, room_id).await;
+    let mut service = service(&client);
+    assert!(
+        service
+            .generic(Err(broadcast::error::RecvError::Lagged(3)))
+            .await
+    );
+    assert!(service.subscriptions.contains_key(room_id));
+}
+
+/// a closed stream left its entry behind, and the room was never followed again
+#[tokio::test]
+async fn a_closed_room_stream_can_be_followed_again() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!closed:x");
+    let client = joined(&server, room_id).await;
+    let mut service = service(&client);
+    service.subscribe(room_id.to_owned()).await;
+    assert!(service.subscriptions.contains_key(room_id));
+    service
+        .room_update(room_id.to_owned(), RoomUpdate::Closed)
+        .await;
+    assert!(!service.subscriptions.contains_key(room_id));
+    assert!(service.generic(Ok(room_id.to_owned())).await);
+    assert!(service.subscriptions.contains_key(room_id));
+}

Modified file: src/search/store/indexeddb.rs+19 additions−4 deletions@@ -136,19 +136,34 @@ impl Backend {
             .map_err(|error| storage_error("writing the search database", error))
     }
 
-    pub(super) async fn clear(&self) -> Result<(), Error> {
+    /// everything goes and `puts` arrive in the same transaction
+    pub(super) async fn replace(&self, puts: Vec<(String, Vec<u8>)>) -> Result<(), Error> {
         let database = self.database().await?;
-        let clear = async {
+        let replace = async {
             let transaction = database
                 .transaction(BLOBS)
                 .with_mode(TransactionMode::Readwrite)
                 .build()?;
             transaction.object_store(BLOBS)?.clear()?.await?;
+            let requests = (|| -> Result<(), IndexedDbError> {
+                let store = transaction.object_store(BLOBS)?;
+                for (key, value) in &puts {
+                    store
+                        .put(Uint8ArraySlice::new(value))
+                        .with_key(key.as_str())
+                        .build()?;
+                }
+                Ok(())
+            })();
+            if let Err(error) = requests {
+                let _ = transaction.abort().await;
+                return Err(error);
+            }
             transaction.commit().await
         };
-        clear
+        replace
             .await
-            .map_err(|error| storage_error("clearing the search database", error))
+            .map_err(|error| storage_error("replacing the search database", error))
     }
 }
 

Modified file: src/search/store/mod.rs+71 additions−48 deletions@@ -24,13 +24,30 @@ struct Owner {
     user_id: String,
 }
 
-/// blobs sealed with a `StoreCipher` under the store's passphrase. every blob carries its
-/// own key inside the seal, so one can't be swapped for another
+/// blobs sealed with a `StoreCipher` under the store's passphrase, stored under a keyed hash
+/// of their name like the mentions store's rows. every blob carries its own name inside the
+/// seal, so one can't be swapped for another
 pub(crate) struct SearchStore {
     backend: Backend,
     cipher: StoreCipher,
 }
 
+/// what a load found, still sealed. `take` hands each over once, and what's never taken
+/// is nothing the index points at
+pub(crate) struct Blobs {
+    sealed: BTreeMap<String, Vec<u8>>,
+}
+
+impl Blobs {
+    pub(crate) fn is_empty(&self) -> bool {
+        self.sealed.is_empty()
+    }
+
+    pub(crate) fn leftover(self) -> Vec<String> {
+        self.sealed.into_keys().collect()
+    }
+}
+
 impl std::fmt::Debug for SearchStore {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         formatter
@@ -51,7 +68,7 @@ impl SearchStore {
         let opened = blobs.remove(CIPHER).and_then(|exported| {
             let cipher = StoreCipher::import(config.passphrase(), &exported).ok()?;
             let stored: Owner =
-                serde_json::from_slice(&unseal(&cipher, ACCOUNT, &blobs.remove(ACCOUNT)?)?).ok()?;
+                serde_json::from_slice(&unseal(&cipher, ACCOUNT, blobs.remove(ACCOUNT)?)?).ok()?;
             (stored.homeserver == owner.homeserver && stored.user_id == owner.user_id)
                 .then_some(cipher)
         });
@@ -63,78 +80,84 @@ impl SearchStore {
         let exported = cipher
             .export(config.passphrase())
             .map_err(|error| storage_error("sealing the search cipher", error))?;
-        let account = serde_json::to_vec(&owner)
+        let owner = serde_json::to_vec(&owner)
             .map_err(|error| storage_error("encoding the search owner", error))?;
-        let store = Self { backend, cipher };
-        store.backend.clear().await?;
-        let sealed = store.seal(ACCOUNT, account)?;
-        store
-            .backend
-            .commit(
-                vec![(CIPHER.to_owned(), exported), (ACCOUNT.to_owned(), sealed)],
-                Vec::new(),
-            )
+        let sealed = seal(&cipher, ACCOUNT, owner)?;
+        backend
+            .replace(vec![
+                (CIPHER.to_owned(), exported),
+                (ACCOUNT.to_owned(), sealed),
+            ])
             .await?;
-        Ok(store)
+        Ok(Self { backend, cipher })
     }
 
-    /// every blob but the store's own, still sealed
-    pub(crate) async fn load(&self) -> Result<BTreeMap<String, Vec<u8>>, Error> {
-        let mut blobs = self.backend.load_all().await?;
-        blobs.remove(CIPHER);
-        blobs.remove(ACCOUNT);
-        Ok(blobs)
+    pub(crate) async fn load(&self) -> Result<Blobs, Error> {
+        let mut sealed = self.backend.load_all().await?;
+        sealed.remove(CIPHER);
+        sealed.remove(ACCOUNT);
+        Ok(Blobs { sealed })
     }
 
-    pub(crate) fn open_blob(&self, key: &str, sealed: &[u8]) -> Option<Vec<u8>> {
-        unseal(&self.cipher, key, sealed)
+    pub(crate) fn take(&self, blobs: &mut Blobs, name: &str) -> Option<Vec<u8>> {
+        let sealed = blobs.sealed.remove(&self.storage_key(name))?;
+        unseal(&self.cipher, name, sealed)
     }
 
+    /// `purge` are storage keys straight from `Blobs::leftover`
     pub(crate) async fn commit(
         &self,
         puts: Vec<(String, Vec<u8>)>,
         deletes: Vec<String>,
+        purge: Vec<String>,
     ) -> Result<(), Error> {
         let mut sealed = Vec::with_capacity(puts.len());
-        for (key, value) in puts {
-            let value = self.seal(&key, value)?;
-            sealed.push((key, value));
+        for (name, value) in puts {
+            let value = seal(&self.cipher, &name, value)?;
+            sealed.push((self.storage_key(&name), value));
         }
-        self.backend.commit(sealed, deletes).await
+        let mut removed: Vec<String> = deletes.iter().map(|name| self.storage_key(name)).collect();
+        removed.extend(purge);
+        self.backend.commit(sealed, removed).await
     }
 
-    /// keeps the cipher and owner, drops everything else
+    /// keeps the cipher and owner and drops everything else, in one transaction
     pub(crate) async fn reset(&self) -> Result<(), Error> {
         let keep = self.backend.load(&[CIPHER, ACCOUNT]).await?;
-        self.backend.clear().await?;
-        self.backend
-            .commit(keep.into_iter().collect(), Vec::new())
-            .await
+        self.backend.replace(keep.into_iter().collect()).await
     }
 
-    fn seal(&self, key: &str, value: Vec<u8>) -> Result<Vec<u8>, Error> {
-        let mut plain = Vec::with_capacity(key.len() + 1 + value.len());
-        plain.extend_from_slice(key.as_bytes());
-        plain.push(0);
-        plain.extend_from_slice(&value);
-        drop(value);
-        let encrypted = self
-            .cipher
-            .encrypt_value_base64_data(plain)
-            .map_err(|error| storage_error("sealing a search blob", error))?;
-        serde_json::to_vec(&encrypted)
-            .map_err(|error| storage_error("encoding a search blob", error))
+    fn storage_key(&self, name: &str) -> String {
+        self.cipher
+            .hash_key("macltok-search", name.as_bytes())
+            .iter()
+            .map(|byte| format!("{byte:02x}"))
+            .collect()
     }
 }
 
-fn unseal(cipher: &StoreCipher, key: &str, sealed: &[u8]) -> Option<Vec<u8>> {
-    let encrypted: EncryptedValueBase64 = serde_json::from_slice(sealed).ok()?;
+/// the name goes after the value, so a value packed with a little spare capacity seals
+/// without being copied first
+fn seal(cipher: &StoreCipher, name: &str, mut value: Vec<u8>) -> Result<Vec<u8>, Error> {
+    value.push(0);
+    value.extend_from_slice(name.as_bytes());
+    let encrypted = cipher
+        .encrypt_value_base64_data(value)
+        .map_err(|error| storage_error("sealing a search blob", error))?;
+    serde_json::to_vec(&encrypted).map_err(|error| storage_error("encoding a search blob", error))
+}
+
+fn unseal(cipher: &StoreCipher, name: &str, sealed: Vec<u8>) -> Option<Vec<u8>> {
+    let encrypted: EncryptedValueBase64 = serde_json::from_slice(&sealed).ok()?;
+    drop(sealed);
     let mut plain = cipher.decrypt_value_base64_data(encrypted).ok()?;
-    let prefix = key.len() + 1;
-    if plain.len() < prefix || &plain[..key.len()] != key.as_bytes() || plain[key.len()] != 0 {
+    let suffix = name.len() + 1;
+    let split = plain.len().checked_sub(suffix)?;
+    if plain[split] != 0 || &plain[split + 1..] != name.as_bytes() {
         return None;
     }
-    Some(plain.split_off(prefix))
+    plain.truncate(split);
+    Some(plain)
 }
 
 pub(super) fn storage_error(

Modified file: src/search/store/sqlite.rs+70 additions−27 deletions@@ -1,6 +1,7 @@
 use std::{
     collections::BTreeMap,
-    fs,
+    fs::{self, OpenOptions},
+    io,
     path::{Path, PathBuf},
     time::Duration,
 };
@@ -74,42 +75,54 @@ impl Backend {
     ) -> Result<(), Error> {
         let path = self.path.clone();
         blocking("writing the search store", move || {
-            let mut connection = connect(&path)?;
-            let transaction = connection
-                .transaction_with_behavior(TransactionBehavior::Immediate)
-                .map_err(|error| storage_error("writing the search store", error))?;
-            for key in deletes {
-                transaction
-                    .execute("DELETE FROM blobs WHERE key = ?1", [key])
-                    .map_err(|error| storage_error("writing the search store", error))?;
-            }
-            for (key, value) in puts {
-                transaction
-                    .execute(
-                        "INSERT INTO blobs (key, value) VALUES (?1, ?2) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
-                        params![key, value],
-                    )
-                    .map_err(|error| storage_error("writing the search store", error))?;
-            }
-            transaction
-                .commit()
-                .map_err(|error| storage_error("writing the search store", error))
+            write(&path, false, puts, deletes)
         })
         .await
     }
 
-    pub(super) async fn clear(&self) -> Result<(), Error> {
+    /// everything goes and `puts` arrive in the same transaction
+    pub(super) async fn replace(&self, puts: Vec<(String, Vec<u8>)>) -> Result<(), Error> {
         let path = self.path.clone();
-        blocking("clearing the search store", move || {
-            connect(&path)?
-                .execute("DELETE FROM blobs", [])
-                .map(drop)
-                .map_err(|error| storage_error("clearing the search store", error))
+        blocking("replacing the search store", move || {
+            write(&path, true, puts, Vec::new())
         })
         .await
     }
 }
 
+fn write(
+    path: &Path,
+    clear: bool,
+    puts: Vec<(String, Vec<u8>)>,
+    deletes: Vec<String>,
+) -> Result<(), Error> {
+    let mut connection = connect(path)?;
+    let transaction = connection
+        .transaction_with_behavior(TransactionBehavior::Immediate)
+        .map_err(|error| storage_error("writing the search store", error))?;
+    if clear {
+        transaction
+            .execute("DELETE FROM blobs", [])
+            .map_err(|error| storage_error("clearing the search store", error))?;
+    }
+    for key in deletes {
+        transaction
+            .execute("DELETE FROM blobs WHERE key = ?1", [key])
+            .map_err(|error| storage_error("writing the search store", error))?;
+    }
+    for (key, value) in puts {
+        transaction
+            .execute(
+                "INSERT INTO blobs (key, value) VALUES (?1, ?2) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+                params![key, value],
+            )
+            .map_err(|error| storage_error("writing the search store", error))?;
+    }
+    transaction
+        .commit()
+        .map_err(|error| storage_error("writing the search store", error))
+}
+
 async fn blocking<T: Send + 'static>(
     context: &'static str,
     work: impl FnOnce() -> Result<T, Error> + Send + 'static,
@@ -127,6 +140,8 @@ fn initialise(path: &Path) -> Result<(), Error> {
         fs::create_dir_all(parent)
             .map_err(|error| storage_error("creating the search store directory", error))?;
     }
+    create_private_file(path)
+        .map_err(|error| storage_error("creating the search store file", error))?;
     connect(path)?
         .execute(
             "CREATE TABLE IF NOT EXISTS blobs (key TEXT PRIMARY KEY, value BLOB NOT NULL)",
@@ -136,6 +151,34 @@ fn initialise(path: &Path) -> Result<(), Error> {
         .map_err(|error| storage_error("initialising the search store", error))
 }
 
+/// sqlite gives its wal and shm files the database's own mode, so this covers all three
+#[cfg(unix)]
+fn create_private_file(path: &Path) -> io::Result<()> {
+    use std::os::unix::fs::OpenOptionsExt;
+
+    match OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .mode(0o600)
+        .open(path)
+    {
+        Ok(file) => drop(file),
+        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
+        Err(error) => return Err(error),
+    }
+    Ok(())
+}
+
+#[cfg(not(unix))]
+fn create_private_file(path: &Path) -> io::Result<()> {
+    match OpenOptions::new().write(true).create_new(true).open(path) {
+        Ok(file) => drop(file),
+        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
+        Err(error) => return Err(error),
+    }
+    Ok(())
+}
+
 fn connect(path: &Path) -> Result<Connection, Error> {
     let connection =
         Connection::open(path).map_err(|error| storage_error("opening the search store", error))?;

Added file: src/search/text.rs+328 additions@@ -0,0 +1,328 @@
+use tantivy::{
+    Term,
+    query::{BooleanQuery, Occur, PhraseQuery, Query, TermQuery},
+    schema::{Field, IndexRecordOption},
+    tokenizer::{
+        AsciiFoldingFilter, LowerCaser, RawTokenizer, TextAnalyzer, Token, TokenStream, Tokenizer,
+    },
+};
+
+use crate::{Error, ErrorKind};
+
+pub(crate) const TOKENIZER: &str = "macltok";
+/// longer runs are hashes, keys or noise, and in chars so cyrillic and greek words fit
+const MAX_WORD_CHARS: usize = 64;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum Kind {
+    Word,
+    /// one char of a han, kana, hangul or thai run, or an emoji
+    Single,
+    /// two neighbouring chars of such a run, sharing the first one's position
+    Pair,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+struct Piece {
+    from: usize,
+    to: usize,
+    position: usize,
+    kind: Kind,
+    /// a single inside a run of two or more, which its pairs already cover in a query
+    paired: bool,
+}
+
+/// scripts written without spaces, where a char is the only boundary there is
+fn unspaced(c: char) -> bool {
+    matches!(c,
+        '\u{3040}'..='\u{30FF}' | '\u{31F0}'..='\u{31FF}' | '\u{3400}'..='\u{4DBF}'
+        | '\u{4E00}'..='\u{9FFF}' | '\u{F900}'..='\u{FAFF}' | '\u{FF66}'..='\u{FF9F}'
+        | '\u{1100}'..='\u{11FF}' | '\u{3130}'..='\u{318F}' | '\u{AC00}'..='\u{D7AF}'
+        | '\u{0E00}'..='\u{0E7F}' | '\u{20000}'..='\u{2FA1F}')
+}
+
+fn emoji(c: char) -> bool {
+    matches!(c, '\u{1F300}'..='\u{1FAFF}' | '\u{2600}'..='\u{27BF}')
+}
+
+fn pieces(text: &str) -> Vec<Piece> {
+    let mut pieces = Vec::new();
+    let mut position = 0;
+    let mut chars = text.char_indices().peekable();
+    while let Some((from, c)) = chars.next() {
+        if unspaced(c) {
+            let mut run = vec![(from, from + c.len_utf8())];
+            while let Some(&(at, next)) = chars.peek() {
+                if !unspaced(next) {
+                    break;
+                }
+                run.push((at, at + next.len_utf8()));
+                chars.next();
+            }
+            let paired = run.len() > 1;
+            for (index, &(from, to)) in run.iter().enumerate() {
+                pieces.push(Piece {
+                    from,
+                    to,
+                    position,
+                    kind: Kind::Single,
+                    paired,
+                });
+                if let Some(&(_, next_to)) = run.get(index + 1) {
+                    pieces.push(Piece {
+                        from,
+                        to: next_to,
+                        position,
+                        kind: Kind::Pair,
+                        paired: false,
+                    });
+                }
+                position += 1;
+            }
+        } else if emoji(c) {
+            pieces.push(Piece {
+                from,
+                to: from + c.len_utf8(),
+                position,
+                kind: Kind::Single,
+                paired: false,
+            });
+            position += 1;
+        } else if c.is_alphanumeric() {
+            let mut to = from + c.len_utf8();
+            let mut count = 1;
+            while let Some(&(at, next)) = chars.peek() {
+                if !next.is_alphanumeric() || unspaced(next) {
+                    break;
+                }
+                to = at + next.len_utf8();
+                count += 1;
+                chars.next();
+            }
+            if count <= MAX_WORD_CHARS {
+                pieces.push(Piece {
+                    from,
+                    to,
+                    position,
+                    kind: Kind::Word,
+                    paired: false,
+                });
+                position += 1;
+            }
+        }
+    }
+    pieces
+}
+
+#[derive(Clone)]
+pub(crate) struct MacltokTokenizer;
+
+pub(crate) struct PieceStream {
+    tokens: std::vec::IntoIter<Token>,
+    token: Token,
+}
+
+impl TokenStream for PieceStream {
+    fn advance(&mut self) -> bool {
+        match self.tokens.next() {
+            Some(token) => {
+                self.token = token;
+                true
+            }
+            None => false,
+        }
+    }
+
+    fn token(&self) -> &Token {
+        &self.token
+    }
+
+    fn token_mut(&mut self) -> &mut Token {
+        &mut self.token
+    }
+}
+
+impl Tokenizer for MacltokTokenizer {
+    type TokenStream<'a> = PieceStream;
+
+    fn token_stream<'a>(&'a mut self, text: &'a str) -> PieceStream {
+        let tokens: Vec<_> = pieces(text)
+            .into_iter()
+            .map(|piece| Token {
+                offset_from: piece.from,
+                offset_to: piece.to,
+                position: piece.position,
+                text: text[piece.from..piece.to].to_owned(),
+                position_length: 1,
+            })
+            .collect();
+        PieceStream {
+            tokens: tokens.into_iter(),
+            token: Token::default(),
+        }
+    }
+}
+
+pub(crate) fn analyzer() -> TextAnalyzer {
+    TextAnalyzer::builder(MacltokTokenizer)
+        .filter(LowerCaser)
+        .filter(AsciiFoldingFilter)
+        .build()
+}
+
+fn fold(text: &str) -> String {
+    let mut folder = TextAnalyzer::builder(RawTokenizer::default())
+        .filter(LowerCaser)
+        .filter(AsciiFoldingFilter)
+        .build();
+    let mut stream = folder.token_stream(text);
+    if stream.advance() {
+        stream.token().text.clone()
+    } else {
+        String::new()
+    }
+}
+
+/// the byte ranges in `text` whose analysed token is one of `terms`, overlaps joined
+pub(crate) fn match_ranges(text: &str, terms: &[String]) -> Vec<(usize, usize)> {
+    if terms.is_empty() {
+        return Vec::new();
+    }
+    let mut analyzer = analyzer();
+    let mut stream = analyzer.token_stream(text);
+    let mut ranges: Vec<(usize, usize)> = Vec::new();
+    while stream.advance() {
+        let token = stream.token();
+        if !terms.iter().any(|term| *term == token.text) {
+            continue;
+        }
+        match ranges.last_mut() {
+            Some(last) if token.offset_from < last.1 => last.1 = last.1.max(token.offset_to),
+            _ => ranges.push((token.offset_from, token.offset_to)),
+        }
+    }
+    ranges
+}
+
+enum Clause {
+    Words(String),
+    Phrase(String),
+}
+
+/// quotes make a phrase, a leading `-` excludes, anything else is a word that must match.
+/// no field syntax, so `re: deploy` or a url is just its words
+fn clauses(query: &str) -> Vec<(bool, Clause)> {
+    let mut clauses = Vec::new();
+    let mut rest = query;
+    loop {
+        rest = rest.trim_start();
+        if rest.is_empty() {
+            return clauses;
+        }
+        let (excluded, body) = match rest.strip_prefix('-') {
+            Some(body) if body.starts_with(|c: char| !c.is_whitespace()) => (true, body),
+            _ => (false, rest),
+        };
+        if let Some(quoted) = body.strip_prefix('"') {
+            let end = quoted.find('"').unwrap_or(quoted.len());
+            clauses.push((excluded, Clause::Phrase(quoted[..end].to_owned())));
+            rest = quoted.get(end + 1..).unwrap_or("");
+        } else {
+            let end = body.find(char::is_whitespace).unwrap_or(body.len());
+            clauses.push((excluded, Clause::Words(body[..end].to_owned())));
+            rest = &body[end..];
+        }
+    }
+}
+
+fn term(field: Field, text: &str) -> Box<dyn Query> {
+    Box::new(TermQuery::new(
+        Term::from_field_text(field, text),
+        IndexRecordOption::WithFreqs,
+    ))
+}
+
+/// every term the clause needs, and its query, `None` when nothing in it is searchable
+fn clause_query(field: Field, clause: &Clause) -> Option<(Box<dyn Query>, Vec<String>)> {
+    let text = match clause {
+        Clause::Words(text) | Clause::Phrase(text) => text,
+    };
+    let found = pieces(text);
+    if let Clause::Phrase(_) = clause {
+        let ordered: Vec<_> = found
+            .iter()
+            .filter(|piece| piece.kind != Kind::Pair)
+            .map(|piece| (piece.position, fold(&text[piece.from..piece.to])))
+            .filter(|(_, folded)| !folded.is_empty())
+            .collect();
+        let terms: Vec<String> = ordered.iter().map(|(_, folded)| folded.clone()).collect();
+        return match ordered.len() {
+            0 => None,
+            1 => Some((term(field, &ordered[0].1), terms)),
+            _ => {
+                let first = ordered[0].0;
+                let offsets = ordered
+                    .iter()
+                    .map(|(position, folded)| {
+                        (position - first, Term::from_field_text(field, folded))
+                    })
+                    .collect();
+                Some((Box::new(PhraseQuery::new_with_offset(offsets)), terms))
+            }
+        };
+    }
+    let mut terms: Vec<String> = found
+        .iter()
+        .filter(|piece| !piece.paired)
+        .map(|piece| fold(&text[piece.from..piece.to]))
+        .filter(|folded| !folded.is_empty())
+        .collect();
+    terms.dedup();
+    match terms.len() {
+        0 => None,
+        1 => Some((term(field, &terms[0]), terms)),
+        _ => {
+            let musts = terms
+                .iter()
+                .map(|text| (Occur::Must, term(field, text)))
+                .collect();
+            Some((Box::new(BooleanQuery::new(musts)), terms))
+        }
+    }
+}
+
+/// the query and the terms to highlight. one with nothing left to find is `Invalid`,
+/// rather than everything in scope
+pub(crate) fn build_query(
+    field: Field,
+    query: &str,
+) -> Result<(Box<dyn Query>, Vec<String>), Error> {
+    let mut subqueries = Vec::new();
+    let mut highlights = Vec::new();
+    let mut positive = false;
+    for (excluded, clause) in clauses(query) {
+        let Some((subquery, terms)) = clause_query(field, &clause) else {
+            continue;
+        };
+        if excluded {
+            subqueries.push((Occur::MustNot, subquery));
+        } else {
+            positive = true;
+            highlights.extend(terms);
+            subqueries.push((Occur::Must, subquery));
+        }
+    }
+    if !positive {
+        return Err(Error::new(
+            ErrorKind::Invalid,
+            "search needs at least one word that isn't excluded",
+        ));
+    }
+    highlights.sort();
+    highlights.dedup();
+    Ok((Box::new(BooleanQuery::new(subqueries)), highlights))
+}
+
+#[cfg(test)]
+#[path = "text_tests.rs"]
+mod tests;

Added file: src/search/text_tests.rs+68 additions@@ -0,0 +1,68 @@
+use super::*;
+
+fn tokens(text: &str) -> Vec<String> {
+    let mut analyzer = analyzer();
+    let mut stream = analyzer.token_stream(text);
+    let mut found = Vec::new();
+    while stream.advance() {
+        found.push(stream.token().text.clone());
+    }
+    found
+}
+
+fn highlights(query: &str) -> Result<Vec<String>, ErrorKind> {
+    build_query(Field::from_field_id(0), query)
+        .map(|(_, terms)| terms)
+        .map_err(|error| error.kind)
+}
+
+#[test]
+fn unspaced_scripts_become_singles_and_pairs() {
+    assert_eq!(tokens("東京都"), vec!["東", "東京", "京", "京都", "都"]);
+    assert_eq!(tokens("ok 日本"), vec!["ok", "日", "日本", "本"]);
+}
+
+#[test]
+fn the_word_limit_counts_chars_not_bytes() {
+    let sha = "a".repeat(40);
+    let cyrillic = "щ".repeat(40);
+    assert_eq!(tokens(&format!("{sha} {cyrillic}")).len(), 2);
+    assert!(tokens(&"x".repeat(65)).is_empty());
+}
+
+/// a colon used to mean a field to the query parser, so "re: deploy" lost both words
+#[test]
+fn queries_have_no_field_syntax() {
+    assert_eq!(highlights("re: deploy").unwrap(), vec!["deploy", "re"]);
+    assert_eq!(
+        highlights("https://example.org/x").unwrap(),
+        vec!["example", "https", "org", "x"]
+    );
+    assert_eq!(highlights("sender:@a").unwrap(), vec!["a", "sender"]);
+}
+
+#[test]
+fn nothing_but_exclusions_or_wildcards_is_invalid() {
+    assert_eq!(highlights("-spam"), Err(ErrorKind::Invalid));
+    assert_eq!(highlights("*"), Err(ErrorKind::Invalid));
+    assert_eq!(highlights("   "), Err(ErrorKind::Invalid));
+    assert_eq!(highlights("ham -spam").unwrap(), vec!["ham"]);
+}
+
+#[test]
+fn a_cjk_word_query_needs_its_pairs() {
+    assert_eq!(highlights("東京").unwrap(), vec!["東京"]);
+    assert_eq!(highlights("東").unwrap(), vec!["東"]);
+}
+
+#[test]
+fn match_ranges_join_overlapping_pairs() {
+    assert_eq!(
+        match_ranges("Café, CAFE and cafés", &["cafe".to_owned()]),
+        vec![(0, 5), (7, 11)]
+    );
+    assert_eq!(
+        match_ranges("東京都", &["東京".to_owned(), "京都".to_owned()]),
+        vec![(0, 9)]
+    );
+}

Modified file: src/search_model.rs+24 additions−2 deletions@@ -34,7 +34,9 @@ pub enum SearchCommand {
         limit: u16,
     },
     /// frees the search's media. its hits' media ids stop resolving
-    Close { search_id: String },
+    Close {
+        search_id: String,
+    },
     Coverage(SearchScope),
 }
 
@@ -47,6 +49,13 @@ pub struct SearchSession {
     pub page: SearchPage,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum SearchCloseReason {
+    /// opening a ninth search closed the oldest
+    Evicted,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub struct SearchPage {
@@ -82,10 +91,23 @@ pub struct SearchCoverage {
     pub rooms: Vec<RoomSearchCoverage>,
     /// false when the index lives in memory and starts over with the next engine
     pub persistent: bool,
-    /// messages that couldn't be decrypted yet, so can't be found yet
+    /// why it lives in memory, `None` exactly when `persistent`
+    pub in_memory: Option<SearchInMemory>,
+    /// messages in scope that couldn't be decrypted yet, so can't be found yet
     pub undecrypted: usize,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum SearchInMemory {
+    /// the engine was built without a store, `Engine::from_client` or no store feature
+    NoStore,
+    OpenFailed,
+    ReadFailed,
+    /// five commits in a row failed, so this engine stopped trying
+    CommitFailed,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub struct RoomSearchCoverage {

Modified file: src/transport.rs+12 additions−2 deletions@@ -37,8 +37,8 @@ use crate::{
     PowerLevelChanges, Profile, PushNotification, PushNotificationEvent, PushNotificationStatus,
     RoomCommand, RoomListFilter, RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels,
     RoomSummary, SavedMentionContext, SearchCommand, SearchCoverage, SearchPage, SearchScope,
-    SearchSession, SpaceChild, ThreadListCommand, TimelineCommand, TimelineId,
-    VerificationAction, VerificationTarget,
+    SearchSession, SpaceChild, ThreadListCommand, TimelineCommand, TimelineId, VerificationAction,
+    VerificationTarget,
     encryption::encryption_state,
     engine::{MediaService, StoreLock},
     model::{PushNotificationRoom, PushNotificationSender},
@@ -1660,6 +1660,16 @@ pub(crate) struct EventSender {
 }
 
 impl EventSender {
+    #[cfg(all(test, feature = "sqlite"))]
+    pub(crate) fn detached(capacity: usize) -> (Self, mpsc::Receiver<Event>) {
+        let (events, receiver) = mpsc::channel(capacity);
+        let sender = Self {
+            events,
+            publication: Arc::new(PublicationRegistry::default()),
+        };
+        (sender, receiver)
+    }
+
     pub(crate) async fn send(&self, event: Event) -> Result<(), SendEventError> {
         self.events.send(event).await.map_err(|_| SendEventError)
     }

Modified file: tests/history_repair.rs+6 additions−2 deletions@@ -40,7 +40,9 @@ async fn start_room(room_id: &RoomId) -> TestRoom {
     server.mock_room_state_encryption().plain().mount().await;
     server.sync_joined_room(&client, room_id).await;
     let sync_client = client.clone();
-    let config = EngineConfig::default().with_initial_history_limit(0);
+    let config = EngineConfig::default()
+        .with_initial_history_limit(0)
+        .with_search_backfill_pages_per_room(0);
     let (engine, handle, events) = Engine::from_client(client, config).await.unwrap();
     let running = tokio::spawn(engine.run());
     TestRoom {
@@ -304,7 +306,9 @@ async fn fill_gap_returns_after_the_mirror_contains_the_committed_rows() {
     let middle_id = event_id!("$gap-middle:localhost");
     let (engine, handle, events) = Engine::from_client(
         client,
-        EngineConfig::default().with_initial_history_limit(0),
+        EngineConfig::default()
+            .with_initial_history_limit(0)
+            .with_search_backfill_pages_per_room(0),
     )
     .await
     .unwrap();

Modified file: tests/mentions_inbox.rs+30 additions−12 deletions@@ -334,7 +334,9 @@ async fn unopened_room_mentions_are_deduplicated_and_read_through_uses_a_revisio
 
     let (engine, handle, mut events) = Engine::from_client(
         client,
-        EngineConfig::default().with_initial_history_limit(0),
+        EngineConfig::default()
+            .with_initial_history_limit(0)
+            .with_search_backfill_pages_per_room(0),
     )
     .await
     .unwrap();
@@ -448,7 +450,9 @@ async fn edits_can_remove_and_restore_a_mention_until_redaction_removes_its_targ
 
     let (engine, handle, mut events) = Engine::from_client(
         client,
-        EngineConfig::default().with_initial_history_limit(0),
+        EngineConfig::default()
+            .with_initial_history_limit(0)
+            .with_search_backfill_pages_per_room(0),
     )
     .await
     .unwrap();
@@ -591,7 +595,9 @@ async fn room_mentions_require_power_while_muted_direct_mentions_remain_and_igno
 
     let (engine, handle, mut events) = Engine::from_client(
         client,
-        EngineConfig::default().with_initial_history_limit(0),
+        EngineConfig::default()
+            .with_initial_history_limit(0)
+            .with_search_backfill_pages_per_room(0),
     )
     .await
     .unwrap();
@@ -679,7 +685,9 @@ async fn event_context_uses_independent_direction_limits_and_saved_mention_conte
     feed.mount(&server).await;
     let (engine, handle, mut events) = Engine::from_client(
         client,
-        EngineConfig::default().with_initial_history_limit(0),
+        EngineConfig::default()
+            .with_initial_history_limit(0)
+            .with_search_backfill_pages_per_room(0),
     )
     .await
     .unwrap();
@@ -836,7 +844,9 @@ async fn highlighting_keywords_are_mentions_and_quiet_ones_are_not() {
 
     let (engine, handle, mut events) = Engine::from_client(
         client,
-        EngineConfig::default().with_initial_history_limit(0),
+        EngineConfig::default()
+            .with_initial_history_limit(0)
+            .with_search_backfill_pages_per_room(0),
     )
     .await
     .unwrap();
@@ -917,9 +927,12 @@ async fn adding_a_keyword_gives_its_rule_a_highlight() {
         .mount(&server)
         .await;
 
-    let (engine, handle, mut events) = Engine::from_client(client, EngineConfig::default())
-        .await
-        .unwrap();
+    let (engine, handle, mut events) = Engine::from_client(
+        client,
+        EngineConfig::default().with_search_backfill_pages_per_room(0),
+    )
+    .await
+    .unwrap();
     let running = tokio::spawn(engine.run());
     handle.add_notification_keyword("deploy").await.unwrap();
     shut_down(&handle, &mut events, running).await;
@@ -963,9 +976,12 @@ async fn a_quiet_keyword_rule_only_gains_the_highlight() {
         .mount(&server)
         .await;
 
-    let (engine, handle, mut events) = Engine::from_client(client, EngineConfig::default())
-        .await
-        .unwrap();
+    let (engine, handle, mut events) = Engine::from_client(
+        client,
+        EngineConfig::default().with_search_backfill_pages_per_room(0),
+    )
+    .await
+    .unwrap();
     let running = tokio::spawn(engine.run());
     handle.add_notification_keyword("deploy").await.unwrap();
     shut_down(&handle, &mut events, running).await;
@@ -1022,7 +1038,9 @@ async fn sqlite_mentions_survive_engine_shutdown_and_session_resume() {
     );
     feed.mount(&server).await;
 
-    let config = EngineConfig::default().with_initial_history_limit(0);
+    let config = EngineConfig::default()
+        .with_initial_history_limit(0)
+        .with_search_backfill_pages_per_room(0);
     let session = Session::login_password(
         PasswordLogin::new(server.uri(), "example", "wordpass").allow_insecure_loopback(true),
         sessions.clone(),

macltok

Commit 74befbc72baf

Commit `74befbc72baf`