macltok

public

Star0

Fork 0

Watch0

23 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `0b8ce8dc9a1c`

tarball

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

5 files changed+115-74

mirror: keep failures in the state and heal diffs that never saw a reset

A failure that only rode on an update was lost to any subscriber that joined or lagged after it, so the last engine failure and the last failed session save now live in State. The engine saves the record itself since oykxwysx, so the mirror only reflects SessionSaveFailed and clears it on the next SessionTokensChanged. A diff for a timeline with no reset is treated as a divergence, since a host wrapper that ate the reset would otherwise never heal.

Modified file: README.md+10 additions−11 deletions@@ -82,17 +82,16 @@ 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
+a batch that doesn't fit leaves the timeline diverged at its last good items,
+and so does a diff for a timeline it never saw reset. 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
+
+failures stay in the `State`, so a late subscriber sees them too. what stopped
+the engine sits in `engine_failure`, and a failed save of the session record in
+`session_save_failure` until the next one works. the mirror never touches the
+store
 
 ## wasm
 

Modified file: docs/design-decisions.md+3 additions−3 deletions@@ -243,9 +243,9 @@ 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
+most. a diff for a timeline with no reset is a divergence too, so a wrapper
+that swallowed the reset still heals. anything a late subscriber needs, failures
+included, lives in the state rather than only in an update
 
 ## public model
 

Modified file: src/mirror.rs+41 additions−40 deletions@@ -49,6 +49,10 @@ pub struct State {
     pub timelines: imbl::HashMap<TimelineId, TimelineState>,
     pub typing: imbl::HashMap<OwnedRoomId, Vec<OwnedUserId>>,
     pub uploads: imbl::HashMap<OwnedTransactionId, UploadState>,
+    /// what stopped the engine, if something went wrong
+    pub engine_failure: Option<EngineFailure>,
+    /// the last save of the session record that failed, until one succeeds
+    pub session_save_failure: Option<String>,
 }
 
 #[derive(Clone, Debug, PartialEq)]
@@ -111,8 +115,9 @@ pub enum StateChange {
     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>),
+    SessionSaveFailed(String),
+    /// saved again after `SessionSaveFailed`
+    SessionSaved,
     Connection(ConnectionState),
     Encryption(EncryptionState),
     Verification(VerificationState),
@@ -164,8 +169,6 @@ pub struct Mirror {
     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 {
@@ -183,7 +186,8 @@ impl Mirror {
         Self::with_capacity(handle, events, DEFAULT_CAPACITY)
     }
 
-    /// `capacity` updates can queue for a subscriber before it lags and starts over
+    /// `capacity` updates can queue for a subscriber before it lags and starts over. a zero
+    /// capacity counts as one
     pub fn with_capacity(
         handle: ClientHandle,
         events: impl Stream<Item = Event> + SendOutsideWasm + 'static,
@@ -200,12 +204,14 @@ impl Mirror {
             timelines: imbl::HashMap::new(),
             typing: imbl::HashMap::new(),
             uploads: imbl::HashMap::new(),
+            engine_failure: None,
+            session_save_failure: None,
         };
         let shared = Arc::new(Shared {
             published: Mutex::new(Published {
                 state: Arc::new(state),
                 sequence: 0,
-                updates: Some(broadcast::channel(capacity).0),
+                updates: Some(broadcast::channel(capacity.max(1)).0),
                 rooms_fetch: RoomsFetch::Idle,
             }),
         });
@@ -219,32 +225,21 @@ impl Mirror {
             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;
+                self.apply(event);
             }
         })
     }
 
