atagen

macltok

public
Star0
Watch0

Commit 002c79e4c6ec

tarball

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

7 files changed+80-222
history: fence every committed completion and drop the joined gap filter

A fill or recent load that committed pages and then failed or was cancelled answered before those pages reached the timeline. The joined gap filter guarded a case the sdk never produces. RetryRecent's fallback pagination reached further back on every retry, so it now only runs on an empty room timeline.
Modified file: docs/segmented-history.md+11 additions−13 deletions
@@ -19,9 +19,10 @@ the fresh window and what was retained.
1919 - `Loaded`, `Cancelled` or `Failed(message)` as it settles
2020 - `Idle` when the room closes, whatever was happening
2121
22-`Failed` stays until `HistoryRepairCommand::RetryRecent` or a reopen. a failed
23-load falls back to paginating `EngineConfig::initial_history_limit` events, so
24-the room isn't left empty.
22+`Failed` stays until `HistoryRepairCommand::RetryRecent`, and closing the room
23+turns it to `Idle` like anything else. a failed load on a room timeline with no
24+events yet falls back to paginating `EngineConfig::initial_history_limit`
25+events, so the room isn't left empty.
2526
2627 ## filling a gap
2728
@@ -55,20 +56,17 @@ gaps appear in `TimelineState.items` as `TimelineItem::HistoryGap`, between the
5556 sections they separate. a row keeps its id while its lifecycle changes, and is
5657 removed once the gap joins. joined gaps never appear.
5758
58-history lands as incremental `TimelineChanged` diffs, never a `TimelineReset`.
59-before answering, the room fences on `Timeline::flush_history`, which applies
60-what the cache already queued without replacing the timeline, so receipts, the
61-read marker and row ids survive.
59+history lands as `TimelineChanged` diffs, never a `TimelineReset`. they are
60+incremental unless the cache's loaded window changed under the operation, after
61+a restart or a shrink, when the sdk sends the room as one `Reset` diff. before
62+answering, the room fences on `Timeline::flush_history`, which applies what the
63+cache already queued without replacing the timeline, so receipts, the read
64+marker and row ids survive. a failed or cancelled operation that committed
65+pages is fenced the same way.
6266
6367 a command's answer comes after its outcome is published, and after the mirror
6468 has applied it when one is attached.
6569
66-## known holes
67-
68-a limited sync for a room with no events in memory records a plain gap instead
69-of a history gap (review 2026-10-25 S1), so an interval can still render as
70-continuous. both `prev_batch` drops in the sdk's `handle_sync` remain.
71-
7270 ## wire shapes
7371
7472 serde enums are externally tagged.
Modified file: src/engine/repair.rs+6 additions−2 deletions
@@ -293,8 +293,12 @@ async fn fill_gap(
293293 let next = tokio::select! {
294294 biased;
295295 _ = cancel.changed() => return Ok(Stopped::Cancelled),
296- result = cache.fill_history_gap_once(report.gap_id.clone()) => {
297- result.map_err(cache_error)?
296+ result = cache.fill_history_gap_once(report.gap_id.clone()) => match result {
297+ Ok(next) => next,
298+ Err(error) => {
299+ committed |= matches!(error, EventCacheError::PaginationError(_));
300+ return Err(cache_error(error));
301+ }
298302 }
299303 };
300304 let (gap, events) = match next {
Modified file: src/engine/room.rs+6 additions−2 deletions
@@ -157,12 +157,15 @@ impl RoomActor {
157157
158158 async fn finish_repair(&self, mut completion: Completion, timelines: &mut Timelines) {
159159 let mut cancellation = Ok(());
160+ // pages a failed or cancelled operation already committed still have to be on
161+ // screen before anyone hears it stopped
160162 if completion.committed
161163 && !completion.closing
162- && completion.result.is_ok()
163164 && let Err(error) = timelines.room.publish_history().await
164165 {
165- completion.unpublished(error.clone());
166+ if completion.result.is_ok() {
167+ completion.unpublished(error.clone());
168+ }
166169 cancellation = Err(error);
167170 }
168171 let event = match &completion.outcome {
@@ -183,6 +186,7 @@ impl RoomActor {
183186 completion.outcome,
184187 Outcome::Recent(RecentHistoryState::Failed(_))
185188 ) && limit > 0
189+ && !timelines.room.has_events().await
186190 {
187191 timelines.room.paginate(limit, None);
188192 }
Modified file: src/engine/timeline.rs+11 additions−8 deletions
@@ -144,6 +144,14 @@ impl TimelineSession {
144144 drained.await.map_err(|_| Error::interrupted())
145145 }
146146
147+ pub(super) async fn has_events(&self) -> bool {
148+ self.timeline
149+ .items()
150+ .await
151+ .iter()
152+ .any(|item| item.as_event().is_some())
153+ }
154+
147155 pub(super) fn paginate(&self, limit: u16, reply: Option<Responder>) {
148156 if let Err(error) = self.pagination.try_send(PaginationRequest { limit, reply }) {
149157 let (request, error) = match error {
@@ -415,8 +423,7 @@ async fn start_watcher(
415423 ) -> Result<(JoinHandle<()>, mpsc::UnboundedSender<Fence>), SendEventError> {
416424 let (items, mut stream) = timeline.subscribe().await;
417425 let current_user_id = context.current_user_id.clone();
418- let (normalised, mut hidden) =
419- normalise_timeline(items.iter(), current_user_id.as_deref(), scope);
426+ let normalised = normalise_timeline(items.iter(), current_user_id.as_deref(), scope);
420427 context.media.register(scope, normalised.media).await;
421428 context
422429 .events
@@ -456,12 +463,8 @@ async fn start_watcher(
456463 Some(_) = replies.fetches.join_next(), if !replies.fetches.is_empty() => continue,
457464 };
458465 for diffs in diffs {
459- let normalised = normalise_timeline_diffs(
460- diffs,
461- current_user_id.as_deref(),
462- &scope,
463- &mut hidden,
464- );
466+ let normalised =
467+ normalise_timeline_diffs(diffs, current_user_id.as_deref(), &scope);
465468 media.register(&scope, normalised.media).await;
466469 replies.fetch(normalised.missing_replies);
467470 if normalised.diffs.is_empty() {
Modified file: src/model.rs+1 additions−1 deletions
@@ -2131,7 +2131,7 @@ pub enum RecentHistoryState {
21312131 Loading,
21322132 Loaded,
21332133 Cancelled,
2134- /// stays until a retry or the room is reopened
2134+ /// stays until a retry, or until closing the room makes it `Idle`
21352135 Failed(String),
21362136 }
21372137
Modified file: src/normalise/timeline.rs+26 additions−195 deletions
@@ -52,16 +52,10 @@ struct Normaliser<'a> {
5252 }
5353
5454 impl Normaliser<'_> {
55- /// `None` for rows macltok leaves out
56- fn item(&mut self, item: &SdkTimelineItem) -> Option<TimelineItem> {
57- if let TimelineItemKind::Virtual(VirtualTimelineItem::HistoryGap(gap)) = item.kind()
58- && gap.status == matrix_sdk::event_cache::HistoryGapStatus::Joined
59- {
60- return None;
61- }
55+ fn item(&mut self, item: &SdkTimelineItem) -> TimelineItem {
6256 self.media
6357 .extend(media_registrations_for_item(item, self.scope));
64- Some(match item.kind() {
58+ match item.kind() {
6559 TimelineItemKind::Event(event) => {
6660 TimelineItem::Event(Box::new(self.event(item, event)))
6761 }
@@ -77,6 +71,7 @@ impl Normaliser<'_> {
7771 TimelineItemKind::Virtual(VirtualTimelineItem::HistoryGap(gap)) => {
7872 use matrix_sdk::event_cache::HistoryGapStatus;
7973 let (lifecycle, failure) = match &gap.status {
74+ // the sdk drops joined gaps before they reach a timeline
8075 HistoryGapStatus::Idle | HistoryGapStatus::Joined => {
8176 (HistoryGapLifecycle::Idle, None)
8277 }
@@ -94,7 +89,7 @@ impl Normaliser<'_> {
9489 failure,
9590 })
9691 }
97- })
92+ }
9893 }
9994
10095 fn event(&mut self, item: &SdkTimelineItem, event: &EventTimelineItem) -> TimelineEvent {
@@ -154,28 +149,25 @@ pub(crate) fn normalise_timeline<'a>(
154149 items: impl Iterator<Item = &'a Arc<SdkTimelineItem>>,
155150 current_user_id: Option<&UserId>,
156151 scope: &MediaScope,
157-) -> (NormalisedTimeline, HiddenRows) {
152+) -> NormalisedTimeline {
158153 let mut normaliser = Normaliser {
159154 current_user_id,
160155 scope,
161156 media: Vec::new(),
162157 missing_replies: Vec::new(),
163158 };
164- let mut hidden = HiddenRows::default();
165- let items = hidden.reset(items.map(|item| normaliser.item(item)));
166- let normalised = NormalisedTimeline {
167- items: items.into_iter().collect(),
159+ let items = items.map(|item| normaliser.item(item)).collect();
160+ NormalisedTimeline {
161+ items,
168162 media: normaliser.media,
169163 missing_replies: normaliser.missing_replies,
170- };
171- (normalised, hidden)
164+ }
172165 }
173166
174167 pub(crate) fn normalise_timeline_diffs(
175168 diffs: Vec<VectorDiff<Arc<SdkTimelineItem>>>,
176169 current_user_id: Option<&UserId>,
177170 scope: &MediaScope,
178- hidden: &mut HiddenRows,
179171 ) -> NormalisedTimelineDiffs {
180172 let mut normaliser = Normaliser {
181173 current_user_id,
@@ -185,7 +177,23 @@ pub(crate) fn normalise_timeline_diffs(
185177 };
186178 let diffs = diffs
187179 .into_iter()
188- .filter_map(|diff| hidden.translate(diff.map(|item| normaliser.item(&item))))
180+ .map(|diff| match diff.map(|item| normaliser.item(&item)) {
181+ VectorDiff::Append { values } => TimelineDiff::Append {
182+ values: values.into_iter().collect(),
183+ },
184+ VectorDiff::Clear => TimelineDiff::Clear,
185+ VectorDiff::PushFront { value } => TimelineDiff::PushFront { value },
186+ VectorDiff::PushBack { value } => TimelineDiff::PushBack { value },
187+ VectorDiff::PopFront => TimelineDiff::PopFront,
188+ VectorDiff::PopBack => TimelineDiff::PopBack,
189+ VectorDiff::Insert { index, value } => TimelineDiff::Insert { index, value },
190+ VectorDiff::Set { index, value } => TimelineDiff::Set { index, value },
191+ VectorDiff::Remove { index } => TimelineDiff::Remove { index },
192+ VectorDiff::Truncate { length } => TimelineDiff::Truncate { length },
193+ VectorDiff::Reset { values } => TimelineDiff::Reset {
194+ values: values.into_iter().collect(),
195+ },
196+ })
189197 .collect();
190198 NormalisedTimelineDiffs {
191199 diffs,
@@ -194,135 +202,6 @@ pub(crate) fn normalise_timeline_diffs(
194202 }
195203 }
196204
197-/// which sdk rows were left out, so diff indices can step over them
198-#[derive(Default)]
199-pub(crate) struct HiddenRows {
200- rows: imbl::Vector<bool>,
201- hidden: usize,
202-}
203-
204-impl HiddenRows {
205- fn visible_index(&self, index: usize) -> usize {
206- if self.hidden == 0 {
207- return index;
208- }
209- self.rows
210- .iter()
211- .take(index)
212- .filter(|hidden| !**hidden)
213- .count()
214- }
215-
216- fn push(&mut self, front: bool, item: &Option<TimelineItem>) {
217- self.hidden += usize::from(item.is_none());
218- if front {
219- self.rows.push_front(item.is_none());
220- } else {
221- self.rows.push_back(item.is_none());
222- }
223- }
224-
225- fn forget(&mut self, hidden: bool) -> bool {
226- self.hidden -= usize::from(hidden);
227- !hidden
228- }
229-
230- fn reset(&mut self, items: impl Iterator<Item = Option<TimelineItem>>) -> Vec<TimelineItem> {
231- *self = Self::default();
232- items
233- .filter_map(|item| {
234- self.push(false, &item);
235- item
236- })
237- .collect()
238- }
239-
240- fn translate(&mut self, diff: VectorDiff<Option<TimelineItem>>) -> Option<TimelineDiff> {
241- Some(match diff {
242- VectorDiff::Append { values } => {
243- let values: Vec<_> = values
244- .into_iter()
245- .filter_map(|item| {
246- self.push(false, &item);
247- item
248- })
249- .collect();
250- if values.is_empty() {
251- return None;
252- }
253- TimelineDiff::Append { values }
254- }
255- VectorDiff::Clear => {
256- *self = Self::default();
257- TimelineDiff::Clear
258- }
259- VectorDiff::PushFront { value } => {
260- self.push(true, &value);
261- TimelineDiff::PushFront { value: value? }
262- }
263- VectorDiff::PushBack { value } => {
264- self.push(false, &value);
265- TimelineDiff::PushBack { value: value? }
266- }
267- VectorDiff::PopFront => {
268- let hidden = self.rows.pop_front()?;
269- self.forget(hidden).then_some(TimelineDiff::PopFront)?
270- }
271- VectorDiff::PopBack => {
272- let hidden = self.rows.pop_back()?;
273- self.forget(hidden).then_some(TimelineDiff::PopBack)?
274- }
275- VectorDiff::Insert { index, value } => {
276- let visible = self.visible_index(index);
277- self.hidden += usize::from(value.is_none());
278- self.rows.insert(index, value.is_none());
279- TimelineDiff::Insert {
280- index: visible,
281- value: value?,
282- }
283- }
284- VectorDiff::Set { index, value } => {
285- let visible = self.visible_index(index);
286- let was_hidden = self.rows.set(index, value.is_none());
287- self.hidden = self.hidden + usize::from(value.is_none()) - usize::from(was_hidden);
288- match (was_hidden, value) {
289- (false, Some(value)) => TimelineDiff::Set {
290- index: visible,
291- value,
292- },
293- (false, None) => TimelineDiff::Remove { index: visible },
294- (true, Some(value)) => TimelineDiff::Insert {
295- index: visible,
296- value,
297- },
298- (true, None) => return None,
299- }
300- }
301- VectorDiff::Remove { index } => {
302- let visible = self.visible_index(index);
303- let hidden = self.rows.remove(index);
304- self.forget(hidden)
305- .then_some(TimelineDiff::Remove { index: visible })?
306- }
307- VectorDiff::Truncate { length } => {
308- let visible = self.visible_index(length);
309- let dropped = self
310- .rows
311- .iter()
312- .skip(length)
313- .filter(|hidden| **hidden)
314- .count();
315- self.hidden -= dropped;
316- self.rows.truncate(length);
317- TimelineDiff::Truncate { length: visible }
318- }
319- VectorDiff::Reset { values } => TimelineDiff::Reset {
320- values: self.reset(values.into_iter()),
321- },
322- })
323- }
324-}
325-
326205 /// `media_identity` is `None` for embedded reply content, which never gets fetchable media of its own
327206 fn content(content: &TimelineItemContent, media_identity: Option<&str>) -> TimelineContent {
328207 match content {
@@ -593,51 +472,3 @@ pub(super) fn formatted_body(message: &MessageType) -> Option<String> {
593472 (formatted.format == MessageFormat::Html).then(|| formatted.body.clone())
594473 }
595474
596-#[cfg(test)]
597-mod tests {
598- use super::*;
599-
600- fn row(n: u64) -> Option<TimelineItem> {
601- Some(TimelineItem::DayDivider { timestamp_ms: n })
602- }
603-
604- #[test]
605- fn hidden_rows_keep_diff_indices_on_the_visible_timeline() {
606- let mut sdk: imbl::Vector<Option<TimelineItem>> = imbl::Vector::new();
607- let mut hidden = HiddenRows::default();
608- let mut visible: imbl::Vector<TimelineItem> = imbl::Vector::new();
609- let diffs = [
610- VectorDiff::Reset {
611- values: [row(0), None, row(1)].into_iter().collect(),
612- },
613- VectorDiff::Insert {
614- index: 1,
615- value: row(2),
616- },
617- VectorDiff::Set {
618- index: 2,
619- value: row(3),
620- },
621- VectorDiff::Set {
622- index: 0,
623- value: None,
624- },
625- VectorDiff::PushFront { value: None },
626- VectorDiff::Remove { index: 3 },
627- VectorDiff::Append {
628- values: [None, row(4)].into_iter().collect(),
629- },
630- VectorDiff::PopFront,
631- VectorDiff::Truncate { length: 2 },
632- VectorDiff::PopBack,
633- ];
634- for diff in diffs {
635- diff.clone().apply(&mut sdk);
636- if let Some(diff) = hidden.translate(diff) {
637- diff.try_apply(&mut visible).unwrap();
638- }
639- let expected: imbl::Vector<_> = sdk.iter().flatten().cloned().collect();
640- assert_eq!(visible, expected);
641- }
642- }
643-}
Modified file: tests/history_repair.rs+19 additions−1 deletions
@@ -7,7 +7,7 @@ use macltok::Mirror;
77 use macltok::{
88 ClientHandle, Engine, EngineConfig, EngineError, Event, EventStream, HistoryGap,
99 HistoryRepairPhase, HistoryRepairReport, HistoryRepairStop, RecentHistoryState, Termination,
10- TimelineId, TimelineItem,
10+ TimelineDiff, TimelineId, TimelineItem,
1111 };
1212 use matrix_sdk::{
1313 Client,
@@ -74,6 +74,10 @@ async fn next_until(room: &mut TestRoom, mut predicate: impl FnMut(&Event) -> bo
7474 diffs,
7575 } => {
7676 for diff in diffs.clone() {
77+ assert!(
78+ !matches!(diff, TimelineDiff::Reset { .. }),
79+ "history replaced the room timeline"
80+ );
7781 diff.try_apply(&mut room.items)
7882 .expect("diff did not fit the timeline");
7983 }
@@ -577,6 +581,20 @@ async fn closing_during_a_recent_load_settles_idle_not_failed() {
577581 .mount()
578582 .await;
579583 open_room(&mut room, room_id).await;
584+ tokio::time::timeout(Duration::from_secs(5), async {
585+ while !room
586+ .server
587+ .received_requests()
588+ .await
589+ .unwrap()
590+ .iter()
591+ .any(|request| request.url.path().ends_with("/messages"))
592+ {
593+ tokio::task::yield_now().await;
594+ }
595+ })
596+ .await
597+ .expect("the recent load never asked for messages");
580598 room.handle.close_room(room_id.to_owned()).await.unwrap();
581599 let mut states = Vec::new();
582600 next_until(&mut room, |event| {