macltok

public

Star0

Fork 0

Watch0

57 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `80381e06a573`

tarball

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

15 files changed+2215-66

timeline: expose bounded room history repair

Modified file: Cargo.lock+10 additions−10 deletions@@ -2002,7 +2002,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "anymap2",
  "aquamarine",
@@ -2065,7 +2065,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-base"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "as_variant",
  "assert_matches",
@@ -2095,7 +2095,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-common"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "eyeball-im",
  "futures-core",
@@ -2118,7 +2118,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-crypto"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "aes",
  "aquamarine",
@@ -2159,7 +2159,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-indexeddb"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "async-trait",
  "base64",
@@ -2191,7 +2191,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-sqlite"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "as_variant",
  "async-trait",
@@ -2218,7 +2218,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-store-encryption"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "base64",
  "blake3",
@@ -2239,7 +2239,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-test"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "as_variant",
  "ctor",
@@ -2262,7 +2262,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-test-macros"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "quote",
  "syn 2.0.119",
@@ -2271,7 +2271,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-ui"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=9b9b16982dcab5dd499547bb87e121b9d0cba6b5#9b9b16982dcab5dd499547bb87e121b9d0cba6b5"
 dependencies = [
  "as_variant",
  "async-rx",

Modified file: Cargo.toml+9 additions−9 deletions@@ -26,11 +26,11 @@ indexeddb = [
 futures-util = { version = "0.3.31", default-features = false, features = ["std"] }
 http = "1"
 imbl = { version = "6.1.0", default-features = false }
-matrix-sdk = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48", default-features = false, features = ["e2e-encryption"] }
-matrix-sdk-base = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48", default-features = false }
-matrix-sdk-crypto = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48", default-features = false }
-matrix-sdk-store-encryption = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48", optional = true }
-matrix-sdk-ui = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48", default-features = false }
+matrix-sdk = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5", default-features = false, features = ["e2e-encryption"] }
+matrix-sdk-base = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5", default-features = false }
+matrix-sdk-crypto = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5", default-features = false }
+matrix-sdk-store-encryption = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5", optional = true }
+matrix-sdk-ui = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5", default-features = false }
 ruma = { version = "0.16.0", default-features = false, features = ["unstable-msc2545", "unstable-msc4095"] }
 serde = { version = "1.0.219", features = ["derive", "rc"], optional = true }
 serde_json = "1.0.143"
@@ -48,8 +48,8 @@ getrandom3 = { package = "getrandom", version = "0.3", features = ["wasm_js"] }
 getrandom4 = { package = "getrandom", version = "0.4", features = ["wasm_js"] }
 indexed_db_futures = { package = "matrix_indexed_db_futures", version = "0.7.0", optional = true, features = ["typed-arrays"] }
 js-sys = { version = "0.3.105", optional = true }
-matrix-sdk = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48", default-features = false, features = ["js"] }
-matrix-sdk-ui = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48", default-features = false, features = ["js"] }
+matrix-sdk = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5", default-features = false, features = ["js"] }
+matrix-sdk-ui = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5", default-features = false, features = ["js"] }
 # the sdk's own wasm reqwest leaves out stream, which the bounded media download needs
 reqwest = { version = "0.13.1", default-features = false, features = ["stream"] }
 uuid = { version = "1", features = ["js"] }
@@ -64,8 +64,8 @@ web-sys = { version = "0.3.105", optional = true, features = [
 ] }
 
 [target.'cfg(not(target_family = "wasm"))'.dev-dependencies]
-matrix-sdk = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48", default-features = false, features = ["testing"] }
-matrix-sdk-test = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "7aace883fbe03d4c3c118b55188fe9519ba8ce48" }
+matrix-sdk = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5", default-features = false, features = ["testing"] }
+matrix-sdk-test = { git = "ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk", rev = "9b9b16982dcab5dd499547bb87e121b9d0cba6b5" }
 tokio = { version = "1.47.1", features = ["macros", "rt-multi-thread", "signal", "time"] }
 wiremock = "0.6.5"
 

Modified file: src/engine/mod.rs+53 additions@@ -5,6 +5,7 @@ mod media;
 mod mutations;
 mod packs;
 mod prepare;
+mod repair;
 mod room;
 mod session;
 mod timeline;
@@ -325,6 +326,8 @@ pub struct Engine {
     mentions: crate::mentions::service::Handle,
     mentions_service: Option<crate::mentions::service::Service>,
     mentions_updates: Option<watch::Receiver<crate::MentionsPage>>,
+    repair_gate: Arc<tokio::sync::Mutex<()>>,
+    repair_shutdown: Arc<std::sync::atomic::AtomicBool>,
 }
 
 impl Engine {
@@ -434,6 +437,8 @@ impl Engine {
             mentions,
             mentions_service: Some(mentions_service),
             mentions_updates: Some(mentions_updates),
+            repair_gate: Arc::default(),
+            repair_shutdown: Arc::default(),
         };
         Ok((engine, handle, events))
     }
@@ -838,6 +843,7 @@ impl Engine {
                         reply.send(mentions.execute(command).await);
                     });
             }
+            Command::HistoryRepair(command) => self.route_repair(command, reply),
             Command::Media(MediaCommand::Upload(upload)) => self.upload(upload, reply),
             Command::Encryption(command) => enqueue(&self.encryption, Job { command, reply }),
             Command::Notification(command) => {
@@ -976,6 +982,44 @@ impl Engine {
         }
     }
 
