macltok

public

Star0

Fork 0

Watch0

58 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `c9f0fd0ee1cc`

tarball

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

16 files changed+1096-1069

timeline: retain segmented room history

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

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

Added file: docs/segmented-history.md+164 additions@@ -0,0 +1,164 @@
+# Segmented history implementation
+
+The durable implementation is in the macltok `segmented-history` feature
+worktree. Its SDK dependency has been published at
+`49a9cd1aff19df1de7e46710cb6c984a1bddc30e` on the SDK `segmented-history`
+bookmark. The macltok manifest and lockfile are being updated to that revision.
+The macltok feature bookmark will also be named `segmented-history`. Malice has
+not yet been pinned to the macltok revision.
+
+## 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
+
+`HistoryRepairCommand::FillGap { room_id, gap_id }` is the supported targeted
+fill operation, sent through `ClientHandle::execute` as
+`Command::HistoryRepair`. It returns `CommandOutput::Done` on success. 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.
+
+Targeted fill progress is published as a `HistoryRepairReport` with `gap_id`
+set and `phase` set to `Bridging`. This report supplies the loading state while
+the fill runs. 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.
+
+```json
+{
+  "FillGap": {
+    "room_id": "!room:example.org",
+    "gap_id": "opaque-gap-transaction-id"
+  }
+}
+```
+
+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.
+
+## Remaining publication and Malice work
+
+Publish the SDK revision on the user's fork, then publish the macltok revision
+that pins it. Update the Malice macltok dependency in both native and worker
+manifests and regenerate the applicable lockfiles against that same revision.
+The current worktree has not recorded a final publishable commit or revision
+for these changes.
+
+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. `nix flake check` is the aggregate gate and remains
+pending.
+
+Seven focused SDK recent-history tests and seven segmented-history tests pass.
+The SDK WebAssembly compile passes with IndexedDB enabled. The SDK UI timeline
+checks pass with 123 unit tests, seven room 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 flake gate remains
+pending. Complete worker restart and encrypted room recovery have not been
+tested.

