macltok

public

Star0

Fork 0

Watch0

15 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `a319bae745d8`

tarball

atagen <boss@atagen.co>2026-09-27 01:42 UTC

8 files changed+1088-3

mirror: fold the event stream into shared state for many subscribers

Every host was writing the same drain loop, and malice got it wrong in ways that only show under load, with colliding thread timelines, half-applied batches and a resync awaited where it can deadlock the engine. Mirror owns that loop behind the mirror feature, and snapshot and subscription share one lock and sequence number so a late subscriber sees no gap and nothing twice. It takes any event stream, so a host can still inspect the raw events.

Modified file: Cargo.toml+1

@@ -6,6 +6,7 @@ description = "A compact application backend on top of the Matrix Rust SDK"

77 [features]

88 default = []

9+mirror = []

910 serde = ["dep:serde", "imbl/serde"]

1011 sqlite = ["matrix-sdk/sqlite", "serde", "dep:url", "dep:matrix-sdk-store-encryption"]

1112 bundled-sqlite = ["sqlite", "matrix-sdk/bundled-sqlite"]

Modified file: README.md+28 additions@@ -60,6 +60,34 @@ nothing, and the login overwrites it with a live record. `ShutDown` and
 one store directory per login, the sdk won't share a crypto store between
 devices
 
+## The mirror
+
+the loop above is easy to get subtly wrong, so the `mirror` feature ships one.
+`Mirror::new` takes a `ClientHandle` and the events, and `Mirror::run` drains
+them into one `State` with the rooms, every open timeline, connection,
+encryption, verification, typing, uploads and where the session stands. spawn
+it the way you spawn `Engine::run`. it takes any stream of events, so wrap the
+`EventStream` if you want to watch the raw ones go past
+
+`MirrorHandle` is cheap to clone, one per frontend. `snapshot` hands back the
+current `State`, and `subscribe` opens with the whole `State` and then every
+`Update` after it, numbered, with nothing missed and nothing twice. a timeline
+update carries its diffs and the items they left, so apply the diffs or
+re-render the list, whichever suits. a subscriber that falls behind gets a fresh
+`State` instead of an error
+
+a batch that doesn't fit leaves the timeline diverged at its last good items.
+the mirror asks for the resync itself, off the draining path, and keeps asking
+until the reset or a close turns up. rooms are refetched on `RoomsChanged`, one
+fetch at a time with at most one more owed
+
+with a store feature, `with_session_store` saves the record on every
+`SessionTokensChanged` before the next event is touched, and a failed save shows
+up as an update. without one, catch `SessionTokensChanged` on the raw stream
+and save it yourself, a subscriber that lags could miss it. `save` is all the
+mirror calls. `soft_log_out` and `discard` wait for `Engine::run`'s `Termination`
+once the client is gone
+
 ## wasm
 
 macltok builds for `wasm32-unknown-unknown`. the `indexeddb` feature gives you

Modified file: docs/design-decisions.md+16 additions@@ -217,6 +217,22 @@ state, since the sdk only reports what changes after you ask. the watcher and
 the action paths claim the terminal event, so a racing
 cancel can't turn `Done` into `Cancelled`
 
+## mirror
+
+the mirror is a consumer like any other, behind the `mirror` feature, and it
+takes any stream of events rather than the `EventStream` itself so a host keeps
+the raw events without a second channel. state, its sequence number and the
+broadcast send all move under one lock, which is how `subscribe` gets a
+snapshot and a receiver with nothing between them. a lagged subscriber starts
+over from a fresh state under the same lock
+
+nothing on the draining path waits for an engine answer. a resync runs on its
+own task, retrying with backoff until a reset or a close for its timeline
+aborts it. rooms are fetched on a task too, one at a time with one more owed at
+most. the only thing it does wait for is `SessionStore::save`, since a refresh
+token spent before the save lands loses the login. `SessionTokensChanged`
+never becomes an update, a lagging subscriber would lose it
+
 ## public model
 
 owned ruma ids and typed data wherever there's a stable shape, `Raw` where

Modified file: src/engine/executor.rs+2 additions−2 deletions@@ -5,9 +5,9 @@ use matrix_sdk::SendOutsideWasm;
 pub(super) use matrix_sdk::executor::{AbortHandle, JoinError, JoinHandle, spawn};
 
 #[cfg(not(target_family = "wasm"))]
-pub(super) type EngineBoxFuture<'a, T> = futures_util::future::BoxFuture<'a, T>;
+pub(crate) type EngineBoxFuture<'a, T> = futures_util::future::BoxFuture<'a, T>;
 #[cfg(target_family = "wasm")]
