Deliver realm events to synapse through one bounded, paced queue per process

docs/realm-server-health-signals.md

@@ -94,6 +94,43 @@ 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`. 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.
+- 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 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 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
+sample.
+
 ## `_readiness-check` is not one of them
 
 `GET /<realm>/_readiness-check` shares the word "readiness" but answers a different

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 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": [
+            {
+              "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=(?<ageMs>\\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": "min",
+                "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": "0s",
+          "isPaused": false,
+          "notification_settings": null
+        },
         {
           "uid": "rs-search-saturation",
           "orgID": 1,

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);

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`);

packages/realm-server/node-realm.ts

@@ -55,6 +55,12 @@ 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 {
+  isBackPressure,
+  realmEventDispatcher,
+  type RealmEventDispatcher,
+  type RealmEventSend,
+} from './realm-event-dispatcher.ts';
 
 const realmEventsLog = logger('realm:events');
 
@@ -88,10 +94,20 @@ function lastModifiedSeconds(stat: Stats): number {
 export class NodeAdapter implements RealmAdapter {
   private realmDir: string;
   private enableFileWatcher?: boolean;
-
-  constructor(realmDir: string, 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,
+    enableFileWatcher?: boolean,
+    dispatcher: RealmEventDispatcher = realmEventDispatcher(),
+  ) {
     this.realmDir = realmDir;
     this.enableFileWatcher = enableFileWatcher;
+    this.dispatcher = dispatcher;
   }
 
   get dir(): string {
@@ -421,6 +437,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,33 +457,70 @@ 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<RealmEventSend[]> {
+    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()) {
         realmEventsLog.error(
           'Matrix client has no user ID after login, unable to broadcast realm event',
           event,
         );
-        return;
+        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;
+      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));
 
-    for (let userId of Object.keys(dmRooms)) {
-      let roomId = dmRooms[userId];
-      try {
+    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,
-          eventWithRealmURL,
+          event,
+          txnId,
         );
-      } catch (e) {
+      },
+      onFailure: async (e: unknown) => {
         if (isRealmServerNotInRoomError(e, roomId)) {
           try {
             let cleared = await clearSessionRoom(dbAdapter, userId, roomId);
@@ -483,15 +539,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(