feat: queuing of calls to python process

.changeset/gentle-clouds-bet.md

@@ -0,0 +1,5 @@
+---
+"apollo": patch
+---
+
+Queue requests to the python process

.env.example

@@ -45,6 +45,16 @@
 # See services/name_rules.py.
 APOLLO_UNICODE_STEP_NAMES=false
 
+# Max concurrent service calls bridge.ts runs at once, per Apollo process
+# (python or TypeScript - not python-specific). Calls past this cap queue in
+# FIFO order rather than running immediately. Defaults to 10.
+# See platform/src/util/python-queue.ts.
+# APOLLO_CONCURRENCY=10
+
+# How long a queued call waits for a free slot before giving up (ms).
+# Defaults to 5 minutes.
+# APOLLO_QUEUE_TIMEOUT_MS=300000
+
 ANTHROPIC_API_KEY=sk-YOUR-API-KEY-HERE
 
 OPENAI_API_KEY=sk-YOUR-API-KEY-HERE

README.md

@@ -65,7 +65,12 @@ bun dev
 
 To see an index of the available language services, head to `localhost:3000`.
 
-The chat services search an index of the OpenFn documentation, which every environment populates itself. Two backends are supported: Pinecone (the default) and Postgres with the `pgvector` extension (opt-in, see [Database](#database)). Run `bun py embed_docsite` to build yours; see [embed_docsite](services/embed_docsite/README.md) for the write-target options and [search_docsite](services/search_docsite/README.md) for backend selection.
+The chat services search an index of the OpenFn documentation, which every
+environment populates itself. Two backends are supported: Pinecone (the default)
+and Postgres with the `pgvector` extension (opt-in, see [Database](#database)).
+Run `bun py embed_docsite` to build yours; see
+[embed_docsite](services/embed_docsite/README.md) for the write-target options
+and [search_docsite](services/search_docsite/README.md) for backend selection.
 
 ## Python Setup
 
@@ -155,10 +160,10 @@ Controls which characters a workflow step name may contain. Apollo sanitises
 step names on the way out and Lightning validates them on the way in, so the two
 rules have to agree. The default is `false`.
 
-| Value | Rule |
-| --- | --- |
-| `false` (default) | ASCII only. Letters, digits, spaces, hyphens, underscores. Accents are folded (`Café` becomes `Cafe`) and anything else is dropped. This is the rule Lightning enforces today. |
-| `true` | Anything except control characters. Letters and marks from any script, all punctuation and symbols, emoji, `/`, `:`, `>`, `&`, quotes and apostrophes. `Vérifier l'état` and `患者確認` survive exactly as typed. |
+| Value             | Rule                                                                                                                                                                                                              |
+| ----------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `false` (default) | ASCII only. Letters, digits, spaces, hyphens, underscores. Accents are folded (`Café` becomes `Cafe`) and anything else is dropped. This is the rule Lightning enforces today.                                    |
+| `true`            | Anything except control characters. Letters and marks from any script, all punctuation and symbols, emoji, `/`, `:`, `>`, `&`, quotes and apostrophes. `Vérifier l'état` and `患者確認` survive exactly as typed. |
 
 Leave it off until Lightning ships Unicode step names (Lightning#4577), then
 turn it on. Turning it on first means Apollo emits names Lightning rejects.
@@ -170,18 +175,18 @@ Lightning is the worse of the two failures: Lightning rejecting a name is loud
 and recoverable, whereas Apollo quietly renaming a valid name is the silent
 vandalism this flag exists to prevent.
 
-The rejected control set is the same in both modes: C0 (`U+0000`-`U+001F`,
-NUL included), DEL (`U+007F`), C1 (`U+0080`-`U+009F`), the noncharacters
-`U+FFFE` and `U+FFFF`, the surrogates `U+D800`-`U+DFFF`, and the line and
-paragraph separators `U+2028` and `U+2029`. A NUL byte in a name crashes the Postgres insert on
-Lightning's side. Names are NFC-normalised in both modes so that Apollo and
-Lightning agree on how to spell an accent, which is what step lookup matches
-on, and capped at 100 graphemes because that is what Ecto's `validate_length`
-counts.
+The rejected control set is the same in both modes: C0 (`U+0000`-`U+001F`, NUL
+included), DEL (`U+007F`), C1 (`U+0080`-`U+009F`), the noncharacters `U+FFFE`
+and `U+FFFF`, the surrogates `U+D800`-`U+DFFF`, and the line and paragraph
+separators `U+2028` and `U+2029`. A NUL byte in a name crashes the Postgres
+insert on Lightning's side. Names are NFC-normalised in both modes so that
+Apollo and Lightning agree on how to spell an accent, which is what step lookup
+matches on, and capped at 100 graphemes because that is what Ecto's
+`validate_length` counts.
 
 The rule lives in `services/name_rules.py`, and everything that states or
-enforces it is derived from there: the sanitiser, the workflow-generation
-prompt (`describe_rule_for_prompt`), the acceptance-test judges
+enforces it is derived from there: the sanitiser, the workflow-generation prompt
+(`describe_rule_for_prompt`), the acceptance-test judges
 (`describe_rule_for_judge`, substituted into the rubric markdown by
 `judges.load_judge`), and the `assert_no_special_chars` test assertion. Change
 the rule in that one file and all four follow.
@@ -195,15 +200,15 @@ an emoji ZWJ run at the joiner unless a pictograph follows) and the whole point
 is to agree with Elixir, not with the spec.
 
 `tools/unicode_parity` is the harness that checks it. Run `python3 edges.py`,
-then `elixir probe.exs`, then `python3 check.py` with the Elixir version Lightning runs; `--tables`
-prints the literals to paste back into `name_rules`. It checks five things:
-every codepoint's break class, the `Extended_Pictographic` set, the trim set,
-what a GB11 emoji run may be separated from its joiner by, and cluster
-boundaries over a generated corpus. Normalisation is not among them:
-`normalize_nfc` is the standard library's, so there is no table of ours to
-check against Elixir.
-
-`Extended_Pictographic` needs its own check because it is *not* a break class,
+then `elixir probe.exs`, then `python3 check.py` with the Elixir version
+Lightning runs; `--tables` prints the literals to paste back into `name_rules`.
+It checks five things: every codepoint's break class, the
+`Extended_Pictographic` set, the trim set, what a GB11 emoji run may be
+separated from its joiner by, and cluster boundaries over a generated corpus.
+Normalisation is not among them: `normalize_nfc` is the standard library's, so
+there is no table of ours to check against Elixir.
+
+`Extended_Pictographic` needs its own check because it is _not_ a break class,
 so a per-codepoint sweep cannot see it — an over-broad set there silently
 changes clustering either side of a ZWJ and nothing else notices. That is
 exactly how a hand-written table with 531 wrong codepoints survived two rounds
@@ -250,6 +255,29 @@ service:
 curl http://localhost:3000/services/status
 ```
 
+### Capacity testing
+
+`tools/capacity-test.js` fires parallel requests at the `echo` service of a
+running server and logs each request's number, duration and response code,
+followed by the total time and the count and percentage of requests that
+succeeded (a 2xx response). It sends 100 requests by default; pass a number as
+the first argument to change that:
+
+```bash
+bun run test:capacity
+bun run test:capacity 500
+```
+
+It reads `APOLLO_ENDPOINT` (default `http://localhost:3000`) and sends
+`AI_ASSISTANT_API_KEY` as the `api_key` in each payload. Requests are not
+throttled, and log lines appear in completion order rather than request order.
+
+To restrict memory usage locally, launch with Docker:
+
+```
+docker run --memory=512m --memory-swap=512m -p 3000:3000 --network host --env-file .env apollo
+```
+
 ## Troubleshooting
 
 If you get errors like `poetry: command not found` (error code 127), and poetry
@@ -305,9 +333,8 @@ them is safe:
     `load_adaptor_docs` runs, so applying it by hand is optional.
 - `lightning_clients` - created and kept current by the migration runner
   (`platform/src/db/migrate.ts`, migrations under
-  [`migrations/clients/`](migrations/clients/)). It is applied
-  automatically at Apollo startup when a database URL is set; no manual `psql`
-  step is needed.
+  [`migrations/clients/`](migrations/clients/)). It is applied automatically at
+  Apollo startup when a database URL is set; no manual `psql` step is needed.
 
 First, make sure you've configured your desired `POSTGRES_URL` in your `.env`
 file.

package.json

@@ -8,6 +8,7 @@
     "dev": "bun  --watch platform/src/index.ts",
     "build": "bun build platform/src/index.ts",
     "test": "bun test platform/test ",
+    "test:capacity": "bun tools/capacity-test.js",
     "client": "bun platform/src/auth/client/cli.ts",
     "migrate": "bun platform/src/db/migrate.ts",
     "py": "poetry run python services/entry.py "

platform/src/bridge.ts

@@ -6,13 +6,15 @@ import { getInternalToken } from "./auth/internal-token";
 import {
   emptyResult,
   malformedResult,
+  queueTimeout,
   subprocessCancelled,
   subprocessFailed,
   subprocessKilled,
   subprocessSpawnFailed,
 } from "./util/errors";
+import { enqueue, QueueTimeoutError } from "./util/python-queue";
 import pkg from "../../package.json";
-import { USE_FORK_SERVER, runForked } from "./bridge.fork";
+import { USE_FORK_SERVER, runForked as runWithFork } from "./bridge.fork";
 import { LOG_LINE } from "./util/log-line";
 
 /**
@@ -38,6 +40,10 @@ export const run = async (
   const inputPath = tmpfile.replace("{}", "input");
   const outputPath = tmpfile.replace("{}", "output");
 
+  // Cleanup shared by every path below that never spawns.
+  const removeTmpFiles = () =>
+    Promise.all([rm(inputPath).catch(() => {}), rm(outputPath).catch(() => {})]);
+
   // Outside the promise, deliberately. The Promise constructor only catches a
   // synchronous throw from its executor, so an await that rejects in there -
   // a full disk, a read-only tmp - leaves the promise pending for ever and
@@ -52,17 +58,48 @@ export const run = async (
 
     await Bun.write(outputPath, "");
   } catch (error) {
-    // Removed rather than left behind by a half-finished setup, for the same
-    // reason it is 0600 above.
-    await rm(inputPath).catch(() => {});
-    await rm(outputPath).catch(() => {});
+    await removeTmpFiles();
     throw subprocessSpawnFailed(scriptName, error);
   }
 
-  if (USE_FORK_SERVER) {
-    return runForked(scriptName, port, inputPath, outputPath, onLog, onEvent, signal);
+  // One cap for both paths below (see util/python-queue.ts): forking is cheap
+  // to start, but the work it does is just as heavy as a fresh interpreter.
+  const runFn = USE_FORK_SERVER ? runWithFork : runWithSpawn;
+
+  let started = false;
+  const { promise, cancel } = enqueue(scriptName, () => {
+    started = true;
+    return runFn(scriptName, port, inputPath, outputPath, onLog, onEvent, signal);
+  });
+
+  if (signal) {
+    if (signal.aborted) cancel();
+    else signal.addEventListener("abort", cancel, { once: true });
+  }
+
+  try {
+    return await promise;
+  } catch (error) {
+    if (!started) {
+      await removeTmpFiles();
+      if (error instanceof QueueTimeoutError) {
+        throw queueTimeout(scriptName);
+      }
+      throw subprocessCancelled(scriptName, "QUEUED");
+    }
+    throw error;
   }
+};
 
+const runWithSpawn = (
+  scriptName: string,
+  port: number,
+  inputPath: string,
+  outputPath: string,
+  onLog?: (str: string) => void,
+  onEvent?: (type: string, payload: any) => void,
+  signal?: AbortSignal
+): Promise<JSON | null> => {
   return new Promise<JSON | null>((resolve, reject) => {
     const proc = spawn(
       "poetry",

platform/src/util/errors.ts

@@ -94,6 +94,15 @@ export function subprocessCancelled(
   );
 }
 
+export function queueTimeout(service: string): ApolloThrowable {
+  return new ApolloThrowable(
+    503,
+    "QUEUE_TIMEOUT",
+    `Service "${service}" timed out waiting for a free slot`,
+    { service }
+  );
+}
+
 /** Something outside Apollo killed the service - most often the OOM killer.
  *  Distinct from a cancellation, which is us, and from a non-zero exit, which
  *  is the service deciding to stop. */

platform/src/util/python-queue.ts

@@ -0,0 +1,99 @@
+// FIFO queue capping how many service calls bridge.ts runs at once.
+// Calls past the cap wait here instead of running immediately.
+
+import { captureMessage } from "./sentry";
+
+const DEFAULT_CONCURRENCY = 10;
+
+export const maxConcurrency = (): number =>
+  Number(process.env.APOLLO_CONCURRENCY) || DEFAULT_CONCURRENCY;
+
+const DEFAULT_QUEUE_TIMEOUT_MS = 5 * 60 * 1000;
+
+export const queueTimeoutMs = (): number =>
+  Number(process.env.APOLLO_QUEUE_TIMEOUT_MS) || DEFAULT_QUEUE_TIMEOUT_MS;
+
+let active = 0;
+const pending: Array<() => void> = [];
+
+export const activeCount = (): number => active;
+export const queueDepth = (): number => pending.length;
+
+// Distinct from cancel()'s rejection, so a caller can tell "the server gave
+// up" apart from "the client went away".
+export class QueueTimeoutError extends Error {
+  constructor() {
+    super("timed out waiting in the queue");
+    this.name = "QueueTimeoutError";
+  }
+}
+
+export type Enqueued<T> = {
+  promise: Promise<T>;
+  cancel: () => void;
+};
+
+// cancel() drops a still-queued task without ever running it; calling it
+// once the task has started is a no-op.
+export const enqueue = <T>(
+  scriptName: string,
+  task: () => Promise<T>
+): Enqueued<T> => {
+  let removeFromQueue: ((reason: Error) => void) | null = null;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+
+  const promise = new Promise<T>((resolve, reject) => {
+    const run = () => {
+      if (timer) clearTimeout(timer);
+      active++;
+      task().then(
+        (value) => {
+          active--;
+          pending.shift()?.();
+          resolve(value);
+        },
+        (error) => {
+          active--;
+          pending.shift()?.();
+          reject(error);
+        }
+      );
+    };
+
+    if (active < maxConcurrency()) {
+      run();
+      return;
+    }
+
+    console.log(
+      `queueing ${scriptName}: ${active}/${maxConcurrency()} running, ${
+        pending.length + 1
+      } waiting`
+    );
+    captureMessage(`service call queued: ${scriptName}`, {
+      scriptName,
+      active,
+      queueDepth: pending.length + 1,
+    });
+
+    pending.push(run);
+    timer = setTimeout(
+      () => removeFromQueue?.(new QueueTimeoutError()),
+      queueTimeoutMs()
+    );
+    timer.unref?.();
+
+    removeFromQueue = (reason: Error) => {
+      const i = pending.indexOf(run);
+      if (i === -1) return; // already started - too late to cancel
+      pending.splice(i, 1);
+      if (timer) clearTimeout(timer);
+      reject(reason);
+    };
+  });
+
+  return {
+    promise,
+    cancel: () => removeFromQueue?.(new Error("queue wait cancelled")),
+  };
+};

platform/src/util/sentry.ts

@@ -46,3 +46,13 @@ export const captureException = (
   if (!enabled) return;
   Sentry.captureException(err, extras ? { extra: extras } : undefined);
 };
+
+// Same no-op-when-disabled contract as captureException, for events that
+// aren't an error.
+export const captureMessage = (
+  message: string,
+  extras?: Record<string, unknown>
+): void => {
+  if (!enabled) return;
+  Sentry.captureMessage(message, extras ? { extra: extras } : undefined);
+};