From 197563434c0056e07c21d387c1278432f7708f78 Mon Sep 17 00:00:00 2001 From: Hassan Abdel-Rahman Date: Wed, 21 Oct 2026 18:23:00 -0400 Subject: [PATCH 1/3] Deliver realm events through one bounded, paced line per process NodeAdapter.broadcastRealmEvent started an independent send loop per event, so a burst of events was pushed at synapse with no limit. Route every broadcast through a per-process RealmEventDispatcher that caps sends in flight, rate-limits them, takes realms in turn, keeps each realm's events in order, merges queued initiation and update events, bounds queued bytes, and treats 429/502/503/504 or an unreachable homeserver as back-pressure: the line pauses with exponential backoff and the send is retried with the same transaction id, so synapse dedupes a retry of a send it already completed. MatrixClient.sendEvent takes an optional txnId and throws a MatrixRequestError carrying the status, without JSON-parsing a proxy's HTML error page. The queue's depth, age, send latency, back-pressure, merges and drops go on the realm:health line, and a Grafana rule alerts when the oldest queued event has waited over 60s for 5 minutes. Co-Authored-By: Claude Opus 5.5 --- docs/realm-server-health-signals.md | 30 + .../realm-server-availability-group.json | 119 ++++ packages/realm-server/health-sampler.ts | 18 + packages/realm-server/node-realm.ts | 67 +- .../realm-server/realm-event-dispatcher.ts | 588 ++++++++++++++++++ .../realm-server/tests/node-realm-test.ts | 60 +- .../tests/realm-event-dispatcher-test.ts | 418 +++++++++++++ packages/runtime-common/matrix-client.ts | 45 +- 8 files changed, 1320 insertions(+), 25 deletions(-) create mode 100644 packages/realm-server/realm-event-dispatcher.ts create mode 100644 packages/realm-server/tests/realm-event-dispatcher-test.ts diff --git a/docs/realm-server-health-signals.md b/docs/realm-server-health-signals.md index 99dcb28767a..c722155b634 100644 --- a/docs/realm-server-health-signals.md +++ b/docs/realm-server-health-signals.md @@ -94,6 +94,36 @@ Silent on an idle, healthy server — so its presence in the logs is itself the signal. Lag climbing in step with `inFlightSearch` is the fingerprint of the loop being starved by concurrent search serialization. +The line also reports the realm event queue: the per-process line every realm +event takes to its subscribers' session rooms on synapse +(`realm-event-dispatcher.ts`): + +``` +realmEventsQueued=… realmEventQueuedKB=… realmEventQueueAgeMs=… realmEventSendsInFlight=… realmEventSends=… realmEventSendMs(mean/max)=… realmEventFailedSends=… realmEventBackPressure=… realmEventsCoalesced=… realmEventsDropped=… +``` + +`realmEventQueueAgeMs` is how long the oldest event still waiting to start has +waited, which is the delay a browser sees on top of the send itself. A burst +that drains keeps it low however deep the queue gets, so it, not the depth, is +what the "Realm Server Realm Event Backlog" alert reads. When it climbs, the +next field to read is `realmEventBackPressure`, the count of sends in the window +that synapse or the proxy in front of it answered with 429, 502, 503 or 504: + +- Nonzero: synapse is overloaded, and the queue is pausing to let it recover. + Look at synapse CPU and at what is writing in bulk. +- Zero: the queue is running at its configured rate. That rate is set by + `REALM_EVENT_SENDS_PER_SECOND` (default 50) and + `REALM_EVENT_MAX_SENDS_IN_FLIGHT` (default 8). + +`realmEventsDropped` counts events discarded because the queue passed +`REALM_EVENT_MAX_QUEUED_BYTES` (default 64 MiB). The queue drops the oldest +event of the realm with the longest backlog. + +The counters (`realmEventSends`, `realmEventFailedSends`, +`realmEventBackPressure`, `realmEventsCoalesced`, `realmEventsDropped`) cover +the window since the previous line. The others are read at the moment of the +sample. + ## `_readiness-check` is not one of them `GET //_readiness-check` shares the word "readiness" but answers a different diff --git a/packages/observability/provisioning/alerting/realm-server-availability-group.json b/packages/observability/provisioning/alerting/realm-server-availability-group.json index 38d0d042fb2..d22df6e8c80 100644 --- a/packages/observability/provisioning/alerting/realm-server-availability-group.json +++ b/packages/observability/provisioning/alerting/realm-server-availability-group.json @@ -474,6 +474,125 @@ "isPaused": false, "notification_settings": null }, + { + "uid": "rs-realm-event-backlog", + "orgID": 1, + "folderUID": "defd2d156sav4d", + "ruleGroup": "realm-server-availability-group", + "title": "Realm Server Realm Event Backlog", + "condition": "B", + "annotations": { + "summary": "A realm-server process's realm event queue held an event for more than 60s before it started sending, at every 1-minute sample for 5 minutes. Sourced from realmEventQueueAgeMs on the periodic health line: the wait of the oldest queued event, which a queue that drains keeps near zero however deep a burst is. A held backlog means realm events (index, update) reach browsers late, so cards and searches look stale. Read realmEventBackPressure on the same line: nonzero means synapse is overloaded and the queue is pacing itself to it, so look at synapse CPU and what is writing in bulk; zero means the queue is at its configured send rate (REALM_EVENT_SENDS_PER_SECOND, REALM_EVENT_MAX_SENDS_IN_FLIGHT). realmEventsDropped counts events discarded once the queue reached REALM_EVENT_MAX_QUEUED_BYTES." + }, + "data": [ + { + "refId": "A", + "queryType": "", + "relativeTimeRange": { + "from": 300, + "to": 0 + }, + "datasourceUid": "cef5x9o3yzawwf", + "model": { + "datasource": { + "type": "cloudwatch", + "uid": "cef5x9o3yzawwf" + }, + "dimensions": {}, + "expression": "filter @message like /realmEventQueueAgeMs=/\n| parse @message /realmEventQueueAgeMs=(?\\d+)/\n| stats max(ageMs) as maxQueueAgeMs by bin(1m)", + "id": "", + "intervalMs": 1000, + "label": "", + "logGroups": [ + { + "accountId": "${REALM_SERVER_LOG_GROUP_ACCOUNT_ID}", + "arn": "arn:aws:logs:us-east-1:${REALM_SERVER_LOG_GROUP_ACCOUNT_ID}:log-group:${REALM_SERVER_LOG_GROUP_NAME}", + "name": "${REALM_SERVER_LOG_GROUP_NAME}" + } + ], + "matchExact": true, + "maxDataPoints": 43200, + "metricEditorMode": 0, + "metricName": "", + "metricQueryType": 0, + "namespace": "", + "period": "", + "queryMode": "Logs", + "refId": "A", + "region": "default", + "sqlExpression": "", + "statistic": "Average" + } + }, + { + "refId": "C", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "datasource": { + "name": "Expression", + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "A", + "intervalMs": 1000, + "maxDataPoints": 43200, + "reducer": "max", + "refId": "C", + "type": "reduce" + } + }, + { + "refId": "B", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "conditions": [ + { + "evaluator": { + "params": [60000, 0], + "type": "gt" + }, + "operator": { + "type": "and" + }, + "query": { + "params": [] + }, + "reducer": { + "params": [], + "type": "avg" + }, + "type": "query" + } + ], + "datasource": { + "name": "Expression", + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "C", + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "B", + "type": "threshold" + } + } + ], + "noDataState": "OK", + "execErrState": "Alerting", + "for": "5m", + "isPaused": false, + "notification_settings": null + }, { "uid": "rs-search-saturation", "orgID": 1, diff --git a/packages/realm-server/health-sampler.ts b/packages/realm-server/health-sampler.ts index 15a2e21624b..fca03551f78 100644 --- a/packages/realm-server/health-sampler.ts +++ b/packages/realm-server/health-sampler.ts @@ -10,6 +10,7 @@ import { heapTelemetry, formatHeapTelemetry, } from './prerender/heap-telemetry.ts'; +import { realmEventDispatcher } from './realm-event-dispatcher.ts'; // Periodically samples the realm-server process's event-loop health, in-flight // search count, and heap usage, emitting a `realm:health` line every interval. @@ -93,6 +94,13 @@ export function startHealthSampler( // overlapping compiles each count their whole span, so `total` can exceed // the window. Read it beside `eventLoopLagMs`, not instead of it. let compiles = takeModuleCompileStats(); + // The realm event line toward synapse. `realmEventQueueAgeMs` is how long + // the oldest event still waiting to start has waited — the delay a + // subscriber sees on top of the send itself — and is the number to alert + // on: a queue that drains keeps it near zero however deep the burst. + // `realmEventBackPressure` counts sends synapse (or the proxy in front of + // it) answered as overloaded in this window. + let realmEvents = realmEventDispatcher().takeStats(); log.info( `eventLoopLagMs(mean/p99/max)=${meanLagMs.toFixed(0)}/${p99LagMs.toFixed(0)}/${maxLagMs.toFixed(0)} ` + `inFlightSearch=${inFlightSearch} searchRequests=${searchRequests} ` + @@ -105,6 +113,16 @@ export function startHealthSampler( : '') + `transpiles=${compiles.count} ` + `transpileMs(total/max)=${compiles.totalMs.toFixed(0)}/${compiles.maxMs.toFixed(0)} ` + + `realmEventsQueued=${realmEvents.queuedEvents} ` + + `realmEventQueuedKB=${(realmEvents.queuedBytes / 1024).toFixed(0)} ` + + `realmEventQueueAgeMs=${realmEvents.oldestQueuedMs.toFixed(0)} ` + + `realmEventSendsInFlight=${realmEvents.sendsInFlight} ` + + `realmEventSends=${realmEvents.sends} ` + + `realmEventSendMs(mean/max)=${realmEvents.sendMsMean.toFixed(0)}/${realmEvents.sendMsMax.toFixed(0)} ` + + `realmEventFailedSends=${realmEvents.failedSends} ` + + `realmEventBackPressure=${realmEvents.backPressure} ` + + `realmEventsCoalesced=${realmEvents.coalesced} ` + + `realmEventsDropped=${realmEvents.dropped} ` + `${formatHeapTelemetry(heap)}`, ); }, intervalMs); diff --git a/packages/realm-server/node-realm.ts b/packages/realm-server/node-realm.ts index 4e90ae9c76e..77a30ac5396 100644 --- a/packages/realm-server/node-realm.ts +++ b/packages/realm-server/node-realm.ts @@ -55,6 +55,11 @@ import type { } from '@cardstack/base/matrix-event'; import { APP_BOXEL_REALM_EVENT_TYPE } from '@cardstack/runtime-common/matrix-constants'; import { createJWT, verifyJWT } from './jwt.ts'; +import { + realmEventDispatcher, + type RealmEventDispatcher, + type RealmEventSend, +} from './realm-event-dispatcher.ts'; const realmEventsLog = logger('realm:events'); @@ -88,10 +93,16 @@ function lastModifiedSeconds(stat: Stats): number { export class NodeAdapter implements RealmAdapter { private realmDir: string; private enableFileWatcher?: boolean; + private dispatcher: RealmEventDispatcher; - constructor(realmDir: string, enableFileWatcher?: boolean) { + constructor( + realmDir: string, + enableFileWatcher?: boolean, + dispatcher: RealmEventDispatcher = realmEventDispatcher(), + ) { this.realmDir = realmDir; this.enableFileWatcher = enableFileWatcher; + this.dispatcher = dispatcher; } get dir(): string { @@ -421,6 +432,9 @@ export class NodeAdapter implements RealmAdapter { }; } + // Resolves once the event has reached every session room subscribed to the + // realm, or has been merged or dropped by the process's realm event line + // (see realm-event-dispatcher.ts). Never rejects: a failed send is logged. async broadcastRealmEvent( event: RealmEventContent, realmUrl: string, @@ -438,6 +452,26 @@ export class NodeAdapter implements RealmAdapter { ); return; } + await this.dispatcher.broadcast({ + realmURL: realmUrl, + event: eventWithRealmURL, + newTxnId: () => matrixClient.nextTxnId(), + prepare: (content) => + this.prepareRoomSends(content, realmUrl, matrixClient, dbAdapter), + }); + } + + private async prepareRoomSends( + event: RealmEventContent, + realmUrl: string, + matrixClient: MatrixClient, + dbAdapter: DBAdapter, + ): Promise { + if (dbAdapter.isClosed) { + return []; + } + // A logged-in client reuses its session; this logs in only on the first + // broadcast, or after the session was revoked. try { await matrixClient.login(); if (!matrixClient.getUserId()) { @@ -445,26 +479,28 @@ export class NodeAdapter implements RealmAdapter { 'Matrix client has no user ID after login, unable to broadcast realm event', event, ); - return; + return []; } } catch (e) { realmEventsLog.error('Error logging into matrix. Skipping broadcast', e); - return; + return []; } let dmRooms = await this.waitForSessionRooms(dbAdapter, realmUrl); realmEventsLog.debug('Sending to dm rooms', Object.values(dmRooms)); - for (let userId of Object.keys(dmRooms)) { - let roomId = dmRooms[userId]; - try { + return Object.entries(dmRooms).map(([userId, roomId]) => ({ + roomId, + send: async (txnId: string) => { await matrixClient.sendEvent( roomId, APP_BOXEL_REALM_EVENT_TYPE, - eventWithRealmURL, + event, + txnId, ); - } catch (e) { + }, + onFailure: async (e: unknown) => { if (isRealmServerNotInRoomError(e, roomId)) { try { let cleared = await clearSessionRoom(dbAdapter, userId, roomId); @@ -483,15 +519,18 @@ export class NodeAdapter implements RealmAdapter { e, ); } - continue; + return; } + // The error's message carries the status and the homeserver's answer, + // which is all a failed send has to say; its stack is the same for + // every room. realmEventsLog.error( - `Unable to send event in room ${roomId} for user ${userId}`, - event, - e, + `Unable to send event in room ${roomId} for user ${userId}: ${ + e instanceof Error ? e.message : String(e) + }`, ); - } - } + }, + })); } private async waitForSessionRooms( diff --git a/packages/realm-server/realm-event-dispatcher.ts b/packages/realm-server/realm-event-dispatcher.ts new file mode 100644 index 00000000000..641042d4058 --- /dev/null +++ b/packages/realm-server/realm-event-dispatcher.ts @@ -0,0 +1,588 @@ +import { logger } from '@cardstack/runtime-common'; +import type { + RealmEventContent, + UpdateRealmEventContent, +} from '@cardstack/base/matrix-event'; + +const log = logger('realm:events'); + +// Delivers realm events to their subscribers' session rooms through one +// per-process line, so a burst of events reaches synapse at a pace synapse can +// absorb rather than as fast as the realms produce them. +// +// Synapse serves every human user's sync, message sends and AI assistant +// stream from the same process the realm events land on. An event fans out to +// one send per session room, so a batch that writes many files turns into +// events × rooms sends; started all at once, they queue inside synapse and +// every user waits behind them. A proxy that times a slow send out does not +// cancel it either: synapse completes the send regardless, so timeouts shed no +// load. The only throttle that works is the sender's. +// +// The shape of the line: +// +// - At most `maxInFlight` sends are outstanding at once, and sends start no +// faster than `sendsPerSecond`. +// - Realms take turns. Each send slot goes to the next realm in rotation that +// has a send ready, so a realm with a large backlog takes an equal share of +// the line rather than all of it, and an event from a quiet realm waits +// behind at most one send per busy realm. +// - Within a realm, events go out in the order they were broadcast: a realm's +// next event starts only once every room has the one before it. A +// subscriber reads an `incremental-index-initiation` as "indexing started" +// and the `incremental` after it as "indexing finished", so reordering them +// would leave it believing a finished pass is still running. +// - A queued event that a newer one supersedes merges into it (see +// `coalesce`). Only the realm's most recently queued event merges, so the +// merge never moves an event past one of another kind. +// - A send that synapse answers with 429, 502, 503 or 504, or that cannot +// reach synapse at all, is back-pressure: the whole line pauses, and the +// send is retried with the same transaction id, so a retry of a send that +// synapse did complete is not stored twice. The pause doubles on each +// consecutive back-pressure answer and resets on the next success. +// - The bytes queued are bounded. Past the bound, the oldest event of the +// realm with the longest backlog is dropped, so the realm causing the +// backlog is the one that loses events. +// +// The queue is in memory, so a process that exits with events queued does not +// deliver them. + +export interface RealmEventSend { + roomId: string; + // Called once per attempt, with the same transaction id each time. + send(txnId: string): Promise; + // Called once, when the send fails with an error that is not + // back-pressure, or when back-pressure outlasts `maxAttempts`. + onFailure(error: unknown): Promise; +} + +export interface RealmEventBroadcast { + realmURL: string; + event: RealmEventContent; + // Resolves the sends the event fans out to. Runs when the event reaches the + // head of its realm's line, with the event as merged by then, so the rooms + // are the ones subscribed at send time. + prepare(event: RealmEventContent): Promise; + newTxnId(): string; +} + +export interface RealmEventDispatcherOptions { + maxInFlight?: number; + sendsPerSecond?: number; + maxQueuedBytes?: number; + maxAttempts?: number; + initialBackoffMs?: number; + maxBackoffMs?: number; + now?: () => number; +} + +export interface RealmEventDispatcherStats { + // Gauges, read at the moment of the call. + queuedEvents: number; + queuedBytes: number; + sendsInFlight: number; + // How long the oldest event still waiting to start has waited. + oldestQueuedMs: number; + // Counters, since the previous call. + sends: number; + sendMsMean: number; + sendMsMax: number; + failedSends: number; + backPressure: number; + coalesced: number; + dropped: number; +} + +interface Entry { + realmURL: string; + event: RealmEventContent; + bytes: number; + enqueuedAt: number; + prepare: RealmEventBroadcast['prepare']; + newTxnId: RealmEventBroadcast['newTxnId']; + waiters: (() => void)[]; +} + +interface PendingSend { + send: RealmEventSend; + txnId: string; + attempts: number; +} + +interface Active { + entry: Entry; + // Undefined while `prepare` runs. + ready: PendingSend[] | undefined; + outstanding: number; +} + +interface Line { + pending: Entry[]; + active: Active | undefined; +} + +const BACK_PRESSURE_STATUSES = new Set([429, 502, 503, 504]); + +export function isBackPressure(error: unknown): boolean { + let status = (error as { status?: unknown } | undefined)?.status; + if (typeof status === 'number') { + return BACK_PRESSURE_STATUSES.has(status); + } + // fetch rejects with a TypeError when it cannot reach the host at all + // (connection refused or reset, DNS failure, socket closed mid-response). + return error instanceof TypeError && /fetch failed/i.test(error.message); +} + +export class RealmEventDispatcher { + readonly #maxInFlight: number; + readonly #sendsPerSecond: number; + readonly #maxQueuedBytes: number; + readonly #maxAttempts: number; + readonly #initialBackoffMs: number; + readonly #maxBackoffMs: number; + readonly #now: () => number; + + // Insertion order is the rotation order; `#cursor` is the realm that gets + // the next look. + #lines = new Map(); + #cursor: string | undefined; + #inFlight = 0; + #queuedEvents = 0; + #queuedBytes = 0; + #tokens: number; + #tokensAt: number; + #pausedUntil = 0; + #backoffMs = 0; + #timer: ReturnType | undefined; + #timerAt = 0; + #idleWaiters: (() => void)[] = []; + + #sends = 0; + #sendMsTotal = 0; + #sendMsMax = 0; + #failedSends = 0; + #backPressure = 0; + #coalesced = 0; + #dropped = 0; + #droppedSinceWarning = 0; + #lastDropWarningAt = 0; + + constructor(opts: RealmEventDispatcherOptions = {}) { + this.#maxInFlight = opts.maxInFlight ?? 8; + this.#sendsPerSecond = opts.sendsPerSecond ?? 50; + this.#maxQueuedBytes = opts.maxQueuedBytes ?? 64 * 1024 * 1024; + this.#maxAttempts = opts.maxAttempts ?? 5; + this.#initialBackoffMs = opts.initialBackoffMs ?? 1000; + this.#maxBackoffMs = opts.maxBackoffMs ?? 30_000; + this.#now = opts.now ?? (() => Date.now()); + // The bucket holds one second's worth of sends, so an idle line can start + // a short burst at once and a long burst settles to the steady rate. + this.#tokens = this.#sendsPerSecond; + this.#tokensAt = this.#now(); + } + + // Resolves once the event has gone to every room it fans out to, has merged + // into an event that has, or has been dropped. It never rejects: delivery is + // best-effort, and a failed send is reported through its `onFailure`. + broadcast(broadcast: RealmEventBroadcast): Promise { + return new Promise((resolve) => { + let line = this.#lines.get(broadcast.realmURL); + if (!line) { + line = { pending: [], active: undefined }; + this.#lines.set(broadcast.realmURL, line); + } + let tail = line.pending[line.pending.length - 1]; + let merged = tail ? coalesce(tail.event, broadcast.event) : undefined; + if (tail && merged) { + let bytes = byteLength(merged); + this.#queuedBytes += bytes - tail.bytes; + tail.event = merged; + tail.bytes = bytes; + tail.prepare = broadcast.prepare; + tail.newTxnId = broadcast.newTxnId; + tail.waiters.push(resolve); + this.#coalesced++; + } else { + let entry: Entry = { + realmURL: broadcast.realmURL, + event: broadcast.event, + bytes: byteLength(broadcast.event), + enqueuedAt: this.#now(), + prepare: broadcast.prepare, + newTxnId: broadcast.newTxnId, + waiters: [resolve], + }; + line.pending.push(entry); + this.#queuedEvents++; + this.#queuedBytes += entry.bytes; + } + this.#enforceBound(); + this.#pump(); + }); + } + + // Resolves once nothing is queued or in flight. + async idle(): Promise { + if (this.#isIdle()) { + return; + } + await new Promise((resolve) => this.#idleWaiters.push(resolve)); + } + + takeStats(): RealmEventDispatcherStats { + let now = this.#now(); + let oldest = now; + for (let line of this.#lines.values()) { + let head = line.pending[0]; + if (head && head.enqueuedAt < oldest) { + oldest = head.enqueuedAt; + } + } + let stats: RealmEventDispatcherStats = { + queuedEvents: this.#queuedEvents, + queuedBytes: this.#queuedBytes, + sendsInFlight: this.#inFlight, + oldestQueuedMs: now - oldest, + sends: this.#sends, + sendMsMean: this.#sends ? this.#sendMsTotal / this.#sends : 0, + sendMsMax: this.#sendMsMax, + failedSends: this.#failedSends, + backPressure: this.#backPressure, + coalesced: this.#coalesced, + dropped: this.#dropped, + }; + this.#sends = 0; + this.#sendMsTotal = 0; + this.#sendMsMax = 0; + this.#failedSends = 0; + this.#backPressure = 0; + this.#coalesced = 0; + this.#dropped = 0; + return stats; + } + + #isIdle() { + return this.#queuedEvents === 0 && this.#lines.size === 0; + } + + #enforceBound() { + while (this.#queuedBytes > this.#maxQueuedBytes) { + let longest: Line | undefined; + for (let line of this.#lines.values()) { + if ( + line.pending.length > 0 && + (!longest || line.pending.length > longest.pending.length) + ) { + longest = line; + } + } + if (!longest) { + return; + } + let dropped = longest.pending.shift()!; + this.#queuedEvents--; + this.#queuedBytes -= dropped.bytes; + this.#dropped++; + this.#droppedSinceWarning++; + for (let resolve of dropped.waiters) { + resolve(); + } + this.#retireIfDone(dropped.realmURL); + } + // One line per window rather than one per event: under a backlog large + // enough to drop, a line per event is its own flood. + let now = this.#now(); + if ( + this.#droppedSinceWarning > 0 && + now - this.#lastDropWarningAt > 10_000 + ) { + log.warn( + `realm event queue is over ${this.#maxQueuedBytes} bytes; dropped ${this.#droppedSinceWarning} queued event(s) from the realms with the longest backlogs`, + ); + this.#droppedSinceWarning = 0; + this.#lastDropWarningAt = now; + } + } + + #pump() { + while (this.#inFlight < this.#maxInFlight) { + let now = this.#now(); + if (now < this.#pausedUntil) { + this.#wakeAt(this.#pausedUntil); + return; + } + this.#refill(now); + let next = this.#nextSend(); + if (!next) { + return; + } + if (this.#tokens < 1) { + // Put it back at the front of its realm's sends; the rotation has + // already moved on, so this realm is not favored when the line wakes. + next.active.ready!.unshift(next.pending); + this.#wakeAt( + now + Math.ceil(((1 - this.#tokens) * 1000) / this.#sendsPerSecond), + ); + return; + } + this.#tokens--; + this.#run(next.active, next.pending); + } + } + + #refill(now: number) { + let elapsed = now - this.#tokensAt; + if (elapsed > 0) { + this.#tokens = Math.min( + this.#sendsPerSecond, + this.#tokens + (elapsed * this.#sendsPerSecond) / 1000, + ); + this.#tokensAt = now; + } + } + + #wakeAt(at: number) { + if (this.#timer && this.#timerAt <= at) { + return; + } + if (this.#timer) { + clearTimeout(this.#timer); + } + this.#timerAt = at; + this.#timer = setTimeout( + () => { + this.#timer = undefined; + this.#pump(); + }, + Math.max(0, at - this.#now()), + ); + // A process with nothing else to do is not kept alive by a queued + // realm event. + this.#timer.unref?.(); + } + + // The next send in rotation, starting realms' next events along the way. + #nextSend(): { active: Active; pending: PendingSend } | undefined { + let realms = [...this.#lines.keys()]; + if (realms.length === 0) { + return undefined; + } + let start = this.#cursor ? realms.indexOf(this.#cursor) : 0; + if (start < 0) { + start = 0; + } + for (let i = 0; i < realms.length; i++) { + let realmURL = realms[(start + i) % realms.length]; + let line = this.#lines.get(realmURL)!; + if (!line.active && line.pending.length > 0) { + this.#start(line); + } + let active = line.active; + if (active?.ready && active.ready.length > 0) { + this.#cursor = realms[(start + i + 1) % realms.length]; + return { active, pending: active.ready.shift()! }; + } + } + return undefined; + } + + #start(line: Line) { + let entry = line.pending.shift()!; + this.#queuedEvents--; + this.#queuedBytes -= entry.bytes; + let active: Active = { entry, ready: undefined, outstanding: 0 }; + line.active = active; + entry + .prepare(entry.event) + .catch((e) => { + log.error( + `Unable to resolve the rooms for a realm event in ${entry.realmURL}; skipping it`, + e, + ); + return []; + }) + .then((sends) => { + active.ready = sends.map((send) => ({ + send, + txnId: entry.newTxnId(), + attempts: 0, + })); + this.#finishIfDone(line); + this.#pump(); + }); + } + + #run(active: Active, pending: PendingSend) { + this.#inFlight++; + active.outstanding++; + pending.attempts++; + let startedAt = this.#now(); + let realmURL = active.entry.realmURL; + void pending.send + .send(pending.txnId) + .then( + () => { + let ms = this.#now() - startedAt; + this.#sends++; + this.#sendMsTotal += ms; + this.#sendMsMax = Math.max(this.#sendMsMax, ms); + this.#backoffMs = 0; + }, + async (error: unknown) => { + if (isBackPressure(error)) { + this.#backPressure++; + this.#backoffMs = Math.min( + this.#maxBackoffMs, + this.#backoffMs ? this.#backoffMs * 2 : this.#initialBackoffMs, + ); + this.#pausedUntil = Math.max( + this.#pausedUntil, + this.#now() + this.#backoffMs, + ); + if (pending.attempts < this.#maxAttempts) { + active.ready!.push(pending); + return; + } + } + this.#failedSends++; + try { + await pending.send.onFailure(error); + } catch (e) { + log.error( + `Failure handler threw for a realm event send to room ${pending.send.roomId}`, + e, + ); + } + }, + ) + .finally(() => { + this.#inFlight--; + active.outstanding--; + let line = this.#lines.get(realmURL); + if (line) { + this.#finishIfDone(line); + } + this.#pump(); + }); + } + + #finishIfDone(line: Line) { + let active = line.active; + if ( + !active || + !active.ready || + active.ready.length > 0 || + active.outstanding > 0 + ) { + return; + } + line.active = undefined; + for (let resolve of active.entry.waiters) { + resolve(); + } + this.#retireIfDone(active.entry.realmURL); + } + + #retireIfDone(realmURL: string) { + let line = this.#lines.get(realmURL); + if (line && !line.active && line.pending.length === 0) { + if (this.#cursor === realmURL) { + let realms = [...this.#lines.keys()]; + let next = realms[(realms.indexOf(realmURL) + 1) % realms.length]; + this.#cursor = next === realmURL ? undefined : next; + } + this.#lines.delete(realmURL); + } + if (this.#isIdle()) { + let waiters = this.#idleWaiters; + this.#idleWaiters = []; + for (let resolve of waiters) { + resolve(); + } + } + } +} + +// The event a queued `queued` and a newly broadcast `next` from the same realm +// merge into, or undefined when the two have to go out separately. +// +// - Two `incremental-index-initiation`s merge into the newer one. A subscriber +// reads the event as "this realm is indexing"; a second one before any +// completion tells it nothing the first did not, and the newer `updatedFile` +// names the most recent write. +// - Two `update`s merge into one listing every path either listed. A +// subscriber reads the lists as "these paths changed" and refreshes what +// contains them, so the union says everything the two said. +// +// Completion events (`incremental`, `full`) and `prerender_html` carry what a +// subscriber reconciles per event — request ids, versions, generations — and +// always go out as broadcast. +export function coalesce( + queued: RealmEventContent, + next: RealmEventContent, +): RealmEventContent | undefined { + if ( + queued.eventName === 'index' && + next.eventName === 'index' && + queued.indexType === 'incremental-index-initiation' && + next.indexType === 'incremental-index-initiation' + ) { + return next; + } + if (queued.eventName === 'update' && next.eventName === 'update') { + let merged: UpdateRealmEventContent = { + eventName: 'update', + realmURL: next.realmURL, + }; + for (let key of ['added', 'updated', 'removed'] as const) { + let paths = union(queued[key], next[key]); + if (paths) { + merged[key] = paths; + } + } + return merged; + } + return undefined; +} + +function union( + a: string[] | undefined, + b: string[] | undefined, +): string[] | undefined { + if (!a && !b) { + return undefined; + } + return [...new Set([...(a ?? []), ...(b ?? [])])]; +} + +function byteLength(event: RealmEventContent): number { + return Buffer.byteLength(JSON.stringify(event)); +} + +function positiveNumberFromEnv(name: string): number | undefined { + let raw = process.env[name]; + if (raw === undefined || raw === '') { + return undefined; + } + let value = Number(raw); + if (!Number.isFinite(value) || value <= 0) { + log.warn(`ignoring ${name}=${raw}: expected a positive number`); + return undefined; + } + return value; +} + +let processDispatcher: RealmEventDispatcher | undefined; + +// The one line every realm in this process broadcasts through. Tuned by +// REALM_EVENT_MAX_SENDS_IN_FLIGHT, REALM_EVENT_SENDS_PER_SECOND and +// REALM_EVENT_MAX_QUEUED_BYTES. +export function realmEventDispatcher(): RealmEventDispatcher { + if (!processDispatcher) { + processDispatcher = new RealmEventDispatcher({ + maxInFlight: positiveNumberFromEnv('REALM_EVENT_MAX_SENDS_IN_FLIGHT'), + sendsPerSecond: positiveNumberFromEnv('REALM_EVENT_SENDS_PER_SECOND'), + maxQueuedBytes: positiveNumberFromEnv('REALM_EVENT_MAX_QUEUED_BYTES'), + }); + } + return processDispatcher; +} diff --git a/packages/realm-server/tests/node-realm-test.ts b/packages/realm-server/tests/node-realm-test.ts index a1e8631c9a7..f5623ae8e88 100644 --- a/packages/realm-server/tests/node-realm-test.ts +++ b/packages/realm-server/tests/node-realm-test.ts @@ -13,6 +13,8 @@ import type { MatrixClient } from '@cardstack/runtime-common/matrix-client'; import type { RealmEventContent } from '@cardstack/base/matrix-event'; import type { LocalPath } from '@cardstack/runtime-common/paths'; import { NodeAdapter } from '../node-realm.ts'; +import { RealmEventDispatcher } from '../realm-event-dispatcher.ts'; +import { MatrixRequestError } from '@cardstack/runtime-common/matrix-client'; import { insertUser, setupDB } from './helpers/index.ts'; module(basename(import.meta.filename), function (hooks) { @@ -56,6 +58,7 @@ module(basename(import.meta.filename), function (hooks) { let matrixClient = { login: async () => undefined, getUserId: () => '@realm_server:localhost', + nextTxnId: () => 'txn-1', sendEvent: async () => { sendEventCalls++; throw new Error( @@ -64,7 +67,7 @@ module(basename(import.meta.filename), function (hooks) { }, } as unknown as MatrixClient; - let adapter = new NodeAdapter('/tmp'); + let adapter = new NodeAdapter('/tmp', false, new RealmEventDispatcher()); await adapter.broadcastRealmEvent( makeEvent(), realmURL.href, @@ -90,6 +93,7 @@ module(basename(import.meta.filename), function (hooks) { let matrixClient = { login: async () => undefined, getUserId: () => '@realm_server:localhost', + nextTxnId: () => 'txn-1', sendEvent: async () => { throw new Error( `Unable to send room event 'app.boxel.realm-event' to room ${staleRoomId}: status 500 - {"errcode":"M_UNKNOWN","error":"boom"}`, @@ -97,7 +101,7 @@ module(basename(import.meta.filename), function (hooks) { }, } as unknown as MatrixClient; - let adapter = new NodeAdapter('/tmp'); + let adapter = new NodeAdapter('/tmp', false, new RealmEventDispatcher()); await adapter.broadcastRealmEvent( makeEvent(), realmURL.href, @@ -128,6 +132,7 @@ module(basename(import.meta.filename), function (hooks) { let matrixClient = { login: async () => undefined, getUserId: () => '@realm_server:localhost', + nextTxnId: () => 'txn-1', sendEvent: async () => { failCleanup = true; throw new Error( @@ -136,7 +141,7 @@ module(basename(import.meta.filename), function (hooks) { }, } as unknown as MatrixClient; - let adapter = new NodeAdapter('/tmp'); + let adapter = new NodeAdapter('/tmp', false, new RealmEventDispatcher()); await adapter.broadcastRealmEvent( makeEvent(), @@ -147,6 +152,55 @@ module(basename(import.meta.filename), function (hooks) { assert.true(true, 'broadcast resolves even if stale room cleanup fails'); }); + + test('retries a send synapse answers as overloaded, with the same transaction id', async function (assert) { + await insertSessionForAlice(); + + let txnIds: string[] = []; + let matrixClient = { + login: async () => undefined, + getUserId: () => '@realm_server:localhost', + nextTxnId: () => `txn-${txnIds.length}`, + sendEvent: async ( + _roomId: string, + _type: string, + _content: unknown, + txnId: string, + ) => { + txnIds.push(txnId); + if (txnIds.length === 1) { + throw new MatrixRequestError( + `Unable to send room event 'app.boxel.realm-event' to room ${staleRoomId}: status 504 - non-JSON body (122 bytes)`, + 504, + ); + } + return '$event'; + }, + } as unknown as MatrixClient; + + let adapter = new NodeAdapter( + '/tmp', + false, + new RealmEventDispatcher({ initialBackoffMs: 1 }), + ); + await adapter.broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + ); + + assert.deepEqual( + txnIds, + ['txn-0', 'txn-0'], + 'the retry reuses the transaction id of the first attempt', + ); + assert.strictEqual( + await fetchSessionRoom(dbAdapter, '@alice:localhost'), + staleRoomId, + 'an overloaded homeserver does not clear the session room', + ); + }); }); module(`${basename(import.meta.filename)} | file stat probing`, function () { diff --git a/packages/realm-server/tests/realm-event-dispatcher-test.ts b/packages/realm-server/tests/realm-event-dispatcher-test.ts new file mode 100644 index 00000000000..ea0310f9d72 --- /dev/null +++ b/packages/realm-server/tests/realm-event-dispatcher-test.ts @@ -0,0 +1,418 @@ +import QUnit from 'qunit'; +const { module, test } = QUnit; +import { basename } from 'path'; +import type { RealmEventContent } from '@cardstack/base/matrix-event'; +import { MatrixRequestError } from '@cardstack/runtime-common/matrix-client'; +import { + RealmEventDispatcher, + coalesce, + isBackPressure, + type RealmEventBroadcast, +} from '../realm-event-dispatcher.ts'; + +const REALM_A = 'http://127.0.0.1:4444/a/'; +const REALM_B = 'http://127.0.0.1:4444/b/'; + +interface Sent { + realmURL: string; + roomId: string; + event: RealmEventContent; + txnId: string; +} + +function deferred() { + let resolve!: () => void; + let reject!: (e: unknown) => void; + let promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +function initiation(realmURL: string, updatedFile: string): RealmEventContent { + return { + eventName: 'index', + indexType: 'incremental-index-initiation', + updatedFile, + realmURL, + }; +} + +function incremental(realmURL: string, card: string): RealmEventContent { + return { + eventName: 'index', + indexType: 'incremental', + invalidations: [`${realmURL}${card}`], + realmURL, + }; +} + +// A harness whose sends are held open until the test releases them, so a test +// can look at what the dispatcher has started at any moment. +function harness( + opts: ConstructorParameters[0] = {}, +) { + let dispatcher = new RealmEventDispatcher({ + sendsPerSecond: 1000, + ...opts, + }); + let started: (Sent & { done: ReturnType })[] = []; + let failures: { roomId: string; error: unknown }[] = []; + let txn = 0; + let autoRelease = false; + + function broadcast( + realmURL: string, + event: RealmEventContent, + rooms: string[] = ['!room-1'], + send?: (sent: Sent) => Promise, + ) { + let request: RealmEventBroadcast = { + realmURL, + event, + newTxnId: () => `txn-${txn++}`, + prepare: async (content) => + rooms.map((roomId) => ({ + roomId, + send: async (txnId: string) => { + let sent = { realmURL, roomId, event: content, txnId }; + if (send) { + return await send(sent); + } + let done = deferred(); + started.push({ ...sent, done }); + if (autoRelease) { + done.resolve(); + } + await done.promise; + }, + onFailure: async (error: unknown) => { + failures.push({ roomId, error }); + }, + })), + }; + return dispatcher.broadcast(request); + } + + async function settle() { + for (let i = 0; i < 10; i++) { + await new Promise((resolve) => setImmediate(resolve)); + } + } + + function releaseAll() { + autoRelease = true; + for (let s of started) { + s.done.resolve(); + } + } + + return { dispatcher, broadcast, started, failures, settle, releaseAll }; +} + +module(basename(import.meta.filename), function () { + test('holds no more than maxInFlight sends outstanding at once', async function (assert) { + let h = harness({ maxInFlight: 2 }); + let delivered = h.broadcast(REALM_A, incremental(REALM_A, 'card'), [ + '!r1', + '!r2', + '!r3', + '!r4', + ]); + await h.settle(); + assert.strictEqual(h.started.length, 2, 'two sends start'); + assert.strictEqual(h.dispatcher.takeStats().sendsInFlight, 2); + + h.started[0].done.resolve(); + await h.settle(); + assert.strictEqual(h.started.length, 3, 'a finished send frees a slot'); + + h.releaseAll(); + await delivered; + assert.deepEqual( + h.started.map((s) => s.roomId), + ['!r1', '!r2', '!r3', '!r4'], + 'every room gets the event', + ); + }); + + test("starts a realm's next event only once every room has the one before it", async function (assert) { + let h = harness({ maxInFlight: 8 }); + void h.broadcast(REALM_A, initiation(REALM_A, `${REALM_A}card`), [ + '!r1', + '!r2', + ]); + let second = h.broadcast(REALM_A, incremental(REALM_A, 'card'), [ + '!r1', + '!r2', + ]); + await h.settle(); + assert.deepEqual( + h.started.map((s) => (s.event as { indexType: string }).indexType), + ['incremental-index-initiation', 'incremental-index-initiation'], + 'only the first event is sending', + ); + + h.started[0].done.resolve(); + await h.settle(); + assert.strictEqual( + h.started.length, + 2, + 'the second event waits while a room still lacks the first', + ); + + h.releaseAll(); + await second; + assert.deepEqual( + h.started.map((s) => (s.event as { indexType: string }).indexType), + [ + 'incremental-index-initiation', + 'incremental-index-initiation', + 'incremental', + 'incremental', + ], + ); + }); + + test('gives each realm with a send ready a turn, so a backlog does not hold up another realm', async function (assert) { + let h = harness({ maxInFlight: 1 }); + void h.broadcast(REALM_A, incremental(REALM_A, 'one'), [ + '!a1', + '!a2', + '!a3', + ]); + void h.broadcast(REALM_A, incremental(REALM_A, 'two'), ['!a1']); + let quiet = h.broadcast(REALM_B, incremental(REALM_B, 'card'), ['!b1']); + await h.settle(); + + for (let i = 0; i < 2; i++) { + h.started[h.started.length - 1].done.resolve(); + await h.settle(); + } + await quiet; + assert.deepEqual( + h.started.map((s) => s.roomId), + ['!a1', '!b1', '!a2'], + "realm B's send goes out after one send of realm A's, not after all of them", + ); + h.releaseAll(); + await h.dispatcher.idle(); + }); + + test('merges queued initiations and updates, but never past an event of another kind', async function (assert) { + let h = harness({ maxInFlight: 1 }); + void h.broadcast(REALM_A, incremental(REALM_A, 'sending')); + await h.settle(); + + let waits = [ + h.broadcast(REALM_A, initiation(REALM_A, `${REALM_A}one`)), + h.broadcast(REALM_A, initiation(REALM_A, `${REALM_A}two`)), + h.broadcast(REALM_A, incremental(REALM_A, 'two')), + h.broadcast(REALM_A, initiation(REALM_A, `${REALM_A}three`)), + h.broadcast(REALM_A, { + eventName: 'update', + added: ['three.json'], + realmURL: REALM_A, + }), + h.broadcast(REALM_A, { + eventName: 'update', + added: ['four.json'], + removed: ['three.json'], + realmURL: REALM_A, + }), + ]; + assert.strictEqual( + h.dispatcher.takeStats().coalesced, + 2, + 'the second initiation and the second update merge', + ); + + h.releaseAll(); + await Promise.all(waits); + assert.deepEqual( + h.started.map((s) => s.event), + [ + incremental(REALM_A, 'sending'), + initiation(REALM_A, `${REALM_A}two`), + incremental(REALM_A, 'two'), + initiation(REALM_A, `${REALM_A}three`), + { + eventName: 'update', + added: ['three.json', 'four.json'], + removed: ['three.json'], + realmURL: REALM_A, + }, + ], + 'the event already sending and the completion between the initiations are left alone', + ); + }); + + test('pauses and retries a send synapse answers as overloaded, with the same transaction id', async function (assert) { + let attempts: string[] = []; + let h = harness({ initialBackoffMs: 20 }); + let startedAt = Date.now(); + await h.broadcast( + REALM_A, + incremental(REALM_A, 'card'), + ['!r1'], + async ({ txnId }) => { + attempts.push(txnId); + if (attempts.length < 3) { + throw new MatrixRequestError('status 504 - non-JSON body', 504); + } + }, + ); + assert.deepEqual(attempts, ['txn-0', 'txn-0', 'txn-0']); + assert.true( + Date.now() - startedAt >= 55, + 'each retry waits out a pause that doubles', + ); + assert.deepEqual(h.failures, [], 'a retried send is not a failure'); + let stats = h.dispatcher.takeStats(); + assert.strictEqual(stats.backPressure, 2); + assert.strictEqual(stats.failedSends, 0); + }); + + test('reports a send that fails for another reason once, without retrying it', async function (assert) { + let attempts = 0; + let h = harness(); + let error = new MatrixRequestError( + 'status 500 - {"errcode":"M_UNKNOWN"}', + 500, + ); + await h.broadcast( + REALM_A, + incremental(REALM_A, 'card'), + ['!r1'], + async () => { + attempts++; + throw error; + }, + ); + assert.strictEqual(attempts, 1); + assert.deepEqual(h.failures, [{ roomId: '!r1', error }]); + }); + + test('gives up on a send that is still overloaded after maxAttempts', async function (assert) { + let attempts = 0; + let h = harness({ maxAttempts: 3, initialBackoffMs: 1 }); + await h.broadcast( + REALM_A, + incremental(REALM_A, 'card'), + ['!r1'], + async () => { + attempts++; + throw new MatrixRequestError('status 503 - non-JSON body', 503); + }, + ); + assert.strictEqual(attempts, 3); + assert.strictEqual(h.failures.length, 1, 'the last attempt is reported'); + }); + + test('past the byte bound, drops the oldest event of the realm with the longest backlog', async function (assert) { + let eventBytes = Buffer.byteLength( + JSON.stringify(incremental(REALM_A, 'card-0')), + ); + let h = harness({ maxInFlight: 1, maxQueuedBytes: eventBytes * 3 }); + void h.broadcast(REALM_A, incremental(REALM_A, 'sending')); + await h.settle(); + + let dropped = false; + void h + .broadcast(REALM_A, incremental(REALM_A, 'card-0')) + .then(() => (dropped = true)); + void h.broadcast(REALM_A, incremental(REALM_A, 'card-1')); + void h.broadcast(REALM_B, incremental(REALM_B, 'card-0')); + assert.false(dropped, 'within the bound nothing is dropped'); + + void h.broadcast(REALM_B, incremental(REALM_B, 'card-1')); + await h.settle(); + assert.true(dropped, "the dropped event's broadcast resolves"); + let stats = h.dispatcher.takeStats(); + assert.strictEqual(stats.dropped, 1); + assert.strictEqual(stats.queuedEvents, 3); + + h.releaseAll(); + await h.dispatcher.idle(); + assert.deepEqual( + h.started + .map((s) => (s.event as { invalidations: string[] }).invalidations[0]) + .sort(), + [ + `${REALM_A}card-1`, + `${REALM_A}sending`, + `${REALM_B}card-0`, + `${REALM_B}card-1`, + ], + "realm B's backlog is delivered whole", + ); + }); + + test('starts sends no faster than sendsPerSecond', async function (assert) { + let h = harness({ sendsPerSecond: 10, maxInFlight: 100 }); + h.releaseAll(); + let startedAt = Date.now(); + await h.broadcast( + REALM_A, + incremental(REALM_A, 'card'), + Array.from({ length: 13 }, (_, i) => `!r${i}`), + ); + // The bucket starts with a second's worth of sends; the three past it + // wait 100ms apiece. + assert.true( + Date.now() - startedAt >= 250, + `13 sends at 10/s took ${Date.now() - startedAt}ms`, + ); + assert.strictEqual(h.started.length, 13); + }); + + test('a realm whose rooms cannot be resolved is skipped without stalling the line', async function (assert) { + let h = harness(); + h.releaseAll(); + await h.dispatcher.broadcast({ + realmURL: REALM_A, + event: incremental(REALM_A, 'card'), + newTxnId: () => 'txn', + prepare: async () => { + throw new Error('database is unreachable'); + }, + }); + await h.broadcast(REALM_A, incremental(REALM_A, 'next')); + assert.strictEqual(h.started.length, 1); + assert.strictEqual(h.dispatcher.takeStats().queuedEvents, 0); + }); + + module('isBackPressure', function () { + test('treats overload and unreachable answers as back-pressure, and nothing else', function (assert) { + for (let status of [429, 502, 503, 504]) { + assert.true(isBackPressure(new MatrixRequestError('x', status))); + } + for (let status of [400, 403, 404, 500]) { + assert.false(isBackPressure(new MatrixRequestError('x', status))); + } + assert.true(isBackPressure(new TypeError('fetch failed'))); + assert.false(isBackPressure(new Error('boom'))); + }); + }); + + module('coalesce', function () { + test('leaves completion events alone', function (assert) { + assert.strictEqual( + coalesce(incremental(REALM_A, 'a'), incremental(REALM_A, 'b')), + undefined, + ); + assert.strictEqual( + coalesce( + { eventName: 'index', indexType: 'full', realmURL: REALM_A }, + { eventName: 'index', indexType: 'full', realmURL: REALM_A }, + ), + undefined, + ); + assert.strictEqual( + coalesce(initiation(REALM_A, `${REALM_A}a`), incremental(REALM_A, 'a')), + undefined, + ); + }); + }); +}); diff --git a/packages/runtime-common/matrix-client.ts b/packages/runtime-common/matrix-client.ts index d3afb0f9628..1605d852b58 100644 --- a/packages/runtime-common/matrix-client.ts +++ b/packages/runtime-common/matrix-client.ts @@ -31,6 +31,18 @@ const joinedRoomsRequests = new WeakMap>(); // sharing the device must never reuse a transaction id; see nextTxnId. export const SERVER_MATRIX_DEVICE_ID = 'boxel-server'; +// A homeserver request that came back with a non-2xx status. The status is +// carried as a field so a caller can tell an overloaded or unreachable +// homeserver (429, 502, 503, 504) from a request the homeserver refused. +export class MatrixRequestError extends Error { + readonly status: number; + constructor(message: string, status: number) { + super(message); + this.name = 'MatrixRequestError'; + this.status = status; + } +} + export interface MatrixAccess { accessToken: string; deviceId: string; @@ -333,11 +345,19 @@ export class MatrixClient { return await fetchMatrixProfile(this.matrixURL, userId); } - async sendEvent(roomId: string, type: string, content: T) { + // A caller that retries a send passes the same `txnId` on every attempt, so + // synapse answers a retry of a send it already stored with the stored event + // rather than storing a second copy. That matters when a proxy in front of + // synapse times the request out while synapse goes on to complete it. + async sendEvent( + roomId: string, + type: string, + content: T, + txnId: string = this.nextTxnId(), + ) { if (!this.access) { throw new Error(`Missing matrix access token`); } - let txnId = this.nextTxnId(); let response = await this.request( `_matrix/client/v3/rooms/${roomId}/send/${type}/${txnId}`, @@ -345,14 +365,23 @@ export class MatrixClient { { body: JSON.stringify(content) }, ); - let json = (await response.json()) as { event_id: string }; if (!response.ok) { - throw new Error( - `Unable to send room event '${type}' to room ${roomId}: status ${ - response.status - } - ${JSON.stringify(json)}`, + // A proxy in front of synapse answers a timeout or an unreachable + // upstream with an HTML page, not a matrix error, so the body is read as + // text and kept out of the message unless it is JSON. + let body = await response.text().catch(() => ''); + let detail: string; + try { + detail = JSON.stringify(JSON.parse(body)); + } catch { + detail = `non-JSON body (${body.length} bytes)`; + } + throw new MatrixRequestError( + `Unable to send room event '${type}' to room ${roomId}: status ${response.status} - ${detail}`, + response.status, ); } + let json = (await response.json()) as { event_id: string }; return json.event_id; } @@ -530,7 +559,7 @@ export class MatrixClient { return uint8ArrayToHex(await hash.digest()); } - private nextTxnId() { + nextTxnId() { // Unique per client instance, and within it even when several events are // sent in the same millisecond. Every process logs in to the same device, // and synapse answers a repeated (device, transaction id) send with the From 102f4b8b7d85565aa2eed1f7ce5abbc75a15552d Mon Sep 17 00:00:00 2001 From: Hassan Abdel-Rahman Date: Wed, 21 Oct 2026 18:38:15 -0400 Subject: [PATCH 2/3] Keep realms' turns under the rate limit; pace back-pressure per episode Address review findings on the realm event dispatcher: - Check for a token before choosing a send, so waiting on the rate limit no longer moves the rotation past a realm and starves it. - Escalate the backoff once per overload episode: only a send started after the latest pause doubles it or resets it. - Honor a rate-limit answer's retry_after_ms / Retry-After, carried on MatrixRequestError.retryAfterMs, as the minimum pause. - Size the token bucket at least one, so a rate under one a second still sends. - Past the byte bound, drop an event a subscriber can lose before an indexing pass's completion. - Skip the room-lookup retries after a realm's last lookup found no rooms. - Wait for an in-flight re-login before a send instead of failing it. - The backlog alert reduces its window with min and drops `for`, so it fires only when every 1-minute bin is over the threshold. Co-Authored-By: Claude Opus 5.5 --- docs/realm-server-health-signals.md | 5 +- .../realm-server-availability-group.json | 6 +- packages/realm-server/node-realm.ts | 16 ++- .../realm-server/realm-event-dispatcher.ts | 91 +++++++++---- .../realm-server/tests/node-realm-test.ts | 4 + .../tests/realm-event-dispatcher-test.ts | 124 ++++++++++++++++++ packages/runtime-common/matrix-client.ts | 21 ++- 7 files changed, 234 insertions(+), 33 deletions(-) diff --git a/docs/realm-server-health-signals.md b/docs/realm-server-health-signals.md index c722155b634..43cacdfcefb 100644 --- a/docs/realm-server-health-signals.md +++ b/docs/realm-server-health-signals.md @@ -116,8 +116,9 @@ that synapse or the proxy in front of it answered with 429, 502, 503 or 504: `REALM_EVENT_MAX_SENDS_IN_FLIGHT` (default 8). `realmEventsDropped` counts events discarded because the queue passed -`REALM_EVENT_MAX_QUEUED_BYTES` (default 64 MiB). The queue drops the oldest -event of the realm with the longest backlog. +`REALM_EVENT_MAX_QUEUED_BYTES` (default 64 MiB). The realm with the longest +backlog loses its oldest queued event, preferring one that isn't the completion +of an indexing pass (`incremental` or `full`). The counters (`realmEventSends`, `realmEventFailedSends`, `realmEventBackPressure`, `realmEventsCoalesced`, `realmEventsDropped`) cover diff --git a/packages/observability/provisioning/alerting/realm-server-availability-group.json b/packages/observability/provisioning/alerting/realm-server-availability-group.json index d22df6e8c80..b06f23029b9 100644 --- a/packages/observability/provisioning/alerting/realm-server-availability-group.json +++ b/packages/observability/provisioning/alerting/realm-server-availability-group.json @@ -482,7 +482,7 @@ "title": "Realm Server Realm Event Backlog", "condition": "B", "annotations": { - "summary": "A realm-server process's realm event queue held an event for more than 60s before it started sending, at every 1-minute sample for 5 minutes. Sourced from realmEventQueueAgeMs on the periodic health line: the wait of the oldest queued event, which a queue that drains keeps near zero however deep a burst is. A held backlog means realm events (index, update) reach browsers late, so cards and searches look stale. Read realmEventBackPressure on the same line: nonzero means synapse is overloaded and the queue is pacing itself to it, so look at synapse CPU and what is writing in bulk; zero means the queue is at its configured send rate (REALM_EVENT_SENDS_PER_SECOND, REALM_EVENT_MAX_SENDS_IN_FLIGHT). realmEventsDropped counts events discarded once the queue reached REALM_EVENT_MAX_QUEUED_BYTES." + "summary": "A realm-server process's realm event queue held an event for more than 60s before it started sending in every 1-minute bin of the last 5 minutes (the rule reduces the window with min, so one bin under the threshold keeps it quiet). Sourced from realmEventQueueAgeMs on the periodic health line: the wait of the oldest queued event, which a queue that drains keeps near zero however deep a burst is. A held backlog means realm events (index, update) reach browsers late, so cards and searches look stale. Read realmEventBackPressure on the same line: nonzero means synapse is overloaded and the queue is pacing itself to it, so look at synapse CPU and what is writing in bulk; zero means the queue is at its configured send rate (REALM_EVENT_SENDS_PER_SECOND, REALM_EVENT_MAX_SENDS_IN_FLIGHT). realmEventsDropped counts events discarded once the queue reached REALM_EVENT_MAX_QUEUED_BYTES." }, "data": [ { @@ -541,7 +541,7 @@ "expression": "A", "intervalMs": 1000, "maxDataPoints": 43200, - "reducer": "max", + "reducer": "min", "refId": "C", "type": "reduce" } @@ -589,7 +589,7 @@ ], "noDataState": "OK", "execErrState": "Alerting", - "for": "5m", + "for": "0s", "isPaused": false, "notification_settings": null }, diff --git a/packages/realm-server/node-realm.ts b/packages/realm-server/node-realm.ts index 77a30ac5396..e7c6eb78b8c 100644 --- a/packages/realm-server/node-realm.ts +++ b/packages/realm-server/node-realm.ts @@ -94,6 +94,10 @@ export class NodeAdapter implements RealmAdapter { private realmDir: string; private enableFileWatcher?: boolean; private dispatcher: RealmEventDispatcher; + // Set when the realm's last room lookup found no session rooms. The lookup + // waits and retries only when this is clear, so a realm nobody has open + // pays the wait once rather than on every queued event. + private lastLookupFoundNoRooms = false; constructor( realmDir: string, @@ -486,13 +490,23 @@ export class NodeAdapter implements RealmAdapter { return []; } - let dmRooms = await this.waitForSessionRooms(dbAdapter, realmUrl); + let dmRooms = await this.waitForSessionRooms( + dbAdapter, + realmUrl, + this.lastLookupFoundNoRooms ? 0 : undefined, + ); + this.lastLookupFoundNoRooms = Object.keys(dmRooms).length === 0; realmEventsLog.debug('Sending to dm rooms', Object.values(dmRooms)); return Object.entries(dmRooms).map(([userId, roomId]) => ({ roomId, send: async (txnId: string) => { + // A send can start while a revoked session is being replaced, when + // the client holds no token; wait for the new one rather than fail. + if (!matrixClient.isLoggedIn()) { + await matrixClient.login(); + } await matrixClient.sendEvent( roomId, APP_BOXEL_REALM_EVENT_TYPE, diff --git a/packages/realm-server/realm-event-dispatcher.ts b/packages/realm-server/realm-event-dispatcher.ts index 641042d4058..7986c36a3fc 100644 --- a/packages/realm-server/realm-event-dispatcher.ts +++ b/packages/realm-server/realm-event-dispatcher.ts @@ -37,11 +37,16 @@ const log = logger('realm:events'); // - A send that synapse answers with 429, 502, 503 or 504, or that cannot // reach synapse at all, is back-pressure: the whole line pauses, and the // send is retried with the same transaction id, so a retry of a send that -// synapse did complete is not stored twice. The pause doubles on each -// consecutive back-pressure answer and resets on the next success. -// - The bytes queued are bounded. Past the bound, the oldest event of the -// realm with the longest backlog is dropped, so the realm causing the -// backlog is the one that loses events. +// synapse did complete is not stored twice. The pause is at least as long +// as synapse asked for in a rate-limit answer. It doubles for each pause in +// a row that ends in back-pressure again, and resets on a success. Only a +// send started after the latest pause counts toward either, so the sends +// that were already in flight when synapse became overloaded add up to one +// pause rather than doubling it once each. +// - The bytes queued are bounded. Past the bound, the realm with the longest +// backlog loses its oldest queued event, so the realm causing the backlog +// is the one that loses events. Events whose loss a subscriber recovers +// from go first (see `isCompletion`). // // The queue is in memory, so a process that exits with events queued does not // deliver them. @@ -135,6 +140,7 @@ export function isBackPressure(error: unknown): boolean { export class RealmEventDispatcher { readonly #maxInFlight: number; readonly #sendsPerSecond: number; + readonly #bucketSize: number; readonly #maxQueuedBytes: number; readonly #maxAttempts: number; readonly #initialBackoffMs: number; @@ -151,6 +157,10 @@ export class RealmEventDispatcher { #tokens: number; #tokensAt: number; #pausedUntil = 0; + // When the latest pause was set. A send started at or before it is part of + // the episode that set it; no send starts during a pause, so any later one + // started after the pause ended. + #pausedAt = 0; #backoffMs = 0; #timer: ReturnType | undefined; #timerAt = 0; @@ -175,8 +185,10 @@ export class RealmEventDispatcher { this.#maxBackoffMs = opts.maxBackoffMs ?? 30_000; this.#now = opts.now ?? (() => Date.now()); // The bucket holds one second's worth of sends, so an idle line can start - // a short burst at once and a long burst settles to the steady rate. - this.#tokens = this.#sendsPerSecond; + // a short burst at once and a long burst settles to the steady rate. It + // holds at least one, or a rate under one a second would never fill it. + this.#bucketSize = Math.max(1, this.#sendsPerSecond); + this.#tokens = this.#bucketSize; this.#tokensAt = this.#now(); } @@ -278,7 +290,10 @@ export class RealmEventDispatcher { if (!longest) { return; } - let dropped = longest.pending.shift()!; + let index = longest.pending.findIndex( + (entry) => !isCompletion(entry.event), + ); + let [dropped] = longest.pending.splice(Math.max(0, index), 1); this.#queuedEvents--; this.#queuedBytes -= dropped.bytes; this.#dropped++; @@ -311,19 +326,18 @@ export class RealmEventDispatcher { return; } this.#refill(now); - let next = this.#nextSend(); - if (!next) { - return; - } + // Checked before a send is chosen: choosing one moves the rotation on, + // so choosing a send that then has to wait would cost its realm a turn. if (this.#tokens < 1) { - // Put it back at the front of its realm's sends; the rotation has - // already moved on, so this realm is not favored when the line wakes. - next.active.ready!.unshift(next.pending); this.#wakeAt( now + Math.ceil(((1 - this.#tokens) * 1000) / this.#sendsPerSecond), ); return; } + let next = this.#nextSend(); + if (!next) { + return; + } this.#tokens--; this.#run(next.active, next.pending); } @@ -333,7 +347,7 @@ export class RealmEventDispatcher { let elapsed = now - this.#tokensAt; if (elapsed > 0) { this.#tokens = Math.min( - this.#sendsPerSecond, + this.#bucketSize, this.#tokens + (elapsed * this.#sendsPerSecond) / 1000, ); this.#tokensAt = now; @@ -425,19 +439,14 @@ export class RealmEventDispatcher { this.#sends++; this.#sendMsTotal += ms; this.#sendMsMax = Math.max(this.#sendMsMax, ms); - this.#backoffMs = 0; + if (startedAt > this.#pausedAt) { + this.#backoffMs = 0; + } }, async (error: unknown) => { if (isBackPressure(error)) { this.#backPressure++; - this.#backoffMs = Math.min( - this.#maxBackoffMs, - this.#backoffMs ? this.#backoffMs * 2 : this.#initialBackoffMs, - ); - this.#pausedUntil = Math.max( - this.#pausedUntil, - this.#now() + this.#backoffMs, - ); + this.#pauseFor(error, startedAt); if (pending.attempts < this.#maxAttempts) { active.ready!.push(pending); return; @@ -465,6 +474,25 @@ export class RealmEventDispatcher { }); } + #pauseFor(error: unknown, startedAt: number) { + let now = this.#now(); + if (startedAt > this.#pausedAt) { + this.#backoffMs = Math.min( + this.#maxBackoffMs, + this.#backoffMs ? this.#backoffMs * 2 : this.#initialBackoffMs, + ); + this.#pausedAt = now; + } + let retryAfterMs = (error as { retryAfterMs?: unknown }).retryAfterMs; + let pauseMs = Math.max( + this.#backoffMs, + typeof retryAfterMs === 'number' && Number.isFinite(retryAfterMs) + ? retryAfterMs + : 0, + ); + this.#pausedUntil = Math.max(this.#pausedUntil, now + pauseMs); + } + #finishIfDone(line: Line) { let active = line.active; if ( @@ -544,6 +572,19 @@ export function coalesce( return undefined; } +// A completion is the event that ends an indexing pass. A subscriber that +// misses one keeps showing the realm as indexing and keeps the cards the pass +// changed as they were, until a later completion arrives. Losing any other +// event costs less: an initiation is followed by its completion, an `update`'s +// paths are re-read on the next directory listing, and `prerender_html` is +// followed by the next pass's completion. +function isCompletion(event: RealmEventContent): boolean { + return ( + event.eventName === 'index' && + (event.indexType === 'incremental' || event.indexType === 'full') + ); +} + function union( a: string[] | undefined, b: string[] | undefined, diff --git a/packages/realm-server/tests/node-realm-test.ts b/packages/realm-server/tests/node-realm-test.ts index f5623ae8e88..3f2f898ab13 100644 --- a/packages/realm-server/tests/node-realm-test.ts +++ b/packages/realm-server/tests/node-realm-test.ts @@ -58,6 +58,7 @@ module(basename(import.meta.filename), function (hooks) { let matrixClient = { login: async () => undefined, getUserId: () => '@realm_server:localhost', + isLoggedIn: () => true, nextTxnId: () => 'txn-1', sendEvent: async () => { sendEventCalls++; @@ -93,6 +94,7 @@ module(basename(import.meta.filename), function (hooks) { let matrixClient = { login: async () => undefined, getUserId: () => '@realm_server:localhost', + isLoggedIn: () => true, nextTxnId: () => 'txn-1', sendEvent: async () => { throw new Error( @@ -132,6 +134,7 @@ module(basename(import.meta.filename), function (hooks) { let matrixClient = { login: async () => undefined, getUserId: () => '@realm_server:localhost', + isLoggedIn: () => true, nextTxnId: () => 'txn-1', sendEvent: async () => { failCleanup = true; @@ -160,6 +163,7 @@ module(basename(import.meta.filename), function (hooks) { let matrixClient = { login: async () => undefined, getUserId: () => '@realm_server:localhost', + isLoggedIn: () => true, nextTxnId: () => `txn-${txnIds.length}`, sendEvent: async ( _roomId: string, diff --git a/packages/realm-server/tests/realm-event-dispatcher-test.ts b/packages/realm-server/tests/realm-event-dispatcher-test.ts index ea0310f9d72..eeb4b3c1650 100644 --- a/packages/realm-server/tests/realm-event-dispatcher-test.ts +++ b/packages/realm-server/tests/realm-event-dispatcher-test.ts @@ -383,6 +383,130 @@ module(basename(import.meta.filename), function () { assert.strictEqual(h.dispatcher.takeStats().queuedEvents, 0); }); + test('keeps realms taking turns while the rate limit holds sends back', async function (assert) { + let h = harness({ sendsPerSecond: 100, maxInFlight: 2 }); + h.releaseAll(); + let rooms = (realm: string) => + Array.from({ length: 80 }, (_, i) => `!${realm}${i}`); + let realms = [REALM_A, REALM_B, 'http://127.0.0.1:4444/c/']; + await Promise.all( + realms.map((realm, i) => + h.broadcast(realm, incremental(realm, 'card'), rooms(`${i}-`)), + ), + ); + // The first second's worth of sends goes out as a burst; the ones after + // it are paced, which is where a realm could lose its turn. + let paced = h.started.slice(100, 160); + for (let realm of realms) { + let count = paced.filter((s) => s.realmURL === realm).length; + let fair = count >= 18 && count <= 22; + assert.true(fair, `${realm} got ${count} of 60 paced sends`); + } + }); + + test('sends that were in flight when synapse became overloaded add up to one pause, not one doubling each', async function (assert) { + let gate = deferred(); + let attempts = new Map(); + let h = harness({ maxInFlight: 4, initialBackoffMs: 50 }); + let startedAt = Date.now(); + let delivered = h.broadcast( + REALM_A, + incremental(REALM_A, 'card'), + ['!r1', '!r2', '!r3', '!r4'], + async ({ roomId }) => { + let n = (attempts.get(roomId) ?? 0) + 1; + attempts.set(roomId, n); + if (n === 1) { + await gate.promise; + throw new MatrixRequestError('status 503 - non-JSON body', 503); + } + }, + ); + await h.settle(); + gate.resolve(); + await delivered; + let elapsed = Date.now() - startedAt; + let pausedOnce = elapsed >= 45 && elapsed < 300; + assert.true( + pausedOnce, + `paused once for the initial backoff (${elapsed}ms), not 50+100+200+400`, + ); + assert.strictEqual(h.dispatcher.takeStats().backPressure, 4); + }); + + test('waits at least as long as a rate-limit answer asks', async function (assert) { + let attempts = 0; + let h = harness({ initialBackoffMs: 1 }); + let startedAt = Date.now(); + await h.broadcast( + REALM_A, + incremental(REALM_A, 'card'), + ['!r1'], + async () => { + attempts++; + if (attempts === 1) { + throw new MatrixRequestError( + 'status 429 - {"errcode":"M_LIMIT_EXCEEDED"}', + 429, + 200, + ); + } + }, + ); + assert.strictEqual(attempts, 2); + assert.true( + Date.now() - startedAt >= 195, + 'the retry waits out retry_after_ms rather than the shorter backoff', + ); + }); + + test('a rate under one send a second still sends', async function (assert) { + let h = harness({ sendsPerSecond: 0.5 }); + h.releaseAll(); + await h.broadcast(REALM_A, incremental(REALM_A, 'card')); + assert.strictEqual(h.started.length, 1); + }); + + test('past the byte bound, drops an event a subscriber can lose before a completion', async function (assert) { + let completion = incremental(REALM_A, 'card-0'); + let update: RealmEventContent = { + eventName: 'update', + added: ['x'.repeat(200)], + realmURL: REALM_A, + }; + let other = incremental(REALM_B, 'card-0'); + let bytes = (e: RealmEventContent) => Buffer.byteLength(JSON.stringify(e)); + let h = harness({ + maxInFlight: 1, + maxQueuedBytes: bytes(completion) + bytes(update) + bytes(other), + }); + void h.broadcast(REALM_A, incremental(REALM_A, 'sending')); + await h.settle(); + + void h.broadcast(REALM_A, completion); + void h.broadcast(REALM_A, update); + void h.broadcast(REALM_B, other); + void h.broadcast(REALM_B, incremental(REALM_B, 'card-1')); + assert.strictEqual(h.dispatcher.takeStats().dropped, 1); + + h.releaseAll(); + await h.dispatcher.idle(); + let sent = h.started.map((s) => s.event); + assert.false( + sent.some((e) => e.eventName === 'update'), + 'the update is dropped', + ); + assert.true( + sent.some( + (e) => + e.eventName === 'index' && + e.indexType === 'incremental' && + e.invalidations[0] === `${REALM_A}card-0`, + ), + 'the completion queued before it is delivered', + ); + }); + module('isBackPressure', function () { test('treats overload and unreachable answers as back-pressure, and nothing else', function (assert) { for (let status of [429, 502, 503, 504]) { diff --git a/packages/runtime-common/matrix-client.ts b/packages/runtime-common/matrix-client.ts index 1605d852b58..65586bfe957 100644 --- a/packages/runtime-common/matrix-client.ts +++ b/packages/runtime-common/matrix-client.ts @@ -34,12 +34,17 @@ export const SERVER_MATRIX_DEVICE_ID = 'boxel-server'; // A homeserver request that came back with a non-2xx status. The status is // carried as a field so a caller can tell an overloaded or unreachable // homeserver (429, 502, 503, 504) from a request the homeserver refused. +// `retryAfterMs` is how long the homeserver asked the caller to wait before +// trying again, from a rate-limit answer's `retry_after_ms` or its +// `Retry-After` header. export class MatrixRequestError extends Error { readonly status: number; - constructor(message: string, status: number) { + readonly retryAfterMs: number | undefined; + constructor(message: string, status: number, retryAfterMs?: number) { super(message); this.name = 'MatrixRequestError'; this.status = status; + this.retryAfterMs = retryAfterMs; } } @@ -371,14 +376,26 @@ export class MatrixClient { // text and kept out of the message unless it is JSON. let body = await response.text().catch(() => ''); let detail: string; + let retryAfterMs: number | undefined; try { - detail = JSON.stringify(JSON.parse(body)); + let json = JSON.parse(body); + detail = JSON.stringify(json); + if (typeof json?.retry_after_ms === 'number') { + retryAfterMs = json.retry_after_ms; + } } catch { detail = `non-JSON body (${body.length} bytes)`; } + if (retryAfterMs === undefined) { + let seconds = Number(response.headers.get('Retry-After') ?? NaN); + if (Number.isFinite(seconds) && seconds >= 0) { + retryAfterMs = seconds * 1000; + } + } throw new MatrixRequestError( `Unable to send room event '${type}' to room ${roomId}: status ${response.status} - ${detail}`, response.status, + retryAfterMs, ); } let json = (await response.json()) as { event_id: string }; From fb6cdccb070ecf80a28751d914d163d9b2c1da2d Mon Sep 17 00:00:00 2001 From: Hassan Abdel-Rahman Date: Thu, 22 Oct 2026 02:15:29 -0400 Subject: [PATCH 3/3] Bound how long a stuck send or overloaded login can hold the line; drain on shutdown Room sends and logins now give synapse 30s to answer, and a timeout counts as back-pressure, so one unanswered send no longer holds its realm's line for the HTTP client's multi-minute default. A login that synapse or its proxy answers with 429/502/503/504 now throws MatrixRequestError, and the dispatcher prepares the event again after the pause instead of skipping it. A server stopping on purpose waits up to 10s for queued realm events before closing its database. Co-Authored-By: Claude Opus 5.5 --- docs/realm-server-health-signals.md | 10 +- packages/realm-server/main.ts | 15 +++ packages/realm-server/node-realm.ts | 6 ++ .../realm-server/realm-event-dispatcher.ts | 55 ++++++++++- .../realm-server/tests/node-realm-test.ts | 40 ++++++++ .../tests/realm-event-dispatcher-test.ts | 92 +++++++++++++++++++ packages/runtime-common/matrix-client.ts | 83 ++++++++++------- 7 files changed, 263 insertions(+), 38 deletions(-) diff --git a/docs/realm-server-health-signals.md b/docs/realm-server-health-signals.md index 43cacdfcefb..355ea01a47a 100644 --- a/docs/realm-server-health-signals.md +++ b/docs/realm-server-health-signals.md @@ -106,8 +106,9 @@ realmEventsQueued=… realmEventQueuedKB=… realmEventQueueAgeMs=… realmEvent waited, which is the delay a browser sees on top of the send itself. A burst that drains keeps it low however deep the queue gets, so it, not the depth, is what the "Realm Server Realm Event Backlog" alert reads. When it climbs, the -next field to read is `realmEventBackPressure`, the count of sends in the window -that synapse or the proxy in front of it answered with 429, 502, 503 or 504: +next field to read is `realmEventBackPressure`. It counts the sends and logins in +the window that synapse, or the proxy in front of it, answered with 429, 502, +503 or 504, did not answer within 30 seconds, or could not be reached for: - Nonzero: synapse is overloaded, and the queue is pausing to let it recover. Look at synapse CPU and at what is writing in bulk. @@ -120,6 +121,11 @@ that synapse or the proxy in front of it answered with 429, 502, 503 or 504: backlog loses its oldest queued event, preferring one that isn't the completion of an indexing pass (`incremental` or `full`). +The queue is in memory. A server stopping on purpose waits up to 10 seconds for +it to empty before closing its database. If it is not empty by then, the server +logs `realm events still undelivered at shutdown` with the counts it is giving +up on. + The counters (`realmEventSends`, `realmEventFailedSends`, `realmEventBackPressure`, `realmEventsCoalesced`, `realmEventsDropped`) cover the window since the previous line. The others are read at the moment of the diff --git a/packages/realm-server/main.ts b/packages/realm-server/main.ts index 47322356f9d..dd4627c5693 100644 --- a/packages/realm-server/main.ts +++ b/packages/realm-server/main.ts @@ -53,6 +53,7 @@ import { ModuleCacheCoordinator } from './lib/module-cache-coordination.ts'; import { JobsFinishedListener } from './lib/jobs-finished-listener.ts'; import { JobScopedSearchCache } from './job-scoped-search-cache.ts'; import { startHealthSampler } from './health-sampler.ts'; +import { realmEventDispatcher } from './realm-event-dispatcher.ts'; import { healthCheckUserAgents } from './middleware/index.ts'; import { isDeployedServerURL, @@ -152,6 +153,10 @@ const AI_BOT_DELEGATION_SECRET = process.env.AI_BOT_DELEGATION_SECRET; const MATRIX_ADMIN_USERNAME = process.env.MATRIX_ADMIN_USERNAME; const MATRIX_ADMIN_PASSWORD = process.env.MATRIX_ADMIN_PASSWORD; +// How long shutdown waits for queued realm events to go out. Kept well inside +// the grace period a deploy gives a stopping task before it is killed. +const REALM_EVENT_DRAIN_MS = 10_000; + // The Google client the secure window signs in with: the Matrix server's own, // so a user sees no new consent. Both unset, the secure window offers only a // password. GOOGLE_OAUTH_ISSUER points it at a stand-in provider in tests. @@ -986,6 +991,16 @@ const reportHostShellToManager = async (dbAdapter: PgAdapter) => { // still turning while the shutdown above runs, and a check arriving in // that window should read a beating heartbeat rather than a frozen one. heartbeat.stop(); + // Queued realm events still need the database to resolve their rooms, + // so they get a bounded chance to go out before it closes. Losing one + // can leave open tabs showing a realm as still indexing. + if (!(await realmEventDispatcher().drain(REALM_EVENT_DRAIN_MS))) { + let { queuedEvents, sendsInFlight } = + realmEventDispatcher().takeStats(); + console.warn( + `realm events still undelivered at shutdown: ${queuedEvents} queued, ${sendsInFlight} sends in flight`, + ); + } queue.destroy(); // warning this is async dbAdapter.close(); // warning this is async console.log(`realm server on port ${stopPort} has stopped`); diff --git a/packages/realm-server/node-realm.ts b/packages/realm-server/node-realm.ts index e7c6eb78b8c..29efabfc740 100644 --- a/packages/realm-server/node-realm.ts +++ b/packages/realm-server/node-realm.ts @@ -56,6 +56,7 @@ import type { import { APP_BOXEL_REALM_EVENT_TYPE } from '@cardstack/runtime-common/matrix-constants'; import { createJWT, verifyJWT } from './jwt.ts'; import { + isBackPressure, realmEventDispatcher, type RealmEventDispatcher, type RealmEventSend, @@ -486,6 +487,11 @@ export class NodeAdapter implements RealmAdapter { return []; } } catch (e) { + // An overloaded or unreachable homeserver is back-pressure: the + // dispatcher pauses and prepares the event again rather than skip it. + if (isBackPressure(e)) { + throw e; + } realmEventsLog.error('Error logging into matrix. Skipping broadcast', e); return []; } diff --git a/packages/realm-server/realm-event-dispatcher.ts b/packages/realm-server/realm-event-dispatcher.ts index 7986c36a3fc..1d262533949 100644 --- a/packages/realm-server/realm-event-dispatcher.ts +++ b/packages/realm-server/realm-event-dispatcher.ts @@ -48,8 +48,9 @@ const log = logger('realm:events'); // is the one that loses events. Events whose loss a subscriber recovers // from go first (see `isCompletion`). // -// The queue is in memory, so a process that exits with events queued does not -// deliver them. +// The queue is in memory. A process stopping on purpose waits a bounded time +// for it to empty (see `drain`); a process that exits any other way, or whose +// queue outlasts that wait, does not deliver what is still queued. export interface RealmEventSend { roomId: string; @@ -65,7 +66,9 @@ export interface RealmEventBroadcast { event: RealmEventContent; // Resolves the sends the event fans out to. Runs when the event reaches the // head of its realm's line, with the event as merged by then, so the rooms - // are the ones subscribed at send time. + // are the ones subscribed at send time. When it rejects with back-pressure + // the event goes back to the head of its line and is prepared again after + // the pause, up to `maxAttempts` times; any other rejection skips the event. prepare(event: RealmEventContent): Promise; newTxnId(): string; } @@ -105,6 +108,7 @@ interface Entry { prepare: RealmEventBroadcast['prepare']; newTxnId: RealmEventBroadcast['newTxnId']; waiters: (() => void)[]; + prepareAttempts: number; } interface PendingSend { @@ -132,6 +136,14 @@ export function isBackPressure(error: unknown): boolean { if (typeof status === 'number') { return BACK_PRESSURE_STATUSES.has(status); } + // A request given a deadline rejects with a TimeoutError (or an AbortError, + // depending on the runtime) when the host does not answer in time. + if ( + error instanceof Error && + (error.name === 'TimeoutError' || error.name === 'AbortError') + ) { + return true; + } // fetch rejects with a TypeError when it cannot reach the host at all // (connection refused or reset, DNS failure, socket closed mid-response). return error instanceof TypeError && /fetch failed/i.test(error.message); @@ -222,6 +234,7 @@ export class RealmEventDispatcher { prepare: broadcast.prepare, newTxnId: broadcast.newTxnId, waiters: [resolve], + prepareAttempts: 0, }; line.pending.push(entry); this.#queuedEvents++; @@ -240,6 +253,21 @@ export class RealmEventDispatcher { await new Promise((resolve) => this.#idleWaiters.push(resolve)); } + // Resolves true once nothing is queued or in flight, or false if that takes + // longer than `timeoutMs`. A process stopping on purpose waits on this so the + // events it has queued are not lost with it. + async drain(timeoutMs: number): Promise { + let timer: ReturnType | undefined; + let timedOut = new Promise((resolve) => { + timer = setTimeout(() => resolve(false), timeoutMs); + }); + try { + return await Promise.race([this.idle().then(() => true), timedOut]); + } finally { + clearTimeout(timer); + } + } + takeStats(): RealmEventDispatcherStats { let now = this.#now(); let oldest = now; @@ -405,9 +433,18 @@ export class RealmEventDispatcher { this.#queuedBytes -= entry.bytes; let active: Active = { entry, ready: undefined, outstanding: 0 }; line.active = active; + entry.prepareAttempts++; + let startedAt = this.#now(); entry .prepare(entry.event) - .catch((e) => { + .catch((e: unknown) => { + if (isBackPressure(e)) { + this.#backPressure++; + this.#pauseFor(e, startedAt); + if (entry.prepareAttempts < this.#maxAttempts) { + return undefined; + } + } log.error( `Unable to resolve the rooms for a realm event in ${entry.realmURL}; skipping it`, e, @@ -415,6 +452,16 @@ export class RealmEventDispatcher { return []; }) .then((sends) => { + if (!sends) { + // Back to the head of the line, to be prepared again once the + // pause is over. + line.active = undefined; + line.pending.unshift(entry); + this.#queuedEvents++; + this.#queuedBytes += entry.bytes; + this.#pump(); + return; + } active.ready = sends.map((send) => ({ send, txnId: entry.newTxnId(), diff --git a/packages/realm-server/tests/node-realm-test.ts b/packages/realm-server/tests/node-realm-test.ts index 3f2f898ab13..d0c43dc266b 100644 --- a/packages/realm-server/tests/node-realm-test.ts +++ b/packages/realm-server/tests/node-realm-test.ts @@ -205,6 +205,46 @@ module(basename(import.meta.filename), function (hooks) { 'an overloaded homeserver does not clear the session room', ); }); + + test('logs in again after a login synapse answers as overloaded, rather than skip the event', async function (assert) { + await insertSessionForAlice(); + + let logins = 0; + let sent: string[] = []; + let matrixClient = { + login: async () => { + logins++; + if (logins === 1) { + throw new MatrixRequestError( + `Unable to login to matrix: status 502 - non-JSON body (150 bytes)`, + 502, + ); + } + }, + getUserId: () => '@realm_server:localhost', + isLoggedIn: () => true, + nextTxnId: () => 'txn-1', + sendEvent: async (roomId: string) => { + sent.push(roomId); + return '$event'; + }, + } as unknown as MatrixClient; + + let adapter = new NodeAdapter( + '/tmp', + false, + new RealmEventDispatcher({ initialBackoffMs: 1 }), + ); + await adapter.broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + ); + + assert.strictEqual(logins, 2, 'the login is tried again after a pause'); + assert.deepEqual(sent, [staleRoomId], 'the event still reaches the room'); + }); }); module(`${basename(import.meta.filename)} | file stat probing`, function () { diff --git a/packages/realm-server/tests/realm-event-dispatcher-test.ts b/packages/realm-server/tests/realm-event-dispatcher-test.ts index eeb4b3c1650..7c455172799 100644 --- a/packages/realm-server/tests/realm-event-dispatcher-test.ts +++ b/packages/realm-server/tests/realm-event-dispatcher-test.ts @@ -383,6 +383,93 @@ module(basename(import.meta.filename), function () { assert.strictEqual(h.dispatcher.takeStats().queuedEvents, 0); }); + test('retries a send that outlived its deadline, with the same transaction id', async function (assert) { + let attempts: string[] = []; + let h = harness({ initialBackoffMs: 1 }); + await h.broadcast( + REALM_A, + incremental(REALM_A, 'card'), + ['!r1'], + async ({ txnId }) => { + attempts.push(txnId); + if (attempts.length === 1) { + throw new DOMException('timed out', 'TimeoutError'); + } + }, + ); + assert.deepEqual(attempts, ['txn-0', 'txn-0']); + assert.deepEqual(h.failures, []); + }); + + test('prepares an event again after its rooms could not be resolved for back-pressure', async function (assert) { + let h = harness({ initialBackoffMs: 20 }); + h.releaseAll(); + let prepares = 0; + let startedAt = Date.now(); + let delivered: string[] = []; + await h.dispatcher.broadcast({ + realmURL: REALM_A, + event: incremental(REALM_A, 'card'), + newTxnId: () => 'txn', + prepare: async () => { + prepares++; + if (prepares === 1) { + throw new MatrixRequestError('status 503 - non-JSON body', 503); + } + return [ + { + roomId: '!r1', + send: async () => { + delivered.push('!r1'); + }, + onFailure: async () => undefined, + }, + ]; + }, + }); + assert.strictEqual(prepares, 2); + assert.deepEqual(delivered, ['!r1']); + assert.true(Date.now() - startedAt >= 18, 'it waits out the pause first'); + let stats = h.dispatcher.takeStats(); + assert.strictEqual(stats.backPressure, 1); + assert.strictEqual(stats.queuedEvents, 0); + }); + + test('skips an event whose rooms stay unresolvable for back-pressure past maxAttempts', async function (assert) { + let h = harness({ maxAttempts: 3, initialBackoffMs: 1 }); + h.releaseAll(); + let prepares = 0; + await h.dispatcher.broadcast({ + realmURL: REALM_A, + event: incremental(REALM_A, 'card'), + newTxnId: () => 'txn', + prepare: async () => { + prepares++; + throw new MatrixRequestError('status 502 - non-JSON body', 502); + }, + }); + await h.broadcast(REALM_A, incremental(REALM_A, 'next')); + assert.strictEqual(prepares, 3); + assert.strictEqual(h.started.length, 1, 'the next event still goes out'); + }); + + test('drain resolves true once the queue empties, and false if it outlasts the wait', async function (assert) { + let h = harness(); + assert.true(await h.dispatcher.drain(10), 'an empty queue is drained'); + + let delivered = h.broadcast(REALM_A, incremental(REALM_A, 'card')); + await h.settle(); + assert.false( + await h.dispatcher.drain(20), + 'a send still outstanding outlasts the wait', + ); + + let drained = h.dispatcher.drain(1000); + h.releaseAll(); + assert.true(await drained); + await delivered; + }); + test('keeps realms taking turns while the rate limit holds sends back', async function (assert) { let h = harness({ sendsPerSecond: 100, maxInFlight: 2 }); h.releaseAll(); @@ -516,6 +603,11 @@ module(basename(import.meta.filename), function () { assert.false(isBackPressure(new MatrixRequestError('x', status))); } assert.true(isBackPressure(new TypeError('fetch failed'))); + assert.true( + isBackPressure(new DOMException('timed out', 'TimeoutError')), + 'a request that outlived its deadline', + ); + assert.true(isBackPressure(new DOMException('aborted', 'AbortError'))); assert.false(isBackPressure(new Error('boom'))); }); }); diff --git a/packages/runtime-common/matrix-client.ts b/packages/runtime-common/matrix-client.ts index 65586bfe957..941131f75d6 100644 --- a/packages/runtime-common/matrix-client.ts +++ b/packages/runtime-common/matrix-client.ts @@ -48,6 +48,45 @@ export class MatrixRequestError extends Error { } } +// Builds the error for a non-2xx answer. A proxy in front of synapse answers a +// timeout or an unreachable upstream with an HTML page, not a matrix error, so +// the body is read as text and kept out of the message unless it is JSON. +async function matrixRequestError( + response: Response, + message: string, +): Promise { + let body = await response.text().catch(() => ''); + let detail: string; + let retryAfterMs: number | undefined; + try { + let json = JSON.parse(body); + detail = JSON.stringify(json); + if (typeof json?.retry_after_ms === 'number') { + retryAfterMs = json.retry_after_ms; + } + } catch { + detail = `non-JSON body (${body.length} bytes)`; + } + if (retryAfterMs === undefined) { + let seconds = Number(response.headers.get('Retry-After') ?? NaN); + if (Number.isFinite(seconds) && seconds >= 0) { + retryAfterMs = seconds * 1000; + } + } + return new MatrixRequestError( + `${message}: status ${response.status} - ${detail}`, + response.status, + retryAfterMs, + ); +} + +// How long a login or a room send waits for synapse to answer. A request +// synapse accepts but never answers would otherwise wait out the HTTP client's +// own headers timeout (minutes), holding up everything queued behind it. When +// it fires the request rejects with a `TimeoutError`, which a caller can retry: +// a send retried with the same transaction id is not stored twice. +const MATRIX_WRITE_TIMEOUT_MS = 30_000; + export interface MatrixAccess { accessToken: string; deviceId: string; @@ -196,19 +235,18 @@ export class MatrixClient { type: 'm.login.password', device_id: SERVER_MATRIX_DEVICE_ID, }), + signal: AbortSignal.timeout(MATRIX_WRITE_TIMEOUT_MS), }, false, ); - let json = await response.json(); - if (!response.ok) { - throw new Error( - `Unable to login to matrix ${this.matrixURL.href} as user ${ - this.username - }: status ${response.status} - ${JSON.stringify(json)}`, + throw await matrixRequestError( + response, + `Unable to login to matrix ${this.matrixURL.href} as user ${this.username}`, ); } + let json = await response.json(); let { access_token: accessToken, device_id: deviceId, @@ -367,35 +405,16 @@ export class MatrixClient { let response = await this.request( `_matrix/client/v3/rooms/${roomId}/send/${type}/${txnId}`, 'PUT', - { body: JSON.stringify(content) }, + { + body: JSON.stringify(content), + signal: AbortSignal.timeout(MATRIX_WRITE_TIMEOUT_MS), + }, ); if (!response.ok) { - // A proxy in front of synapse answers a timeout or an unreachable - // upstream with an HTML page, not a matrix error, so the body is read as - // text and kept out of the message unless it is JSON. - let body = await response.text().catch(() => ''); - let detail: string; - let retryAfterMs: number | undefined; - try { - let json = JSON.parse(body); - detail = JSON.stringify(json); - if (typeof json?.retry_after_ms === 'number') { - retryAfterMs = json.retry_after_ms; - } - } catch { - detail = `non-JSON body (${body.length} bytes)`; - } - if (retryAfterMs === undefined) { - let seconds = Number(response.headers.get('Retry-After') ?? NaN); - if (Number.isFinite(seconds) && seconds >= 0) { - retryAfterMs = seconds * 1000; - } - } - throw new MatrixRequestError( - `Unable to send room event '${type}' to room ${roomId}: status ${response.status} - ${detail}`, - response.status, - retryAfterMs, + throw await matrixRequestError( + response, + `Unable to send room event '${type}' to room ${roomId}`, ); } let json = (await response.json()) as { event_id: string };