macltok

public

Star0

Fork 0

Watch0

68 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `d9b96952ec60`

tarball

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

13 files changed+756-927

history: drop the dead repair api and report every edge once

Start, Continue, the policy and budget types and most report fields were validated, echoed or never emitted, so consumers branched on values that could not occur. A recent load now settles only through RecentHistoryChanged and a fill only through HistoryRepairReport, and cancel or close publish Cancelled or Idle instead of Failed. Completions no longer resync the room timeline, since the watcher drains what the flush queued before the publication barrier, and joined gaps are dropped in normalisation rather than shown as Verified. The mentions repair policy goes too, as it was stored and never read.

Modified file: docs/segmented-history.md+91 additions−176 deletions@@ -1,198 +1,113 @@
-# Segmented history implementation
-
-The durable implementation is in the macltok `segmented-history` feature
-worktree. The combined SDK revision
-`9b2c17fb10703d90b177b02588912a34cecfa11f` is published on its
-`segmented-history` bookmark. It includes the reaction compatibility fix from
-`c0a` and segmented history from `49a9`. The macltok history branch includes
-pack fixes from `e4bb841a` and the `macltok-pack-fixes` worktree through
-`d1313f47`. All nine SDK pins and both lockfiles use the combined SDK revision.
-The unified macltok bookmark is named `segmented-history`.
-
-## Current behavior
-
-Opening a room loads retained history and its gap rows, then fetches a bounded
-recent window from a fresh server position. The recent fetch is capped at four
-pages of 50 raw events. It preserves retained sections and their older cursors.
-If it cannot establish continuity, it installs a durable gap between the
-sections. It stops when continuity is established or accessible history ends.
-
-`HistoryRepairCommand::FillGap` takes a room ID and stable gap transaction ID.
-It fetches up to four backward pages of 50 events and advances only that gap's
-saved cursor. Fill progress and events commit atomically. Concurrent history
-operations for the room return `RoomBusy`. An empty page with an advancing
-cursor keeps the gap. A repeated cursor pauses it. Accessible history ending
-before the retained boundary marks the interval unavailable. Only reaching the
-retained boundary joins sections. Already joined gaps are harmless to stale
-callers.
-
-The cache stores gap identity, boundary evidence, cursor, and status alongside
-the encrypted linked event chunks. SQLite supports legacy token data and the
-new metadata. IndexedDB stores the optional encrypted metadata field. Room
-close and restart retain sections and fill cursors. Timeline snapshots and
-incremental updates project gaps as `TimelineItem::HistoryGap`. Gap rows retain
-their identity while status changes. Joined markers disappear and unavailable
-intervals remain visible.
-
-The recent fetch can complete while live sync continues. Cache installation
-uses a candidate state and commits events, gap metadata, and cursor together.
-Sync publication is serialized with gap installation. Existing loaded timelines
-receive incremental insertions that preserve retained item identities and
-pending sends. Loading history outside the current timeline window still needs
-a reset.
-
-Each SDK room cache owns a coordinator task. Recent callers share an in-flight
-request. Cancelling one caller preserves the operation for other subscribers,
-and cancelling all callers stops fetching before installation. Started commits
-finish before the coordinator accepts conflicting work. No mutex guard crosses
-the network request. The idle task holds only a weak cache reference, so it does
-not keep a room cache alive. The SDK UI calls `wait_for_history` as a fence,
-then takes an atomic event and gap snapshot while replacing its cache
-subscription. The macltok watcher is stopped before its timeline reset is sent,
-so old diffs cannot arrive after the reset. A mirror publication acknowledgement
-completes before the command response is released. The room actor starts recent
-loading in the background when the room opens, so opening does not wait for the
-network request.
-
-Limited sync results persist explicit gap markers in the event cache. Later sync
-updates preserve those markers until the pagination chain reaches the retained
-boundary. The integration uses the room's live timeline flush and watcher reset
-path. It does not reseed the cache. Thread and pinned timelines remain open
-through this update path.
-
-## Public and wire shapes
-
-`ClientHandle::fill_history_gap(room_id, gap_id)` is the convenience method for
-the targeted fill. It returns `Result<(), Error>`. The underlying command is
-`HistoryRepairCommand::FillGap { room_id, gap_id }`, sent as
-`Command::HistoryRepair`. The older policy and staged repair command variants
-remain available for serde compatibility, but room opening no longer consults
-the saved policy. `Continue` is invalid because continuation must identify its
-gap. `GetPolicy` and `SetPolicy` remain compatibility operations and do not
-control opening.
-
-The inline gap lifecycle value `Fetching` is reserved and is not currently
-emitted by timeline projection. During a targeted fill, loading is signaled by
-a `HistoryRepairReport` with `gap_id` set and `phase` set to `Bridging`. Recent
-loading has its own room keyed `RecentHistoryState` with
-`Idle`, `Loading`, `Loaded`, or `Failed` values. The mirror exposes that state in
-`recent_history`, separately from gap continuity and repair progress.
-
-`TimelineItem::HistoryGap` carries a `HistoryGap` with `id`,
-`older_boundary`, `newer_boundary`, `lifecycle`, `continuity`, and `failure`.
-Lifecycle values are `Idle`, `Fetching`, `Paused`, `Failed`, and `Unavailable`.
-Continuity values are `Unresolved`, `Verified`, and `AccessibleStart`. The gap
-ID is an owned transaction ID. Boundary values are optional owned event IDs.
-Failure is an optional displayable string. `TimelineState.items` contains these
-items in order with event rows. Mirror snapshots serialize the same item shape,
-and incremental timeline changes carry it through the normal reset or changed
-diff.
-
-Serde enums use externally tagged variants. A command payload is shaped as
-follows, with the outer `Command` adding its `HistoryRepair` tag.
+# segmented history
+
+a room's history is sections of events with durable gaps between them. the sdk
+event cache keeps the sections, gap identity, boundaries and each gap's cursor,
+encrypted at rest, across room close and restart. macltok shows gaps inline and
+fills them on request.
+
+## opening a room
+
+opening loads retained history and its gaps, then starts a recent load in the
+background, so opening never waits on the network. the recent load fetches up
+to four pages of 50 events from a fresh server position. if it reaches the
+newest retained event the sections join. if it doesn't, a gap is left between
+the fresh window and what was retained.
+
+`Event::RecentHistoryChanged` reports it, one event per edge.
+
+- `Loading` as it starts
+- `Loaded`, `Cancelled` or `Failed(message)` as it settles
+- `Idle` when the room closes, whatever was happening
+
+`Failed` stays until `HistoryRepairCommand::RetryRecent` or a reopen. a failed
+load falls back to paginating `EngineConfig::initial_history_limit` events, so
+the room isn't left empty.
+
+## filling a gap
+
+`HistoryRepairCommand::FillGap { room_id, gap_id }` walks one gap backwards, up
+to four pages of 50, each page committing its events and the gap's new cursor
+together. `Event::HistoryRepairChanged` carries a `HistoryRepairReport`, once
+with phase `Bridging` as the fill starts and once more as it settles.
+
+- `Ready`, continuity `Verified`. the older section was reached and the gap is gone
+- `Paused` with a stop. `Budget` after four pages, `CursorStalled` when the
+  server repeats a cursor, `HistoryUnavailable` (continuity `AccessibleStart`)
+  when accessible history ends first, `Cancelled`
+- `Paused` with a `failure`, and the gap row goes `Failed`
+
+a gap that is already joined or unavailable answers at once and counts no page.
+
+## one operation per room
+
+a room runs one history operation at a time. `RetryRecent` joins a recent load
+that is already running. everything else that overlaps, a fill during a recent
+load, a recent load during a fill, a second fill, answers `RoomBusy`.
+
+`HistoryRepairCommand::Cancel` stops whatever is running. its own answer is
+`Ok`, the cancelled command answers `Interrupted`, and the published outcome is
+`Cancelled`, never a failure. pages already committed stay.
+
+## the timeline
+
+gaps appear in `TimelineState.items` as `TimelineItem::HistoryGap`, between the
+sections they separate. a row keeps its id while its lifecycle changes, and is
+removed once the gap joins. joined gaps never appear.
+
+history lands as `TimelineChanged` diffs, not a `TimelineReset`. the sdk
+currently replays the whole live timeline when it applies committed history,
+so those diffs can span the timeline until it gets a non-replacing fence.
+
+a command's answer comes after its outcome is published, and after the mirror
+has applied it when one is attached.
+
+## known holes
+
+a limited sync for a room with no events in memory records a plain gap instead
+of a history gap (review 2026-10-25 S1), so an interval can still render as
+continuous. both `prev_batch` drops in the sdk's `handle_sync` remain.
+
+## wire shapes
+
+serde enums are externally tagged.
 
 ```json
-{
-  "HistoryRepair": {
-    "FillGap": {
-      "room_id": "!room:example.org",
-      "gap_id": "opaque-gap-transaction-id"
-    }
-  }
-}
+{ "HistoryRepair": { "FillGap": { "room_id": "!room:example.org", "gap_id": "opaque" } } }
+{ "HistoryRepair": { "RetryRecent": { "room_id": "!room:example.org" } } }
+{ "HistoryRepair": { "Cancel": { "room_id": "!room:example.org" } } }
 ```
 
-A gap row in `TimelineState.items` has the following shape.
+a gap row.
 
 ```json
 {
   "HistoryGap": {
-    "id": "opaque-gap-transaction-id",
+    "id": "opaque",
     "older_boundary": "$older:example.org",
     "newer_boundary": "$newer:example.org",
     "lifecycle": "Idle",
-    "continuity": "Unresolved",
     "failure": null
   }
 }
 ```
 
-The mirror snapshot stores recent loading separately by room ID. This fragment
-shows the serialized `State.recent_history` field.
+lifecycle is `Idle`, `Paused`, `Failed` or `Unavailable`. `failure` is set with
+`Failed`.
+
+a settled fill.
 
 ```json
 {
-  "recent_history": {
-    "!room:example.org": "Loading"
-  }
+  "room_id": "!room:example.org",
+  "gap_id": "opaque",
+  "pages": 2,
+  "events": 63,
+  "phase": "Ready",
+  "continuity": "Verified",
+  "stop": null,
+  "failure": null
 }
 ```
 
