Analysis gateway: a share per client when places are scarce

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}

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.
+- [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
 
 ---
 

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<typeof clientOf>[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();
+    }
+  });
+});

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<LaneName, { limit: number; active: number; waiting: number }>;
   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<LaneName, number>;
   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<string>(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<string, number>();
+  /** 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),