From 8c2a2e2c9a851fe93a974abe75466c97de807b5b Mon Sep 17 00:00:00 2001 From: Hassan Abdel-Rahman Date: Wed, 21 Oct 2026 19:00:18 -0400 Subject: [PATCH 1/2] Announce a write that changes nothing only to the writer that sent it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When every file in a commit is byte-identical to what is stored, the commit queues no index pass, yet it still broadcast an `incremental` event with no invalidations to every session room subscribed to the realm — every user's room, for a public realm. Bulk writers that resend unchanged content (the nightly OpenRouter sync, the skills and homepage syncs) paid one fan-out per request for nothing. Such a commit now announces nothing when the request named no client request id, since nobody can be waiting for that event. When it did name one, the event goes only to the writer's own session room: the host's AI-assistant waiter keys its wait on the event carrying its request id, and only the writer's client can be holding that wait. A module flush that ran ahead of byte-identical instances, or a shared pass this writer was chosen to announce, is still announced to every subscriber. RealmAdapter.broadcastRealmEvent takes an optional delivery argument whose `recipients` narrows the send to those matrix users' session rooms. Co-Authored-By: Claude Opus 5.5 --- packages/realm-server/node-realm.ts | 8 + .../tests/atomic-endpoints-test.ts | 196 ++++++++++++++++++ .../realm-server/tests/node-realm-test.ts | 54 +++++ packages/runtime-common/index.ts | 1 + packages/runtime-common/realm.ts | 82 ++++++-- 5 files changed, 321 insertions(+), 20 deletions(-) diff --git a/packages/realm-server/node-realm.ts b/packages/realm-server/node-realm.ts index 4e90ae9c76e..c018be95308 100644 --- a/packages/realm-server/node-realm.ts +++ b/packages/realm-server/node-realm.ts @@ -1,5 +1,6 @@ import type { RealmAdapter, + RealmEventDelivery, Kind, FileRef, DBAdapter, @@ -426,6 +427,7 @@ export class NodeAdapter implements RealmAdapter { realmUrl: string, matrixClient: MatrixClient, dbAdapter: DBAdapter, + opts?: RealmEventDelivery, ): Promise { realmEventsLog.debug('Broadcasting realm event', event); const eventWithRealmURL: RealmEventContent = { @@ -453,6 +455,12 @@ export class NodeAdapter implements RealmAdapter { } let dmRooms = await this.waitForSessionRooms(dbAdapter, realmUrl); + if (opts?.recipients) { + let recipients = new Set(opts.recipients); + dmRooms = Object.fromEntries( + Object.entries(dmRooms).filter(([userId]) => recipients.has(userId)), + ); + } realmEventsLog.debug('Sending to dm rooms', Object.values(dmRooms)); diff --git a/packages/realm-server/tests/atomic-endpoints-test.ts b/packages/realm-server/tests/atomic-endpoints-test.ts index 8112daaf63f..315ed31a6ce 100644 --- a/packages/realm-server/tests/atomic-endpoints-test.ts +++ b/packages/realm-server/tests/atomic-endpoints-test.ts @@ -925,6 +925,202 @@ module(basename(import.meta.filename), function () { } }); + test('a write that changes nothing announces its index event only to its own writer', async function (assert) { + let source = ` + import { field, CardDef, contains } from "@cardstack/base/card-api"; + import StringField from "@cardstack/base/string"; + export class Place extends CardDef { + static displayName = 'Place'; + @field name = contains(StringField); + } + `.trim(); + let doc = (op: 'add' | 'update') => ({ + 'atomic:operations': [ + { + op, + href: 'place-modules/place-silent.gts', + data: { + type: 'source', + attributes: { content: source }, + meta: {}, + }, + }, + ], + }); + let post = (op: 'add' | 'update', clientRequestId?: string) => { + let req = request + .post('/_atomic') + .set('Accept', SupportedMimeType.JSONAPI) + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'user', ['read', 'write'])}`, + ); + if (clientRequestId) { + req = req.set('X-Boxel-Client-Request-Id', clientRequestId); + } + return req.send(JSON.stringify(doc(op))); + }; + + let addResponse = await post('add'); + assert.strictEqual(addResponse.status, 201, 'initial write succeeds'); + // The initial write's own index event lands before the broadcasts + // below are recorded. + await testRealm.incrementalIndexing(); + + let broadcasts: { + event: RealmEventContent; + recipients: string[] | undefined; + }[] = []; + let originalBroadcast = + testRealmAdapter.broadcastRealmEvent.bind(testRealmAdapter); + testRealmAdapter.broadcastRealmEvent = (async ( + event, + realmUrl, + matrixClient, + dbAdapter, + opts, + ) => { + broadcasts.push({ event, recipients: opts?.recipients }); + return originalBroadcast( + event, + realmUrl, + matrixClient, + dbAdapter, + opts, + ); + }) as RealmAdapter['broadcastRealmEvent']; + let indexEvents = () => + broadcasts.filter( + ({ event }) => + event.eventName === 'index' && + 'indexType' in event && + event.indexType === 'incremental', + ); + + try { + let anonymous = await post('update'); + assert.strictEqual(anonymous.status, 201, 'no-op write succeeds'); + assert.deepEqual( + broadcasts.map(({ event }) => event.eventName), + [], + 'a no-op write that names no request id broadcasts nothing', + ); + + let identified = await post('update', 'test-noop-request'); + assert.strictEqual(identified.status, 201, 'no-op write succeeds'); + let events = indexEvents(); + assert.strictEqual( + events.length, + 1, + 'a no-op write that names a request id announces its index event once', + ); + assert.strictEqual( + (events[0]?.event as { clientRequestId?: string | null }) + .clientRequestId, + 'test-noop-request', + 'the event carries the request id', + ); + assert.deepEqual( + events[0]?.recipients, + ['user'], + "the event goes to the writer's own session room only", + ); + assert.strictEqual( + broadcasts.length, + 1, + 'no initiation or file update event goes out for a no-op write', + ); + + // A module that changes ahead of a byte-identical instance runs + // an index pass of its own, and that pass is every subscriber's + // news even though the instance changed nothing. + let instance = JSON.stringify({ + data: { + type: 'card', + attributes: { name: 'Silent' }, + meta: { + adoptsFrom: { + module: './place-modules/place-silent', + name: 'Place', + }, + }, + }, + }); + let mixed = (moduleSource: string) => ({ + 'atomic:operations': [ + { + op: 'update', + href: 'place-modules/place-silent.gts', + data: { + type: 'source', + attributes: { content: moduleSource }, + meta: {}, + }, + }, + { + op: 'add', + href: 'place-silent-1.json', + data: { + type: 'source', + attributes: { content: instance }, + meta: {}, + }, + }, + ], + }); + let send = (body: object) => + request + .post('/_atomic') + .set('Accept', SupportedMimeType.JSONAPI) + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'user', ['read', 'write'])}`, + ) + .send(JSON.stringify(body)); + assert.strictEqual( + (await send(mixed(source))).status, + 201, + 'the instance is written beside the unchanged module', + ); + await testRealm.incrementalIndexing(); + + broadcasts = []; + let changedSource = source.replace( + "static displayName = 'Place';", + "static displayName = 'Silent Place';", + ); + let mixedBody = mixed(changedSource); + mixedBody['atomic:operations'][1].op = 'update'; + assert.strictEqual( + (await send(mixedBody)).status, + 201, + 'the module changes and the instance does not', + ); + await testRealm.incrementalIndexing(); + let flushEvents = indexEvents(); + assert.strictEqual( + flushEvents.length, + 1, + "the module flush's pass is announced", + ); + assert.strictEqual( + flushEvents[0]?.recipients, + undefined, + 'to every subscriber', + ); + assert.true( + ( + flushEvents[0]?.event as { invalidations?: string[] } + ).invalidations?.some((url) => + url.includes('place-modules/place-silent'), + ), + 'naming the module the flush invalidated', + ); + } finally { + testRealmAdapter.broadcastRealmEvent = originalBroadcast; + } + }); + test('waitForIndex write preserves compound field values backed by an unexported FieldDef', async function (assert) { let logSource = ` import { diff --git a/packages/realm-server/tests/node-realm-test.ts b/packages/realm-server/tests/node-realm-test.ts index a1e8631c9a7..f90115bd45c 100644 --- a/packages/realm-server/tests/node-realm-test.ts +++ b/packages/realm-server/tests/node-realm-test.ts @@ -147,6 +147,60 @@ module(basename(import.meta.filename), function (hooks) { assert.true(true, 'broadcast resolves even if stale room cleanup fails'); }); + + test('delivers only to the named recipients when the broadcast names any', async function (assert) { + await insertSessionForAlice(); + await insertUser( + dbAdapter, + '@carol:localhost', + 'cus_carol', + 'c@example.com', + ); + await upsertSessionRoom( + dbAdapter, + '@carol:localhost', + '!room-carol:localhost', + ); + await insertPermissions(dbAdapter, realmURL, { + '@carol:localhost': ['read'], + }); + + let sentTo: string[] = []; + let matrixClient = { + login: async () => undefined, + getUserId: () => '@realm_server:localhost', + sendEvent: async (roomId: string) => { + sentTo.push(roomId); + }, + } as unknown as MatrixClient; + + let adapter = new NodeAdapter('/tmp'); + await adapter.broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + ); + assert.deepEqual( + sentTo.sort(), + ['!room-alice:localhost', '!room-carol:localhost'], + 'with no recipients, every subscriber receives the event', + ); + + sentTo = []; + await adapter.broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + { recipients: ['@carol:localhost', '@nobody:localhost'] }, + ); + assert.deepEqual( + sentTo, + ['!room-carol:localhost'], + "only the named subscriber's room receives the event", + ); + }); }); module(`${basename(import.meta.filename)} | file stat probing`, function () { diff --git a/packages/runtime-common/index.ts b/packages/runtime-common/index.ts index 7569dc89415..894dbfe8dcb 100644 --- a/packages/runtime-common/index.ts +++ b/packages/runtime-common/index.ts @@ -2020,6 +2020,7 @@ export { RealmAuthDataSource } from './realm-auth-data-source.ts'; export type { Kind, RealmAdapter, + RealmEventDelivery, FileRef, RealmIndexCounts, RealmInfo, diff --git a/packages/runtime-common/realm.ts b/packages/runtime-common/realm.ts index 6d5bedca5d7..caadfee3ff0 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -2337,6 +2337,14 @@ export interface WriteOptions { onDurable?: () => void; } +// Who a realm event is delivered to. By default it goes to the session room +// of every user who can read the realm. `recipients` narrows that to the named +// matrix users' session rooms, for an event that concerns only them; a +// recipient with no session room for the realm receives nothing. +export interface RealmEventDelivery { + recipients?: string[]; +} + export interface RealmAdapter { readdir( path: LocalPath, @@ -2434,6 +2442,7 @@ export interface RealmAdapter { realmUrl: string, matrixClient: MatrixClient, dbAdapter: DBAdapter, + opts?: RealmEventDelivery, ): Promise; // optional, set this to override _lint endpoint behavior in tests @@ -4154,6 +4163,9 @@ export class Realm { // False for a writer that asked not to announce its write (see // `WriteOptions.broadcastRealmEvents`). broadcastRealmEvents?: boolean; + // Narrows delivery to these matrix users' session rooms (see + // `RealmEventDelivery`). Absent, every subscriber receives the event. + recipients?: string[]; }, ): void { // A pass indexed alongside other publishes is announced once, by the @@ -4244,7 +4256,10 @@ export class Realm { versions = versionsOnlyThisWriterChanged(passes, versions); content = this.#withCoalescedWrites(content, coalescedWrites); } - this.broadcastRealmEvent(withVersionsIfItFits(content, versions)); + this.broadcastRealmEvent( + withVersionsIfItFits(content, versions), + opts?.recipients ? { recipients: opts.recipients } : undefined, + ); } // `coalescedWrites` on a shared pass's event, as much of it as fits. @@ -5754,9 +5769,25 @@ export class Realm { }); } } else { - // Nothing changed on disk (e.g., every file's content was already what - // the caller staged). Preserve the pre-existing always-broadcast - // behavior. + // Nothing changed on disk: every file's content was already what the + // caller staged. + // + // A module flush that ran ahead of byte-identical instances is the only + // pass that can stand behind this event, and what it invalidated is + // owed to every subscriber like any other pass — unless it was shared + // with a writer that announces it, whose event already carries these + // invalidations, so `passes` goes along and this broadcast stands down. + // + // Without such a pass the request changed nothing anyone else could be + // holding, so it is not announced to every subscriber of the realm: a + // public realm has a session room for every user, and a bulk writer + // re-sending unchanged content would cost each of them an event per + // request. The one party that can still be waiting on it is the writer, + // whose client may key a wait on the event carrying its request id. So + // a request that named an id hears it back in its writer's own session + // room, and one that named none hears nothing. A writer the realm cannot + // identify gets the event as every subscriber does, as the only way to + // reach it. // // The versions are still reported, and they are still the file's own: a // write that found the bytes it staged already there left the file @@ -5767,21 +5798,28 @@ export class Realm { // whatever a mid-loop flush put there, and usually nothing at all. That // is why the member is documented as describing what the request wrote // rather than as a companion to the list beside it. - // - // A module flush that ran ahead of these byte-identical instances is the - // only pass behind this broadcast, and when it was shared with a writer - // that announces it, that writer's event already carries this one's - // invalidations — so `passes` goes along, and this broadcast stands down. - this.broadcastIncrementalInvalidationEvent([...invalidations], { - clientRequestId, - ...(clientAuthored ? { clientAuthored } : {}), - versions, - generation: indexGeneration, - invalidatedTypes: invalidatedTypes.value, - passes, - ownChanges, - broadcastRealmEvents, - }); + let passOwed = + invalidations.size > 0 || + passes.some((pass) => pass !== undefined && !pass.announcedByPeer); + if (passOwed || clientRequestId) { + this.broadcastIncrementalInvalidationEvent([...invalidations], { + clientRequestId, + ...(clientAuthored ? { clientAuthored } : {}), + versions, + generation: indexGeneration, + invalidatedTypes: invalidatedTypes.value, + passes, + ownChanges, + broadcastRealmEvents, + ...(!passOwed && initiatingUser + ? { recipients: [initiatingUser] } + : {}), + }); + } else { + this.#log.debug( + `not broadcasting index event for ${this.url}: the write changed nothing and named no request id`, + ); + } } // A mixed batch whose instances all turned out to be byte-identical // leaves the flush's deferred set with no later pass to fold it into. @@ -17612,12 +17650,16 @@ export class Realm { }); } - private async broadcastRealmEvent(event: RealmEventContent): Promise { + private async broadcastRealmEvent( + event: RealmEventContent, + delivery?: RealmEventDelivery, + ): Promise { this.#adapter.broadcastRealmEvent( event, this.url, this.#matrixClient, this.#dbAdapter, + delivery, ); } From 39a8edf6691a3296e9f60bd78d682c1f96cbc159 Mon Sep 17 00:00:00 2001 From: Hassan Abdel-Rahman Date: Wed, 21 Oct 2026 19:48:38 -0400 Subject: [PATCH 2/2] Name the anonymous writer the no-op event's fallback reaches The fallback that sends a no-op write's event to every subscriber exists for a request admitted without a user: an anonymous write to a world-writable realm has no session room to narrow to. Say so, and describe default delivery as what fetchRealmSessionRooms returns rather than as every user who can read the realm. Co-Authored-By: Claude Opus 5.5 --- packages/runtime-common/realm.ts | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/packages/runtime-common/realm.ts b/packages/runtime-common/realm.ts index caadfee3ff0..cfbd3bff949 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -2337,10 +2337,10 @@ export interface WriteOptions { onDurable?: () => void; } -// Who a realm event is delivered to. By default it goes to the session room -// of every user who can read the realm. `recipients` narrows that to the named -// matrix users' session rooms, for an event that concerns only them; a -// recipient with no session room for the realm receives nothing. +// Who a realm event is delivered to. By default it goes to every session room +// subscribed to the realm (see `fetchRealmSessionRooms`). `recipients` narrows +// that to the named matrix users' session rooms, for an event that concerns +// only them; a recipient with no subscribed session room receives nothing. export interface RealmEventDelivery { recipients?: string[]; } @@ -5785,9 +5785,10 @@ export class Realm { // request. The one party that can still be waiting on it is the writer, // whose client may key a wait on the event carrying its request id. So // a request that named an id hears it back in its writer's own session - // room, and one that named none hears nothing. A writer the realm cannot - // identify gets the event as every subscriber does, as the only way to - // reach it. + // room, and one that named none hears nothing. A request admitted + // without a user — an anonymous write to a world-writable realm — names + // no room to narrow to, so its event goes to every subscriber as the + // only way to reach a writer that may be waiting on it. // // The versions are still reported, and they are still the file's own: a // write that found the bytes it staged already there left the file