diff --git a/packages/realm-server/handlers/handle-search.ts b/packages/realm-server/handlers/handle-search.ts index fcbd75f9fc3..6055c280443 100644 --- a/packages/realm-server/handlers/handle-search.ts +++ b/packages/realm-server/handlers/handle-search.ts @@ -22,6 +22,7 @@ import { } from '@cardstack/runtime-common'; import { fetchRequestFromContext, + releaseSearchAdmission, sendResponseForBadRequest, setContextResponse, } from '../middleware/index.ts'; @@ -381,16 +382,25 @@ async function respondWithJobScopedSearchCache( query, opts: { ...(args.opts as Record), generations }, populate: runSearch, + // A joiner or a hit holds no result document of its own, so it stops + // counting toward the search admission ceiling here rather than when + // its response ends; the ceiling is then a bound on concurrent + // computations, which is what holds the heap. + onOutcome: (decided) => { + if (decided !== 'miss') { + releaseSearchAdmission(ctxt); + } + }, }); - await setContextResponse( - ctxt, - new Response(body, { - headers: { - 'content-type': SupportedMimeType.CardJson, - [LIVE_SEARCH_CACHE_HEADER]: outcome, - }, - }), - ); + // The body is a string the cache may be handing to many requests at once, + // so it goes to Koa as-is. Wrapping it in a `Response` would encode it into + // a stream that `setContextResponse` decodes back into a per-request copy; + // this way a joiner's or a hit's cost on the way out is Koa's own write of + // the shared string and nothing more. + ctxt.status = 200; + ctxt.set('content-type', SupportedMimeType.CardJson); + ctxt.set(LIVE_SEARCH_CACHE_HEADER, outcome); + ctxt.body = body; emitTimeline(); return; } diff --git a/packages/realm-server/live-search-cache.ts b/packages/realm-server/live-search-cache.ts index 319fe4c2f81..bb2fcd0fc6b 100644 --- a/packages/realm-server/live-search-cache.ts +++ b/packages/realm-server/live-search-cache.ts @@ -175,11 +175,18 @@ export class LiveSearchCache { this.#emitTelemetry = opts?.emitTelemetry ?? defaultEmitTelemetry; } + // `onOutcome` fires the moment the cache decides how it will satisfy the + // request — before a joiner starts waiting on the in-flight compute, before + // a hit returns, and as a miss starts its populate — so a caller can act on + // the decision while the request is still in progress. The realm-server's + // search admission uses it to hand back the slot of a request that holds no + // result document of its own. async getOrPopulate(args: { realms: string[]; query: Query; opts: unknown | undefined; populate: () => Promise; + onOutcome?: (outcome: LiveSearchOutcome) => void; }): Promise<{ body: string; outcome: LiveSearchOutcome }> { try { return await this.#getOrPopulate(args); @@ -193,9 +200,11 @@ export class LiveSearchCache { query: Query; opts: unknown | undefined; populate: () => Promise; + onOutcome?: (outcome: LiveSearchOutcome) => void; }): Promise<{ body: string; outcome: LiveSearchOutcome }> { this.#reapExpiredHead(); let key = searchRequestKeyHash(args.realms, args.query, args.opts); + let onOutcome = args.onOutcome ?? (() => {}); let entry = this.#entries.get(key); if (entry) { @@ -206,6 +215,7 @@ export class LiveSearchCache { this.#entries.set(key, entry); this.#counters.hits += 1; this.#counters.hitBytes += entry.body.length; + onOutcome('hit'); return { body: entry.body, outcome: 'hit' }; } this.#delete(key, entry); @@ -215,11 +225,13 @@ export class LiveSearchCache { let inFlight = this.#inFlight.get(key); if (inFlight) { this.#counters.joins += 1; + onOutcome('join'); let body = await inFlight; this.#counters.joinBytes += body.length; return { body, outcome: 'join' }; } + onOutcome('miss'); let promise = args.populate(); this.#inFlight.set(key, promise); try { diff --git a/packages/realm-server/middleware/index.ts b/packages/realm-server/middleware/index.ts index bc84d857d9a..367e95eddf9 100644 --- a/packages/realm-server/middleware/index.ts +++ b/packages/realm-server/middleware/index.ts @@ -222,12 +222,34 @@ export function httpLogging(ctxt: Koa.Context, next: Koa.Next) { return next(); } -// Puts a search through the admission gate (`search-inflight.ts`) and holds -// its slot for the request's full lifecycle (parse → SQL → serialize → send), -// which is the window in which it holds heap and in which a saturated event -// loop would leave it unserviced. Mounted after CORS so that a shed response -// carries the headers a cross-origin client needs to read its status and -// Retry-After; before the body is parsed so that a shed costs nothing. +// Where `searchAdmission` leaves the release for the slot it granted, so a +// handler can hand the slot back before the response ends. +const SEARCH_ADMISSION_RELEASE = 'searchAdmissionRelease'; + +// Hand back the admission slot a search request holds, if it holds one. A +// request that the live-search cache satisfies from another request's +// computation — a `join` or a `hit` — builds no result document of its own, so +// it calls this the moment the cache says so and stops counting toward the +// ceiling; only the request doing the computing keeps its slot until its +// response ends. That holds for an indexing-lane admission too: an in-render +// search served from another request's computation holds no document either, +// and the count is of computations, whichever lane admitted them. Idempotent, +// and a no-op for requests the gate never saw. +export function releaseSearchAdmission(ctxt: Koa.Context): void { + let release = ctxt.state[SEARCH_ADMISSION_RELEASE]; + if (typeof release === 'function') { + release(); + } +} + +// Puts a search through the admission gate (`search-inflight.ts`). A search +// that computes its own result holds its slot for the request's full lifecycle +// (parse → SQL → serialize → send), which is the window in which it holds heap +// and in which a saturated event loop would leave it unserviced; one that the +// live-search cache serves from another's computation hands the slot back +// early via `releaseSearchAdmission`. Mounted after CORS so that a shed +// response carries the headers a cross-origin client needs to read its status +// and Retry-After; before the body is parsed so that a shed costs nothing. export async function searchAdmission(ctxt: Koa.Context, next: Koa.Next) { if ( !SEARCH_PATH_PATTERN.test(ctxt.path) || @@ -249,6 +271,7 @@ export async function searchAdmission(ctxt: Koa.Context, next: Koa.Next) { release?.(); release = undefined; }; + ctxt.state[SEARCH_ADMISSION_RELEASE] = releaseSlot; // `finish` fires on a fully-sent response; `close` covers a connection // torn down before that, so a slot can't leak on an abort. ctxt.res.on('finish', releaseSlot); diff --git a/packages/realm-server/search-inflight.ts b/packages/realm-server/search-inflight.ts index 1ed7871d22e..92978b30423 100644 --- a/packages/realm-server/search-inflight.ts +++ b/packages/realm-server/search-inflight.ts @@ -4,14 +4,18 @@ import { } from '@cardstack/runtime-common'; // Admission gate for the realm-server's search endpoints (`/_search`, -// `/_federated-search`). Every in-flight search holds tens of MB of heap while -// its result set is assembled, and the process is a single event loop, so the -// number of searches running at once is what decides whether the heap survives -// a burst. The gate bounds it: up to `limit` searches run concurrently, -// arrivals above that wait up to a bounded time for a slot in FIFO order, and a -// request still waiting when its time is up is shed — the middleware answers -// 429 + Retry-After without having parsed a body or touched the index, so a -// shed costs the process almost nothing. +// `/_federated-search`). A search that assembles its own result document holds +// tens of MB of heap while it does, and the process is a single event loop, so +// the number of such computations running at once is what decides whether the +// heap survives a burst. The gate bounds it: up to `limit` searches hold a +// slot at once, arrivals above that wait up to a bounded time for a slot in +// FIFO order, and a request still waiting when its time is up is shed — the +// middleware answers 429 + Retry-After without having parsed a body or +// touched the index, so a shed costs the process almost nothing. A request +// that the live-search cache serves from another request's computation hands +// its slot back as soon as the cache decides so, so in steady state the slots +// are held by computations plus the requests briefly between admission and +// the cache lookup. // // Indexing traffic (a request stamped with a prerender job id or the // during-prerender header) is admitted unconditionally: shedding an in-render diff --git a/packages/realm-server/tests/live-search-cache-test.ts b/packages/realm-server/tests/live-search-cache-test.ts index 034c5c3dadb..b434ba68de3 100644 --- a/packages/realm-server/tests/live-search-cache-test.ts +++ b/packages/realm-server/tests/live-search-cache-test.ts @@ -67,6 +67,42 @@ module(basename(import.meta.filename), function () { assert.strictEqual(a.body, b.body, 'both callers share the body'); }); + test('onOutcome fires when the cache decides, not when the body resolves', async function (assert) { + let cache = new LiveSearchCache({ ttlMs: 60_000 }); + let deferred = deferredPopulate('{"data":[1]}'); + let realms = ['http://a/']; + let decided: string[] = []; + let load = () => + cache.getOrPopulate({ + realms, + query: personQuery(), + opts: undefined, + populate: deferred.populate, + onOutcome: (outcome) => decided.push(outcome), + }); + + let first = load(); + assert.deepEqual(decided, ['miss'], 'the miss is announced as it starts'); + let second = load(); + assert.deepEqual( + decided, + ['miss', 'join'], + 'the join is announced before the joiner has anything to wait on', + ); + assert.strictEqual(deferred.calls, 1, 'one populate'); + + deferred.resolve(); + await Promise.all([first, second]); + + let third = load(); + assert.deepEqual( + decided, + ['miss', 'join', 'hit'], + 'a hit is announced synchronously on the call', + ); + assert.strictEqual((await third).outcome, 'hit'); + }); + test('different queries do not coalesce', async function (assert) { let cache = new LiveSearchCache({ ttlMs: 60_000 }); let realms = ['http://a/']; diff --git a/packages/realm-server/tests/search-admission-test.ts b/packages/realm-server/tests/search-admission-test.ts index c64817086b7..f2b4cc4ee27 100644 --- a/packages/realm-server/tests/search-admission-test.ts +++ b/packages/realm-server/tests/search-admission-test.ts @@ -12,7 +12,11 @@ import { resetSearchAdmissionForTests, setSearchAdmissionForTests, } from '../search-inflight.ts'; -import { httpLogging, searchAdmission } from '../middleware/index.ts'; +import { + httpLogging, + releaseSearchAdmission, + searchAdmission, +} from '../middleware/index.ts'; // The admission gate is what stands between a burst of searches and a heap // exhausted by their concurrent result sets. These tests pin the contract the @@ -177,6 +181,9 @@ module(basename(import.meta.filename), function () { let app = new Koa(); let router = new Router(); let search = async (ctxt: Koa.Context) => { + if (ctxt.query.releaseEarly) { + releaseSearchAdmission(ctxt); + } if (ctxt.query.hold) { await new Promise((resolve) => { holds.push(resolve); @@ -225,7 +232,8 @@ module(basename(import.meta.filename), function () { headers: Record = {}, ) { let held = new Promise((resolve) => (onHeld = resolve)); - let response = send(app, `${path}?hold=1`, headers); + let separator = path.includes('?') ? '&' : '?'; + let response = send(app, `${path}${separator}hold=1`, headers); return { response, held }; } @@ -252,12 +260,11 @@ module(basename(import.meta.filename), function () { return [first.response, second.response]; } - function assert_inFlight(expected: number) { - QUnit.assert.strictEqual( - getSearchInFlight(), - expected, - `inFlight=${expected}`, - ); + function assert_inFlight( + expected: number, + message = `inFlight=${expected}`, + ) { + QUnit.assert.strictEqual(getSearchInFlight(), expected, message); } test('a search arriving above the ceiling is shed with 429 and Retry-After', async function (assert) { @@ -353,6 +360,41 @@ module(basename(import.meta.filename), function () { await Promise.all(held); }); + test('a handler can hand its slot back before its response ends', async function (assert) { + setSearchAdmissionForTests({ limit: 2, waitMs: 2000 }); + let app = buildApp(); + let releasing = holdSearch(app, '/_federated-search?releaseEarly=1'); + await releasing.held; + assert_inFlight(0, 'released while the response is still open'); + + // The freed slot is real: with the gate full again, a waiter is blocked + // by the two computing searches, not by the early-released one. + let held = await fillGate(app); + let waiting = send(app, '/_federated-search'); + await wait(50); + let blocked = await settledWithin(waiting, 20); + assert.false(blocked.settled, 'the gate is full'); + + // Ending the early-released response hands back nothing more. + holds.shift()!(); + await releasing.response; + let stillBlocked = await settledWithin(waiting, 20); + assert.false(stillBlocked.settled, 'no second release on finish'); + assert_inFlight(2); + + holds.shift()!(); + assert.strictEqual( + (await waiting).status, + 200, + 'a computing search ending admits the waiter', + ); + for (let release of holds) { + release(); + } + await Promise.all(held); + assert_inFlight(0); + }); + test('a non-search request is not counted', async function (assert) { let app = buildApp(); let { response, held } = holdSearch(app, '/some-realm/cards'); diff --git a/packages/realm-server/tests/server-endpoints/search-test.ts b/packages/realm-server/tests/server-endpoints/search-test.ts index 01802d40e36..5fb2237b776 100644 --- a/packages/realm-server/tests/server-endpoints/search-test.ts +++ b/packages/realm-server/tests/server-endpoints/search-test.ts @@ -24,6 +24,7 @@ import type { PgAdapter } from '@cardstack/postgres'; import { resetCatalogRealms } from '../../handlers/handle-fetch-catalog-realms.ts'; import { LIVE_SEARCH_CACHE_HEADER } from '../../handlers/handle-search.ts'; import { LiveSearchCache } from '../../live-search-cache.ts'; +import { getSearchInFlight } from '../../search-inflight.ts'; import { closeServer, createVirtualNetwork, @@ -213,6 +214,18 @@ module(`server-endpoints/${basename(import.meta.filename)}`, function (_hooks) { return postSearchAs(ownerToken(), body); } + // Poll for a condition that a request in flight will bring about, failing + // rather than hanging if it never does. + async function waitUntil(condition: () => boolean, what: string) { + let deadline = Date.now() + 5_000; + while (!condition()) { + if (Date.now() > deadline) { + throw new Error(`timed out waiting for ${what}`); + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + } + test('QUERY /_federated-search federates entry results across realms', async function (assert) { let response = await postSearch({ filter: personFilter(), @@ -701,6 +714,90 @@ module(`server-endpoints/${basename(import.meta.filename)}`, function (_hooks) { ); }); + test('a coalesced live search releases its admission slot while the compute is still running', async function (assert) { + // A cache whose next compute waits on the test, so a second identical + // request is guaranteed to arrive while the first is still computing. + class HoldableLiveSearchCache extends LiveSearchCache { + hold: Promise | undefined; + onComputeStarted: (() => void) | undefined; + override getOrPopulate( + args: Parameters[0], + ) { + let { hold, onComputeStarted } = this; + return super.getOrPopulate({ + ...args, + populate: async () => { + onComputeStarted?.(); + if (hold) { + await hold; + } + return args.populate(); + }, + }); + } + } + let cache = new HoldableLiveSearchCache({ + ttlMs: 60_000, + telemetryIntervalMs: 0, + }); + await stopSearchRealmServer(); + await startSearchRealmServer({ + dbAdapter, + publisher, + runner, + liveSearchCache: cache, + }); + await settlePrerenderHtmlJobs(dbAdapter, testRealm.url); + await settlePrerenderHtmlJobs(dbAdapter, secondaryRealm.url); + + let releaseCompute!: () => void; + cache.hold = new Promise((resolve) => (releaseCompute = resolve)); + let computeStarted = new Promise( + (resolve) => (cache.onComputeStarted = resolve), + ); + let searchBody = { + filter: personFilter(), + realms: [testRealm.url, secondaryRealm.url], + }; + + // supertest sends lazily; `.then` starts each request now. + let first = postSearch(searchBody).then((response) => response); + await computeStarted; + assert.strictEqual( + getSearchInFlight(), + 1, + 'the computing request holds a slot', + ); + + let second = postSearch(searchBody).then((response) => response); + await waitUntil( + () => cache.stats.joins === 1, + 'the second request joins', + ); + assert.strictEqual( + getSearchInFlight(), + 1, + 'the joiner handed its slot back while the compute is still running', + ); + + releaseCompute(); + let [a, b] = await Promise.all([first, second]); + assert.strictEqual(a.status, 200); + assert.strictEqual(b.status, 200); + assert.strictEqual(a.headers[LIVE_SEARCH_CACHE_HEADER], 'miss'); + assert.strictEqual(b.headers[LIVE_SEARCH_CACHE_HEADER], 'join'); + assert.strictEqual(b.text, a.text, 'the joiner got the shared body'); + assert.strictEqual( + getSearchInFlight(), + 0, + 'the computing request released on completion, and only once', + ); + + let third = await postSearch(searchBody); + assert.strictEqual(third.headers[LIVE_SEARCH_CACHE_HEADER], 'hit'); + assert.strictEqual(getSearchInFlight(), 0, 'a hit holds no slot after'); + }); + test('a write to a searched realm invalidates the live search cache', async function (assert) { let searchBody = { filter: personFilter(), diff --git a/packages/runtime-common/search-bounds.ts b/packages/runtime-common/search-bounds.ts index dd2ebf07305..1fcb3417362 100644 --- a/packages/runtime-common/search-bounds.ts +++ b/packages/runtime-common/search-bounds.ts @@ -152,14 +152,18 @@ export const SEARCH_CONCURRENCY_CAP = parsePositiveInt( MIN_CONCURRENCY, ); -// Max searches the realm-server process runs at once, across every caller. -// Enforced server-side at admission (see the realm-server's -// `search-inflight.ts`). Sized against the per-search heap cost: a few dozen -// concurrent federated searches exhaust a 2 GB heap, so the default keeps a -// process on the default heap alive and leaves headroom on a larger one. -// Indexing traffic is admitted regardless of this ceiling (it is bounded -// upstream by the prerender pool), so the effective room for interactive -// searches is whatever indexing isn't using. +// Max search admission slots the realm-server process hands out at once, +// across every caller. Enforced server-side at admission (see the realm-server's +// `search-inflight.ts`), before the request body is read. A request that the +// live-search cache serves from another request's computation hands its slot +// back as soon as the cache says so, so the slots are held by searches +// assembling their own result document — the ones that hold heap — plus the +// requests briefly between admission and the cache lookup. Sized so that a +// full gate of distinct computations fits a 2 GB heap: each holds tens of MB +// while it assembles, and a few dozen exhaust that heap. Raise it per +// environment where the heap allows. Indexing traffic is admitted regardless +// of this ceiling (it is bounded upstream by the prerender pool), so the +// effective room for interactive searches is whatever indexing isn't using. export const SERVER_MAX_IN_FLIGHT_SEARCHES = parsePositiveInt( env.SERVER_MAX_IN_FLIGHT_SEARCHES, DEFAULT_SERVER_MAX_IN_FLIGHT_SEARCHES,