diff --git a/tools/content-node/interactors-endpoints.ts b/tools/content-node/interactors-endpoints.ts index a29e6adb..b385e5ba 100644 --- a/tools/content-node/interactors-endpoints.ts +++ b/tools/content-node/interactors-endpoints.ts @@ -15,6 +15,7 @@ import { } from './spring.ts'; import { RawJson, + deepShare, javaError, neo4jFailureMessage, nonEmptyJson, @@ -315,6 +316,9 @@ export async function refreshPsicquicResources({ timeout = 60_000 } = {}) { } } +/** Batches of molecule lookups running at once (POST /interactors/static/molecules/details). */ +const MOLECULE_BATCHES = deepShare(4); + export const interactorsEndpoints: Endpoint[] = [ { path: '/ContentService/interactors/static/molecule/{id}/details', @@ -464,12 +468,36 @@ export const interactorsEndpoints: Endpoint[] = [ const proteins = plainText(request.body, contentType); const accs = new JavaHashMap(); for (const acc of plainIds(proteins)) accs.set(acc, true); + // One query per accession, one at a time as Java asks: a diagram's + // proteins (273 for R-HSA-69620) take a second, and a 10MB body holds + // hundreds of thousands. Batches have a share of their own, and stop + // when nobody is left to read the answer (as the xrefs batch does). + const gone = new AbortController(); + request.res?.on('close', () => { + if (!request.res?.writableFinished) gone.abort(); + }); + if (!MOLECULE_BATCHES.enter()) + return { + ...javaError( + request, + 503, + 'SERVICE_UNAVAILABLE', + 'The service is busy; try again shortly' + ), + headers: { 'Retry-After': '5' }, + }; const entities: [string, InteractionRow[]][] = []; - for (const acc of accs.keys()) { - const rows = await interactionsOf(request, acc, window); - if ('answer' in rows) return rows.answer; - if (rows.some((row) => row.accession === null)) return unexpected(request); - entities.push([acc, rows]); + try { + for (const acc of accs.keys()) { + // Nobody to answer: whatever is written goes nowhere. + if (gone.signal.aborted) return { status: 500 }; + const rows = await interactionsOf(request, acc, window); + if ('answer' in rows) return rows.answer; + if (rows.some((row) => row.accession === null)) return unexpected(request); + entities.push([acc, rows]); + } + } finally { + MOLECULE_BATCHES.leave(); } if (entities.every(([, rows]) => rows.length === 0)) return notFound(request, `No interactors found for accession: ${proteins}`); diff --git a/tools/content-node/molecule-details-bound.spec.mjs b/tools/content-node/molecule-details-bound.spec.mjs new file mode 100644 index 00000000..4fbe3da6 --- /dev/null +++ b/tools/content-node/molecule-details-bound.spec.mjs @@ -0,0 +1,109 @@ +// @vitest-environment node +// +// POST /interactors/static/molecules/details runs one query per accession, one +// at a time, and a 10MB body holds hundreds of thousands: its batches have a +// share of their own, and stop when the client is gone. A stand-in graph +// holds every query open until the test lets it go. +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; +import { request as httpRequest } from 'node:http'; + +const held = []; +let asked = 0; +vi.mock('./graph.ts', () => ({ + read: () => { + asked += 1; + return new Promise((resolve) => held.push(() => resolve([]))); + }, + configured: () => true, + close: async () => {}, + envSetting: () => '', +})); + +const { app } = await import('./server.ts'); + +let server; +let port; +beforeAll(async () => { + await new Promise((resolve) => { + server = app().listen(0, '127.0.0.1', resolve); + }); + port = server.address().port; +}); +afterAll(() => new Promise((resolve) => server.close(resolve))); + +/** A POST of these accessions: its response, and a way to drop the connection. */ +function post(accessions) { + const sent = httpRequest({ + host: '127.0.0.1', + port, + // A connection of its own, closed with its answer: the server's count of + // open connections then says when a dropped one has gone. + agent: false, + method: 'POST', + path: '/ContentService/interactors/static/molecules/details', + headers: { 'Content-Type': 'text/plain' }, + }); + const answered = new Promise((resolve, reject) => { + sent.on('response', (response) => { + let body = ''; + response.on('data', (chunk) => (body += chunk)); + response.on('end', () => resolve({ status: response.statusCode, response, body })); + }); + sent.on('error', reject); + }); + sent.end(accessions.join(',')); + return { answered, drop: () => sent.destroy() }; +} + +const until = async (condition) => { + for (let i = 0; i < 200 && !condition(); i += 1) await new Promise((r) => setTimeout(r, 10)); + expect(condition()).toBe(true); +}; +const release = () => { + for (const go of held.splice(0)) go(); +}; + +describe('the molecule details batch', () => { + it('runs at most four at once; a fifth is the 503 to try again shortly', async () => { + const running = [1, 2, 3, 4].map((n) => post([`P0000${n}`])); + await until(() => held.length === 4); + const fifth = await post(['P00005']).answered; + expect(fifth.status).toBe(503); + expect(fifth.response.headers['retry-after']).toBe('5'); + expect(JSON.parse(fifth.body).messages).toEqual(['The service is busy; try again shortly']); + release(); + // None found for any is Java's 404, and each batch gives its place back. + for (const { answered } of running) expect((await answered).status).toBe(404); + const after = post(['P00006']); + await until(() => held.length === 1); + release(); + expect((await after.answered).status).toBe(404); + }); + + it('stops asking once the client is gone, and gives its place back', async () => { + asked = 0; + const { answered, drop } = post(['P1', 'P2', 'P3', 'P4', 'P5']); + answered.catch(() => {}); + await until(() => held.length === 1); + drop(); + // The server has seen the close before the first lookup returns. + const open = () => + new Promise((resolve) => server.getConnections((_error, count) => resolve(count))); + let connections = 1; + for (let i = 0; i < 200 && connections > 0; i += 1) { + connections = await open(); + if (connections > 0) await new Promise((r) => setTimeout(r, 10)); + } + expect(connections).toBe(0); + release(); + await new Promise((r) => setTimeout(r, 20)); + expect(asked).toBe(1); + // All four places are free again: four run, and a fifth is refused. A + // dropped batch that kept its place would leave three. + const four = [1, 2, 3, 4].map((n) => post([`Q0000${n}`])); + await until(() => held.length === 4); + expect((await post(['Q00005']).answered).status).toBe(503); + release(); + for (const { answered: each } of four) expect((await each).status).toBe(404); + }); +});