From 385c10ebd3ca77035e64b49cf463b5c394876575 Mon Sep 17 00:00:00 2001 From: Buck Doyle Date: Thu, 10 Sep 2026 08:55:16 -0400 Subject: [PATCH 1/3] Let coalesced live searches hand back their admission slot The search admission gate bounds concurrent requests, but the quantity that holds heap is concurrent computations: a live search the LiveSearchCache resolves by joining an in-flight compute or serving a cached body builds no result document of its own. Counting those requests against the ceiling for their whole lifetime made the gate shed exactly the traffic the cache had made cheap, and left the ceiling tracking something other than heap. The cache now announces its decision through an onOutcome hook the moment it makes it, and the search handler uses that to release the request's admission slot on a join or hit, so only the computing request keeps its slot until its response ends. The middleware leaves the release on the request state for the handler to find. The ceiling's comments now describe it as a bound on computations, sized so a full gate of distinct ones fits a 2 GB heap. Co-Authored-By: Claude Fable 5.1 --- .../realm-server/handlers/handle-search.ts | 10 ++++ packages/realm-server/live-search-cache.ts | 12 ++++ packages/realm-server/middleware/index.ts | 32 ++++++++-- packages/realm-server/search-inflight.ts | 20 ++++--- .../tests/live-search-cache-test.ts | 36 ++++++++++++ .../tests/search-admission-test.ts | 58 ++++++++++++++++--- packages/runtime-common/search-bounds.ts | 20 ++++--- 7 files changed, 158 insertions(+), 30 deletions(-) diff --git a/packages/realm-server/handlers/handle-search.ts b/packages/realm-server/handlers/handle-search.ts index fcbd75f9fc3..8eced00c281 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,6 +382,15 @@ 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, 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..aa4e3ea61bc 100644 --- a/packages/realm-server/middleware/index.ts +++ b/packages/realm-server/middleware/index.ts @@ -222,12 +222,31 @@ 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. 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 +268,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/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, From cddac3a9bb9758105bbb5665681e91f4e3ba04b8 Mon Sep 17 00:00:00 2001 From: Buck Doyle Date: Thu, 10 Sep 2026 16:12:37 -0400 Subject: [PATCH 2/3] Send live-search bodies to Koa without a Response round trip The live search path wrapped the cache's body string in a Response, which encodes it into a stream that setContextResponse then decodes back into a per-request string, so every response to a shared body made two more copies of it before Koa wrote a third. With joiners and hits no longer holding an admission slot, that per-response cost is what bounds a burst of identical searches on the way out, so the handler now hands the shared string to Koa directly and each response costs its write alone. Co-Authored-By: Claude Fable 5.1 --- .../realm-server/handlers/handle-search.ts | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/packages/realm-server/handlers/handle-search.ts b/packages/realm-server/handlers/handle-search.ts index 8eced00c281..6055c280443 100644 --- a/packages/realm-server/handlers/handle-search.ts +++ b/packages/realm-server/handlers/handle-search.ts @@ -392,15 +392,15 @@ async function respondWithJobScopedSearchCache( } }, }); - 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; } From 4866dd584d0e366fdba572cfebc2b615430df55a Mon Sep 17 00:00:00 2001 From: Buck Doyle Date: Thu, 10 Sep 2026 16:12:40 -0400 Subject: [PATCH 3/3] Cover slot release through the real search route The unit tests pin the cache's onOutcome timing and the middleware's release helper separately; this drives the production wiring end to end. A cache whose compute waits on the test holds the first request mid-compute, a second identical request joins it, and the gate's count stays at one while both are in flight. Also notes on the release helper that it applies to indexing-lane admissions for the same reason. Co-Authored-By: Claude Fable 5.1 --- packages/realm-server/middleware/index.ts | 5 +- .../tests/server-endpoints/search-test.ts | 97 +++++++++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) diff --git a/packages/realm-server/middleware/index.ts b/packages/realm-server/middleware/index.ts index aa4e3ea61bc..367e95eddf9 100644 --- a/packages/realm-server/middleware/index.ts +++ b/packages/realm-server/middleware/index.ts @@ -231,7 +231,10 @@ const SEARCH_ADMISSION_RELEASE = 'searchAdmissionRelease'; // 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. Idempotent, and a no-op for requests the gate never saw. +// 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') { 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(),