-The final history repair publication acknowledgement is internal, skipped by
-serde, and not part of application wire data. The mirror applies the report
-before acknowledging it, so command completion follows publication.
-
-## Malice integration
-
-Pin the published macltok `segmented-history` revision in Malice's native and
-worker manifests and regenerate their lockfiles. The SDK dependency is already
-pinned by macltok.
-
-Malice should remove the saved policy and threshold controls from room opening.
-Render each inline gap between its neighboring event sections. A fill should
-preserve the visible event anchor, composer draft, reply or edit state, pending
-sends, and encryption identity. Keep event-context windows independent from
-live room history.
-
-## Lost interval investigation
-
-The reported failure showed retained events followed by newly sent messages,
-with an omitted interval of roughly eight to ten hours presented as continuous.
-Older pagination continued before the retained buffer, and a page refresh did
-not recover the missing range. A refresh can reconnect to the existing
-SharedWorker, so it does not prove that the engine restarted.
-
-The earlier SDK cache path could discard a limited-sync boundary token when a
-timeline was reported unlimited despite cached events. It could also discard a
-token when all incoming events were duplicates. Both cases can hide a real
-discontinuity from older pagination. Initial pagination could also finish
-against retained history before fresh sync arrived. These are concrete failure
-paths, but none has been confirmed as the production cause. The durable gaps
-preserve discontinuities when observed. Startup, subscription, and sync
-diagnostics are still needed to identify where this particular interval was
-lost.
-
-## Validation commands
-
-Run native checks from `../macltok-history-repair` with the repository's Nix
-toolchain.
-
-```sh
-nix develop --command cargo test --all-targets
-nix develop --command cargo test --all-targets --features sqlite
-nix develop --command cargo clippy --all-targets -- -D warnings
-nix develop --command cargo clippy --all-targets --features sqlite -- -D warnings
-```
-
-For the browser worker runtime, use the web devshell. The harness starts a local
-homeserver and runs both worker targets in Firefox.
-
-```sh
-nix develop .#web -c nu harness/web/run.nu
-```
-
-The flake checks include native and WebAssembly compilation with IndexedDB and
-the web worker runtime. The default native gate passed all 86 tests. The
-aggregate `nix flake check` run stopped during the SQLite and mirror build.
-The remaining aggregate gates were waived for this handoff and have not passed.
-The focused native history repair and engine suites passed all 11 tests with
-SQLite, mirror, and serde enabled against the published combined SDK revision.
-
-Seven focused SDK recent-history tests and seven segmented-history tests pass.
-The SDK WebAssembly compile passes with IndexedDB enabled. SDK UI timeline
-checks pass with 123 unit tests. Separate SDK room checks pass with seven unit
-tests and 14 timed integration tests. Two headless Firefox tests pass for the
-IndexedDB shared gap storage contract with plaintext and encrypted stores. The
-full SQLite and mirror gate, aggregate WebAssembly gate, and browser worker
-runtime gate remain unverified. Complete worker restart and encrypted room
-recovery remain integration acceptance checks.
+the mirror keeps `State.recent_history`, a room without an entry being idle,
+and `State.repairs`, the last fill report per room. closing a room drops both.

Modified file: src/engine/mod.rs+3 additions−24 deletions@@ -61,6 +61,7 @@ use crate::{
 pub struct EngineConfig {
     pub command_capacity: usize,
     pub event_capacity: usize,
+    /// what a thread paginates as it opens, and a room if its recent load fails. 0 for neither
     pub initial_history_limit: u16,
     pub max_media_bytes: usize,
     pub max_media_sources: usize,
@@ -981,30 +982,9 @@ 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 }
+            HistoryRepairCommand::Cancel { room_id }
+            | HistoryRepairCommand::RetryRecent { room_id }
             | HistoryRepairCommand::FillGap { 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, .. }) => {
@@ -1080,7 +1060,6 @@ 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(),
             },
             panics,
             opening: vec![reply],