-    async fn apply(&mut self, event: Event) {
+    fn apply(&mut self, event: Event) {
         match event {
             Event::ConnectionChanged(connection) => self.publish(|state| {
                 state.connection = connection;
@@ -254,10 +249,14 @@ impl Mirror {
                 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::SessionSaveFailed { message } => self.publish(|state| {
+                state.session_save_failure = Some(message.clone());
+                Some(StateChange::SessionSaveFailed(message))
+            }),
+            Event::SessionTokensChanged => self.publish(|state| {
+                state.session_save_failure.take()?;
+                Some(StateChange::SessionSaved)
+            }),
             Event::SessionAccessRejected { soft_logout } => self.publish(|state| {
                 state.session = match soft_logout {
                     Some(device_id) => SessionStatus::SoftLoggedOut { device_id },
@@ -279,7 +278,10 @@ impl Mirror {
                     Some(StateChange::Session(SessionStatus::Stopped))
                 });
             }
-            Event::Failed(failure) => self.publish(|_| Some(StateChange::Failed(failure))),
+            Event::Failed(failure) => self.publish(|state| {
+                state.engine_failure = Some(failure.clone());
+                Some(StateChange::Failed(failure))
+            }),
             Event::NotificationSettingsChanged => {
                 self.publish(|_| Some(StateChange::NotificationSettings));
             }
@@ -406,7 +408,21 @@ impl Mirror {
     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)?;
+            let Some(timeline) = state.timelines.get_mut(&timeline_id) else {
+                diverged = true;
+                state.timelines.insert(
+                    timeline_id.clone(),
+                    TimelineState {
+                        items: Vector::new(),
+                        history: HistoryState::Idle,
+                        diverged: true,
+                    },
+                );
+                return Some(StateChange::Timeline {
+                    timeline_id: timeline_id.clone(),
+                    change: TimelineChange::Diverged,
+                });
+            };
             if timeline.diverged {
                 return None;
             }
@@ -434,21 +450,6 @@ impl Mirror {
         }
     }
 
-    #[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 {

Modified file: src/model.rs+4 additions−3 deletions@@ -1324,17 +1324,18 @@ mod tests {
         let mut items = vector![item(1), item(2)];
         let batch = vec![
             TimelineDiff::PushBack { value: item(3) },
+            TimelineDiff::PushBack { value: item(4) },
             TimelineDiff::Remove { index: 0 },
             TimelineDiff::Set {
-                index: 2,
-                value: item(4),
+                index: 3,
+                value: item(5),
             },
         ];
         assert_eq!(
             TimelineDiff::try_apply_all(batch, &mut items)
                 .unwrap_err()
                 .length,
-            2
+            3
         );
         assert_eq!(items, vector![item(1), item(2)]);
 

Modified file: tests/mirror.rs+57 additions−17 deletions@@ -9,7 +9,10 @@ use macltok::{
     UploadOutcome,
 };
 use matrix_sdk::{
-    ruma::{OwnedRoomId, TransactionId, api::client::sync::sync_events::v5, event_id, room_id},
+    ruma::{
+        OwnedRoomId, OwnedTransactionId, TransactionId, api::client::sync::sync_events::v5,
+        event_id, room_id,
+    },
     test_utils::mocks::MatrixMockServer,
 };
 use tokio::sync::mpsc;
@@ -149,6 +152,30 @@ async fn diverged_batch_keeps_last_good_items_until_a_reset() {
 }
 
 #[tokio::test]
+async fn a_diff_for_a_timeline_without_a_reset_diverges_it() {
+    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(Event::TimelineChanged {
+            timeline_id: timeline_id.clone(),
+            diffs: vec![TimelineDiff::PushBack { value: item(1) }],
+        })
+        .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!(timeline.items.is_empty());
+}
+
+#[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());
@@ -183,25 +210,38 @@ async fn room_and_thread_timelines_stay_apart() {
     );
 }
 
-#[tokio::test]
-async fn late_subscriber_gets_state_then_every_update_after_it() {
-    let harness = start(EngineConfig::default(), 1024).await;
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn subscribers_joining_mid_stream_see_no_gap_and_nothing_twice() {
+    const CHUNKS: u64 = 64;
+    const CHUNK: u64 = 32;
+    let harness = start(EngineConfig::default(), 4096).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 followers = Vec::new();
+    for chunk in 0..CHUNKS {
+        for uploaded_bytes in chunk * CHUNK + 1..=(chunk + 1) * CHUNK {
+            harness
+                .inject
+                .send(Event::MediaUploadProgress {
+                    room_id: harness.room_id.clone(),
+                    transaction_id: transaction_id.clone(),
+                    uploaded_bytes,
+                    total_bytes: CHUNKS * CHUNK,
+                    complete: false,
+                })
+                .unwrap();
+        }
+        followers.push(tokio::spawn(follow(
+            harness.mirror.subscribe(),
+            transaction_id.clone(),
+            CHUNKS * CHUNK,
+        )));
     }
-    let mut updates = harness.mirror.subscribe();
-    for uploaded_bytes in 51..=100 {
-        harness.inject.send(progress(uploaded_bytes)).unwrap();
+    for follower in followers {
+        follower.await.unwrap();
     }
+}
 
+async fn follow(mut updates: Subscription, transaction_id: OwnedTransactionId, total: u64) {
     let first = updates.next().await.unwrap();
     let StateChange::State(state) = &first.change else {
         panic!("a subscription opened with {first:?}");
@@ -211,7 +251,7 @@ async fn late_subscriber_gets_state_then_every_update_after_it() {
         .get(&transaction_id)
         .map_or(0, |upload| upload.uploaded_bytes);
     let mut sequence = first.sequence;
-    while seen < 100 {
+    while seen < total {
         let update = next_matching(&mut updates, |_| true).await;
         assert_eq!(update.sequence, sequence + 1);
         sequence = update.sequence;

macltok

Commit 0b8ce8dc9a1c

Commit `0b8ce8dc9a1c`