diff --git a/packages/boxel-cli/plugin/skills/realm-sync/SKILL.md b/packages/boxel-cli/plugin/skills/realm-sync/SKILL.md index df3305695ac..8aed0ae7b20 100644 --- a/packages/boxel-cli/plugin/skills/realm-sync/SKILL.md +++ b/packages/boxel-cli/plugin/skills/realm-sync/SKILL.md @@ -45,6 +45,22 @@ A profile must be active. If `boxel profile list` shows none, the user has to ru `watch` protects local edits without a flag: by default any file whose local hash differs from the sync manifest is skipped (with a yellow `⚠ skipped …` line) instead of overwritten. The warning re-fires on every poll until the user reconciles via `boxel realm sync …` (e.g. `--prefer-newest`) or reruns watch with `--overwrite-local` to accept the remote. +## Pushing without realm events (`--no-realm-events`) + +Every write to a realm normally broadcasts realm events to each session room subscribed to it. Open tabs use them to show that the realm is indexing, then reload the changed cards, modules and live searches. `push --no-realm-events` marks each write it sends so the realm server announces none of them. The files are still written and indexed, and the server's caches are still invalidated. Open tabs just aren't told, so they pick the new content up on their next page load. + +**Use it** when a push replaces a realm's content out of band, for a realm people mostly read rather than edit: + +- A deploy that pushes a git tree to a shared realm, such as the catalog. Without the flag, every open tab on that realm reloads code its user didn't ask for, mid-session, and the realm server fans the events out to every subscribed room. +- Seeding or restoring a realm in bulk when nobody needs to see the change live. + +**Don't use it** when someone needs to see the push as it lands: + +- Pushing your own edits to a workspace you, a collaborator or an agent has open. Those tabs keep showing the old cards and code until reload. Card saves don't check for a newer version, so an open tab's next save of a stale card writes the old content back over the push. +- Small, everyday pushes. Their events cost little, and live refresh is what users expect. + +When unsure, leave it off. Realm servers that predate the flag ignore it and announce as usual. + ## Realm skills reach Claude Code automatically A workspace's user-authored skills live in the realm as `skills//SKILL.md` (see `/boxel-skills:boxel-skill-authoring`). `pull`, `sync`, and `watch` expose them to Claude Code with no extra step: each one is copied into `/.claude/skills/-/`, where `` is the last segment of the realm URL. So a skill named `trip-planner` in `…/alice/experiments/` becomes `/experiments-trip-planner` in the next session. @@ -118,6 +134,7 @@ Push local files to a Boxel realm - `--dry-run` — Show what would be done without making changes - `--force` — Upload all files, even if unchanged - `--changed-since ` — Upload only files that differ from this git revision (one the realm already holds), plus files git does not track and files missing from the realm, instead of comparing against the sync manifest. Repeat it to name every revision the realm may hold part of, such as one a failed sync stopped partway through; a file is skipped only when it matches all of them. A file edited directly on the realm stays as it is until it changes in git. The local directory must be a git checkout whose history reaches each revision +- `--no-realm-events` — Index the pushed changes without announcing them to open tabs, which pick them up on their next load (for deploys) - `--realm-secret-seed` — Administrative auth: prompt for a realm secret seed and mint a JWT locally instead of using a Matrix profile (env: BOXEL_REALM_SECRET_SEED) ### `boxel realm pull ` @@ -179,5 +196,6 @@ List realms accessible to the active profile ## Pitfalls - `boxel realm push --delete` will remove remote files that don't exist locally. Confirm the user intends destructive sync before adding the flag. +- Don't add `--no-realm-events` to a push into a realm someone has open. Their tabs go stale, and a stale tab's next save can overwrite the push. See [Pushing without realm events](#pushing-without-realm-events---no-realm-events). - The `` argument is the realm's _base URL_ (`https://app.boxel.ai/owner/realm/`), not a card URL inside the realm. If the user pastes a card URL, strip the path back to the realm root. - Newly created realms need indexing before they're usable. After `boxel realm create`, follow up with `boxel realm wait-for-ready` (see `/boxel-cli:realm-history`). diff --git a/packages/boxel-cli/src/commands/realm/push.ts b/packages/boxel-cli/src/commands/realm/push.ts index 9014c4c869f..404aebb8fc8 100644 --- a/packages/boxel-cli/src/commands/realm/push.ts +++ b/packages/boxel-cli/src/commands/realm/push.ts @@ -425,6 +425,11 @@ export interface PushCommandOptions { delete?: boolean; dryRun?: boolean; force?: boolean; + /** + * False to push without the realm announcing the changes to its + * subscribers (see `SyncOptions.realmEvents`). + */ + realmEvents?: boolean; atomicBatchSize?: number; /** * Git revisions the realm's files may hold. Only files that differ from any @@ -464,6 +469,10 @@ export function registerPushCommand(realm: Command): void { 'Upload only files that differ from this git revision (one the realm already holds), plus files git does not track and files missing from the realm, instead of comparing against the sync manifest. Repeat it to name every revision the realm may hold part of, such as one a failed sync stopped partway through; a file is skipped only when it matches all of them. A file edited directly on the realm stays as it is until it changes in git. The local directory must be a git checkout whose history reaches each revision', (revision: string, previous: string[] = []) => [...previous, revision], ) + .option( + '--no-realm-events', + 'Index the pushed changes without announcing them to open tabs, which pick them up on their next load (for deploys)', + ) .option( '--realm-secret-seed', 'Administrative auth: prompt for a realm secret seed and mint a JWT locally instead of using a Matrix profile (env: BOXEL_REALM_SECRET_SEED)', @@ -476,6 +485,7 @@ export function registerPushCommand(realm: Command): void { delete?: boolean; dryRun?: boolean; force?: boolean; + realmEvents?: boolean; changedSince?: string[]; realmSecretSeed?: boolean; }, @@ -487,6 +497,7 @@ export function registerPushCommand(realm: Command): void { delete: options.delete, dryRun: options.dryRun, force: options.force, + realmEvents: options.realmEvents, changedSince: options.changedSince, realmSecretSeed, }); @@ -542,6 +553,7 @@ export async function pushCommand( deleteRemote: options.delete, dryRun: options.dryRun, force: options.force, + realmEvents: options.realmEvents, atomicBatchSize: options.atomicBatchSize, changedSince, }, diff --git a/packages/boxel-cli/src/lib/realm-sync-base.ts b/packages/boxel-cli/src/lib/realm-sync-base.ts index 31e0f8c98c1..1740b89d835 100644 --- a/packages/boxel-cli/src/lib/realm-sync-base.ts +++ b/packages/boxel-cli/src/lib/realm-sync-base.ts @@ -97,6 +97,14 @@ export interface SyncOptions { * default. */ waitForIndex?: boolean; + /** + * When false, every write this sync sends carries + * `X-Boxel-Realm-Events: none`, so the realm indexes the changes without + * announcing them to its subscribers. Open tabs pick the new content up on + * their next load instead of resetting mid-session. For pushes that replace + * a realm's content wholesale, such as a deploy. On by default. + */ + realmEvents?: boolean; /** Files per /_atomic request. Defaults to ATOMIC_BATCH_SIZE. */ atomicBatchSize?: number; } @@ -158,6 +166,16 @@ export abstract class RealmSyncBase { return `${this.normalizedRealmUrl}${cleanDir}/`; } + // Headers for a request that writes or removes a file, with the realm-event + // opt-out added when this sync asked for it. + protected writeHeaders( + headers: Record, + ): Record { + return this.options.realmEvents === false + ? { ...headers, 'X-Boxel-Realm-Events': 'none' } + : headers; + } + protected buildFileUrl(relativePath: string): string { const cleanPath = relativePath.replace(/^\/+/, ''); return `${this.normalizedRealmUrl}${cleanPath}`; @@ -451,10 +469,10 @@ export abstract class RealmSyncBase { const response = await this.authenticator.authedRealmFetch(url, { method: 'POST', - headers: { + headers: this.writeHeaders({ 'Content-Type': 'text/plain;charset=UTF-8', Accept: SupportedMimeType.CardSource, - }, + }), body: content, }); @@ -481,9 +499,9 @@ export abstract class RealmSyncBase { const response = await this.authenticator.authedRealmFetch(url, { method: 'POST', - headers: { + headers: this.writeHeaders({ 'Content-Type': SupportedMimeType.OctetStream, - }, + }), body: bytes, }); @@ -724,10 +742,10 @@ export abstract class RealmSyncBase { try { response = await this.authenticator.authedRealmFetch(url, { method: 'POST', - headers: { + headers: this.writeHeaders({ 'Content-Type': 'application/vnd.api+json', Accept: 'application/vnd.api+json', - }, + }), body: JSON.stringify({ 'atomic:operations': operations }), }); } catch (err) { @@ -874,9 +892,9 @@ export abstract class RealmSyncBase { try { response = await this.authenticator.authedRealmFetch(url, { method: 'DELETE', - headers: { + headers: this.writeHeaders({ Accept: SupportedMimeType.CardSource, - }, + }), signal: AbortSignal.timeout(DELETE_TIMEOUT_MS), }); } catch (error) { diff --git a/packages/boxel-cli/tests/integration/realm-push.test.ts b/packages/boxel-cli/tests/integration/realm-push.test.ts index 5882b1acbc1..2a7a88e724d 100644 --- a/packages/boxel-cli/tests/integration/realm-push.test.ts +++ b/packages/boxel-cli/tests/integration/realm-push.test.ts @@ -698,6 +698,60 @@ describe('realm push (integration)', () => { expect(await remoteFileExists(realmUrl, 'c.gts')).toBe(true); }); + it('--no-realm-events marks every write it sends, and a plain push marks none', async () => { + // In-process so the requests the push sends can be inspected. + let realmUrl = await createTestRealm(); + let localDir = makeLocalDir(); + writeLocalFile(localDir, 'a.gts', 'export const a = 1;\n'); + writeLocalBytes(localDir, 'pixel.png', TINY_PNG_BYTES); + writeLocalFile(localDir, 'stale.gts', 'export const stale = 1;\n'); + + let profileManager = reloadProfile(home); + let realFetch = profileManager.authedRealmFetch.bind(profileManager); + let writes: { method: string; marker: string | null }[] = []; + let fetchSpy = vi + .spyOn(profileManager, 'authedRealmFetch') + .mockImplementation(async (input, init) => { + let method = init?.method ?? 'GET'; + if (method === 'POST' || method === 'DELETE') { + writes.push({ + method, + marker: new Headers(init?.headers).get('X-Boxel-Realm-Events'), + }); + } + return realFetch(input, init); + }); + + try { + await pushCommand(localDir, realmUrl, { profileManager }); + expect(writes.length).toBeGreaterThan(0); + expect(writes.every(({ marker }) => marker === null)).toBe(true); + + writes = []; + fs.rmSync(path.join(localDir, 'stale.gts')); + writeLocalFile(localDir, 'a.gts', 'export const a = 2;\n'); + await pushCommand(localDir, realmUrl, { + profileManager, + delete: true, + force: true, + realmEvents: false, + }); + // The upload batch, the binary upload and the removal of the file + // that no longer exists locally. + expect(writes.map(({ method }) => method).sort()).toEqual([ + 'DELETE', + 'POST', + 'POST', + ]); + expect(writes.every(({ marker }) => marker === 'none')).toBe(true); + } finally { + fetchSpy.mockRestore(); + } + + expect(await fetchRemoteFile(realmUrl, 'a.gts')).toContain('a = 2'); + expect(await remoteFileExists(realmUrl, 'stale.gts')).toBe(false); + }); + it('splits an upload into /_atomic requests, modules first', async () => { // In-process so the /_atomic requests can be read. let realmUrl = await createTestRealm(); diff --git a/packages/realm-server/server.ts b/packages/realm-server/server.ts index 7a4697ed1bc..936f0b67407 100644 --- a/packages/realm-server/server.ts +++ b/packages/realm-server/server.ts @@ -959,7 +959,7 @@ export class RealmServer { // here the preflight fails and no browser client can invoke a card // operation cross-origin at all. allowHeaders: - 'Accept, Authorization, Content-Type, If-Match, If-None-Match, If-Range, Range, X-Requested-With, X-Boxel-Client-Request-Id, X-Boxel-Assume-User, X-HTTP-Method-Override, X-Boxel-Disable-Module-Cache, X-Filename, X-Boxel-During-Prerender, X-Boxel-Skip-Index-Wait, X-Boxel-Consuming-Realm, X-Boxel-Job-Id, X-Boxel-Job-Priority, X-Boxel-Logging-Correlation-Id, X-Boxel-Link-Shape, X-Grafana-Device-Id, X-Grafana-Action', + 'Accept, Authorization, Content-Type, If-Match, If-None-Match, If-Range, Range, X-Requested-With, X-Boxel-Client-Request-Id, X-Boxel-Assume-User, X-HTTP-Method-Override, X-Boxel-Disable-Module-Cache, X-Filename, X-Boxel-During-Prerender, X-Boxel-Skip-Index-Wait, X-Boxel-Realm-Events, X-Boxel-Consuming-Realm, X-Boxel-Job-Id, X-Boxel-Job-Priority, X-Boxel-Logging-Correlation-Id, X-Boxel-Link-Shape, X-Grafana-Device-Id, X-Grafana-Action', // Without an explicit expose list, @koa/cors only emits the // CORS-safelisted response headers (cache-control, content-*, // expires, last-modified, pragma). ETag is not on that list, diff --git a/packages/realm-server/tests/atomic-endpoints-test.ts b/packages/realm-server/tests/atomic-endpoints-test.ts index 1d04e068498..8112daaf63f 100644 --- a/packages/realm-server/tests/atomic-endpoints-test.ts +++ b/packages/realm-server/tests/atomic-endpoints-test.ts @@ -11,6 +11,11 @@ import { readFileAsText, SupportedMimeType, } from '@cardstack/runtime-common'; +import type { + IncrementalIndexEventContent, + IncrementalIndexInitiationContent, + RealmEventContent, +} from '@cardstack/base/matrix-event'; import { setupPermissionedRealmCached, createJWT, @@ -1048,6 +1053,209 @@ module(basename(import.meta.filename), function () { ); }); }); + module('realm events', function (hooks) { + setupPermissionedRealmCached(hooks, { + fixture: 'simple', + permissions: { + '*': ['read', 'write'], + }, + realmURL, + onRealmSetup, + }); + + // Every realm event the realm sends passes through its adapter, so + // wrapping the adapter's broadcast records each one a write produced. + let restoreBroadcast: (() => void) | undefined; + function recordRealmEvents() { + let original = testRealmAdapter.broadcastRealmEvent; + let events: RealmEventContent[] = []; + testRealmAdapter.broadcastRealmEvent = async (event, ...rest) => { + events.push(event); + return await original.call(testRealmAdapter, event, ...rest); + }; + restoreBroadcast = () => { + testRealmAdapter.broadcastRealmEvent = original; + }; + return events; + } + hooks.afterEach(function () { + restoreBroadcast?.(); + restoreBroadcast = undefined; + }); + + function eventKinds(events: RealmEventContent[]) { + return events.map((event) => + 'indexType' in event + ? `${event.eventName}:${event.indexType}` + : event.eventName, + ); + } + + function peopleBatch(names: string[]) { + return { + 'atomic:operations': names.map((name) => ({ + op: 'add', + href: `${name}.json`, + data: { + type: 'card', + attributes: { firstName: name }, + meta: { + adoptsFrom: { module: './person', name: 'Person' }, + }, + }, + })), + }; + } + + async function postAtomic( + doc: unknown, + headers: Record = {}, + ) { + let pending = request + .post('/_atomic?waitForIndex=true') + .set('Accept', SupportedMimeType.JSONAPI) + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'user', ['read', 'write'])}`, + ); + for (let [name, value] of Object.entries(headers)) { + pending = pending.set(name, value); + } + return await pending.send(JSON.stringify(doc)); + } + + test('a multi-file batch sends one initiation, one update and one incremental event', async function (assert) { + let events = recordRealmEvents(); + let response = await postAtomic( + peopleBatch(['batch-a', 'batch-b', 'batch-c']), + ); + assert.strictEqual(response.status, 201); + await testRealm.incrementalIndexing(); + + assert.deepEqual( + eventKinds(events), + [ + 'index:incremental-index-initiation', + 'update', + 'index:incremental', + ], + 'the batch is announced once, not once per file', + ); + let initiation = events[0] as IncrementalIndexInitiationContent; + assert.strictEqual( + initiation.updatedFile, + `${testRealmHref}batch-a.json`, + 'the initiation names the first file the batch changed', + ); + let incremental = events[2] as IncrementalIndexEventContent; + assert.deepEqual( + [...incremental.invalidations].sort(), + [ + `${testRealmHref}batch-a`, + `${testRealmHref}batch-b`, + `${testRealmHref}batch-c`, + ], + 'the completion event names every card the batch wrote', + ); + }); + + test('a batch sent with X-Boxel-Realm-Events: none is indexed without any realm event', async function (assert) { + let events = recordRealmEvents(); + let response = await postAtomic(peopleBatch(['quiet-a', 'quiet-b']), { + 'X-Boxel-Realm-Events': 'none', + }); + assert.strictEqual(response.status, 201); + await testRealm.incrementalIndexing(); + + assert.deepEqual(eventKinds(events), [], 'no realm event was sent'); + let card = await request + .get('/quiet-b') + .set('Accept', SupportedMimeType.CardJson); + assert.strictEqual(card.status, 200, 'the card is indexed'); + assert.strictEqual( + (card.body as LooseSingleCardDocument).data.attributes?.firstName, + 'quiet-b', + ); + + // The same bytes again leave every file as it is, which otherwise + // still sends a completion event. + let repeat = await postAtomic( + { + 'atomic:operations': peopleBatch(['quiet-a'])[ + 'atomic:operations' + ].map((operation) => ({ ...operation, op: 'update' })), + }, + { 'X-Boxel-Realm-Events': 'none' }, + ); + assert.strictEqual(repeat.status, 201); + await testRealm.incrementalIndexing(); + assert.deepEqual( + eventKinds(events), + [], + 'an unchanged batch sends nothing either', + ); + }); + + test('per-file source writes and removals sent with X-Boxel-Realm-Events: none send no realm event', async function (assert) { + let events = recordRealmEvents(); + let auth = `Bearer ${createJWT(testRealm, 'user', ['read', 'write'])}`; + let write = await request + .post('/quiet-notes.txt') + .set('Accept', SupportedMimeType.CardSource) + .set('Content-Type', 'text/plain;charset=UTF-8') + .set('Authorization', auth) + .set('X-Boxel-Realm-Events', 'none') + .send('hello'); + assert.strictEqual(write.status, 204); + let binary = await request + .post('/quiet-bytes.bin') + .set('Content-Type', 'application/octet-stream') + .set('Authorization', auth) + .set('X-Boxel-Realm-Events', 'none') + .send(Buffer.from([1, 2, 3])); + assert.strictEqual(binary.status, 204); + let removal = await request + .delete('/quiet-notes.txt') + .set('Accept', SupportedMimeType.CardSource) + .set('Authorization', auth) + .set('X-Boxel-Realm-Events', 'none'); + assert.strictEqual(removal.status, 204); + await testRealm.incrementalIndexing(); + + assert.deepEqual(eventKinds(events), [], 'no realm event was sent'); + assert.notOk( + await testRealmAdapter.exists('quiet-notes.txt'), + 'the removal landed', + ); + assert.ok( + await testRealmAdapter.exists('quiet-bytes.bin'), + 'the binary write landed', + ); + }); + + test('removing several files at once sends one initiation', async function (assert) { + await postAtomic(peopleBatch(['gone-a', 'gone-b', 'gone-c'])); + await testRealm.incrementalIndexing(); + let events = recordRealmEvents(); + + await testRealm.deleteAll([ + 'gone-a.json', + 'gone-b.json', + 'gone-c.json', + ]); + await testRealm.incrementalIndexing(); + + assert.deepEqual( + eventKinds(events), + [ + 'index:incremental-index-initiation', + 'update', + 'index:incremental', + ], + 'the removal is announced once, not once per file', + ); + }); + }); module('error handling', function (hooks) { setupPermissionedRealmCached(hooks, { fixture: 'simple', diff --git a/packages/realm-server/tests/card-conditional-write-test.ts b/packages/realm-server/tests/card-conditional-write-test.ts index 66c2042989d..a57358c6210 100644 --- a/packages/realm-server/tests/card-conditional-write-test.ts +++ b/packages/realm-server/tests/card-conditional-write-test.ts @@ -102,8 +102,8 @@ module(basename(import.meta.filename), function () { } // Both halves of an incremental index pass. The initiation event is - // broadcast per written file *before* indexing runs, so it is the earliest - // signal that something was staged — a test asserting a write was refused + // broadcast once per commit, as its first file changes and *before* + // indexing runs, so it is the earliest signal that something was staged — a test asserting a write was refused // has to count it, or it only checks that the refusal failed to finish. function indexEvents(messages: MatrixEvent[], indexType: string) { return messages.filter( @@ -408,9 +408,9 @@ module(basename(import.meta.filename), function () { await waitForIncrementalIndexEvent(getMessagesSince, since); let messages = await getMessagesSince(since); - // Both halves are checked: the initiation event is broadcast per - // written file *before* indexing runs, so a refusal that staged - // anything shows up there first. + // Both halves are checked: the initiation event is broadcast once per + // commit, as its first file changes and *before* indexing runs, so a + // refusal that staged anything shows up there first. // // The assertion is about which card the window's events name, not how // many there are: the control's presence proves the window is one an diff --git a/packages/realm-server/tests/card-operations-commit-test.ts b/packages/realm-server/tests/card-operations-commit-test.ts index bc7f8eedb16..de2d33c061c 100644 --- a/packages/realm-server/tests/card-operations-commit-test.ts +++ b/packages/realm-server/tests/card-operations-commit-test.ts @@ -335,6 +335,8 @@ function makeFileSystem(): Record { 'shared-pass-both', 'shared-pass-mixed-card', 'shared-pass-mixed-peer', + 'shared-pass-silent', + 'shared-pass-announcing', 'version-target', 'patch-over-http', 'patch-over-batch', @@ -811,6 +813,105 @@ module(basename(import.meta.filename), function (hooks) { ); }); + test('a pass shared by a write that asked for no realm events and one that did is announced by the one that did', async function (assert) { + let silent = `${testRealmHref}shared-pass-silent`; + let announcing = `${testRealmHref}shared-pass-announcing`; + let jobsBefore = await indexJobIds(); + let { holder, release } = await holdIndexingLane(); + let since = Date.now(); + let writes: Promise[] = []; + try { + // The silent write joins the pending pass first, which is the position + // that makes a caller the pass's announcer. + writes.push( + Promise.resolve( + request + .post('/shared-pass-silent.json') + .set('Accept', 'application/vnd.card+source') + .set('Content-Type', 'text/plain;charset=UTF-8') + .set('X-Boxel-Client-Request-Id', 'source:silent-deploy') + .set('X-Boxel-Realm-Events', 'none') + .send( + JSON.stringify({ + data: { + type: 'card', + attributes: { firstName: 'Silent' }, + meta: { adoptsFrom: PERSON }, + }, + }), + ), + ), + ); + await waitUntil(async () => + (await pendingIndexCallers()).includes('source:silent-deploy'), + ); + writes.push( + Promise.resolve( + request + .patch('/shared-pass-announcing') + .set('Accept', 'application/vnd.card+json') + .set('X-Boxel-Client-Request-Id', 'instance:announcing-tab') + .send({ + data: { + type: 'card', + attributes: { firstName: 'Announced' }, + meta: { adoptsFrom: PERSON }, + }, + }), + ), + ); + await waitUntil(async () => + (await pendingIndexCallers()).includes('instance:announcing-tab'), + ); + } finally { + release.fulfill(); + } + let responses = (await Promise.all(writes)) as { status: number }[]; + await holder.done; + await realm.incrementalIndexing(); + assert.deepEqual( + responses.map((response) => response.status), + [204, 200], + 'both writes are served', + ); + let newJobs = (await indexJobIds()).filter( + (id) => !jobsBefore.includes(id), + ); + assert.strictEqual( + newJobs.length, + 1, + `the two writes share one index pass (got ${newJobs.length})`, + ); + + let eventsForPass = async () => + (await incrementalIndexEventsSince(since)).filter( + (event) => + event.invalidations.includes(silent) || + event.invalidations.includes(announcing), + ); + await waitUntil(async () => (await eventsForPass()).length > 0); + await new Promise((resolve) => setTimeout(resolve, 500)); + let events = await eventsForPass(); + assert.strictEqual( + events.length, + 1, + `the shared pass is announced once (got ${events.length})`, + ); + let [event] = events; + assert.strictEqual( + event.clientRequestId, + 'instance:announcing-tab', + 'the writer that wants realm events announces the pass', + ); + assert.deepEqual( + (event.coalescedWrites ?? []) + .map(({ clientRequestId }) => clientRequestId) + .sort(), + ['instance:announcing-tab', 'source:silent-deploy'], + 'and names the silent write among the writes the pass indexed', + ); + }); + test('a write whose module flush was shared names its own instances in the event it announces', async function (assert) { let instance = `${testRealmHref}shared-pass-mixed-card`; let peerCard = `${testRealmHref}shared-pass-mixed-peer`; diff --git a/packages/realm-server/tests/index-commit-validation-test.ts b/packages/realm-server/tests/index-commit-validation-test.ts index 7b67a7de367..3facb7be149 100644 --- a/packages/realm-server/tests/index-commit-validation-test.ts +++ b/packages/realm-server/tests/index-commit-validation-test.ts @@ -761,6 +761,7 @@ module(basename(import.meta.filename), function (hooks) { function incrementalArgs( coalescedCallers: IncrementalArgs['coalescedCallers'], + realmEvents = true, ): IncrementalArgs { return { realmURL: testRealm, @@ -771,6 +772,7 @@ module(basename(import.meta.filename), function (hooks) { deferPrerenderHtml: false, carriedPrerenderHtmlChanges: [], readsOwnWrite: false, + realmEvents, }; } @@ -798,6 +800,18 @@ module(basename(import.meta.filename), function (hooks) { ); }); + test('which follows writes that asked for no realm events, tells the replicas but not the hosts', async function (assert) { + let { realmEvents, indexUpdates } = await runIncrementalTask( + incrementalArgs([], false), + ); + assert.deepEqual(realmEvents, [], 'hosts hear nothing'); + assert.deepEqual( + indexUpdates, + [{ channel: REALM_INDEX_UPDATED_CHANNEL, payload: testRealm }], + "every replica is still told to drop the realm's index-derived caches", + ); + }); + test('a pass a write waits on leaves the announcement to that write', async function (assert) { let { realmEvents, indexUpdates } = await runIncrementalTask( incrementalArgs([ diff --git a/packages/realm-server/tests/prerender-html-split-test.ts b/packages/realm-server/tests/prerender-html-split-test.ts index 8a75c1bfe7b..9eef31d688d 100644 --- a/packages/realm-server/tests/prerender-html-split-test.ts +++ b/packages/realm-server/tests/prerender-html-split-test.ts @@ -74,6 +74,7 @@ function prerenderHtmlArgs( spawningJobId: 100, coalescedPublishes: null, preWarm: false, + announce: true, ...overrides, }; } @@ -359,6 +360,77 @@ module(basename(import.meta.filename), function () { assert.deepEqual(decision, { type: 'join', jobId: 9 }); }); + test('a merged job announces when either publish does', function (assert) { + for (let [existingAnnounces, incomingAnnounces, merged] of [ + [true, false, true], + [false, true, true], + [false, false, false], + ]) { + let decision = coalesce( + context({ + incoming: spec(prerenderHtmlArgs({ announce: incomingAnnounces })), + candidates: [ + candidate(42, prerenderHtmlArgs({ announce: existingAnnounces })), + ], + }), + ); + if (decision.type !== 'join') { + throw new Error('expected a join decision'); + } + assert.strictEqual( + (decision.update?.args as PrerenderHtmlArgs).announce, + merged, + `existing ${existingAnnounces}, incoming ${incomingAnnounces}`, + ); + } + }); + + test('a job enqueued without the announce flag announces', function (assert) { + let { announce: _announce, ...withoutFlag } = prerenderHtmlArgs(); + let decision = coalesce( + context({ + incoming: spec(prerenderHtmlArgs({ announce: false })), + candidates: [candidate(42, withoutFlag)], + }), + ); + if (decision.type !== 'join') { + throw new Error('expected a join decision'); + } + assert.true((decision.update?.args as PrerenderHtmlArgs).announce); + }); + + test('an announcing publish does not ride on an in-flight job that announces nothing', function (assert) { + let inFlight = prerenderHtmlArgs({ + spawningIndexPasses: passes(100, 200), + announce: false, + }); + let announcing = coalesce( + context({ + incoming: spec( + prerenderHtmlArgs({ spawningIndexPasses: passes(200) }), + ), + inFlightCandidates: [candidate(9, inFlight)], + }), + ); + assert.deepEqual(announcing, { type: 'insert' }); + let silent = coalesce( + context({ + incoming: spec( + prerenderHtmlArgs({ + spawningIndexPasses: passes(200), + announce: false, + }), + ), + inFlightCandidates: [candidate(9, inFlight)], + }), + ); + assert.deepEqual( + silent, + { type: 'join', jobId: 9 }, + 'a publish that announces nothing still rides on it', + ); + }); + test('inserts rather than joining an in-flight job that does not wait on the incoming spawning pass', function (assert) { // The running job may already be past its wait, so it cannot be relied // on to render after a pass it never waited for. diff --git a/packages/realm-server/tests/queue-test.ts b/packages/realm-server/tests/queue-test.ts index 679a1786e61..01d1fb766e9 100644 --- a/packages/realm-server/tests/queue-test.ts +++ b/packages/realm-server/tests/queue-test.ts @@ -89,6 +89,7 @@ module(basename(import.meta.filename), function () { clientRequestId: string | null; priority?: number; initiatedBy?: string[]; + realmEvents?: boolean; }) { let priority = args.priority ?? userInitiatedPriority; return await publisher.publish({ @@ -99,6 +100,9 @@ module(basename(import.meta.filename), function () { args: makeIncrementalArgsWithCallerMetadata( args.args, args.clientRequestId, + args.realmEvents === undefined + ? undefined + : { realmEvents: args.realmEvents }, ), ...(args.initiatedBy ? { initiatedBy: args.initiatedBy } : {}), mapResult: mapIncrementalDoneResult(args.clientRequestId), @@ -1621,6 +1625,79 @@ module(basename(import.meta.filename), function () { } }); + test('incremental dedup: a publish that announces does not attach to a running pass that announces nothing', async function (assert) { + // A running job's args can no longer change, so a publish attaching to + // a pass whose writes asked for no realm events would have the render + // job that pass spawns render its cards without announcing them. + await runner.destroy(); + let realmURL = 'http://example.com/in-flight-silent/'; + let started = new Deferred(); + let release = new Deferred(); + + let worker = new PgQueueRunner({ + adapter, + workerId: 'in-flight-silent-worker', + }); + worker.register( + 'incremental-index', + async (args: { changes: { url: string }[] }) => { + started.fulfill(); + await release.promise; + return { + invalidations: args.changes.map((change) => change.url), + ignoreData: {}, + stats: { + instancesIndexed: 0, + filesIndexed: 0, + instanceErrors: 0, + fileErrors: 0, + totalIndexEntries: 0, + }, + }; + }, + ); + let changeArgs = () => ({ + realmURL, + realmUsername: 'owner', + ignoreData: {}, + changes: [{ url: `${realmURL}a`, operation: 'update' as const }], + }); + + try { + await worker.start(); + let running = await publishIncrementalIndexJob({ + clientRequestId: 'silent-1', + args: changeArgs(), + realmEvents: false, + }); + await started.promise; + + let silent = await publishIncrementalIndexJob({ + clientRequestId: 'silent-2', + args: changeArgs(), + realmEvents: false, + }); + assert.strictEqual( + silent.id, + running.id, + 'a publish that announces nothing attaches to the running pass', + ); + + let announcing = await publishIncrementalIndexJob({ + clientRequestId: 'announcing', + args: changeArgs(), + }); + assert.notStrictEqual( + announcing.id, + running.id, + 'a publish that announces waits for a pass that will announce it', + ); + } finally { + release.fulfill(); + await worker.destroy(); + } + }); + test('incremental does not coalesce onto pending from-scratch in same group', async function (assert) { await runner.destroy(); diff --git a/packages/runtime-common/card-operations/coordinator.ts b/packages/runtime-common/card-operations/coordinator.ts index a7e18a5febd..6b801df7af7 100644 --- a/packages/runtime-common/card-operations/coordinator.ts +++ b/packages/runtime-common/card-operations/coordinator.ts @@ -176,6 +176,8 @@ export interface BatchCore { waitForIndex?: boolean; // See `CommitBatchOptions.indexWaitBudgetMs`. indexWaitBudgetMs?: number; + // See `CommitBatchOptions.broadcastRealmEvents`. + broadcastRealmEvents?: boolean; initiatingUser?: string | null; // Where the commit stamps the stages it owns. The durable write, the // index enqueue, the wait for the worker and the invalidation that @@ -295,6 +297,11 @@ export interface CommitBatchOptions { // drain ahead of staging is what the batch's serialization resolves // against. indexWaitBudgetMs?: number; + // When false, the batch is written and indexed as usual but announces + // nothing to the realm's subscribers. For writers replacing a realm's + // content out of band, whose subscribers pick it up on their next load + // rather than mid-session. Absent, the batch announces as any write does. + broadcastRealmEvents?: boolean; // A caller's own precondition, run inside the write lock and before // anything is staged. It exists because a precondition evaluated before // `commitBatch` is not binding: the lock is acquired here, so a request @@ -2422,6 +2429,9 @@ async function commitStaged( ...(opts.indexWaitBudgetMs === undefined ? {} : { indexWaitBudgetMs: opts.indexWaitBudgetMs }), + ...(opts.broadcastRealmEvents === undefined + ? {} + : { broadcastRealmEvents: opts.broadcastRealmEvents }), // The batch's index job is tagged with the user whose request produced // it, the same as every other write path, so a reader draining its own // writes waits for this job rather than returning ahead of it. diff --git a/packages/runtime-common/jobs/indexing.ts b/packages/runtime-common/jobs/indexing.ts index f4384bc1e45..fd8ee0b4679 100644 --- a/packages/runtime-common/jobs/indexing.ts +++ b/packages/runtime-common/jobs/indexing.ts @@ -768,6 +768,8 @@ export function makeIncrementalArgsWithCallerMetadata( // See CoalescedCaller. clientAuthored?: string[] | null; announcesPass?: boolean; + // See IncrementalArgs.realmEvents. Absent, the pass announces. + realmEvents?: boolean; }, ): IncrementalArgs { let waiterId = uuidv4(); @@ -789,6 +791,7 @@ export function makeIncrementalArgsWithCallerMetadata( deferPrerenderHtml: args.deferPrerenderHtml === true, carriedPrerenderHtmlChanges: args.carriedPrerenderHtmlChanges ?? [], readsOwnWrite: args.readsOwnWrite === true, + realmEvents: caller?.realmEvents !== false, }; } diff --git a/packages/runtime-common/jobs/prerender-html.ts b/packages/runtime-common/jobs/prerender-html.ts index 5e4170f8580..b741f23ed6f 100644 --- a/packages/runtime-common/jobs/prerender-html.ts +++ b/packages/runtime-common/jobs/prerender-html.ts @@ -118,6 +118,8 @@ export interface PrerenderHtmlEnqueueArgs { // module pre-warm sweep — O(realm module count) — runs at the start of the // job only when set; incremental spawns leave it false. preWarm: boolean; + // See PrerenderHtmlArgs. Absent, the job announces. + announce?: boolean; // The lane the job runs in: the spawning writer's lane of the realm's // prerender-html family (`prerenderHtmlLaneFollowing` / // `prerenderHtmlWriterLane`). Absent, the job is the family's exclusive work, @@ -349,6 +351,7 @@ export async function enqueuePrerenderHtmlJob( spawningPriority, timeoutSec, preWarm, + announce, awaitedByPublish, lane, }: PrerenderHtmlEnqueueArgs, @@ -363,6 +366,7 @@ export async function enqueuePrerenderHtmlJob( spawningJobId, coalescedPublishes: null, preWarm, + announce: announce !== false, }; return await queuePublisher.publish({ jobType: 'prerender_html', diff --git a/packages/runtime-common/realm-index-updater.ts b/packages/runtime-common/realm-index-updater.ts index 92834af214f..969cf623e52 100644 --- a/packages/runtime-common/realm-index-updater.ts +++ b/packages/runtime-common/realm-index-updater.ts @@ -86,6 +86,10 @@ export interface IncrementalIndexOptions { // Whether the caller announces the pass the moment it lands, rather than // after later passes of the same write. See CoalescedCaller. announcesPass?: boolean; + // False when the write asked for no realm events, so neither the pass's + // render job nor a follow-up pass announces anything on its behalf. See + // `IncrementalArgs.realmEvents`. + realmEvents?: boolean; // Matrix user whose HTTP write produced this job, when known. Scopes // the read endpoints' read-your-writes drain — see // #incrementalIndexingDeferreds. @@ -626,6 +630,7 @@ export class RealmIndexUpdater { { clientAuthored: opts?.clientAuthored ?? null, announcesPass: opts?.announcesPass === true, + realmEvents: opts?.realmEvents !== false, }, ); job = await this.#queue.publish({ @@ -742,6 +747,7 @@ export class RealmIndexUpdater { | 'clientRequestId' | 'clientAuthored' | 'announcesPass' + | 'realmEvents' | 'initiatedBy' | 'deferPrerenderHtml' | 'carriedPrerenderHtmlChanges' diff --git a/packages/runtime-common/realm.ts b/packages/runtime-common/realm.ts index 9d95ca35619..6d5bedca5d7 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -913,6 +913,26 @@ function isSkipIndexWaitRequest(request: Request): boolean { return (request.headers.get(SKIP_INDEX_WAIT_HEADER) ?? '').length > 0; } +// A write sent with `X-Boxel-Realm-Events: none` lands and is indexed like any +// other, but announces nothing to the realm's subscribers: no +// `incremental-index-initiation`, no file `update`, no `incremental` index +// event, and no `prerender_html` event from the render job its pass spawns +// (see `IncrementalArgs.realmEvents`). A pass it shares with writes that do +// announce is announced on their behalf, naming this write among them. It is for writers that replace a realm's content wholesale out of +// band — a deploy pushing a git tree — where every open tab reacting to the +// push would reset its loader mid-session for code its user did not ask for. +// Those tabs pick the new content up on their next load instead. +// +// Server-side state is unaffected: byte caches, definition caches and peer +// replicas are invalidated through their own channels, and the index is as +// current as for any write. Only the Matrix broadcast is withheld. +export const REALM_EVENTS_HEADER = 'x-boxel-realm-events'; +function realmEventsSuppressed(request: Request): boolean { + return ( + request.headers.get(REALM_EVENTS_HEADER)?.trim().toLowerCase() === 'none' + ); +} + export interface FileRef { path: LocalPath; content: ReadableStream | Readable | Uint8Array | string; @@ -2277,6 +2297,12 @@ export interface WriteOptions { // a mixed batch's module flush are what the write's serialization resolves // against, so a write cannot proceed without them. indexWaitBudgetMs?: number; + // When false, the commit broadcasts none of its realm events — the + // initiation, the file `update` and the `incremental` index event — while + // writing and indexing exactly as it otherwise would. An index pass it + // shares with writers that do announce is still announced, by one of them. + // Defaults to true. See `REALM_EVENTS_HEADER`. + broadcastRealmEvents?: boolean; // The matrix user whose request produced this write — the effective user // (post assume-user) that `checkPermission` records on the request context // as `authenticatedUser`. Tags the resulting incremental index job so read @@ -3931,6 +3957,10 @@ export class Realm { // unless the pass is only a step toward a later one the same write // runs; see `CoalescedCaller.announcesPass`. announcesPass?: boolean; + // False for a caller that asked for no realm events (see + // `WriteOptions.broadcastRealmEvents`): it announces neither this pass + // nor anything the pass spawns. + broadcastRealmEvents?: boolean; // See `IncrementalIndexOptions.modulesWithoutSchemaChange`. modulesWithoutSchemaChange?: string[]; }, @@ -3952,7 +3982,9 @@ export class Realm { await this.#realmIndexUpdater.updateChanges(changes, { clientRequestId: opts?.clientRequestId ?? null, ...(opts?.clientAuthored ? { clientAuthored: opts.clientAuthored } : {}), - announcesPass: opts?.announcesPass !== false, + announcesPass: + opts?.announcesPass !== false && opts?.broadcastRealmEvents !== false, + realmEvents: opts?.broadcastRealmEvents !== false, initiatedBy: opts?.initiatedBy ?? null, ...(opts?.deferPrerenderHtml ? { deferPrerenderHtml: true } : {}), ...(opts?.carriedPrerenderHtmlChanges?.length @@ -4018,9 +4050,14 @@ export class Realm { changes: IndexChange[], opts: { clientRequestId?: string | null; - // See updateIndexAndCollectInvalidations. This form always announces - // the pass as it lands, from `onSettled`. + // See updateIndexAndCollectInvalidations. This form announces the pass + // as it lands, from `onSettled`, unless the caller asked for no realm + // events. clientAuthored?: string[]; + // False for a caller that asked for no realm events (see + // `WriteOptions.broadcastRealmEvents`): it neither announces the pass + // nor broadcasts the failure of one. + broadcastRealmEvents?: boolean; initiatedBy?: string | null; // Invalidation sets earlier passes of this same write deferred, folded // into the prerender_html job this pass spawns. @@ -4047,7 +4084,8 @@ export class Realm { let { settled } = await this.#realmIndexUpdater.enqueueChanges(changes, { clientRequestId: opts?.clientRequestId ?? null, ...(opts?.clientAuthored ? { clientAuthored: opts.clientAuthored } : {}), - announcesPass: true, + announcesPass: opts.broadcastRealmEvents !== false, + realmEvents: opts.broadcastRealmEvents !== false, initiatedBy: opts?.initiatedBy ?? null, ...(opts?.carriedPrerenderHtmlChanges?.length ? { carriedPrerenderHtmlChanges: opts.carriedPrerenderHtmlChanges } @@ -4085,7 +4123,10 @@ export class Realm { await this.clearRealmIndexCachesAndBroadcast(); this.broadcastIncrementalInvalidationEvent( changes.map(({ url }) => url.href.replace(/\.json$/, '')), - { clientRequestId: opts?.clientRequestId ?? null }, + { + clientRequestId: opts?.clientRequestId ?? null, + broadcastRealmEvents: opts.broadcastRealmEvents, + }, ); }, }); @@ -4110,6 +4151,9 @@ export class Realm { // `coalescedWrites` beside the others' when only some of its passes were // shared — the rest name it nowhere else. ownChanges?: string[]; + // False for a writer that asked not to announce its write (see + // `WriteOptions.broadcastRealmEvents`). + broadcastRealmEvents?: boolean; }, ): void { // A pass indexed alongside other publishes is announced once, by the @@ -4118,6 +4162,18 @@ export class Realm { // announced that way: one pass nobody else speaks for and the event is // still owed, carrying the others' invalidations again along with it. let passes = opts?.passes ?? []; + // A writer that asked for no realm events joins any pass it shares as a + // caller that does not announce it (`CoalescedCaller.announcesPass`), so + // it is never the one the others stand down for: a pass it shares with a + // writer that does announce is announced by that writer, naming this one + // among its coalesced writes, and a pass only such writers share is + // announced by nobody. + if (opts?.broadcastRealmEvents === false) { + this.#log.debug( + `not broadcasting index event for ${this.url} (request ${opts?.clientRequestId ?? 'none'}): the writer asked for no realm events`, + ); + return; + } if ( passes.length > 0 && passes.every((pass) => pass?.announcedByPeer === true) @@ -5054,6 +5110,21 @@ export class Realm { // nothing — so the two stay distinguishable all the way to the event. let clientAuthored: string[] | undefined = options?.clientAuthored; let initiatingUser: string | null = options?.initiatingUser ?? null; + let broadcastRealmEvents = options?.broadcastRealmEvents !== false; + // One initiation for the whole commit, sent as the first file actually + // changes, however many files the batch writes, appends to or removes. A + // subscriber learns only that the realm is indexing from it — the + // completion event says what changed — so one per file would cost every + // subscribed room a send per file and tell it nothing more. A commit that + // changes nothing sends none. + let initiationSent = false; + let announceIndexing = (url: URL) => { + if (initiationSent || !broadcastRealmEvents) { + return; + } + initiationSent = true; + this.sendIndexInitiationEvent(url.href); + }; // The module→instance flush below runs an index pass whose invalidation // set is every dependent of the modules written so far — including the // instances this very batch is about to write, whose on-disk content at @@ -5089,6 +5160,7 @@ export class Realm { // A deferring pass is the module flush ahead of this write's own // pass, which is the one it announces after. announcesPass: !opts?.deferPrerenderHtml, + broadcastRealmEvents, initiatedBy: initiatingUser, // The write answers from the index once this lands, so it cannot be // settled by a pass that read the file before these bytes did. @@ -5141,6 +5213,7 @@ export class Realm { // to the index ahead of it — nothing about it is serialized — nor // stands in for the instance that does. `lastWriteType` is therefore // left as the last write that did serialize. + announceIndexing(url); let written = await this.#writeSplicedUnlocked(path, content); results.push(written.result); fileMetaRows.push(written.row); @@ -5288,7 +5361,7 @@ export class Realm { ) { modulesWithoutSchemaChange.add(url.href); } - this.sendIndexInitiationEvent(url.href); + announceIndexing(url); await this.trackOwnWrite(path); let { lastModified } = await this.#adapter.write(path, content); (isNewFile ? addedFiles : updatedFiles).push(path); @@ -5389,9 +5462,10 @@ export class Realm { } // The append leg. Each path gets the same per-file treatment a write does - // — the ceiling, the initiation event, the own-write tracking, the - // byte-cache drop, the peer notification, and a `realm_file_meta` row — - // with two differences, both of which follow from never holding the file. + // — the ceiling, the commit's initiation if none has gone out yet, the + // own-write tracking, the byte-cache drop, the peer notification, and a + // `realm_file_meta` row — with two differences, both of which follow from + // never holding the file. // // The ceiling is applied to what is being added rather than to what the // file will hold, and it is the file ceiling whatever the path's @@ -5411,7 +5485,7 @@ export class Realm { let url = this.paths.fileURL(path); this.assertWriteSize(content, 'file', path); let existed = await this.#adapter.exists(path); - this.sendIndexInitiationEvent(url.href); + announceIndexing(url); await this.trackOwnWrite(path); let { lastModified, size } = await this.#adapter.append(path, content); this.invalidateCache(path); @@ -5457,14 +5531,14 @@ export class Realm { } // The removal leg. Each path gets the same per-file treatment a write - // does — the initiation event, the own-write tracking that stops the file - // watcher from re-reporting this replica's own change, the byte-cache - // drop, and the peer notification — and then the whole batch's file - // changes are announced together below. + // does — the commit's initiation if none has gone out yet, the own-write + // tracking that stops the file watcher from re-reporting this replica's + // own change, the byte-cache drop, and the peer notification — and then + // the whole batch's file changes are announced together below. let deleteURLs: URL[] = []; for (let path of deletes) { let url = this.paths.fileURL(path); - this.sendIndexInitiationEvent(url.href); + announceIndexing(url); await this.trackOwnWrite(path, { isDelete: true }); await this.#adapter.remove(path); this.invalidateCache(path); @@ -5485,6 +5559,13 @@ export class Realm { ) { this.invalidateCachedRealmInfo(); } + } + if ( + broadcastRealmEvents && + (addedFiles.length > 0 || + updatedFiles.length > 0 || + removedFiles.length > 0) + ) { this.broadcastRealmEvent({ eventName: 'update', ...(addedFiles.length ? { added: addedFiles } : {}), @@ -5559,6 +5640,7 @@ export class Realm { invalidatedTypes: invalidatedTypes.value, passes, ownChanges, + broadcastRealmEvents, }); // Announcing what the pass invalidated is the last of the // invalidation, so it accumulates into the same stage as the hooks @@ -5581,6 +5663,7 @@ export class Realm { this.enqueueDeferredPrerenderHtml( deferredPrerenderHtml, initiatingUser, + broadcastRealmEvents, ), ); answeredAhead = true; @@ -5619,6 +5702,7 @@ export class Realm { { clientRequestId, ...(clientAuthored ? { clientAuthored } : {}), + broadcastRealmEvents, initiatedBy: initiatingUser, ...(carried.length ? { carriedPrerenderHtmlChanges: carried } : {}), modulesWithoutSchemaChange: [...modulesWithoutSchemaChange], @@ -5647,6 +5731,7 @@ export class Realm { invalidatedTypes: types.value, passes: [...priorPasses, meta.sharedPass], ownChanges, + broadcastRealmEvents, }, ); }, @@ -5695,6 +5780,7 @@ export class Realm { invalidatedTypes: invalidatedTypes.value, passes, ownChanges, + broadcastRealmEvents, }); } // A mixed batch whose instances all turned out to be byte-identical @@ -5706,6 +5792,7 @@ export class Realm { await this.enqueueDeferredPrerenderHtml( deferredPrerenderHtml, initiatingUser, + broadcastRealmEvents, ); } return { @@ -5731,6 +5818,8 @@ export class Realm { private async enqueueDeferredPrerenderHtml( deferred: DeferredPrerenderHtml | undefined, initiatingUser: string | null, + // False when the write asked for no realm events. + announce = true, ): Promise { if (!deferred || deferred.changes.length === 0) { return; @@ -5754,6 +5843,7 @@ export class Realm { }), timeoutSec: INCREMENTAL_INDEX_JOB_TIMEOUT_SEC, preWarm: false, + announce, lane: prerenderHtmlWriterLane(this.url, initiatingUser), }); } catch (e: any) { @@ -6125,6 +6215,7 @@ export class Realm { serializeFile: true, waitForIndex, initiatingUser: requestContext.authenticatedUser ?? null, + broadcastRealmEvents: !realmEventsSuppressed(request), }); } catch (e: any) { if (e instanceof CardError) { @@ -7159,12 +7250,11 @@ export class Realm { // Write a file whose content is described as an edit of its own bytes. // // Everything the byte path does per file it does too, in the same order — - // the ceiling, the initiation event, the own-write tracking, the byte-cache - // drop and the peer notification — with two differences, both of which - // follow from never holding the content. The unchanged-bytes short circuit - // is gone: telling whether the result matches what is on disk would mean - // reading what is on disk, and an edit that adds items never matches it - // anyway. And the fingerprint is assembled from bounded reads of the file + // the ceiling, the own-write tracking, the byte-cache drop and the peer + // notification — with two differences, both of which follow from never + // holding the content. The unchanged-bytes short circuit is gone: telling + // whether the result matches what is on disk would mean reading what is on + // disk, and an edit that adds items never matches it anyway. And the fingerprint is assembled from bounded reads of the file // once it is written, rather than computed from bytes in hand; it is the // same value either way, which is what lets a version computed here validate // against one computed the other way. @@ -7179,8 +7269,6 @@ export class Realm { row: { path: LocalPath; contentHash?: string; contentSize?: number }; }> { this.assertWriteSize(content, 'card', path); - let url = this.paths.fileURL(path); - this.sendIndexInitiationEvent(url.href); await this.trackOwnWrite(path); // The content is assembled beside the file before it is renamed onto it, // so a watching realm sees that file appear and vanish. Both are this @@ -7257,7 +7345,12 @@ export class Realm { async delete( path: LocalPath, - options?: { waitForIndex?: boolean; initiatingUser?: string | null }, + options?: { + waitForIndex?: boolean; + initiatingUser?: string | null; + // See `WriteOptions.broadcastRealmEvents`. + broadcastRealmEvents?: boolean; + }, ): Promise { await this.#dbAdapter.withFileWriteLocks(this.url, [path], () => this._deleteUnlocked(path, options), @@ -7266,17 +7359,26 @@ export class Realm { private async _deleteUnlocked( path: LocalPath, - options?: { waitForIndex?: boolean; initiatingUser?: string | null }, + options?: { + waitForIndex?: boolean; + initiatingUser?: string | null; + broadcastRealmEvents?: boolean; + }, ): Promise { let url = this.paths.fileURL(path); - this.sendIndexInitiationEvent(url.href); + let broadcastRealmEvents = options?.broadcastRealmEvents !== false; + if (broadcastRealmEvents) { + this.sendIndexInitiationEvent(url.href); + } await this.trackOwnWrite(path, { isDelete: true }); await this.#adapter.remove(path); - this.broadcastRealmEvent({ - eventName: 'update', - removed: [path], - realmURL: this.url, - }); + if (broadcastRealmEvents) { + this.broadcastRealmEvent({ + eventName: 'update', + removed: [path], + realmURL: this.url, + }); + } this.invalidateCache(path); await this.#notifyFileChange(path); // Remove file meta for this path @@ -7286,12 +7388,16 @@ export class Realm { let { invalidations, generation, invalidatedTypes, sharedPass } = await this.updateIndexAndCollectInvalidations( [{ url, operation: 'delete' }], - { initiatedBy: options?.initiatingUser ?? null }, + { + initiatedBy: options?.initiatingUser ?? null, + broadcastRealmEvents, + }, ); this.broadcastIncrementalInvalidationEvent(invalidations, { generation, invalidatedTypes, passes: [sharedPass], + broadcastRealmEvents, }); } else { // Mirrors the write() waitForIndex:false path: await the durable @@ -7304,11 +7410,13 @@ export class Realm { [{ url, operation: 'delete' }], { initiatedBy: options?.initiatingUser ?? null, + broadcastRealmEvents, onSettled: (deferredInvalidations, meta) => { this.broadcastIncrementalInvalidationEvent(deferredInvalidations, { generation: meta.generation, invalidatedTypes: meta.invalidatedTypes, passes: [meta.sharedPass], + broadcastRealmEvents, }); }, }, @@ -7339,10 +7447,13 @@ export class Realm { let trackPromises: Promise[] = []; let removePromises: Promise[] = []; + // One initiation for the whole removal, as for a batch commit. + if (paths.length > 0) { + this.sendIndexInitiationEvent(this.paths.fileURL(paths[0]).href); + } for (let path of paths) { let url = this.paths.fileURL(path); urls.push(url); - this.sendIndexInitiationEvent(url.href); trackPromises.push(this.trackOwnWrite(path, { isDelete: true })); removePromises.push(this.#adapter.remove(path)); this.invalidateCache(path); @@ -11598,6 +11709,7 @@ export class Realm { clientRequestId: request.headers.get('X-Boxel-Client-Request-Id'), actor: requestContext.authenticatedUser ?? undefined, waitForIndex: false, + broadcastRealmEvents: !realmEventsSuppressed(request), }, ); } catch (err: unknown) { @@ -11971,6 +12083,7 @@ export class Realm { await this.delete(handle.path, { waitForIndex: false, initiatingUser: requestContext.authenticatedUser ?? null, + broadcastRealmEvents: !realmEventsSuppressed(request), }); return createResponse({ body: null, @@ -16958,6 +17071,7 @@ export class Realm { deferPrerenderHtml: false, carriedPrerenderHtmlChanges: [], readsOwnWrite: false, + realmEvents: true, }; let job = await this.#queue.publish({ jobType: 'incremental-index', diff --git a/packages/runtime-common/tasks/indexer.ts b/packages/runtime-common/tasks/indexer.ts index 974cace0586..64eb49ffa91 100644 --- a/packages/runtime-common/tasks/indexer.ts +++ b/packages/runtime-common/tasks/indexer.ts @@ -110,6 +110,13 @@ export interface IncrementalArgs extends WorkerArgs { // `chooseIncrementalCoalesceDecision`. Non-optional for the same reason // `deferPrerenderHtml` is. readsOwnWrite: boolean; + // Whether anything this pass does is announced to the realm's subscribers + // on behalf of the writes behind it: the `prerender_html` event of the + // render job it spawns, and its own `incremental` event when no publisher + // waits on it (a follow-up pass). False only when every write behind the + // job asked for no realm events, so OR-merged. Read as true when absent, as + // on a job enqueued by a realm server predating the field. + realmEvents: boolean; } // An invalidation set an index pass computed but deliberately did not enqueue @@ -295,6 +302,7 @@ function parseIncrementalArgsForCoalesce( deferPrerenderHtml, carriedPrerenderHtmlChanges, readsOwnWrite, + realmEvents, } = args; if ( typeof realmURL !== 'string' || @@ -319,6 +327,7 @@ function parseIncrementalArgsForCoalesce( ? (carriedPrerenderHtmlChanges as IncrementalChange[]) : [], readsOwnWrite: readsOwnWrite === true, + realmEvents: realmEvents !== false, }; } @@ -398,6 +407,9 @@ function chooseIncrementalCoalesceDecision( // enough to make that true of the merged job. readsOwnWrite: existingArgs.readsOwnWrite || incomingArgs.readsOwnWrite, + // OR, because one write that announces is enough for the pass to + // announce what it does. + realmEvents: existingArgs.realmEvents || incomingArgs.realmEvents, }, }, }; @@ -432,7 +444,12 @@ function chooseIncrementalCoalesceDecision( if (!existingArgs) { continue; } - if (incrementalChangesCover(existingArgs.changes, incomingArgs.changes)) { + if ( + incrementalChangesCover(existingArgs.changes, incomingArgs.changes) && + // A running job's args can no longer change, so a publish that + // announces cannot ride on one whose render job won't. + (existingArgs.realmEvents || !incomingArgs.realmEvents) + ) { return { type: 'join', jobId: candidate.id }; } } @@ -557,10 +574,12 @@ function spawningIndexPassFor( // The args of the follow-up job a pass's commit enqueues when peer commits // keep leaving its rows stale (see `CommitValidation.followUpJobArgs`). No // publisher waits on it, so it carries no callers, and it announces its own -// pass when it lands (see `incrementalIndex`). +// pass when it lands (see `incrementalIndex`) — unless the writes behind the +// pass it follows asked for no realm events, which it inherits. function followUpIncrementalArgs( realmURL: string, realmUsername: string, + realmEvents: boolean, ): ( changes: IncrementalChange[], ignoreData: Record, @@ -574,6 +593,7 @@ function followUpIncrementalArgs( deferPrerenderHtml: false, carriedPrerenderHtmlChanges: [], readsOwnWrite: false, + realmEvents, }); } @@ -683,7 +703,7 @@ const fromScratchIndex: Task = ({ ); }); }, - followUpJobArgs: followUpIncrementalArgs(realmURL, realmUsername), + followUpJobArgs: followUpIncrementalArgs(realmURL, realmUsername, true), auth, fetch: _fetch, prerenderer, @@ -846,6 +866,7 @@ const incrementalIndex: Task = ({ // Incremental: no realm-wide sweep — its cost is O(realm module // count), deliberately not paid on incrementals. preWarm: false, + announce: args.realmEvents !== false, // The writer's own render lane, the one the bulk-write render hold // names, so another writer's render work neither waits behind this // one nor merges into it. @@ -856,7 +877,11 @@ const incrementalIndex: Task = ({ ); }); }, - followUpJobArgs: followUpIncrementalArgs(realmURL, realmUsername), + followUpJobArgs: followUpIncrementalArgs( + realmURL, + realmUsername, + args.realmEvents !== false, + ), auth, fetch: _fetch, prerenderer, @@ -891,13 +916,20 @@ const incrementalIndex: Task = ({ // replica drops the index-derived caches a publisher's realm would have // dropped after its own pass. The event names no types, so every live // query re-runs rather than trusting a set nothing bounded. - reportRealmEvent?.({ - eventName: 'index', - indexType: 'incremental', - invalidations: invalidations.map((href) => href.replace(/\.json$/, '')), - ...(generation !== undefined ? { generation } : {}), - realmURL, - }); + // The replicas drop their caches whatever the writes asked; only the + // announcement to hosts is withheld from a pass following writes that + // asked for no realm events. + if (args.realmEvents !== false) { + reportRealmEvent?.({ + eventName: 'index', + indexType: 'incremental', + invalidations: invalidations.map((href) => + href.replace(/\.json$/, ''), + ), + ...(generation !== undefined ? { generation } : {}), + realmURL, + }); + } await notifyRealmIndexUpdated(dbAdapter, realmURL); } reportStatus(jobInfo, 'finish'); diff --git a/packages/runtime-common/tasks/prerender-html.ts b/packages/runtime-common/tasks/prerender-html.ts index 44dc3621f5d..41a2b43131f 100644 --- a/packages/runtime-common/tasks/prerender-html.ts +++ b/packages/runtime-common/tasks/prerender-html.ts @@ -74,6 +74,12 @@ export interface PrerenderHtmlArgs extends WorkerArgs { // once at the start of the pass. Incremental spawns leave it false — the // sweep is O(realm module count), deliberately not paid on incrementals. preWarm: boolean; + // Whether the job tells the realm's subscribers its HTML is live (the + // `prerender_html` realm event). False only when every write behind it + // asked for no realm events (see `IncrementalArgs.realmEvents`). OR-merged, + // so one announcing publish makes the merged job announce. Absent from the + // args of a job enqueued by a worker predating the field, which announces. + announce: boolean; } export interface PrerenderHtmlResult extends JSONTypes.Object { @@ -132,6 +138,7 @@ function parsePrerenderHtmlArgsForCoalesce( spawningJobId, coalescedPublishes, preWarm, + announce, } = args; if ( typeof realmURL !== 'string' || @@ -154,6 +161,7 @@ function parsePrerenderHtmlArgsForCoalesce( coalescedPublishes: typeof coalescedPublishes === 'number' ? coalescedPublishes : null, preWarm: preWarm === true, + announce: announce !== false, legacy: passes === undefined, }; } @@ -291,6 +299,7 @@ function choosePrerenderHtmlCoalesceDecision( // job still runs the realm-wide sweep. The pass runs the sweep once // at its start regardless of how many publishes coalesced into it. preWarm: existingArgs.preWarm || incomingArgs.preWarm, + announce: existingArgs.announce || incomingArgs.announce, }, }, }; @@ -309,7 +318,10 @@ function choosePrerenderHtmlCoalesceDecision( if ( waitsOnEverySpawner(existingArgs, incomingArgs) && existingArgs.loaderEpoch === incomingArgs.loaderEpoch && - incrementalChangesCover(existingArgs.changes, incomingArgs.changes) + incrementalChangesCover(existingArgs.changes, incomingArgs.changes) && + // A running job's args can no longer change, so a publish that + // announces cannot ride on one that won't. + (existingArgs.announce || !incomingArgs.announce) ) { return { type: 'join', jobId: candidate.id }; } @@ -444,13 +456,16 @@ const prerenderHtml: Task = ({ // re-run and pick up the new renderings / corrected full-text // membership. Rides the worker→manager→realm-server event bridge. The // generation is the committed one the renders were stamped against, which - // is the floor the host's live search reads at. - reportRealmEvent?.({ - eventName: 'prerender_html', - realmURL, - generation: pass.generation, - invalidations, - }); + // is the floor the host's live search reads at. A job whose writes all + // asked for no realm events renders without saying so. + if (args.announce !== false) { + reportRealmEvent?.({ + eventName: 'prerender_html', + realmURL, + generation: pass.generation, + invalidations, + }); + } reportStatus(jobInfo, 'finish'); log.debug(