Modified file: src/engine/repair.rs+189 additions−256 deletions@@ -1,4 +1,8 @@
-use matrix_sdk::{Room, event_cache::HistoryGapFillOutcome};
+use matrix_sdk::{
+    Room,
+    event_cache::{EventCacheError, HistoryGapFillOutcome, HistoryGapStatus},
+};
+use ruma::{OwnedRoomId, OwnedTransactionId};
 use tokio::sync::watch;
 
 use super::{
@@ -6,18 +10,19 @@ use super::{
     timeline::SessionContext,
 };
 use crate::{
-    CommandOutput, Error, ErrorKind, HistoryContinuity, HistoryRepairBudget, HistoryRepairCommand,
-    HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairReport, HistoryRepairStop,
-    HistoryRepairStrategy, RecentHistoryState, transport::Responder,
+    CommandOutput, Error, ErrorKind, Event, HistoryContinuity, HistoryRepairCommand,
+    HistoryRepairPhase, HistoryRepairReport, HistoryRepairStop, RecentHistoryState,
+    transport::Responder,
 };
 
 const MAX_GAP_PAGES: u32 = 4;
 
 struct Running {
-    task: JoinHandle<CompletionResult>,
+    task: JoinHandle<Finished>,
     cancel: watch::Sender<bool>,
     responders: Vec<PendingResponder>,
-    recent: bool,
+    gap_id: Option<OwnedTransactionId>,
+    closing: bool,
 }
 
 struct PendingResponder {
@@ -25,14 +30,33 @@ struct PendingResponder {
     cancel_ack: bool,
 }
 
+pub(super) enum Outcome {
+    Recent(RecentHistoryState),
+    Fill(HistoryRepairReport),
+}
+
 pub(super) struct Completion {
-    pub(super) report: HistoryRepairReport,
-    pub(super) recent_state: Option<RecentHistoryState>,
+    pub(super) outcome: Outcome,
+    /// history went into the cache, so the room timeline has to show it before anyone hears
+    pub(super) committed: bool,
+    pub(super) closing: bool,
     pub(super) result: Result<CommandOutput, Error>,
     responders: Vec<PendingResponder>,
 }
 
 impl Completion {
+    /// the room timeline couldn't show what was committed
+    pub(super) fn unpublished(&mut self, error: Error) {
+        match &mut self.outcome {
+            Outcome::Recent(state) => *state = RecentHistoryState::Failed(error.message.clone()),
+            Outcome::Fill(report) => {
+                report.phase = HistoryRepairPhase::Paused;
+                report.failure = Some(error.clone());
+            }
+        }
+        self.result = Err(error);
+    }
+
     pub(super) fn resolve(
         self,
         result: Result<CommandOutput, Error>,
@@ -48,10 +72,16 @@ impl Completion {
     }
 }
 
-struct CompletionResult {
-    report: HistoryRepairReport,
-    recent_state: Option<RecentHistoryState>,
-    result: Result<CommandOutput, Error>,
+enum Stopped {
+    Done,
+    Cancelled,
+    Failed(Error),
+}
+
+struct Finished {
+    stopped: Stopped,
+    committed: bool,
+    report: Option<HistoryRepairReport>,
 }
 
 pub(super) struct RepairController {
@@ -74,8 +104,7 @@ impl RepairController {
     }
 
     pub(super) async fn start_recent(&mut self) {
-        self.start_recent_with(HistoryRepairPolicy::default(), None, true, Vec::new())
-            .await;
+        self.start_recent_with(None).await;
     }
 
     pub(super) async fn execute(&mut self, command: HistoryRepairCommand, reply: Responder) {
@@ -91,56 +120,17 @@ impl RepairController {
                     reply.done(Ok(()));
                 }
             }
-            HistoryRepairCommand::Start { policy, budget, .. } => {
-                if let Err(error) = validate_budget(&budget) {
-                    reply.done(Err(error));
-                    return;
-                }
-                let policy = policy.unwrap_or_default();
-                self.start_recent_with(policy, Some(reply), true, Vec::new())
-                    .await;
-            }
-            HistoryRepairCommand::StartRecent { .. } => {
-                self.start_recent_with(
-                    HistoryRepairPolicy::default(),
-                    Some(reply),
-                    true,
-                    Vec::new(),
-                )
-                .await;
-            }
+            HistoryRepairCommand::RetryRecent { .. } => self.start_recent_with(Some(reply)).await,
             HistoryRepairCommand::FillGap { gap_id, .. } => {
                 self.start_gap_fill(gap_id, reply).await;
             }
-            HistoryRepairCommand::Continue { .. } => reply.done(Err(Error::new(
-                ErrorKind::Invalid,
-                "continue requires a history gap ID, use FillGap",
-            ))),
-            HistoryRepairCommand::GetPolicy => {
-                reply.send(
-                    self.context
-                        .mentions
-                        .repair_policy()
-                        .await
-                        .map(CommandOutput::HistoryRepairPolicy),
-                );
-            }
-            HistoryRepairCommand::SetPolicy(policy) => {
-                reply.done(self.context.mentions.set_repair_policy(policy).await);
-            }
         }
     }
 
-    async fn start_recent_with(
-        &mut self,
-        policy: HistoryRepairPolicy,
-        responder: Option<Responder>,
-        publish_recent: bool,
-        mut responders: Vec<PendingResponder>,
-    ) {
+    async fn start_recent_with(&mut self, responder: Option<Responder>) {
         if let Some(running) = &mut self.running {
             if let Some(responder) = responder {
-                if running.recent {
+                if running.gap_id.is_none() {
                     running.responders.push(PendingResponder {
                         responder,
                         cancel_ack: false,
@@ -151,55 +141,45 @@ impl RepairController {
             }
             return;
         }
-        if let Some(responder) = responder {
-            responders.push(PendingResponder {
-                responder,
-                cancel_ack: false,
-            });
-        }
-        if publish_recent
-            && self
-                .context
-                .events
-                .send(crate::Event::RecentHistoryChanged {
-                    room_id: self.room.room_id().to_owned(),
-                    state: RecentHistoryState::Loading,
-                })
-                .await
-                .is_err()
-        {
-            for responder in responders {
-                responder.responder.done(Err(Error::interrupted()));
+        let event = Event::RecentHistoryChanged {
+            room_id: self.room.room_id().to_owned(),
+            state: RecentHistoryState::Loading,
+        };
+        if self.context.events.send(event).await.is_err() {
+            if let Some(responder) = responder {
+                responder.done(Err(Error::interrupted()));
             }
             return;
         }
         let (cancel, receiver) = watch::channel(false);
         let room = self.room.clone();
-        let room_id = self.room.room_id().to_owned();
-        let task = spawn(async move { load_recent(room, room_id, policy, receiver).await });
+        let task = spawn(async move { load_recent(room, receiver).await });
         self.running = Some(Running {
             task,
             cancel,
-            responders,
-            recent: true,
+            responders: responder
+                .map(|responder| PendingResponder {
+                    responder,
+                    cancel_ack: false,
+                })
+                .into_iter()
+                .collect(),
+            gap_id: None,
+            closing: false,
         });
     }
 
-    async fn start_gap_fill(&mut self, gap_id: ruma::OwnedTransactionId, responder: Responder) {
+    async fn start_gap_fill(&mut self, gap_id: OwnedTransactionId, responder: Responder) {
         if self.is_running() {
             responder.done(Err(room_busy()));
             return;
         }
         let room = self.room.clone();
-        let room_id = self.room.room_id().to_owned();
-        let mut progress = report(room_id.clone(), HistoryRepairPolicy::default());
-        progress.gap_id = Some(gap_id.clone());
-        progress.strategy = HistoryRepairStrategy::Bridge;
-        progress.phase = HistoryRepairPhase::Bridging;
+        let progress = report(self.room.room_id().to_owned(), gap_id.clone());
         if self
             .context
             .events
-            .send(crate::Event::HistoryRepairChanged(progress))
+            .send(Event::HistoryRepairChanged(progress.clone()))
             .await
             .is_err()
         {
@@ -207,7 +187,7 @@ impl RepairController {
             return;
         }
         let (cancel, receiver) = watch::channel(false);
-        let task = spawn(async move { fill_gap(room, room_id, gap_id, receiver).await });
+        let task = spawn(async move { fill_gap(room, progress, receiver).await });
         self.running = Some(Running {
             task,
             cancel,
@@ -215,7 +195,8 @@ impl RepairController {
                 responder,
                 cancel_ack: false,
             }],
-            recent: false,
+            gap_id: Some(gap_id),
+            closing: false,
         });
     }
 
@@ -227,222 +208,174 @@ impl RepairController {
 
     pub(super) async fn shutdown(&mut self) -> Option<Completion> {
         let running = self.running.as_mut()?;
+        running.closing = true;
         running.cancel.send_replace(true);
         let result = (&mut running.task).await;
         Some(self.finish_task(result))
     }
 
-    fn finish_task(&mut self, result: Result<CompletionResult, JoinError>) -> Completion {
+    fn finish_task(&mut self, result: Result<Finished, JoinError>) -> Completion {
         let running = self.running.take().expect("repair task exists");
-        let result = match result {
-            Ok(result) => result,
-            Err(_) => CompletionResult {
-                report: failed_report(
-                    self.room.room_id().to_owned(),
-                    HistoryRepairPolicy::default(),
-                    Error::interrupted(),
-                ),
-                recent_state: running.recent.then(|| {
-                    RecentHistoryState::Failed(
-                        "recent history operation was interrupted".to_owned(),
-                    )
-                }),
-                result: Err(Error::interrupted()),
-            },
+        let finished = result.unwrap_or(Finished {
+            stopped: Stopped::Failed(Error::interrupted()),
+            committed: false,
+            report: None,
+        });
+        let result = match &finished.stopped {
+            Stopped::Done => Ok(CommandOutput::Done),
+            Stopped::Cancelled => Err(Error::interrupted()),
+            Stopped::Failed(error) => Err(error.clone()),
+        };
+        let outcome = match running.gap_id {
+            None => Outcome::Recent(match finished.stopped {
+                _ if running.closing => RecentHistoryState::Idle,
+                Stopped::Done => RecentHistoryState::Loaded,
+                Stopped::Cancelled => RecentHistoryState::Cancelled,
+                Stopped::Failed(error) => RecentHistoryState::Failed(error.message),
+            }),
+            Some(gap_id) => {
+                let mut report = finished
+                    .report
+                    .unwrap_or_else(|| report(self.room.room_id().to_owned(), gap_id));
+                match finished.stopped {
+                    Stopped::Done => {}
+                    Stopped::Cancelled => report.stop = Some(HistoryRepairStop::Cancelled),
+                    Stopped::Failed(error) => report.failure = Some(error),
+                }
+                report.phase = if report.stop.is_none() && report.failure.is_none() {
+                    HistoryRepairPhase::Ready
+                } else {
+                    HistoryRepairPhase::Paused
+                };
+                Outcome::Fill(report)
+            }
         };
         Completion {
-            report: result.report,
-            recent_state: result.recent_state,
-            result: result.result,
+            outcome,
+            committed: finished.committed,
+            closing: running.closing,
+            result,
             responders: running.responders,
         }
     }
 }
 
-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(())
-}
-
-async fn load_recent(
-    room: Room,
-    room_id: ruma::OwnedRoomId,
-    policy: HistoryRepairPolicy,
-    mut cancel: watch::Receiver<bool>,
-) -> CompletionResult {
-    let base = report(room_id.clone(), policy);
+async fn load_recent(room: Room, mut cancel: watch::Receiver<bool>) -> Finished {
     let outcome = tokio::select! {
         biased;
-        _ = cancel.changed() => Err(Error::interrupted()),
+        _ = cancel.changed() => None,
         result = async {
-            let (cache, _handles) = room.event_cache().await.map_err(cache_error)?;
-            cache.load_recent_history().await.map_err(cache_error)
-        } => result,
+            let (cache, _handles) = room.event_cache().await?;
+            cache.load_recent_history().await
+        } => Some(result),
     };
-    match outcome {
-        Ok(outcome) => {
-            let mut report = base;
-            report.pages = outcome.pages;
-            report.events = outcome.events.min(u32::MAX as usize) as u32;
-            report.recent_available = true;
-            report.phase = HistoryRepairPhase::Ready;
-            match outcome.gap.map(|gap| gap.status) {
-                None => report.continuity = HistoryContinuity::AccessibleStart,
-                Some(matrix_sdk::event_cache::HistoryGapStatus::Joined) => {
-                    report.continuity = HistoryContinuity::Verified
-                }
-                Some(matrix_sdk::event_cache::HistoryGapStatus::Unavailable) => {
-                    report.continuity = HistoryContinuity::AccessibleStart;
-                    report.stop = Some(HistoryRepairStop::HistoryUnavailable);
-                    report.phase = HistoryRepairPhase::Paused;
-                }
-                Some(matrix_sdk::event_cache::HistoryGapStatus::Paused) => {
-                    report.stop = Some(HistoryRepairStop::CursorStalled);
-                    report.phase = HistoryRepairPhase::Paused;
-                }
-                Some(matrix_sdk::event_cache::HistoryGapStatus::Failed { message }) => {
-                    let error = Error::new(ErrorKind::Network, message.clone());
-                    report.failure = Some(error.clone());
-                    report.phase = HistoryRepairPhase::Paused;
-                }
-                Some(matrix_sdk::event_cache::HistoryGapStatus::Idle) => {}
-            }
-            CompletionResult {
-                report: report.clone(),
-                recent_state: Some(RecentHistoryState::Loaded),
-                result: Ok(CommandOutput::HistoryRepair(report)),
-            }
-        }
-        Err(error) => CompletionResult {
-            report: failed_report(room_id, base.requested_policy, error.clone()),
-            recent_state: Some(RecentHistoryState::Failed(error.message.clone())),
-            result: Err(error),
-        },
+    let stopped = match outcome {
+        None => Stopped::Cancelled,
+        Some(Ok(_)) => Stopped::Done,
+        Some(Err(error)) => Stopped::Failed(cache_error(error)),
+    };
+    Finished {
+        committed: matches!(stopped, Stopped::Done),
+        stopped,
+        report: None,
     }
 }
 
 async fn fill_gap(
     room: Room,
-    room_id: ruma::OwnedRoomId,
-    gap_id: ruma::OwnedTransactionId,
+    mut report: HistoryRepairReport,
     mut cancel: watch::Receiver<bool>,
-) -> CompletionResult {
-    let mut report = report(room_id, HistoryRepairPolicy::default());
-    report.gap_id = Some(gap_id.clone());
-    report.strategy = HistoryRepairStrategy::Bridge;
-    let outcome = async {
+) -> Finished {
+    let mut committed = false;
+    let stopped = async {
         let (cache, _handles) = room.event_cache().await.map_err(cache_error)?;
-        let mut last_gap = None;
         for _ in 0..MAX_GAP_PAGES {
             let next = tokio::select! {
                 biased;
-                _ = cancel.changed() => return Err(Error::interrupted()),
-                result = cache.fill_history_gap_once(gap_id.clone()) => result.map_err(cache_error)?,
+                _ = cancel.changed() => return Ok(Stopped::Cancelled),
+                result = cache.fill_history_gap_once(report.gap_id.clone()) => {
+                    result.map_err(cache_error)?
+                }
             };
-            match next {
+            let (gap, events) = match next {
                 HistoryGapFillOutcome::NotFound => {
-                    return Err(Error::new(ErrorKind::EventNotFound, "history gap was not found"));
+                    return Err(Error::new(
+                        ErrorKind::EventNotFound,
+                        "history gap was not found",
+                    ));
                 }
                 HistoryGapFillOutcome::Busy => return Err(room_busy()),
-                HistoryGapFillOutcome::Updated { gap, events } => {
-                    report.pages += 1;
-                    report.events = report.events.saturating_add(events.min(u32::MAX as usize) as u32);
-                    match gap.status.clone() {
-                        matrix_sdk::event_cache::HistoryGapStatus::Joined => {
-                            report.continuity = HistoryContinuity::Verified;
-                            last_gap = Some(gap);
-                            break;
-                        }
-                        matrix_sdk::event_cache::HistoryGapStatus::Unavailable => {
-                            report.continuity = HistoryContinuity::AccessibleStart;
-                            report.stop = Some(HistoryRepairStop::HistoryUnavailable);
-                            last_gap = Some(gap);
-                            break;
-                        }
-                        matrix_sdk::event_cache::HistoryGapStatus::Paused => {
-                            report.stop = Some(HistoryRepairStop::CursorStalled);
-                            last_gap = Some(gap);
-                            break;
-                        }
-                        matrix_sdk::event_cache::HistoryGapStatus::Failed { message } => {
-                            return Err(Error::new(ErrorKind::Network, message));
-                        }
-                        matrix_sdk::event_cache::HistoryGapStatus::Idle => last_gap = Some(gap),
-                    }
+                HistoryGapFillOutcome::Updated { gap, events } => (gap, events),
+            };
+            // the sdk answers an already joined or unavailable gap without asking the server.
+            // an empty first page that settles the gap looks the same and goes uncounted too
+            let settled_before = report.pages == 0
+                && events == 0
+                && matches!(
+                    gap.status,
+                    HistoryGapStatus::Joined | HistoryGapStatus::Unavailable
+                );
+            if !settled_before {
+                report.pages += 1;
+                committed = true;
+            }
+            report.events = report
+                .events
+                .saturating_add(u32::try_from(events).unwrap_or(u32::MAX));
+            match gap.status {
+                HistoryGapStatus::Joined => {
+                    report.continuity = HistoryContinuity::Verified;
+                    return Ok(Stopped::Done);
                 }
+                HistoryGapStatus::Unavailable => {
+                    report.continuity = HistoryContinuity::AccessibleStart;
+                    report.stop = Some(HistoryRepairStop::HistoryUnavailable);
+                    return Ok(Stopped::Done);
+                }
+                HistoryGapStatus::Paused => {
+                    report.stop = Some(HistoryRepairStop::CursorStalled);
+                    return Ok(Stopped::Done);
+                }
+                HistoryGapStatus::Failed { message } => {
+                    return Err(Error::new(ErrorKind::Network, message));
+                }
+                HistoryGapStatus::Idle => {}
             }
         }
-        if report.pages == MAX_GAP_PAGES
-            && report.continuity == HistoryContinuity::Unresolved
-            && report.stop.is_none()
-        {
-            report.stop = Some(HistoryRepairStop::Budget);
-        }
-        Ok::<_, Error>(last_gap)
+        report.stop = Some(HistoryRepairStop::Budget);
+        Ok(Stopped::Done)
     }
-    .await;
-    let command_result = match outcome {
-        Ok(_) => {
-            report.recent_available = true;
-            report.phase = if report.stop.is_some() {
-                HistoryRepairPhase::Paused
-            } else {
-                HistoryRepairPhase::Ready
-            };
-            Ok(CommandOutput::Done)
-        }
-        Err(error) => {
-            report.failure = Some(error.clone());
-            report.phase = HistoryRepairPhase::Paused;
-            Err(error)
-        }
-    };
-    CompletionResult {
-        report,
-        recent_state: None,
-        result: command_result,
+    .await
+    .unwrap_or_else(Stopped::Failed);
+    Finished {
+        stopped,
+        committed,
+        report: Some(report),
     }
 }
 
-fn report(room_id: ruma::OwnedRoomId, policy: HistoryRepairPolicy) -> HistoryRepairReport {
+fn report(room_id: OwnedRoomId, gap_id: OwnedTransactionId) -> HistoryRepairReport {
     HistoryRepairReport {
         room_id,
-        gap_id: None,
-        requested_policy: policy,
-        strategy: HistoryRepairStrategy::Recent,
-        coverage_duration_ms: None,
-        threshold_evaluated: false,
+        gap_id,
         pages: 0,
         events: 0,
-        phase: HistoryRepairPhase::FetchingRecent,
+        phase: HistoryRepairPhase::Bridging,
         continuity: HistoryContinuity::Unresolved,
-        recent_available: false,
         stop: None,
         failure: None,
     }
 }
 
-fn failed_report(
-    room_id: ruma::OwnedRoomId,
-    policy: HistoryRepairPolicy,
-    error: Error,
-) -> HistoryRepairReport {
-    let mut report = report(room_id, policy);
-    report.phase = HistoryRepairPhase::Paused;
-    report.failure = Some(error);
-    report
-}
-
-fn cache_error(source: matrix_sdk::event_cache::EventCacheError) -> Error {
-    Error::with_source(ErrorKind::Store, "room history operation failed", source)
+fn cache_error(source: EventCacheError) -> Error {
+    let kind = match &source {
+        EventCacheError::PaginationError(error) => crate::error::classify_sdk_error(error),
+        EventCacheError::InitialPaginationError(_) => ErrorKind::Network,
+        EventCacheError::RoomNotFound { .. } => ErrorKind::RoomUnknown,
+        EventCacheError::ClientDropped => ErrorKind::Interrupted,
+        _ => ErrorKind::Store,
+    };
+    Error::with_source(kind, "room history operation failed", source)
 }
 
 fn room_busy() -> Error {

Modified file: src/engine/room.rs+43 additions−34 deletions@@ -6,7 +6,7 @@ use ruma::OwnedRoomId;
 use tokio::sync::mpsc;
 
 use super::executor::{AbortHandle, JoinError, TaskSet, is_panic, resume_unwind};
-use super::repair::{Completion, RepairController};
+use super::repair::{Completion, Outcome, RepairController};
 use super::timeline::{SessionContext, TaskGuard, TimelineSession};
 use crate::{
     Error, ErrorKind, Event, HistoryRepairCommand, RecentHistoryState, TimelineCommand, TimelineId,
@@ -109,8 +109,23 @@ impl RoomActor {
             reply.done(Ok(()));
         }
         let close = self.serve_open(&mut timelines, &mut repair).await;
-        if let Some(completion) = repair.shutdown().await {
-            self.finish_repair(completion, &mut timelines).await;
+        let recent_settled = match repair.shutdown().await {
+            Some(completion) => {
+                let recent = matches!(completion.outcome, Outcome::Recent(_));
+                self.finish_repair(completion, &mut timelines).await;
+                recent
+            }
+            None => false,
+        };
+        if !recent_settled {
+            let _ = self
+                .context
+                .events
+                .send(Event::RecentHistoryChanged {
+                    room_id: self.room_id.clone(),
+                    state: RecentHistoryState::Idle,
+                })
+                .await;
         }
         if close.is_none() {
             self.requests.close();
@@ -141,42 +156,36 @@ impl RoomActor {
     }
 
     async fn finish_repair(&self, mut completion: Completion, timelines: &mut Timelines) {
-        let publication = timelines.room.publish_history().await;
-        let mut cancellation = publication.clone();
-        if let Err(error) = publication {
-            completion.report.failure = Some(error.clone());
-            completion.report.phase = crate::HistoryRepairPhase::Paused;
-            completion.result = Err(error.clone());
-            if completion.recent_state.is_some() {
-                completion.recent_state = Some(RecentHistoryState::Failed(error.message));
-            }
+        let mut cancellation = Ok(());
+        if completion.committed
+            && !completion.closing
+            && completion.result.is_ok()
+            && let Err(error) = timelines.room.publish_history().await
+        {
+            completion.unpublished(error.clone());
+            cancellation = Err(error);
         }
+        let event = match &completion.outcome {
+            Outcome::Recent(state) => Event::RecentHistoryChanged {
+                room_id: self.room_id.clone(),
+                state: state.clone(),
+            },
+            Outcome::Fill(report) => Event::HistoryRepairChanged(report.clone()),
+        };
         let mut resolved = completion.result.clone();
-        if let Some(state) = completion.recent_state.clone() {
-            if self
-                .context
-                .events
-                .send(Event::RecentHistoryChanged {
-                    room_id: self.room_id.clone(),
-                    state,
-                })
-                .await
-                .is_err()
-            {
-                resolved = Err(Error::interrupted());
-                cancellation = Err(Error::interrupted());
-            }
-        }
-        if self
-            .context
-            .events
-            .publish_repair(completion.report.clone())
-            .await
-            .is_err()
-        {
+        let events = &self.context.events;
+        if events.send(event).await.is_err() || events.publish_barrier().await.is_err() {
             resolved = Err(Error::interrupted());
             cancellation = Err(Error::interrupted());
         }
+        let limit = self.context.initial_history_limit;
+        if matches!(
+            completion.outcome,
+            Outcome::Recent(RecentHistoryState::Failed(_))
+        ) && limit > 0
+        {
+            timelines.room.paginate(limit, None);
+        }
         completion.resolve(resolved, cancellation);
     }
 

Modified file: src/engine/timeline.rs+57 additions−28 deletions@@ -14,7 +14,7 @@ use matrix_sdk::{
 };
 use matrix_sdk_ui::timeline::{Timeline, TimelineEventItemId};
 use ruma::{OwnedEventId, OwnedUserId};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
 
 use super::executor::{EngineBoxFuture, JoinHandle, TaskSet, spawn};
 use crate::{
@@ -66,7 +66,6 @@ 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) current_user_id: Option<OwnedUserId>,
     pub(super) initial_history_limit: u16,
     pub(super) supervisor: Supervisor,
@@ -78,6 +77,7 @@ pub(super) struct TimelineSession {
     context: SessionContext,
     scope: MediaScope,
     watcher: JoinHandle<()>,
+    fences: mpsc::UnboundedSender<Fence>,
     pagination: mpsc::Sender<PaginationRequest>,
     pagination_worker: JoinHandle<()>,
     member_fetch: JoinHandle<()>,
@@ -96,7 +96,7 @@ impl TimelineSession {
     ) -> Result<Self, SendEventError> {
         let timeline = Arc::new(timeline);
         let scope = MediaScope::new();
-        let watcher = start_watcher(&id, &timeline, &context, &scope).await?;
+        let (watcher, fences) = start_watcher(&id, &timeline, &context, &scope).await?;
         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
@@ -109,6 +109,7 @@ impl TimelineSession {
             timeline,
             scope,
             watcher,
+            fences,
             pagination,
             pagination_worker,
             member_fetch,
@@ -125,23 +126,22 @@ impl TimelineSession {
     pub(super) async fn resync(&mut self) -> Result<(), Error> {
         self.watcher.abort();
         let _ = (&mut self.watcher).await;
-        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.watcher, self.fences) =
+            start_watcher(&self.id, &self.timeline, &self.context, &self.scope)
+                .await
+                .map_err(|_| Error::new(ErrorKind::Timeline, "timeline reset could not be sent"))?;
         Ok(())
     }
 
-    pub(super) async fn publish_history(&mut self) -> Result<(), Error> {
+    /// returns once the watcher has passed on the history the room cache committed
+    pub(super) async fn publish_history(&self) -> Result<(), Error> {
         self.timeline
             .flush_history()
             .await
             .map_err(timeline_error)?;
-        self.resync().await?;
-        self.context
-            .events
-            .publish_barrier()
-            .await
-            .map_err(|_| Error::interrupted())
+        let (fence, drained) = oneshot::channel();
+        self.fences.send(fence).map_err(|_| Error::interrupted())?;
+        drained.await.map_err(|_| Error::interrupted())
     }
 
     pub(super) fn paginate(&self, limit: u16, reply: Option<Responder>) {
@@ -404,15 +404,19 @@ impl ReplyFetcher {
     }
 }
 
+/// asks the watcher to pass on every diff the timeline already holds, then answers
+type Fence = oneshot::Sender<()>;
+
 async fn start_watcher(
     id: &TimelineId,
     timeline: &Arc<Timeline>,
     context: &SessionContext,
     scope: &MediaScope,
-) -> Result<JoinHandle<()>, SendEventError> {
+) -> Result<(JoinHandle<()>, mpsc::UnboundedSender<Fence>), SendEventError> {
     let (items, mut stream) = timeline.subscribe().await;
     let current_user_id = context.current_user_id.clone();
-    let normalised = normalise_timeline(items.iter(), current_user_id.as_deref(), scope);
+    let (normalised, mut hidden) =
+        normalise_timeline(items.iter(), current_user_id.as_deref(), scope);
     context.media.register(normalised.media).await;
     context
         .events
@@ -427,31 +431,56 @@ async fn start_watcher(
         fetches: TaskSet::new(),
     };
     replies.fetch(normalised.missing_replies);
+    let (fences, mut fence_requests) = mpsc::unbounded_channel::<Fence>();
     let id = id.clone();
     let events = context.events.clone();
     let media = context.media.clone();
     let scope = scope.clone();
-    Ok(context.supervisor.spawn("timeline watcher", async move {
+    let task = context.supervisor.spawn("timeline watcher", async move {
         loop {
-            let diffs = tokio::select! {
+            let (diffs, fence) = tokio::select! {
                 diffs = stream.next() => match diffs {
-                    Some(diffs) => diffs,
+                    Some(diffs) => (vec![diffs], None),
                     None => return,
                 },
+                Some(fence) = fence_requests.recv() => {
+                    let mut ready = Vec::new();
+                    while let Some(diffs) = stream.next().now_or_never() {
+                        match diffs {
+                            Some(diffs) => ready.push(diffs),
+                            None => return,
+                        }
+                    }
+                    (ready, Some(fence))
+                }
                 Some(_) = replies.fetches.join_next(), if !replies.fetches.is_empty() => continue,
             };
-            let normalised = normalise_timeline_diffs(diffs, current_user_id.as_deref(), &scope);
-            media.register(normalised.media).await;
-            replies.fetch(normalised.missing_replies);
-            let event = Event::TimelineChanged {
-                timeline_id: id.clone(),
-                diffs: normalised.diffs,
-            };
-            if events.send(event).await.is_err() {
-                return;
+            for diffs in diffs {
+                let normalised = normalise_timeline_diffs(
+                    diffs,
+                    current_user_id.as_deref(),
+                    &scope,
+                    &mut hidden,
+                );
+                media.register(normalised.media).await;
+                replies.fetch(normalised.missing_replies);
+                if normalised.diffs.is_empty() {
+                    continue;
+                }
+                let event = Event::TimelineChanged {
+                    timeline_id: id.clone(),
+                    diffs: normalised.diffs,
+                };
+                if events.send(event).await.is_err() {
+                    return;
+                }
+            }
+            if let Some(fence) = fence {
+                let _ = fence.send(());
             }
         }
-    }))
+    });
+    Ok((task, fences))
 }
 
 async fn run_pagination(

Modified file: src/lib.rs+2 additions−3 deletions@@ -74,9 +74,8 @@ pub use model::{
     VerificationAction, VerificationFailure, VerificationState, VerificationTarget,
 };
 pub use repair_model::{
-    HistoryContinuity, HistoryGap, HistoryGapLifecycle, HistoryRepairBudget, HistoryRepairCommand,
-    HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairReport, HistoryRepairStop,
-    HistoryRepairStrategy,
+    HistoryContinuity, HistoryGap, HistoryGapLifecycle, HistoryRepairCommand, HistoryRepairPhase,
+    HistoryRepairReport, HistoryRepairStop,
 };
 pub use ruma;
 #[doc(hidden)]

Modified file: src/mirror.rs+9 additions−14 deletions@@ -50,7 +50,7 @@ pub struct State {
     pub rooms: Vector<RoomSummary>,
     pub image_packs: ImagePacks,
     pub timelines: imbl::HashMap<TimelineId, TimelineState>,
-    #[cfg_attr(feature = "serde", serde(default))]
+    /// a room without an entry is idle. closing a room drops its entries here and in `repairs`
     pub recent_history: imbl::HashMap<OwnedRoomId, RecentHistoryState>,
     pub repairs: imbl::HashMap<OwnedRoomId, HistoryRepairReport>,
     pub typing: imbl::HashMap<OwnedRoomId, Vec<OwnedUserId>>,
@@ -375,7 +375,11 @@ impl Mirror {
                 room_id,
                 state: recent,
             } => self.publish(|state| {
-                state.recent_history.insert(room_id.clone(), recent.clone());
+                if recent == RecentHistoryState::Idle {
+                    state.recent_history.remove(&room_id);
+                } else {
+                    state.recent_history.insert(room_id.clone(), recent.clone());
+                }
                 Some(StateChange::RecentHistory {
                     room_id,
                     state: recent,
@@ -385,18 +389,6 @@ impl Mirror {
                 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::PublicationBarrier { acknowledgement } => {
                 if let Some(acknowledgement) = acknowledgement {
                     acknowledgement.complete(true);
@@ -406,6 +398,9 @@ impl Mirror {
                 self.cancel_resync(&timeline_id);
                 self.publish(|state| {
                     state.timelines.remove(&timeline_id)?;
+                    if let TimelineId::Room(room_id) = &timeline_id {
+                        state.repairs.remove(room_id);
+                    }
                     Some(StateChange::Timeline {
                         timeline_id,
                         change: TimelineChange::Closed,

Modified file: src/model.rs+3 additions−8 deletions@@ -672,8 +672,6 @@ pub enum CommandOutput {
     EventContextPage(EventContextPage),
     MentionsPage(crate::MentionsPage),
     MentionContext(crate::SavedMentionContext),
-    HistoryRepair(crate::HistoryRepairReport),
-    HistoryRepairPolicy(crate::HistoryRepairPolicy),
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -916,12 +914,6 @@ pub enum Event {
     },
     HistoryRepairChanged(crate::HistoryRepairReport),
     #[doc(hidden)]
-    HistoryRepairPublished {
-        report: crate::HistoryRepairReport,
-        #[cfg_attr(feature = "serde", serde(skip))]
-        acknowledgement: Option<crate::PublicationAck>,
-    },
-    #[doc(hidden)]
     PublicationBarrier {
         #[cfg_attr(feature = "serde", serde(skip))]
         acknowledgement: Option<crate::PublicationAck>,
@@ -2125,9 +2117,12 @@ pub enum HistoryState {
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum RecentHistoryState {
+    /// nothing running and nothing to show, which is also what closing the room leaves
     Idle,
     Loading,
     Loaded,
+    Cancelled,
+    /// stays until a retry or the room is reopened
     Failed(String),
 }
 

Modified file: src/normalise/timeline.rs+208 additions−53 deletions@@ -19,10 +19,9 @@ use ruma::{
 };
 
 use crate::{
-    Change, HistoryContinuity, HistoryGap, HistoryGapLifecycle, MediaId, MembershipChange, Message,
-    MessageKind, MessageMentions, Reaction, ReplyDetails, ReplyReference, SendState, Sender,
-    ThreadSummary, TimelineContent, TimelineDiff, TimelineEvent, TimelineItem, TimelineItemId,
-    UtdCause,
+    Change, HistoryGap, HistoryGapLifecycle, MediaId, MembershipChange, Message, MessageKind,
+    MessageMentions, Reaction, ReplyDetails, ReplyReference, SendState, Sender, ThreadSummary,
+    TimelineContent, TimelineDiff, TimelineEvent, TimelineItem, TimelineItemId, UtdCause,
 };
 
 use super::link_preview::sender_previews;
@@ -53,10 +52,16 @@ struct Normaliser<'a> {
 }
 
 impl Normaliser<'_> {
-    fn item(&mut self, item: &SdkTimelineItem) -> TimelineItem {
+    /// `None` for rows macltok leaves out
+    fn item(&mut self, item: &SdkTimelineItem) -> Option<TimelineItem> {
+        if let TimelineItemKind::Virtual(VirtualTimelineItem::HistoryGap(gap)) = item.kind()
+            && gap.status == matrix_sdk::event_cache::HistoryGapStatus::Joined
+        {
+            return None;
+        }
         self.media
             .extend(media_registrations_for_item(item, self.scope));
-        match item.kind() {
+        Some(match item.kind() {
             TimelineItemKind::Event(event) => {
                 TimelineItem::Event(Box::new(self.event(item, event)))
             }
@@ -70,41 +75,26 @@ impl Normaliser<'_> {
                 TimelineItem::TimelineStart
             }
             TimelineItemKind::Virtual(VirtualTimelineItem::HistoryGap(gap)) => {
-                let (lifecycle, continuity, failure) = match &gap.status {
-                    matrix_sdk::event_cache::HistoryGapStatus::Idle => (
-                        HistoryGapLifecycle::Idle,
-                        HistoryContinuity::Unresolved,
-                        None,
-                    ),
-                    matrix_sdk::event_cache::HistoryGapStatus::Paused => (
-                        HistoryGapLifecycle::Paused,
-                        HistoryContinuity::Unresolved,
-                        None,
-                    ),
-                    matrix_sdk::event_cache::HistoryGapStatus::Failed { message } => (
-                        HistoryGapLifecycle::Failed,
-                        HistoryContinuity::Unresolved,
-                        Some(message.clone()),
-                    ),
-                    matrix_sdk::event_cache::HistoryGapStatus::Unavailable => (
-                        HistoryGapLifecycle::Unavailable,
-                        HistoryContinuity::AccessibleStart,
-                        None,
-                    ),
-                    matrix_sdk::event_cache::HistoryGapStatus::Joined => {
-                        (HistoryGapLifecycle::Idle, HistoryContinuity::Verified, None)
+                use matrix_sdk::event_cache::HistoryGapStatus;
+                let (lifecycle, failure) = match &gap.status {
+                    HistoryGapStatus::Idle | HistoryGapStatus::Joined => {
+                        (HistoryGapLifecycle::Idle, None)
+                    }
+                    HistoryGapStatus::Paused => (HistoryGapLifecycle::Paused, None),
+                    HistoryGapStatus::Failed { message } => {
+                        (HistoryGapLifecycle::Failed, Some(message.clone()))
                     }
+                    HistoryGapStatus::Unavailable => (HistoryGapLifecycle::Unavailable, None),
                 };
                 TimelineItem::HistoryGap(HistoryGap {
                     id: gap.id.clone(),
                     older_boundary: gap.older_boundary.clone(),
                     newer_boundary: gap.newer_boundary.clone(),
                     lifecycle,
-                    continuity,
                     failure,
                 })
             }
-        }
+        })
     }
 
     fn event(&mut self, item: &SdkTimelineItem, event: &EventTimelineItem) -> TimelineEvent {
@@ -163,25 +153,28 @@ pub(crate) fn normalise_timeline<'a>(
     items: impl Iterator<Item = &'a Arc<SdkTimelineItem>>,
     current_user_id: Option<&UserId>,
     scope: &MediaScope,
-) -> NormalisedTimeline {
+) -> (NormalisedTimeline, HiddenRows) {
     let mut normaliser = Normaliser {
         current_user_id,
         scope,
         media: Vec::new(),
         missing_replies: Vec::new(),
     };
-    let items = items.map(|item| normaliser.item(item)).collect();
-    NormalisedTimeline {
-        items,
+    let mut hidden = HiddenRows::default();
+    let items = hidden.reset(items.map(|item| normaliser.item(item)));
+    let normalised = NormalisedTimeline {
+        items: items.into_iter().collect(),
         media: normaliser.media,
         missing_replies: normaliser.missing_replies,
-    }
+    };
+    (normalised, hidden)
 }
 
 pub(crate) fn normalise_timeline_diffs(
     diffs: Vec<VectorDiff<Arc<SdkTimelineItem>>>,
     current_user_id: Option<&UserId>,
     scope: &MediaScope,
+    hidden: &mut HiddenRows,
 ) -> NormalisedTimelineDiffs {
     let mut normaliser = Normaliser {
         current_user_id,
@@ -191,23 +184,7 @@ pub(crate) fn normalise_timeline_diffs(
     };
     let diffs = diffs
         .into_iter()
-        .map(|diff| match diff.map(|item| normaliser.item(&item)) {
-            VectorDiff::Append { values } => TimelineDiff::Append {
-                values: values.into_iter().collect(),
-            },
-            VectorDiff::Clear => TimelineDiff::Clear,
-            VectorDiff::PushFront { value } => TimelineDiff::PushFront { value },
-            VectorDiff::PushBack { value } => TimelineDiff::PushBack { value },
-            VectorDiff::PopFront => TimelineDiff::PopFront,
-            VectorDiff::PopBack => TimelineDiff::PopBack,
-            VectorDiff::Insert { index, value } => TimelineDiff::Insert { index, value },
-            VectorDiff::Set { index, value } => TimelineDiff::Set { index, value },
-            VectorDiff::Remove { index } => TimelineDiff::Remove { index },
-            VectorDiff::Truncate { length } => TimelineDiff::Truncate { length },
-            VectorDiff::Reset { values } => TimelineDiff::Reset {
-                values: values.into_iter().collect(),
-            },
-        })
+        .filter_map(|diff| hidden.translate(diff.map(|item| normaliser.item(&item))))
         .collect();
     NormalisedTimelineDiffs {
         diffs,
@@ -216,6 +193,135 @@ pub(crate) fn normalise_timeline_diffs(
     }
 }
 
+/// which sdk rows were left out, so diff indices can step over them
+#[derive(Default)]
+pub(crate) struct HiddenRows {
+    rows: imbl::Vector<bool>,
+    hidden: usize,
+}
+
+impl HiddenRows {
+    fn visible_index(&self, index: usize) -> usize {
+        if self.hidden == 0 {
+            return index;
+        }
+        self.rows
+            .iter()
+            .take(index)
+            .filter(|hidden| !**hidden)
+            .count()
+    }
+
+    fn push(&mut self, front: bool, item: &Option<TimelineItem>) {
+        self.hidden += usize::from(item.is_none());
+        if front {
+            self.rows.push_front(item.is_none());
+        } else {
+            self.rows.push_back(item.is_none());
+        }
+    }
+
+    fn forget(&mut self, hidden: bool) -> bool {
+        self.hidden -= usize::from(hidden);
+        !hidden
+    }
+
+    fn reset(&mut self, items: impl Iterator<Item = Option<TimelineItem>>) -> Vec<TimelineItem> {
+        *self = Self::default();
+        items
+            .filter_map(|item| {
+                self.push(false, &item);
+                item
+            })
+            .collect()
+    }
+
+    fn translate(&mut self, diff: VectorDiff<Option<TimelineItem>>) -> Option<TimelineDiff> {
+        Some(match diff {
+            VectorDiff::Append { values } => {
+                let values: Vec<_> = values
+                    .into_iter()
+                    .filter_map(|item| {
+                        self.push(false, &item);
+                        item
+                    })
+                    .collect();
+                if values.is_empty() {
+                    return None;
+                }
+                TimelineDiff::Append { values }
+            }
+            VectorDiff::Clear => {
+                *self = Self::default();
+                TimelineDiff::Clear
+            }
+            VectorDiff::PushFront { value } => {
+                self.push(true, &value);
+                TimelineDiff::PushFront { value: value? }
+            }
+            VectorDiff::PushBack { value } => {
+                self.push(false, &value);
+                TimelineDiff::PushBack { value: value? }
+            }
+            VectorDiff::PopFront => {
+                let hidden = self.rows.pop_front()?;
+                self.forget(hidden).then_some(TimelineDiff::PopFront)?
+            }
+            VectorDiff::PopBack => {
+                let hidden = self.rows.pop_back()?;
+                self.forget(hidden).then_some(TimelineDiff::PopBack)?
+            }
+            VectorDiff::Insert { index, value } => {
+                let visible = self.visible_index(index);
+                self.hidden += usize::from(value.is_none());
+                self.rows.insert(index, value.is_none());
+                TimelineDiff::Insert {
+                    index: visible,
+                    value: value?,
+                }
+            }
+            VectorDiff::Set { index, value } => {
+                let visible = self.visible_index(index);
+                let was_hidden = self.rows.set(index, value.is_none());
+                self.hidden = self.hidden + usize::from(value.is_none()) - usize::from(was_hidden);
+                match (was_hidden, value) {
+                    (false, Some(value)) => TimelineDiff::Set {
+                        index: visible,
+                        value,
+                    },
+                    (false, None) => TimelineDiff::Remove { index: visible },
+                    (true, Some(value)) => TimelineDiff::Insert {
+                        index: visible,
+                        value,
+                    },
+                    (true, None) => return None,
+                }
+            }
+            VectorDiff::Remove { index } => {
+                let visible = self.visible_index(index);
+                let hidden = self.rows.remove(index);
+                self.forget(hidden)
+                    .then_some(TimelineDiff::Remove { index: visible })?
+            }
+            VectorDiff::Truncate { length } => {
+                let visible = self.visible_index(length);
+                let dropped = self
+                    .rows
+                    .iter()
+                    .skip(length)
+                    .filter(|hidden| **hidden)
+                    .count();
+                self.hidden -= dropped;
+                self.rows.truncate(length);
+                TimelineDiff::Truncate { length: visible }
+            }
+            VectorDiff::Reset { values } => TimelineDiff::Reset {
+                values: self.reset(values.into_iter()),
+            },
+        })
+    }
+}
+
 /// `media_identity` is `None` for embedded reply content, which never gets fetchable media of its own
 fn content(content: &TimelineItemContent, media_identity: Option<&str>) -> TimelineContent {
     match content {
@@ -485,3 +591,52 @@ pub(super) fn formatted_body(message: &MessageType) -> Option<String> {
     }?;
     (formatted.format == MessageFormat::Html).then(|| formatted.body.clone())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn row(n: u64) -> Option<TimelineItem> {
+        Some(TimelineItem::DayDivider { timestamp_ms: n })
+    }
+
+    #[test]
+    fn hidden_rows_keep_diff_indices_on_the_visible_timeline() {
+        let mut sdk: imbl::Vector<Option<TimelineItem>> = imbl::Vector::new();
+        let mut hidden = HiddenRows::default();
+        let mut visible: imbl::Vector<TimelineItem> = imbl::Vector::new();
+        let diffs = [
+            VectorDiff::Reset {
+                values: [row(0), None, row(1)].into_iter().collect(),
+            },
+            VectorDiff::Insert {
+                index: 1,
+                value: row(2),
+            },
+            VectorDiff::Set {
+                index: 2,
+                value: row(3),
+            },
+            VectorDiff::Set {
+                index: 0,
+                value: None,
+            },
+            VectorDiff::PushFront { value: None },
+            VectorDiff::Remove { index: 3 },
+            VectorDiff::Append {
+                values: [None, row(4)].into_iter().collect(),
+            },
+            VectorDiff::PopFront,
+            VectorDiff::Truncate { length: 2 },
+            VectorDiff::PopBack,
+        ];
+        for diff in diffs {
+            diff.clone().apply(&mut sdk);
+            if let Some(diff) = hidden.translate(diff) {
+                diff.try_apply(&mut visible).unwrap();
+            }
+            let expected: imbl::Vector<_> = sdk.iter().flatten().cloned().collect();
+            assert_eq!(visible, expected);
+        }
+    }
+}

Modified file: src/repair_model.rs+9 additions−109 deletions@@ -5,62 +5,19 @@ use serde::{Deserialize, Serialize};
 
 use crate::Error;
 
-#[derive(Clone, Debug, PartialEq, Eq)]
-#[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: 4,
-            max_events: 200,
-            max_duration_ms: 15_000,
-        }
-    }
-}
-
 #[derive(Clone, Debug, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
 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,
-    },
+    /// runs the open-time recent load again, joining one already running. the outcome
+    /// arrives as `RecentHistoryChanged`
+    RetryRecent { room_id: OwnedRoomId },
     FillGap {
         room_id: OwnedRoomId,
         gap_id: OwnedTransactionId,
     },
-    GetPolicy,
-    SetPolicy(HistoryRepairPolicy),
+    /// stops whichever history operation the room is running
+    Cancel { room_id: OwnedRoomId },
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -70,7 +27,6 @@ pub struct HistoryGap {
     pub older_boundary: Option<OwnedEventId>,
     pub newer_boundary: Option<OwnedEventId>,
     pub lifecycle: HistoryGapLifecycle,
-    pub continuity: HistoryContinuity,
     pub failure: Option<String>,
 }
 
@@ -78,22 +34,14 @@ pub struct HistoryGap {
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum HistoryGapLifecycle {
     Idle,
-    Fetching,
     Paused,
     Failed,
+    /// accessible history ends before the older section, so this interval stays
     Unavailable,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
-pub enum HistoryRepairStrategy {
-    Recent,
-    Overlap,
-    Bridge,
-}
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum HistoryContinuity {
     Unresolved,
     Verified,
@@ -103,7 +51,6 @@ pub enum HistoryContinuity {
 #[derive(Clone, Debug, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum HistoryRepairPhase {
-    FetchingRecent,
     Bridging,
     Ready,
     Paused,
@@ -115,66 +62,19 @@ pub enum HistoryRepairStop {
     Cancelled,
     Budget,
     HistoryUnavailable,
-    NoAnchor,
     CursorStalled,
 }
 
+/// one gap fill. `Bridging` when it starts, then once more when it settles
 #[derive(Clone, Debug, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub struct HistoryRepairReport {
     pub room_id: OwnedRoomId,
-    #[cfg_attr(feature = "serde", serde(default))]
-    pub gap_id: Option<OwnedTransactionId>,
-    pub requested_policy: HistoryRepairPolicy,
-    pub strategy: HistoryRepairStrategy,
-    pub coverage_duration_ms: Option<u64>,
-    pub threshold_evaluated: bool,
+    pub gap_id: OwnedTransactionId,
     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 {
-            gap_id: None,
-            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+20 additions−135 deletions@@ -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, 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,
+    EventContextPage, EventRef, HistoryRepairCommand, 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},
@@ -332,42 +332,6 @@ 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,
@@ -387,38 +351,10 @@ impl ClientHandle {
         .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,
-        )))
+    pub async fn retry_recent_history(&self, room_id: OwnedRoomId) -> Result<(), Error> {
+        self.run(Command::HistoryRepair(HistoryRepairCommand::RetryRecent {
+            room_id,
+        }))
         .await
     }
 
@@ -1689,30 +1625,6 @@ impl EventSender {
         }
     }
 
-    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.events.reserve().await.map_err(|_| SendEventError)
     }
@@ -1874,31 +1786,9 @@ fn unexpected(wanted: &str, output: &CommandOutput) -> Error {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{
-        HistoryContinuity, HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairStrategy,
-    };
-    use ruma::room_id;
-
-    fn report() -> HistoryRepairReport {
-        HistoryRepairReport {
-            gap_id: None,
-            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() {
+    async fn publish_barrier_waits_for_mirror_publication() {
         let publication = Arc::new(PublicationRegistry::default());
         let _registration = publication.register_mirror();
         let (events, mut received) = mpsc::channel(1);
@@ -1906,18 +1796,14 @@ mod tests {
             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 publisher = tokio::spawn(async move { sender.publish_barrier().await });
 
-        let Some(Event::HistoryRepairPublished {
-            report,
+        let Some(Event::PublicationBarrier {
             acknowledgement: Some(acknowledgement),
         }) = received.recv().await
         else {
-            panic!("repair event was not published");
+            panic!("barrier was not published");
         };
-        assert_eq!(report, expected);
         tokio::task::yield_now().await;
         assert!(!publisher.is_finished());
 
@@ -1926,7 +1812,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn publish_repair_fails_when_last_mirror_drops() {
+    async fn publish_barrier_fails_when_last_mirror_drops() {
         let publication = Arc::new(PublicationRegistry::default());
         let registration = publication.register_mirror();
         let (events, mut received) = mpsc::channel(1);
@@ -1934,13 +1820,12 @@ mod tests {
             events,
             publication,
         };
-        let publisher = tokio::spawn(async move { sender.publish_repair(report()).await });
+        let publisher = tokio::spawn(async move { sender.publish_barrier().await });
 
         assert!(matches!(
             received.recv().await,
-            Some(Event::HistoryRepairPublished {
+            Some(Event::PublicationBarrier {
                 acknowledgement: Some(_),
-                ..
             })
         ));
         drop(registration);

Modified file: tests/engine.rs+2 additions−12 deletions@@ -4,8 +4,7 @@ use std::time::Duration;
 
 use macltok::{
     ClientHandle, Command, Engine, EngineConfig, ErrorKind, Event, EventStream, HistoryState,
-    MediaSize, MessageContent, RecentHistoryState, RoomCommand, TimelineCommand, TimelineId,
-    TimelineItem,
+    MediaSize, MessageContent, RoomCommand, TimelineCommand, TimelineId, TimelineItem,
 };
 use matrix_sdk::{
     ruma::{api::client::sync::sync_events::v5, event_id, mxc_uri, room_id},
@@ -318,7 +317,6 @@ async fn closes_and_reopens_a_room_in_command_order() {
 
     // biased select drains buffered events before it reports ready replies
     let mut resets = 0;
-    let mut recent_active = false;
     let mut order = Vec::new();
     let (mut close_done, mut replacement_open_done, mut second_close_done, mut final_open_done) =
         (false, false, false, false);
@@ -328,15 +326,7 @@ async fn closes_and_reopens_a_room_in_command_order() {
                 biased;
                 event = events.recv() => {
                     match event.expect("event stream closed") {
-                        Event::RecentHistoryChanged { room_id: changed, state } if changed == room_id => {
-                            recent_active = matches!(state, RecentHistoryState::Loading);
-                        }
-                        Event::TimelineClosed { timeline_id } if timeline_id.room_id() == room_id => {
-                            recent_active = false;
-                        }
-                        Event::TimelineReset { timeline_id, .. }
-                            if timeline_id.room_id() == room_id && !recent_active =>
-                        {
+                        Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id => {
                             resets += 1;
                             assert!(resets <= 2, "a room generation reset more than once");
                         }

Modified file: tests/history_repair.rs+120 additions−75 deletions@@ -2,19 +2,20 @@ mod support;
 
 use std::time::Duration;
 
+#[cfg(feature = "mirror")]
+use macltok::Mirror;
 use macltok::{
     ClientHandle, Engine, EngineConfig, EngineError, Event, EventStream, HistoryGap,
-    HistoryRepairPhase, HistoryRepairPolicy, RecentHistoryState, Termination, TimelineItem,
+    HistoryRepairPhase, HistoryRepairReport, HistoryRepairStop, RecentHistoryState, Termination,
+    TimelineId, 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 support::shut_down;
 use wiremock::ResponseTemplate;
 
 struct TestRoom {
@@ -23,6 +24,9 @@ struct TestRoom {
     handle: ClientHandle,
     events: EventStream,
     running: tokio::task::JoinHandle<Result<Termination, EngineError>>,
+    /// the open room's timeline, kept current by every helper that reads events
+    items: imbl::Vector<TimelineItem>,
+    opened: bool,
 }
 
 async fn start_room(room_id: &RoomId) -> TestRoom {
@@ -45,12 +49,62 @@ async fn start_room(room_id: &RoomId) -> TestRoom {
         handle,
         events,
         running,
+        items: imbl::Vector::new(),
+        opened: false,
     }
 }
 
+/// the next event that satisfies `predicate`. only the open itself may reset the room
+/// timeline, history after it has to arrive as diffs
+async fn next_until(room: &mut TestRoom, mut predicate: impl FnMut(&Event) -> bool) -> Event {
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let event = room.events.recv().await.expect("event stream closed");
+            match &event {
+                Event::TimelineReset {
+                    timeline_id: TimelineId::Room(_),
+                    items,
+                } => {
+                    assert!(!room.opened, "history reset the room timeline");
+                    room.opened = true;
+                    room.items = items.clone();
+                }
+                Event::TimelineChanged {
+                    timeline_id: TimelineId::Room(_),
+                    diffs,
+                } => {
+                    for diff in diffs.clone() {
+                        diff.try_apply(&mut room.items)
+                            .expect("diff did not fit the timeline");
+                    }
+                }
+                _ => {}
+            }
+            if predicate(&event) {
+                return event;
+            }
+        }
+    })
+    .await
+    .expect("expected event did not arrive")
+}
+
+async fn timeline_until(
+    room: &mut TestRoom,
+    predicate: impl Fn(&imbl::Vector<TimelineItem>) -> bool,
+) {
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while !predicate(&room.items) {
+            next_until(room, |_| true).await;
+        }
+    })
+    .await
+    .expect("the timeline never got there");
+}
+
 async fn open_room(room: &mut TestRoom, room_id: &RoomId) {
     room.handle.open_room(room_id.to_owned()).await.unwrap();
-    receive_until(&mut room.events, |event| {
+    next_until(room, |event| {
         matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
     })
     .await;
@@ -84,7 +138,7 @@ fn history_event(
 }
 
 async fn wait_for_recent(room: &mut TestRoom, room_id: &RoomId) -> RecentHistoryState {
-    let event = receive_until(&mut room.events, |event| {
+    let event = next_until(room, |event| {
         matches!(event, Event::RecentHistoryChanged { room_id: id, state, .. }
             if id == room_id && !matches!(state, RecentHistoryState::Loading))
     })
@@ -95,11 +149,11 @@ async fn wait_for_recent(room: &mut TestRoom, room_id: &RoomId) -> RecentHistory
     }
 }
 
-async fn wait_for_report(room: &mut TestRoom, room_id: &RoomId) -> macltok::HistoryRepairReport {
-    match receive_until(
-        &mut room.events,
-        |event| matches!(event, Event::HistoryRepairChanged(report) if &report.room_id == room_id),
-    )
+async fn wait_for_settled_fill(room: &mut TestRoom) -> HistoryRepairReport {
+    match next_until(room, |event| {
+        matches!(event, Event::HistoryRepairChanged(report)
+            if report.phase != HistoryRepairPhase::Bridging)
+    })
     .await
     {
         Event::HistoryRepairChanged(report) => report,
@@ -107,6 +161,12 @@ async fn wait_for_report(room: &mut TestRoom, room_id: &RoomId) -> macltok::Hist
     }
 }
 
+fn has_event(items: &imbl::Vector<TimelineItem>, id: &EventId) -> bool {
+    items.iter().any(
+        |item| matches!(item, TimelineItem::Event(event) if event.event_id.as_deref() == Some(id)),
+    )
+}
+
 fn gap_in(items: &imbl::Vector<TimelineItem>) -> Option<HistoryGap> {
     items.iter().find_map(|item| match item {
         TimelineItem::HistoryGap(gap) => Some(gap.clone()),
@@ -125,10 +185,6 @@ async fn automatic_recent_load_is_bounded_and_keeps_cached_history_behind_a_gap(
         event_id!("$fresh-d:localhost"),
     ];
     let mut room = start_room(room_id).await;
-    room.handle
-        .set_history_repair_policy(HistoryRepairPolicy::Bridge)
-        .await
-        .unwrap();
     for (index, id) in new_ids.iter().enumerate() {
         let token = format!("recent-{index}");
         let response = RoomMessagesResponseTemplate::default()
@@ -154,40 +210,16 @@ async fn automatic_recent_load_is_bounded_and_keeps_cached_history_behind_a_gap(
     }
     cache_event_before_open(&room, room_id, old_id, "retained old event").await;
     open_room(&mut room, room_id).await;
-    let reset = receive_until(&mut room.events, |event| {
-        let items = match event {
-            Event::TimelineReset { timeline_id, items } if timeline_id.room_id() == room_id => {
-                items
-            }
-            _ => return false,
-        };
+    timeline_until(&mut room, |items| {
         gap_in(items).is_some()
-            && items.iter().any(|item| {
-                matches!(item, TimelineItem::Event(event)
-                if event.event_id.as_deref() == Some(old_id))
-            })
-            && new_ids.iter().all(|id| {
-                items.iter().any(|item| {
-                    matches!(item,
-                TimelineItem::Event(event) if event.event_id.as_deref() == Some(id))
-                })
-            })
+            && has_event(items, old_id)
+            && new_ids.iter().all(|id| has_event(items, id))
     })
     .await;
-    let Event::TimelineReset { items, .. } = reset else {
-        unreachable!();
-    };
-    assert!(gap_in(&items).is_some());
     assert_eq!(
         wait_for_recent(&mut room, room_id).await,
         RecentHistoryState::Loaded
     );
-    let report = wait_for_report(&mut room, room_id).await;
-    assert_eq!(report.requested_policy, HistoryRepairPolicy::StartRecent);
-    assert_eq!(report.phase, HistoryRepairPhase::Ready);
-    assert_eq!(report.pages, 4);
-    assert_eq!(report.events, 4);
-    assert!(report.recent_available);
     let requests = room.server.received_requests().await.unwrap();
     let messages = requests
         .iter()
@@ -230,13 +262,10 @@ async fn explicit_recent_start_coalesces_with_the_automatic_load() {
     .await
     .expect("automatic recent request did not start");
 
-    let report = room
-        .handle
-        .start_recent_history(room_id.to_owned())
+    room.handle
+        .retry_recent_history(room_id.to_owned())
         .await
         .unwrap();
-    assert_eq!(report.pages, 1);
-    assert_eq!(report.events, 1);
     assert_eq!(
         wait_for_recent(&mut room, room_id).await,
         RecentHistoryState::Loaded
@@ -358,8 +387,7 @@ async fn fill_gap_returns_after_the_mirror_contains_the_committed_rows() {
         loop {
             let snapshot = mirror_handle.snapshot();
             let active_gap = snapshot.repairs.get(room_id).is_some_and(|report| {
-                report.gap_id.as_ref() == Some(&target_gap_id)
-                    && report.phase == HistoryRepairPhase::Bridging
+                report.gap_id == target_gap_id && report.phase == HistoryRepairPhase::Bridging
             });
             if active_gap {
                 break;
@@ -416,15 +444,8 @@ async fn cancelling_a_gap_fill_keeps_committed_progress_for_retry() {
         .await;
     cache_event_before_open(&room, room_id, old_id, "old cached").await;
     open_room(&mut room, room_id).await;
-    let reset = receive_until(&mut room.events, |event| {
-        matches!(event, Event::TimelineReset { timeline_id, items }
-            if timeline_id.room_id() == room_id && gap_in(items).is_some())
-    })
-    .await;
-    let gap = match reset {
-        Event::TimelineReset { items, .. } => gap_in(&items).unwrap(),
-        _ => unreachable!(),
-    };
+    timeline_until(&mut room, |items| gap_in(items).is_some()).await;
+    let gap = gap_in(&room.items).unwrap();
     assert_eq!(
         wait_for_recent(&mut room, room_id).await,
         RecentHistoryState::Loaded
@@ -487,6 +508,9 @@ async fn cancelling_a_gap_fill_keeps_committed_progress_for_retry() {
         .unwrap();
     let error = fill.await.unwrap().unwrap_err();
     assert_eq!(error.kind, macltok::ErrorKind::Interrupted);
+    let report = wait_for_settled_fill(&mut room).await;
+    assert_eq!(report.stop, Some(HistoryRepairStop::Cancelled));
+    assert!(report.failure.is_none());
 
     room.server
         .mock_room_messages()
@@ -505,22 +529,13 @@ async fn cancelling_a_gap_fill_keeps_committed_progress_for_retry() {
         .fill_history_gap(room_id.to_owned(), gap.id)
         .await
         .unwrap();
-    let final_reset = receive_until(&mut room.events, |event| {
-        matches!(event, Event::TimelineReset { timeline_id, items }
-            if timeline_id.room_id() == room_id
-                && items.iter().any(|item| matches!(item, TimelineItem::Event(event)
-                    if event.event_id.as_deref() == Some(middle_id)))
-                && items.iter().any(|item| matches!(item, TimelineItem::Event(event)
-                    if event.event_id.as_deref() == Some(old_id))))
+    timeline_until(&mut room, |items| {
+        [fresh_id, middle_id, old_id]
+            .iter()
+            .all(|id| has_event(items, id))
+            && gap_in(items).is_none()
     })
     .await;
-    let ids = event_messages(&final_reset)
-        .iter()
-        .filter_map(|event| event.event_id.as_deref())
-        .collect::<Vec<_>>();
-    assert!(ids.contains(&fresh_id));
-    assert!(ids.contains(&middle_id));
-    assert!(ids.contains(&old_id));
     shut_down(&room.handle, &mut room.events, room.running).await;
 }
 
@@ -542,8 +557,38 @@ async fn failed_automatic_recent_load_is_public_and_keeps_the_cached_timeline() 
     open_room(&mut room, room_id).await;
     let state = wait_for_recent(&mut room, room_id).await;
     assert!(matches!(state, RecentHistoryState::Failed(_)));
-    let report = wait_for_report(&mut room, room_id).await;
-    assert_eq!(report.phase, HistoryRepairPhase::Paused);
-    assert!(report.failure.is_some());
+    shut_down(&room.handle, &mut room.events, room.running).await;
+}
+
+#[tokio::test]
+async fn closing_during_a_recent_load_settles_idle_not_failed() {
+    let room_id = room_id!("!close-recent:localhost");
+    let mut room = start_room(room_id).await;
+    room.server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .with_delay(Duration::from_secs(5))
+            .events(vec![history_event(
+                room_id,
+                event_id!("$never:localhost"),
+                "too late",
+            )]))
+        .mock_once()
+        .mount()
+        .await;
+    open_room(&mut room, room_id).await;
+    room.handle.close_room(room_id.to_owned()).await.unwrap();
+    let mut states = Vec::new();
+    next_until(&mut room, |event| {
+        if let Event::RecentHistoryChanged { state, .. } = event {
+            states.push(state.clone());
+        }
+        matches!(event, Event::TimelineClosed { timeline_id } if timeline_id.room_id() == room_id)
+    })
+    .await;
+    assert_eq!(
+        states,
+        [RecentHistoryState::Loading, RecentHistoryState::Idle]
+    );
     shut_down(&room.handle, &mut room.events, room.running).await;
 }

macltok

Commit d9b96952ec60

Commit `d9b96952ec60`