client: watchdog, resend on reload, outbox for every send, no in-place client restart
Sends have no timeout and sync polls do, so a dead connection leaves a send
SENDING forever with no request reaching the server while its echo looked
delivered. A watchdog (send pending past 20 s, sync ERROR/STOPPED for 30 s)
calls a guarded reload; what the reload interrupts is sent again under its
own txn id. Uploads and stickers go through the outbox. Resend takes a fresh
txn id while the SDK still holds the failed echo. The resume-sync recovery no
longer calls stopClient/startClient, which closes the rust crypto for good
("null pointer passed to rust" on later encrypted sends).
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Void
parent
a92f6c4294
commit
77edebff9c
14 changed files with 523 additions and 155 deletions
|
|
@ -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<HTMLDivElement, RoomInputProps>(
|
|||
});
|
||||
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<string, unknown>));
|
||||
};
|
||||
|
||||
const submit = useCallback(() => {
|
||||
|
|
@ -492,11 +493,7 @@ export const RoomInput = forwardRef<HTMLDivElement, RoomInputProps>(
|
|||
await getImageUrlBlob(stickerUrl)
|
||||
);
|
||||
|
||||
mx.sendEvent(roomId, EventType.Sticker, {
|
||||
body: label,
|
||||
url: mxc,
|
||||
info,
|
||||
});
|
||||
sendTracked(mx, roomId, { body: label, url: mxc, info }, EventType.Sticker);
|
||||
};
|
||||
|
||||
return (
|
||||
|
|
|
|||
|
|
@ -1,57 +1,24 @@
|
|||
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<string, string>();
|
||||
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<unknown>) =>
|
||||
({ 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: 'Still sending: slow connection',
|
||||
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/);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -3,32 +3,40 @@ 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<void> =>
|
||||
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<string, unknown>
|
||||
): Promise<void> => {
|
||||
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<OutboxEntry, 'status' | 'ts' | 'error'>,
|
||||
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', failed: false, canResend: true };
|
||||
}
|
||||
return { label: 'Sending…', failed: false, canResend: false };
|
||||
};
|
||||
|
||||
export const useRoomOutbox = (userId: string, roomId: string): OutboxEntry[] => {
|
||||
|
|
@ -42,19 +50,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 +81,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 (
|
||||
<Box key={entry.txnId} alignItems="Center" gap="200" data-outbox-status={entry.status}>
|
||||
<Box grow="Yes" direction="Column">
|
||||
<Text
|
||||
size="T200"
|
||||
style={{ color: failed ? color.Critical.Main : color.Surface.OnContainer }}
|
||||
style={{ color: view.failed ? color.Critical.Main : color.Surface.OnContainer }}
|
||||
>
|
||||
{(() => {
|
||||
if (failed) return `Not sent${entry.error ? `: ${entry.error}` : ''}`;
|
||||
return interrupted ? 'Interrupted by a reload: may not have been sent' : 'Sending…';
|
||||
})()}
|
||||
{view.label}
|
||||
</Text>
|
||||
<Text size="T300" truncate>
|
||||
{String(entry.content.body ?? '')}
|
||||
</Text>
|
||||
</Box>
|
||||
{canResend && (
|
||||
{view.canResend && (
|
||||
<>
|
||||
<Chip
|
||||
variant="Surface"
|
||||
radii="Pill"
|
||||
onClick={() => {
|
||||
retryOutbox(userId, entry.txnId);
|
||||
attempt(mx, userId, entry);
|
||||
}}
|
||||
>
|
||||
<Text size="B300">Resend</Text>
|
||||
</Chip>
|
||||
<Chip variant="Surface" radii="Pill" onClick={() => removeOutbox(userId, entry.txnId)}>
|
||||
<Text size="B300">Discard</Text>
|
||||
</Chip>
|
||||
<Chip variant="Surface" radii="Pill" onClick={() => resendTracked(mx, entry)}>
|
||||
<Text size="B300">Resend</Text>
|
||||
</Chip>
|
||||
<Chip
|
||||
variant="Surface"
|
||||
radii="Pill"
|
||||
onClick={() => removeOutbox(userId, entry.txnId)}
|
||||
>
|
||||
<Text size="B300">Discard</Text>
|
||||
</Chip>
|
||||
</>
|
||||
)}
|
||||
</Box>
|
||||
|
|
|
|||
|
|
@ -664,7 +664,14 @@ export function RoomTimeline({ room, eventId, roomInputRef, editor }: RoomTimeli
|
|||
|
||||
useLocalEchoRefresh(
|
||||
room,
|
||||
useCallback(() => setTimeline((ct) => ({ ...ct })), [])
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -88,11 +88,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);
|
||||
|
|
|
|||
|
|
@ -15,6 +15,8 @@ export type OutboxStatus = 'pending' | 'failed';
|
|||
export type OutboxEntry = {
|
||||
txnId: string;
|
||||
roomId: string;
|
||||
/** Event type; absent means m.room.message. */
|
||||
type?: string;
|
||||
content: Record<string, unknown>;
|
||||
status: OutboxStatus;
|
||||
error?: string;
|
||||
|
|
@ -155,16 +157,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 }
|
||||
: e
|
||||
),
|
||||
store
|
||||
);
|
||||
|
|
|
|||
132
src/app/state/room/outboxSend.test.ts
Normal file
132
src/app/state/room/outboxSend.test.ts
Normal file
|
|
@ -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<string, string>();
|
||||
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<unknown>) =>
|
||||
({ 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');
|
||||
});
|
||||
});
|
||||
66
src/app/state/room/outboxSend.ts
Normal file
66
src/app/state/room/outboxSend.ts
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
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<void> => {
|
||||
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<string, unknown>,
|
||||
type?: string
|
||||
): Promise<void> => {
|
||||
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<void> => {
|
||||
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<void[]> => {
|
||||
const userId = mx.getUserId() ?? '';
|
||||
return Promise.all(
|
||||
loadOutbox(userId)
|
||||
.filter((e) => e.status === 'pending' && e.ts < since)
|
||||
.map((e) => resendTracked(mx, e))
|
||||
);
|
||||
};
|
||||
20
src/client/initMatrix.test.ts
Normal file
20
src/client/initMatrix.test.ts
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
import { describe, expect, it, vi } from 'vitest';
|
||||
import { RECOVER_RELOAD_MIN_GAP_MS, reloadToRecoverSync } from './initMatrix';
|
||||
|
||||
const memory = () => {
|
||||
const m = new Map<string, string>();
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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 { loadOutbox } from '../app/state/room/composerStore';
|
||||
import { resendInterrupted } from '../app/state/room/outboxSend';
|
||||
|
||||
type Session = {
|
||||
baseUrl: string;
|
||||
|
|
@ -46,9 +56,42 @@ const startSync = (mx: MatrixClient) =>
|
|||
lazyLoadMembers: true,
|
||||
});
|
||||
|
||||
const RECOVER_RELOAD_KEY = 'ratatoskr.recoverReloadAt';
|
||||
export const RECOVER_RELOAD_MIN_GAP_MS = 60_000;
|
||||
|
||||
/** Reload to bring a dead sync back, at most once a minute, so a server that is down cannot loop us. */
|
||||
export const reloadToRecoverSync = (
|
||||
now: number = Date.now(),
|
||||
store: Pick<Storage, 'getItem' | 'setItem'> = sessionStorage,
|
||||
reload: () => void = () => window.location.reload()
|
||||
): boolean => {
|
||||
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));
|
||||
reload();
|
||||
return true;
|
||||
};
|
||||
|
||||
export const startClient = async (mx: MatrixClient) => {
|
||||
await startSync(mx);
|
||||
startResumeSync(mx, { restart: startSync });
|
||||
const recover = () => reloadToRecoverSync();
|
||||
startResumeSync(mx, { recover });
|
||||
startWatchdog({
|
||||
getSyncState: () => mx.getSyncState(),
|
||||
pendingAges: () =>
|
||||
loadOutbox(mx.getUserId() ?? '')
|
||||
.filter((e) => e.status === 'pending')
|
||||
.map((e) => Date.now() - e.ts),
|
||||
isVisible: () => document.visibilityState === 'visible',
|
||||
recover,
|
||||
});
|
||||
// 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) => {
|
||||
|
|
|
|||
|
|
@ -12,11 +12,10 @@ const setup = (state: string | null = 'SYNCING') => {
|
|||
state,
|
||||
getSyncState: () => mx.state,
|
||||
retryImmediately: vi.fn(() => true),
|
||||
stopClient: vi.fn(),
|
||||
};
|
||||
const restart = vi.fn(async () => {});
|
||||
const recover = vi.fn();
|
||||
const stop = startResumeSync(mx, {
|
||||
restart,
|
||||
recover,
|
||||
doc: doc as never,
|
||||
win: win as never,
|
||||
});
|
||||
|
|
@ -28,7 +27,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, recover, stop, hide, show };
|
||||
};
|
||||
|
||||
const settle = () => vi.advanceTimersByTimeAsync(RESUME_DEBOUNCE_MS);
|
||||
|
|
@ -88,34 +87,18 @@ describe('startResumeSync', () => {
|
|||
const t = setup(state);
|
||||
t.win.dispatchEvent(new Event('online'));
|
||||
await vi.advanceTimersByTimeAsync(RESUME_DEBOUNCE_MS + HEALTH_CHECK_MS);
|
||||
expect(t.restart).not.toHaveBeenCalled();
|
||||
expect(t.recover).not.toHaveBeenCalled();
|
||||
t.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it('does one stop/start when sync is still unhealthy, never overlapping', async () => {
|
||||
it('calls recover once when sync is still unhealthy, and never stops the client itself', async () => {
|
||||
const t = setup('RECONNECTING');
|
||||
let release!: () => void;
|
||||
t.restart.mockImplementation(
|
||||
() =>
|
||||
new Promise<void>((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);
|
||||
expect(t.recover).toHaveBeenCalledTimes(1);
|
||||
expect(t.recover).toHaveBeenCalledWith(t.mx);
|
||||
expect(Object.keys(t.mx)).not.toContain('stopClient');
|
||||
});
|
||||
|
||||
it('removes listeners and cancels pending work on dispose', async () => {
|
||||
|
|
@ -124,7 +107,7 @@ describe('startResumeSync', () => {
|
|||
await settle();
|
||||
t.stop();
|
||||
await vi.advanceTimersByTimeAsync(HEALTH_CHECK_MS);
|
||||
expect(t.restart).not.toHaveBeenCalled();
|
||||
expect(t.recover).not.toHaveBeenCalled();
|
||||
t.win.dispatchEvent(new Event('online'));
|
||||
await settle();
|
||||
expect(t.mx.retryImmediately).toHaveBeenCalledTimes(1);
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@
|
|||
export type SyncHost = {
|
||||
getSyncState(): string | null;
|
||||
retryImmediately(): boolean;
|
||||
stopClient(): void;
|
||||
};
|
||||
|
||||
// matrix-js-sdk SyncState values for a live sync loop.
|
||||
|
|
@ -16,43 +15,40 @@ type ResumeSyncOptions<Host> = {
|
|||
/** Event sources and visibility; default to the browser globals. */
|
||||
doc?: Pick<Document, 'addEventListener' | 'removeEventListener' | 'visibilityState'>;
|
||||
win?: Pick<Window, 'addEventListener' | 'removeEventListener'>;
|
||||
/** Restart the client after stopClient; must resolve once startClient has been issued. */
|
||||
restart: (mx: Host) => Promise<void>;
|
||||
/**
|
||||
* Called once when sync is still unhealthy after the nudge. It must not stop and restart the
|
||||
* client in place: stopClient() closes the rust crypto's OlmMachine for good and startClient()
|
||||
* does not reopen it, so every later encrypted send fails with "null pointer passed to rust".
|
||||
* A page reload is the only restart that brings crypto back.
|
||||
*/
|
||||
recover: (mx: Host) => void;
|
||||
};
|
||||
|
||||
/**
|
||||
* 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; if sync is still unhealthy after HEALTH_CHECK_MS, call recover once. Returns a disposer that removes every listener and timer.
|
||||
*/
|
||||
export const startResumeSync = <Host extends SyncHost>(
|
||||
mx: Host,
|
||||
{ restart, doc = document, win = window }: ResumeSyncOptions<Host>
|
||||
{ recover, doc = document, win = window }: ResumeSyncOptions<Host>
|
||||
): (() => void) => {
|
||||
let hiddenAt: number | undefined;
|
||||
let debounce: ReturnType<typeof setTimeout> | undefined;
|
||||
let healthCheck: ReturnType<typeof setTimeout> | undefined;
|
||||
let restarting = false;
|
||||
let disposed = false;
|
||||
|
||||
const check = async () => {
|
||||
const check = () => {
|
||||
healthCheck = undefined;
|
||||
if (disposed || restarting || HEALTHY.has(mx.getSyncState())) return;
|
||||
restarting = true;
|
||||
try {
|
||||
mx.stopClient();
|
||||
await restart(mx);
|
||||
} finally {
|
||||
restarting = false;
|
||||
}
|
||||
if (disposed || HEALTHY.has(mx.getSyncState())) return;
|
||||
recover(mx);
|
||||
};
|
||||
|
||||
const resume = () => {
|
||||
if (debounce !== undefined) clearTimeout(debounce);
|
||||
debounce = setTimeout(() => {
|
||||
debounce = undefined;
|
||||
if (disposed || restarting || healthCheck !== undefined) return;
|
||||
if (disposed || healthCheck !== undefined) return;
|
||||
mx.retryImmediately();
|
||||
healthCheck = setTimeout(check, HEALTH_CHECK_MS);
|
||||
}, RESUME_DEBOUNCE_MS);
|
||||
|
|
|
|||
83
src/client/watchdog.test.ts
Normal file
83
src/client/watchdog.test.ts
Normal file
|
|
@ -0,0 +1,83 @@
|
|||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { SEND_STALL_MS, SYNC_BAD_TICKS, WATCHDOG_INTERVAL_MS, startWatchdog } from './watchdog';
|
||||
|
||||
const setup = (over: Partial<Parameters<typeof startWatchdog>[0]> = {}) => {
|
||||
const recover = vi.fn();
|
||||
const o = {
|
||||
getSyncState: () => 'SYNCING' as string | null,
|
||||
pendingAges: () => [] as number[],
|
||||
isVisible: () => true,
|
||||
recover,
|
||||
...over,
|
||||
};
|
||||
const stop = startWatchdog(o);
|
||||
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 sync is healthy and 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', () => {
|
||||
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.each(['ERROR', 'STOPPED'])('recovers when sync sits in %s for the bad-tick count, not before', (state) => {
|
||||
const t = setup({ getSyncState: () => state });
|
||||
tick(SYNC_BAD_TICKS - 1);
|
||||
expect(t.recover).not.toHaveBeenCalled();
|
||||
tick(1);
|
||||
expect(t.recover).toHaveBeenCalledTimes(1);
|
||||
t.stop();
|
||||
});
|
||||
|
||||
it('a recovered sync resets the count; RECONNECTING and a null state are js-sdk working, not bad', () => {
|
||||
let state: string | null = 'ERROR';
|
||||
const t = setup({ getSyncState: () => state });
|
||||
tick(SYNC_BAD_TICKS - 1);
|
||||
state = 'SYNCING';
|
||||
tick(1);
|
||||
state = 'ERROR';
|
||||
tick(SYNC_BAD_TICKS - 1);
|
||||
state = 'RECONNECTING';
|
||||
tick(50);
|
||||
state = null;
|
||||
tick(50);
|
||||
expect(t.recover).not.toHaveBeenCalled();
|
||||
t.stop();
|
||||
});
|
||||
|
||||
it('does nothing while the page is hidden', () => {
|
||||
const t = setup({ isVisible: () => false, pendingAges: () => [1e9], getSyncState: () => 'STOPPED' });
|
||||
tick(50);
|
||||
expect(t.recover).not.toHaveBeenCalled();
|
||||
t.stop();
|
||||
});
|
||||
|
||||
it('stop() ends it', () => {
|
||||
const t = setup({ pendingAges: () => [1e9] });
|
||||
t.stop();
|
||||
tick(10);
|
||||
expect(t.recover).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
48
src/client/watchdog.ts
Normal file
48
src/client/watchdog.ts
Normal file
|
|
@ -0,0 +1,48 @@
|
|||
/**
|
||||
* The page can go on looking alive while its connection is dead: sync polls get cut by their
|
||||
* own timeout, and a send has no timeout at all, so it hangs unsent while its echo sits in the
|
||||
* composer's room. This notices the two signs and calls `recover` (a guarded reload; the
|
||||
* outbox then resends what was interrupted).
|
||||
* - a send pending past `stallMs`
|
||||
* - sync stuck in ERROR or STOPPED for `syncBadTicks` consecutive ticks
|
||||
* Nothing fires while the page is hidden: a frozen background tab is not a dead connection.
|
||||
*/
|
||||
export type WatchdogOptions = {
|
||||
getSyncState: () => string | null;
|
||||
/** Ages in ms of sends still pending. */
|
||||
pendingAges: () => number[];
|
||||
isVisible: () => boolean;
|
||||
recover: () => void;
|
||||
intervalMs?: number;
|
||||
stallMs?: number;
|
||||
syncBadTicks?: number;
|
||||
};
|
||||
|
||||
export const WATCHDOG_INTERVAL_MS = 5000;
|
||||
export const SEND_STALL_MS = 20_000;
|
||||
export const SYNC_BAD_TICKS = 6;
|
||||
|
||||
export const startWatchdog = ({
|
||||
getSyncState,
|
||||
pendingAges,
|
||||
isVisible,
|
||||
recover,
|
||||
intervalMs = WATCHDOG_INTERVAL_MS,
|
||||
stallMs = SEND_STALL_MS,
|
||||
syncBadTicks = SYNC_BAD_TICKS,
|
||||
}: WatchdogOptions): (() => void) => {
|
||||
let badTicks = 0;
|
||||
const id = setInterval(() => {
|
||||
if (!isVisible()) {
|
||||
badTicks = 0;
|
||||
return;
|
||||
}
|
||||
const state = getSyncState();
|
||||
badTicks = state === 'ERROR' || state === 'STOPPED' ? badTicks + 1 : 0;
|
||||
if (badTicks >= syncBadTicks || pendingAges().some((age) => age >= stallMs)) {
|
||||
badTicks = 0;
|
||||
recover();
|
||||
}
|
||||
}, intervalMs);
|
||||
return () => clearInterval(id);
|
||||
};
|
||||
Loading…
Reference in a new issue