From 5512f825eae5b5f93030b66bfc5a3bf463632350 Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Tue, 20 Oct 2026 19:02:47 +0000 Subject: [PATCH 1/2] Analysis gateway: a share per client when places are scarce From the review of everything merged since #485 (spec 010 Story 6, T065). The lanes' places and waiting list were first come, first served across all clients, so one address keeping a lane and its waiting list full had everyone else refused. Now a client has at most PER_CLIENT (default 8) computations running or waiting, in all lanes; beyond that, one that would have to wait is refused, while one that finds a place free still runs -- a class or an institution behind one address is not refused by an idle service. A client is the address nginx saw (X-Real-IP, which nginx sets from the connection; the gateway listens on loopback, so only nginx can send it), and an IPv6 client its /64, any address in which it can choose. /health reports the share and how many clients are inside. Each rule shown failing its spec when mutated away: the share, the share held back while places are free, and the place given back on a timeout and on leaving. Co-Authored-By: Claude Opus 5.5 --- docker-compose.yml | 1 + specs/010-java-services/tasks.md | 1 + tools/analysis-gateway/gateway.spec.ts | 136 ++++++++++++++++++++++++- tools/analysis-gateway/gateway.ts | 75 ++++++++++++-- 4 files changed, 206 insertions(+), 7 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 99aacacc..afbaa49e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -250,6 +250,7 @@ services: - ANALYSIS_LIMIT=${ANALYSIS_LIMIT:-4} - REPORT_LIMIT=${REPORT_LIMIT:-1} - QUEUE_LENGTH=${QUEUE_LENGTH:-40} + - PER_CLIENT=${PER_CLIENT:-8} - MAX_WAIT_MS=${MAX_WAIT_MS:-60000} - RETRY_AFTER=${RETRY_AFTER:-10} - UPSTREAM_TIMEOUT_MS=${UPSTREAM_TIMEOUT_MS:-300000} diff --git a/specs/010-java-services/tasks.md b/specs/010-java-services/tasks.md index 6e10e9e7..3bf5a4a4 100644 --- a/specs/010-java-services/tasks.md +++ b/specs/010-java-services/tasks.md @@ -221,6 +221,7 @@ Each cutover is one PR. It adds the group's patterns to node's table, one nginx - [x] T062 [US6] Measure on this host, with the site's example data and nothing sent outside: the burst of research R12 through the gateway, and a stored result read during it. Record the timings in R12. — #539 - [x] T063 [US6] Have the site say "the service is busy, trying again" on a 503 with Retry-After, and retry once after the time it names. Test it in the site's analysis service, failing first. — #539 - [x] T064 [US6] Review, CI, merge, deploy to beta. Check with `curl --resolve` that an analysis, a stored result and a download pass through unchanged, and that `/health` shows the queue. — #539, deployed 19 Oct 2026 after two reviews: an analysis took a lane place; a stored result, its not-found list and both downloads (CSV and JSON) were byte-identical through the gateway and direct, and `/health` showed the lanes and the counts (1 admitted, 4 passed). Ticked in #540, after the merge. +- [ ] T065 [US6] A share per client (review of 20 Oct): with places scarce, one address keeping a lane and its waiting list full had everyone else refused. A client -- nginx's X-Real-IP, an IPv6 one by its /64 -- has at most `PER_CLIENT` (8) computations running or waiting; beyond that, one that would have to wait is refused, while one that finds a place free still runs, so a class behind one address is not refused by an idle service. Each rule shown failing by mutation. --- diff --git a/tools/analysis-gateway/gateway.spec.ts b/tools/analysis-gateway/gateway.spec.ts index 72d169cc..5f0c776c 100644 --- a/tools/analysis-gateway/gateway.spec.ts +++ b/tools/analysis-gateway/gateway.spec.ts @@ -7,7 +7,7 @@ import { createServer, request as httpRequest, type Server } from 'node:http'; import type { AddressInfo } from 'node:net'; import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; -import { classify, gateway, type GatewaySettings, type Health } from './gateway.ts'; +import { classify, clientOf, gateway, type GatewaySettings, type Health } from './gateway.ts'; const seen: string[] = []; /** Every request the service was sent, recorded as it arrives, body or not. */ @@ -668,3 +668,137 @@ describe('the second review of #539', () => { } }); }); + +describe('the review of 20 Oct', () => { + const from = (address: string, header?: string) => + clientOf({ + headers: header === undefined ? {} : { 'x-real-ip': header }, + socket: { remoteAddress: address }, + } as unknown as Parameters[0]); + + it("knows a client by nginx's X-Real-IP, an IPv6 one by its /64", () => { + expect(from('127.0.0.1', '203.0.113.9')).toBe('203.0.113.9'); + expect(from('127.0.0.1')).toBe('127.0.0.1'); + expect(from('::ffff:127.0.0.1')).toBe('::ffff:127.0.0.1'); + // One /64, however written: every address in it is one client's. + expect( + new Set([ + from('127.0.0.1', '2001:db8:0:1::1'), + from('127.0.0.1', '2001:DB8:0000:0001:ffff:1:2:3'), + from('127.0.0.1', '2001:db8:0:1:0:0:0:9'), + ]) + ).toEqual(new Set(['2001:db8:0:1::/64'])); + expect(from('127.0.0.1', '2001:db8:0:2::1')).toBe('2001:db8:0:2::/64'); + expect(from('127.0.0.1', '2001:db8::')).toBe('2001:db8:0:0::/64'); + }); + + const as = (base: string, who: string, n: number, signal?: AbortSignal) => + fetch(`${base}/AnalysisService/identifiers/?n=${n}`, { + method: 'POST', + body: 'P04637', + headers: { 'Content-Type': 'text/plain', 'X-Real-IP': who }, + ...(signal && { signal }), + }); + + it('refuses a client beyond its share when it would wait, while others still get in', async () => { + const { base, close } = await start({ + limits: { analysis: 2, report: 1 }, + queueLength: 10, + perClient: 2, + // Long enough that only the share can refuse the third, never its wait. + maxWaitMs: 30_000, + }); + try { + const first = [as(base, '203.0.113.1', 1), as(base, '203.0.113.1', 2)]; + await until(() => held.length === 2); + // A third from the same client, with the lane full: refused at once, + // though the waiting list has room. + const started = Date.now(); + const third = await as(base, '203.0.113.1', 3); + expect(Date.now() - started).toBeLessThan(1000); + expect([third.status, third.headers.get('retry-after')]).toEqual([503, '7']); + // Another client waits its turn as before. + const other = as(base, '203.0.113.2', 4); + for (let i = 0; i < 200 && (await health(base)).lanes.analysis.waiting < 1; i++) + await new Promise((r) => setTimeout(r, 10)); + expect((await health(base)).clients).toBe(2); + letGo(); + await until(() => held.length === 2); + // One of the first client's done: it has a place again. + const again = as(base, '203.0.113.1', 5); + letGo(); + letGo(); + await until(() => held.length === 1); + letGo(); + expect((await Promise.all([...first, other, again])).map((r) => r.status)).toEqual([ + 200, 200, 200, 200, + ]); + expect(arrived).not.toContain('/AnalysisService/identifiers/?n=3'); + } finally { + await close(); + } + }); + + it('runs a client beyond its share while places are free: the share is for scarce places', async () => { + const { base, close } = await start({ + limits: { analysis: 3, report: 1 }, + perClient: 1, + }); + try { + const two = [as(base, '203.0.113.1', 1), as(base, '203.0.113.1', 2)]; + await until(() => held.length === 2); + letGo(); + letGo(); + expect((await Promise.all(two)).map((r) => r.status)).toEqual([200, 200]); + } finally { + await close(); + } + }); + + it('gives a waiting client its place back when it times out', async () => { + const { base, close } = await start({ + limits: { analysis: 1, report: 1 }, + queueLength: 10, + perClient: 2, + maxWaitMs: 200, + }); + try { + const running = as(base, '203.0.113.1', 1); + await until(() => held.length === 1); + expect((await as(base, '203.0.113.1', 2)).status).toBe(503); + expect((await health(base)).clients).toBe(1); + letGo(); + expect((await running).status).toBe(200); + expect((await health(base)).clients).toBe(0); + } finally { + await close(); + } + }); + + it('gives a waiting client its place back when it leaves', async () => { + const { base, close } = await start({ + limits: { analysis: 1, report: 1 }, + queueLength: 10, + perClient: 2, + maxWaitMs: 30_000, + }); + try { + const running = as(base, '203.0.113.1', 1); + await until(() => held.length === 1); + const leaving = new AbortController(); + const waited = as(base, '203.0.113.1', 2, leaving.signal).catch(() => undefined); + for (let i = 0; i < 200 && (await health(base)).lanes.analysis.waiting < 1; i++) + await new Promise((r) => setTimeout(r, 10)); + leaving.abort(); + await waited; + for (let i = 0; i < 200 && (await health(base)).lanes.analysis.waiting > 0; i++) + await new Promise((r) => setTimeout(r, 10)); + expect((await health(base)).clients).toBe(1); + letGo(); + expect((await running).status).toBe(200); + expect((await health(base)).clients).toBe(0); + } finally { + await close(); + } + }); +}); diff --git a/tools/analysis-gateway/gateway.ts b/tools/analysis-gateway/gateway.ts index 3e66c429..e4f5316d 100644 --- a/tools/analysis-gateway/gateway.ts +++ b/tools/analysis-gateway/gateway.ts @@ -19,6 +19,9 @@ * ANALYSIS_LIMIT analyses at once, default 4 (below analysis-core's pool of 5) * REPORT_LIMIT PDF reports at once, default 1 * QUEUE_LENGTH waiting per lane before refusing, default 40 + * PER_CLIENT one client's computations running or waiting at once, in + * all lanes, default 8: beyond it, one that would have to + * wait is refused (with places free, it still runs) * MAX_WAIT_MS longest wait, default 60000 (wait + run inside the edge's 100 s) * RETRY_AFTER seconds a refused reader is told to wait, default 10 * WAITING_BYTES bodies held while waiting, in all, default 268435456 (256 MB) @@ -49,6 +52,8 @@ interface Waiting { req: IncomingMessage; res: ServerResponse; body: HeldBody; + /** Gives its client's place back; called once, when it is done here. */ + out: () => void; /** Stops reading into `body` and gives its bytes back to the budget. */ settle: () => void; timer: NodeJS.Timeout; @@ -67,6 +72,9 @@ export interface Health { lanes: Record; queueLength: number; maxWaitMs: number; + perClient: number; + /** Clients with a computation running or waiting. */ + clients: number; heldBytes: number; passed: number; admitted: number; @@ -81,6 +89,7 @@ export interface GatewaySettings { limits?: Record; queueLength?: number; maxWaitMs?: number; + perClient?: number; retryAfter?: number; waitingBytes?: number; upstreamTimeoutMs?: number; @@ -144,6 +153,28 @@ const without = (raw: string[]): string[] => { return kept; }; +/** + * Who a request is from, for the per-client share: the address nginx saw + * (X-Real-IP, which it sets from the connection; the gateway listens on + * loopback, so only nginx can send it), else the connection's own. An IPv6 + * client holds a /64 and can choose any address in it, so it is that /64. + */ +export function clientOf(req: IncomingMessage): string { + const header = req.headers['x-real-ip']; + const address = (typeof header === 'string' && header.trim()) || req.socket.remoteAddress || ''; + if (!address.includes(':') || address.startsWith('::ffff:')) return address; + const [head = '', tail = ''] = address.split('::'); + const left = head ? head.split(':') : []; + const right = tail ? tail.split(':') : []; + const groups = address.includes('::') + ? [...left, ...Array(Math.max(0, 8 - left.length - right.length)).fill('0'), ...right] + : left; + return `${groups + .slice(0, 4) + .map((group) => group.toLowerCase().replace(/^0+(?=.)/, '')) + .join(':')}::/64`; +} + /** The service's own error envelope, as it writes a 404 or a 410. */ function busy(res: ServerResponse, retryAfter: number): void { if (res.headersSent || res.destroyed) return; @@ -165,6 +196,7 @@ export function gateway({ limits = { analysis: 4, report: 1 }, queueLength = 40, maxWaitMs = 60_000, + perClient = 8, retryAfter = 10, waitingBytes = 256 * 1024 * 1024, upstreamTimeoutMs = 300_000, @@ -176,6 +208,20 @@ export function gateway({ }; const counts = { passed: 0, admitted: 0, waited: 0, refused: 0, left: 0, failed: 0 }; let held = 0; + /** Each client's computations running or waiting, in all lanes. */ + const inside = new Map(); + /** A place for `client`, and the call that gives it back (once). */ + function enter(client: string): () => void { + inside.set(client, (inside.get(client) ?? 0) + 1); + let given = false; + return () => { + if (given) return; + given = true; + const left = (inside.get(client) ?? 1) - 1; + if (left > 0) inside.set(client, left); + else inside.delete(client); + }; + } /** * Send the request on, and its answer back as it came; `done` once the @@ -258,12 +304,19 @@ export function gateway({ } else req.pipe(call); } - function admit(lane: Lane, req: IncomingMessage, res: ServerResponse, body?: HeldBody): void { + function admit( + lane: Lane, + req: IncomingMessage, + res: ServerResponse, + out: () => void, + body?: HeldBody + ): void { lane.active++; counts.admitted++; pass(req, res, { ...(body && { body }), done: () => { + out(); lane.active--; next(lane); }, @@ -275,11 +328,11 @@ export function gateway({ const entry = lane.waiting.shift(); if (!entry) return; entry.settle(); - admit(lane, entry.req, entry.res, entry.body); + admit(lane, entry.req, entry.res, entry.out, entry.body); } } - function wait(lane: Lane, req: IncomingMessage, res: ServerResponse): void { + function wait(lane: Lane, req: IncomingMessage, res: ServerResponse, out: () => void): void { counts.waited++; // Read while waiting, so a reader who leaves is seen leaving: with the // body unread, a closed connection goes unnoticed and the analysis runs @@ -309,6 +362,7 @@ export function gateway({ req, res, body, + out, settle: release, timer: setTimeout(() => { if (leave()) { @@ -322,6 +376,7 @@ export function gateway({ if (at === -1) return false; lane.waiting.splice(at, 1); release(); + out(); return true; }; // A reader who gives up while waiting loses their place; nothing is run. @@ -347,6 +402,8 @@ export function gateway({ lanes: { analysis: state(lanes.analysis), report: state(lanes.report) }, queueLength, maxWaitMs, + perClient, + clients: inside.size, heldBytes: held, ...counts, }; @@ -360,16 +417,21 @@ export function gateway({ pass(req, res); return; } + const client = clientOf(req); if (lane.active < lane.limit) { - admit(lane, req, res); + admit(lane, req, res, enter(client)); return; } - if (lane.waiting.length >= queueLength) { + // No one client takes every place: without a share, one address keeping + // the lane and its waiting list full had everyone else refused. Only + // when places are scarce -- a class behind one address, with the service + // otherwise idle, still runs. + if ((inside.get(client) ?? 0) >= perClient || lane.waiting.length >= queueLength) { counts.refused++; busy(res, retryAfter); return; } - wait(lane, req, res); + wait(lane, req, res, enter(client)); }); return server; } @@ -399,6 +461,7 @@ if (startedDirectly()) { limits: { analysis: number('ANALYSIS_LIMIT', 4), report: number('REPORT_LIMIT', 1) }, queueLength: number('QUEUE_LENGTH', 40), maxWaitMs: number('MAX_WAIT_MS', 60_000), + perClient: number('PER_CLIENT', 8), retryAfter: number('RETRY_AFTER', 10), waitingBytes: number('WAITING_BYTES', 256 * 1024 * 1024), upstreamTimeoutMs: number('UPSTREAM_TIMEOUT_MS', 300_000), From ddb56431dac686ab6365e4c8660bd057ce2cbc21 Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Tue, 20 Oct 2026 19:42:48 +0000 Subject: [PATCH 2/2] Spec 010: tick T065 (#574) Co-Authored-By: Claude Opus 5.5 --- specs/010-java-services/tasks.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/specs/010-java-services/tasks.md b/specs/010-java-services/tasks.md index 3bf5a4a4..9d6b605e 100644 --- a/specs/010-java-services/tasks.md +++ b/specs/010-java-services/tasks.md @@ -221,7 +221,7 @@ Each cutover is one PR. It adds the group's patterns to node's table, one nginx - [x] T062 [US6] Measure on this host, with the site's example data and nothing sent outside: the burst of research R12 through the gateway, and a stored result read during it. Record the timings in R12. — #539 - [x] T063 [US6] Have the site say "the service is busy, trying again" on a 503 with Retry-After, and retry once after the time it names. Test it in the site's analysis service, failing first. — #539 - [x] T064 [US6] Review, CI, merge, deploy to beta. Check with `curl --resolve` that an analysis, a stored result and a download pass through unchanged, and that `/health` shows the queue. — #539, deployed 19 Oct 2026 after two reviews: an analysis took a lane place; a stored result, its not-found list and both downloads (CSV and JSON) were byte-identical through the gateway and direct, and `/health` showed the lanes and the counts (1 admitted, 4 passed). Ticked in #540, after the merge. -- [ ] T065 [US6] A share per client (review of 20 Oct): with places scarce, one address keeping a lane and its waiting list full had everyone else refused. A client -- nginx's X-Real-IP, an IPv6 one by its /64 -- has at most `PER_CLIENT` (8) computations running or waiting; beyond that, one that would have to wait is refused, while one that finds a place free still runs, so a class behind one address is not refused by an idle service. Each rule shown failing by mutation. +- [x] T065 [US6] A share per client (review of 20 Oct): with places scarce, one address keeping a lane and its waiting list full had everyone else refused. A client -- nginx's X-Real-IP, an IPv6 one by its /64 -- has at most `PER_CLIENT` (8) computations running or waiting; beyond that, one that would have to wait is refused, while one that finds a place free still runs, so a class behind one address is not refused by an idle service. Each rule shown failing by mutation. — #574 ---