diff --git a/crates/rustmail-tui/src/app.rs b/crates/rustmail-tui/src/app.rs index 4ccc8a6..03b2109 100644 --- a/crates/rustmail-tui/src/app.rs +++ b/crates/rustmail-tui/src/app.rs @@ -788,9 +788,7 @@ impl App { self.messages = resp.messages; self.total = resp.total; self.view_stale = pending.overflowed; - for delta in pending.events { - self.replay_delta(delta); - } + self.replay_deltas(pending.events); self.error = None; self.error_ticks = 0; let moved = self.reanchor_selection(anchor); @@ -1124,6 +1122,25 @@ impl App { self.messages.iter_mut().find(|m| m.id == id) } + /// Re-applies the deltas recorded during an in-flight fetch, in order, onto + /// the snapshot that fetch returned. + /// + /// Arrivals are announced in the order they commit, so every arrival up to + /// the last one the page lists is already in the snapshot, counted in its + /// total even when it sits past the page; only the arrivals after it are new. + fn replay_deltas(&mut self, deltas: Vec) { + let listed_through = deltas.iter().rposition( + |delta| matches!(delta, WsEvent::MessageNew(summary) if self.is_listed(&summary.id)), + ); + for (index, delta) in deltas.into_iter().enumerate() { + let in_snapshot = listed_through.is_some_and(|last| index <= last); + if in_snapshot && matches!(delta, WsEvent::MessageNew(_)) { + continue; + } + self.replay_delta(delta); + } + } + /// Re-applies a delta recorded during an in-flight fetch onto the snapshot /// that fetch returned. Selection is left alone: the caller re-resolves it /// by message id once every delta has been replayed. A delta whose effect @@ -1864,6 +1881,21 @@ mod tests { assert_eq!(app.total, 4); } + #[tokio::test] + async fn arrival_the_snapshot_holds_past_its_page_is_not_counted_again() { + let mut app = in_flight_fetch(0); + app.page_size = 2; + + for id in ["live-0", "live-1", "live-2"] { + app.handle_ws_message(&new_message_event(id)).await; + } + land_snapshot(&mut app, &["live-2", "live-1"], 3).await; + + let ids: Vec<&str> = app.messages.iter().map(|m| m.id.as_str()).collect(); + assert_eq!(ids, vec!["live-2", "live-1"]); + assert_eq!(app.total, 3); + } + #[tokio::test] async fn update_during_in_flight_fetch_is_replayed_onto_the_snapshot() { let mut app = in_flight_fetch(1); diff --git a/ui/src/stores/messages.test.ts b/ui/src/stores/messages.test.ts index e6c906a..ec93f8e 100644 --- a/ui/src/stores/messages.test.ts +++ b/ui/src/stores/messages.test.ts @@ -1411,6 +1411,21 @@ describe("arrivals while the reader is scrolled away", () => { expect(filteredMessages().map((m) => m.is_starred)).toEqual([false, true]); }); + it("lets in a match a held resync finds between loaded rows", async () => { + const starred = (n: number) => message(n, { is_starred: true }); + serve([starred(0), starred(2), starred(4)]); + toggleFilter("starred"); + await vi.waitFor(() => expect(loading()).toBe(false)); + setLiveHeld(true); + serve([starred(0), starred(1), starred(2), starred(4)]); + + await resync(); + + expect(ids()).toEqual(["id-0", "id-1", "id-2", "id-4"]); + expect(heldArrivals()).toBe(0); + expect(total()).toBe(4); + }); + it("reconciles loaded rows past the first page on a held resync", async () => { const rows = range(PAGE_SIZE + 50); serve(rows); @@ -1491,6 +1506,25 @@ describe("arrivals while the reader is scrolled away", () => { expect(ids()).toEqual(["id-0", "id-1"]); }); + it("counts arrivals a held resync finds past what it holds once", async () => { + setLiveHeld(true); + const arrivals = Array.from({ length: MAX_LIVE_ROWS + 1 }, (_, i) => + message(1000 + i), + ); + const rows = [...arrivals.toReversed(), ...range(2)]; + serve(rows); + + openSocket(); + for (const m of arrivals) { + receive(JSON.stringify({ type: "message:new", data: m })); + } + nextFrame(); + await vi.waitFor(() => expect(loading()).toBe(false)); + + expect(heldArrivals()).toBe(arrivals.length); + expect(total()).toBe(rows.length); + }); + it("puts the held arrivals on top once the reader is back there", () => { setLiveHeld(true); deliver(arrival(8)); @@ -1702,6 +1736,29 @@ describe("resync when the socket opens", () => { expect(total()).toBe(3); }); + it("counts an arrival the read holds past its first page once", async () => { + const read = deferred(); + listMessages.mockReturnValueOnce(read.promise); + connectWebSocket(); + openSocket(); + const arrivals = Array.from({ length: PAGE_SIZE + 1 }, (_, i) => + message(1000 + i), + ); + + for (const m of arrivals) { + receive(JSON.stringify({ type: "message:new", data: m })); + } + nextFrame(); + const newest = arrivals.slice(1).toReversed(); + read.resolve({ messages: newest, total: arrivals.length }); + + await vi.waitFor(() => expect(loading()).toBe(false)); + expect(filteredMessages().map((m) => m.id)).toEqual( + newest.map((m) => m.id), + ); + expect(total()).toBe(arrivals.length); + }); + it("keeps a flag change made while the read is in flight", async () => { const read = deferred(); listMessages.mockReturnValueOnce(read.promise); diff --git a/ui/src/stores/messages.ts b/ui/src/stores/messages.ts index 4e2b5d7..c14beff 100644 --- a/ui/src/stores/messages.ts +++ b/ui/src/stores/messages.ts @@ -828,8 +828,15 @@ async function fetchMessages(): Promise { await readFirstPage(); } -/** A list read that has landed: the count it read, and how to show it. */ -type LandedRead = { total: number; show: () => void }; +/** + * A list read that has landed: the count it read, the ids of the rows it + * returned, and how to show it. + */ +type LandedRead = { + total: number; + ids: ReadonlySet; + show: () => void; +}; async function readTopPage( view: View, @@ -841,6 +848,7 @@ async function readTopPage( ); return { total: res.total, + ids: new Set(res.messages.map((m) => m.id)), show: () => { replaceRows(res.messages); dropHeldArrivals(); @@ -854,9 +862,12 @@ async function readTopPage( * * This is the read while live mail is held: the reader is scrolled into these * rows, so replacing them with the first page would move the list under them. - * Rows above the first loaded one are arrivals and wait behind the pill. A - * loaded row the read meets takes the server's copy, and one it passes without - * meeting is gone from the view. The read stops at the loaded tail, at the end + * Rows above the first loaded one are arrivals and wait behind the pill. From + * the first loaded row the read meets to the last, the server's rows replace + * the loaded ones: a loaded row the read meets takes the server's copy, one it + * passes without meeting is gone from the view, and a row between them that + * was never loaded, such as one starred elsewhere under the starred filter, + * joins the list where it falls. The read stops at the loaded tail, at the end * of the list, after {@link PAGE_SIZE} rows past the last loaded one it met, * which is where a deleted tail leaves it, or once more has arrived than the * pill keeps, which only a fresh read on the way back to the top can show. @@ -901,6 +912,7 @@ async function readLoadedWindow( if (passedAll || outranLoaded) { return { total: res.total, + ids: new Set(read.map((m) => m.id)), show: () => reconcileWindow(read, passedAll, ended, view), }; } @@ -918,21 +930,14 @@ function reconcileWindow( ended: boolean, view: View, ): void { - const server = new Map(read.map((m) => [m.id, m])); const loaded = messages(); const firstLoaded = read.findIndex((m) => positions.has(m.id)); + const lastLoaded = read.findLastIndex((m) => positions.has(m.id)); const arrivals = firstLoaded < 0 ? read : read.slice(0, firstLoaded); - let passedUpTo = passedAll ? loaded.length - 1 : -1; - if (!passedAll) { - loaded.forEach((m, i) => { - if (server.has(m.id)) passedUpTo = i; - }); - } - const kept = loaded.flatMap((m, i) => { - const fresh = server.get(m.id); - if (fresh !== undefined) return [fresh]; - return i <= passedUpTo ? [] : [m]; - }); + const met = firstLoaded < 0 ? [] : read.slice(firstLoaded, lastLoaded + 1); + const lastMet = lastLoaded < 0 ? -1 : indexOfRow(read[lastLoaded].id); + const unmet = passedAll ? [] : loaded.slice(lastMet + 1); + const kept = [...met, ...unmet]; const last = kept.at(-1) ?? arrivals.at(-1); const lastRead = last === undefined ? -1 : read.findIndex((m) => m.id === last.id); @@ -972,7 +977,7 @@ async function readFirstPage(): Promise { landed.show(); setStoredTotal(landed.total); searchStale = false; - applyEvents(eventsDuringRead ?? []); + replayEvents(eventsDuringRead ?? [], landed.ids); }); if (raced) void refreshTotal(); return true; @@ -1241,6 +1246,27 @@ function applyEvents(events: readonly WsEvent[]): void { admitArrivals(arrivals); } +/** + * Applies the events that arrived while a list read was in flight over the + * list it returned. + * + * Arrivals are announced in the order they commit, so every arrival up to the + * last one among the rows the read returned, `read`, was already in the read, + * and counted in its total even when it sits past the rows it returned; only + * the arrivals after it are new. + */ +function replayEvents( + events: readonly ReplayableEvent[], + read: ReadonlySet, +): void { + const lastInRead = events.findLastIndex( + (event) => event.type === "message:new" && read.has(event.data.id), + ); + applyEvents( + events.filter((event, i) => i > lastInRead || event.type !== "message:new"), + ); +} + function applyEvent(event: Exclude): void { switch (event.type) { case "message:read":