diff --git a/packages/realm-server/tests/atomic-endpoints-test.ts b/packages/realm-server/tests/atomic-endpoints-test.ts index 8a36bb1f009..f53007d23a5 100644 --- a/packages/realm-server/tests/atomic-endpoints-test.ts +++ b/packages/realm-server/tests/atomic-endpoints-test.ts @@ -340,8 +340,11 @@ module(basename(import.meta.filename), function () { }, ], }; + // The follow-up GET is credential-less, so it doesn't share this + // authenticated write's read-your-writes principal and won't wait + // on its indexing — make the write itself synchronous instead. let response = await request - .post('/_atomic') + .post('/_atomic?waitForIndex=true') .set('Accept', SupportedMimeType.JSONAPI) .set( 'Authorization', @@ -394,8 +397,11 @@ module(basename(import.meta.filename), function () { ], }; + // Synchronous write for the same reason as the single-instance + // test: the credential-less GETs below don't wait on this + // authenticated write's indexing. let response = await request - .post('/_atomic') + .post('/_atomic?waitForIndex=true') .set('Accept', SupportedMimeType.JSONAPI) .set( 'Authorization', @@ -697,8 +703,10 @@ module(basename(import.meta.filename), function () { }, ], }; + // Synchronous write: the credential-less card+json GET below + // doesn't wait on this authenticated write's indexing. let response = await request - .post('/_atomic') + .post('/_atomic?waitForIndex=true') .set('Accept', SupportedMimeType.JSONAPI) .set( 'Authorization', @@ -775,8 +783,10 @@ module(basename(import.meta.filename), function () { }, ], }; + // Synchronous write: the credential-less card+json GET below + // doesn't wait on this authenticated write's indexing. let response = await request - .post('/_atomic') + .post('/_atomic?waitForIndex=true') .set('Accept', SupportedMimeType.JSONAPI) .set( 'Authorization', diff --git a/packages/realm-server/tests/card-source-endpoints-test.ts b/packages/realm-server/tests/card-source-endpoints-test.ts index 5add92a9509..d377274f569 100644 --- a/packages/realm-server/tests/card-source-endpoints-test.ts +++ b/packages/realm-server/tests/card-source-endpoints-test.ts @@ -935,11 +935,20 @@ module(basename(import.meta.filename), function () { } let id = maybeId; - // modify field + // modify field. The +source POST indexes deferred, and only an + // identified reader waits on their own pending indexing (a + // credential-less read never waits — anonymous writes are + // unsupported), so this write and the read that depends on it + // authenticate as the same user; authorization still comes from + // the realm's public write permission. { let response = await request .post('/test-card.gts') - .set('Accept', 'application/vnd.card+source').send(` + .set('Accept', 'application/vnd.card+source') + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'john', ['read', 'write'])}`, + ).send(` import { contains, field, CardDef } from '@cardstack/base/card-api'; import StringField from '@cardstack/base/string'; @@ -956,7 +965,11 @@ module(basename(import.meta.filename), function () { { let response = await request .get(new URL(id).pathname) - .set('Accept', 'application/vnd.card+json'); + .set('Accept', 'application/vnd.card+json') + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'john', ['read', 'write'])}`, + ); assert.strictEqual(response.status, 200, 'HTTP 200 status'); let json = response.body; diff --git a/packages/realm-server/tests/helpers/index.ts b/packages/realm-server/tests/helpers/index.ts index e1c783a26f2..34a3826f100 100644 --- a/packages/realm-server/tests/helpers/index.ts +++ b/packages/realm-server/tests/helpers/index.ts @@ -1265,6 +1265,7 @@ export async function createRealm({ fullIndexOnStartup, mediaCacheAdapter, screenshotSyncWaitMs, + readIndexDrainBudgetMs, }: { dir: string; definitionLookup: DefinitionLookup; @@ -1304,6 +1305,9 @@ export async function createRealm({ // Shrinks the `_screenshot/` route's on-demand sync-wait budget so tests // can exercise the 503 + Retry-After path without holding real time. screenshotSyncWaitMs?: number; + // Shrinks the card read endpoints' read-your-writes indexing-drain budget + // so tests can exercise the bounded-wait path without holding real time. + readIndexDrainBudgetMs?: number; }): Promise<{ realm: Realm; adapter: RealmAdapter }> { await insertPermissions(dbAdapter, new URL(realmURL), permissions); @@ -1386,6 +1390,9 @@ export async function createRealm({ { ...(fullIndexOnStartup ? { fullIndexOnStartup: true as const } : {}), ...(screenshotSyncWaitMs !== undefined ? { screenshotSyncWaitMs } : {}), + ...(readIndexDrainBudgetMs !== undefined + ? { readIndexDrainBudgetMs } + : {}), }, ); if (worker) { @@ -1435,6 +1442,7 @@ export async function runTestRealmServer({ }, prerenderer: providedPrerenderer, mediaCacheAdapter, + readIndexDrainBudgetMs, }: { testRealmDir: string; realmsRootPath: string; @@ -1458,6 +1466,7 @@ export async function runTestRealmServer({ }; prerenderer?: Prerenderer; mediaCacheAdapter?: MediaCacheAdapter; + readIndexDrainBudgetMs?: number; }) { stripTlsEnvVars(); let prerenderer = providedPrerenderer ?? (await getTestPrerenderer()); @@ -1499,6 +1508,7 @@ export async function runTestRealmServer({ audioSizeLimitBytes, videoSizeLimitBytes, mediaCacheAdapter, + readIndexDrainBudgetMs, }); await testRealm.logInToMatrix(); @@ -2165,6 +2175,7 @@ type InternalPermissionedRealmSetupOptions = { audioSizeLimitBytes?: number; videoSizeLimitBytes?: number; mediaCacheAdapter?: MediaCacheAdapter; + readIndexDrainBudgetMs?: number; }; async function startPermissionedRealmFixture( @@ -2184,6 +2195,7 @@ async function startPermissionedRealmFixture( audioSizeLimitBytes, videoSizeLimitBytes, mediaCacheAdapter, + readIndexDrainBudgetMs, }: InternalPermissionedRealmSetupOptions, ): Promise<{ testRealmServer: Awaited>; @@ -2254,6 +2266,7 @@ async function startPermissionedRealmFixture( videoSizeLimitBytes, prerenderer, mediaCacheAdapter, + readIndexDrainBudgetMs, }); let request = supertest(testRealmServer.testRealmHttpServer); @@ -2323,6 +2336,7 @@ export function setupPermissionedRealm( audioSizeLimitBytes, videoSizeLimitBytes, mediaCacheAdapter, + readIndexDrainBudgetMs, }: { permissions: RealmPermissions; realmURL?: URL; @@ -2352,6 +2366,7 @@ export function setupPermissionedRealm( audioSizeLimitBytes?: number; videoSizeLimitBytes?: number; mediaCacheAdapter?: MediaCacheAdapter; + readIndexDrainBudgetMs?: number; }, ) { let testRealmServer: Awaited>; @@ -2382,6 +2397,7 @@ export function setupPermissionedRealm( audioSizeLimitBytes, videoSizeLimitBytes, mediaCacheAdapter, + readIndexDrainBudgetMs, }); testRealmServer = server; @@ -2436,6 +2452,9 @@ function permissionedRealmTemplateCacheKey( fileSizeLimitBytes: options.fileSizeLimitBytes ?? null, audioSizeLimitBytes: options.audioSizeLimitBytes ?? null, videoSizeLimitBytes: options.videoSizeLimitBytes ?? null, + // `readIndexDrainBudgetMs` is deliberately absent: it tunes request-time + // wait behavior on the live realm and leaves no trace in the template + // database, so keying on it would only fragment the template cache. prerenderer: prerendererCacheKeyPart(options.prerenderer), }); } diff --git a/packages/realm-server/tests/read-index-drain-test.ts b/packages/realm-server/tests/read-index-drain-test.ts new file mode 100644 index 00000000000..35c84565e52 --- /dev/null +++ b/packages/realm-server/tests/read-index-drain-test.ts @@ -0,0 +1,650 @@ +import QUnit from 'qunit'; +const { module, test } = QUnit; +import { basename } from 'path'; +import type { RealmHttpServer as Server } from '../server.ts'; +import type { Realm } from '@cardstack/runtime-common'; +import { + setupPermissionedRealmCached, + closeServer, + createJWT, + withRealmPath, + type RealmRequest, +} from './helpers/index.ts'; +import { resetCatalogRealms } from '../handlers/handle-fetch-catalog-realms.ts'; + +// The card read endpoints (card+json / card+html GET) gate on the +// requester's OWN in-flight incremental indexing — scoped and bounded — via +// Realm.drainRequestersOwnIndexing. These tests pin the gate's three +// behaviors at the HTTP boundary: another user's pending indexing never +// holds a read, the writer's own pending indexing does, and the hold is +// bounded by the realm's readIndexDrainBudgetMs. +// +// The updater's gates are stubbed per test rather than racing real jobs: a +// real incremental job settles as fast as the worker runs it, so "the read +// did not wait" could never be asserted deterministically against one. +// (RealmIndexUpdater's own tagging/filtering is covered in +// realm-index-updater-test.ts.) + +const DRAIN_BUDGET_MS = 4_000; +// A stubbed gate that resolves after this long distinguishes "the read +// waited for the gate" (elapsed >= GATE_RESOLVE_MS) from "the read skipped +// it" (elapsed well under), with margin against a slow cold request. +const GATE_RESOLVE_MS = 500; +// "Did not wait" assertions allow up to this much for the request itself; +// far above a warm request's actual cost, far below the gate/budget. +const NO_WAIT_CEILING_MS = 2_000; + +function resolveAfter(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +const NEVER: Promise = new Promise(() => {}); + +// Shadow the updater's gate methods with instance properties; delete them to +// restore the prototype implementations. +function stubUpdaterGates( + realm: Realm, + stubs: { + initiatedBy?: (user: string) => Promise | undefined; + all?: () => Promise | undefined; + }, +): () => void { + let updater = realm.realmIndexUpdater as any; + if (stubs.initiatedBy) { + updater.incrementalIndexingInitiatedBy = stubs.initiatedBy; + } + if (stubs.all) { + updater.incrementalIndexing = stubs.all; + } + return () => { + delete updater.incrementalIndexingInitiatedBy; + delete updater.incrementalIndexing; + }; +} + +module(basename(import.meta.filename), function () { + module('card read drain | permissioned realm', function (hooks) { + let realmURL = new URL('http://127.0.0.1:4444/test/'); + let testRealm: Realm; + let testRealmHttpServer: Server; + let request: RealmRequest; + + hooks.afterEach(async function () { + await closeServer(testRealmHttpServer); + resetCatalogRealms(); + }); + + setupPermissionedRealmCached(hooks, { + fixture: 'realistic', + realmURL, + permissions: { + hassan: ['read', 'write'], + john: ['read'], + '@node-test_realm:localhost': ['read', 'realm-owner'], + }, + readIndexDrainBudgetMs: DRAIN_BUDGET_MS, + onRealmSetup(args: { + testRealm: Realm; + testRealmHttpServer: Server; + request: any; + }) { + testRealm = args.testRealm; + testRealmHttpServer = args.testRealmHttpServer; + request = withRealmPath(args.request, realmURL); + }, + }); + + async function getPersonAs( + user: string, + permissions: ('read' | 'write')[], + accept = 'application/vnd.card+json', + ) { + return request + .get('/person-1') + .set('Accept', accept) + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, user, permissions)}`, + ); + } + + test("a reader is not held by another user's in-flight indexing", async function (assert) { + // Warm both endpoints so the timed requests below measure the drain, + // not cold module/doc assembly. + let warm = await getPersonAs('john', ['read']); + assert.strictEqual(warm.status, 200, `warm-up GET: ${warm.text}`); + let warmHtml = await getPersonAs( + 'john', + ['read'], + 'application/vnd.card+html', + ); + assert.strictEqual( + warmHtml.status, + 200, + `card+html warm-up GET: ${warmHtml.text}`, + ); + + // hassan has a (never-settling) job in flight; the unscoped gate is + // also held open, so a read that (wrongly) consulted it would hold for + // the full budget. + let restore = stubUpdaterGates(testRealm, { + initiatedBy: (user) => (user === 'hassan' ? NEVER : undefined), + all: () => NEVER, + }); + try { + let startedAt = Date.now(); + let response = await getPersonAs('john', ['read']); + let elapsed = Date.now() - startedAt; + assert.strictEqual(response.status, 200, `HTTP 200: ${response.text}`); + assert.true( + elapsed < NO_WAIT_CEILING_MS, + `read returned without waiting on the writer's job (took ${elapsed}ms)`, + ); + + startedAt = Date.now(); + let htmlResponse = await getPersonAs( + 'john', + ['read'], + 'application/vnd.card+html', + ); + elapsed = Date.now() - startedAt; + assert.strictEqual( + htmlResponse.status, + 200, + `card+html HTTP 200: ${htmlResponse.text}`, + ); + assert.true( + elapsed < NO_WAIT_CEILING_MS, + `card+html read returned without waiting on the writer's job (took ${elapsed}ms)`, + ); + } finally { + restore(); + } + }); + + test("the writer's own read waits for their in-flight indexing", async function (assert) { + let warm = await getPersonAs('hassan', ['read', 'write']); + assert.strictEqual(warm.status, 200, `warm-up GET: ${warm.text}`); + + let gateResolved = false; + let gate = resolveAfter(GATE_RESOLVE_MS).then(() => { + gateResolved = true; + }); + // The unscoped gate must report the job as pending too: a real tagged + // job always appears in both gates, and the drain's fast-path probe + // reads the unscoped one. + let restore = stubUpdaterGates(testRealm, { + initiatedBy: (user) => (user === 'hassan' ? gate : undefined), + all: () => gate, + }); + try { + let startedAt = Date.now(); + let response = await getPersonAs('hassan', ['read', 'write']); + let elapsed = Date.now() - startedAt; + assert.strictEqual(response.status, 200, `HTTP 200: ${response.text}`); + assert.true( + gateResolved, + 'the read did not return before the gate settled', + ); + assert.true( + elapsed >= GATE_RESOLVE_MS - 20, + `read held until the writer's own job settled (took ${elapsed}ms)`, + ); + } finally { + restore(); + } + }); + + test("the writer's read proceeds on the current index generation once the budget expires", async function (assert) { + let warm = await getPersonAs('hassan', ['read', 'write']); + assert.strictEqual(warm.status, 200, `warm-up GET: ${warm.text}`); + + let restore = stubUpdaterGates(testRealm, { + initiatedBy: (user) => (user === 'hassan' ? NEVER : undefined), + all: () => NEVER, + }); + try { + let startedAt = Date.now(); + let response = await getPersonAs('hassan', ['read', 'write']); + let elapsed = Date.now() - startedAt; + assert.strictEqual(response.status, 200, `HTTP 200: ${response.text}`); + assert.true( + elapsed >= DRAIN_BUDGET_MS - 50, + `read held for the drain budget (took ${elapsed}ms)`, + ); + assert.true( + elapsed < DRAIN_BUDGET_MS + NO_WAIT_CEILING_MS, + `read was released by the budget, not the gate (took ${elapsed}ms)`, + ); + } finally { + restore(); + } + }); + + // End-to-end (no stubs): the authenticated write handlers must tag their + // deferred indexing jobs with the writer, or this same-user + // write-then-read sees the pre-write schema. Only an authenticated flow + // can exercise the tagging — a credential-less write produces an + // untagged job and a credential-less read skips the gate entirely + // (anonymous writes are unsupported). + test("an authenticated writer's follow-up read sees their own deferred write indexed", async function (assert) { + let auth = () => + `Bearer ${createJWT(testRealm, 'hassan', ['read', 'write'])}`; + let cardSource = (secondField: string) => ` + import { contains, field, CardDef } from '@cardstack/base/card-api'; + import StringField from '@cardstack/base/string'; + + export class DrainTestCard extends CardDef { + @field field1 = contains(StringField); + @field ${secondField} = contains(StringField); + } + `; + + let createDef = await request + .post('/drain-test-card.gts') + .set('Accept', 'application/vnd.card+source') + .set('Authorization', auth()) + .send(cardSource('field2')); + assert.strictEqual(createDef.status, 204, `HTTP 204: ${createDef.text}`); + + let createInstance = await request + .post('/') + .set('Accept', 'application/vnd.card+json') + .set('Authorization', auth()) + .send({ + data: { + type: 'card', + attributes: { field1: 'a', field2: 'b' }, + meta: { + adoptsFrom: { + module: `${realmURL.href}drain-test-card`, + name: 'DrainTestCard', + }, + }, + }, + }); + assert.strictEqual( + createInstance.status, + 201, + `HTTP 201: ${createInstance.text}`, + ); + let id = createInstance.body.data.id as string; + + // The +source POST answers once the bytes are durable; its indexing is + // the deferred job the follow-up GET must wait out. + let renameField = await request + .post('/drain-test-card.gts') + .set('Accept', 'application/vnd.card+source') + .set('Authorization', auth()) + .send(cardSource('field2a')); + assert.strictEqual( + renameField.status, + 204, + `HTTP 204: ${renameField.text}`, + ); + + let readBack = await request + .get(new URL(id).pathname) + .set('Accept', 'application/vnd.card+json') + .set('Authorization', auth()); + assert.strictEqual(readBack.status, 200, `HTTP 200: ${readBack.text}`); + let attributes = readBack.body.data.attributes; + assert.true( + 'field2a' in attributes, + `post-rename schema is served: ${JSON.stringify(attributes)}`, + ); + assert.false( + 'field2' in attributes, + `pre-rename field is gone: ${JSON.stringify(attributes)}`, + ); + }); + + // Pins the wiring between requestContext.authenticatedUser and the job + // tag for every HTTP write route: a handler that drops its + // `initiatingUser` argument silently disables read-your-writes for that + // route, which no black-box read test can catch deterministically. The + // spy shadows `enqueueUpdate` (the synchronous `update` path calls + // through it, so one spy covers both), delegates to the prototype, and + // captures each job's `initiatedBy`. + test('every HTTP write route tags its indexing job with the writer', async function (assert) { + let auth = () => + `Bearer ${createJWT(testRealm, 'hassan', ['read', 'write'])}`; + let updater = testRealm.realmIndexUpdater as any; + let proto = Object.getPrototypeOf(updater); + let tags = new Map(); + let currentOp = 'none'; + updater.enqueueUpdate = function (urls: URL[], opts?: any) { + tags.set(currentOp, opts?.initiatedBy); + return proto.enqueueUpdate.call(this, urls, opts); + }; + let drainJobs = async () => { + let pending = testRealm.incrementalIndexing(); + if (pending) { + await pending; + } + }; + try { + currentOp = 'source POST'; + let sourcePost = await request + .post('/tag-probe.gts') + .set('Accept', 'application/vnd.card+source') + .set('Authorization', auth()).send(` + import { contains, field, CardDef } from '@cardstack/base/card-api'; + import StringField from '@cardstack/base/string'; + + export class TagProbe extends CardDef { + @field name = contains(StringField); + } + `); + assert.strictEqual( + sourcePost.status, + 204, + `source POST: ${sourcePost.text}`, + ); + await drainJobs(); + + currentOp = 'card POST'; + let cardPost = await request + .post('/') + .set('Accept', 'application/vnd.card+json') + .set('Authorization', auth()) + .send({ + data: { + type: 'card', + attributes: { name: 'probe' }, + meta: { + adoptsFrom: { + module: `${realmURL.href}tag-probe`, + name: 'TagProbe', + }, + }, + }, + }); + assert.strictEqual(cardPost.status, 201, `card POST: ${cardPost.text}`); + let cardPath = new URL(cardPost.body.data.id as string).pathname; + + currentOp = 'card PATCH'; + let cardPatch = await request + .patch(cardPath) + .set('Accept', 'application/vnd.card+json') + .set('Authorization', auth()) + .send({ + data: { + type: 'card', + attributes: { name: 'probe2' }, + meta: { + adoptsFrom: { + module: `${realmURL.href}tag-probe`, + name: 'TagProbe', + }, + }, + }, + }); + assert.strictEqual( + cardPatch.status, + 200, + `card PATCH: ${cardPatch.text}`, + ); + + currentOp = '_invalidate POST'; + let invalidatePost = await request + .post('/_invalidate') + .set('Accept', 'application/vnd.api+json') + .set('Authorization', auth()) + .send({ + data: { attributes: { urls: [`${realmURL.href}tag-probe.gts`] } }, + }); + assert.strictEqual( + invalidatePost.status, + 204, + `_invalidate POST: ${invalidatePost.text}`, + ); + + currentOp = '_atomic POST'; + let atomicPost = await request + .post('/_atomic') + .set('Accept', 'application/vnd.api+json') + .set('Authorization', auth()) + .send({ + 'atomic:operations': [ + { + op: 'add', + href: 'atomic-tag-probe.txt', + data: { + type: 'source', + attributes: { content: 'tag probe payload' }, + meta: {}, + }, + }, + ], + }); + assert.strictEqual( + atomicPost.status, + 201, + `_atomic POST: ${atomicPost.text}`, + ); + await drainJobs(); + + currentOp = 'card DELETE'; + let cardDelete = await request + .delete(cardPath) + .set('Accept', 'application/vnd.card+json') + .set('Authorization', auth()); + assert.strictEqual( + cardDelete.status, + 204, + `card DELETE: ${cardDelete.text}`, + ); + + currentOp = 'source DELETE'; + let sourceDelete = await request + .delete('/tag-probe.gts') + .set('Accept', 'application/vnd.card+source') + .set('Authorization', auth()); + assert.strictEqual( + sourceDelete.status, + 204, + `source DELETE: ${sourceDelete.text}`, + ); + await drainJobs(); + + for (let op of [ + 'source POST', + 'card POST', + 'card PATCH', + '_invalidate POST', + '_atomic POST', + 'card DELETE', + 'source DELETE', + ]) { + assert.strictEqual( + tags.get(op), + 'hassan', + `${op} tagged its indexing job with the writer (got ${JSON.stringify( + tags.get(op), + )})`, + ); + } + } finally { + await drainJobs(); + delete updater.enqueueUpdate; + } + }); + }); + + module('card read drain | public readable realm', function (hooks) { + let realmURL = new URL('http://127.0.0.1:4444/test/'); + let testRealm: Realm; + let testRealmHttpServer: Server; + let request: RealmRequest; + + hooks.afterEach(async function () { + await closeServer(testRealmHttpServer); + resetCatalogRealms(); + }); + + setupPermissionedRealmCached(hooks, { + fixture: 'realistic', + realmURL, + permissions: { + '*': ['read'], + john: ['read'], + '@node-test_realm:localhost': ['read', 'realm-owner'], + }, + readIndexDrainBudgetMs: DRAIN_BUDGET_MS, + onRealmSetup(args: { + testRealm: Realm; + testRealmHttpServer: Server; + request: any; + }) { + testRealm = args.testRealm; + testRealmHttpServer = args.testRealmHttpServer; + request = withRealmPath(args.request, realmURL); + }, + }); + + test('an anonymous reader never waits on pending indexing', async function (assert) { + let warm = await request + .get('/person-1') + .set('Accept', 'application/vnd.card+json'); + assert.strictEqual(warm.status, 200, `warm-up GET: ${warm.text}`); + + // A provably credential-less caller can have no write in flight + // (anonymous writes are unsupported), so the gate skips them without + // consulting any scope — every gate is held open here, and the read + // must still return immediately. + let restore = stubUpdaterGates(testRealm, { + initiatedBy: () => NEVER, + all: () => NEVER, + }); + try { + let startedAt = Date.now(); + let response = await request + .get('/person-1') + .set('Accept', 'application/vnd.card+json'); + let elapsed = Date.now() - startedAt; + assert.strictEqual(response.status, 200, `HTTP 200: ${response.text}`); + assert.true( + elapsed < NO_WAIT_CEILING_MS, + `anonymous read skipped every pending-indexing hold (took ${elapsed}ms)`, + ); + } finally { + restore(); + } + }); + + test('a reader with an unverifiable token conservatively waits, bounded, on any pending indexing', async function (assert) { + let warm = await request + .get('/person-1') + .set('Accept', 'application/vnd.card+json'); + assert.strictEqual(warm.status, 200, `warm-up GET: ${warm.text}`); + + // A token that fails verification leaves the requester unknown — they + // presented credentials, so they might be the writer — and the drain + // covers every pending incremental job. + let gateResolved = false; + let gate = resolveAfter(GATE_RESOLVE_MS).then(() => { + gateResolved = true; + }); + let restore = stubUpdaterGates(testRealm, { + all: () => gate, + }); + try { + let startedAt = Date.now(); + let response = await request + .get('/person-1') + .set('Accept', 'application/vnd.card+json') + .set('Authorization', 'Bearer not-a-real-token'); + let elapsed = Date.now() - startedAt; + assert.strictEqual(response.status, 200, `HTTP 200: ${response.text}`); + assert.true( + gateResolved, + 'the unidentified read did not return before the gate settled', + ); + assert.true( + elapsed >= GATE_RESOLVE_MS - 20, + `unidentified read held for pending indexing (took ${elapsed}ms)`, + ); + } finally { + restore(); + } + }); + + test('a token accompanied by X-Boxel-Assume-User is not trusted for identity on the public path', async function (assert) { + let warm = await request + .get('/person-1') + .set('Accept', 'application/vnd.card+json'); + assert.strictEqual(warm.status, 200, `warm-up GET: ${warm.text}`); + + // Public read authorizes without the assume-user permission check, so + // the indirection can't be honored — and recording the bearer instead + // would desynchronize the read identity from the assumed-user tag the + // same client's writes carry. The requester stays unknown and takes + // the conservative bounded hold. + let gateResolved = false; + let gate = resolveAfter(GATE_RESOLVE_MS).then(() => { + gateResolved = true; + }); + let restore = stubUpdaterGates(testRealm, { + initiatedBy: () => undefined, + all: () => gate, + }); + try { + let startedAt = Date.now(); + let response = await request + .get('/person-1') + .set('Accept', 'application/vnd.card+json') + .set('X-Boxel-Assume-User', 'someone-else') + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'john', ['read'])}`, + ); + let elapsed = Date.now() - startedAt; + assert.strictEqual(response.status, 200, `HTTP 200: ${response.text}`); + assert.true( + gateResolved, + 'the assume-user read did not return before the gate settled', + ); + assert.true( + elapsed >= GATE_RESOLVE_MS - 20, + `assume-user read held for pending indexing (took ${elapsed}ms)`, + ); + } finally { + restore(); + } + }); + + test("a token-carrying reader on a public realm is identified and skips other writers' indexing", async function (assert) { + let warm = await request + .get('/person-1') + .set('Accept', 'application/vnd.card+json'); + assert.strictEqual(warm.status, 200, `warm-up GET: ${warm.text}`); + + // Public read authorizes without parsing the token, but the drain still + // learns who the requester is from the token they sent — so john is + // ruled out as the writer and skips both gates. + let restore = stubUpdaterGates(testRealm, { + initiatedBy: (user) => (user === 'hassan' ? NEVER : undefined), + all: () => NEVER, + }); + try { + let startedAt = Date.now(); + let response = await request + .get('/person-1') + .set('Accept', 'application/vnd.card+json') + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'john', ['read'])}`, + ); + let elapsed = Date.now() - startedAt; + assert.strictEqual(response.status, 200, `HTTP 200: ${response.text}`); + assert.true( + elapsed < NO_WAIT_CEILING_MS, + `identified reader skipped the pending-indexing hold (took ${elapsed}ms)`, + ); + } finally { + restore(); + } + }); + }); +}); diff --git a/packages/realm-server/tests/realm-index-updater-test.ts b/packages/realm-server/tests/realm-index-updater-test.ts index 441318fd8ab..05d83e16980 100644 --- a/packages/realm-server/tests/realm-index-updater-test.ts +++ b/packages/realm-server/tests/realm-index-updater-test.ts @@ -199,6 +199,74 @@ module(basename(import.meta.filename), function (hooks) { ); }); + test('incrementalIndexingInitiatedBy scopes the gate to the tagged user', async function (assert) { + let { queue, waiters } = makeStubQueue(); + let updater = new RealmIndexUpdater({ + realm: makeStubRealm(), + dbAdapter: {} as DBAdapter, + queue, + }); + + let { settled } = await updater.enqueueUpdate( + [new URL(`${realmURL}doomed.txt`)], + { initiatedBy: '@test-writer:localhost' }, + ); + settled.catch(() => {}); + + assert.notStrictEqual( + updater.incrementalIndexingInitiatedBy('@test-writer:localhost'), + undefined, + "the writer's own gate reflects their in-flight job", + ); + assert.strictEqual( + updater.incrementalIndexingInitiatedBy('@test-bystander:localhost'), + undefined, + "another user's gate does not see the job", + ); + assert.notStrictEqual( + updater.incrementalIndexing(), + undefined, + 'the unscoped gate still covers every pending job', + ); + + waiters[0].rejectFromResult(serializedWorkerError); + await updater.incrementalIndexing(); + + assert.strictEqual( + updater.incrementalIndexingInitiatedBy('@test-writer:localhost'), + undefined, + "the writer's gate is drained once the job settles", + ); + }); + + test('an untagged job is invisible to every user-scoped gate', async function (assert) { + let { queue, waiters } = makeStubQueue(); + let updater = new RealmIndexUpdater({ + realm: makeStubRealm(), + dbAdapter: {} as DBAdapter, + queue, + }); + + let { settled } = await updater.enqueueUpdate([ + new URL(`${realmURL}doomed.txt`), + ]); + settled.catch(() => {}); + + assert.strictEqual( + updater.incrementalIndexingInitiatedBy('@test-writer:localhost'), + undefined, + 'a system-originated job never blocks a user-scoped gate', + ); + assert.notStrictEqual( + updater.incrementalIndexing(), + undefined, + 'the unscoped gate covers the untagged job', + ); + + waiters[0].rejectFromResult(serializedWorkerError); + await updater.incrementalIndexing(); + }); + test('a failing copy job throws from copy() and resolves the gate', async function (assert) { let { queue, waiters } = makeStubQueue(); let updater = new RealmIndexUpdater({ diff --git a/packages/runtime-common/realm-index-updater.ts b/packages/runtime-common/realm-index-updater.ts index 947b0c0cd2e..b997a66a899 100644 --- a/packages/runtime-common/realm-index-updater.ts +++ b/packages/runtime-common/realm-index-updater.ts @@ -65,7 +65,16 @@ export class RealmIndexUpdater { // unrelated request happens to be awaiting the write-path gate, and // (b) become an unhandled promise rejection whenever a deferred-indexing // job fails while nothing is draining the gate. - #incrementalIndexingDeferreds = new Set>(); + // Each incremental/copy deferred is tagged with the matrix user whose HTTP + // write produced the job, when known. Read endpoints use the tag to scope + // their read-your-writes drain to the requesting user's own writes + // (`incrementalIndexingInitiatedBy`) instead of parking every reader behind + // whatever indexing happens to be in flight. System-originated jobs (file + // watcher, realm copy) carry no tag. + #incrementalIndexingDeferreds = new Map< + Deferred, + { initiatedBy?: string } + >(); #fullIndexingDeferreds = new Set>(); constructor({ @@ -116,7 +125,7 @@ export class RealmIndexUpdater { // hundreds of jobs and stall every PATCH for hours). indexing() { let pending = [ - ...this.#incrementalIndexingDeferreds, + ...this.#incrementalIndexingDeferreds.keys(), ...this.#fullIndexingDeferreds, ]; if (pending.length === 0) { @@ -137,12 +146,28 @@ export class RealmIndexUpdater { return undefined; } return Promise.all( - [...this.#incrementalIndexingDeferreds].map( + [...this.#incrementalIndexingDeferreds.keys()].map( (deferred) => deferred.promise, ), ).then(() => undefined); } + // Awaits only the incremental jobs whose write was initiated by `user` — + // the read-your-writes slice of `incrementalIndexing()`. Returns undefined + // when this user has nothing in flight, even while other users' or + // system-originated jobs are pending: those jobs can only make the caller's + // read fresher-than-requested, never wrong, because the production index + // rows stay live (and consistent) until the working-table swap lands. + incrementalIndexingInitiatedBy(user: string): Promise | undefined { + let pending = [...this.#incrementalIndexingDeferreds.entries()] + .filter(([, { initiatedBy }]) => initiatedBy === user) + .map(([deferred]) => deferred.promise); + if (pending.length === 0) { + return undefined; + } + return Promise.all(pending).then(() => undefined); + } + publishFullIndex( priority = systemInitiatedPriority, opts?: { clearLastModified?: boolean; awaitedByPublish?: boolean }, @@ -247,10 +272,16 @@ export class RealmIndexUpdater { // job's own rejection still propagates through `settled`. onFailed?: (error: unknown) => Promise | void; clientRequestId?: string | null; + // Matrix user whose HTTP write produced this job, when known. Scopes + // the read endpoints' read-your-writes drain — see + // #incrementalIndexingDeferreds. + initiatedBy?: string | null; }, ): Promise<{ settled: Promise }> { let indexingDeferred = new Deferred(); - this.#incrementalIndexingDeferreds.add(indexingDeferred); + this.#incrementalIndexingDeferreds.set(indexingDeferred, { + initiatedBy: opts?.initiatedBy ?? undefined, + }); let snapshotVersion = this.#ignoreDataVersion; let job: Job; try { @@ -328,6 +359,7 @@ export class RealmIndexUpdater { meta: { generation?: number }, ) => Promise; clientRequestId?: string | null; + initiatedBy?: string | null; }, ): Promise { let { settled } = await this.enqueueUpdate(urls, opts); @@ -339,7 +371,7 @@ export class RealmIndexUpdater { onInvalidation?: (invalidatedURLs: URL[]) => Promise, ): Promise<{ generation?: number }> { let indexingDeferred = new Deferred(); - this.#incrementalIndexingDeferreds.add(indexingDeferred); + this.#incrementalIndexingDeferreds.set(indexingDeferred, {}); try { let args: CopyArgs = { realmURL: this.#realm.url, diff --git a/packages/runtime-common/realm.ts b/packages/runtime-common/realm.ts index 9c2b0792721..05358808df7 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -418,6 +418,15 @@ const ARCHIVED_SEAL_EXEMPT_PATHS = new Set(['_readiness-check', '_session']); // carries its own, longer one — a published realm's HTML can take minutes and // its callers are pollers with deadlines to match. const READINESS_REQUEST_BUDGET_MS = 10_000; +// How long a card read holds its connection on the read-your-writes indexing +// drain (see `drainRequestersOwnIndexing`) before serving the current index +// generation anyway. Bounded for the same reason the readiness gates are: an +// unbounded hold is worse than a slightly stale answer. The index stays +// consistent throughout — incremental jobs write into the working table and +// only swap on completion — so a read that outlives the budget serves the +// previous generation, and the index event that follows the swap refreshes +// live clients. +const READ_INDEX_DRAIN_BUDGET_MS = 10_000; const MODULE_ETAG_VARIANT = 'module'; const SOURCE_ETAG_VARIANT = 'source'; // Card+JSON ETag is `"-[-]:card"` @@ -850,6 +859,14 @@ export interface WriteOptions { // the JSON-API card handlers do not, writing a single file and instances // respectively. waitForIndex?: boolean | null; + // 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 + // endpoints can scope their read-your-writes drain to this user's own + // reads. Absent for system-originated writes and for credential-less + // writes, whose jobs no reader waits on (anonymous writes are unsupported; + // see `drainRequestersOwnIndexing`). + initiatingUser?: string | null; } export interface RealmAdapter { @@ -938,6 +955,12 @@ interface Options { // SCREENSHOT_SYNC_WAIT_BUDGET_MS; tests shrink it to exercise the timeout // path without holding real time. screenshotSyncWaitMs?: number; + // How long a card read (card+json / card+html GET) holds its connection + // waiting on the requester's own in-flight incremental indexing before + // serving the current index generation anyway. Defaults to + // READ_INDEX_DRAIN_BUDGET_MS; tests shrink it to exercise the timeout path + // without holding real time. + readIndexDrainBudgetMs?: number; } interface UpdateItem { @@ -950,7 +973,29 @@ export interface MatrixConfig { username: string; } -export type RequestContext = { realm: Realm; permissions: RealmPermissions }; +export type RequestContext = { + realm: Realm; + permissions: RealmPermissions; + // The effective matrix user this request runs as (post `X-Boxel-Assume-User` + // indirection), recorded by `checkPermission` when it sees a verifiable + // token. Undefined when the request was authorized without one — a public + // endpoint or public-permission realm where no (valid) token accompanied + // the request, or a realm-internal (isLocal) dispatch. Used to scope the + // read endpoints' read-your-writes indexing drain to the requester's own + // writes; it is identity, not authority — authorization decisions never + // read it. + authenticatedUser?: string; + // Set by `checkPermission` when the request presented no Authorization + // header at all. Anonymous writes are unsupported (no realm grants `*` + // write in practice), so a provably credential-less caller has no + // read-your-writes claim and the read gate skips them outright — unlike a + // caller whose identity is merely unknown: a token that failed + // verification, an assume-user indirection the public path cannot + // validate, or a realm-internal dispatch that never ran `checkPermission`, + // all of which take the conservative bounded hold. Identity, not + // authority, like `authenticatedUser`. + anonymous?: true; +}; export class Realm { #startedUp = new Deferred(); @@ -963,6 +1008,9 @@ export class Realm { #adapter: RealmAdapter; #router: Router; #log = logger('realm'); + // One line per card read that arrives while incremental indexing is + // pending — see drainRequestersOwnIndexing for the outcome grammar. + #readGateLog = logger('realm:read-index-gate'); #perfLog = logger('perf'); #updateItems: UpdateItem[] = []; #flushUpdateEvents: Promise | undefined; @@ -1064,6 +1112,7 @@ export class Realm { #virtualNetwork: VirtualNetwork; #mediaCacheAdapter: MediaCacheAdapter | undefined; #screenshotSyncWaitMs: number; + #readIndexDrainBudgetMs: number; #cachedRealmInfo: RealmInfo | null = null; // md5 of the JSON-stringified `#cachedRealmInfo`. Folded into the // card+json ETag so any path that nulls `#cachedRealmInfo` (e.g. @@ -1199,6 +1248,8 @@ export class Realm { this.#mediaCacheAdapter = mediaCacheAdapter; this.#screenshotSyncWaitMs = opts?.screenshotSyncWaitMs ?? SCREENSHOT_SYNC_WAIT_BUDGET_MS; + this.#readIndexDrainBudgetMs = + opts?.readIndexDrainBudgetMs ?? READ_INDEX_DRAIN_BUDGET_MS; let owner: string | undefined; let _fetch = fetcher( virtualNetwork.fetch, @@ -1705,6 +1756,7 @@ export class Realm { opts?: { delete?: true; clientRequestId?: string | null; + initiatedBy?: string | null; }, ): Promise<{ invalidations: string[]; generation?: number }> { if (urls.length === 0) { @@ -1716,6 +1768,7 @@ export class Realm { await this.#realmIndexUpdater.update(urls, { ...(opts?.delete ? { delete: true } : {}), clientRequestId: opts?.clientRequestId ?? null, + initiatedBy: opts?.initiatedBy ?? null, onInvalidation: async (invalidatedURLs: URL[], meta) => { // Drop the searchCards in-flight map: the worker's batch.done() // swap landed in this realm's boxel_index, so any pending @@ -1755,6 +1808,7 @@ export class Realm { opts: { delete?: true; clientRequestId?: string | null; + initiatedBy?: string | null; onSettled?: ( invalidations: string[], meta: { generation?: number }, @@ -1773,6 +1827,7 @@ export class Realm { let { settled } = await this.#realmIndexUpdater.enqueueUpdate(urls, { ...(opts?.delete ? { delete: true } : {}), clientRequestId: opts?.clientRequestId ?? null, + initiatedBy: opts?.initiatedBy ?? null, onInvalidation: async (invalidatedURLs: URL[], meta) => { await this.clearRealmIndexCachesAndBroadcast(); await this.touchSourceRealmUpdatedAt(); @@ -1882,7 +1937,9 @@ export class Realm { } let { invalidations, generation } = - await this.updateIndexAndCollectInvalidations(urls); + await this.updateIndexAndCollectInvalidations(urls, { + initiatedBy: requestContext.authenticatedUser ?? null, + }); this.broadcastIncrementalInvalidationEvent(invalidations, { generation }); return createResponse({ @@ -2336,10 +2393,12 @@ export class Realm { let invalidations: Set = new Set(); let indexGeneration: number | undefined; let clientRequestId: string | null = options?.clientRequestId ?? null; + let initiatingUser: string | null = options?.initiatingUser ?? null; let performIndex = async () => { let { invalidations: workingInvalidations, generation } = await this.updateIndexAndCollectInvalidations(urls, { clientRequestId, + initiatedBy: initiatingUser, }); invalidations = new Set([...invalidations, ...workingInvalidations]); indexGeneration = generation ?? indexGeneration; @@ -2573,6 +2632,7 @@ export class Realm { urls, { clientRequestId, + initiatedBy: initiatingUser, // Route the post-worker broadcast through onSettled so it runs // INSIDE the indexing deferred lifecycle. Without this, the // broadcast would fire from an outer .then() after the deferred @@ -2932,6 +2992,7 @@ export class Realm { clientRequestId: request.headers.get('X-Boxel-Client-Request-Id'), serializeFile: true, waitForIndex, + initiatingUser: requestContext.authenticatedUser ?? null, }); } catch (e: any) { if (e instanceof CardError) { @@ -3048,7 +3109,7 @@ export class Realm { async delete( path: LocalPath, - options?: { waitForIndex?: boolean }, + options?: { waitForIndex?: boolean; initiatingUser?: string | null }, ): Promise { await this.#dbAdapter.withWriteLock(this.url, () => this._deleteUnlocked(path, options), @@ -3057,7 +3118,7 @@ export class Realm { private async _deleteUnlocked( path: LocalPath, - options?: { waitForIndex?: boolean }, + options?: { waitForIndex?: boolean; initiatingUser?: string | null }, ): Promise { let url = this.paths.fileURL(path); this.sendIndexInitiationEvent(url.href); @@ -3077,6 +3138,7 @@ export class Realm { let { invalidations, generation } = await this.updateIndexAndCollectInvalidations([url], { delete: true, + initiatedBy: options?.initiatingUser ?? null, }); this.broadcastIncrementalInvalidationEvent(invalidations, { generation }); } else { @@ -3090,6 +3152,7 @@ export class Realm { [url], { delete: true, + initiatedBy: options?.initiatingUser ?? null, onSettled: (deferredInvalidations, meta) => { this.broadcastIncrementalInvalidationEvent(deferredInvalidations, { generation: meta.generation, @@ -4969,6 +5032,33 @@ export class Realm { (requiredPermission === 'write' && realmPermissions['*']?.includes('write'))) ) { + // Authorized without needing a token. Record what we can about who the + // caller is for the read endpoints' indexing gate (identity, not + // authority — see RequestContext): a verifiable token identifies them, + // and no Authorization header at all marks them anonymous. Two cases + // deliberately leave both fields unset, landing the caller in the + // gate's conservative bucket: a token that fails verification (must + // not fail a request that public permissions already authorized), and + // a request carrying `X-Boxel-Assume-User` — honoring the indirection + // requires the assume-user permission check the main path runs (and + // identity capture must stay free of matrix round-trips), while + // recording the bearer instead would desynchronize this identity from + // the assumed-user tag the same client's writes carry, since writes + // always take the main path. + let publicAuthHeader = request.headers.get('Authorization'); + if (!publicAuthHeader) { + requestContext.anonymous = true; + } else if (!request.headers.get('X-Boxel-Assume-User')) { + try { + let publicToken = this.#adapter.verifyJWT( + publicAuthHeader.replace('Bearer ', ''), + this.#realmSecretSeed, + ); + requestContext.authenticatedUser = publicToken.user; + } catch (e) { + // fall through with no identity + } + } return; } @@ -5045,6 +5135,7 @@ export class Realm { AuthenticationErrorMessages.PermissionMismatch, ); } + requestContext.authenticatedUser = user; return; } @@ -5060,6 +5151,7 @@ export class Realm { // if the client is the realm matrix user then we permit all actions if (user === this.#matrixClientUserId) { + requestContext.authenticatedUser = user; return; } @@ -5085,6 +5177,7 @@ export class Realm { 'Insufficient permissions to perform this action', ); } + requestContext.authenticatedUser = user; } catch (e: any) { if (e?.constructor?.name === 'TokenExpiredError') { this.#log.warn( @@ -5118,6 +5211,7 @@ export class Realm { clientRequestId: request.headers.get('X-Boxel-Client-Request-Id'), serializeFile: false, waitForIndex: false, + initiatingUser: requestContext.authenticatedUser ?? null, }, ); return createResponse({ @@ -5148,6 +5242,7 @@ export class Realm { clientRequestId: request.headers.get('X-Boxel-Client-Request-Id'), serializeFile: false, waitForIndex: false, + initiatingUser: requestContext.authenticatedUser ?? null, }, ); return createResponse({ @@ -5402,7 +5497,10 @@ export class Realm { if (!handle) { return notFound(request, requestContext, `${localName} not found`); } - await this.delete(handle.path, { waitForIndex: false }); + await this.delete(handle.path, { + waitForIndex: false, + initiatingUser: requestContext.authenticatedUser ?? null, + }); return createResponse({ body: null, init: { status: 204 }, @@ -5945,6 +6043,7 @@ export class Realm { } let [{ lastModified, created }] = await this.writeMany(files, { clientRequestId: request.headers.get('X-Boxel-Client-Request-Id'), + initiatingUser: requestContext.authenticatedUser ?? null, ...(duringPrerender ? { waitForIndex: false } : {}), }); @@ -6276,6 +6375,7 @@ export class Realm { // connection). let [{ lastModified, created }] = await this._batchWriteUnlocked(files, { clientRequestId: request.headers.get('X-Boxel-Client-Request-Id'), + initiatingUser: requestContext.authenticatedUser ?? null, ...(duringPrerender ? { waitForIndex: false } : {}), }); let doc: SingleCardDocument; @@ -6478,21 +6578,112 @@ export class Realm { } } + // Read-your-writes gate for the card read endpoints (card+json / + // card+html GET). The +source POST indexes deferred (it returns once the + // bytes are durable), so a GET that immediately follows the same client's + // definition rewrite would otherwise read a stale snapshot — e.g. a + // post-rename instance still serialized under the old schema. Waiting is a freshness courtesy, not a correctness + // requirement: incremental jobs write into the working table and the + // production rows stay live (and mutually consistent) until the completed + // batch swaps in, so a read during indexing serves the previous + // generation, never a torn one. That shapes both bounds here: + // + // - Scoped to the requester. Only the principal whose own write is in + // flight has a read-your-writes expectation; every other reader takes + // the current generation immediately. A write-heavy user (or a module + // edit whose invalidation set spans the realm's module graph) must not + // park every reader of the realm behind their job. A provably + // credential-less caller has no read-your-writes claim at all — + // anonymous writes are unsupported — so an anonymous read skips the + // gate outright and never parks behind an identified user's reindex. + // System-originated jobs (file watcher, realm copy) + // are untagged and hold no scoped reader — no one has a + // read-your-writes claim on them. Only when the requester is genuinely + // unknown — a token that failed verification, an assume-user + // indirection the public path cannot validate, or a realm-internal + // dispatch — does the drain conservatively cover all pending + // incremental jobs, since the writer might be behind any of them. + // - Bounded. The job being awaited can also sit queued behind other + // realms' work in a saturated worker pool; past the budget the read + // proceeds on the current generation and the index event that follows + // the swap refreshes the client. + // + // A prerender-originated request skips the gate entirely: its tab holds a + // render slot that the pending job may itself be waiting on, so any wait + // here is at best dead time and at worst a deadlock held for the budget. + // + // Every read that arrives while incremental indexing is pending emits one + // `realm:read-index-gate` key=value line (`outcome=` skipped-prerender / + // skipped-not-writer / settled / budget-expired, with `waitMs=` and the + // requester principal on the waited outcomes) — the skipped outcomes count + // reads an unscoped gate would have parked, the waited ones measure what + // the requester-scoped hold actually costs. Steady-state reads (nothing + // pending) emit nothing. budget-expired logs at warn; the rest at info. + private async drainRequestersOwnIndexing( + request: Request, + requestContext: RequestContext, + ): Promise { + let anyPending = this.incrementalIndexing(); + if (!anyPending) { + return; + } + let emit = ( + level: 'info' | 'warn', + outcome: string, + fragment: string = '', + ) => + this.#readGateLog[level]( + `outcome=${outcome}${fragment} url=${request.url}`, + ); + if (isDuringPrerenderRequest(request)) { + emit('info', 'skipped-prerender'); + return; + } + if (requestContext.anonymous) { + // A provably credential-less caller has no write in flight to wait on + // (anonymous writes are unsupported), so they are never the writer. + emit('info', 'skipped-not-writer'); + return; + } + let requester = requestContext.authenticatedUser; + let pending: Promise | undefined; + let scope: 'own' | 'all'; + if (requester !== undefined) { + pending = + this.#realmIndexUpdater.incrementalIndexingInitiatedBy(requester); + if (!pending) { + emit('info', 'skipped-not-writer'); + return; + } + scope = 'own'; + } else { + pending = anyPending; + scope = 'all'; + } + let waitStartedAt = Date.now(); + let settled = await settledBy( + pending, + waitStartedAt + this.#readIndexDrainBudgetMs, + ); + let fragment = + ` scope=${scope} waitMs=${Date.now() - waitStartedAt}` + + (requester !== undefined ? ` user=${requester}` : ''); + if (settled) { + emit('info', 'settled', fragment); + } else { + // Past the budget the read proceeds on the current index generation. + emit('warn', 'budget-expired', fragment); + } + } + private async getCard( request: Request, requestContext: RequestContext, ): Promise { - // Drain any in-flight incremental indexing before reading the card from - // the index. With CS-11003's deferred +source POST, an immediately- - // following GET +json after a definition rewrite would otherwise read - // a stale snapshot — e.g. a post-rename instance still serialized - // under the old schema. Read endpoints serve the realm's canonical - // indexed view; the small wait when indexing is genuinely pending is - // the right tradeoff vs returning stale state. - let pending = this.incrementalIndexing(); - if (pending) { - await pending; - } + // Read-your-writes: wait (scoped, bounded) on the requester's own + // in-flight incremental indexing before reading the card from the + // index. See drainRequestersOwnIndexing. + await this.drainRequestersOwnIndexing(request, requestContext); let requestedLocalPath = this.paths.local(new URL(request.url)); let requestedHadJsonExtension = requestedLocalPath.endsWith('.json'); // `.json` requests always 302 to the canonical no-extension URL, @@ -6771,12 +6962,10 @@ export class Realm { ? SupportedMimeType.FileMetaHtml : SupportedMimeType.CardHtml; - // Read endpoints serve the realm's canonical indexed view — drain any - // in-flight incremental indexing first, exactly as `getCard` does. - let pending = this.incrementalIndexing(); - if (pending) { - await pending; - } + // Read-your-writes: wait (scoped, bounded) on the requester's own + // in-flight incremental indexing first, exactly as `getCard` does. See + // drainRequestersOwnIndexing. + await this.drainRequestersOwnIndexing(request, requestContext); let htmlQuery: HtmlQuery; let fieldset: SearchEntryFieldset; @@ -6944,7 +7133,9 @@ export class Realm { return notFound(request, requestContext); } let path = this.paths.local(url) + '.json'; - await this.delete(path); + await this.delete(path, { + initiatingUser: requestContext.authenticatedUser ?? null, + }); return createResponse({ body: null, init: { status: 204 }, @@ -7763,7 +7954,7 @@ export class Realm { // pushes a fix and immediately polls this endpoint could otherwise // see a stale snapshot — either still reporting an error the just- // pushed fix cleared, or missing a fresh failure from the same write. - // Same hazard publishability() guards against (see realm.ts:5629). + // Same hazard publishability() guards against. let pending = this.incrementalIndexing(); if (pending) { await pending;