+    fn route_repair(&mut self, command: crate::HistoryRepairCommand, reply: Responder) {
+        use crate::HistoryRepairCommand;
+        let room_id = match &command {
+            HistoryRepairCommand::Start { room_id, .. }
+            | HistoryRepairCommand::Continue { room_id, .. }
+            | HistoryRepairCommand::Cancel { room_id }
+            | HistoryRepairCommand::StartRecent { room_id } => room_id,
+            HistoryRepairCommand::GetPolicy | HistoryRepairCommand::SetPolicy(_) => {
+                let mentions = self.mentions.clone();
+                self.tasks
+                    .spawn(TaskContext::Command("history repair policy"), async move {
+                        let result = match command {
+                            HistoryRepairCommand::GetPolicy => mentions
+                                .repair_policy()
+                                .await
+                                .map(crate::CommandOutput::HistoryRepairPolicy),
+                            HistoryRepairCommand::SetPolicy(policy) => mentions
+                                .set_repair_policy(policy)
+                                .await
+                                .map(|_| crate::CommandOutput::Done),
+                            _ => unreachable!(),
+                        };
+                        reply.send(result);
+                    });
+                return;
+            }
+        };
+        match self.rooms.get(room_id) {
+            Some(RoomSlot::Active { sender, .. }) => {
+                send_actor(sender, ActorRequest::Repair(command, reply))
+            }
+            Some(RoomSlot::Closing { .. }) => {
+                reply.done(Err(Error::new(ErrorKind::RoomBusy, "room is closing")))
+            }
+            None => reply.done(Err(Error::new(ErrorKind::RoomNotOpen, "room is not open"))),
+        }
+    }
+
     async fn open_room(&mut self, room_id: OwnedRoomId, reply: Responder) {
         match self.rooms.get_mut(&room_id) {
             Some(RoomSlot::Active { sender, .. }) => send_actor(sender, ActorRequest::Open(reply)),
@@ -1039,6 +1083,10 @@ impl Engine {
                 current_user_id: self.client.user_id().map(ToOwned::to_owned),
                 initial_history_limit: self.config.initial_history_limit,
                 supervisor,
+                mentions: self.mentions.clone(),
+                sync: self.sync.clone(),
+                repair_gate: self.repair_gate.clone(),
+                repair_shutdown: self.repair_shutdown.clone(),
             },
             panics,
             opening: vec![reply],
@@ -1094,6 +1142,7 @@ impl Engine {
                 ActorRequest::Open(reply) => self.open_room(room_id.clone(), reply).await,
                 ActorRequest::Close(reply) => reply.done(Ok(())),
                 ActorRequest::Execute(command, reply) => self.route_timeline(command, reply),
+                ActorRequest::Repair(command, reply) => self.route_repair(command, reply),
             }
         }
         for deferred in pending {
@@ -1116,11 +1165,15 @@ impl Engine {
     }
 
     async fn shutdown(&mut self) {
+        self.repair_shutdown
+            .store(true, std::sync::atomic::Ordering::Release);
         if let Some(handler) = self.verification_handler.take() {
             self.client.remove_event_handler(handler);
         }
         self.port.close();
+        let repair_gate = self.repair_gate.lock().await;
         self.sync.stop().await;
+        drop(repair_gate);
         if self.mentions_service.is_none() {
             self.mentions.stop().await;
         }

Added file: src/engine/repair.rs+608 additions@@ -0,0 +1,608 @@
+use std::{
+    collections::HashSet,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::Duration,
+};
+
+use matrix_sdk::{Room, deserialized_responses::TimelineEvent, room::MessagesOptions};
+use matrix_sdk_ui::sync_service::SyncService;
+use ruma::{OwnedEventId, time::Instant};
+use tokio::sync::{OwnedMutexGuard, mpsc, oneshot, watch};
+
+use super::{
+    executor::{JoinHandle, spawn},
+    timeline::SessionContext,
+};
+use crate::{
+    CommandOutput, Error, ErrorKind, Event, HistoryContinuity, HistoryRepairBudget,
+    HistoryRepairCommand, HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairReport,
+    HistoryRepairStop, HistoryRepairStrategy, transport::Responder,
+};
+
+const PAGE_SIZE: u16 = 50;
+const STAGING_LIMIT: usize = 10_000;
+
+pub(super) struct InstallRequest {
+    pub(super) events: Vec<TimelineEvent>,
+    pub(super) prev_batch: Option<String>,
+    pub(super) reply: oneshot::Sender<Result<(), Error>>,
+}
+
+struct Stage {
+    events: Vec<TimelineEvent>,
+    cursor: Option<String>,
+    anchor: Option<OwnedEventId>,
+    report: HistoryRepairReport,
+}
+
+struct Running {
+    task: JoinHandle<Stage>,
+    cancel: watch::Sender<bool>,
+    reply: Responder,
+}
+
+pub(super) struct RepairController {
+    room: Room,
+    context: SessionContext,
+    installs: mpsc::Sender<InstallRequest>,
+    requests: mpsc::Receiver<InstallRequest>,
+    running: Option<Running>,
+    stage: Option<Stage>,
+}
+
+impl RepairController {
+    pub(super) fn new(room: Room, context: SessionContext) -> Self {
+        let (installs, requests) = mpsc::channel(1);
+        Self {
+            room,
+            context,
+            installs,
+            requests,
+            running: None,
+            stage: None,
+        }
+    }
+
+    pub(super) fn is_running(&self) -> bool {
+        self.running.is_some()
+    }
+
+    pub(super) async fn execute(&mut self, command: HistoryRepairCommand, reply: Responder) {
+        if matches!(command, HistoryRepairCommand::Cancel { .. }) {
+            if let Some(running) = &self.running {
+                running.cancel.send_replace(true);
+            }
+            reply.done(Ok(()));
+            return;
+        }
+        if self.is_running() {
+            reply.done(Err(Error::new(
+                ErrorKind::RoomBusy,
+                "room history repair is running",
+            )));
+            return;
+        }
+        let (stage, budget) = match command {
+            HistoryRepairCommand::Start { policy, budget, .. } => {
+                if let Err(error) = validate_budget(&budget) {
+                    reply.done(Err(error));
+                    return;
+                }
+                let policy = match policy {
+                    Some(policy) => policy,
+                    None => match self.context.mentions.repair_policy().await {
+                        Ok(policy) => policy,
+                        Err(error) => {
+                            reply.done(Err(error));
+                            return;
+                        }
+                    },
+                };
+                let coverage = match self
+                    .context
+                    .mentions
+                    .repair_coverage(self.room.room_id().to_owned())
+                    .await
+                {
+                    Ok(coverage) => coverage,
+                    Err(error) => {
+                        reply.done(Err(error));
+                        return;
+                    }
+                };
+                let duration = coverage.duration_ms();
+                let report = HistoryRepairReport {
+                    room_id: self.room.room_id().to_owned(),
+                    requested_policy: policy.clone(),
+                    strategy: HistoryRepairStrategy::Recent,
+                    coverage_duration_ms: duration,
+                    threshold_evaluated: duration.is_some()
+                        && coverage.unresolved
+                        && matches!(policy, HistoryRepairPolicy::BridgeShortGaps { .. }),
+                    pages: 0,
+                    events: 0,
+                    phase: HistoryRepairPhase::FetchingRecent,
+                    continuity: HistoryContinuity::Unresolved,
+                    recent_available: false,
+                    stop: None,
+                    failure: None,
+                };
+                (
+                    Stage {
+                        events: Vec::new(),
+                        cursor: None,
+                        anchor: None,
+                        report,
+                    },
+                    budget,
+                )
+            }
+            HistoryRepairCommand::Continue { budget, .. } => {
+                if let Err(error) = validate_budget(&budget) {
+                    reply.done(Err(error));
+                    return;
+                }
+                let Some(mut stage) = self.stage.take() else {
+                    reply.done(Err(Error::new(
+                        ErrorKind::Invalid,
+                        "no history bridge to continue",
+                    )));
+                    return;
+                };
+                if !stage.report.recent_available
+                    || stage.report.continuity != HistoryContinuity::Unresolved
+                {
+                    reply.done(Err(Error::new(
+                        ErrorKind::Invalid,
+                        "history bridge is not resumable",
+                    )));
+                    self.stage = Some(stage);
+                    return;
+                }
+                stage.report.strategy = HistoryRepairStrategy::Bridge;
+                stage.report.phase = HistoryRepairPhase::Bridging;
+                stage.report.stop = None;
+                stage.report.failure = None;
+                (stage, budget)
+            }
+            HistoryRepairCommand::StartRecent { .. } => {
+                let Some(mut stage) = self.stage.take() else {
+                    reply.done(Err(Error::new(
+                        ErrorKind::Invalid,
+                        "no recent history repair",
+                    )));
+                    return;
+                };
+                if !stage.report.recent_available {
+                    self.stage = Some(stage);
+                    reply.done(Err(Error::new(
+                        ErrorKind::Invalid,
+                        "recent history is not available, retry the repair",
+                    )));
+                    return;
+                }
+                stage.report.strategy = HistoryRepairStrategy::Recent;
+                stage.report.phase = HistoryRepairPhase::Ready;
+                stage.report.stop = None;
+                stage.report.failure = None;
+                let report = stage.report.clone();
+                let published = self
+                    .context
+                    .events
+                    .publish_repair(report.clone())
+                    .await
+                    .map_err(|_| Error::interrupted());
+                reply.send(published.map(|_| CommandOutput::HistoryRepair(report)));
+                return;
+            }
+            _ => {
+                reply.done(Err(Error::new(
+                    ErrorKind::Invalid,
+                    "repair command was misrouted",
+                )));
+                return;
+            }
+        };
+        let (cancel, receiver) = watch::channel(false);
+        let room = self.room.clone();
+        let context = self.context.clone();
+        let installs = self.installs.clone();
+        let task =
+            spawn(async move { run(room, context, installs, receiver, stage, budget).await });
+        self.running = Some(Running {
+            task,
+            cancel,
+            reply,
+        });
+    }
+
+    pub(super) async fn next_install(&mut self) -> Option<InstallRequest> {
+        let running = self.running.as_mut()?;
+        tokio::select! {
+            request = self.requests.recv() => request,
+            result = &mut running.task => {
+                let running = self.running.take().expect("repair task exists");
+                match result {
+                    Ok(stage) => {
+                        let report = stage.report.clone();
+                        let published = self.context.events.publish_repair(report.clone()).await
+                            .map_err(|_| Error::interrupted());
+                        running.reply.send(published.map(|_| CommandOutput::HistoryRepair(report)));
+                        self.stage = Some(stage);
+                    }
+                    Err(_) => running.reply.done(Err(Error::interrupted())),
+                }
+                None
+            }
+        }
+    }
+
+    pub(super) async fn shutdown(&mut self) {
+        if let Some(mut running) = self.running.take() {
+            running.task.abort();
+            match (&mut running.task).await {
+                Ok(stage) => {
+                    let report = stage.report;
+                    let published = self
+                        .context
+                        .events
+                        .publish_repair(report.clone())
+                        .await
+                        .map_err(|_| Error::interrupted());
+                    running
+                        .reply
+                        .send(published.map(|_| CommandOutput::HistoryRepair(report)));
+                }
+                Err(_) => running.reply.done(Err(Error::interrupted())),
+            }
+        }
+    }
+}
+
+impl Drop for RepairController {
+    fn drop(&mut self) {
+        if let Some(running) = &self.running {
+            running.task.abort();
+        }
+    }
+}
+
+fn validate_budget(budget: &HistoryRepairBudget) -> Result<(), Error> {
+    if !(1..=100).contains(&budget.max_pages)
+        || !(50..=10_000).contains(&budget.max_events)
+        || !(1..=60_000).contains(&budget.max_duration_ms)
+    {
+        return Err(Error::new(
+            ErrorKind::Invalid,
+            "repair budget requires 1 to 100 pages, 50 to 10000 events, and 1 to 60000 milliseconds",
+        ));
+    }
+    Ok(())
+}
+
+fn should_bridge(policy: &HistoryRepairPolicy, duration: Option<u64>, boundary: bool) -> bool {
+    match policy {
+        HistoryRepairPolicy::Bridge => true,
+        HistoryRepairPolicy::StartRecent => false,
+        HistoryRepairPolicy::BridgeShortGaps { threshold_ms } => {
+            boundary
+                && *threshold_ms > 0
+                && duration.is_some_and(|duration| duration <= *threshold_ms)
+        }
+    }
+}
+
+async fn run(
+    room: Room,
+    context: SessionContext,
+    installs: mpsc::Sender<InstallRequest>,
+    mut cancel: watch::Receiver<bool>,
+    mut stage: Stage,
+    budget: HistoryRepairBudget,
+) -> Stage {
+    let started = Instant::now();
+    let initial_pages = stage.report.pages;
+    let initial_events = stage.report.events;
+    let result = async {
+        if !stage.report.recent_available {
+            let (cache, _handles) = room.event_cache().await.map_err(cache_error)?;
+            stage.anchor = cache
+                .events()
+                .await
+                .map_err(cache_error)?
+                .iter()
+                .find_map(TimelineEvent::event_id);
+            let _ = context
+                .events
+                .send(Event::HistoryRepairChanged(stage.report.clone()))
+                .await;
+            let guard = pause_sync(&context, &mut cancel, &started, &budget).await?;
+            let page = fetch(&room, None, &mut cancel, &started, &budget).await?;
+            stage.report.pages += 1;
+            stage.report.events += page.chunk.len() as u32;
+            stage.cursor = page.end;
+            stage.events = page.chunk.into_iter().rev().collect();
+            deduplicate(&mut stage.events);
+            install(&installs, &stage).await?;
+            stage.report.recent_available = true;
+            if contains_anchor(&stage) {
+                stage.report.strategy = HistoryRepairStrategy::Overlap;
+                stage.report.continuity = HistoryContinuity::Verified;
+            } else if stage.cursor.is_none() {
+                stage.report.continuity = HistoryContinuity::AccessibleStart;
+            }
+            if stage.report.continuity != HistoryContinuity::Unresolved {
+                context
+                    .mentions
+                    .repair_resolved(room.room_id().to_owned())
+                    .await?;
+            }
+            guard.finish().await;
+            let _ = context
+                .events
+                .send(Event::HistoryRepairChanged(stage.report.clone()))
+                .await;
+            if stage.report.continuity != HistoryContinuity::Unresolved {
+                return Ok(());
+            }
+            if !should_bridge(
+                &stage.report.requested_policy,
+                stage.report.coverage_duration_ms,
+                stage.report.threshold_evaluated,
+            ) {
+                return Ok(());
+            }
+            stage.report.strategy = HistoryRepairStrategy::Bridge;
+        }
+        stage.report.phase = HistoryRepairPhase::Bridging;
+        let _ = context
+            .events
+            .send(Event::HistoryRepairChanged(stage.report.clone()))
+            .await;
+        if contains_anchor(&stage) {
+            stage.report.continuity = HistoryContinuity::Verified;
+        } else if stage.cursor.is_none() {
+            stage.report.continuity = HistoryContinuity::AccessibleStart;
+        }
+        while stage.report.continuity == HistoryContinuity::Unresolved {
+            if *cancel.borrow() {
+                stage.report.stop = Some(HistoryRepairStop::Cancelled);
+                break;
+            }
+            if stage.report.pages - initial_pages + 1 >= u32::from(budget.max_pages)
+                || stage.report.events - initial_events + 2 * u32::from(PAGE_SIZE)
+                    > budget.max_events
+                || stage.events.len() + 2 * usize::from(PAGE_SIZE) > STAGING_LIMIT
+                || started.elapsed() >= Duration::from_millis(budget.max_duration_ms)
+            {
+                stage.report.stop = Some(HistoryRepairStop::Budget);
+                break;
+            }
+            if stage.anchor.is_none() {
+                stage.report.stop = Some(HistoryRepairStop::NoAnchor);
+                break;
+            }
+            let Some(cursor) = stage.cursor.clone() else {
+                break;
+            };
+            let page = fetch(&room, Some(&cursor), &mut cancel, &started, &budget).await?;
+            stage.report.pages += 1;
+            stage.report.events += page.chunk.len() as u32;
+            if page.end.as_deref() == Some(&cursor) {
+                stage.report.stop = Some(HistoryRepairStop::CursorStalled);
+                break;
+            }
+            let mut older: Vec<_> = page.chunk.into_iter().rev().collect();
+            older.append(&mut stage.events);
+            deduplicate(&mut older);
+            stage.events = older;
+            stage.cursor = page.end;
+            if contains_anchor(&stage) {
+                stage.report.continuity = HistoryContinuity::Verified;
+            } else if stage.cursor.is_none() {
+                stage.report.continuity = HistoryContinuity::AccessibleStart;
+            }
+            if stage.report.continuity == HistoryContinuity::Unresolved {
+                let _ = context
+                    .events
+                    .send(Event::HistoryRepairChanged(stage.report.clone()))
+                    .await;
+            }
+        }
+        if stage.report.continuity != HistoryContinuity::Unresolved
+            && stage.report.strategy == HistoryRepairStrategy::Bridge
+        {
+            let guard = pause_sync(&context, &mut cancel, &started, &budget).await?;
+            let head = fetch(&room, None, &mut cancel, &started, &budget).await?;
+            stage.report.pages += 1;
+            stage.report.events += head.chunk.len() as u32;
+            let head: Vec<_> = head.chunk.into_iter().rev().collect();
+            if !extend_head(&mut stage.events, head) {
+                stage.report.continuity = HistoryContinuity::Unresolved;
+                stage.report.stop = Some(HistoryRepairStop::HistoryUnavailable);
+                return Ok(());
+            }
+            install(&installs, &stage).await?;
+            context
+                .mentions
+                .repair_resolved(room.room_id().to_owned())
+                .await?;
+            guard.finish().await;
+        }
+        Ok::<_, Error>(())
+    }
+    .await;
+    if let Err(error) = result {
+        if error.kind == ErrorKind::Interrupted {
+            stage.report.stop = Some(HistoryRepairStop::Cancelled);
+        } else if error.kind == ErrorKind::RoomBusy {
+            stage.report.stop = Some(HistoryRepairStop::Budget);
+        } else {
+            stage.report.failure = Some(error);
+        }
+        stage.report.continuity = HistoryContinuity::Unresolved;
+    }
+    stage.report.phase = if stage.report.stop.is_some() || stage.report.failure.is_some() {
+        HistoryRepairPhase::Paused
+    } else {
+        HistoryRepairPhase::Ready
+    };
+    stage
+}
+
+fn contains_anchor(stage: &Stage) -> bool {
+    stage.anchor.as_ref().is_some_and(|anchor| {
+        stage
+            .events
+            .iter()
+            .any(|event| event.event_id().as_ref() == Some(anchor))
+    })
+}
+
+fn deduplicate(events: &mut Vec<TimelineEvent>) {
+    let mut seen = HashSet::new();
+    events.retain(|event| event.event_id().is_some_and(|id| seen.insert(id)));
+}
+
+fn extend_head(events: &mut Vec<TimelineEvent>, head: Vec<TimelineEvent>) -> bool {
+    let ids: HashSet<_> = events.iter().filter_map(TimelineEvent::event_id).collect();
+    let Some(overlap) = head
+        .iter()
+        .position(|event| event.event_id().is_some_and(|id| ids.contains(&id)))
+    else {
+        return events.is_empty() && head.is_empty();
+    };
+    events.extend(head.into_iter().skip(overlap + 1));
+    deduplicate(events);
+    true
+}
+
+async fn install(installs: &mpsc::Sender<InstallRequest>, stage: &Stage) -> Result<(), Error> {
+    let (reply, response) = oneshot::channel();
+    installs
+        .send(InstallRequest {
+            events: stage.events.clone(),
+            prev_batch: stage.cursor.clone(),
+            reply,
+        })
+        .await
+        .map_err(|_| Error::interrupted())?;
+    response.await.map_err(|_| Error::interrupted())?
+}
+
+async fn fetch(
+    room: &Room,
+    from: Option<&str>,
+    cancel: &mut watch::Receiver<bool>,
+    started: &Instant,
+    budget: &HistoryRepairBudget,
+) -> Result<matrix_sdk::room::Messages, Error> {
+    let remaining = Duration::from_millis(budget.max_duration_ms).saturating_sub(started.elapsed());
+    if *cancel.borrow() {
+        return Err(Error::interrupted());
+    }
+    let mut options = MessagesOptions::backward().from(from);
+    options.limit = PAGE_SIZE.into();
+    tokio::select! {
+        _ = cancel.changed() => Err(Error::interrupted()),
+        result = matrix_sdk::timeout::timeout(room.messages(options), remaining) => {
+            let page = result.map_err(|_| Error::new(ErrorKind::RoomBusy, "repair work duration exhausted"))?
+                .map_err(|source| Error::from_sdk("fetching repair history failed", source))?;
+            if page.chunk.len() > usize::from(PAGE_SIZE) {
+                return Err(Error::new(ErrorKind::Invalid, "server exceeded the repair page limit"));
+            }
+            Ok(page)
+        }
+    }
+}
+
+struct SyncPause {
+    sync: Arc<SyncService>,
+    shutdown: Arc<AtomicBool>,
+    gate: Option<OwnedMutexGuard<()>>,
+}
+
+impl SyncPause {
+    async fn finish(mut self) {
+        if !self.shutdown.load(Ordering::Acquire) {
+            self.sync.start().await;
+        }
+        self.gate.take();
+    }
+}
+
+impl Drop for SyncPause {
+    fn drop(&mut self) {
+        if let Some(gate) = self.gate.take() {
+            let sync = self.sync.clone();
+            let shutdown = self.shutdown.clone();
+            spawn(async move {
+                if !shutdown.load(Ordering::Acquire) {
+                    sync.start().await;
+                }
+                drop(gate);
+            });
+        }
+    }
+}
+
+async fn pause_sync(
+    context: &SessionContext,
+    cancel: &mut watch::Receiver<bool>,
+    started: &Instant,
+    budget: &HistoryRepairBudget,
+) -> Result<SyncPause, Error> {
+    let remaining = Duration::from_millis(budget.max_duration_ms).saturating_sub(started.elapsed());
+    let gate = tokio::select! {
+        _ = cancel.changed() => return Err(Error::interrupted()),
+        result = matrix_sdk::timeout::timeout(context.repair_gate.clone().lock_owned(), remaining) =>
+            result.map_err(|_| Error::new(ErrorKind::RoomBusy, "repair work duration exhausted"))?,
+    };
+    if context.repair_shutdown.load(Ordering::Acquire) {
+        return Err(Error::interrupted());
+    }
+    let guard = SyncPause {
+        sync: context.sync.clone(),
+        shutdown: context.repair_shutdown.clone(),
+        gate: Some(gate),
+    };
+    context.sync.stop().await;
+    Ok(guard)
+}
+
+fn cache_error(source: matrix_sdk::event_cache::EventCacheError) -> Error {
+    Error::with_source(
+        ErrorKind::Store,
+        "room history cache operation failed",
+        source,
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn threshold_requires_known_coverage_and_an_independent_boundary() {
+        let policy = HistoryRepairPolicy::BridgeShortGaps { threshold_ms: 100 };
+        assert!(should_bridge(&policy, Some(100), true));
+        assert!(should_bridge(&policy, Some(99), true));
+        assert!(!should_bridge(&policy, Some(101), true));
+        assert!(!should_bridge(&policy, None, true));
+        assert!(!should_bridge(&policy, Some(1), false));
+        assert!(!should_bridge(
+            &HistoryRepairPolicy::BridgeShortGaps { threshold_ms: 0 },
+            Some(0),
+            true
+        ));
+        assert!(should_bridge(&HistoryRepairPolicy::Bridge, None, false));
+        assert!(!should_bridge(
+            &HistoryRepairPolicy::StartRecent,
+            Some(1),
+            true
+        ));
+    }
+}

Modified file: src/engine/room.rs+106 additions−4 deletions@@ -6,22 +6,28 @@ use ruma::OwnedRoomId;
 use tokio::sync::mpsc;
 
 use super::executor::{AbortHandle, JoinError, TaskSet, is_panic, resume_unwind};
+use super::repair::{InstallRequest, RepairController};
 use super::timeline::{SessionContext, TaskGuard, TimelineSession};
 use crate::{
-    Error, ErrorKind, Event, TimelineCommand, TimelineId, TypingState, transport::Responder,
+    Error, ErrorKind, Event, HistoryRepairCommand, TimelineCommand, TimelineId, TypingState,
+    transport::Responder,
 };
 
 pub(super) enum ActorRequest {
     /// opening an open room answers with a fresh reset
     Open(Responder),
     Execute(TimelineCommand, Responder),
+    Repair(HistoryRepairCommand, Responder),
     Close(Responder),
 }
 
 impl ActorRequest {
     pub(super) fn into_reply(self) -> Responder {
         match self {
-            Self::Open(reply) | Self::Execute(_, reply) | Self::Close(reply) => reply,
+            Self::Open(reply)
+            | Self::Execute(_, reply)
+            | Self::Repair(_, reply)
+            | Self::Close(reply) => reply,
         }
     }
 }
@@ -100,7 +106,9 @@ impl RoomActor {
             threads: Threads::new(self.requests.max_capacity()),
             work: TaskSet::new(),
         };
-        let close = self.serve_open(&mut timelines).await;
+        let mut repair = RepairController::new(self.room.clone(), self.context.clone());
+        let close = self.serve_open(&mut timelines, &mut repair).await;
+        repair.shutdown().await;
         if close.is_none() {
             self.requests.close();
             let _ = self.notes.send(ActorNote::Leaving {
@@ -130,7 +138,11 @@ impl RoomActor {
     }
 
     /// returns the close request's responder, if a close is why we stopped
-    async fn serve_open(&mut self, timelines: &mut Timelines) -> Option<Responder> {
+    async fn serve_open(
+        &mut self,
+        timelines: &mut Timelines,
+        repair: &mut RepairController,
+    ) -> Option<Responder> {
         let mut room_info = self.room.subscribe_info();
         // a leave that landed while the timeline was building won't come round again
         if matches!(self.room.state(), RoomState::Left | RoomState::Banned) {
@@ -151,7 +163,15 @@ impl RoomActor {
                     Some(ActorRequest::Execute(command, reply)) => {
                         timelines.execute(command, reply, &self.room).await;
                     }
+                    Some(ActorRequest::Repair(command, reply)) => {
+                        repair.execute(command, reply).await;
+                    }
                 },
+                install = repair.next_install(), if repair.is_running() => {
+                    if let Some(install) = install {
+                        timelines.install_history(install, &self.room, &self.context).await;
+                    }
+                }
                 Some(built) = timelines.threads.builds.join_next(), if !timelines.threads.builds.is_empty() => {
                     timelines.thread_built(built, &self.room_id, &self.context).await;
                 }
@@ -220,6 +240,88 @@ struct Timelines {
 }
 
 impl Timelines {
+    async fn install_history(
+        &mut self,
+        install: InstallRequest,
+        room: &Room,
+        context: &SessionContext,
+    ) {
+        let InstallRequest {
+            events,
+            prev_batch,
+            reply,
+        } = install;
+        let result = self
+            .install_history_inner(events, prev_batch, room, context)
+            .await;
+        let _ = reply.send(result);
+    }
+
+    async fn install_history_inner(
+        &mut self,
+        events: Vec<matrix_sdk::deserialized_responses::TimelineEvent>,
+        prev_batch: Option<String>,
+        room: &Room,
+        context: &SessionContext,
+    ) -> Result<(), Error> {
+        let threads_capacity = self.threads.capacity;
+        self.threads.close().await;
+        self.threads = Threads::new(threads_capacity);
+        self.room.pause_for_reseed().await;
+
+        let reseed = async {
+            let (cache, _drop_handles) = room.event_cache().await.map_err(|source| {
+                Error::with_source(ErrorKind::Timeline, "history event cache failed", source)
+            })?;
+            cache.reseed(events, prev_batch).await.map_err(|source| {
+                Error::with_source(ErrorKind::Timeline, "history cache reseed failed", source)
+            })
+        }
+        .await;
+        if let Err(error) = reseed {
+            return match self.room.resume_after_failed_reseed().await {
+                Ok(()) => Err(error),
+                Err(restart_error) => Err(restart_error),
+            };
+        }
+
+        let timeline = match build_timeline(
+            room,
+            TimelineFocus::Live {
+                hide_threaded_events: true,
+            },
+        )
+        .await
+        {
+            Ok(timeline) => timeline,
+            Err(source) => {
+                let error =
+                    Error::with_source(ErrorKind::Timeline, "room timeline rebuild failed", source);
+                return match self.room.resume_after_failed_reseed().await {
+                    Ok(()) => Err(error),
+                    Err(restart_error) => Err(restart_error),
+                };
+            }
+        };
+        let timeline_id = self.room.id.clone();
+        let mut context = context.clone();
+        context.initial_history_limit = 0;
+        let timeline = match TimelineSession::start(timeline_id, timeline, context).await {
+            Ok(timeline) => timeline,
+            Err(source) => {
+                let error =
+                    Error::with_source(ErrorKind::Timeline, "room timeline reset failed", source);
+                return match self.room.resume_after_failed_reseed().await {
+                    Ok(()) => Err(error),
+                    Err(restart_error) => Err(restart_error),
+                };
+            }
+        };
+        let previous = std::mem::replace(&mut self.room, timeline);
+        drop(previous);
+        Ok(())
+    }
+
     async fn execute(&mut self, command: TimelineCommand, reply: Responder, room: &Room) {
         match command {
             TimelineCommand::OpenThread {

Modified file: src/engine/timeline.rs+50 additions−14 deletions@@ -12,9 +12,10 @@ use matrix_sdk::{
         },
     },
 };
+use matrix_sdk_ui::sync_service::SyncService;
 use matrix_sdk_ui::timeline::{Timeline, TimelineEventItemId};
 use ruma::{OwnedEventId, OwnedUserId};
-use tokio::sync::mpsc;
+use tokio::sync::{Mutex, mpsc};
 
 use super::executor::{EngineBoxFuture, JoinHandle, TaskSet, spawn};
 use crate::{
@@ -66,6 +67,10 @@ impl Drop for TaskGuard {
 pub(super) struct SessionContext {
     pub(super) events: EventSender,
     pub(super) media: Arc<MediaService>,
+    pub(super) mentions: crate::mentions::service::Handle,
+    pub(super) sync: Arc<SyncService>,
+    pub(super) repair_gate: Arc<Mutex<()>>,
+    pub(super) repair_shutdown: Arc<std::sync::atomic::AtomicBool>,
     pub(super) current_user_id: Option<OwnedUserId>,
     pub(super) initial_history_limit: u16,
     pub(super) supervisor: Supervisor,
@@ -96,16 +101,7 @@ impl TimelineSession {
         let timeline = Arc::new(timeline);
         let scope = MediaScope::new();
         let watcher = start_watcher(&id, &timeline, &context, &scope).await?;
-        let (pagination, requests) = mpsc::channel(PAGINATION_QUEUE);
-        let pagination_worker = context.supervisor.spawn(
-            "pagination worker",
-            run_pagination(
-                requests,
-                timeline.clone(),
-                context.events.clone(),
-                id.clone(),
-            ),
-        );
+        let (pagination, pagination_worker) = start_pagination_worker(&id, &timeline, &context);
         // under lazy-loaded members the sdk leaves sender profiles unavailable until someone
         // asks, and a /members fetched any other way never reaches the timeline's items
         let member_fetch = context.supervisor.spawn("member fetch", {
@@ -157,15 +153,37 @@ impl TimelineSession {
         }
     }
 
-    /// pending paginations answer `Interrupted` as their responders drop
-    pub(super) async fn close(mut self) {
+    pub(super) async fn pause_for_reseed(&mut self) {
+        self.stop_without_close().await;
+    }
+
+    async fn stop_without_close(&mut self) {
         self.member_fetch.abort();
         self.pagination_worker.abort();
-        let _ = (&mut self.pagination_worker).await;
         self.watcher.abort();
+        let _ = (&mut self.member_fetch).await;
+        let _ = (&mut self.pagination_worker).await;
         let _ = (&mut self.watcher).await;
         // after the watcher, or a last diff could register media for a timeline that's gone
         self.context.media.release(&self.scope).await;
+    }
+
+    pub(super) async fn resume_after_failed_reseed(&mut self) -> Result<(), Error> {
+        self.watcher = start_watcher(&self.id, &self.timeline, &self.context, &self.scope)
+            .await
+            .map_err(|_| Error::new(ErrorKind::Timeline, "timeline reset could not be sent"))?;
+        (self.pagination, self.pagination_worker) =
+            start_pagination_worker(&self.id, &self.timeline, &self.context);
+        self.member_fetch = self.context.supervisor.spawn("member fetch", {
+            let timeline = self.timeline.clone();
+            async move { timeline.fetch_members().await }
+        });
+        Ok(())
+    }
+
+    /// pending paginations answer `Interrupted` as their responders drop
+    pub(super) async fn close(mut self) {
+        self.stop_without_close().await;
         let _ = self
             .context
             .events
@@ -339,6 +357,24 @@ impl TimelineSession {
     }
 }
 
+fn start_pagination_worker(
+    id: &TimelineId,
+    timeline: &Arc<Timeline>,
+    context: &SessionContext,
+) -> (mpsc::Sender<PaginationRequest>, JoinHandle<()>) {
+    let (pagination, requests) = mpsc::channel(PAGINATION_QUEUE);
+    let worker = context.supervisor.spawn(
+        "pagination worker",
+        run_pagination(
+            requests,
+            timeline.clone(),
+            context.events.clone(),
+            id.clone(),
+        ),
+    );
+    (pagination, worker)
+}
+
 impl Drop for TimelineSession {
     fn drop(&mut self) {
         self.watcher.abort();

Modified file: src/lib.rs+7 additions@@ -9,6 +9,7 @@ mod mentions_model;
 mod mirror;
 mod model;
 mod normalise;
+mod repair_model;
 mod space_order;
 mod transport;
 mod uiaa;
@@ -71,7 +72,13 @@ pub use model::{
     UploadDetails, UploadOutcome, UploadThumbnail, UtdCause, VerificationAction,
     VerificationFailure, VerificationState, VerificationTarget,
 };
+pub use repair_model::{
+    HistoryContinuity, HistoryRepairBudget, HistoryRepairCommand, HistoryRepairPhase,
+    HistoryRepairPolicy, HistoryRepairReport, HistoryRepairStop, HistoryRepairStrategy,
+};
 pub use ruma;
+#[doc(hidden)]
+pub use transport::PublicationAck;
 pub use transport::{ClientHandle, EventStream, Pending, SpaceHierarchy};
 pub use uiaa::{
     InteractiveAuth, InteractiveAuthChallenge, InteractiveAuthFailure, InteractiveAuthFlow,

Modified file: src/mentions/service.rs+105 additions−6 deletions@@ -10,18 +10,18 @@ use matrix_sdk::{
     sliding_sync::{ResponseCommitBatch, ResponseCommitHook},
 };
 use ruma::{
-    OwnedRoomId, OwnedUserId,
+    OwnedEventId, OwnedRoomId, OwnedUserId,
     events::{AnySyncTimelineEvent, room::encrypted::OriginalSyncRoomEncryptedEvent},
     serde::Raw,
 };
 use serde_json::Value;
 use tokio::sync::{mpsc, oneshot, watch};
 
-use super::state::{Candidate, Data, SavedEvent, Window, id, invalid, now};
+use super::state::{Candidate, Data, RepairCoverage, SavedEvent, Window, id, invalid, now};
 use crate::{
-    CommandOutput, EngineConfig, Error, ErrorKind, Event, MentionCaptureState, MentionEntry,
-    MentionGap, MentionKey, MentionReason, MentionsCommand, MentionsPage, MentionsUpdate,
-    TimelineContent, TimelineItem,
+    CommandOutput, EngineConfig, Error, ErrorKind, Event, HistoryRepairPolicy, MentionCaptureState,
+    MentionEntry, MentionGap, MentionKey, MentionReason, MentionsCommand, MentionsPage,
+    MentionsUpdate, TimelineContent, TimelineItem,
     engine::{MediaService, MentionsStore, context, executor::EngineBoxFuture},
     normalise::{MediaRegistration, normalise_observed},
     transport::EventSender,
@@ -40,6 +40,10 @@ enum Request {
         MentionsCommand,
         oneshot::Sender<Result<CommandOutput, Error>>,
     ),
+    GetRepairPolicy(oneshot::Sender<Result<HistoryRepairPolicy, Error>>),
+    SetRepairPolicy(HistoryRepairPolicy, oneshot::Sender<Result<(), Error>>),
+    GetRepairCoverage(OwnedRoomId, oneshot::Sender<Result<RepairCoverage, Error>>),
+    RepairResolved(OwnedRoomId, oneshot::Sender<Result<(), Error>>),
     Stop(oneshot::Sender<()>),
 }
 
@@ -100,6 +104,45 @@ impl Handle {
         response.await.map_err(|_| Error::interrupted())?
     }
 
+    pub(crate) async fn repair_policy(&self) -> Result<HistoryRepairPolicy, Error> {
+        let (reply, response) = oneshot::channel();
+        self.0
+            .send(Request::GetRepairPolicy(reply))
+            .await
+            .map_err(|_| Error::interrupted())?;
+        response.await.map_err(|_| Error::interrupted())?
+    }
+
+    pub(crate) async fn set_repair_policy(&self, policy: HistoryRepairPolicy) -> Result<(), Error> {
+        let (reply, response) = oneshot::channel();
+        self.0
+            .send(Request::SetRepairPolicy(policy, reply))
+            .await
+            .map_err(|_| Error::interrupted())?;
+        response.await.map_err(|_| Error::interrupted())?
+    }
+
+    pub(crate) async fn repair_coverage(
+        &self,
+        room_id: OwnedRoomId,
+    ) -> Result<RepairCoverage, Error> {
+        let (reply, response) = oneshot::channel();
+        self.0
+            .send(Request::GetRepairCoverage(room_id, reply))
+            .await
+            .map_err(|_| Error::interrupted())?;
+        response.await.map_err(|_| Error::interrupted())?
+    }
+
+    pub(crate) async fn repair_resolved(&self, room_id: OwnedRoomId) -> Result<(), Error> {
+        let (reply, response) = oneshot::channel();
+        self.0
+            .send(Request::RepairResolved(room_id, reply))
+            .await
+            .map_err(|_| Error::interrupted())?;
+        response.await.map_err(|_| Error::interrupted())?
+    }
+
     pub(crate) async fn stop(&self) {
         let (reply, response) = oneshot::channel();
         if self.0.send(Request::Stop(reply)).await.is_ok() {
@@ -225,6 +268,23 @@ impl Service {
         Ok(())
     }
 
+    async fn set_repair_policy(&mut self, policy: HistoryRepairPolicy) -> Result<(), Error> {
+        let mut next = self.data.clone();
+        next.repair_policy = policy;
+        self.save(next).await
+    }
+
+    async fn repair_resolved(&mut self, room_id: OwnedRoomId) -> Result<(), Error> {
+        let mut next = self.data.clone();
+        let coverage = next.repair_coverage.entry(room_id.clone()).or_default();
+        coverage.boundary_ms = None;
+        coverage.last_confirmed_ms = Some(now());
+        coverage.reliable = false;
+        coverage.unresolved = false;
+        next.observed_rooms.insert(room_id);
+        self.save(next).await
+    }
+
     async fn command(&mut self, command: MentionsCommand) -> Result<CommandOutput, Error> {
         match command {
             MentionsCommand::Page { cursor, limit } => {
@@ -340,13 +400,20 @@ impl Service {
         if next.checkpoints.get(&batch.connection_id) == Some(&batch.next_pos) {
             return Ok(());
         }
-        if next.checkpoints.get(&batch.connection_id) != batch.previous_pos.as_ref() {
+        let observed_at_ms = now();
+        let checkpoint_discontinuity =
+            next.checkpoints.get(&batch.connection_id) != batch.previous_pos.as_ref();
+        if checkpoint_discontinuity {
             Self::gap(
                 &mut next,
                 None,
                 "sync checkpoint continuity is unknown",
                 None,
             );
+            let observed_rooms: Vec<_> = next.observed_rooms.iter().cloned().collect();
+            for room_id in observed_rooms {
+                Self::mark_unresolved(&mut next, &room_id, observed_at_ms);
+            }
         }
         for (room_id, raw) in &batch.raw_rooms {
             if raw.limited || !next.observed_rooms.contains(room_id) {
@@ -376,8 +443,20 @@ impl Service {
             {
                 continue;
             }
+            let previously_observed = next.observed_rooms.contains(room_id);
+            let limited = batch.raw_rooms.get(room_id).is_some_and(|raw| raw.limited);
+            if checkpoint_discontinuity || limited || !previously_observed {
+                Self::mark_unresolved(&mut next, room_id, observed_at_ms);
+            } else {
+                let coverage = next.repair_coverage.entry(room_id.clone()).or_default();
+                if !coverage.unresolved {
+                    coverage.last_confirmed_ms = Some(observed_at_ms);
+                }
+                coverage.reliable = false;
+            }
             next.observed_rooms.insert(room_id.clone());
             let mut observed = BTreeMap::new();
+            let mut latest_anchor = None;
             if let Some(raw) = batch.raw_rooms.get(room_id) {
                 for event in &raw.timeline {
                     let value: Value =
@@ -385,10 +464,17 @@ impl Service {
                             Error::with_source(ErrorKind::Invalid, "invalid observed event", error)
                         })?;
                     if let Some(event_id) = value["event_id"].as_str() {
+                        latest_anchor = event_id.parse::<OwnedEventId>().ok();
                         observed.insert(event_id.to_owned(), value);
                     }
                 }
             }
+            if let Some(latest_anchor) = latest_anchor {
+                next.repair_coverage
+                    .entry(room_id.clone())
+                    .or_default()
+                    .latest_anchor = Some(latest_anchor);
+            }
             if let Some(update) = batch.response.rooms.joined.get(room_id) {
                 for event in &update.timeline.events {
                     let value: Value =
@@ -443,6 +529,15 @@ impl Service {
         self.save(next).await
     }
 
+    fn mark_unresolved(data: &mut Data, room_id: &OwnedRoomId, observed_at_ms: u64) {
+        let coverage = data.repair_coverage.entry(room_id.clone()).or_default();
+        if !coverage.unresolved {
+            coverage.boundary_ms = Some(observed_at_ms);
+            coverage.unresolved = true;
+        }
+        coverage.reliable = false;
+    }
+
     async fn filter_ignored(&self, next: &mut Data) {
         let mut removed = Vec::new();
         for (key, entry) in &next.entries {
@@ -853,6 +948,10 @@ impl Service {
                     Some(Request::Stop(reply)) => { let _ = reply.send(()); return; }
                     Some(Request::Batch(batch, reply)) => { let result = self.batch(batch).await; let _ = reply.send(result); }
                     Some(Request::Command(command, reply)) => { let result = self.command(command).await; let _ = reply.send(result); }
+                    Some(Request::GetRepairPolicy(reply)) => { let _ = reply.send(Ok(self.data.repair_policy.clone())); }
+                    Some(Request::SetRepairPolicy(policy, reply)) => { let result = self.set_repair_policy(policy).await; let _ = reply.send(result); }
+                    Some(Request::GetRepairCoverage(room_id, reply)) => { let coverage = self.data.repair_coverage.get(&room_id).cloned().unwrap_or_default(); let _ = reply.send(Ok(coverage)); }
+                    Some(Request::RepairResolved(room_id, reply)) => { let result = self.repair_resolved(room_id).await; let _ = reply.send(result); }
                     None => return,
                 },
                 Some(result) = work.next(), if !work.is_empty() => {

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

Modified file: src/mentions/state.rs+26 additions−3 deletions@@ -1,11 +1,11 @@
 use std::collections::{BTreeMap, BTreeSet};
 
-use ruma::OwnedUserId;
+use ruma::{OwnedEventId, OwnedUserId};
 use serde_json::Value;
 
 use crate::{
-    Error, ErrorKind, EventContext, MentionCaptureState, MentionCoverage, MentionEntry, MentionKey,
-    MentionsBoundary, MentionsPage, SavedMentionContext, TimelineItem,
+    Error, ErrorKind, EventContext, HistoryRepairPolicy, MentionCaptureState, MentionCoverage,
+    MentionEntry, MentionKey, MentionsBoundary, MentionsPage, SavedMentionContext, TimelineItem,
     normalise::MediaRegistration,
 };
 
@@ -36,6 +36,25 @@ pub(super) struct Window {
     pub end: Option<String>,
 }
 
+#[derive(Clone, Debug, Default)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+pub(crate) struct RepairCoverage {
+    pub(crate) last_confirmed_ms: Option<u64>,
+    pub(crate) boundary_ms: Option<u64>,
+    pub(crate) reliable: bool,
+    pub(crate) unresolved: bool,
+    pub(crate) latest_anchor: Option<OwnedEventId>,
+}
+
+impl RepairCoverage {
+    pub(crate) fn duration_ms(&self) -> Option<u64> {
+        if !self.reliable {
+            return None;
+        }
+        self.boundary_ms?.checked_sub(self.last_confirmed_ms?)
+    }
+}
+
 #[derive(Clone, Default)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub(super) struct Data {
@@ -50,6 +69,10 @@ pub(super) struct Data {
     pub coverage: MentionCoverage,
     pub checkpoints: BTreeMap<String, String>,
     pub observed_rooms: BTreeSet<ruma::OwnedRoomId>,
+    #[cfg_attr(feature = "serde", serde(default))]
+    pub repair_policy: HistoryRepairPolicy,
+    #[cfg_attr(feature = "serde", serde(default))]
+    pub repair_coverage: BTreeMap<ruma::OwnedRoomId, RepairCoverage>,
     pub history_cursor: Option<String>,
     pub history_exhausted: bool,
     pub latest_cursor: Option<String>,

Modified file: src/mirror.rs+25 additions−3 deletions@@ -20,9 +20,9 @@ use tokio::sync::broadcast;
 
 use crate::{
     ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
-    HistoryState, ImagePacks, RoomSummary, TimelineCommand, TimelineDiff, TimelineId, TimelineItem,
-    TypingState, UploadOutcome, VerificationState, VerificationTarget,
-    engine::executor::EngineBoxFuture,
+    HistoryRepairReport, HistoryState, ImagePacks, RoomSummary, TimelineCommand, TimelineDiff,
+    TimelineId, TimelineItem, TypingState, UploadOutcome, VerificationState, VerificationTarget,
+    engine::executor::EngineBoxFuture, transport::PublicationRegistration,
 };
 
 #[cfg(not(target_family = "wasm"))]
@@ -49,6 +49,7 @@ pub struct State {
     pub rooms: Vector<RoomSummary>,
     pub image_packs: ImagePacks,
     pub timelines: imbl::HashMap<TimelineId, TimelineState>,
+    pub repairs: imbl::HashMap<OwnedRoomId, HistoryRepairReport>,
     pub typing: imbl::HashMap<OwnedRoomId, Vec<OwnedUserId>>,
     pub uploads: imbl::HashMap<OwnedTransactionId, UploadState>,
     /// what stopped the engine, if something went wrong
@@ -135,6 +136,7 @@ pub enum StateChange {
         timeline_id: TimelineId,
         state: HistoryState,
     },
+    HistoryRepair(HistoryRepairReport),
     Typing(TypingState),
     Upload {
         transaction_id: OwnedTransactionId,
@@ -168,6 +170,7 @@ pub struct Mirror {
     handle: ClientHandle,
     events: EngineBoxStream<Event>,
     shared: Arc<Shared>,
+    _publication: PublicationRegistration,
     resyncs: HashMap<TimelineId, AbortHandle>,
     rooms_fetch: Option<AbortHandle>,
     packs_fetch: Option<AbortHandle>,
@@ -206,6 +209,7 @@ impl Mirror {
             rooms: Vector::new(),
             image_packs: ImagePacks::default(),
             timelines: imbl::HashMap::new(),
+            repairs: imbl::HashMap::new(),
             typing: imbl::HashMap::new(),
             uploads: imbl::HashMap::new(),
             engine_failure: None,
@@ -220,6 +224,7 @@ impl Mirror {
                 packs_fetch: Fetch::Idle,
             }),
         });
+        let publication = handle.register_publication_mirror();
         #[cfg(not(target_family = "wasm"))]
         let events = events.boxed();
         #[cfg(target_family = "wasm")]
@@ -228,6 +233,7 @@ impl Mirror {
             handle,
             events,
             shared: shared.clone(),
+            _publication: publication,
             resyncs: HashMap::new(),
             rooms_fetch: None,
             packs_fetch: None,
@@ -357,6 +363,22 @@ impl Mirror {
                     state: history,
                 })
             }),
+            Event::HistoryRepairChanged(report) => self.publish(|state| {
+                state.repairs.insert(report.room_id.clone(), report.clone());
+                Some(StateChange::HistoryRepair(report))
+            }),
+            Event::HistoryRepairPublished {
+                report,
+                acknowledgement,
+            } => {
+                self.publish(|state| {
+                    state.repairs.insert(report.room_id.clone(), report.clone());
+                    Some(StateChange::HistoryRepair(report))
+                });
+                if let Some(acknowledgement) = acknowledgement {
+                    acknowledgement.complete(true);
+                }
+            }
             Event::TimelineClosed { timeline_id } => {
                 self.cancel_resync(&timeline_id);
                 self.publish(|state| {

Modified file: src/model.rs+12 additions−2 deletions@@ -30,6 +30,7 @@ pub enum Command {
     Notification(NotificationCommand),
     Lifecycle(LifecycleCommand),
     Pack(PackCommand),
+    HistoryRepair(crate::HistoryRepairCommand),
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -653,7 +654,7 @@ pub enum LifecycleCommand {
     Shutdown,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum CommandOutput {
     Done,
@@ -668,6 +669,8 @@ pub enum CommandOutput {
     EventContextPage(EventContextPage),
     MentionsPage(crate::MentionsPage),
     MentionContext(crate::SavedMentionContext),
+    HistoryRepair(crate::HistoryRepairReport),
+    HistoryRepairPolicy(crate::HistoryRepairPolicy),
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -864,7 +867,7 @@ impl MessageMentions {
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum Event {
     ConnectionChanged(ConnectionState),
@@ -904,6 +907,13 @@ pub enum Event {
         timeline_id: TimelineId,
         state: HistoryState,
     },
+    HistoryRepairChanged(crate::HistoryRepairReport),
+    #[doc(hidden)]
+    HistoryRepairPublished {
+        report: crate::HistoryRepairReport,
+        #[cfg_attr(feature = "serde", serde(skip))]
+        acknowledgement: Option<crate::PublicationAck>,
+    },
     /// the timeline was closed, or its room was left, and nothing more comes for it
     /// until another `TimelineReset`
     TimelineClosed {

Added file: src/repair_model.rs+152 additions@@ -0,0 +1,152 @@
+use ruma::OwnedRoomId;
+
+#[cfg(feature = "serde")]
+use serde::{Deserialize, Serialize};
+
+use crate::Error;
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum HistoryRepairPolicy {
+    StartRecent,
+    Bridge,
+    BridgeShortGaps { threshold_ms: u64 },
+}
+
+impl Default for HistoryRepairPolicy {
+    fn default() -> Self {
+        Self::StartRecent
+    }
+}
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub struct HistoryRepairBudget {
+    pub max_pages: u16,
+    pub max_events: u32,
+    pub max_duration_ms: u64,
+}
+
+impl Default for HistoryRepairBudget {
+    fn default() -> Self {
+        Self {
+            max_pages: 10,
+            max_events: 1000,
+            max_duration_ms: 15_000,
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum HistoryRepairCommand {
+    Start {
+        room_id: OwnedRoomId,
+        policy: Option<HistoryRepairPolicy>,
+        budget: HistoryRepairBudget,
+    },
+    Continue {
+        room_id: OwnedRoomId,
+        budget: HistoryRepairBudget,
+    },
+    Cancel {
+        room_id: OwnedRoomId,
+    },
+    StartRecent {
+        room_id: OwnedRoomId,
+    },
+    GetPolicy,
+    SetPolicy(HistoryRepairPolicy),
+}
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum HistoryRepairStrategy {
+    Recent,
+    Overlap,
+    Bridge,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum HistoryContinuity {
+    Unresolved,
+    Verified,
+    AccessibleStart,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum HistoryRepairPhase {
+    FetchingRecent,
+    Bridging,
+    Ready,
+    Paused,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum HistoryRepairStop {
+    Cancelled,
+    Budget,
+    HistoryUnavailable,
+    NoAnchor,
+    CursorStalled,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub struct HistoryRepairReport {
+    pub room_id: OwnedRoomId,
+    pub requested_policy: HistoryRepairPolicy,
+    pub strategy: HistoryRepairStrategy,
+    pub coverage_duration_ms: Option<u64>,
+    pub threshold_evaluated: bool,
+    pub pages: u32,
+    pub events: u32,
+    pub phase: HistoryRepairPhase,
+    pub continuity: HistoryContinuity,
+    pub recent_available: bool,
+    pub stop: Option<HistoryRepairStop>,
+    pub failure: Option<Error>,
+}
+
+#[cfg(all(test, feature = "serde"))]
+mod tests {
+    use super::*;
+    use crate::ErrorKind;
+    use ruma::room_id;
+
+    #[test]
+    fn history_repair_types_round_trip_through_serde() {
+        let command = HistoryRepairCommand::Start {
+            room_id: room_id!("!room:example.org").to_owned(),
+            policy: Some(HistoryRepairPolicy::BridgeShortGaps {
+                threshold_ms: 86_400_000,
+            }),
+            budget: HistoryRepairBudget::default(),
+        };
+        let encoded_command = serde_json::to_value(&command).unwrap();
+        let decoded_command: HistoryRepairCommand =
+            serde_json::from_value(encoded_command).unwrap();
+        assert_eq!(decoded_command, command);
+
+        let report = HistoryRepairReport {
+            room_id: room_id!("!room:example.org").to_owned(),
+            requested_policy: HistoryRepairPolicy::Bridge,
+            strategy: HistoryRepairStrategy::Overlap,
+            coverage_duration_ms: Some(86_400_000),
+            threshold_evaluated: false,
+            pages: 3,
+            events: 42,
+            phase: HistoryRepairPhase::Paused,
+            continuity: HistoryContinuity::Unresolved,
+            recent_available: true,
+            stop: Some(HistoryRepairStop::Budget),
+            failure: Some(Error::new(ErrorKind::Timeline, "repair failed")),
+        };
+        let encoded_report = serde_json::to_value(&report).unwrap();
+        let decoded_report: HistoryRepairReport = serde_json::from_value(encoded_report).unwrap();
+        assert_eq!(decoded_report, report);
+    }
+}

Modified file: src/transport.rs+302 additions−15 deletions@@ -3,7 +3,7 @@ use std::{
     fmt,
     future::Future,
     pin::Pin,
-    sync::{Arc, Mutex},
+    sync::{Arc, Mutex, Weak},
     task::{Context, Poll},
 };
 
@@ -29,15 +29,15 @@ use tokio::sync::{OnceCell, mpsc, oneshot, watch};
 use crate::{
     AccountCommand, AttachmentUpload, Command, CommandOutput, ConnectionState, CreateRoomOptions,
     EncryptionCommand, EncryptionState, Error, ErrorKind, Event, EventContext, EventContextCommand,
-    EventContextPage, EventRef, ImagePackSource, ImagePacks, InteractiveAuth,
-    InteractiveAuthOutcome, LifecycleCommand, LinkPreview, Media, MediaCommand, MediaId,
-    MediaLocation, MediaSize, MemberFilter, MemberSummary, MentionKey, MentionsBoundary,
-    MentionsCommand, MentionsPage, MessageContent, MessageEditContent, NotificationCommand,
-    OwnDevice, PackCommand, PackImageInfo, PackUsage, Patch, PowerLevelChanges, Profile,
-    PushNotification, PushNotificationEvent, PushNotificationStatus, RoomCommand, RoomListFilter,
-    RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomSummary,
-    SavedMentionContext, SpaceChild, TimelineCommand, TimelineId, VerificationAction,
-    VerificationTarget,
+    EventContextPage, EventRef, HistoryRepairBudget, HistoryRepairCommand, HistoryRepairPolicy,
+    HistoryRepairReport, ImagePackSource, ImagePacks, InteractiveAuth, InteractiveAuthOutcome,
+    LifecycleCommand, LinkPreview, Media, MediaCommand, MediaId, MediaLocation, MediaSize,
+    MemberFilter, MemberSummary, MentionKey, MentionsBoundary, MentionsCommand, MentionsPage,
+    MessageContent, MessageEditContent, NotificationCommand, OwnDevice, PackCommand, PackImageInfo,
+    PackUsage, Patch, PowerLevelChanges, Profile, PushNotification, PushNotificationEvent,
+    PushNotificationStatus, RoomCommand, RoomListFilter, RoomNotificationMode,
+    RoomNotificationSettings, RoomPowerLevels, RoomSummary, SavedMentionContext, SpaceChild,
+    TimelineCommand, TimelineId, VerificationAction, VerificationTarget,
     encryption::encryption_state,
     engine::{MediaService, StoreLock},
     model::{PushNotificationRoom, PushNotificationSender},
@@ -54,6 +54,7 @@ pub struct ClientHandle {
     media: Arc<MediaService>,
     spaces: Arc<OnceCell<SpaceService>>,
     members_checked: Arc<Mutex<HashSet<OwnedRoomId>>>,
+    publication: Arc<PublicationRegistry>,
     _store_lock: Option<Arc<StoreLock>>,
 }
 
@@ -81,6 +82,10 @@ impl Future for Pending {
 }
 
 impl ClientHandle {
+    pub(crate) fn register_publication_mirror(&self) -> PublicationRegistration {
+        self.publication.register_mirror()
+    }
+
     /// current credentials, it holds secrets. the engine saves it for you unless the
     /// session was built `without_persistence`
     #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
@@ -327,6 +332,84 @@ impl ClientHandle {
         .await
     }
 
+    pub async fn history_repair(
+        &self,
+        room_id: OwnedRoomId,
+        policy: Option<HistoryRepairPolicy>,
+        budget: HistoryRepairBudget,
+    ) -> Result<HistoryRepairReport, Error> {
+        match self
+            .execute(Command::HistoryRepair(HistoryRepairCommand::Start {
+                room_id,
+                policy,
+                budget,
+            }))
+            .await?
+        {
+            CommandOutput::HistoryRepair(report) => Ok(report),
+            output => Err(unexpected("history repair", &output)),
+        }
+    }
+
+    pub async fn continue_history_repair(
+        &self,
+        room_id: OwnedRoomId,
+        budget: HistoryRepairBudget,
+    ) -> Result<HistoryRepairReport, Error> {
+        match self
+            .execute(Command::HistoryRepair(HistoryRepairCommand::Continue {
+                room_id,
+                budget,
+            }))
+            .await?
+        {
+            CommandOutput::HistoryRepair(report) => Ok(report),
+            output => Err(unexpected("history repair", &output)),
+        }
+    }
+
+    pub async fn cancel_history_repair(&self, room_id: OwnedRoomId) -> Result<(), Error> {
+        self.run(Command::HistoryRepair(HistoryRepairCommand::Cancel {
+            room_id,
+        }))
+        .await
+    }
+
+    pub async fn start_recent_history(
+        &self,
+        room_id: OwnedRoomId,
+    ) -> Result<HistoryRepairReport, Error> {
+        match self
+            .execute(Command::HistoryRepair(HistoryRepairCommand::StartRecent {
+                room_id,
+            }))
+            .await?
+        {
+            CommandOutput::HistoryRepair(report) => Ok(report),
+            output => Err(unexpected("history repair", &output)),
+        }
+    }
+
+    pub async fn history_repair_policy(&self) -> Result<HistoryRepairPolicy, Error> {
+        match self
+            .execute(Command::HistoryRepair(HistoryRepairCommand::GetPolicy))
+            .await?
+        {
+            CommandOutput::HistoryRepairPolicy(policy) => Ok(policy),
+            output => Err(unexpected("history repair policy", &output)),
+        }
+    }
+
+    pub async fn set_history_repair_policy(
+        &self,
+        policy: HistoryRepairPolicy,
+    ) -> Result<(), Error> {
+        self.run(Command::HistoryRepair(HistoryRepairCommand::SetPolicy(
+            policy,
+        )))
+        .await
+    }
+
     pub async fn event_context(
         &self,
         room_id: &RoomId,
@@ -1430,29 +1513,153 @@ pub(crate) struct DriverPort {
     requests: mpsc::Receiver<Request>,
     events: mpsc::Sender<Event>,
     connection: watch::Sender<ConnectionState>,
+    publication: Arc<PublicationRegistry>,
+}
+
+#[doc(hidden)]
+#[derive(Clone, Debug)]
+pub struct PublicationAck(Arc<PublicationAckInner>);
+
+#[derive(Debug)]
+struct PublicationAckInner {
+    sender: Mutex<Option<oneshot::Sender<bool>>>,
+}
+
+impl PartialEq for PublicationAck {
+    fn eq(&self, other: &Self) -> bool {
+        Arc::ptr_eq(&self.0, &other.0)
+    }
+}
+
+impl Eq for PublicationAck {}
+
+impl PublicationAck {
+    pub(crate) fn complete(&self, published: bool) {
+        let sender = self
+            .0
+            .sender
+            .lock()
+            .unwrap_or_else(|error| error.into_inner())
+            .take();
+        if let Some(sender) = sender {
+            let _ = sender.send(published);
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct PublicationRegistry {
+    state: Mutex<PublicationRegistryState>,
+}
+
+#[derive(Debug, Default)]
+struct PublicationRegistryState {
+    mirrors: usize,
+    pending: Vec<Weak<PublicationAckInner>>,
+}
+
+impl PublicationRegistry {
+    fn register_mirror(self: &Arc<Self>) -> PublicationRegistration {
+        self.state
+            .lock()
+            .unwrap_or_else(|error| error.into_inner())
+            .mirrors += 1;
+        PublicationRegistration {
+            registry: self.clone(),
+        }
+    }
+
+    fn acknowledgement(&self) -> Option<(PublicationAck, oneshot::Receiver<bool>)> {
+        let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+        if state.mirrors == 0 {
+            return None;
+        }
+        state.pending.retain(|pending| pending.strong_count() > 0);
+        let (sender, receiver) = oneshot::channel();
+        let acknowledgement = PublicationAck(Arc::new(PublicationAckInner {
+            sender: Mutex::new(Some(sender)),
+        }));
+        state.pending.push(Arc::downgrade(&acknowledgement.0));
+        Some((acknowledgement, receiver))
+    }
+}
+
+pub(crate) struct PublicationRegistration {
+    registry: Arc<PublicationRegistry>,
+}
+
+impl Drop for PublicationRegistration {
+    fn drop(&mut self) {
+        let pending = {
+            let mut state = self
+                .registry
+                .state
+                .lock()
+                .unwrap_or_else(|error| error.into_inner());
+            state.mirrors -= 1;
+            if state.mirrors == 0 {
+                std::mem::take(&mut state.pending)
+            } else {
+                Vec::new()
+            }
+        };
+        for acknowledgement in pending.into_iter().filter_map(|ack| ack.upgrade()) {
+            PublicationAck(acknowledgement).complete(false);
+        }
+    }
 }
 
 #[derive(Clone, Debug)]
-pub(crate) struct EventSender(mpsc::Sender<Event>);
+pub(crate) struct EventSender {
+    events: mpsc::Sender<Event>,
+    publication: Arc<PublicationRegistry>,
+}
 
 impl EventSender {
     pub(crate) async fn send(&self, event: Event) -> Result<(), SendEventError> {
-        self.0.send(event).await.map_err(|_| SendEventError)
+        self.events.send(event).await.map_err(|_| SendEventError)
+    }
+
+    pub(crate) async fn publish_repair(
+        &self,
+        report: crate::HistoryRepairReport,
+    ) -> Result<(), SendEventError> {
+        let Some((acknowledgement, published)) = self.publication.acknowledgement() else {
+            return self.send(Event::HistoryRepairChanged(report)).await;
+        };
+        if let Err(error) = self
+            .send(Event::HistoryRepairPublished {
+                report,
+                acknowledgement: Some(acknowledgement.clone()),
+            })
+            .await
+        {
+            acknowledgement.complete(false);
+            return Err(error);
+        }
+        drop(acknowledgement);
+        match published.await {
+            Ok(true) => Ok(()),
+            Ok(false) | Err(_) => Err(SendEventError),
+        }
     }
 
     pub(crate) async fn reserve(&self) -> Result<mpsc::Permit<'_, Event>, SendEventError> {
-        self.0.reserve().await.map_err(|_| SendEventError)
+        self.events.reserve().await.map_err(|_| SendEventError)
     }
 
     /// resolves once every `EventStream` is gone
     pub(crate) async fn closed(&self) {
-        self.0.closed().await;
+        self.events.closed().await;
     }
 }
 
 impl DriverPort {
     pub(crate) fn event_sender(&self) -> EventSender {
-        EventSender(self.events.clone())
+        EventSender {
+            events: self.events.clone(),
+            publication: self.publication.clone(),
+        }
     }
 
     pub(crate) fn connection_sender(&self) -> watch::Sender<ConnectionState> {
@@ -1488,6 +1695,7 @@ pub(crate) fn transport_pair(
     let (request_tx, request_rx) = mpsc::channel(command_capacity);
     let (event_tx, event_rx) = mpsc::channel(event_capacity);
     let (connection_tx, connection_rx) = watch::channel(ConnectionState::Idle);
+    let publication = Arc::new(PublicationRegistry::default());
     (
         ClientHandle {
             requests: request_tx,
@@ -1498,12 +1706,14 @@ pub(crate) fn transport_pair(
             media,
             spaces: Arc::new(OnceCell::new()),
             members_checked: Arc::default(),
+            publication: publication.clone(),
             _store_lock: store_lock,
         },
         DriverPort {
             requests: request_rx,
             events: event_tx,
             connection: connection_tx,
+            publication,
         },
         EventStream { events: event_rx },
     )
@@ -1591,3 +1801,80 @@ fn unexpected(wanted: &str, output: &CommandOutput) -> Error {
         format!("expected {wanted} but got {output:?}"),
     )
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        HistoryContinuity, HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairStrategy,
+    };
+    use ruma::room_id;
+
+    fn report() -> HistoryRepairReport {
+        HistoryRepairReport {
+            room_id: room_id!("!room:example.org").to_owned(),
+            requested_policy: HistoryRepairPolicy::Bridge,
+            strategy: HistoryRepairStrategy::Bridge,
+            coverage_duration_ms: None,
+            threshold_evaluated: false,
+            pages: 0,
+            events: 0,
+            phase: HistoryRepairPhase::Ready,
+            continuity: HistoryContinuity::Verified,
+            recent_available: true,
+            stop: None,
+            failure: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn publish_repair_waits_for_mirror_publication() {
+        let publication = Arc::new(PublicationRegistry::default());
+        let _registration = publication.register_mirror();
+        let (events, mut received) = mpsc::channel(1);
+        let sender = EventSender {
+            events,
+            publication,
+        };
+        let expected = report();
+        let expected_for_task = expected.clone();
+        let publisher = tokio::spawn(async move { sender.publish_repair(expected_for_task).await });
+
+        let Some(Event::HistoryRepairPublished {
+            report,
+            acknowledgement: Some(acknowledgement),
+        }) = received.recv().await
+        else {
+            panic!("repair event was not published");
+        };
+        assert_eq!(report, expected);
+        tokio::task::yield_now().await;
+        assert!(!publisher.is_finished());
+
+        acknowledgement.complete(true);
+        assert!(matches!(publisher.await.unwrap(), Ok(())));
+    }
+
+    #[tokio::test]
+    async fn publish_repair_fails_when_last_mirror_drops() {
+        let publication = Arc::new(PublicationRegistry::default());
+        let registration = publication.register_mirror();
+        let (events, mut received) = mpsc::channel(1);
+        let sender = EventSender {
+            events,
+            publication,
+        };
+        let publisher = tokio::spawn(async move { sender.publish_repair(report()).await });
+
+        assert!(matches!(
+            received.recv().await,
+            Some(Event::HistoryRepairPublished {
+                acknowledgement: Some(_),
+                ..
+            })
+        ));
+        drop(registration);
+
+        assert!(matches!(publisher.await.unwrap(), Err(SendEventError)));
+    }
+}

Added file: tests/history_repair.rs+611 additions@@ -0,0 +1,611 @@
+mod support;
+
+use std::time::Duration;
+
+use macltok::{
+    ClientHandle, Engine, EngineConfig, EngineError, Event, EventStream, HistoryContinuity,
+    HistoryRepairBudget, HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairStop,
+    HistoryRepairStrategy, Termination, TimelineItem,
+};
+#[cfg(feature = "mirror")]
+use macltok::{Mirror, TimelineId};
+use matrix_sdk::{
+    Client,
+    ruma::{EventId, RoomId, api::client::sync::sync_events::v5, event_id, room_id},
+    test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate},
+};
+use matrix_sdk_test::{ALICE, JoinedRoomBuilder, event_factory::EventFactory};
+use support::{event_messages, receive_until, shut_down};
+use wiremock::ResponseTemplate;
+
+struct TestRoom {
+    server: MatrixMockServer,
+    sync_client: Client,
+    handle: ClientHandle,
+    events: EventStream,
+    running: tokio::task::JoinHandle<Result<Termination, EngineError>>,
+}
+
+async fn open_room(room_id: &RoomId) -> TestRoom {
+    let server = MatrixMockServer::new().await;
+    server
+        .mock_sliding_sync()
+        .ok(v5::Response::new("1".to_owned()))
+        .mount()
+        .await;
+    let client = server.client_builder().build().await;
+    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 (engine, handle, mut events) = Engine::from_client(client, config).await.unwrap();
+    let running = tokio::spawn(engine.run());
+
+    handle.open_room(room_id.to_owned()).await.unwrap();
+    receive_until(&mut events, |event| {
+        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
+    })
+    .await;
+
+    TestRoom {
+        server,
+        sync_client,
+        handle,
+        events,
+        running,
+    }
+}
+
+async fn sync_cached_event(room: &mut TestRoom, room_id: &RoomId, id: &EventId, body: &str) {
+    room.server
+        .sync_room(
+            &room.sync_client,
+            JoinedRoomBuilder::new(room_id).add_timeline_event(
+                EventFactory::new()
+                    .text_msg(body)
+                    .event_id(id)
+                    .sender(&ALICE),
+            ),
+        )
+        .await;
+    receive_until(&mut room.events, |event| {
+        event_messages(event)
+            .into_iter()
+            .any(|event| event.event_id.as_deref() == Some(id))
+    })
+    .await;
+}
+
+fn history_event(
+    room_id: &RoomId,
+    id: &EventId,
+    body: &str,
+) -> matrix_sdk::ruma::serde::Raw<matrix_sdk::ruma::events::AnyTimelineEvent> {
+    EventFactory::new()
+        .room(room_id)
+        .text_msg(body)
+        .event_id(id)
+        .sender(&ALICE)
+        .into_raw_timeline()
+}
+
+fn report_budget(max_pages: u16) -> HistoryRepairBudget {
+    HistoryRepairBudget {
+        max_pages,
+        ..HistoryRepairBudget::default()
+    }
+}
+
+#[tokio::test]
+async fn recent_repair_starts_without_a_cursor_and_keeps_its_end_for_pagination() {
+    let room_id = room_id!("!recent-repair:localhost");
+    let stale_id = event_id!("$stale:localhost");
+    let fresh_id = event_id!("$fresh:localhost");
+    let older_id = event_id!("$older:localhost");
+    let mut room = open_room(room_id).await;
+    sync_cached_event(&mut room, room_id, stale_id, "cached before repair").await;
+
+    room.server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("recent-prev")
+            .events(vec![history_event(room_id, fresh_id, "fresh page")]))
+        .mock_once()
+        .mount()
+        .await;
+    let report = room
+        .handle
+        .history_repair(
+            room_id.to_owned(),
+            Some(HistoryRepairPolicy::StartRecent),
+            HistoryRepairBudget::default(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(report.strategy, HistoryRepairStrategy::Recent);
+    assert_eq!(report.phase, HistoryRepairPhase::Ready);
+    assert!(report.recent_available);
+
+    let reset = receive_until(&mut room.events, |event| {
+        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
+            && event_messages(event)
+                .into_iter()
+                .any(|event| event.event_id.as_deref() == Some(fresh_id))
+    })
+    .await;
+    let reset_ids = event_messages(&reset)
+        .into_iter()
+        .filter_map(|event| event.event_id.as_ref().map(ToString::to_string))
+        .collect::<Vec<_>>();
+    assert_eq!(reset_ids, vec![fresh_id.to_string()]);
+
+    let requests = room.server.received_requests().await.unwrap();
+    let messages = requests
+        .iter()
+        .filter(|request| request.url.path().ends_with("/messages"))
+        .collect::<Vec<_>>();
+    assert_eq!(messages.len(), 1);
+    assert!(!messages[0].url.query_pairs().any(|(key, _)| key == "from"));
+
+    room.server
+        .mock_room_messages()
+        .match_from("recent-prev")
+        .ok(
+            RoomMessagesResponseTemplate::default().events(vec![history_event(
+                room_id,
+                older_id,
+                "older page",
+            )]),
+        )
+        .mock_once()
+        .mount()
+        .await;
+    room.handle
+        .paginate(room_id.to_owned().into(), 10)
+        .await
+        .unwrap();
+    receive_until(&mut room.events, |event| {
+        event_messages(event).into_iter().any(|event| {
+            support::message(event).is_some_and(|message| message.body == "older page")
+        })
+    })
+    .await;
+
+    shut_down(&room.handle, &mut room.events, room.running).await;
+}
+
+#[cfg(feature = "mirror")]
+#[tokio::test]
+async fn repair_ack_waits_until_the_mirror_has_applied_the_reset_and_report() {
+    let server = MatrixMockServer::new().await;
+    server
+        .mock_sliding_sync()
+        .ok(v5::Response::new("1".to_owned()))
+        .mount()
+        .await;
+    let client = server.client_builder().build().await;
+    let room_id = room_id!("!repair-mirror:localhost");
+    server.mock_room_state_encryption().plain().mount().await;
+    server.sync_joined_room(&client, room_id).await;
+    let sync_client = client.clone();
+    let (engine, handle, events) = Engine::from_client(
+        client,
+        EngineConfig::default().with_initial_history_limit(0),
+    )
+    .await
+    .unwrap();
+    let running = tokio::spawn(engine.run());
+    let (mirror, mirror_handle) = Mirror::new(handle.clone(), events);
+    let mirror_running = tokio::spawn(mirror.run());
+
+    handle.open_room(room_id.to_owned()).await.unwrap();
+    let stale_id = event_id!("$mirror-stale:localhost");
+    server
+        .sync_room(
+            &sync_client,
+            JoinedRoomBuilder::new(room_id).add_timeline_event(
+                EventFactory::new()
+                    .text_msg("cached event")
+                    .event_id(stale_id)
+                    .sender(&ALICE),
+            ),
+        )
+        .await;
+    let timeline_id = TimelineId::from(room_id.to_owned());
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let has_cached_event = mirror_handle
+                .snapshot()
+                .timelines
+                .get(&timeline_id)
+                .is_some_and(|timeline| {
+                    timeline.items.iter().any(|item| {
+                        matches!(item, TimelineItem::Event(event)
+                            if event.event_id.as_deref() == Some(stale_id))
+                    })
+                });
+            if has_cached_event {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("the cached timeline event did not reach the mirror");
+
+    let fresh_id = event_id!("$mirror-fresh:localhost");
+    server
+        .mock_room_messages()
+        .ok(
+            RoomMessagesResponseTemplate::default().events(vec![history_event(
+                room_id,
+                fresh_id,
+                "fresh event",
+            )]),
+        )
+        .mock_once()
+        .mount()
+        .await;
+    let report = handle
+        .history_repair(
+            room_id.to_owned(),
+            Some(HistoryRepairPolicy::StartRecent),
+            HistoryRepairBudget::default(),
+        )
+        .await
+        .unwrap();
+
+    let state = mirror_handle.snapshot();
+    assert_eq!(state.repairs.get(room_id), Some(&report));
+    let timeline = state
+        .timelines
+        .get(&timeline_id)
+        .expect("the repaired timeline was not applied before acknowledgement");
+    let ids = timeline
+        .items
+        .iter()
+        .filter_map(|item| match item {
+            TimelineItem::Event(event) => event.event_id.as_ref().map(ToString::to_string),
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(ids, vec![fresh_id.to_string()]);
+
+    handle.shutdown().await.unwrap();
+    assert_eq!(running.await.unwrap().unwrap(), Termination::ShutDown);
+    mirror_running.await.unwrap();
+}
+
+#[tokio::test]
+async fn failed_recent_fetch_leaves_the_existing_timeline_in_place() {
+    let room_id = room_id!("!failed-repair:localhost");
+    let stale_id = event_id!("$kept:localhost");
+    let mut room = open_room(room_id).await;
+    sync_cached_event(&mut room, room_id, stale_id, "keep this event").await;
+
+    room.server
+        .mock_room_messages()
+        .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({
+            "errcode": "M_FORBIDDEN",
+            "error": "history is unavailable",
+        })))
+        .mock_once()
+        .mount()
+        .await;
+    let report = room
+        .handle
+        .history_repair(
+            room_id.to_owned(),
+            Some(HistoryRepairPolicy::StartRecent),
+            HistoryRepairBudget::default(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(report.phase, HistoryRepairPhase::Paused);
+    assert!(report.failure.is_some());
+
+    let mut reset_after_failure = false;
+    while let Ok(event) = room.events.try_recv() {
+        reset_after_failure |= matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id);
+    }
+    assert!(!reset_after_failure);
+
+    shut_down(&room.handle, &mut room.events, room.running).await;
+}
+
+#[tokio::test]
+async fn cancelling_a_delayed_bridge_page_preserves_the_recent_stage() {
+    let room_id = room_id!("!cancel-repair:localhost");
+    let anchor_id = event_id!("$cancel-anchor:localhost");
+    let recent_id = event_id!("$cancel-recent:localhost");
+    let mut room = open_room(room_id).await;
+    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;
+
+    room.server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("cancel-prev")
+            .events(vec![history_event(room_id, recent_id, "recent")]))
+        .mock_once()
+        .mount()
+        .await;
+    let initial = room
+        .handle
+        .history_repair(
+            room_id.to_owned(),
+            Some(HistoryRepairPolicy::Bridge),
+            report_budget(1),
+        )
+        .await
+        .unwrap();
+    assert_eq!(initial.stop, Some(HistoryRepairStop::Budget));
+    assert!(initial.recent_available);
+    receive_until(&mut room.events, |event| {
+        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
+            && event_messages(event)
+                .into_iter()
+                .any(|event| event.event_id.as_deref() == Some(recent_id))
+    })
+    .await;
+
+    room.server
+        .mock_room_messages()
+        .match_from("cancel-prev")
+        .ok(RoomMessagesResponseTemplate::default()
+            .with_delay(Duration::from_secs(5))
+            .events(vec![history_event(room_id, anchor_id, "cached anchor")]))
+        .mock_once()
+        .mount()
+        .await;
+    let repair_handle = room.handle.clone();
+    let repair = tokio::spawn(async move {
+        repair_handle
+            .continue_history_repair(room_id.to_owned(), HistoryRepairBudget::default())
+            .await
+    });
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let requested = room
+                .server
+                .received_requests()
+                .await
+                .unwrap()
+                .iter()
+                .any(|request| {
+                    request.url.path().ends_with("/messages")
+                        && request
+                            .url
+                            .query_pairs()
+                            .any(|(key, value)| key == "from" && value == "cancel-prev")
+                });
+            if requested {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("bridge request did not start");
+    room.handle
+        .cancel_history_repair(room_id.to_owned())
+        .await
+        .unwrap();
+    let report = repair.await.unwrap().unwrap();
+    assert_eq!(report.phase, HistoryRepairPhase::Paused);
+    assert_eq!(report.stop, Some(HistoryRepairStop::Cancelled));
+    assert!(report.recent_available);
+    assert_eq!(report.pages, 1);
+
+    let mut reset_after_cancel = false;
+    while let Ok(event) = room.events.try_recv() {
+        reset_after_cancel |= matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id);
+    }
+    assert!(!reset_after_cancel);
+
+    shut_down(&room.handle, &mut room.events, room.running).await;
+}
+
+#[tokio::test]
+async fn bridge_reaches_the_cached_anchor_deduplicates_overlap_and_can_continue_after_budget() {
+    let room_id = room_id!("!bridge-repair:localhost");
+    let anchor_id = event_id!("$anchor:localhost");
+    let middle_id = event_id!("$middle:localhost");
+    let recent_id = event_id!("$recent:localhost");
+    let latest_id = event_id!("$latest:localhost");
+    let mut room = open_room(room_id).await;
+    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;
+
+    room.server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("recent-prev")
+            .events(vec![
+                history_event(room_id, recent_id, "recent"),
+                history_event(room_id, middle_id, "middle"),
+            ]))
+        .mock_once()
+        .mount()
+        .await;
+    let initial = room
+        .handle
+        .history_repair(
+            room_id.to_owned(),
+            Some(HistoryRepairPolicy::Bridge),
+            report_budget(1),
+        )
+        .await
+        .unwrap();
+    assert_eq!(initial.phase, HistoryRepairPhase::Paused);
+    assert_eq!(initial.stop, Some(HistoryRepairStop::Budget));
+    assert!(initial.recent_available);
+
+    wiremock::Mock::given(|request: &wiremock::Request| {
+        request.url.path().ends_with("/messages")
+            && !request.url.query_pairs().any(|(key, _)| key == "from")
+    })
+    .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+        "start": "head-start",
+        "chunk": [
+            history_event(room_id, latest_id, "latest"),
+            history_event(room_id, recent_id, "recent overlap"),
+            history_event(room_id, middle_id, "middle overlap"),
+        ],
+        "state": [],
+    })))
+    .up_to_n_times(1)
+    .expect(1)
+    .mount(room.server.server())
+    .await;
+    room.server
+        .mock_room_messages()
+        .match_from("recent-prev")
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("older-prev")
+            .events(vec![
+                history_event(room_id, middle_id, "middle overlap"),
+                history_event(room_id, anchor_id, "cached anchor overlap"),
+            ]))
+        .mock_once()
+        .mount()
+        .await;
+
+    let report = room
+        .handle
+        .continue_history_repair(room_id.to_owned(), HistoryRepairBudget::default())
+        .await
+        .unwrap();
+    assert_eq!(report.strategy, HistoryRepairStrategy::Bridge);
+    assert_eq!(report.phase, HistoryRepairPhase::Ready, "{report:?}");
+    assert_eq!(report.continuity, HistoryContinuity::Verified);
+    assert_eq!(report.pages, 3);
+
+    let reset = receive_until(&mut room.events, |event| {
+        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
+            && event_messages(event)
+                .into_iter()
+                .any(|event| event.event_id.as_deref() == Some(latest_id))
+    })
+    .await;
+    let reset_ids = event_messages(&reset)
+        .into_iter()
+        .filter_map(|event| event.event_id.as_ref().map(ToString::to_string))
+        .collect::<Vec<_>>();
+    assert_eq!(
+        reset_ids,
+        vec![
+            anchor_id.to_string(),
+            middle_id.to_string(),
+            recent_id.to_string(),
+            latest_id.to_string(),
+        ]
+    );
+
+    shut_down(&room.handle, &mut room.events, room.running).await;
+}
+
+#[tokio::test]
+async fn a_recent_window_containing_the_cached_anchor_overlaps_under_every_policy() {
+    let cases = [
+        (
+            room_id!("!policy-recent:localhost"),
+            event_id!("$policy-recent-anchor:localhost"),
+            HistoryRepairPolicy::StartRecent,
+        ),
+        (
+            room_id!("!policy-bridge:localhost"),
+            event_id!("$policy-bridge-anchor:localhost"),
+            HistoryRepairPolicy::Bridge,
+        ),
+        (
+            room_id!("!policy-short-gap:localhost"),
+            event_id!("$policy-short-anchor:localhost"),
+            HistoryRepairPolicy::BridgeShortGaps { threshold_ms: 0 },
+        ),
+    ];
+
+    for (room_id, anchor_id, policy) in cases {
+        let mut room = open_room(room_id).await;
+        sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;
+        room.server
+            .mock_room_messages()
+            .ok(
+                RoomMessagesResponseTemplate::default().events(vec![history_event(
+                    room_id,
+                    anchor_id,
+                    "same anchor in recent history",
+                )]),
+            )
+            .mock_once()
+            .mount()
+            .await;
+
+        let report = room
+            .handle
+            .history_repair(
+                room_id.to_owned(),
+                Some(policy.clone()),
+                HistoryRepairBudget::default(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(report.requested_policy, policy);
+        assert_eq!(report.strategy, HistoryRepairStrategy::Overlap);
+        assert_eq!(report.continuity, HistoryContinuity::Verified);
+        assert_eq!(report.pages, 1);
+        assert!(report.recent_available);
+
+        let requests = room.server.received_requests().await.unwrap();
+        assert_eq!(
+            requests
+                .iter()
+                .filter(|request| request.url.path().ends_with("/messages"))
+                .count(),
+            1
+        );
+        shut_down(&room.handle, &mut room.events, room.running).await;
+    }
+}
+
+#[tokio::test]
+async fn short_gap_policy_stays_recent_when_coverage_is_unknown() {
+    let room_id = room_id!("!unknown-coverage:localhost");
+    let anchor_id = event_id!("$unknown-anchor:localhost");
+    let fresh_id = event_id!("$unknown-fresh:localhost");
+    let mut room = open_room(room_id).await;
+    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;
+    room.server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("recent-prev")
+            .events(vec![history_event(room_id, fresh_id, "recent window")]))
+        .mock_once()
+        .mount()
+        .await;
+
+    let report = room
+        .handle
+        .history_repair(
+            room_id.to_owned(),
+            Some(HistoryRepairPolicy::BridgeShortGaps {
+                threshold_ms: 60_000,
+            }),
+            HistoryRepairBudget::default(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(report.coverage_duration_ms, None);
+    assert!(!report.threshold_evaluated);
+    assert_eq!(report.strategy, HistoryRepairStrategy::Recent);
+    assert_eq!(report.continuity, HistoryContinuity::Unresolved);
+    assert_eq!(report.pages, 1);
+
+    let requests = room.server.received_requests().await.unwrap();
+    assert_eq!(
+        requests
+            .iter()
+            .filter(|request| request.url.path().ends_with("/messages"))
+            .count(),
+        1
+    );
+    shut_down(&room.handle, &mut room.events, room.running).await;
+}

macltok

Commit 80381e06a573

Commit `80381e06a573`