From b76ba40e04373bf40f470952ac6de822467ef786 Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Tue, 10 Nov 2026 13:18:49 +1100 Subject: [PATCH] fix(messages): end the New messages hold when no repair can sample Follow-up to #3071, which made a channel longer than one page wait for the unread store's repair before placing "New messages", holding the channel's own reading meanwhile. The hold keyed on `sampled` alone. A repair that failed, overflowed its evidence budget, or was abandoned by a dropped connection or an access change left `sampled` unset with `requested` already true, so `ensure()` started nothing. Once later live evidence made the store observed again, the timeline held reading until some unrelated repair succeeded. Channel snapshots now report `sampling` while a repair that can set `sampled` is reading, about to restart, or not yet requested. The timeline holds only while it is set. Otherwise the visit settles with no line and reading resumes, including when the channel is entered after such a repair has already ended. The line inside incomplete history needed a read row above it, but any read row counted, including one read only by its own `msg:` receipt (dwell short of the bottom), which says nothing about the rows above it. `attention()` now reports `prefixRead` when the channel frontier, the `activity:` catch-up mark (ordinary rows) or the retention floor reaches the row, and only such a row lets the line sit below it. A channel read part-way by scrolling now shows no line on return. Unread mentions, broadcasts, broadcast replies and marked-unread rows above the window remain the documented residual. docs/unread.md records both. docs/buzz-help.md scopes the sentence about waiting to mark anything read to the repair check, and says the line is absent when the rows above were only read one by one. Each new regression test fails without its fix: the hold after a failed repair that becomes observed again, entry after a failed or overflowed repair, the receipt-only witness, and the store's `sampling` and `prefixRead` contracts, including the restart gap. Signed-off-by: Matt Toohey Co-Authored-By: Claude Opus 5.5 --- docs/buzz-help.md | 14 +- docs/unread.md | 37 +++-- .../messages/ChannelTimeline.restore.test.tsx | 143 ++++++++++++++---- src/features/messages/ChannelTimeline.tsx | 24 +-- src/features/relay/unread-startup.test.ts | 73 +++++++++ src/features/relay/unread.test.ts | 64 +++++++- src/features/relay/unread.ts | 32 +++- 7 files changed, 323 insertions(+), 64 deletions(-) diff --git a/docs/buzz-help.md b/docs/buzz-help.md index 8a2bd22326..bdd9405e72 100644 --- a/docs/buzz-help.md +++ b/docs/buzz-help.md @@ -409,11 +409,15 @@ revokes all of them, and a pairing unused for 30 days expires. you enter. Once established, the divider stays in place for that visit as you read, reveal another message, or receive new messages. In a channel with more history than loads on entry, it appears when the oldest unread message is - among the rows loaded on entry, below a message you have read; when the unread - messages start at the first loaded row or further up, there is no divider for - that visit, even if older rows load later. The divider is settled before the - channel marks anything read, so opening a channel does not hide its own - divider. It may also be absent when unread evidence is incomplete, or when + among the rows loaded on entry, below a message you read by reaching the + bottom of the channel or marking the channel read. When the unread messages + start at the first loaded row or further up, or the messages above them were + only read one by one as you scrolled past, there is no divider for that visit, + even if older rows load later. While Buzz is still checking such a channel's + older history for unread messages, the channel waits to mark anything read, + so opening it then does not hide its own divider; if that check fails, there + is no divider for that visit and reading resumes. It may also be absent when + unread evidence is incomplete, or when read intent or the read evidence needed to locate the boundary changes elsewhere before it is established. - **Opening a channel or DM** while Buzz is visible and focused clears its sidebar row: the bold name, the mention diff --git a/docs/unread.md b/docs/unread.md index 43d39181df..faa169a4be 100644 --- a/docs/unread.md +++ b/docs/unread.md @@ -142,21 +142,28 @@ viewport): a channel entered that scope: evidence then reaches past what windows loaded, up to each batch's 500-row limit. A cache clear or newly accessible channel starts that repair at once; an overflow waits for the next recovery tick. - Later repairs, refreshes included, leave `sampled` set. When entry did not - capture a verified, idle window with all of its history (rows sit above it, or - it was cached, refreshing or still paging), the **New messages** divider waits - for `sampled`, then appears only at `oldestUnreadId` among the rows loaded on - entry and only below a read one. While it waits, the timeline holds its own - reading (dwell and bottom catch-up), which would otherwise read the loaded - rows and erase the evidence it waits for; the hold ends when the repair - finishes or fails, and reading then starts at once. A batch that filled its - limit can still stop short of the true oldest unread message, and the - read-row condition covers only ordinary rows: catch-up's `activity:` mark - reads all of them up to a timestamp, but never mentions, broadcasts, - broadcast replies or messages marked unread, which only their own `msg:` or - `thread:` marks read. A short sample can hide an unread one of those above - the window, and the divider then lands below it. A row above read only by - its own mark says nothing about older rows either. + Later repairs, refreshes included, leave `sampled` set. Until it is set, + `sampling` says a repair that can set it is reading, about to restart, or not + yet asked for; one that fails, overflows or is abandoned (a dropped + connection, an access change) without a restart clears it until the next + starts. When entry did not capture a verified, idle window with all of its + history (rows sit above it, or it was cached, refreshing or still paging), the + **New messages** divider waits for `sampled`, then appears only at + `oldestUnreadId` among the rows loaded on entry and only below one that a + channel-wide read reaches (`attention()` reports `prefixRead`: the channel + frontier, its `activity:` catch-up mark or the retention floor), because each + of those also reads every older ordinary row. While it waits with `sampling` + set, the timeline holds its own reading (dwell and bottom catch-up), which + would otherwise read the loaded rows and erase the evidence it waits for. Once + `sampled` is set the divider is placed and reading starts at once; if + `sampling` clears first, the visit settles with no divider and reading resumes + without waiting for a retry. A batch that filled its limit can still stop + short of the true oldest unread message. `prefixRead` rules that out for + ordinary rows, but not for mentions, broadcasts, broadcast replies or messages + marked unread, which catch-up's `activity:` mark never reads: a short sample + can hide an unread one of those above the window, and the divider then lands + below it. Rows above read only by their own `msg:` or `thread:` marks, as + dwell short of the bottom leaves them, place no divider. - Both keys use verified event timestamps, never wall time. Neither enters generic channel/thread inheritance or remote override baselines. Channel catch-up ends a local manual unread on the channel itself; message and thread manual-unread diff --git a/src/features/messages/ChannelTimeline.restore.test.tsx b/src/features/messages/ChannelTimeline.restore.test.tsx index ca87cd4d71..794e4cc7e2 100644 --- a/src/features/messages/ChannelTimeline.restore.test.tsx +++ b/src/features/messages/ChannelTimeline.restore.test.tsx @@ -271,6 +271,8 @@ function boundaryFixture({ reading = false } = {}) { unknownIds = new Set(), pendingIds = new Set(), ineligibleIds = new Set(), + // Read only by their own receipt, not by a channel-wide read. + receiptIds = new Set(), listeners = new Set<() => void>(); const syncListeners = new Set<() => void>(); let readRevision = owner.session.unread.revision(); @@ -313,6 +315,8 @@ function boundaryFixture({ reading = false } = {}) { const built: ChannelMessage[] = []; // The store's repair has finished, so that evidence reaches past the window. let sampled = true; + // Unsampled: a repair that can still sample is running. + let repairing = true; let freshness: "observed" | "stale" = "observed"; let error: string | undefined; const snapshot = (channelId: string) => { @@ -330,7 +334,11 @@ function boundaryFixture({ reading = false } = {}) { return { target: { kind: "channel" as const, channelId }, ...(oldest ? { oldestUnreadId: oldest.id } : {}), - ...(sampled ? { sampled: true as const } : {}), + ...(sampled + ? { sampled: true as const } + : repairing + ? { sampling: true as const } + : {}), observedCount: unreadIds.size, attentionCount: 0, coverage: "observed" as const, @@ -383,6 +391,9 @@ function boundaryFixture({ reading = false } = {}) { unread: unreadIds.has(id), forced: forcedIds.has(id), viewing: false, + ...(!unreadIds.has(id) && !receiptIds.has(id) + ? { prefixRead: true as const } + : {}), }; }, revision: () => readRevision, @@ -488,6 +499,7 @@ function boundaryFixture({ reading = false } = {}) { unknownIds, pendingIds, ineligibleIds, + receiptIds, enterChannel, subscribers: () => ({ unread: listeners.size, sync: syncListeners.size }), completeReadStateSnapshot, @@ -520,6 +532,10 @@ function boundaryFixture({ reading = false } = {}) { setSampled(value: boolean) { sampled = value; }, + /** Whether an unsampled store still has a repair that can sample. */ + setRepairing(value: boolean) { + repairing = value; + }, setFreshness(value: "observed" | "stale", cause?: string) { freshness = value; error = cause; @@ -710,11 +726,11 @@ it.each([{ olderUnread: false }, { olderUnread: true }])( "waits for the unread repair before placing a boundary inside incomplete history: %j", async ({ olderUnread }) => { // Before the repair returns, the store knows only the loaded rows: its - // oldest unread message sits in the window below a row read by its own - // receipt, whatever lies above the window. + // oldest unread message sits in the window below a row catch-up read, + // whatever lies above the window. const h = boundaryFixture(); h.setSampled(false); - const read = h.row(message(h.peer, "c", "Read by its own receipt", 2)); + const read = h.row(message(h.peer, "c", "Read by catch-up", 2)); const first = h.row(message(h.peer, "c", "First loaded unread", 3)); const later = h.row(message(h.peer, "c", "Later loaded unread", 4)); h.unreadIds.add(first.id); @@ -740,26 +756,33 @@ it.each([{ olderUnread: false }, { olderUnread: true }])( screen.queryByRole("separator", { name: "New messages" }), ).toBeNull(); } else { - // The store cannot tell this from a sample that stopped short of older - // unread rows; docs/unread.md records that residual. + // Catch-up read every ordinary row above the read one, but a sample + // that stopped short could still miss an unread mention, broadcast or + // marked message there; docs/unread.md records that residual. expect(precedes(divider(), screen.getByText("First loaded unread"))).toBe( true, ); - expect( - precedes(screen.getByText("Read by its own receipt"), divider()), - ).toBe(true); + expect(precedes(screen.getByText("Read by catch-up"), divider())).toBe( + true, + ); } mounted.unmount(); }, ); -/** Rows above the window, the store's repair unfinished: the oldest unread - * row it knows sits below a receipt-read row, and the reader is at the bottom. */ -async function heldReader() { +/** Rows above the window, the store unsampled: the oldest unread row it knows + * sits below a row catch-up read, and the reader is at the bottom. Without + * `failed` its repair is still running; with it, a repair already ended + * without sampling and live evidence has since made the store observed. */ +async function heldReader({ failed }: { failed?: string } = {}) { vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); const h = boundaryFixture({ reading: true }); h.setSampled(false); - const read = h.row(message(h.peer, "c", "Read by its own receipt", 2)); + if (failed) { + h.setRepairing(false); + h.setFreshness("observed", failed); + } + const read = h.row(message(h.peer, "c", "Read by catch-up", 2)); const first = h.row(message(h.peer, "c", "First loaded unread", 3)); const later = h.row(message(h.peer, "c", "Later loaded unread", 4)); h.unreadIds.add(first.id); @@ -790,9 +813,9 @@ it("holds this list's reading while the boundary waits for the unread repair, th expect(precedes(divider(), screen.getByText("First loaded unread"))).toBe( true, ); - expect( - precedes(screen.getByText("Read by its own receipt"), divider()), - ).toBe(true); + expect(precedes(screen.getByText("Read by catch-up"), divider())).toBe( + true, + ); expect(h.catchUp).not.toHaveBeenCalled(); // The hold lifting starts dwell on its own. @@ -808,22 +831,82 @@ it("holds this list's reading while the boundary waits for the unread repair, th } }); -it("resumes this list's reading with no boundary when the unread repair fails", async () => { - const { h, mounted, later } = await heldReader(); - try { - expect(h.catchUp).not.toHaveBeenCalled(); - h.setFreshness("stale", "Unread observation failed"); +it.each([{ observedAgain: false }, { observedAgain: true }])( + "resumes this list's reading with no boundary when the unread repair fails: %j", + async ({ observedAgain }) => { + const { h, mounted, later } = await heldReader(); + try { + expect(h.catchUp).not.toHaveBeenCalled(); + h.setRepairing(false); + h.setFreshness("stale", "Unread observation failed"); + h.publish(); + // A window read or live message marks the evidence observed again + // before any dwell, with the error kept and nothing sampled. Dwelling + // while stale would instead settle the visit through its own read. + if (observedAgain) { + h.setFreshness("observed", "Unread observation failed"); + h.publish(); + } + await h.dwell(); + expect(h.catchUp).toHaveBeenCalledWith(later.id, undefined); + expect( + screen.queryByRole("separator", { name: "New messages" }), + ).toBeNull(); + } finally { + mounted.unmount(); + vi.useRealTimers(); + } + }, +); + +it.each([ + "Unread observation failed", + "Unread observation capacity reached; refresh available", +])( + "reads at once with no boundary when entered after a repair ended without sampling: %s", + async (failed) => { + const { h, mounted, later } = await heldReader({ failed }); + try { + // No repair is coming, so waiting would hold reading until a retry. + expect(h.catchUp).toHaveBeenCalledWith(later.id, undefined); + expect(h.ensure).not.toHaveBeenCalled(); + expect( + screen.queryByRole("separator", { name: "New messages" }), + ).toBeNull(); + } finally { + mounted.unmount(); + vi.useRealTimers(); + } + }, +); + +it.each([{ receipt: false }, { receipt: true }])( + "places a boundary inside incomplete history only below a row a channel-wide read reaches: %j", + async ({ receipt }) => { + // A row read only by its own receipt says nothing about the ordinary rows + // above it, which a sample that stopped short may not have seen. + const h = boundaryFixture(); + const read = h.row(message(h.peer, "c", "Read in loaded window", 2)); + const first = h.row(message(h.peer, "c", "First loaded unread", 3)); + const later = h.row(message(h.peer, "c", "Later loaded unread", 4)); + if (receipt) h.receiptIds.add(read.id); + h.unreadIds.add(first.id); + h.unreadIds.add(later.id); + h.completeReadStateSnapshot(); h.publish(); - await h.dwell(); - expect(h.catchUp).toHaveBeenCalledWith(later.id, undefined); - expect( - screen.queryByRole("separator", { name: "New messages" }), - ).toBeNull(); - } finally { + const mounted = render(h.tree("c", [read, first, later], true)); + await h.completeEntry(); + if (receipt) + expect( + screen.queryByRole("separator", { name: "New messages" }), + ).toBeNull(); + else + expect(precedes(divider(), screen.getByText("First loaded unread"))).toBe( + true, + ); mounted.unmount(); - vi.useRealTimers(); - } -}); + }, +); it("a read-intent change from elsewhere during the held wait still settles with no boundary and releases reading", async () => { const { h, mounted, later } = await heldReader(); diff --git a/src/features/messages/ChannelTimeline.tsx b/src/features/messages/ChannelTimeline.tsx index 8ae3bb5462..86c256b604 100644 --- a/src/features/messages/ChannelTimeline.tsx +++ b/src/features/messages/ChannelTimeline.tsx @@ -470,21 +470,25 @@ function Timeline({ // history it may know only the loaded rows, and its oldest unread one // can sit in the window below any read row while older ones are unread. // This is the only gate that holds this list's own reading: every - // earlier one has passed, so the repair alone is awaited. + // earlier one has passed, so the repair alone is awaited. A repair that + // ended without sampling (failed, overflowed, abandoned) settles the + // visit with no line, so reading never waits for a later retry. if (!unreadSnapshot.sampled) - return { ready: false, awaitingRepair: true }; + return unreadSnapshot.sampling + ? { ready: false, awaitingRepair: true } + : { ready: true }; const oldest = unreadSnapshot.oldestUnreadId; if (!oldest || !baseline.messageIds.has(oldest)) return { ready: true }; } if (!chronologicalRows.length) return { ready: noUnreadEvidence }; // A repair batch that fills its limit can stop inside the window, leaving - // the first loaded row the oldest unread message the store knows. Requiring - // a read entry row first covers only ordinary rows: catch-up's activity: - // mark reads all of them up to a timestamp. It never reads mentions, - // broadcasts, broadcast replies or messages marked unread; only their own - // msg: or thread: marks do, so a short sample can still hide an unread one - // above the window. A row read by its own mark says nothing about the rows - // above it either. + // the first loaded row the oldest unread message the store knows. So the + // line needs an entry row above it that a channel-wide read reaches, which + // reads every older ordinary row; one read only by its own msg: or thread: + // mark says nothing about them. That still leaves mentions, broadcasts, + // broadcast replies and messages marked unread, which catch-up's activity: + // mark never reads, so a short sample can hide an unread one of those + // above the window. let afterRead = messageVisit.entryHistoryComplete; for (const { row, attention } of entryAttention) { if (attention.status === "unknown" || attention.pending) @@ -503,7 +507,7 @@ function Timeline({ // Ineligible means no attention category, not that an incoming row is read. if (attention.forced || attention.unread) return afterRead ? { ready: true, messageId: row.id } : { ready: true }; - afterRead = true; + if (attention.prefixRead) afterRead = true; } return { ready: true }; })(); diff --git a/src/features/relay/unread-startup.test.ts b/src/features/relay/unread-startup.test.ts index 55e553c040..b0134cde84 100644 --- a/src/features/relay/unread-startup.test.ts +++ b/src/features/relay/unread-startup.test.ts @@ -1138,6 +1138,79 @@ it("an evidence overflow is repaired on the next recovery tick", async () => { } }); +it("a channel snapshot is sampling only while a repair that can sample it is owed or reading", async () => { + const h = setup(); + let fail = true; + let gate: Promise | undefined; + h.query.mockImplementation(async (filters) => { + if (filters[0]?.kinds?.includes(39002)) + return [ + roster(h.relay, "other", [h.viewer.pubkey], 10), + metadata(h.relay, "other", "other", 10), + ]; + if (!filters[0]?.kinds?.includes(9)) return []; + await gate; + if (fail) throw new Error("Transient evidence failure"); + return [h.row]; + }); + // Live evidence before anything asked: `ensure` would start the repair. + h.receive([message(h.peer, "other", "before any repair", 12)]); + expect(h.snapshot()).toMatchObject({ + coverage: "observed", + sampling: true, + }); + + // The first repair fails. Live traffic marks the evidence observed again + // but samples nothing, and no repair is reading until the recovery tick. + await h.session.unread.ensure(); + expect(h.snapshot()).toMatchObject({ + freshness: "stale", + error: "Transient evidence failure", + }); + expect(h.snapshot().sampling).toBeUndefined(); + h.receive([message(h.peer, "other", "after the failure", 13)]); + expect(h.snapshot()).toMatchObject({ + freshness: "observed", + error: "Transient evidence failure", + }); + expect(h.snapshot().sampled).toBeUndefined(); + expect(h.snapshot().sampling).toBeUndefined(); + + // A retry is sampling while it reads, and sampled once it returns. + fail = false; + const retry = deferred(); + gate = retry.promise; + const reads = h.query.mock.calls.length; + const repair = h.session.unread.refresh(); + await vi.waitFor(() => + expect( + h.query.mock.calls + .slice(reads) + .some(([filters]) => filters[0]?.include_aux), + ).toBe(true), + ); + expect(h.snapshot().sampling).toBe(true); + retry.release(); + await repair; + expect(h.snapshot()).toMatchObject({ sampled: true, freshness: "observed" }); + expect(h.snapshot().sampling).toBeUndefined(); + gate = undefined; + + // An overflow empties the sample; until the next tick nothing reads. + h.receive( + Array.from({ length: 19 }, (_, i) => + message(h.peer, "other", "x".repeat(450000), i + 20), + ), + ); + h.receive([message(h.peer, "other", "after the overflow", 40)]); + expect(h.snapshot()).toMatchObject({ + freshness: "observed", + error: "Unread observation capacity reached; refresh available", + }); + expect(h.snapshot().sampled).toBeUndefined(); + expect(h.snapshot().sampling).toBeUndefined(); +}); + it("a covered reconnect pays the drop's debt: the next recovery tick reads nothing", async () => { vi.useFakeTimers({ toFake: ["Date"], now: 1_700_000_000_000 }); try { diff --git a/src/features/relay/unread.test.ts b/src/features/relay/unread.test.ts index 6b665fb9d5..e02e89fe36 100644 --- a/src/features/relay/unread.test.ts +++ b/src/features/relay/unread.test.ts @@ -8,7 +8,11 @@ import { type ReadStateStorage, } from "./read-state-storage"; import { eventDto, type RelayEvent } from "./events"; -import type { ThreadActivitySnapshot, UnreadCapability } from "./unread"; +import type { + ThreadActivitySnapshot, + UnreadCapability, + UnreadSnapshot, +} from "./unread"; import type { SavedHead } from "./persistence"; import type { ReadStateSigning } from "./read-state-host"; import { retainReadState } from "./read-state-retention"; @@ -506,7 +510,7 @@ it("the channel snapshot names the oldest unread message its timeline shows", as expect(h.snapshot().oldestUnreadId).toBeUndefined(); }); -it("the channel snapshot says sampled once a repair finishes, keeps it through later repairs and regains it after a cache clear", async () => { +it("the channel snapshot says sampled once a repair finishes, keeps it through later repairs and regains it after a cache clear; sampling until then", async () => { const h = setup(); const rows = Array.from({ length: 30 }, (_, i) => message(h.alice, "room", `history ${i}`, i + 11), @@ -558,6 +562,8 @@ it("the channel snapshot says sampled once a repair finishes, keeps it through l freshness: "observed", observedCount: 20, oldestUnreadId: cursor.id, + // Nothing asked for a repair yet; `ensure` would start one. + sampling: true, }); expect(h.snapshot().sampled).toBeUndefined(); @@ -565,6 +571,7 @@ it("the channel snapshot says sampled once a repair finishes, keeps it through l let repair = h.session.unread.ensure(); await flush(); expect(h.snapshot().sampled).toBeUndefined(); + expect(h.snapshot().sampling).toBe(true); release(); await repair; expect(h.snapshot()).toMatchObject({ @@ -572,6 +579,7 @@ it("the channel snapshot says sampled once a repair finishes, keeps it through l oldestUnreadId: rows[0]?.id, sampled: true, }); + expect(h.snapshot().sampling).toBeUndefined(); // A later repair only adds evidence: the sample stands while it runs, and // subscribers see no change. @@ -603,6 +611,7 @@ it("the channel snapshot says sampled once a repair finishes, keeps it through l ); expect(repairs).toBe(started + 1); expect(h.snapshot().sampled).toBeUndefined(); + expect(h.snapshot().sampling).toBe(true); release(); await vi.waitFor(() => expect(h.snapshot()).toMatchObject({ @@ -634,7 +643,12 @@ it("the channel snapshot says sampled once a repair finishes, keeps it through l // Access gained while a repair is out: subscribers see the reset at once, // and the sample returns only when the repair restarted after it finishes. - const stopRoom = h.session.unread.subscribe(h.target, () => {}); + // None of them sees a snapshot that no repair can sample, the restart's gap + // included. + const seen: UnreadSnapshot[] = []; + const stopRoom = h.session.unread.subscribe(h.target, () => + seen.push(h.snapshot()), + ); release = repairing(); started = repairs; void h.session.unread.refresh(); @@ -642,15 +656,59 @@ it("the channel snapshot says sampled once a repair finishes, keeps it through l expect(h.snapshot().sampled).toBe(true); h.grant("third"); expect(h.snapshot().sampled).toBeUndefined(); + expect(h.snapshot().sampling).toBe(true); release(); release = repairing(); await vi.waitFor(() => expect(repairs).toBe(started + 2)); expect(h.snapshot().sampled).toBeUndefined(); release(); await vi.waitFor(() => expect(h.snapshot().sampled).toBe(true)); + expect(seen.length).toBeGreaterThan(1); + expect(seen.filter((seen) => !seen.sampled && !seen.sampling)).toEqual([]); stopRoom(); }); +it("attention says a row is prefix-read only when a channel-wide read reaches it", async () => { + const h = setup(); + h.grant("room"); + // Catch-up needs the channel's type: a DM's is its channel read. + h.emit([metadata(h.relay, "room", "room", 11, [["t", "stream"]])]); + const first = message(h.alice, "room", "first", 11); + const second = message(h.alice, "room", "second", 12); + const mention = message(h.alice, "room", "mention", 13, [ + ["p", h.viewer.pubkey], + ]); + const third = message(h.alice, "room", "third", 14); + h.emit([first, second, mention, third]); + const attention = (event: RelayEvent) => + h.session.unread.attention("room", event.id); + + // A row's own receipt reads it alone, not the ordinary row above it. + await h.session.unread.markMessageRead("room", second.id); + expect(attention(second)).toMatchObject({ unread: false }); + expect(attention(second).prefixRead).toBeUndefined(); + expect(attention(first)).toMatchObject({ unread: true }); + + // Catch-up's mark reads ordinary rows through a timestamp; the mention it + // reads by its own receipt. + await h.session.unread.markTimelineRead("room", [ + first.id, + second.id, + mention.id, + ]); + expect(attention(first)).toMatchObject({ unread: false, prefixRead: true }); + expect(attention(second)).toMatchObject({ unread: false, prefixRead: true }); + expect(attention(mention)).toMatchObject({ unread: false }); + expect(attention(mention).prefixRead).toBeUndefined(); + expect(attention(third)).toMatchObject({ unread: true }); + expect(attention(third).prefixRead).toBeUndefined(); + + // The channel's own frontier reads every message through it. + await h.session.unread.markThrough(h.target, mention.id); + expect(attention(mention)).toMatchObject({ unread: false, prefixRead: true }); + expect(attention(third).prefixRead).toBeUndefined(); +}); + it("counts replies only in the viewer's conversations and threads; others stay quiet until joined or mentioned", async () => { const h = setup(); h.grant("room"); diff --git a/src/features/relay/unread.ts b/src/features/relay/unread.ts index 17eece5a27..102091c592 100644 --- a/src/features/relay/unread.ts +++ b/src/features/relay/unread.ts @@ -44,6 +44,12 @@ export type UnreadSnapshot = Readonly<{ * reaches past the rows channel windows loaded, up to the repair's per-batch * limit. Later repairs keep it. */ sampled?: true; + /** Channel targets without `sampled`: a repair that can set it is reading, + * is about to start again, or has not been asked for (`ensure` starts one). + * Absent once one ends without sampling and none is due (it failed, evidence + * overflowed, or a dropped connection or access change abandoned it) until + * another starts. */ + sampling?: true; /** null means unobserved/denied, never a fabricated zero or an exact relay total. */ observedCount: number | null; attentionCount: number | null; @@ -75,6 +81,11 @@ export type MessageAttention = Readonly<{ /** This row is explicitly forced unread during the current channel visit. */ forced: boolean; viewing: boolean; + /** A channel-wide read reaches this row: the channel's read frontier, its + * catch-up mark (`activity:`) for an ordinary top-level message, or the + * retention floor. Each also reads every older ordinary top-level message, + * which the row's own `msg:` or `thread:` mark does not. */ + prefixRead?: true; }>; export type ThreadActivityItem = Readonly<{ channelId: string; @@ -956,6 +967,15 @@ export function createUnread({ view.channelId === channelId && view.visible(messageId, view.ids.has(messageId)), ); + const state = reads.state(); + const prefixRead = + Math.max( + state.frontiers[channelId] ?? -1, + timeline(entry, dm) + ? (state.frontiers[`activity:${channelId}`] ?? -1) + : -1, + reads.floor(), + ) >= event.created_at; return Object.freeze({ status: kind ? "eligible" @@ -966,9 +986,10 @@ export function createUnread({ ...(pending ? { pending: true } : {}), ...(entry.mentioned ? { mentioned: true } : {}), ...(entry.rootId ? { rootId: entry.rootId } : {}), - unread: isUnread(entry, reads.state(), dm), + unread: isUnread(entry, state, dm), forced: forcedMessages.get(channelId)?.has(messageId) ?? false, viewing, + ...(prefixRead ? { prefixRead: true as const } : {}), }); } function compute(target: ReadTarget): UnreadSnapshot { @@ -1058,6 +1079,14 @@ export function createUnread({ ...(evidence && target.kind === "channel" && sampled ? { sampled: true as const } : {}), + // `repairAgain` spans a restart's gap: the settling repair publishes + // before the next one begins. + ...(evidence && + target.kind === "channel" && + !sampled && + (!requested || refresh || repairAgain) + ? { sampling: true as const } + : {}), observedCount: evidence ? count : null, attentionCount: evidence ? attention : null, coverage: evidence ? "observed" : "unknown", @@ -1079,6 +1108,7 @@ export function createUnread({ a.latestMessage?.id === b.latestMessage?.id && a.oldestUnreadId === b.oldestUnreadId && a.sampled === b.sampled && + a.sampling === b.sampling && a.observedCount === b.observedCount && a.attentionCount === b.attentionCount && a.row?.observedCount === b.row?.observedCount &&