-pub(super) type EngineBoxFuture<'a, T> = futures_util::future::LocalBoxFuture<'a, T>;
+pub(crate) type EngineBoxFuture<'a, T> = futures_util::future::LocalBoxFuture<'a, T>;
 
 #[cfg(not(target_family = "wasm"))]
 pub(super) fn is_panic(error: &JoinError) -> bool {

Modified file: src/engine/mod.rs+1−1

@@ -1,5 +1,5 @@

11 mod commands;

2-mod executor;

2+pub(crate) mod executor;

33 mod media;

44 mod mutations;

55 mod room;

Modified file: src/lib.rs+7 additions@@ -3,6 +3,8 @@
 mod encryption;
 mod engine;
 mod error;
+#[cfg(feature = "mirror")]
+mod mirror;
 mod model;
 mod normalise;
 mod transport;
@@ -30,6 +32,11 @@ pub use session_store::{EncryptedSessionStore, SessionStore};
 pub use engine::{Engine, EngineConfig, EngineError};
 pub use error::{Error, ErrorKind};
 pub use imbl;
+#[cfg(feature = "mirror")]
+pub use mirror::{
+    Mirror, MirrorHandle, SessionStatus, State, StateChange, Subscription, TimelineChange,
+    TimelineState, Update, UploadFailure, UploadState,
+};
 pub use model::{
     AccountCommand, AttachmentUpload, Change, Command, CommandOutput, ConnectionState,
     CreateRoomOptions, EncryptionCommand, EncryptionState, EngineErrorKind, EngineFailure, Event,

Added file: src/mirror.rs+638 additions@@ -0,0 +1,638 @@
+use std::{
+    collections::HashMap,
+    fmt,
+    pin::Pin,
+    sync::{Arc, Mutex, MutexGuard, PoisonError},
+    task::{Context, Poll},
+    time::Duration,
+};
+
+use futures_util::{Stream, StreamExt};
+use imbl::Vector;
+use matrix_sdk::{
+    SendOutsideWasm,
+    executor::{AbortHandle, spawn},
+};
+use ruma::{OwnedDeviceId, OwnedRoomId, OwnedTransactionId, OwnedUserId};
+#[cfg(feature = "serde")]
+use serde::{Deserialize, Serialize};
+use tokio::sync::broadcast;
+
+use crate::{
+    ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
+    HistoryState, RoomSummary, TimelineCommand, TimelineDiff, TimelineId, TimelineItem,
+    TypingState, UploadOutcome, VerificationState, VerificationTarget,
+    engine::executor::EngineBoxFuture,
+};
+
+#[cfg(not(target_family = "wasm"))]
+type EngineBoxStream<T> = futures_util::stream::BoxStream<'static, T>;
+#[cfg(target_family = "wasm")]
+type EngineBoxStream<T> = futures_util::stream::LocalBoxStream<'static, T>;
+
+const DEFAULT_CAPACITY: usize = 256;
+const RESYNC_BACKOFF: Duration = Duration::from_millis(100);
+const RESYNC_BACKOFF_CAP: Duration = Duration::from_secs(10);
+
+/// everything the engine has said so far, folded together
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct State {
+    pub user_id: Option<OwnedUserId>,
+    pub session: SessionStatus,
+    pub connection: ConnectionState,
+    pub encryption: Option<EncryptionState>,
+    pub verification: Option<VerificationState>,
+    pub verification_target: Option<VerificationTarget>,
+    pub rooms: Vector<RoomSummary>,
+    pub timelines: imbl::HashMap<TimelineId, TimelineState>,
+    pub typing: imbl::HashMap<OwnedRoomId, Vec<OwnedUserId>>,
+    pub uploads: imbl::HashMap<OwnedTransactionId, UploadState>,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct TimelineState {
+    pub items: Vector<TimelineItem>,
+    pub history: HistoryState,
+    /// a batch didn't fit, so `items` is the last good copy until the next reset
+    pub diverged: bool,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct UploadState {
+    pub room_id: OwnedRoomId,
+    pub uploaded_bytes: u64,
+    pub total_bytes: u64,
+    pub complete: bool,
+    pub failure: Option<UploadFailure>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct UploadFailure {
+    pub kind: ErrorKind,
+    pub will_retry: bool,
+}
+
+/// the `Termination` it'll end in, as far as the events tell
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub enum SessionStatus {
+    Active,
+    SoftLoggedOut {
+        device_id: OwnedDeviceId,
+    },
+    AccessRevoked,
+    LoggedOut,
+    /// stopped with the session still good
+    Stopped,
+}
+
+#[derive(Clone, Debug)]
+#[non_exhaustive]
+pub struct Update {
+    /// one more than the update before it, and a `State` carries the one it's current as of
+    pub sequence: u64,
+    pub change: StateChange,
+}
+
+#[derive(Clone, Debug)]
+#[non_exhaustive]
+pub enum StateChange {
+    /// start over from this. a subscription opens with one and gets another after lagging
+    State(Arc<State>),
+    Authenticated(OwnedUserId),
+    Session(SessionStatus),
+    /// the record on disk may now hold a spent refresh token
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    SessionSaveFailed(Arc<crate::AccountError>),
+    Connection(ConnectionState),
+    Encryption(EncryptionState),
+    Verification(VerificationState),
+    VerificationTarget(VerificationTarget),
+    /// room modes arrive with the `Rooms` that follows
+    NotificationSettings,
+    Rooms(Vector<RoomSummary>),
+    Timeline {
+        timeline_id: TimelineId,
+        change: TimelineChange,
+    },
+    History {
+        timeline_id: TimelineId,
+        state: HistoryState,
+    },
+    Typing(TypingState),
+    Upload {
+        transaction_id: OwnedTransactionId,
+        upload: UploadState,
+    },
+    UploadFinished {
+        room_id: OwnedRoomId,
+        transaction_id: OwnedTransactionId,
+        outcome: UploadOutcome,
+    },
+    Failed(EngineFailure),
+}
+
+#[derive(Clone, Debug)]
+#[non_exhaustive]
+pub enum TimelineChange {
+    Reset {
+        items: Vector<TimelineItem>,
+    },
+    /// `items` is what applying `diffs` left
+    Changed {
+        diffs: Arc<[TimelineDiff]>,
+        items: Vector<TimelineItem>,
+    },
+    /// a batch didn't fit. a resync is on its way and nothing changes until its reset
+    Diverged,
+    Closed,
+}
+
+/// drains the event stream into a `State` that any number of `MirrorHandle`s watch
+pub struct Mirror {
+    handle: ClientHandle,
+    events: EngineBoxStream<Event>,
+    shared: Arc<Shared>,
+    resyncs: HashMap<TimelineId, AbortHandle>,
+    rooms_fetch: Option<AbortHandle>,
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    session_store: Option<Arc<dyn crate::SessionStore>>,
+}
+
+impl fmt::Debug for Mirror {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.debug_struct("Mirror").finish_non_exhaustive()
+    }
+}
+
+impl Mirror {
+    /// takes any stream of the engine's events, so a host can still look at them on the way past
+    pub fn new(
+        handle: ClientHandle,
+        events: impl Stream<Item = Event> + SendOutsideWasm + 'static,
+    ) -> (Self, MirrorHandle) {
+        Self::with_capacity(handle, events, DEFAULT_CAPACITY)
+    }
+
+    /// `capacity` updates can queue for a subscriber before it lags and starts over
+    pub fn with_capacity(
+        handle: ClientHandle,
+        events: impl Stream<Item = Event> + SendOutsideWasm + 'static,
+        capacity: usize,
+    ) -> (Self, MirrorHandle) {
+        let state = State {
+            user_id: None,
+            session: SessionStatus::Active,
+            connection: *handle.connection_state().borrow(),
+            encryption: handle.encryption_state(),
+            verification: None,
+            verification_target: None,
+            rooms: Vector::new(),
+            timelines: imbl::HashMap::new(),
+            typing: imbl::HashMap::new(),
+            uploads: imbl::HashMap::new(),
+        };
+        let shared = Arc::new(Shared {
+            published: Mutex::new(Published {
+                state: Arc::new(state),
+                sequence: 0,
+                updates: Some(broadcast::channel(capacity).0),
+                rooms_fetch: RoomsFetch::Idle,
+            }),
+        });
+        #[cfg(not(target_family = "wasm"))]
+        let events = events.boxed();
+        #[cfg(target_family = "wasm")]
+        let events = events.boxed_local();
+        let mirror = Self {
+            handle,
+            events,
+            shared: shared.clone(),
+            resyncs: HashMap::new(),
+            rooms_fetch: None,
+            #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+            session_store: None,
+        };
+        (mirror, MirrorHandle { shared })
+    }
+
+    /// saves the record on every `SessionTokensChanged` before taking another event, and
+    /// calls nothing but `save`. the store can't go while the client holds it open
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    #[must_use]
+    pub fn with_session_store(mut self, store: Arc<dyn crate::SessionStore>) -> Self {
+        self.session_store = Some(store);
+        self
+    }
+
+    /// resolves once the event stream ends, which is just after `Stopped`
+    pub fn run(mut self) -> EngineBoxFuture<'static, ()> {
+        Box::pin(async move {
+            self.refetch_rooms();
+            while let Some(event) = self.events.next().await {
+                self.apply(event).await;
+            }
+        })
+    }
+
+    async fn apply(&mut self, event: Event) {
+        match event {
+            Event::ConnectionChanged(connection) => self.publish(|state| {
+                state.connection = connection;
+                Some(StateChange::Connection(connection))
+            }),
+            Event::Authenticated { user_id } => self.publish(|state| {
+                state.user_id = Some(user_id.clone());
+                Some(StateChange::Authenticated(user_id))
+            }),
+            #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+            Event::SessionTokensChanged => self.save_session().await,
+            #[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
+            Event::SessionTokensChanged => {}
+            Event::SessionAccessRejected { soft_logout } => self.publish(|state| {
+                state.session = match soft_logout {
+                    Some(device_id) => SessionStatus::SoftLoggedOut { device_id },
+                    None => SessionStatus::AccessRevoked,
+                };
+                Some(StateChange::Session(state.session.clone()))
+            }),
+            Event::LoggedOut => self.publish(|state| {
+                state.session = SessionStatus::LoggedOut;
+                Some(StateChange::Session(SessionStatus::LoggedOut))
+            }),
+            Event::Stopped => {
+                self.cancel_resyncs();
+                self.publish(|state| {
+                    if state.session != SessionStatus::Active {
+                        return None;
+                    }
+                    state.session = SessionStatus::Stopped;
+                    Some(StateChange::Session(SessionStatus::Stopped))
+                });
+            }
+            Event::Failed(failure) => self.publish(|_| Some(StateChange::Failed(failure))),
+            Event::NotificationSettingsChanged => {
+                self.publish(|_| Some(StateChange::NotificationSettings));
+            }
+            Event::RoomsChanged => self.refetch_rooms(),
+            Event::EncryptionChanged(encryption) => self.publish(|state| {
+                state.encryption = Some(encryption.clone());
+                Some(StateChange::Encryption(encryption))
+            }),
+            Event::VerificationChanged(verification) => self.publish(|state| {
+                state.verification = Some(verification.clone());
+                Some(StateChange::Verification(verification))
+            }),
+            Event::VerificationTargetChanged(target) => self.publish(|state| {
+                state.verification_target = Some(target.clone());
+                Some(StateChange::VerificationTarget(target))
+            }),
+            Event::TimelineReset { timeline_id, items } => {
+                self.cancel_resync(&timeline_id);
+                self.publish(|state| {
+                    let history = state
+                        .timelines
+                        .get(&timeline_id)
+                        .map_or(HistoryState::Idle, |timeline| timeline.history.clone());
+                    let timeline = TimelineState {
+                        items: items.clone(),
+                        history,
+                        diverged: false,
+                    };
+                    state.timelines.insert(timeline_id.clone(), timeline);
+                    Some(StateChange::Timeline {
+                        timeline_id,
+                        change: TimelineChange::Reset { items },
+                    })
+                });
+            }
+            Event::TimelineChanged { timeline_id, diffs } => self.apply_diffs(timeline_id, diffs),
+            Event::HistoryChanged {
+                timeline_id,
+                state: history,
+            } => self.publish(|state| {
+                state.timelines.get_mut(&timeline_id)?.history = history.clone();
+                Some(StateChange::History {
+                    timeline_id,
+                    state: history,
+                })
+            }),
+            Event::TimelineClosed { timeline_id } => {
+                self.cancel_resync(&timeline_id);
+                self.publish(|state| {
+                    state.timelines.remove(&timeline_id)?;
+                    Some(StateChange::Timeline {
+                        timeline_id,
+                        change: TimelineChange::Closed,
+                    })
+                });
+            }
+            Event::TypingChanged(typing) => self.publish(|state| {
+                if typing.users.is_empty() {
+                    state.typing.remove(&typing.room_id);
+                } else {
+                    state
+                        .typing
+                        .insert(typing.room_id.clone(), typing.users.clone());
+                }
+                Some(StateChange::Typing(typing))
+            }),
+            Event::MediaUploadProgress {
+                room_id,
+                transaction_id,
+                uploaded_bytes,
+                total_bytes,
+                complete,
+            } => self.publish(|state| {
+                let upload = UploadState {
+                    room_id,
+                    uploaded_bytes,
+                    total_bytes,
+                    complete,
+                    failure: None,
+                };
+                state.uploads.insert(transaction_id.clone(), upload.clone());
+                Some(StateChange::Upload {
+                    transaction_id,
+                    upload,
+                })
+            }),
+            Event::MediaUploadFailed {
+                room_id,
+                transaction_id,
+                kind,
+                will_retry,
+            } => self.publish(|state| {
+                let upload = state
+                    .uploads
+                    .entry(transaction_id.clone())
+                    .or_insert_with(|| UploadState {
+                        room_id,
+                        uploaded_bytes: 0,
+                        total_bytes: 0,
+                        complete: false,
+                        failure: None,
+                    });
+                upload.failure = Some(UploadFailure { kind, will_retry });
+                Some(StateChange::Upload {
+                    transaction_id,
+                    upload: upload.clone(),
+                })
+            }),
+            Event::MediaUploadFinished {
+                room_id,
+                transaction_id,
+                outcome,
+            } => self.publish(|state| {
+                state.uploads.remove(&transaction_id);
+                Some(StateChange::UploadFinished {
+                    room_id,
+                    transaction_id,
+                    outcome,
+                })
+            }),
+        }
+    }
+
+    fn apply_diffs(&mut self, timeline_id: TimelineId, diffs: Vec<TimelineDiff>) {
+        let mut diverged = false;
+        self.publish(|state| {
+            let timeline = state.timelines.get_mut(&timeline_id)?;
+            if timeline.diverged {
+                return None;
+            }
+            let diffs: Arc<[TimelineDiff]> = diffs.into();
+            let change = match TimelineDiff::try_apply_all(diffs.to_vec(), &mut timeline.items) {
+                Ok(()) => TimelineChange::Changed {
+                    diffs,
+                    items: timeline.items.clone(),
+                },
+                Err(error) => {
+                    tracing::debug!(?timeline_id, length = error.length, "timeline diverged");
+                    timeline.diverged = true;
+                    diverged = true;
+                    TimelineChange::Diverged
+                }
+            };
+            Some(StateChange::Timeline {
+                timeline_id: timeline_id.clone(),
+                change,
+            })
+        });
+        if diverged {
+            let task = spawn(resync(self.handle.clone(), timeline_id.clone()));
+            self.resyncs.insert(timeline_id, task.abort_handle());
+        }
+    }
+
+    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
+    async fn save_session(&mut self) {
+        let Some(store) = &self.session_store else {
+            return;
+        };
+        let saved = match self.handle.session_record() {
+            Ok(record) => store.save(&record).await,
+            Err(error) => Err(error),
+        };
+        if let Err(error) = saved {
+            tracing::warn!(%error, "saving the session record failed");
+            self.publish(|_| Some(StateChange::SessionSaveFailed(Arc::new(error))));
+        }
+    }
+
+    fn refetch_rooms(&mut self) {
+        let mut published = self.shared.lock();
+        match published.rooms_fetch {
+            RoomsFetch::Idle => {
+                published.rooms_fetch = RoomsFetch::Running;
+                drop(published);
+                let task = spawn(fetch_rooms(self.handle.clone(), self.shared.clone()));
+                self.rooms_fetch = Some(task.abort_handle());
+            }
+            RoomsFetch::Running => published.rooms_fetch = RoomsFetch::Again,
+            RoomsFetch::Again => {}
+        }
+    }
+
+    fn cancel_resync(&mut self, timeline_id: &TimelineId) {
+        if let Some(task) = self.resyncs.remove(timeline_id) {
+            task.abort();
+        }
+    }
+
+    fn cancel_resyncs(&mut self) {
+        for (_, task) in self.resyncs.drain() {
+            task.abort();
+        }
+    }
+
+    fn publish(&self, change: impl FnOnce(&mut State) -> Option<StateChange>) {
+        self.shared.publish(change);
+    }
+}
+
+impl Drop for Mirror {
+    fn drop(&mut self) {
+        self.cancel_resyncs();
+        if let Some(task) = self.rooms_fetch.take() {
+            task.abort();
+        }
+        self.shared.lock().updates = None;
+    }
+}
+
+/// asks until the engine answers, since a busy room or a full queue is only ever for now.
+/// a reset or a close for the timeline aborts it, and the engine sends a reset before it answers
+async fn resync(handle: ClientHandle, timeline_id: TimelineId) {
+    let mut backoff = RESYNC_BACKOFF;
+    loop {
+        let command = Command::Timeline(TimelineCommand::Resync {
+            timeline_id: timeline_id.clone(),
+        });
+        match handle.execute(command).await {
+            Ok(_) => return,
+            Err(error) if error.kind == ErrorKind::Interrupted => return,
+            Err(error) => {
+                tracing::debug!(%error, ?timeline_id, "resync refused, trying again");
+                matrix_sdk::sleep::sleep(backoff).await;
+                backoff = (backoff * 2).min(RESYNC_BACKOFF_CAP);
+            }
+        }
+    }
+}
+
+async fn fetch_rooms(handle: ClientHandle, shared: Arc<Shared>) {
+    loop {
+        let rooms: Vector<RoomSummary> = handle.rooms().await.into();
+        shared.publish(|state| {
+            state.rooms = rooms.clone();
+            Some(StateChange::Rooms(rooms))
+        });
+        let mut published = shared.lock();
+        if published.rooms_fetch != RoomsFetch::Again {
+            published.rooms_fetch = RoomsFetch::Idle;
+            return;
+        }
+        published.rooms_fetch = RoomsFetch::Running;
+    }
+}
+
+/// cheap to clone, one per frontend
+#[derive(Clone, Debug)]
+pub struct MirrorHandle {
+    shared: Arc<Shared>,
+}
+
+impl MirrorHandle {
+    #[must_use]
+    pub fn snapshot(&self) -> Arc<State> {
+        self.shared.lock().state.clone()
+    }
+
+    /// opens with the current state, then every update after it. ends once the mirror stops
+    #[must_use]
+    pub fn subscribe(&self) -> Subscription {
+        let (first, receiving) = self.shared.clone().subscribe();
+        let rest = futures_util::stream::unfold(receiving, |(shared, receiver)| async move {
+            let mut receiver = receiver?;
+            match receiver.recv().await {
+                Ok(update) => Some((update, (shared, Some(receiver)))),
+                Err(broadcast::error::RecvError::Lagged(missed)) => {
+                    tracing::debug!(missed, "mirror subscriber lagged, starting it over");
+                    Some(shared.subscribe())
+                }
+                Err(broadcast::error::RecvError::Closed) => None,
+            }
+        });
+        let updates = futures_util::stream::once(std::future::ready(first)).chain(rest);
+        #[cfg(not(target_family = "wasm"))]
+        let updates = updates.boxed();
+        #[cfg(target_family = "wasm")]
+        let updates = updates.boxed_local();
+        Subscription(updates)
+    }
+}
+
+pub struct Subscription(EngineBoxStream<Update>);
+
+impl fmt::Debug for Subscription {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("Subscription")
+            .finish_non_exhaustive()
+    }
+}
+
+impl Stream for Subscription {
+    type Item = Update;
+
+    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Update>> {
+        self.0.poll_next_unpin(context)
+    }
+}
+
+#[derive(Debug)]
+struct Shared {
+    published: Mutex<Published>,
+}
+
+#[derive(Debug)]
+struct Published {
+    state: Arc<State>,
+    sequence: u64,
+    /// gone once the mirror stops, which ends every subscription
+    updates: Option<broadcast::Sender<Update>>,
+    rooms_fetch: RoomsFetch,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum RoomsFetch {
+    Idle,
+    Running,
+    Again,
+}
+
+impl Shared {
+    fn lock(&self) -> MutexGuard<'_, Published> {
+        self.published
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+    }
+
+    /// state, sequence and send move together under the one lock, which is what lets
+    /// `subscribe` take a snapshot and a receiver with nothing between them
+    fn publish(&self, change: impl FnOnce(&mut State) -> Option<StateChange>) {
+        let mut published = self.lock();
+        let Some(change) = change(Arc::make_mut(&mut published.state)) else {
+            return;
+        };
+        published.sequence += 1;
+        let update = Update {
+            sequence: published.sequence,
+            change,
+        };
+        if let Some(updates) = &published.updates {
+            let _ = updates.send(update);
+        }
+    }
+
+    fn subscribe(self: Arc<Self>) -> (Update, (Arc<Self>, Option<broadcast::Receiver<Update>>)) {
+        let published = self.lock();
+        let update = Update {
+            sequence: published.sequence,
+            change: StateChange::State(published.state.clone()),
+        };
+        let receiver = published.updates.as_ref().map(broadcast::Sender::subscribe);
+        drop(published);
+        (update, (self, receiver))
+    }
+}

Added file: tests/mirror.rs+395 additions@@ -0,0 +1,395 @@
+#![cfg(feature = "mirror")]
+
+use std::time::Duration;
+
+use futures_util::StreamExt;
+use macltok::{
+    ClientHandle, Command, Engine, EngineConfig, Event, Mirror, MirrorHandle, StateChange,
+    Subscription, TimelineChange, TimelineCommand, TimelineDiff, TimelineId, TimelineItem, Update,
+    UploadOutcome,
+};
+use matrix_sdk::{
+    ruma::{OwnedRoomId, TransactionId, api::client::sync::sync_events::v5, event_id, room_id},
+    test_utils::mocks::MatrixMockServer,
+};
+use tokio::sync::mpsc;
+
+struct Harness {
+    handle: ClientHandle,
+    mirror: MirrorHandle,
+    inject: mpsc::UnboundedSender<Event>,
+    room_id: OwnedRoomId,
+    _server: MatrixMockServer,
+}
+
+/// the mirror reads the real engine's stream with synthetic events mixed in, so a test can
+/// hand it a diff the engine never would
+async fn start(config: EngineConfig, capacity: usize) -> Harness {
+    let server = MatrixMockServer::new().await;
+    server
+        .mock_sliding_sync()
+        .ok(v5::Response::new("1".to_owned()))
+        .mount()
+        .await;
+    let client = server.client_builder().build().await;
+    let room_id = room_id!("!mirror:localhost");
+    server.mock_room_state_encryption().plain().mount().await;
+    server.sync_joined_room(&client, room_id).await;
+    let (engine, handle, events) =
+        Engine::from_client(client, config.with_initial_history_limit(0))
+            .await
+            .unwrap();
+    tokio::spawn(engine.run());
+    let (inject, mut injected) = mpsc::unbounded_channel();
+    let injected = futures_util::stream::poll_fn(move |context| injected.poll_recv(context));
+    let (mirror, mirror_handle) = Mirror::with_capacity(
+        handle.clone(),
+        futures_util::stream::select(events, injected),
+        capacity,
+    );
+    tokio::spawn(mirror.run());
+    Harness {
+        handle,
+        mirror: mirror_handle,
+        inject,
+        room_id: room_id.to_owned(),
+        _server: server,
+    }
+}
+
+async fn next_matching(
+    subscription: &mut Subscription,
+    predicate: impl Fn(&StateChange) -> bool,
+) -> Update {
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let update = subscription.next().await.expect("the mirror stopped");
+            if predicate(&update.change) {
+                return update;
+            }
+        }
+    })
+    .await
+    .expect("expected update did not arrive")
+}
+
+fn timeline_change(
+    change: &StateChange,
+    id: &TimelineId,
+    predicate: impl Fn(&TimelineChange) -> bool,
+) -> bool {
+    matches!(change, StateChange::Timeline { timeline_id, change } if timeline_id == id && predicate(change))
+}
+
+fn item(timestamp_ms: u64) -> TimelineItem {
+    TimelineItem::DayDivider { timestamp_ms }
+}
+
+fn reset(timeline_id: &TimelineId, items: Vec<TimelineItem>) -> Event {
+    Event::TimelineReset {
+        timeline_id: timeline_id.clone(),
+        items: items.into(),
+    }
+}
+
+/// the room never opens, so every resync is refused and the divergence holds still to be looked at
+#[tokio::test]
+async fn diverged_batch_keeps_last_good_items_until_a_reset() {
+    let harness = start(EngineConfig::default(), 256).await;
+    let timeline_id = TimelineId::from(harness.room_id.clone());
+    let mut updates = harness.mirror.subscribe();
+    harness
+        .inject
+        .send(reset(&timeline_id, vec![item(1), item(2)]))
+        .unwrap();
+    for diffs in [
+        vec![
+            TimelineDiff::PushBack { value: item(3) },
+            TimelineDiff::Remove { index: 7 },
+        ],
+        vec![TimelineDiff::PushBack { value: item(4) }],
+    ] {
+        harness
+            .inject
+            .send(Event::TimelineChanged {
+                timeline_id: timeline_id.clone(),
+                diffs,
+            })
+            .unwrap();
+    }
+    next_matching(&mut updates, |change| {
+        timeline_change(change, &timeline_id, |change| {
+            matches!(change, TimelineChange::Diverged)
+        })
+    })
+    .await;
+    let snapshot = harness.mirror.snapshot();
+    let timeline = &snapshot.timelines[&timeline_id];
+    assert!(timeline.diverged);
+    assert_eq!(timeline.items, [item(1), item(2)].into_iter().collect());
+
+    harness
+        .inject
+        .send(reset(&timeline_id, vec![item(5)]))
+        .unwrap();
+    let update = next_matching(&mut updates, |change| {
+        timeline_change(change, &timeline_id, |_| true)
+    })
+    .await;
+    assert!(
+        timeline_change(&update.change, &timeline_id, |change| {
+            matches!(change, TimelineChange::Reset { .. })
+        }),
+        "a diverged timeline took {update:?}"
+    );
+    let snapshot = harness.mirror.snapshot();
+    let timeline = &snapshot.timelines[&timeline_id];
+    assert!(!timeline.diverged);
+    assert_eq!(timeline.items, [item(5)].into_iter().collect());
+}
+
+#[tokio::test]
+async fn room_and_thread_timelines_stay_apart() {
+    let harness = start(EngineConfig::default(), 256).await;
+    let room = TimelineId::from(harness.room_id.clone());
+    let thread = TimelineId::thread(
+        harness.room_id.clone(),
+        event_id!("$root:localhost").to_owned(),
+    );
+    let mut updates = harness.mirror.subscribe();
+    harness.inject.send(reset(&room, vec![item(1)])).unwrap();
+    harness.inject.send(reset(&thread, vec![item(2)])).unwrap();
+    harness
+        .inject
+        .send(Event::TimelineChanged {
+            timeline_id: thread.clone(),
+            diffs: vec![TimelineDiff::PushBack { value: item(3) }],
+        })
+        .unwrap();
+    next_matching(&mut updates, |change| {
+        timeline_change(change, &thread, |change| {
+            matches!(change, TimelineChange::Changed { .. })
+        })
+    })
+    .await;
+    let snapshot = harness.mirror.snapshot();
+    assert_eq!(
+        snapshot.timelines[&room].items,
+        [item(1)].into_iter().collect()
+    );
+    assert_eq!(
+        snapshot.timelines[&thread].items,
+        [item(2), item(3)].into_iter().collect()
+    );
+}
+
+#[tokio::test]
+async fn late_subscriber_gets_state_then_every_update_after_it() {
+    let harness = start(EngineConfig::default(), 1024).await;
+    let transaction_id = TransactionId::new();
+    let progress = |uploaded_bytes| Event::MediaUploadProgress {
+        room_id: harness.room_id.clone(),
+        transaction_id: transaction_id.clone(),
+        uploaded_bytes,
+        total_bytes: 100,
+        complete: false,
+    };
+    for uploaded_bytes in 1..=50 {
+        harness.inject.send(progress(uploaded_bytes)).unwrap();
+    }
+    let mut updates = harness.mirror.subscribe();
+    for uploaded_bytes in 51..=100 {
+        harness.inject.send(progress(uploaded_bytes)).unwrap();
+    }
+
+    let first = updates.next().await.unwrap();
+    let StateChange::State(state) = &first.change else {
+        panic!("a subscription opened with {first:?}");
+    };
+    let mut seen = state
+        .uploads
+        .get(&transaction_id)
+        .map_or(0, |upload| upload.uploaded_bytes);
+    let mut sequence = first.sequence;
+    while seen < 100 {
+        let update = next_matching(&mut updates, |_| true).await;
+        assert_eq!(update.sequence, sequence + 1);
+        sequence = update.sequence;
+        if let StateChange::Upload { upload, .. } = update.change {
+            assert_eq!(upload.uploaded_bytes, seen + 1);
+            seen += 1;
+        }
+    }
+}
+
+#[tokio::test]
+async fn lagging_subscriber_starts_over_from_a_fresh_state() {
+    let harness = start(EngineConfig::default(), 4).await;
+    let transaction_id = TransactionId::new();
+    let mut updates = harness.mirror.subscribe();
+    assert!(matches!(
+        updates.next().await.unwrap().change,
+        StateChange::State(_)
+    ));
+    for uploaded_bytes in 1..=20 {
+        harness
+            .inject
+            .send(Event::MediaUploadProgress {
+                room_id: harness.room_id.clone(),
+                transaction_id: transaction_id.clone(),
+                uploaded_bytes,
+                total_bytes: 20,
+                complete: uploaded_bytes == 20,
+            })
+            .unwrap();
+    }
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while !harness
+            .mirror
+            .snapshot()
+            .uploads
+            .get(&transaction_id)
+            .is_some_and(|upload| upload.complete)
+        {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("the mirror never caught up");
+
+    let update = updates.next().await.unwrap();
+    let StateChange::State(state) = &update.change else {
+        panic!("a lagged subscription carried on with {update:?}");
+    };
+    assert!(state.uploads[&transaction_id].complete);
+}
+
+#[tokio::test]
+async fn resync_does_not_deadlock_on_full_queues() {
+    let config = EngineConfig::default()
+        .with_command_capacity(1)
+        .with_event_capacity(1);
+    let harness = start(config, 1024).await;
+    let timeline_id = TimelineId::from(harness.room_id.clone());
+    let mut updates = harness.mirror.subscribe();
+    harness
+        .handle
+        .open_room(harness.room_id.clone())
+        .await
+        .unwrap();
+    next_matching(&mut updates, |change| {
+        timeline_change(change, &timeline_id, |change| {
+            matches!(change, TimelineChange::Reset { .. })
+        })
+    })
+    .await;
+
+    let flood = tokio::spawn({
+        let handle = harness.handle.clone();
+        let timeline_id = timeline_id.clone();
+        async move {
+            loop {
+                let command = Command::Timeline(TimelineCommand::Paginate {
+                    timeline_id: timeline_id.clone(),
+                    limit: 1,
+                });
+                if handle.send(command).await.is_err() {
+                    return;
+                }
+            }
+        }
+    });
+    for _ in 0..10 {
+        harness
+            .inject
+            .send(Event::TimelineChanged {
+                timeline_id: timeline_id.clone(),
+                diffs: vec![TimelineDiff::Remove { index: 1000 }],
+            })
+            .unwrap();
+        next_matching(&mut updates, |change| {
+            timeline_change(change, &timeline_id, |change| {
+                matches!(change, TimelineChange::Diverged)
+            })
+        })
+        .await;
+        next_matching(&mut updates, |change| {
+            timeline_change(change, &timeline_id, |change| {
+                matches!(change, TimelineChange::Reset { .. })
+            })
+        })
+        .await;
+    }
+    flood.abort();
+}
+
+#[tokio::test]
+async fn closing_a_timeline_evicts_it() {
+    let harness = start(EngineConfig::default(), 256).await;
+    let timeline_id = TimelineId::from(harness.room_id.clone());
+    let mut updates = harness.mirror.subscribe();
+    harness
+        .handle
+        .open_room(harness.room_id.clone())
+        .await
+        .unwrap();
+    harness
+        .handle
+        .close_room(harness.room_id.clone())
+        .await
+        .unwrap();
+    next_matching(&mut updates, |change| {
+        timeline_change(change, &timeline_id, |change| {
+            matches!(change, TimelineChange::Closed)
+        })
+    })
+    .await;
+    assert!(
+        !harness
+            .mirror
+            .snapshot()
+            .timelines
+            .contains_key(&timeline_id)
+    );
+}
+
+#[tokio::test]
+async fn an_upload_ends_on_finished() {
+    let harness = start(EngineConfig::default(), 256).await;
+    let transaction_id = TransactionId::new();
+    let mut updates = harness.mirror.subscribe();
+    harness
+        .inject
+        .send(Event::MediaUploadProgress {
+            room_id: harness.room_id.clone(),
+            transaction_id: transaction_id.clone(),
+            uploaded_bytes: 5,
+            total_bytes: 10,
+            complete: false,
+        })
+        .unwrap();
+    next_matching(&mut updates, |change| {
+        matches!(change, StateChange::Upload { .. })
+    })
+    .await;
+    assert!(
+        harness
+            .mirror
+            .snapshot()
+            .uploads
+            .contains_key(&transaction_id)
+    );
+    harness
+        .inject
+        .send(Event::MediaUploadFinished {
+            room_id: harness.room_id.clone(),
+            transaction_id: transaction_id.clone(),
+            outcome: UploadOutcome::Cancelled,
+        })
+        .unwrap();
+    next_matching(&mut updates, |change| {
+        matches!(change, StateChange::UploadFinished { .. })
+    })
+    .await;
+    assert!(harness.mirror.snapshot().uploads.is_empty());
+}

macltok

Commit a319bae745d8

Commit `a319bae745d8`