diff --git a/src/app/features/room/RoomInput.tsx b/src/app/features/room/RoomInput.tsx index e92ae575..03926da9 100644 --- a/src/app/features/room/RoomInput.tsx +++ b/src/app/features/room/RoomInput.tsx @@ -28,7 +28,8 @@ import { config, toRem, } from 'folds'; -import { RoomOutbox, sendTracked } from './RoomOutbox'; +import { RoomOutbox } from './RoomOutbox'; +import { sendTracked } from '../../state/room/outboxSend'; import { clearDraft, loadDraft, saveDraft } from '../../state/room/composerStore'; import { useMatrixClient } from '../../hooks/useMatrixClient'; @@ -333,7 +334,7 @@ export const RoomInput = forwardRef( }); handleCancelUpload(uploads); const contents = fulfilledPromiseSettledResult(await Promise.allSettled(contentsPromises)); - contents.forEach((content) => mx.sendMessage(roomId, content as any)); + contents.forEach((content) => sendTracked(mx, roomId, content as Record)); }; const submit = useCallback(() => { @@ -492,11 +493,7 @@ export const RoomInput = forwardRef( await getImageUrlBlob(stickerUrl) ); - mx.sendEvent(roomId, EventType.Sticker, { - body: label, - url: mxc, - info, - }); + sendTracked(mx, roomId, { body: label, url: mxc, info }, EventType.Sticker); }; return ( diff --git a/src/app/features/room/RoomOutbox.test.ts b/src/app/features/room/RoomOutbox.test.ts index df22f9d4..55597694 100644 --- a/src/app/features/room/RoomOutbox.test.ts +++ b/src/app/features/room/RoomOutbox.test.ts @@ -1,57 +1,23 @@ -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { loadOutbox } from '../../state/room/composerStore'; -import { sendTracked } from './RoomOutbox'; +import { describe, expect, it } from 'vitest'; +import { SLOW_SEND_MS, outboxView } from './RoomOutbox'; -const stubLocalStorage = () => { - const m = new Map(); - vi.stubGlobal('localStorage', { - getItem: (k: string) => m.get(k) ?? null, - setItem: (k: string, v: string) => { - m.set(k, v); - }, - removeItem: (k: string) => { - m.delete(k); - }, - }); -}; - -const fakeClient = (send: (...a: unknown[]) => Promise) => - ({ getUserId: () => '@a:x', makeTxnId: () => 'txn-1', sendMessage: send } as never); - -describe('sendTracked', () => { - beforeEach(stubLocalStorage); - afterEach(() => { - vi.unstubAllGlobals(); +describe('outboxView', () => { + it('walks sending -> still sending (resendable) -> failed', () => { + expect(outboxView({ status: 'pending', ts: 100 }, 100, 0)).toMatchObject({ + label: 'Sending…', + canResend: false, + }); + expect(outboxView({ status: 'pending', ts: 100 }, 100 + SLOW_SEND_MS, 0)).toMatchObject({ + label: expect.stringMatching(/^Still sending/), + canResend: true, + }); + expect(outboxView({ status: 'failed', ts: 100, error: 'x' }, 100, 0)).toMatchObject({ + failed: true, + canResend: true, + }); }); - const content = { msgtype: 'm.text', body: 'typed before reload' }; - - it('holds the message in the outbox while the send is in flight, so a reload keeps it', async () => { - let finish!: () => void; - const mx = fakeClient( - () => - new Promise((r) => { - finish = () => r({ event_id: '$e' }); - }) - ); - const done = sendTracked(mx, '!r:x', content); - expect(loadOutbox('@a:x')).toMatchObject([{ txnId: 'txn-1', status: 'pending', content }]); - finish(); - await done; - expect(loadOutbox('@a:x')).toEqual([]); - }); - - it('a failed send stays, marked failed, and is not cleared', async () => { - const mx = fakeClient(() => Promise.reject(new Error('ECONNRESET'))); - await sendTracked(mx, '!r:x', content); - expect(loadOutbox('@a:x')).toMatchObject([ - { txnId: 'txn-1', status: 'failed', error: 'ECONNRESET', content }, - ]); - }); - - it('sends under the stored txn id, so a resend after an unknown outcome deduplicates', async () => { - const send = vi.fn().mockResolvedValue({}); - await sendTracked(fakeClient(send), '!r:x', content); - expect(send).toHaveBeenCalledWith('!r:x', content, 'txn-1'); + it('an entry older than this page load reads as interrupted', () => { + expect(outboxView({ status: 'pending', ts: 5 }, 6, 10).label).toMatch(/Interrupted/); }); }); diff --git a/src/app/features/room/RoomOutbox.tsx b/src/app/features/room/RoomOutbox.tsx index cf164472..1e77df43 100644 --- a/src/app/features/room/RoomOutbox.tsx +++ b/src/app/features/room/RoomOutbox.tsx @@ -3,32 +3,44 @@ import { MatrixClient } from 'matrix-js-sdk'; import { Box, Chip, Text, color, config } from 'folds'; import { OutboxEntry, - enqueueOutbox, - failOutbox, loadOutbox, removeOutbox, - retryOutbox, subscribeOutbox, } from '../../state/room/composerStore'; +import { PAGE_START, resendTracked } from '../../state/room/outboxSend'; -const attempt = (mx: MatrixClient, userId: string, entry: OutboxEntry): Promise => - mx - .sendMessage(entry.roomId, entry.content as never, entry.txnId) - .then(() => removeOutbox(userId, entry.txnId)) - .catch((e: unknown) => - failOutbox(userId, entry.txnId, e instanceof Error ? e.message : String(e)) - ); +/** A send still unconfirmed after this long is offered Resend: the request may be hung. */ +export const SLOW_SEND_MS = 30_000; -/** Send a message, keeping it in the durable outbox until the server confirms it. */ -export const sendTracked = ( - mx: MatrixClient, - roomId: string, - content: Record -): Promise => { - const userId = mx.getUserId() ?? ''; - const entry = { roomId, content, txnId: mx.makeTxnId() }; - enqueueOutbox(userId, entry); - return attempt(mx, userId, { ...entry, status: 'pending', ts: Date.now() }); +export type OutboxView = { label: string; failed: boolean; canResend: boolean }; + +export const outboxView = ( + entry: Pick, + now: number, + since: number = PAGE_START +): OutboxView => { + if (entry.status === 'failed') { + return { + label: `Not sent${entry.error ? `: ${entry.error}` : ''}`, + failed: true, + canResend: true, + }; + } + if (entry.ts < since) { + return { + label: 'Interrupted by a reload: may not have been sent', + failed: false, + canResend: true, + }; + } + if (now - entry.ts >= SLOW_SEND_MS) { + return { + label: 'Still sending: slow connection (Resend may duplicate it if it lands later)', + failed: false, + canResend: true, + }; + } + return { label: 'Sending…', failed: false, canResend: false }; }; export const useRoomOutbox = (userId: string, roomId: string): OutboxEntry[] => { @@ -42,19 +54,28 @@ export const useRoomOutbox = (userId: string, roomId: string): OutboxEntry[] => return entries; }; -/** Entries still 'pending' that predate this page load lost their promise in a reload: outcome unknown. */ -const pageStart = Date.now(); +/** Re-render every few seconds while anything is waiting, so a slow send turns resendable. */ +const useNow = (active: boolean): number => { + const [now, setNow] = useState(Date.now()); + useEffect(() => { + if (!active) return undefined; + setNow(Date.now()); + const id = setInterval(() => setNow(Date.now()), 5000); + return () => clearInterval(id); + }, [active]); + return now; +}; type RoomOutboxProps = { mx: MatrixClient; roomId: string }; /** - * Messages this client sent that the server has not confirmed: still sending, or failed. - * They also appear here after a reload, since the outbox is durable. Resending reuses the - * original txn id, so a send that did land the first time is not duplicated. + * Messages this client sent that the server has not confirmed: sending, slow, interrupted by a + * reload, or failed, each with Resend where one makes sense and Discard. */ export function RoomOutbox({ mx, roomId }: RoomOutboxProps) { const userId = mx.getUserId() ?? ''; const entries = useRoomOutbox(userId, roomId); + const now = useNow(entries.length > 0); if (entries.length === 0) return null; return ( @@ -64,40 +85,32 @@ export function RoomOutbox({ mx, roomId }: RoomOutboxProps) { style={{ padding: `${config.space.S200} ${config.space.S300} 0` }} > {entries.map((entry) => { - const failed = entry.status === 'failed'; - const interrupted = !failed && entry.ts < pageStart; - const canResend = failed || interrupted; + const view = outboxView(entry, now); return ( - {(() => { - if (failed) return `Not sent${entry.error ? `: ${entry.error}` : ''}`; - return interrupted ? 'Interrupted by a reload: may not have been sent' : 'Sending…'; - })()} + {view.label} {String(entry.content.body ?? '')} - {canResend && ( + {view.canResend && ( <> - { - retryOutbox(userId, entry.txnId); - attempt(mx, userId, entry); - }} - > - Resend - - removeOutbox(userId, entry.txnId)}> - Discard - + resendTracked(mx, entry)}> + Resend + + removeOutbox(userId, entry.txnId)} + > + Discard + )} diff --git a/src/app/features/room/RoomTimeline.tsx b/src/app/features/room/RoomTimeline.tsx index e9c13e2a..9bd776f5 100644 --- a/src/app/features/room/RoomTimeline.tsx +++ b/src/app/features/room/RoomTimeline.tsx @@ -47,6 +47,7 @@ import { import { isKeyHotkey } from 'is-hotkey'; import { Opts as LinkifyOpts } from 'linkifyjs'; import { useTranslation } from 'react-i18next'; +import { isUnconfirmedLocalEcho } from './localEcho'; import { eventWithShortcode, factoryEventSentBy, getMxIdLocalPart } from '../../utils/matrix'; import { useMatrixClient } from '../../hooks/useMatrixClient'; import { useVirtualPaginator, ItemRange } from '../../hooks/useVirtualPaginator'; @@ -389,6 +390,19 @@ const useLiveEventArrive = (room: Room, onArrive: (mEvent: MatrixEvent) => void) }, [room, onArrive]); }; +/** A local echo changes status (or becomes its remote echo) without a Timeline event: redraw. */ +const useLocalEchoRefresh = (room: Room, onUpdate: () => void) => { + useEffect(() => { + const handle: RoomEventHandlerMap[RoomEvent.LocalEchoUpdated] = (_evt, r) => { + if (r.roomId === room.roomId) onUpdate(); + }; + room.on(RoomEvent.LocalEchoUpdated, handle); + return () => { + room.removeListener(RoomEvent.LocalEchoUpdated, handle); + }; + }, [room, onUpdate]); +}; + const useLiveTimelineRefresh = (room: Room, onRefresh: () => void) => { useEffect(() => { const handleTimelineRefresh: RoomEventHandlerMap[RoomEvent.TimelineRefresh] = (r) => { @@ -648,6 +662,18 @@ export function RoomTimeline({ room, eventId, roomInputRef, editor }: RoomTimeli ) ); + useLocalEchoRefresh( + room, + useCallback(() => { + // the echo was hidden while unconfirmed, so its arrival scrolled to nothing: scroll now + if (atBottomRef.current) { + scrollToBottomRef.current.count += 1; + scrollToBottomRef.current.smooth = true; + } + setTimeline((ct) => ({ ...ct })); + }, []) + ); + const handleOpenEvent = useCallback( async ( evtId: string, @@ -1646,6 +1672,7 @@ export function RoomTimeline({ room, eventId, roomInputRef, editor }: RoomTimeli if (mEvent.isRedacted() && !showHiddenEvents) { return null; } + if (isUnconfirmedLocalEcho(mEvent)) return null; if (!newDivider && readUptoEventIdRef.current) { newDivider = prevEvent?.getId() === readUptoEventIdRef.current; diff --git a/src/app/features/room/localEcho.test.ts b/src/app/features/room/localEcho.test.ts new file mode 100644 index 00000000..fa44ac9b --- /dev/null +++ b/src/app/features/room/localEcho.test.ts @@ -0,0 +1,20 @@ +import { describe, expect, it } from 'vitest'; +import { EventStatus } from 'matrix-js-sdk'; +import { isUnconfirmedLocalEcho } from './localEcho'; + +describe('isUnconfirmedLocalEcho', () => { + it.each([ + EventStatus.ENCRYPTING, + EventStatus.QUEUED, + EventStatus.SENDING, + EventStatus.NOT_SENT, + EventStatus.CANCELLED, + ])('hides %s', (status) => { + expect(isUnconfirmedLocalEcho({ status })).toBe(true); + }); + + it('shows a server-confirmed event and one the server has accepted (SENT)', () => { + expect(isUnconfirmedLocalEcho({ status: null })).toBe(false); + expect(isUnconfirmedLocalEcho({ status: EventStatus.SENT })).toBe(false); + }); +}); diff --git a/src/app/features/room/localEcho.ts b/src/app/features/room/localEcho.ts new file mode 100644 index 00000000..4f827500 --- /dev/null +++ b/src/app/features/room/localEcho.ts @@ -0,0 +1,10 @@ +import { EventStatus, MatrixEvent } from 'matrix-js-sdk'; + +/** + * A local echo the server has not accepted: the SDK adds it to the live timeline the moment + * sendMessage is called, marks it NOT_SENT on failure, and drops it on reload. Rendered like + * any message it looks delivered and then vanishes. The outbox (RoomOutbox) is the one place + * that shows unconfirmed sends, so the timeline hides these until the status is SENT. + */ +export const isUnconfirmedLocalEcho = (mEvent: Pick): boolean => + mEvent.status !== null && mEvent.status !== undefined && mEvent.status !== EventStatus.SENT; diff --git a/src/app/state/room/composerStore.test.ts b/src/app/state/room/composerStore.test.ts index 92bdcdfd..f5a0f35b 100644 --- a/src/app/state/room/composerStore.test.ts +++ b/src/app/state/room/composerStore.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest'; import { Descendant } from 'slate'; import { + PAGE_ID, clearDraft, enqueueOutbox, failOutbox, @@ -77,7 +78,7 @@ describe('outbox', () => { const store = memoryStore(); enqueueOutbox('@a:x', entry, 5, store); const after = loadOutbox('@a:x', store); - expect(after).toEqual([{ ...entry, status: 'pending', ts: 5 }]); + expect(after).toEqual([{ ...entry, status: 'pending', ts: 5, tab: PAGE_ID }]); removeOutbox('@a:x', 't1', store); expect(loadOutbox('@a:x', store)).toEqual([]); expect(store.size()).toBe(0); @@ -88,11 +89,19 @@ describe('outbox', () => { enqueueOutbox('@a:x', entry, 5, store); failOutbox('@a:x', 't1', 'offline', store); expect(loadOutbox('@a:x', store)[0]).toMatchObject({ status: 'failed', error: 'offline' }); - retryOutbox('@a:x', 't1', store); + retryOutbox('@a:x', 't1', 9, 't1', store); expect(loadOutbox('@a:x', store)[0]).toMatchObject({ status: 'pending', txnId: 't1' }); expect(loadOutbox('@a:x', store)[0].error).toBeUndefined(); }); + it('a resend can rekey the entry and restarts its clock', () => { + const store = memoryStore(); + enqueueOutbox('@a:x', entry, 5, store); + failOutbox('@a:x', 't1', 'offline', store); + retryOutbox('@a:x', 't1', 50, 't2', store); + expect(loadOutbox('@a:x', store)).toMatchObject([{ txnId: 't2', status: 'pending', ts: 50 }]); + }); + it('re-enqueueing the same txn id does not duplicate it; other accounts are separate', () => { const store = memoryStore(); enqueueOutbox('@a:x', entry, 5, store); diff --git a/src/app/state/room/composerStore.ts b/src/app/state/room/composerStore.ts index 1cbe54aa..91933d72 100644 --- a/src/app/state/room/composerStore.ts +++ b/src/app/state/room/composerStore.ts @@ -15,12 +15,18 @@ export type OutboxStatus = 'pending' | 'failed'; export type OutboxEntry = { txnId: string; roomId: string; + /** Event type; absent means m.room.message. */ + type?: string; content: Record; status: OutboxStatus; error?: string; ts: number; + /** The page that last sent it: only that page's watchdog may judge it stalled. */ + tab?: string; }; +export const PAGE_ID = Math.random().toString(36).slice(2); + type Store = Pick; const defaultStore = (): Store | undefined => { @@ -90,7 +96,8 @@ export const saveDraft = ( removeKey(store, draftKey(userId, roomId)); return; } - if (!writeJson(store, draftKey(userId, roomId), draft)) removeKey(store, draftKey(userId, roomId)); + if (!writeJson(store, draftKey(userId, roomId), draft)) + removeKey(store, draftKey(userId, roomId)); }; export const clearDraft = ( @@ -137,7 +144,7 @@ export const enqueueOutbox = ( store: Store | undefined = defaultStore() ): void => { const rest = loadOutbox(userId, store).filter((e) => e.txnId !== entry.txnId); - saveOutbox(userId, [...rest, { ...entry, status: 'pending', ts: now }], store); + saveOutbox(userId, [...rest, { ...entry, status: 'pending', ts: now, tab: PAGE_ID }], store); }; export const failOutbox = ( @@ -155,16 +162,23 @@ export const failOutbox = ( ); }; -/** Retry in flight: failed goes back to pending so it is not shown as dead while it is sent. */ +/** + * A resend begins: the entry is pending again from `now`, under `newTxnId` when the SDK still + * holds the failed attempt's echo under the old one (it refuses to add a second echo for a txn id). + */ export const retryOutbox = ( userId: string, txnId: string, + now: number = Date.now(), + newTxnId: string = txnId, store: Store | undefined = defaultStore() ): void => { saveOutbox( userId, loadOutbox(userId, store).map((e) => - e.txnId === txnId ? { ...e, status: 'pending', error: undefined } : e + e.txnId === txnId + ? { ...e, txnId: newTxnId, status: 'pending', error: undefined, ts: now, tab: PAGE_ID } + : e ), store ); diff --git a/src/app/state/room/outboxSend.test.ts b/src/app/state/room/outboxSend.test.ts new file mode 100644 index 00000000..8f1db8fd --- /dev/null +++ b/src/app/state/room/outboxSend.test.ts @@ -0,0 +1,132 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { enqueueOutbox, failOutbox, loadOutbox } from './composerStore'; +import { resendInterrupted, resendTracked, sendTracked } from './outboxSend'; + +const stubLocalStorage = () => { + const m = new Map(); + vi.stubGlobal('localStorage', { + getItem: (k: string) => m.get(k) ?? null, + setItem: (k: string, v: string) => { + m.set(k, v); + }, + removeItem: (k: string) => { + m.delete(k); + }, + }); +}; + +const fakeClient = (send: (...a: unknown[]) => Promise) => + ({ getUserId: () => '@a:x', makeTxnId: () => 'txn-1', sendEvent: send } as never); + +describe('sendTracked', () => { + beforeEach(stubLocalStorage); + afterEach(() => { + vi.unstubAllGlobals(); + }); + + const content = { msgtype: 'm.text', body: 'typed before reload' }; + + it('holds the message in the outbox while the send is in flight, so a reload keeps it', async () => { + let finish!: () => void; + const mx = fakeClient( + () => + new Promise((r) => { + finish = () => r({ event_id: '$e' }); + }) + ); + const done = sendTracked(mx, '!r:x', content); + expect(loadOutbox('@a:x')).toMatchObject([{ txnId: 'txn-1', status: 'pending', content }]); + finish(); + await done; + expect(loadOutbox('@a:x')).toEqual([]); + }); + + it('a failed send stays, marked failed, and is not cleared', async () => { + const mx = fakeClient(() => Promise.reject(new Error('ECONNRESET'))); + await sendTracked(mx, '!r:x', content); + expect(loadOutbox('@a:x')).toMatchObject([ + { txnId: 'txn-1', status: 'failed', error: 'ECONNRESET', content }, + ]); + }); + + it('sends under the stored txn id, so a resend after an unknown outcome deduplicates', async () => { + const send = vi.fn().mockResolvedValue({}); + await sendTracked(fakeClient(send), '!r:x', content); + expect(send).toHaveBeenCalledWith('!r:x', 'm.room.message', content, 'txn-1'); + }); +}); + +describe('resendTracked', () => { + beforeEach(stubLocalStorage); + afterEach(() => { + vi.unstubAllGlobals(); + }); + const entry = { + txnId: 'txn-old', + roomId: '!r:x', + content: { msgtype: 'm.text', body: 'b' }, + status: 'failed' as const, + ts: 1, + }; + + it('uses a fresh txn id while the SDK still holds the failed echo (it would throw on the old one)', async () => { + const send = vi.fn().mockResolvedValue({}); + const mx = { + getUserId: () => '@a:x', + makeTxnId: () => 'txn-new', + sendEvent: send, + getRoom: () => ({ getEventForTxnId: () => ({}) }), + } as never; + await resendTracked(mx, entry); + expect(send).toHaveBeenCalledWith('!r:x', 'm.room.message', entry.content, 'txn-new'); + }); + + it('reuses the txn id after a reload, when no echo exists, so the server deduplicates', async () => { + const send = vi.fn().mockResolvedValue({}); + const mx = { + getUserId: () => '@a:x', + makeTxnId: () => 'txn-new', + sendEvent: send, + getRoom: () => ({ getEventForTxnId: () => undefined }), + } as never; + await resendTracked(mx, entry); + expect(send).toHaveBeenCalledWith('!r:x', 'm.room.message', entry.content, 'txn-old'); + }); + + it('a synchronous throw from the SDK marks the entry failed instead of escaping', async () => { + const mx = { + getUserId: () => '@a:x', + makeTxnId: () => 'txn-new', + sendEvent: () => { + throw new Error('addPendingEvent called on an event with known txnId'); + }, + getRoom: () => undefined, + } as never; + await sendTracked(mx, '!r:x', { body: 'x' }); + expect(loadOutbox('@a:x')[0]).toMatchObject({ status: 'failed' }); + }); +}); + +describe('resendInterrupted', () => { + beforeEach(stubLocalStorage); + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('resends pending entries older than the page under their own txn id, and leaves failed ones', async () => { + const send = vi.fn().mockResolvedValue({}); + const mx = { + getUserId: () => '@a:x', + makeTxnId: () => 'fresh', + sendEvent: send, + getRoom: () => undefined, + } as never; + enqueueOutbox('@a:x', { txnId: 'old', roomId: '!r:x', content: { body: 'o' } }, 5); + enqueueOutbox('@a:x', { txnId: 'bad', roomId: '!r:x', content: { body: 'b' } }, 5); + failOutbox('@a:x', 'bad', 'nope'); + enqueueOutbox('@a:x', { txnId: 'new', roomId: '!r:x', content: { body: 'n' } }, 500); + await resendInterrupted(mx, 100); + expect(send).toHaveBeenCalledTimes(1); + expect(send).toHaveBeenCalledWith('!r:x', 'm.room.message', { body: 'o' }, 'old'); + }); +}); diff --git a/src/app/state/room/outboxSend.ts b/src/app/state/room/outboxSend.ts new file mode 100644 index 00000000..845e64a0 --- /dev/null +++ b/src/app/state/room/outboxSend.ts @@ -0,0 +1,73 @@ +import { MatrixClient } from 'matrix-js-sdk'; +import { + OutboxEntry, + enqueueOutbox, + failOutbox, + loadOutbox, + removeOutbox, + retryOutbox, +} from './composerStore'; + +/** Entries still 'pending' that predate this page load lost their promise in a reload: outcome unknown. */ +export const PAGE_START = Date.now(); + +const attempt = async (mx: MatrixClient, userId: string, entry: OutboxEntry): Promise => { + try { + await mx.sendEvent( + entry.roomId, + (entry.type ?? 'm.room.message') as never, + entry.content as never, + entry.txnId + ); + removeOutbox(userId, entry.txnId); + } catch (e) { + // sendEvent can also throw synchronously (e.g. a txn id the SDK already holds) + failOutbox(userId, entry.txnId, e instanceof Error ? e.message : String(e)); + } +}; + +/** + * Send, keeping the message in the durable outbox until the server confirms it. The timeline + * hides the SDK's local echo until then, so this is the only place an unconfirmed send shows. + */ +export const sendTracked = ( + mx: MatrixClient, + roomId: string, + content: Record, + type?: string +): Promise => { + const userId = mx.getUserId() ?? ''; + const entry = { roomId, content, txnId: mx.makeTxnId(), ...(type ? { type } : {}) }; + enqueueOutbox(userId, entry); + return attempt(mx, userId, { ...entry, status: 'pending', ts: Date.now() }); +}; + +/** + * Resend. An echo still held by the SDK under the old txn id (a failure in this session) makes + * it refuse the same id, so that attempt is abandoned under a fresh id. After a reload the echo + * is gone and the old id is reused, so a send that did land is deduplicated by the server. + */ +export const resendTracked = (mx: MatrixClient, entry: OutboxEntry): Promise => { + const userId = mx.getUserId() ?? ''; + const held = mx.getRoom(entry.roomId)?.getEventForTxnId(entry.txnId); + const txnId = held ? mx.makeTxnId() : entry.txnId; + retryOutbox(userId, entry.txnId, Date.now(), txnId); + return attempt(mx, userId, { ...entry, txnId, status: 'pending', ts: Date.now() }); +}; + +/** + * After a reload, send again every entry the reload interrupted. They have no SDK echo, so each + * keeps its txn id and the server deduplicates one that did land. Failed entries are left for + * the user: they failed with an answer, these are only unknown. + */ +export const resendInterrupted = ( + mx: MatrixClient, + since: number = PAGE_START +): Promise => { + const userId = mx.getUserId() ?? ''; + return Promise.all( + loadOutbox(userId) + .filter((e) => e.status === 'pending' && e.ts < since) + .map((e) => resendTracked(mx, e)) + ); +}; diff --git a/src/app/state/room/roomInputDrafts.ts b/src/app/state/room/roomInputDrafts.ts index 5d8ec8c7..f3bf9e1c 100644 --- a/src/app/state/room/roomInputDrafts.ts +++ b/src/app/state/room/roomInputDrafts.ts @@ -20,9 +20,7 @@ export type TUploadItem = { export type TUploadListAtom = ReturnType>; -export const roomIdToUploadItemsAtomFamily = atomFamily( - createListAtom -); +export const roomIdToUploadItemsAtomFamily = atomFamily(createListAtom); export const roomUploadAtomFamily = createUploadAtomFamily(); diff --git a/src/client/initMatrix.test.ts b/src/client/initMatrix.test.ts new file mode 100644 index 00000000..95a8d966 --- /dev/null +++ b/src/client/initMatrix.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it, vi } from 'vitest'; +import { PAGE_ID, enqueueOutbox, failOutbox } from '../app/state/room/composerStore'; +import { RECOVER_RELOAD_MIN_GAP_MS, ownPendingAges, reloadToRecoverSync } from './initMatrix'; + +const memory = () => { + const m = new Map(); + return { + getItem: (k: string) => m.get(k) ?? null, + setItem: (k: string, v: string) => { + m.set(k, v); + }, + }; +}; + +describe('reloadToRecoverSync', () => { + it('reloads once, then refuses inside the gap so a down server cannot loop it', () => { + const store = memory(); + const reload = vi.fn(); + expect(reloadToRecoverSync(1_000_000, () => store, reload)).toBe(true); + expect( + reloadToRecoverSync(1_000_000 + RECOVER_RELOAD_MIN_GAP_MS - 1, () => store, reload) + ).toBe(false); + expect(reloadToRecoverSync(1_000_000 + RECOVER_RELOAD_MIN_GAP_MS, () => store, reload)).toBe( + true + ); + expect(reload).toHaveBeenCalledTimes(2); + }); +}); + +describe('reloadToRecoverSync with unusable storage', () => { + it('refuses instead of throwing when sessionStorage throws', () => { + const reload = vi.fn(); + const denied = () => { + throw new Error('SecurityError'); + }; + expect(reloadToRecoverSync(1, denied, reload)).toBe(false); + expect(reload).not.toHaveBeenCalled(); + }); +}); + +describe('ownPendingAges', () => { + it("counts only this page's pending sends, not other pages' or failed ones", () => { + const m = new Map(); + vi.stubGlobal('localStorage', { + getItem: (k: string) => m.get(k) ?? null, + setItem: (k: string, v: string) => { + m.set(k, v); + }, + removeItem: (k: string) => { + m.delete(k); + }, + }); + const e = (txnId: string) => ({ txnId, roomId: '!r:x', content: {} }); + enqueueOutbox('@a:x', e('mine'), 1000); + enqueueOutbox('@a:x', e('failed'), 1000); + failOutbox('@a:x', 'failed', 'x'); + enqueueOutbox('@a:x', e('theirs'), 1000); + const entries = JSON.parse(m.get('ratatoskr.outbox.@a:x') as string); + entries.find((x: { txnId: string }) => x.txnId === 'theirs').tab = `${PAGE_ID}-other`; + m.set('ratatoskr.outbox.@a:x', JSON.stringify(entries)); + expect(ownPendingAges('@a:x', 1500)).toEqual([500]); + vi.unstubAllGlobals(); + }); +}); diff --git a/src/client/initMatrix.ts b/src/client/initMatrix.ts index 871a7625..871c186a 100644 --- a/src/client/initMatrix.ts +++ b/src/client/initMatrix.ts @@ -1,9 +1,19 @@ -import { createClient, MatrixClient, IndexedDBStore, IndexedDBCryptoStore } from 'matrix-js-sdk'; +import { + ClientEvent, + createClient, + MatrixClient, + IndexedDBStore, + IndexedDBCryptoStore, + SyncState, +} from 'matrix-js-sdk'; import { cryptoCallbacks } from './secretStorageKeys'; import { clearNavToActivePathStore } from '../app/state/navToActivePath'; import { pushSessionToSW } from '../sw-session'; import { startResumeSync } from './resumeSync'; +import { startWatchdog } from './watchdog'; +import { PAGE_ID, loadOutbox } from '../app/state/room/composerStore'; +import { resendInterrupted } from '../app/state/room/outboxSend'; type Session = { baseUrl: string; @@ -46,11 +56,54 @@ const startSync = (mx: MatrixClient) => lazyLoadMembers: true, }); -export const startClient = async (mx: MatrixClient) => { - await startSync(mx); - startResumeSync(mx, { restart: startSync }); +const RECOVER_RELOAD_KEY = 'ratatoskr.recoverReloadAt'; +export const RECOVER_RELOAD_MIN_GAP_MS = 60_000; + +/** Reload to bring a dead connection back, at most once a minute, so a server that is down cannot loop us. */ +export const reloadToRecoverSync = ( + now: number = Date.now(), + getStore: () => Pick = () => sessionStorage, + reload: () => void = () => window.location.reload() +): boolean => { + try { + const store = getStore(); + const last = Number(store.getItem(RECOVER_RELOAD_KEY) ?? 0); + if (now - last < RECOVER_RELOAD_MIN_GAP_MS) return false; + store.setItem(RECOVER_RELOAD_KEY, String(now)); + } catch { + return false; + } + reload(); + return true; }; +/** Ages in ms of the sends this page made that are still pending: the only ones it may judge stalled. */ +export const ownPendingAges = (userId: string, now: number = Date.now()): number[] => + loadOutbox(userId) + .filter((e) => e.status === 'pending' && e.tab === PAGE_ID) + .map((e) => now - e.ts); + +const started = new WeakSet(); + +export const startClient = async (mx: MatrixClient) => { + await startSync(mx); + if (started.has(mx)) return; + started.add(mx); + startResumeSync(mx); + startWatchdog({ + pendingAges: () => ownPendingAges(mx.getUserId() ?? ''), + isVisible: () => document.visibilityState === 'visible', + isOnline: () => navigator.onLine, + recover: () => reloadToRecoverSync(), + }); + // What a reload interrupted is sent again once sync is up. + const onSync = (state: SyncState) => { + if (state !== SyncState.Prepared && state !== SyncState.Syncing) return; + mx.off(ClientEvent.Sync, onSync); + resendInterrupted(mx); + }; + mx.on(ClientEvent.Sync, onSync); +}; export const clearCacheAndReload = async (mx: MatrixClient) => { mx.stopClient(); clearNavToActivePathStore(mx.getSafeUserId()); diff --git a/src/client/resumeSync.test.ts b/src/client/resumeSync.test.ts index eada97af..e90ac6a8 100644 --- a/src/client/resumeSync.test.ts +++ b/src/client/resumeSync.test.ts @@ -1,25 +1,15 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { HEALTH_CHECK_MS, MIN_HIDDEN_MS, RESUME_DEBOUNCE_MS, startResumeSync } from './resumeSync'; +import { MIN_HIDDEN_MS, RESUME_DEBOUNCE_MS, startResumeSync } from './resumeSync'; class Target extends EventTarget { visibilityState: 'visible' | 'hidden' = 'visible'; } -const setup = (state: string | null = 'SYNCING') => { +const setup = () => { const doc = new Target(); const win = new Target(); - const mx = { - state, - getSyncState: () => mx.state, - retryImmediately: vi.fn(() => true), - stopClient: vi.fn(), - }; - const restart = vi.fn(async () => {}); - const stop = startResumeSync(mx, { - restart, - doc: doc as never, - win: win as never, - }); + const mx = { retryImmediately: vi.fn(() => true) }; + const stop = startResumeSync(mx, { doc: doc as never, win: win as never }); const hide = () => { doc.visibilityState = 'hidden'; doc.dispatchEvent(new Event('visibilitychange')); @@ -28,7 +18,7 @@ const setup = (state: string | null = 'SYNCING') => { doc.visibilityState = 'visible'; doc.dispatchEvent(new Event('visibilitychange')); }; - return { doc, win, mx, restart, stop, hide, show }; + return { doc, win, mx, stop, hide, show }; }; const settle = () => vi.advanceTimersByTimeAsync(RESUME_DEBOUNCE_MS); @@ -56,7 +46,7 @@ describe('startResumeSync', () => { t.hide(); await vi.advanceTimersByTimeAsync(MIN_HIDDEN_MS - 1); t.show(); - await vi.advanceTimersByTimeAsync(RESUME_DEBOUNCE_MS + HEALTH_CHECK_MS); + await vi.advanceTimersByTimeAsync(RESUME_DEBOUNCE_MS); expect(t.mx.retryImmediately).not.toHaveBeenCalled(); }); @@ -68,7 +58,6 @@ describe('startResumeSync', () => { t.win.dispatchEvent(Object.assign(new Event('pageshow'), { persisted: true })); await settle(); expect(t.mx.retryImmediately).toHaveBeenCalledTimes(1); - await vi.advanceTimersByTimeAsync(HEALTH_CHECK_MS); t.win.dispatchEvent(new Event('online')); await settle(); expect(t.mx.retryImmediately).toHaveBeenCalledTimes(2); @@ -83,50 +72,14 @@ describe('startResumeSync', () => { expect(t.mx.retryImmediately).toHaveBeenCalledTimes(1); }); - it('does not restart when sync is healthy after the check window', async () => { - for (const state of ['PREPARED', 'SYNCING', 'CATCHUP']) { - const t = setup(state); - t.win.dispatchEvent(new Event('online')); - await vi.advanceTimersByTimeAsync(RESUME_DEBOUNCE_MS + HEALTH_CHECK_MS); - expect(t.restart).not.toHaveBeenCalled(); - t.stop(); - } - }); - - it('does one stop/start when sync is still unhealthy, never overlapping', async () => { - const t = setup('RECONNECTING'); - let release!: () => void; - t.restart.mockImplementation( - () => - new Promise((r) => { - release = r; - }) - ); - t.win.dispatchEvent(new Event('online')); - await vi.advanceTimersByTimeAsync(RESUME_DEBOUNCE_MS + HEALTH_CHECK_MS); - expect(t.mx.stopClient).toHaveBeenCalledTimes(1); - expect(t.restart).toHaveBeenCalledTimes(1); - // a trigger while the restart is in flight does nothing - t.win.dispatchEvent(new Event('online')); - await vi.advanceTimersByTimeAsync(RESUME_DEBOUNCE_MS + HEALTH_CHECK_MS); - expect(t.mx.retryImmediately).toHaveBeenCalledTimes(1); - expect(t.restart).toHaveBeenCalledTimes(1); - release(); - await vi.advanceTimersByTimeAsync(0); - t.win.dispatchEvent(new Event('online')); - await settle(); - expect(t.mx.retryImmediately).toHaveBeenCalledTimes(2); - }); - it('removes listeners and cancels pending work on dispose', async () => { - const t = setup('ERROR'); + const t = setup(); t.win.dispatchEvent(new Event('online')); - await settle(); t.stop(); - await vi.advanceTimersByTimeAsync(HEALTH_CHECK_MS); - expect(t.restart).not.toHaveBeenCalled(); + await settle(); + expect(t.mx.retryImmediately).not.toHaveBeenCalled(); t.win.dispatchEvent(new Event('online')); await settle(); - expect(t.mx.retryImmediately).toHaveBeenCalledTimes(1); + expect(t.mx.retryImmediately).not.toHaveBeenCalled(); }); }); diff --git a/src/client/resumeSync.ts b/src/client/resumeSync.ts index 5f81547b..a09f39f8 100644 --- a/src/client/resumeSync.ts +++ b/src/client/resumeSync.ts @@ -1,60 +1,38 @@ /** The slice of MatrixClient this module drives; a MatrixClient satisfies it. */ export type SyncHost = { - getSyncState(): string | null; retryImmediately(): boolean; - stopClient(): void; }; -// matrix-js-sdk SyncState values for a live sync loop. -const HEALTHY: ReadonlySet = new Set(['PREPARED', 'SYNCING', 'CATCHUP']); - export const RESUME_DEBOUNCE_MS = 250; export const MIN_HIDDEN_MS = 5000; -export const HEALTH_CHECK_MS = 3000; -type ResumeSyncOptions = { +type ResumeSyncOptions = { /** Event sources and visibility; default to the browser globals. */ doc?: Pick; win?: Pick; - /** Restart the client after stopClient; must resolve once startClient has been issued. */ - restart: (mx: Host) => Promise; }; /** * A backgrounded tab or PWA leaves the sync long-poll dead until its 30 s timeout. * On return (visible after a real absence, pageshow from bfcache, online) nudge the - * poll with retryImmediately; if sync is still unhealthy after HEALTH_CHECK_MS, do one - * stop/start. Returns a disposer that removes every listener and timer. + * poll with retryImmediately. It never stops and restarts the client: stopClient() closes + * the rust crypto's OlmMachine for good and startClient() does not reopen it. A connection + * that stays dead is the watchdog's business. Returns a disposer that removes every + * listener and timer. */ -export const startResumeSync = ( - mx: Host, - { restart, doc = document, win = window }: ResumeSyncOptions +export const startResumeSync = ( + mx: SyncHost, + { doc = document, win = window }: ResumeSyncOptions = {} ): (() => void) => { let hiddenAt: number | undefined; let debounce: ReturnType | undefined; - let healthCheck: ReturnType | undefined; - let restarting = false; let disposed = false; - const check = async () => { - healthCheck = undefined; - if (disposed || restarting || HEALTHY.has(mx.getSyncState())) return; - restarting = true; - try { - mx.stopClient(); - await restart(mx); - } finally { - restarting = false; - } - }; - const resume = () => { if (debounce !== undefined) clearTimeout(debounce); debounce = setTimeout(() => { debounce = undefined; - if (disposed || restarting || healthCheck !== undefined) return; - mx.retryImmediately(); - healthCheck = setTimeout(check, HEALTH_CHECK_MS); + if (!disposed) mx.retryImmediately(); }, RESUME_DEBOUNCE_MS); }; @@ -78,7 +56,6 @@ export const startResumeSync = ( return () => { disposed = true; if (debounce !== undefined) clearTimeout(debounce); - if (healthCheck !== undefined) clearTimeout(healthCheck); doc.removeEventListener('visibilitychange', onVisibility); win.removeEventListener('pageshow', onPageShow); win.removeEventListener('online', resume); diff --git a/src/client/watchdog.test.ts b/src/client/watchdog.test.ts new file mode 100644 index 00000000..7d80f099 --- /dev/null +++ b/src/client/watchdog.test.ts @@ -0,0 +1,60 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { SEND_STALL_MS, WATCHDOG_INTERVAL_MS, startWatchdog } from './watchdog'; + +const setup = (over: Partial[0]> = {}) => { + const recover = vi.fn(); + const stop = startWatchdog({ + pendingAges: () => [], + isVisible: () => true, + isOnline: () => true, + recover, + ...over, + }); + return { recover, stop }; +}; +const tick = (n: number) => vi.advanceTimersByTime(WATCHDOG_INTERVAL_MS * n); + +describe('watchdog', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it('stays quiet while nothing is pending', () => { + const t = setup(); + tick(100); + expect(t.recover).not.toHaveBeenCalled(); + t.stop(); + }); + + it('recovers when a send has been pending past the stall threshold, not before', () => { + let age = 0; + const t = setup({ pendingAges: () => [age] }); + age = SEND_STALL_MS - 1; + tick(1); + expect(t.recover).not.toHaveBeenCalled(); + age = SEND_STALL_MS; + tick(1); + expect(t.recover).toHaveBeenCalledTimes(1); + t.stop(); + }); + + it('does nothing while the page is hidden or offline', () => { + const hidden = setup({ isVisible: () => false, pendingAges: () => [1e9] }); + const offline = setup({ isOnline: () => false, pendingAges: () => [1e9] }); + tick(50); + expect(hidden.recover).not.toHaveBeenCalled(); + expect(offline.recover).not.toHaveBeenCalled(); + hidden.stop(); + offline.stop(); + }); + + it('stop() ends it', () => { + const t = setup({ pendingAges: () => [1e9] }); + t.stop(); + tick(10); + expect(t.recover).not.toHaveBeenCalled(); + }); +}); diff --git a/src/client/watchdog.ts b/src/client/watchdog.ts new file mode 100644 index 00000000..6b97e8fc --- /dev/null +++ b/src/client/watchdog.ts @@ -0,0 +1,37 @@ +/** + * The page can go on looking alive while its connection is dead: sync polls are cut by their + * own timeout and js-sdk retries them, but a send has no timeout at all, so on a half-open + * connection it hangs SENDING and no request reaches the server. A send pending past + * `stallMs` is the sign, and `recover` (a guarded reload) the answer; the outbox then sends + * again what the reload interrupted. Sync state is not watched: ERROR is also what offline + * looks like, and js-sdk recovers from that itself. + * Nothing fires while the page is hidden (a frozen tab is not a dead connection) or offline + * (a reload there only blanks the page). + */ +export type WatchdogOptions = { + /** Ages in ms of this page's sends still pending. */ + pendingAges: () => number[]; + isVisible: () => boolean; + isOnline: () => boolean; + recover: () => void; + intervalMs?: number; + stallMs?: number; +}; + +export const WATCHDOG_INTERVAL_MS = 5000; +export const SEND_STALL_MS = 20_000; + +export const startWatchdog = ({ + pendingAges, + isVisible, + isOnline, + recover, + intervalMs = WATCHDOG_INTERVAL_MS, + stallMs = SEND_STALL_MS, +}: WatchdogOptions): (() => void) => { + const id = setInterval(() => { + if (!isVisible() || !isOnline()) return; + if (pendingAges().some((age) => age >= stallMs)) recover(); + }, intervalMs); + return () => clearInterval(id); +};