Merge pull request 'fix(status): status changes apply at once instead of waiting out the rate limit' (#469) from fix/status-save-speed into lotus
CI / Secret scan (gitleaks) (push) Successful in 19s
CI / Build & Quality Checks (push) Successful in 3m37s
CI / Docker image build & smoke test (push) Skipped
CI / Trigger Desktop Build (push) Successful in 1m7s
CI / Playwright smoke (e2e) (push) Successful in 29m10s
CI / Secret scan (gitleaks) (push) Successful in 19s
CI / Build & Quality Checks (push) Successful in 3m37s
CI / Docker image build & smoke test (push) Skipped
CI / Trigger Desktop Build (push) Successful in 1m7s
CI / Playwright smoke (e2e) (push) Successful in 29m10s
This commit was merged in pull request #469.
This commit is contained in:
@@ -38,11 +38,7 @@ import { SettingTile } from '../../../components/setting-tile';
|
||||
import { useMatrixClient } from '../../../hooks/useMatrixClient';
|
||||
import { getAccountData, setAccountData } from '../../../utils/accountData';
|
||||
import { presenceStateFromSetting } from '../../../hooks/usePresenceUpdater';
|
||||
import {
|
||||
describePresenceError,
|
||||
setPresenceWithRetry,
|
||||
writePresence,
|
||||
} from '../../../utils/presenceWrite';
|
||||
import { describePresenceError, queuePresenceWrite } from '../../../utils/presenceWrite';
|
||||
import { useSetting } from '../../../state/hooks/settings';
|
||||
import { settingsAtom } from '../../../state/settings';
|
||||
import { UserProfile, useUserProfile } from '../../../hooks/useUserProfile';
|
||||
@@ -419,6 +415,10 @@ function ProfileStatus() {
|
||||
// until our own echo lands or a short window elapses (bounded so a genuinely
|
||||
// dropped echo can't block real cross-device updates forever).
|
||||
const pendingAppliedRef = useRef<{ value: string; ts: number } | null>(null);
|
||||
// The status as this device last applied or synced it. "Save" compares with
|
||||
// this, not with the server echo, so it switches off the moment a status is
|
||||
// applied instead of staying lit until the (rate-limited) write lands.
|
||||
const [savedMsg, setSavedMsg] = useState<string>(initialStatus);
|
||||
const [clearAfter, setClearAfter] = useState('0');
|
||||
const [emojiAnchor, setEmojiAnchor] = useState<RectCords>();
|
||||
const { presets, addPreset, removePreset } = useStatusPresets();
|
||||
@@ -442,7 +442,7 @@ function ProfileStatus() {
|
||||
if (pending) {
|
||||
if (remoteStatus === pending.value) {
|
||||
pendingAppliedRef.current = null; // our own echo landed — accept it
|
||||
} else if (Date.now() - pending.ts < 15_000) {
|
||||
} else if (Date.now() - pending.ts < 30_000) {
|
||||
return; // stale echo of the previous status; ignore within the window
|
||||
} else {
|
||||
pendingAppliedRef.current = null; // window elapsed — accept external changes
|
||||
@@ -451,6 +451,7 @@ function ProfileStatus() {
|
||||
// [Gitea #283] "dnd" is the Do Not Disturb sentinel, not something the user
|
||||
// typed: don't show it in the status field or keep it as their status.
|
||||
if (remoteStatus === DND_STATUS_SENTINEL) return;
|
||||
setSavedMsg(remoteStatus);
|
||||
if (remoteStatus) {
|
||||
setStatusMsg(remoteStatus);
|
||||
localStorage.setItem(STATUS_MSG_KEY(userId), remoteStatus);
|
||||
@@ -467,9 +468,10 @@ function ProfileStatus() {
|
||||
useCallback(
|
||||
(msg: string) =>
|
||||
// Synapse allows ONE presence write per 10 s per user (shared by all
|
||||
// devices) and our heartbeat spends that budget too — wait out a 429
|
||||
// instead of failing the user's save.
|
||||
setPresenceWithRetry(mx, {
|
||||
// devices) and our heartbeat spends that budget too. The shared queue
|
||||
// waits out a 429 and folds quick successive changes into one write;
|
||||
// the UI doesn't wait on it (the status is applied locally at once).
|
||||
queuePresenceWrite(mx, {
|
||||
// Derive presence from the user's chosen setting so writing a status
|
||||
// never overrides Invisible/DND/Idle (e.g. outing an Invisible user).
|
||||
presence: presenceStateFromSetting(presenceStatus, hidePresence),
|
||||
@@ -498,6 +500,7 @@ function ProfileStatus() {
|
||||
(rawMsg: string, clearAfterValue: string) => {
|
||||
statusDirtyRef.current = false;
|
||||
const msg = rawMsg.trim();
|
||||
setSavedMsg(msg);
|
||||
// Guard against a stale presence echo reverting this value (see the sync effect).
|
||||
pendingAppliedRef.current = { value: msg, ts: Date.now() };
|
||||
noteLocalStatusWrite();
|
||||
@@ -527,19 +530,17 @@ function ProfileStatus() {
|
||||
|
||||
const handleSubmit: FormEventHandler<HTMLFormElement> = (evt) => {
|
||||
evt.preventDefault();
|
||||
if (saving) return;
|
||||
applyStatus(statusMsg, clearAfter);
|
||||
};
|
||||
|
||||
// Preset click = one-click apply: reflect it in the inputs and save immediately.
|
||||
const applyPreset = useCallback(
|
||||
(preset: StatusPreset) => {
|
||||
if (saving) return;
|
||||
setStatusMsg(preset.label);
|
||||
setClearAfter(preset.clearAfter);
|
||||
applyStatus(preset.label, preset.clearAfter);
|
||||
},
|
||||
[saving, applyStatus],
|
||||
[applyStatus],
|
||||
);
|
||||
|
||||
const handleSaveCurrent = useCallback(() => {
|
||||
@@ -560,17 +561,18 @@ function ProfileStatus() {
|
||||
pendingAppliedRef.current = { value: '', ts: Date.now() };
|
||||
noteLocalStatusWrite();
|
||||
setStatusMsg('');
|
||||
setSavedMsg('');
|
||||
localStorage.removeItem(STATUS_MSG_KEY(userId));
|
||||
localStorage.removeItem(STATUS_EXPIRY_KEY(userId));
|
||||
writeStatusExpiry(mx, '').catch(() => undefined);
|
||||
// Preserve the user's chosen presence when clearing the status message.
|
||||
writePresence(mx, {
|
||||
queuePresenceWrite(mx, {
|
||||
presence: presenceStateFromSetting(presenceStatus, hidePresence),
|
||||
status_msg: '',
|
||||
}).catch(() => undefined);
|
||||
};
|
||||
|
||||
const hasChanges = statusMsg !== (presence?.status ?? '');
|
||||
const hasChanges = statusMsg.trim() !== savedMsg;
|
||||
|
||||
return (
|
||||
<SettingTile
|
||||
@@ -599,7 +601,6 @@ function ProfileStatus() {
|
||||
variant="Secondary"
|
||||
fill="Soft"
|
||||
radii="Pill"
|
||||
disabled={saving}
|
||||
onClick={() => applyPreset(preset)}
|
||||
>
|
||||
<Text as="span" size="B300">
|
||||
@@ -625,7 +626,6 @@ function ProfileStatus() {
|
||||
variant="Secondary"
|
||||
fill="Soft"
|
||||
radii="Pill"
|
||||
disabled={saving}
|
||||
onClick={() => applyPreset(preset)}
|
||||
>
|
||||
<Text as="span" size="B300">
|
||||
@@ -651,7 +651,7 @@ function ProfileStatus() {
|
||||
fill="Soft"
|
||||
radii="Pill"
|
||||
outlined
|
||||
disabled={saving || !statusMsg.trim()}
|
||||
disabled={!statusMsg.trim()}
|
||||
onClick={handleSaveCurrent}
|
||||
before={<Icon size="100" src={Icons.Plus} />}
|
||||
>
|
||||
@@ -662,7 +662,7 @@ function ProfileStatus() {
|
||||
</Box>
|
||||
</Box>
|
||||
|
||||
<Box as="form" onSubmit={handleSubmit} gap="200" alignItems="Center" aria-disabled={saving}>
|
||||
<Box as="form" onSubmit={handleSubmit} gap="200" alignItems="Center">
|
||||
<Box grow="Yes" direction="Column" gap="100">
|
||||
<Input
|
||||
name="statusMsgInput"
|
||||
@@ -672,7 +672,6 @@ function ProfileStatus() {
|
||||
placeholder="What's on your mind?"
|
||||
variant="Secondary"
|
||||
radii="300"
|
||||
readOnly={saving}
|
||||
maxLength={64}
|
||||
/>
|
||||
<Text
|
||||
@@ -733,13 +732,17 @@ function ProfileStatus() {
|
||||
fill={hasChanges ? 'Solid' : 'Soft'}
|
||||
outlined
|
||||
radii="300"
|
||||
disabled={!hasChanges || saving}
|
||||
disabled={!hasChanges}
|
||||
type="submit"
|
||||
>
|
||||
{saving && <Spinner variant="Success" fill="Solid" size="300" />}
|
||||
<Text size="B400">Save</Text>
|
||||
</Button>
|
||||
</Box>
|
||||
{saving && (
|
||||
<Text size="T200" priority="300" role="status">
|
||||
Status set. Sharing it with others, this can take a few seconds.
|
||||
</Text>
|
||||
)}
|
||||
{saveState.status === AsyncStatus.Error && (
|
||||
<Text size="T200" style={{ color: color.Critical.Main }}>
|
||||
{describePresenceError(saveState.error)}
|
||||
@@ -764,7 +767,6 @@ function ProfileStatus() {
|
||||
radii="300"
|
||||
type="button"
|
||||
onClick={handleClear}
|
||||
disabled={saving}
|
||||
>
|
||||
<Text size="B300">Clear Status</Text>
|
||||
</Button>
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { useEffect, useRef } from 'react';
|
||||
import { SetPresence } from 'matrix-js-sdk';
|
||||
import { PresenceDeduper, PresenceWrite, writePresence } from '../utils/presenceWrite';
|
||||
import { PresenceDeduper, PresenceWrite, queuePresenceWrite } from '../utils/presenceWrite';
|
||||
import { useMatrixClient } from './useMatrixClient';
|
||||
import { getLastLocalStatusWrite, pickHeartbeatStatus } from '../utils/ownStatusSync';
|
||||
import { useSetting } from '../state/hooks/settings';
|
||||
@@ -70,9 +70,25 @@ export function usePresenceUpdater() {
|
||||
// by all devices) and /sync already keeps us online — only send when the
|
||||
// state or status actually changes, so a manual status save isn't 429'd
|
||||
// by a heartbeat that changed nothing.
|
||||
// What the server reported in the last resolveStatus() (consumed by send).
|
||||
let serverNow: { presence: string; status_msg?: string } | undefined;
|
||||
const send = (write: PresenceWrite, label: string) => {
|
||||
const server = serverNow;
|
||||
serverNow = undefined;
|
||||
if (!deduper.shouldSend(write)) return Promise.resolve();
|
||||
return writePresence(mx, write)
|
||||
// The server already shows exactly this (e.g. /sync keeps us online and
|
||||
// the status is unchanged): writing it again would only spend the
|
||||
// 10-second budget a status save may need next.
|
||||
if (
|
||||
server &&
|
||||
server.presence === write.presence &&
|
||||
(server.status_msg ?? '') === (write.status_msg ?? '')
|
||||
) {
|
||||
deduper.sent(write);
|
||||
return Promise.resolve();
|
||||
}
|
||||
// Shared queue with the status save: waits out a 429, newest write wins.
|
||||
return queuePresenceWrite(mx, write)
|
||||
.then(() => deduper.sent(write))
|
||||
.catch((err) => warnPresenceFailure(label, err));
|
||||
};
|
||||
@@ -84,6 +100,7 @@ export function usePresenceUpdater() {
|
||||
const local = readStatus();
|
||||
if (!userId) return local;
|
||||
const server = await mx.getPresence(userId).catch(() => undefined);
|
||||
serverNow = server;
|
||||
const status = pickHeartbeatStatus({
|
||||
local,
|
||||
server,
|
||||
|
||||
@@ -1,7 +1,13 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { MatrixError } from 'matrix-js-sdk';
|
||||
import { PresenceDeduper, describePresenceError, setPresenceWithRetry } from './presenceWrite';
|
||||
import { IPresenceOpts, MatrixError } from 'matrix-js-sdk';
|
||||
import {
|
||||
PresenceDeduper,
|
||||
describePresenceError,
|
||||
isPresenceWritePending,
|
||||
queuePresenceWrite,
|
||||
setPresenceWithRetry,
|
||||
} from './presenceWrite';
|
||||
|
||||
const limited = (retryMs = 3000) =>
|
||||
new MatrixError(
|
||||
@@ -79,3 +85,43 @@ test('error wording', () => {
|
||||
);
|
||||
assert.match(describePresenceError(new TypeError('Failed to fetch')), /couldn't reach/);
|
||||
});
|
||||
|
||||
test('[status save] queuePresenceWrite: newest wins, waits out 429 once, all callers settle', async () => {
|
||||
const sent: string[] = [];
|
||||
let calls = 0;
|
||||
const sleeps: number[] = [];
|
||||
const mx = {
|
||||
setPresence: async (w: IPresenceOpts) => {
|
||||
calls += 1;
|
||||
if (calls === 1) {
|
||||
sent.push(`429:${w.status_msg}`);
|
||||
throw new MatrixError({ errcode: 'M_LIMIT_EXCEEDED', retry_after_ms: 7000 }, 429);
|
||||
}
|
||||
sent.push(`ok:${w.status_msg}`);
|
||||
},
|
||||
};
|
||||
const sleepFn = async (ms: number) => {
|
||||
sleeps.push(ms);
|
||||
};
|
||||
// 'a' goes out at once; 'b' and 'c' arrive while it waits out the limit.
|
||||
const first = queuePresenceWrite(mx, { presence: 'online', status_msg: 'a' }, { sleepFn });
|
||||
const second = queuePresenceWrite(mx, { presence: 'online', status_msg: 'b' }, { sleepFn });
|
||||
const third = queuePresenceWrite(mx, { presence: 'online', status_msg: 'c' }, { sleepFn });
|
||||
assert.equal(isPresenceWritePending(mx), true);
|
||||
await Promise.all([first, second, third]);
|
||||
// 'a' was 429'd; after the wait only the newest ('c') goes out — 'a' isn't
|
||||
// retried and 'b' is never sent.
|
||||
assert.deepEqual(sent, ['429:a', 'ok:c']);
|
||||
assert.deepEqual(sleeps, [7000]);
|
||||
assert.equal(isPresenceWritePending(mx), false);
|
||||
});
|
||||
|
||||
test('[status save] queuePresenceWrite: a failure rejects the callers of that write', async () => {
|
||||
const mx = {
|
||||
setPresence: async () => {
|
||||
throw new MatrixError({ errcode: 'M_FORBIDDEN' }, 403);
|
||||
},
|
||||
};
|
||||
await assert.rejects(queuePresenceWrite(mx, { presence: 'online', status_msg: 'x' }));
|
||||
assert.equal(isPresenceWritePending(mx), false);
|
||||
});
|
||||
|
||||
@@ -94,3 +94,78 @@ export const describePresenceError = (e: unknown): string => {
|
||||
}
|
||||
return "Failed to save status — couldn't reach the server. Try again.";
|
||||
};
|
||||
|
||||
type PresenceQueue = {
|
||||
pending?: PresenceWrite;
|
||||
waiters: { resolve: () => void; reject: (e: unknown) => void }[];
|
||||
running: boolean;
|
||||
};
|
||||
const queues = new WeakMap<object, PresenceQueue>();
|
||||
|
||||
/**
|
||||
* One presence writer per client, newest write wins. The status save and the
|
||||
* heartbeat share Synapse's one-write-per-10-s budget, so instead of each
|
||||
* waiting out 429s on its own (and racing), writes queue here: anything still
|
||||
* waiting is replaced by the newer write, which goes out as soon as the server
|
||||
* allows. Each caller's promise settles with the outcome of the write that
|
||||
* finally carried (or superseded) theirs.
|
||||
*/
|
||||
export function queuePresenceWrite(
|
||||
mx: Pick<MatrixClient, 'setPresence'>,
|
||||
write: PresenceWrite,
|
||||
opts: { maxWaitMs?: number; sleepFn?: (ms: number) => Promise<void> } = {},
|
||||
): Promise<void> {
|
||||
let q = queues.get(mx);
|
||||
if (!q) {
|
||||
q = { waiters: [], running: false };
|
||||
queues.set(mx, q);
|
||||
}
|
||||
const queue = q;
|
||||
queue.pending = write;
|
||||
const settled = new Promise<void>((resolve, reject) => {
|
||||
queue.waiters.push({ resolve, reject });
|
||||
});
|
||||
if (!queue.running) {
|
||||
queue.running = true;
|
||||
const maxWaitMs = opts.maxWaitMs ?? 25_000;
|
||||
const sleepFn = opts.sleepFn ?? sleep;
|
||||
(async () => {
|
||||
let waited = 0;
|
||||
while (queue.pending) {
|
||||
const next = queue.pending;
|
||||
queue.pending = undefined;
|
||||
const waiters = queue.waiters.splice(0);
|
||||
try {
|
||||
// eslint-disable-next-line no-await-in-loop
|
||||
await writePresence(mx, next);
|
||||
waited = 0;
|
||||
if (queue.pending) queue.waiters.unshift(...waiters);
|
||||
else waiters.forEach((w) => w.resolve());
|
||||
} catch (e) {
|
||||
const wait =
|
||||
e instanceof MatrixError && e.httpStatus === 429
|
||||
? Math.min(e.getRetryAfterMs() ?? 10_500, 15_000)
|
||||
: undefined;
|
||||
if (wait !== undefined && waited + wait <= maxWaitMs) {
|
||||
waited += wait;
|
||||
// eslint-disable-next-line no-await-in-loop
|
||||
await sleepFn(wait);
|
||||
// After the wait, send whatever is newest: a change made meanwhile
|
||||
// replaces this one rather than queueing behind it.
|
||||
queue.pending ??= next;
|
||||
queue.waiters.unshift(...waiters);
|
||||
} else if (queue.pending) {
|
||||
queue.waiters.unshift(...waiters);
|
||||
} else {
|
||||
waiters.forEach((w) => w.reject(e));
|
||||
}
|
||||
}
|
||||
}
|
||||
queue.running = false;
|
||||
})();
|
||||
}
|
||||
return settled;
|
||||
}
|
||||
|
||||
/** Whether a presence write is queued or in flight for this client. */
|
||||
export const isPresenceWritePending = (mx: object): boolean => queues.get(mx)?.running ?? false;
|
||||
|
||||
Reference in New Issue
Block a user