From fd9514b0ad6ec486cf7ae704a2938b56548e3aec Mon Sep 17 00:00:00 2001 From: Hosmel Quintana Date: Wed, 4 Nov 2026 17:02:21 -0600 Subject: [PATCH 1/2] fix: read a reply thread once when it opens Opening replies read them, then read them again as soon as the thread started listening for live updates. A listener joining a channel that was already joined catches up when the channel was joined after the page was read, which on a freshly loaded page is every thread opened later, though it had just read its replies. The thread now tells listen() when it started reading the replies it shows, so it only catches up on what was heard since then. It also asks what changed since that start, rather than since shortly before the read landed, so a reply posted during a slow read still shows. --- resources/js/components/commit-thread.tsx | 23 ++++++-- resources/js/lib/commit-changes.ts | 7 ++- resources/js/lib/realtime.ts | 6 +- tests/Unit/Realtime.test.ts | 27 +++++++++ tests/Unit/ThreadRecovery.test.tsx | 68 +++++++++++++++++++++-- 5 files changed, 118 insertions(+), 13 deletions(-) diff --git a/resources/js/components/commit-thread.tsx b/resources/js/components/commit-thread.tsx index 4c36788f0..ca56addcb 100644 --- a/resources/js/components/commit-thread.tsx +++ b/resources/js/components/commit-thread.tsx @@ -399,6 +399,8 @@ export function useThread( (lifetime.current ??= new AbortController()).signal; // When the replies shown were read, on the server's clock, to ask what changed since after missing updates. const readSince = useRef(Math.min(serverNow - 5_000, readAt())); + // When the thread read the replies shown, by the browser's clock, so listening catches up only on what came after. + const repliesReadAt = useRef(undefined); const [isReplying, setIsReplying] = useState(false); const [previousInitial, setPreviousInitial] = useState(initial); @@ -406,6 +408,8 @@ export function useThread( setPreviousInitial(initial); if (initial !== undefined && (replies?.length ?? 0) <= repliesPerPage) { ++replyRequest.current; + readSince.current = Math.min(serverNow - 5_000, readAt()); + repliesReadAt.current = undefined; setIsLoading(false); setThread((current) => ({ ...current, @@ -415,9 +419,15 @@ export function useThread( } } - const applyPage = (loaded: ThreadCommit[], cursor: string | null) => { + const applyPage = ( + loaded: ThreadCommit[], + cursor: string | null, + startedAt?: number, + ) => { // Read over however many requests, so dated from before the first of them could have started. - readSince.current = readAt() - 10_000; + readSince.current = + startedAt === undefined ? readAt() - 10_000 : readAt(startedAt); + repliesReadAt.current = startedAt; setThread((current) => ({ ...current, replies: loaded, @@ -485,6 +495,7 @@ export function useThread( : replies.length + extra; const request = ++replyRequest.current; + const startedAt = Date.now(); setIsLoading(true); return fetchRepliesUntil( @@ -497,7 +508,7 @@ export function useThread( ) .then((page) => { if (!signal.aborted && request === replyRequest.current) { - applyPage(page.data, page.next_cursor); + applyPage(page.data, page.next_cursor, startedAt); } }) .catch(() => undefined) @@ -550,6 +561,7 @@ export function useThread( let isCurrent = true; const request = ++replyRequest.current; const signal = readsLifetime(); + const startedAt = Date.now(); setIsLoading(true); fetchRepliesUntil( @@ -562,7 +574,7 @@ export function useThread( ) .then((page) => { if (isCurrent && request === replyRequest.current) { - onRepliesLoaded(page.data, page.next_cursor); + onRepliesLoaded(page.data, page.next_cursor, startedAt); } }) .catch(() => undefined) @@ -787,6 +799,7 @@ export function useThread( refresh.request(); } }), + repliesReadAt.current, ); // Muting, blocking or unmuting someone, or being blocked, changes which replies the viewer sees, including ones @@ -819,10 +832,12 @@ export function useThread( userChannel(viewerId), { relationships: recoverWhenFree }, isPreview ? undefined : recoverWhenFree, + repliesReadAt.current, ); return () => { isListening = false; + repliesReadAt.current = undefined; stop(); stopRelationships(); refresh.cancel(); diff --git a/resources/js/lib/commit-changes.ts b/resources/js/lib/commit-changes.ts index f82dc773e..492ac46b8 100644 --- a/resources/js/lib/commit-changes.ts +++ b/resources/js/lib/commit-changes.ts @@ -22,10 +22,11 @@ export function setServerTime(serverNow: number): void { } /** - * Now, on the server's clock, minus the margin: when something just read was read, to ask what changed since. + * Now, or the given time on the browser's clock, on the server's clock minus the margin: when something read then was + * read, to ask what changed since. */ -export function readAt(): number { - return Date.now() + serverOffset - margin; +export function readAt(at = Date.now()): number { + return at + serverOffset - margin; } type Waiting = { diff --git a/resources/js/lib/realtime.ts b/resources/js/lib/realtime.ts index 51e494827..3315bf94e 100644 --- a/resources/js/lib/realtime.ts +++ b/resources/js/lib/realtime.ts @@ -142,12 +142,14 @@ function entryFor(channel: RealtimeChannel, socket: RealtimeConnection) { /** * Listens for events on a channel, calling `onSync` whenever the channel is joined, at first and after every - * reconnection, since updates sent in between were never heard. Returns the function that stops listening. + * reconnection, since updates sent in between were never heard. `readAt` is when what it listens for was read, if after + * the page. Returns the function that stops listening. */ export function listen( channel: RealtimeChannel, events: Record, onSync?: () => void, + readAt?: number, ): () => void { const socket = connection; @@ -186,7 +188,7 @@ export function listen( if ( entry.isSubscribed && - mayHaveMissed(entry, Object.keys(events), pageRead) + mayHaveMissed(entry, Object.keys(events), readAt ?? pageRead) ) { queueMicrotask(sync); } diff --git a/tests/Unit/Realtime.test.ts b/tests/Unit/Realtime.test.ts index a53793816..b500c417a 100644 --- a/tests/Unit/Realtime.test.ts +++ b/tests/Unit/Realtime.test.ts @@ -327,6 +327,33 @@ void test('a listener joining a channel listened to since before the page was re assert.equal(busy, 1); }); +void test('a listener for what was read after the page, since the channel was joined, catches up only on what was heard since', async () => { + mock.timers.enable({ apis: ['setTimeout', 'Date'], now: 1_000_000 }); + const fake = fakeConnection(); + connectRealtime(fake.connection); + pageReadAt(Date.now()); + mock.timers.tick(1_000); + listen(userChannel('a'), { notifications: () => {} }); + fake.setStatus('connected'); + fake.joined.get('private-users.a')?.join(); + mock.timers.tick(1_000); + const readAt = Date.now(); + mock.timers.tick(500); + let quiet = 0; + let busy = 0; + let withThePage = 0; + + listen(userChannel('a'), { relationships: () => {} }, () => quiet++, readAt); + fake.joined.get('private-users.a')?.emit('account'); + listen(userChannel('a'), { account: () => {} }, () => busy++, readAt); + listen(userChannel('a'), { relationships: () => {} }, () => withThePage++); + await Promise.resolve(); + + assert.equal(quiet, 0); + assert.equal(busy, 1); + assert.equal(withThePage, 1, 'what came with the page was read before the channel was joined'); +}); + void test('a page coming back from history makes every listener joining it catch up, however late, and reads live props again', async () => { mock.timers.enable({ apis: ['setTimeout', 'Date'], now: 1_000_000 }); const fake = fakeConnection(); diff --git a/tests/Unit/ThreadRecovery.test.tsx b/tests/Unit/ThreadRecovery.test.tsx index 6f059d16a..79b38223e 100644 --- a/tests/Unit/ThreadRecovery.test.tsx +++ b/tests/Unit/ThreadRecovery.test.tsx @@ -25,7 +25,7 @@ vi.doMock('@inertiajs/react', async () => ({ router: { restore: () => undefined, on: () => () => {}, reload: () => {} }, })); -const { connectRealtime } = await import('../../resources/js/lib/realtime'); +const { commitChannel, connectRealtime, listen, userChannel } = await import('../../resources/js/lib/realtime'); const { useThread } = await import('../../resources/js/components/commit-thread'); const reply = (id: string) => @@ -40,8 +40,8 @@ let thread: ReturnType; type Initial = { data: ThreadCommit[]; next_cursor: null; prev_cursor: null }; -function Probe(props: { initial: Initial }) { - thread = useThread(root, props.initial, true); +function Probe(props: { initial?: Initial }) { + thread = useThread(root, props.initial, props.initial !== undefined); return null; } @@ -98,7 +98,7 @@ async function settle() { } } -let props = { initial: initialOf(['r1', 'r2']) }; +let props: { initial?: Initial } = { initial: initialOf(['r1', 'r2']) }; const shownIds = () => thread.replies?.map((shown) => shown.id); @@ -152,3 +152,63 @@ test('catching up on many changes reads again only what was loaded, leaving a la assert.equal(thread.hasMore, true); assert.ok(replyReads.length - before <= 2, `read ${replyReads.length - before} pages`); }); + +test('opening a thread reads its replies once, though the channels it listens on were joined before', async () => { + props = {}; + server = ['r1', 'r2']; + listen(userChannel('me'), {}); + listen(commitChannel('root'), {}); + joinAll(fake); + await vi.advanceTimersByTimeAsync(1_000); + void harness.render(Probe, props); + thread.toggle(); + await settle(); + + assert.deepEqual(shownIds(), ['r1', 'r2']); + assert.equal(replyReads.length, 1); +}); + +test('a reply posted while a slow thread was opening shows once its replies land', async () => { + props = {}; + server = ['r1']; + listen(userChannel('me'), {}); + listen(commitChannel('root'), {}); + joinAll(fake); + await vi.advanceTimersByTimeAsync(1_000); + const answer = globalThis.fetch; + let postedAt = Infinity; + let isHeld = false; + let land = () => {}; + vi.stubGlobal('fetch', async (input: string, init?: RequestInit) => { + const url = new URL(input, 'http://localhost'); + + if (url.pathname === '/commits/changes') { + const replies = Number(url.searchParams.get('since')) <= postedAt ? ['r2'] : []; + + return Response.json({ changes: { root: { changed: replies.length > 0, replies, truncated: false } } }); + } + + const response = answer(input, init); + + if (isHeld || !url.searchParams.has('limit')) { + return response; + } + + isHeld = true; + + return new Promise((resolve) => { + land = () => resolve(response); + }); + }); + void harness.render(Probe, props); + thread.toggle(); + await vi.advanceTimersByTimeAsync(1_000); + server = ['r1', 'r2']; + postedAt = Date.now(); + fake.joined.get('commits.root')?.emit('visibility', { ids: ['r2'] }); + await vi.advanceTimersByTimeAsync(20_000); + land(); + await settle(); + + assert.deepEqual(shownIds(), ['r1', 'r2']); +}); From 3ea2d9cb002fe7d8e95abc11aa33c5f2bb2451b6 Mon Sep 17 00:00:00 2001 From: Hosmel Quintana Date: Wed, 4 Nov 2026 19:17:47 -0600 Subject: [PATCH 2/2] fix: keep thread read marks out of render The branch where replies the page brought replace the thread's set its read marks during render, so a render React discards could move them while the old replies stay shown. They are now set in an effect once those replies show. --- resources/js/components/commit-thread.tsx | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/resources/js/components/commit-thread.tsx b/resources/js/components/commit-thread.tsx index ca56addcb..ef4f4422b 100644 --- a/resources/js/components/commit-thread.tsx +++ b/resources/js/components/commit-thread.tsx @@ -408,8 +408,6 @@ export function useThread( setPreviousInitial(initial); if (initial !== undefined && (replies?.length ?? 0) <= repliesPerPage) { ++replyRequest.current; - readSince.current = Math.min(serverNow - 5_000, readAt()); - repliesReadAt.current = undefined; setIsLoading(false); setThread((current) => ({ ...current, @@ -419,6 +417,17 @@ export function useThread( } } + // Replies the page brought were read with it, whatever the thread read before, once they show. + const onInitialShown = useEffectEvent(() => { + readSince.current = Math.min(serverNow - 5_000, readAt()); + repliesReadAt.current = undefined; + }); + useEffect(() => { + if (initial !== undefined && replies === initial.data) { + onInitialShown(); + } + }, [initial, replies]); + const applyPage = ( loaded: ThreadCommit[], cursor: string | null,