Modified file: harness/web/Cargo.lock+8 additions−8 deletions@@ -1907,7 +1907,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=49a9cd1aff19df1de7e46710cb6c984a1bddc30e#49a9cd1aff19df1de7e46710cb6c984a1bddc30e"
 dependencies = [
  "anymap2",
  "aquamarine",
@@ -1967,7 +1967,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-base"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=49a9cd1aff19df1de7e46710cb6c984a1bddc30e#49a9cd1aff19df1de7e46710cb6c984a1bddc30e"
 dependencies = [
  "as_variant",
  "async-trait",
@@ -1993,7 +1993,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-common"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=49a9cd1aff19df1de7e46710cb6c984a1bddc30e#49a9cd1aff19df1de7e46710cb6c984a1bddc30e"
 dependencies = [
  "eyeball-im",
  "futures-core",
@@ -2016,7 +2016,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-crypto"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=49a9cd1aff19df1de7e46710cb6c984a1bddc30e#49a9cd1aff19df1de7e46710cb6c984a1bddc30e"
 dependencies = [
  "aes",
  "aquamarine",
@@ -2056,7 +2056,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-indexeddb"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=49a9cd1aff19df1de7e46710cb6c984a1bddc30e#49a9cd1aff19df1de7e46710cb6c984a1bddc30e"
 dependencies = [
  "async-trait",
  "base64",
@@ -2088,7 +2088,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-sqlite"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=49a9cd1aff19df1de7e46710cb6c984a1bddc30e#49a9cd1aff19df1de7e46710cb6c984a1bddc30e"
 dependencies = [
  "as_variant",
  "async-trait",
@@ -2115,7 +2115,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-store-encryption"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=49a9cd1aff19df1de7e46710cb6c984a1bddc30e#49a9cd1aff19df1de7e46710cb6c984a1bddc30e"
 dependencies = [
  "base64",
  "blake3",
@@ -2136,7 +2136,7 @@ dependencies = [
 [[package]]
 name = "matrix-sdk-ui"
 version = "0.18.0"
-source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=7aace883fbe03d4c3c118b55188fe9519ba8ce48#7aace883fbe03d4c3c118b55188fe9519ba8ce48"
+source = "git+ssh://brewery@pls.lobotomise.me:33/atagen/matrix-rust-sdk?rev=49a9cd1aff19df1de7e46710cb6c984a1bddc30e#49a9cd1aff19df1de7e46710cb6c984a1bddc30e"
 dependencies = [
  "as_variant",
  "async-rx",

Modified file: src/engine/mod.rs+2 additions−12 deletions@@ -326,8 +326,6 @@ pub struct Engine {
     mentions: crate::mentions::service::Handle,
     mentions_service: Option<crate::mentions::service::Service>,
     mentions_updates: Option<watch::Receiver<crate::MentionsPage>>,
-    repair_gate: Arc<tokio::sync::Mutex<()>>,
-    repair_shutdown: Arc<std::sync::atomic::AtomicBool>,
 }
 
 impl Engine {
@@ -437,8 +435,6 @@ impl Engine {
             mentions,
             mentions_service: Some(mentions_service),
             mentions_updates: Some(mentions_updates),
-            repair_gate: Arc::default(),
-            repair_shutdown: Arc::default(),
         };
         Ok((engine, handle, events))
     }
@@ -988,7 +984,8 @@ impl Engine {
             HistoryRepairCommand::Start { room_id, .. }
             | HistoryRepairCommand::Continue { room_id, .. }
             | HistoryRepairCommand::Cancel { room_id }
-            | HistoryRepairCommand::StartRecent { room_id } => room_id,
+            | HistoryRepairCommand::StartRecent { room_id }
+            | HistoryRepairCommand::FillGap { room_id, .. } => room_id,
             HistoryRepairCommand::GetPolicy | HistoryRepairCommand::SetPolicy(_) => {
                 let mentions = self.mentions.clone();
                 self.tasks
@@ -1084,9 +1081,6 @@ impl Engine {
                 initial_history_limit: self.config.initial_history_limit,
                 supervisor,
                 mentions: self.mentions.clone(),
-                sync: self.sync.clone(),
-                repair_gate: self.repair_gate.clone(),
-                repair_shutdown: self.repair_shutdown.clone(),
             },
             panics,
             opening: vec![reply],
@@ -1165,15 +1159,11 @@ impl Engine {
     }
 
     async fn shutdown(&mut self) {
-        self.repair_shutdown
-            .store(true, std::sync::atomic::Ordering::Release);
         if let Some(handler) = self.verification_handler.take() {
             self.client.remove_event_handler(handler);
         }
         self.port.close();
-        let repair_gate = self.repair_gate.lock().await;
         self.sync.stop().await;
-        drop(repair_gate);
         if self.mentions_service.is_none() {
             self.mentions.stop().await;
         }

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

Modified file: src/engine/room.rs+53 additions−92 deletions@@ -6,11 +6,11 @@ use ruma::OwnedRoomId;
 use tokio::sync::mpsc;
 
 use super::executor::{AbortHandle, JoinError, TaskSet, is_panic, resume_unwind};
-use super::repair::{InstallRequest, RepairController};
+use super::repair::{Completion, RepairController};
 use super::timeline::{SessionContext, TaskGuard, TimelineSession};
 use crate::{
-    Error, ErrorKind, Event, HistoryRepairCommand, TimelineCommand, TimelineId, TypingState,
-    transport::Responder,
+    Error, ErrorKind, Event, HistoryRepairCommand, RecentHistoryState, TimelineCommand, TimelineId,
+    TypingState, transport::Responder,
 };
 
 pub(super) enum ActorRequest {
@@ -93,9 +93,6 @@ impl RoomActor {
         else {
             return;
         };
-        for reply in self.opening.drain(..) {
-            reply.done(Ok(()));
-        }
         let (typing_guard, typing) = self.room.subscribe_to_typing_notifications();
         let mut typing_watcher = TaskGuard(self.context.supervisor.spawn(
             "typing watcher",
@@ -107,8 +104,14 @@ impl RoomActor {
             work: TaskSet::new(),
         };
         let mut repair = RepairController::new(self.room.clone(), self.context.clone());
+        repair.start_recent().await;
+        for reply in self.opening.drain(..) {
+            reply.done(Ok(()));
+        }
         let close = self.serve_open(&mut timelines, &mut repair).await;
-        repair.shutdown().await;
+        if let Some(completion) = repair.shutdown().await {
+            self.finish_repair(completion, &mut timelines).await;
+        }
         if close.is_none() {
             self.requests.close();
             let _ = self.notes.send(ActorNote::Leaving {
@@ -137,6 +140,46 @@ 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 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()
+        {
+            resolved = Err(Error::interrupted());
+            cancellation = Err(Error::interrupted());
+        }
+        completion.resolve(resolved, cancellation);
+    }
+
     /// returns the close request's responder, if a close is why we stopped
     async fn serve_open(
         &mut self,
@@ -167,9 +210,9 @@ impl RoomActor {
                         repair.execute(command, reply).await;
                     }
                 },
-                install = repair.next_install(), if repair.is_running() => {
-                    if let Some(install) = install {
-                        timelines.install_history(install, &self.room, &self.context).await;
+                completion = repair.next_completion(), if repair.is_running() => {
+                    if let Some(completion) = completion {
+                        self.finish_repair(completion, timelines).await;
                     }
                 }
                 Some(built) = timelines.threads.builds.join_next(), if !timelines.threads.builds.is_empty() => {
@@ -240,88 +283,6 @@ struct Timelines {
 }
 
 impl Timelines {
-    async fn install_history(
-        &mut self,
-        install: InstallRequest,
-        room: &Room,
-        context: &SessionContext,
-    ) {
-        let InstallRequest {
-            events,
-            prev_batch,
-            reply,
-        } = install;
-        let result = self
-            .install_history_inner(events, prev_batch, room, context)
-            .await;
-        let _ = reply.send(result);
-    }
-
-    async fn install_history_inner(
-        &mut self,
-        events: Vec<matrix_sdk::deserialized_responses::TimelineEvent>,
-        prev_batch: Option<String>,
-        room: &Room,
-        context: &SessionContext,
-    ) -> Result<(), Error> {
-        let threads_capacity = self.threads.capacity;
-        self.threads.close().await;
-        self.threads = Threads::new(threads_capacity);
-        self.room.pause_for_reseed().await;
-
-        let reseed = async {
-            let (cache, _drop_handles) = room.event_cache().await.map_err(|source| {
-                Error::with_source(ErrorKind::Timeline, "history event cache failed", source)
-            })?;
-            cache.reseed(events, prev_batch).await.map_err(|source| {
-                Error::with_source(ErrorKind::Timeline, "history cache reseed failed", source)
-            })
-        }
-        .await;
-        if let Err(error) = reseed {
-            return match self.room.resume_after_failed_reseed().await {
-                Ok(()) => Err(error),
-                Err(restart_error) => Err(restart_error),
-            };
-        }
-
-        let timeline = match build_timeline(
-            room,
-            TimelineFocus::Live {
-                hide_threaded_events: true,
-            },
-        )
-        .await
-        {
-            Ok(timeline) => timeline,
-            Err(source) => {
-                let error =
-                    Error::with_source(ErrorKind::Timeline, "room timeline rebuild failed", source);
-                return match self.room.resume_after_failed_reseed().await {
-                    Ok(()) => Err(error),
-                    Err(restart_error) => Err(restart_error),
-                };
-            }
-        };
-        let timeline_id = self.room.id.clone();
-        let mut context = context.clone();
-        context.initial_history_limit = 0;
-        let timeline = match TimelineSession::start(timeline_id, timeline, context).await {
-            Ok(timeline) => timeline,
-            Err(source) => {
-                let error =
-                    Error::with_source(ErrorKind::Timeline, "room timeline reset failed", source);
-                return match self.room.resume_after_failed_reseed().await {
-                    Ok(()) => Err(error),
-                    Err(restart_error) => Err(restart_error),
-                };
-            }
-        };
-        let previous = std::mem::replace(&mut self.room, timeline);
-        drop(previous);
-        Ok(())
-    }
-
     async fn execute(&mut self, command: TimelineCommand, reply: Responder, room: &Room) {
         match command {
             TimelineCommand::OpenThread {

Modified file: src/engine/timeline.rs+15 additions−23 deletions@@ -12,10 +12,9 @@ use matrix_sdk::{
         },
     },
 };
-use matrix_sdk_ui::sync_service::SyncService;
 use matrix_sdk_ui::timeline::{Timeline, TimelineEventItemId};
 use ruma::{OwnedEventId, OwnedUserId};
-use tokio::sync::{Mutex, mpsc};
+use tokio::sync::mpsc;
 
 use super::executor::{EngineBoxFuture, JoinHandle, TaskSet, spawn};
 use crate::{
@@ -68,9 +67,6 @@ pub(super) struct SessionContext {
     pub(super) events: EventSender,
     pub(super) media: Arc<MediaService>,
     pub(super) mentions: crate::mentions::service::Handle,
-    pub(super) sync: Arc<SyncService>,
-    pub(super) repair_gate: Arc<Mutex<()>>,
-    pub(super) repair_shutdown: Arc<std::sync::atomic::AtomicBool>,
     pub(super) current_user_id: Option<OwnedUserId>,
     pub(super) initial_history_limit: u16,
     pub(super) supervisor: Supervisor,
@@ -118,7 +114,7 @@ impl TimelineSession {
             member_fetch,
             context,
         };
-        if session.context.initial_history_limit > 0 {
+        if session.context.initial_history_limit > 0 && !matches!(session.id, TimelineId::Room(_)) {
             session.paginate(session.context.initial_history_limit, None);
         }
         Ok(session)
@@ -135,6 +131,19 @@ impl TimelineSession {
         Ok(())
     }
 
+    pub(super) async fn publish_history(&mut 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())
+    }
+
     pub(super) fn paginate(&self, limit: u16, reply: Option<Responder>) {
         if let Err(error) = self.pagination.try_send(PaginationRequest { limit, reply }) {
             let (request, error) = match error {
@@ -153,10 +162,6 @@ impl TimelineSession {
         }
     }
 
-    pub(super) async fn pause_for_reseed(&mut self) {
-        self.stop_without_close().await;
-    }
-
     async fn stop_without_close(&mut self) {
         self.member_fetch.abort();
         self.pagination_worker.abort();
@@ -168,19 +173,6 @@ impl TimelineSession {
         self.context.media.release(&self.scope).await;
     }
 
-    pub(super) async fn resume_after_failed_reseed(&mut self) -> Result<(), Error> {
-        self.watcher = start_watcher(&self.id, &self.timeline, &self.context, &self.scope)
-            .await
-            .map_err(|_| Error::new(ErrorKind::Timeline, "timeline reset could not be sent"))?;
-        (self.pagination, self.pagination_worker) =
-            start_pagination_worker(&self.id, &self.timeline, &self.context);
-        self.member_fetch = self.context.supervisor.spawn("member fetch", {
-            let timeline = self.timeline.clone();
-            async move { timeline.fetch_members().await }
-        });
-        Ok(())
-    }
-
     /// pending paginations answer `Interrupted` as their responders drop
     pub(super) async fn close(mut self) {
         self.stop_without_close().await;

Modified file: src/lib.rs+10 additions−9 deletions@@ -64,17 +64,18 @@ pub use model::{
     MessageEditContent, MessageKind, MessageMentions, NotificationCommand, OwnDevice, PackCommand,
     PackImage, PackImageInfo, PackUsage, Patch, PowerLevelChanges, Profile, PushNotification,
     PushNotificationEvent, PushNotificationRoom, PushNotificationSender, PushNotificationStatus,
-    Reaction, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText, RoomCategory,
-    RoomCommand, RoomHero, RoomListFilter, RoomMatcher, RoomMembership, RoomNotificationMode,
-    RoomNotificationSettings, RoomPowerLevels, RoomPreset, RoomSummary, RoomVisibility, SendState,
-    Sender, SpaceChild, Termination, ThreadSummary, TimelineCommand, TimelineContent, TimelineDiff,
-    TimelineDiffError, TimelineEvent, TimelineId, TimelineItem, TimelineItemId, TypingState,
-    UploadDetails, UploadOutcome, UploadThumbnail, UtdCause, VerificationAction,
-    VerificationFailure, VerificationState, VerificationTarget,
+    Reaction, RecentHistoryState, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText,
+    RoomCategory, RoomCommand, RoomHero, RoomListFilter, RoomMatcher, RoomMembership,
+    RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomPreset, RoomSummary,
+    RoomVisibility, SendState, Sender, SpaceChild, Termination, ThreadSummary, TimelineCommand,
+    TimelineContent, TimelineDiff, TimelineDiffError, TimelineEvent, TimelineId, TimelineItem,
+    TimelineItemId, TypingState, UploadDetails, UploadOutcome, UploadThumbnail, UtdCause,
+    VerificationAction, VerificationFailure, VerificationState, VerificationTarget,
 };
 pub use repair_model::{
-    HistoryContinuity, HistoryRepairBudget, HistoryRepairCommand, HistoryRepairPhase,
-    HistoryRepairPolicy, HistoryRepairReport, HistoryRepairStop, HistoryRepairStrategy,
+    HistoryContinuity, HistoryGap, HistoryGapLifecycle, HistoryRepairBudget, HistoryRepairCommand,
+    HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairReport, HistoryRepairStop,
+    HistoryRepairStrategy,
 };
 pub use ruma;
 #[doc(hidden)]

Modified file: src/mirror.rs+26 additions−3 deletions@@ -20,9 +20,10 @@ use tokio::sync::broadcast;
 
 use crate::{
     ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
-    HistoryRepairReport, HistoryState, ImagePacks, RoomSummary, TimelineCommand, TimelineDiff,
-    TimelineId, TimelineItem, TypingState, UploadOutcome, VerificationState, VerificationTarget,
-    engine::executor::EngineBoxFuture, transport::PublicationRegistration,
+    HistoryRepairReport, HistoryState, ImagePacks, RecentHistoryState, RoomSummary,
+    TimelineCommand, TimelineDiff, TimelineId, TimelineItem, TypingState, UploadOutcome,
+    VerificationState, VerificationTarget, engine::executor::EngineBoxFuture,
+    transport::PublicationRegistration,
 };
 
 #[cfg(not(target_family = "wasm"))]
@@ -49,6 +50,8 @@ pub struct State {
     pub rooms: Vector<RoomSummary>,
     pub image_packs: ImagePacks,
     pub timelines: imbl::HashMap<TimelineId, TimelineState>,
+    #[cfg_attr(feature = "serde", serde(default))]
+    pub recent_history: imbl::HashMap<OwnedRoomId, RecentHistoryState>,
     pub repairs: imbl::HashMap<OwnedRoomId, HistoryRepairReport>,
     pub typing: imbl::HashMap<OwnedRoomId, Vec<OwnedUserId>>,
     pub uploads: imbl::HashMap<OwnedTransactionId, UploadState>,
@@ -136,6 +139,10 @@ pub enum StateChange {
         timeline_id: TimelineId,
         state: HistoryState,
     },
+    RecentHistory {
+        room_id: OwnedRoomId,
+        state: RecentHistoryState,
+    },
     HistoryRepair(HistoryRepairReport),
     Typing(TypingState),
     Upload {
@@ -209,6 +216,7 @@ impl Mirror {
             rooms: Vector::new(),
             image_packs: ImagePacks::default(),
             timelines: imbl::HashMap::new(),
+            recent_history: imbl::HashMap::new(),
             repairs: imbl::HashMap::new(),
             typing: imbl::HashMap::new(),
             uploads: imbl::HashMap::new(),
@@ -363,6 +371,16 @@ impl Mirror {
                     state: history,
                 })
             }),
+            Event::RecentHistoryChanged {
+                room_id,
+                state: recent,
+            } => self.publish(|state| {
+                state.recent_history.insert(room_id.clone(), recent.clone());
+                Some(StateChange::RecentHistory {
+                    room_id,
+                    state: recent,
+                })
+            }),
             Event::HistoryRepairChanged(report) => self.publish(|state| {
                 state.repairs.insert(report.room_id.clone(), report.clone());
                 Some(StateChange::HistoryRepair(report))
@@ -379,6 +397,11 @@ impl Mirror {
                     acknowledgement.complete(true);
                 }
             }
+            Event::PublicationBarrier { acknowledgement } => {
+                if let Some(acknowledgement) = acknowledgement {
+                    acknowledgement.complete(true);
+                }
+            }
             Event::TimelineClosed { timeline_id } => {
                 self.cancel_resync(&timeline_id);
                 self.publish(|state| {

Modified file: src/model.rs+19 additions@@ -907,6 +907,10 @@ pub enum Event {
         timeline_id: TimelineId,
         state: HistoryState,
     },
+    RecentHistoryChanged {
+        room_id: OwnedRoomId,
+        state: RecentHistoryState,
+    },
     HistoryRepairChanged(crate::HistoryRepairReport),
     #[doc(hidden)]
     HistoryRepairPublished {
@@ -914,6 +918,11 @@ pub enum Event {
         #[cfg_attr(feature = "serde", serde(skip))]
         acknowledgement: Option<crate::PublicationAck>,
     },
+    #[doc(hidden)]
+    PublicationBarrier {
+        #[cfg_attr(feature = "serde", serde(skip))]
+        acknowledgement: Option<crate::PublicationAck>,
+    },
     /// the timeline was closed, or its room was left, and nothing more comes for it
     /// until another `TimelineReset`
     TimelineClosed {
@@ -1376,6 +1385,7 @@ pub struct OwnDevice {
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum TimelineItem {
     Event(Box<TimelineEvent>),
+    HistoryGap(crate::HistoryGap),
     DayDivider { timestamp_ms: u64 },
     ReadMarker,
     TimelineStart,
@@ -2111,6 +2121,15 @@ pub enum HistoryState {
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum RecentHistoryState {
+    Idle,
+    Loading,
+    Loaded,
+    Failed(String),
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[non_exhaustive]
 pub struct EncryptionState {
     pub device_id: OwnedDeviceId,

Modified file: src/normalise/timeline.rs+39 additions−3 deletions@@ -19,9 +19,10 @@ use ruma::{
 };
 
 use crate::{
-    Change, MediaId, MembershipChange, Message, MessageKind, MessageMentions, Reaction,
-    ReplyDetails, ReplyReference, SendState, Sender, ThreadSummary, TimelineContent, TimelineDiff,
-    TimelineEvent, TimelineItem, TimelineItemId, UtdCause,
+    Change, HistoryContinuity, 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;
@@ -68,6 +69,41 @@ impl Normaliser<'_> {
             TimelineItemKind::Virtual(VirtualTimelineItem::TimelineStart) => {
                 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)
+                    }
+                };
+                TimelineItem::HistoryGap(HistoryGap {
+                    id: gap.id.clone(),
+                    older_boundary: gap.older_boundary.clone(),
+                    newer_boundary: gap.newer_boundary.clone(),
+                    lifecycle,
+                    continuity,
+                    failure,
+                })
+            }
         }
     }
 

Modified file: src/repair_model.rs+34 additions−6 deletions@@ -1,11 +1,11 @@
-use ruma::OwnedRoomId;
+use ruma::{OwnedEventId, OwnedRoomId, OwnedTransactionId};
 
 #[cfg(feature = "serde")]
 use serde::{Deserialize, Serialize};
 
 use crate::Error;
 
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum HistoryRepairPolicy {
     StartRecent,
@@ -30,8 +30,8 @@ pub struct HistoryRepairBudget {
 impl Default for HistoryRepairBudget {
     fn default() -> Self {
         Self {
-            max_pages: 10,
-            max_events: 1000,
+            max_pages: 4,
+            max_events: 200,
             max_duration_ms: 15_000,
         }
     }
@@ -55,11 +55,36 @@ pub enum HistoryRepairCommand {
     StartRecent {
         room_id: OwnedRoomId,
     },
+    FillGap {
+        room_id: OwnedRoomId,
+        gap_id: OwnedTransactionId,
+    },
     GetPolicy,
     SetPolicy(HistoryRepairPolicy),
 }
 
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub struct HistoryGap {
+    pub id: OwnedTransactionId,
+    pub older_boundary: Option<OwnedEventId>,
+    pub newer_boundary: Option<OwnedEventId>,
+    pub lifecycle: HistoryGapLifecycle,
+    pub continuity: HistoryContinuity,
+    pub failure: Option<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum HistoryGapLifecycle {
+    Idle,
+    Fetching,
+    Paused,
+    Failed,
+    Unavailable,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum HistoryRepairStrategy {
     Recent,
@@ -67,7 +92,7 @@ pub enum HistoryRepairStrategy {
     Bridge,
 }
 
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum HistoryContinuity {
     Unresolved,
@@ -98,6 +123,8 @@ pub enum HistoryRepairStop {
 #[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>,
@@ -132,6 +159,7 @@ mod tests {
         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,

Modified file: src/transport.rs+33 additions@@ -375,6 +375,18 @@ impl ClientHandle {
         .await
     }
 
+    pub async fn fill_history_gap(
+        &self,
+        room_id: OwnedRoomId,
+        gap_id: ruma::OwnedTransactionId,
+    ) -> Result<(), Error> {
+        self.run(Command::HistoryRepair(HistoryRepairCommand::FillGap {
+            room_id,
+            gap_id,
+        }))
+        .await
+    }
+
     pub async fn start_recent_history(
         &self,
         room_id: OwnedRoomId,
@@ -1620,6 +1632,26 @@ impl EventSender {
         self.events.send(event).await.map_err(|_| SendEventError)
     }
 
+    pub(crate) async fn publish_barrier(&self) -> Result<(), SendEventError> {
+        let Some((acknowledgement, published)) = self.publication.acknowledgement() else {
+            return Ok(());
+        };
+        if let Err(error) = self
+            .send(Event::PublicationBarrier {
+                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 publish_repair(
         &self,
         report: crate::HistoryRepairReport,
@@ -1812,6 +1844,7 @@ mod tests {
 
     fn report() -> HistoryRepairReport {
         HistoryRepairReport {
+            gap_id: None,
             room_id: room_id!("!room:example.org").to_owned(),
             requested_policy: HistoryRepairPolicy::Bridge,
             strategy: HistoryRepairStrategy::Bridge,

Modified file: tests/engine.rs+18 additions−8 deletions@@ -4,7 +4,8 @@ use std::time::Duration;
 
 use macltok::{
     ClientHandle, Command, Engine, EngineConfig, ErrorKind, Event, EventStream, HistoryState,
-    MediaSize, MessageContent, RoomCommand, TimelineCommand, TimelineId, TimelineItem,
+    MediaSize, MessageContent, RecentHistoryState, RoomCommand, TimelineCommand, TimelineId,
+    TimelineItem,
 };
 use matrix_sdk::{
     ruma::{api::client::sync::sync_events::v5, event_id, mxc_uri, room_id},
@@ -315,9 +316,9 @@ async fn closes_and_reopens_a_room_in_command_order() {
         .try_send(Command::Room(RoomCommand::Open(room_id.to_owned())))
         .unwrap();
 
-    // biased select drains every buffered reset before it ever reports a ready reply,
-    // so the engine can race ahead and land both resets before either open is observed
+    // 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);
@@ -326,11 +327,20 @@ async fn closes_and_reopens_a_room_in_command_order() {
             tokio::select! {
                 biased;
                 event = events.recv() => {
-                    if let Event::TimelineReset { timeline_id, .. } = event.expect("event stream closed")
-                        && timeline_id.room_id() == room_id
-                    {
-                        resets += 1;
-                        assert!(resets <= 2, "a room generation reset more than once");
+                    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 =>
+                        {
+                            resets += 1;
+                            assert!(resets <= 2, "a room generation reset more than once");
+                        }
+                        _ => {}
                     }
                 }
                 result = &mut close, if !close_done => {

Modified file: tests/history_repair.rs+319 additions−381 deletions@@ -3,9 +3,8 @@ mod support;
 use std::time::Duration;
 
 use macltok::{
-    ClientHandle, Engine, EngineConfig, EngineError, Event, EventStream, HistoryContinuity,
-    HistoryRepairBudget, HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairStop,
-    HistoryRepairStrategy, Termination, TimelineItem,
+    ClientHandle, Engine, EngineConfig, EngineError, Event, EventStream, HistoryGap,
+    HistoryRepairPhase, HistoryRepairPolicy, RecentHistoryState, Termination, TimelineItem,
 };
 #[cfg(feature = "mirror")]
 use macltok::{Mirror, TimelineId};
@@ -26,7 +25,7 @@ struct TestRoom {
     running: tokio::task::JoinHandle<Result<Termination, EngineError>>,
 }
 
-async fn open_room(room_id: &RoomId) -> TestRoom {
+async fn start_room(room_id: &RoomId) -> TestRoom {
     let server = MatrixMockServer::new().await;
     server
         .mock_sliding_sync()
@@ -38,15 +37,8 @@ async fn open_room(room_id: &RoomId) -> TestRoom {
     server.sync_joined_room(&client, room_id).await;
     let sync_client = client.clone();
     let config = EngineConfig::default().with_initial_history_limit(0);
-    let (engine, handle, mut events) = Engine::from_client(client, config).await.unwrap();
+    let (engine, handle, events) = Engine::from_client(client, config).await.unwrap();
     let running = tokio::spawn(engine.run());
-
-    handle.open_room(room_id.to_owned()).await.unwrap();
-    receive_until(&mut events, |event| {
-        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
-    })
-    .await;
-
     TestRoom {
         server,
         sync_client,
@@ -56,7 +48,15 @@ async fn open_room(room_id: &RoomId) -> TestRoom {
     }
 }
 
-async fn sync_cached_event(room: &mut TestRoom, room_id: &RoomId, id: &EventId, body: &str) {
+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| {
+        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
+    })
+    .await;
+}
+
+async fn cache_event_before_open(room: &TestRoom, room_id: &RoomId, id: &EventId, body: &str) {
     room.server
         .sync_room(
             &room.sync_client,
@@ -68,12 +68,6 @@ async fn sync_cached_event(room: &mut TestRoom, room_id: &RoomId, id: &EventId, 
             ),
         )
         .await;
-    receive_until(&mut room.events, |event| {
-        event_messages(event)
-            .into_iter()
-            .any(|event| event.event_id.as_deref() == Some(id))
-    })
-    .await;
 }
 
 fn history_event(
@@ -89,94 +83,178 @@ fn history_event(
         .into_raw_timeline()
 }
 
-fn report_budget(max_pages: u16) -> HistoryRepairBudget {
-    HistoryRepairBudget {
-        max_pages,
-        ..HistoryRepairBudget::default()
+async fn wait_for_recent(room: &mut TestRoom, room_id: &RoomId) -> RecentHistoryState {
+    let event = receive_until(&mut room.events, |event| {
+        matches!(event, Event::RecentHistoryChanged { room_id: id, state, .. }
+            if id == room_id && !matches!(state, RecentHistoryState::Loading))
+    })
+    .await;
+    match event {
+        Event::RecentHistoryChanged { state, .. } => state,
+        _ => unreachable!(),
+    }
+}
+
+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),
+    )
+    .await
+    {
+        Event::HistoryRepairChanged(report) => report,
+        _ => unreachable!(),
     }
 }
 
+fn gap_in(items: &imbl::Vector<TimelineItem>) -> Option<HistoryGap> {
+    items.iter().find_map(|item| match item {
+        TimelineItem::HistoryGap(gap) => Some(gap.clone()),
+        _ => None,
+    })
+}
+
 #[tokio::test]
-async fn recent_repair_starts_without_a_cursor_and_keeps_its_end_for_pagination() {
+async fn automatic_recent_load_is_bounded_and_keeps_cached_history_behind_a_gap() {
     let room_id = room_id!("!recent-repair:localhost");
-    let stale_id = event_id!("$stale:localhost");
-    let fresh_id = event_id!("$fresh:localhost");
-    let older_id = event_id!("$older:localhost");
-    let mut room = open_room(room_id).await;
-    sync_cached_event(&mut room, room_id, stale_id, "cached before repair").await;
-
-    room.server
-        .mock_room_messages()
-        .ok(RoomMessagesResponseTemplate::default()
-            .end_token("recent-prev")
-            .events(vec![history_event(room_id, fresh_id, "fresh page")]))
-        .mock_once()
-        .mount()
-        .await;
-    let report = room
-        .handle
-        .history_repair(
-            room_id.to_owned(),
-            Some(HistoryRepairPolicy::StartRecent),
-            HistoryRepairBudget::default(),
-        )
+    let old_id = event_id!("$old:localhost");
+    let new_ids = [
+        event_id!("$fresh-a:localhost"),
+        event_id!("$fresh-b:localhost"),
+        event_id!("$fresh-c:localhost"),
+        event_id!("$fresh-d:localhost"),
+    ];
+    let mut room = start_room(room_id).await;
+    room.handle
+        .set_history_repair_policy(HistoryRepairPolicy::Bridge)
         .await
         .unwrap();
-    assert_eq!(report.strategy, HistoryRepairStrategy::Recent);
-    assert_eq!(report.phase, HistoryRepairPhase::Ready);
-    assert!(report.recent_available);
-
+    for (index, id) in new_ids.iter().enumerate() {
+        let token = format!("recent-{index}");
+        let response = RoomMessagesResponseTemplate::default()
+            .end_token(token.clone())
+            .events(vec![history_event(room_id, id, "recent event")]);
+        if index > 0 {
+            let previous = format!("recent-{}", index - 1);
+            room.server
+                .mock_room_messages()
+                .match_from(&previous)
+                .ok(response)
+                .mock_once()
+                .mount()
+                .await;
+        } else {
+            room.server
+                .mock_room_messages()
+                .ok(response)
+                .mock_once()
+                .mount()
+                .await;
+        }
+    }
+    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| {
-        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
-            && event_messages(event)
-                .into_iter()
-                .any(|event| event.event_id.as_deref() == Some(fresh_id))
+        let items = match event {
+            Event::TimelineReset { timeline_id, items } if timeline_id.room_id() == room_id => {
+                items
+            }
+            _ => return false,
+        };
+        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))
+                })
+            })
     })
     .await;
-    let reset_ids = event_messages(&reset)
-        .into_iter()
-        .filter_map(|event| event.event_id.as_ref().map(ToString::to_string))
-        .collect::<Vec<_>>();
-    assert_eq!(reset_ids, vec![fresh_id.to_string()]);
-
+    let 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()
         .filter(|request| request.url.path().ends_with("/messages"))
         .collect::<Vec<_>>();
-    assert_eq!(messages.len(), 1);
+    assert_eq!(messages.len(), 4);
     assert!(!messages[0].url.query_pairs().any(|(key, _)| key == "from"));
+    shut_down(&room.handle, &mut room.events, room.running).await;
+}
 
+#[tokio::test]
+async fn explicit_recent_start_coalesces_with_the_automatic_load() {
+    let room_id = room_id!("!recent-coalesce:localhost");
+    let fresh_id = event_id!("$coalesced:localhost");
+    let mut room = start_room(room_id).await;
     room.server
         .mock_room_messages()
-        .match_from("recent-prev")
-        .ok(
-            RoomMessagesResponseTemplate::default().events(vec![history_event(
-                room_id,
-                older_id,
-                "older page",
-            )]),
-        )
+        .ok(RoomMessagesResponseTemplate::default()
+            .with_delay(Duration::from_millis(300))
+            .events(vec![history_event(room_id, fresh_id, "shared result")]))
         .mock_once()
         .mount()
         .await;
-    room.handle
-        .paginate(room_id.to_owned().into(), 10)
-        .await
-        .unwrap();
-    receive_until(&mut room.events, |event| {
-        event_messages(event).into_iter().any(|event| {
-            support::message(event).is_some_and(|message| message.body == "older page")
-        })
+    open_room(&mut room, room_id).await;
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let requested = room
+                .server
+                .received_requests()
+                .await
+                .unwrap()
+                .iter()
+                .any(|request| request.url.path().ends_with("/messages"));
+            if requested {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
     })
-    .await;
+    .await
+    .expect("automatic recent request did not start");
 
+    let report = room
+        .handle
+        .start_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
+    );
+    let requests = room.server.received_requests().await.unwrap();
+    assert_eq!(
+        requests
+            .iter()
+            .filter(|request| request.url.path().ends_with("/messages"))
+            .count(),
+        1
+    );
     shut_down(&room.handle, &mut room.events, room.running).await;
 }
 
 #[cfg(feature = "mirror")]
 #[tokio::test]
-async fn repair_ack_waits_until_the_mirror_has_applied_the_reset_and_report() {
+async fn fill_gap_returns_after_the_mirror_contains_the_committed_rows() {
     let server = MatrixMockServer::new().await;
     server
         .mock_sliding_sync()
@@ -184,10 +262,13 @@ async fn repair_ack_waits_until_the_mirror_has_applied_the_reset_and_report() {
         .mount()
         .await;
     let client = server.client_builder().build().await;
-    let room_id = room_id!("!repair-mirror:localhost");
+    let room_id = room_id!("!fill-gap-mirror:localhost");
     server.mock_room_state_encryption().plain().mount().await;
     server.sync_joined_room(&client, room_id).await;
     let sync_client = client.clone();
+    let old_id = event_id!("$gap-old:localhost");
+    let fresh_id = event_id!("$gap-fresh:localhost");
+    let middle_id = event_id!("$gap-middle:localhost");
     let (engine, handle, events) = Engine::from_client(
         client,
         EngineConfig::default().with_initial_history_limit(0),
@@ -197,79 +278,114 @@ async fn repair_ack_waits_until_the_mirror_has_applied_the_reset_and_report() {
     let running = tokio::spawn(engine.run());
     let (mirror, mirror_handle) = Mirror::new(handle.clone(), events);
     let mirror_running = tokio::spawn(mirror.run());
-
-    handle.open_room(room_id.to_owned()).await.unwrap();
-    let stale_id = event_id!("$mirror-stale:localhost");
+    server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("recent-gap")
+            .events(vec![history_event(room_id, fresh_id, "new window")]))
+        .mock_once()
+        .mount()
+        .await;
+    server
+        .mock_room_messages()
+        .match_from("recent-gap")
+        .ok(RoomMessagesResponseTemplate::default().end_token("recent-gap"))
+        .mock_once()
+        .mount()
+        .await;
     server
         .sync_room(
             &sync_client,
             JoinedRoomBuilder::new(room_id).add_timeline_event(
                 EventFactory::new()
-                    .text_msg("cached event")
-                    .event_id(stale_id)
+                    .text_msg("old retained")
+                    .event_id(old_id)
                     .sender(&ALICE),
             ),
         )
         .await;
+    handle.open_room(room_id.to_owned()).await.unwrap();
     let timeline_id = TimelineId::from(room_id.to_owned());
-    tokio::time::timeout(Duration::from_secs(2), async {
+    let gap = tokio::time::timeout(Duration::from_secs(5), async {
         loop {
-            let has_cached_event = mirror_handle
+            if let Some(gap) = mirror_handle
                 .snapshot()
                 .timelines
                 .get(&timeline_id)
-                .is_some_and(|timeline| {
-                    timeline.items.iter().any(|item| {
-                        matches!(item, TimelineItem::Event(event)
-                            if event.event_id.as_deref() == Some(stale_id))
-                    })
-                });
-            if has_cached_event {
+                .and_then(|timeline| gap_in(&timeline.items))
+            {
+                break gap;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("the recent history gap did not reach the mirror");
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if mirror_handle.snapshot().recent_history.get(room_id)
+                == Some(&RecentHistoryState::Loaded)
+            {
                 break;
             }
             tokio::task::yield_now().await;
         }
     })
     .await
-    .expect("the cached timeline event did not reach the mirror");
-
-    let fresh_id = event_id!("$mirror-fresh:localhost");
+    .expect("automatic recent loading did not finish");
     server
         .mock_room_messages()
-        .ok(
-            RoomMessagesResponseTemplate::default().events(vec![history_event(
-                room_id,
-                fresh_id,
-                "fresh event",
-            )]),
-        )
+        .match_from("recent-gap")
+        .ok(RoomMessagesResponseTemplate::default()
+            .with_delay(Duration::from_millis(300))
+            .events(vec![
+                history_event(room_id, middle_id, "middle page"),
+                history_event(room_id, old_id, "old retained overlap"),
+            ]))
         .mock_once()
         .mount()
         .await;
-    let report = handle
-        .history_repair(
-            room_id.to_owned(),
-            Some(HistoryRepairPolicy::StartRecent),
-            HistoryRepairBudget::default(),
-        )
-        .await
-        .unwrap();
 
-    let state = mirror_handle.snapshot();
-    assert_eq!(state.repairs.get(room_id), Some(&report));
-    let timeline = state
+    let fill_handle = handle.clone();
+    let target_gap_id = gap.id.clone();
+    let fill_gap_id = target_gap_id.clone();
+    let fill = tokio::spawn(async move {
+        fill_handle
+            .fill_history_gap(room_id.to_owned(), fill_gap_id)
+            .await
+    });
+    tokio::time::timeout(Duration::from_secs(5), async {
+        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
+            });
+            if active_gap {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("the active repair report did not identify its target gap");
+    fill.await.unwrap().unwrap();
+    let snapshot = mirror_handle.snapshot();
+    let timeline = snapshot
         .timelines
         .get(&timeline_id)
-        .expect("the repaired timeline was not applied before acknowledgement");
-    let ids = timeline
+        .expect("the room timeline disappeared");
+    let event_ids = timeline
         .items
         .iter()
         .filter_map(|item| match item {
-            TimelineItem::Event(event) => event.event_id.as_ref().map(ToString::to_string),
+            TimelineItem::Event(event) => event.event_id.as_deref(),
             _ => None,
         })
         .collect::<Vec<_>>();
-    assert_eq!(ids, vec![fresh_id.to_string()]);
+    assert!(event_ids.contains(&fresh_id));
+    assert!(event_ids.contains(&middle_id));
+    assert!(event_ids.contains(&old_id));
 
     handle.shutdown().await.unwrap();
     assert_eq!(running.await.unwrap().unwrap(), Termination::ShutDown);
@@ -277,90 +393,69 @@ async fn repair_ack_waits_until_the_mirror_has_applied_the_reset_and_report() {
 }
 
 #[tokio::test]
-async fn failed_recent_fetch_leaves_the_existing_timeline_in_place() {
-    let room_id = room_id!("!failed-repair:localhost");
-    let stale_id = event_id!("$kept:localhost");
-    let mut room = open_room(room_id).await;
-    sync_cached_event(&mut room, room_id, stale_id, "keep this event").await;
-
+async fn cancelling_a_gap_fill_keeps_committed_progress_for_retry() {
+    let room_id = room_id!("!cancel-gap-fill:localhost");
+    let old_id = event_id!("$cancel-gap-old:localhost");
+    let fresh_id = event_id!("$cancel-gap-fresh:localhost");
+    let middle_id = event_id!("$cancel-gap-middle:localhost");
+    let mut room = start_room(room_id).await;
     room.server
         .mock_room_messages()
-        .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({
-            "errcode": "M_FORBIDDEN",
-            "error": "history is unavailable",
-        })))
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("gap-cursor-1")
+            .events(vec![history_event(room_id, fresh_id, "recent")]))
         .mock_once()
         .mount()
         .await;
-    let report = room
-        .handle
-        .history_repair(
-            room_id.to_owned(),
-            Some(HistoryRepairPolicy::StartRecent),
-            HistoryRepairBudget::default(),
-        )
-        .await
-        .unwrap();
-    assert_eq!(report.phase, HistoryRepairPhase::Paused);
-    assert!(report.failure.is_some());
-
-    let mut reset_after_failure = false;
-    while let Ok(event) = room.events.try_recv() {
-        reset_after_failure |= matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id);
-    }
-    assert!(!reset_after_failure);
-
-    shut_down(&room.handle, &mut room.events, room.running).await;
-}
-
-#[tokio::test]
-async fn cancelling_a_delayed_bridge_page_preserves_the_recent_stage() {
-    let room_id = room_id!("!cancel-repair:localhost");
-    let anchor_id = event_id!("$cancel-anchor:localhost");
-    let recent_id = event_id!("$cancel-recent:localhost");
-    let mut room = open_room(room_id).await;
-    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;
-
     room.server
         .mock_room_messages()
-        .ok(RoomMessagesResponseTemplate::default()
-            .end_token("cancel-prev")
-            .events(vec![history_event(room_id, recent_id, "recent")]))
+        .match_from("gap-cursor-1")
+        .ok(RoomMessagesResponseTemplate::default().end_token("gap-cursor-1"))
         .mock_once()
         .mount()
         .await;
-    let initial = room
-        .handle
-        .history_repair(
-            room_id.to_owned(),
-            Some(HistoryRepairPolicy::Bridge),
-            report_budget(1),
-        )
-        .await
-        .unwrap();
-    assert_eq!(initial.stop, Some(HistoryRepairStop::Budget));
-    assert!(initial.recent_available);
-    receive_until(&mut room.events, |event| {
-        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
-            && event_messages(event)
-                .into_iter()
-                .any(|event| event.event_id.as_deref() == Some(recent_id))
+    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!(),
+    };
+    assert_eq!(
+        wait_for_recent(&mut room, room_id).await,
+        RecentHistoryState::Loaded
+    );
     room.server
         .mock_room_messages()
-        .match_from("cancel-prev")
+        .match_from("gap-cursor-1")
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("gap-cursor-2")
+            .events(vec![history_event(
+                room_id,
+                middle_id,
+                "committed progress",
+            )]))
+        .mock_once()
+        .mount()
+        .await;
+    room.server
+        .mock_room_messages()
+        .match_from("gap-cursor-2")
         .ok(RoomMessagesResponseTemplate::default()
             .with_delay(Duration::from_secs(5))
-            .events(vec![history_event(room_id, anchor_id, "cached anchor")]))
+            .events(vec![history_event(room_id, old_id, "old overlap")]))
         .mock_once()
         .mount()
         .await;
-    let repair_handle = room.handle.clone();
-    let repair = tokio::spawn(async move {
-        repair_handle
-            .continue_history_repair(room_id.to_owned(), HistoryRepairBudget::default())
+    let fill_handle = room.handle.clone();
+    let gap_id = gap.id.clone();
+    let fill = tokio::spawn(async move {
+        fill_handle
+            .fill_history_gap(room_id.to_owned(), gap_id)
             .await
     });
     tokio::time::timeout(Duration::from_secs(5), async {
@@ -376,7 +471,7 @@ async fn cancelling_a_delayed_bridge_page_preserves_the_recent_stage() {
                         && request
                             .url
                             .query_pairs()
-                            .any(|(key, value)| key == "from" && value == "cancel-prev")
+                            .any(|(key, value)| key == "from" && value == "gap-cursor-2")
                 });
             if requested {
                 break;
@@ -385,227 +480,70 @@ async fn cancelling_a_delayed_bridge_page_preserves_the_recent_stage() {
         }
     })
     .await
-    .expect("bridge request did not start");
+    .expect("the second gap page did not start");
     room.handle
         .cancel_history_repair(room_id.to_owned())
         .await
         .unwrap();
-    let report = repair.await.unwrap().unwrap();
-    assert_eq!(report.phase, HistoryRepairPhase::Paused);
-    assert_eq!(report.stop, Some(HistoryRepairStop::Cancelled));
-    assert!(report.recent_available);
-    assert_eq!(report.pages, 1);
-
-    let mut reset_after_cancel = false;
-    while let Ok(event) = room.events.try_recv() {
-        reset_after_cancel |= matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id);
-    }
-    assert!(!reset_after_cancel);
-
-    shut_down(&room.handle, &mut room.events, room.running).await;
-}
-
-#[tokio::test]
-async fn bridge_reaches_the_cached_anchor_deduplicates_overlap_and_can_continue_after_budget() {
-    let room_id = room_id!("!bridge-repair:localhost");
-    let anchor_id = event_id!("$anchor:localhost");
-    let middle_id = event_id!("$middle:localhost");
-    let recent_id = event_id!("$recent:localhost");
-    let latest_id = event_id!("$latest:localhost");
-    let mut room = open_room(room_id).await;
-    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;
+    let error = fill.await.unwrap().unwrap_err();
+    assert_eq!(error.kind, macltok::ErrorKind::Interrupted);
 
     room.server
         .mock_room_messages()
-        .ok(RoomMessagesResponseTemplate::default()
-            .end_token("recent-prev")
-            .events(vec![
-                history_event(room_id, recent_id, "recent"),
-                history_event(room_id, middle_id, "middle"),
-            ]))
-        .mock_once()
-        .mount()
-        .await;
-    let initial = room
-        .handle
-        .history_repair(
-            room_id.to_owned(),
-            Some(HistoryRepairPolicy::Bridge),
-            report_budget(1),
+        .match_from("gap-cursor-2")
+        .ok(
+            RoomMessagesResponseTemplate::default().events(vec![history_event(
+                room_id,
+                old_id,
+                "old overlap",
+            )]),
         )
-        .await
-        .unwrap();
-    assert_eq!(initial.phase, HistoryRepairPhase::Paused);
-    assert_eq!(initial.stop, Some(HistoryRepairStop::Budget));
-    assert!(initial.recent_available);
-
-    wiremock::Mock::given(|request: &wiremock::Request| {
-        request.url.path().ends_with("/messages")
-            && !request.url.query_pairs().any(|(key, _)| key == "from")
-    })
-    .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
-        "start": "head-start",
-        "chunk": [
-            history_event(room_id, latest_id, "latest"),
-            history_event(room_id, recent_id, "recent overlap"),
-            history_event(room_id, middle_id, "middle overlap"),
-        ],
-        "state": [],
-    })))
-    .up_to_n_times(1)
-    .expect(1)
-    .mount(room.server.server())
-    .await;
-    room.server
-        .mock_room_messages()
-        .match_from("recent-prev")
-        .ok(RoomMessagesResponseTemplate::default()
-            .end_token("older-prev")
-            .events(vec![
-                history_event(room_id, middle_id, "middle overlap"),
-                history_event(room_id, anchor_id, "cached anchor overlap"),
-            ]))
         .mock_once()
         .mount()
         .await;
-
-    let report = room
-        .handle
-        .continue_history_repair(room_id.to_owned(), HistoryRepairBudget::default())
+    room.handle
+        .fill_history_gap(room_id.to_owned(), gap.id)
         .await
         .unwrap();
-    assert_eq!(report.strategy, HistoryRepairStrategy::Bridge);
-    assert_eq!(report.phase, HistoryRepairPhase::Ready, "{report:?}");
-    assert_eq!(report.continuity, HistoryContinuity::Verified);
-    assert_eq!(report.pages, 3);
-
-    let reset = receive_until(&mut room.events, |event| {
-        matches!(event, Event::TimelineReset { timeline_id, .. } if timeline_id.room_id() == room_id)
-            && event_messages(event)
-                .into_iter()
-                .any(|event| event.event_id.as_deref() == Some(latest_id))
+    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))))
     })
     .await;
-    let reset_ids = event_messages(&reset)
-        .into_iter()
-        .filter_map(|event| event.event_id.as_ref().map(ToString::to_string))
+    let ids = event_messages(&final_reset)
+        .iter()
+        .filter_map(|event| event.event_id.as_deref())
         .collect::<Vec<_>>();
-    assert_eq!(
-        reset_ids,
-        vec![
-            anchor_id.to_string(),
-            middle_id.to_string(),
-            recent_id.to_string(),
-            latest_id.to_string(),
-        ]
-    );
-
+    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;
 }
 
 #[tokio::test]
-async fn a_recent_window_containing_the_cached_anchor_overlaps_under_every_policy() {
-    let cases = [
-        (
-            room_id!("!policy-recent:localhost"),
-            event_id!("$policy-recent-anchor:localhost"),
-            HistoryRepairPolicy::StartRecent,
-        ),
-        (
-            room_id!("!policy-bridge:localhost"),
-            event_id!("$policy-bridge-anchor:localhost"),
-            HistoryRepairPolicy::Bridge,
-        ),
-        (
-            room_id!("!policy-short-gap:localhost"),
-            event_id!("$policy-short-anchor:localhost"),
-            HistoryRepairPolicy::BridgeShortGaps { threshold_ms: 0 },
-        ),
-    ];
-
-    for (room_id, anchor_id, policy) in cases {
-        let mut room = open_room(room_id).await;
-        sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;
-        room.server
-            .mock_room_messages()
-            .ok(
-                RoomMessagesResponseTemplate::default().events(vec![history_event(
-                    room_id,
-                    anchor_id,
-                    "same anchor in recent history",
-                )]),
-            )
-            .mock_once()
-            .mount()
-            .await;
-
-        let report = room
-            .handle
-            .history_repair(
-                room_id.to_owned(),
-                Some(policy.clone()),
-                HistoryRepairBudget::default(),
-            )
-            .await
-            .unwrap();
-        assert_eq!(report.requested_policy, policy);
-        assert_eq!(report.strategy, HistoryRepairStrategy::Overlap);
-        assert_eq!(report.continuity, HistoryContinuity::Verified);
-        assert_eq!(report.pages, 1);
-        assert!(report.recent_available);
-
-        let requests = room.server.received_requests().await.unwrap();
-        assert_eq!(
-            requests
-                .iter()
-                .filter(|request| request.url.path().ends_with("/messages"))
-                .count(),
-            1
-        );
-        shut_down(&room.handle, &mut room.events, room.running).await;
-    }
-}
-
-#[tokio::test]
-async fn short_gap_policy_stays_recent_when_coverage_is_unknown() {
-    let room_id = room_id!("!unknown-coverage:localhost");
-    let anchor_id = event_id!("$unknown-anchor:localhost");
-    let fresh_id = event_id!("$unknown-fresh:localhost");
-    let mut room = open_room(room_id).await;
-    sync_cached_event(&mut room, room_id, anchor_id, "cached anchor").await;
+async fn failed_automatic_recent_load_is_public_and_keeps_the_cached_timeline() {
+    let room_id = room_id!("!failed-recent:localhost");
+    let old_id = event_id!("$failed-old:localhost");
+    let mut room = start_room(room_id).await;
     room.server
         .mock_room_messages()
-        .ok(RoomMessagesResponseTemplate::default()
-            .end_token("recent-prev")
-            .events(vec![history_event(room_id, fresh_id, "recent window")]))
+        .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({
+            "errcode": "M_FORBIDDEN",
+            "error": "history is unavailable",
+        })))
         .mock_once()
         .mount()
         .await;
-
-    let report = room
-        .handle
-        .history_repair(
-            room_id.to_owned(),
-            Some(HistoryRepairPolicy::BridgeShortGaps {
-                threshold_ms: 60_000,
-            }),
-            HistoryRepairBudget::default(),
-        )
-        .await
-        .unwrap();
-    assert_eq!(report.coverage_duration_ms, None);
-    assert!(!report.threshold_evaluated);
-    assert_eq!(report.strategy, HistoryRepairStrategy::Recent);
-    assert_eq!(report.continuity, HistoryContinuity::Unresolved);
-    assert_eq!(report.pages, 1);
-
-    let requests = room.server.received_requests().await.unwrap();
-    assert_eq!(
-        requests
-            .iter()
-            .filter(|request| request.url.path().ends_with("/messages"))
-            .count(),
-        1
-    );
+    cache_event_before_open(&room, room_id, old_id, "kept event").await;
+    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;
 }

macltok

Commit c9f0fd0ee1cc

Commit `c9f0fd0ee1cc`