From 7644450c6fae937a76333908c30be4da22a1ba12 Mon Sep 17 00:00:00 2001 From: doc-han Date: Fri, 2 Oct 2026 09:58:32 +0000 Subject: [PATCH 1/9] feat: forked python process --- platform/src/bridge.ts | 200 +++++++++++++++++++++++++++++++++++++++- services/entry.py | 69 ++++++++------ services/fork_server.py | 109 ++++++++++++++++++++++ 3 files changed, 347 insertions(+), 31 deletions(-) create mode 100644 services/fork_server.py diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index 240e9ade..edceebec 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -1,6 +1,7 @@ import readline from "node:readline"; +import net from "node:net"; import path from "node:path"; -import { spawn } from "node:child_process"; +import { spawn, type ChildProcess } from "node:child_process"; import { chmod, rm } from "node:fs/promises"; import { getInternalToken } from "./auth/internal-token"; import { @@ -17,6 +18,199 @@ import pkg from "../../package.json"; // stream. Only these are forwarded to the caller. const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/; +const USE_FORK_SERVER = ["1", "true", "yes"].includes( + (process.env.APOLLO_FORK_SERVER ?? "").toLowerCase() +); + +interface ForkMaster { + proc: ChildProcess; + socketPath: string; +} +let forkMaster: Promise | null = null; +let forkMasterProc: ChildProcess | null = null; + +function startForkMaster(): Promise { + const socketPath = path.resolve(`tmp/fork_server-${process.pid}.sock`); + const proc = spawn( + "poetry", + ["run", "python", "services/fork_server.py", "--socket", socketPath], + { + env: { + ...process.env, + APOLLO_INTERNAL_TOKEN: getInternalToken(), + APOLLO_VERSION: pkg.version, + OBJC_DISABLE_INITIALIZE_FORK_SAFETY: "YES", + }, + } + ); + forkMasterProc = proc; + + return new Promise((resolve, reject) => { + const timeout = setTimeout( + () => reject(subprocessSpawnFailed("fork_server", new Error("not READY in time"))), + 60_000 + ); + readline.createInterface({ input: proc.stdout!, crlfDelay: Infinity }).on("line", (line) => { + console.log(`fork-server: ${line}`); + if (line.startsWith("READY")) { + clearTimeout(timeout); + resolve({ proc, socketPath }); + } + }); + readline + .createInterface({ input: proc.stderr!, crlfDelay: Infinity }) + .on("line", (line) => console.error(`fork-server: ${line}`)); + + proc.on("error", (err) => { + clearTimeout(timeout); + reject(subprocessSpawnFailed("fork_server", err)); + }); + proc.on("exit", (code, sig) => { + clearTimeout(timeout); + console.error(`fork-server exited (code=${code} signal=${sig})`); + forkMaster = null; // next request starts a fresh master + forkMasterProc = null; + rm(socketPath).catch(() => {}); + }); + }); +} + +function getForkMaster(): Promise { + if (!forkMaster) { + forkMaster = startForkMaster().catch((err) => { + forkMaster = null; + throw err; + }); + } + return forkMaster; +} + +process.once("exit", () => forkMasterProc?.kill("SIGTERM")); + +function runForked( + scriptName: string, + port: number, + inputPath: string, + outputPath: string, + onLog?: (str: string) => void, + onEvent?: (type: string, payload: any) => void, + signal?: AbortSignal +): Promise { + return getForkMaster().then( + (master) => + new Promise((resolve, reject) => { + let childPid: number | null = null; + let exitCode: number | null = null; + let cancelled = false; + let hardKill: ReturnType | undefined; + + const cleanupFiles = async () => { + try { + await rm(inputPath); + await rm(outputPath); + } catch (e) { + console.error("Error removing temporary files"); + console.error(e); + } + }; + + const killChild = () => { + if (childPid == null) return; // killed once the PID: line arrives + try { + process.kill(childPid, "SIGTERM"); + } catch {} + hardKill = setTimeout(() => { + try { + process.kill(childPid!, "SIGKILL"); + } catch {} + }, 5_000); + hardKill.unref?.(); + }; + + const onAbort = () => { + cancelled = true; + console.warn(`cancelling ${scriptName}: client went away`); + killChild(); + conn.destroy(); + }; + + const conn = net.createConnection({ path: master.socketPath }); + conn.on("connect", () => { + conn.write( + JSON.stringify({ service: scriptName, input: inputPath, output: outputPath, port }) + "\n" + ); + }); + + // The child's stream: the service's own stdout, wrapped in PID:/EXIT:. + const rl = readline.createInterface({ input: conn, crlfDelay: Infinity }); + rl.on("line", (line) => { + if (line.startsWith("PID:")) { + childPid = Number(line.slice(4)) || null; + if (cancelled) killChild(); + } else if (line.startsWith("EXIT:")) { + exitCode = Number(line.slice(5)); + } else if (LOG_LINE.test(line)) { + console.log(line); + onLog?.(line); + } else if (/^(EVENT)\:/.test(line)) { + const [_prefix, type, ...payload] = line.split(":"); + let processedPayload: any = payload.join(":"); + try { + processedPayload = JSON.parse(processedPayload); + } catch (e) { + // No json, no problem + } + onEvent?.(type, processedPayload); + } + }); + + if (signal) { + if (signal.aborted) onAbort(); + else signal.addEventListener("abort", onAbort, { once: true }); + } + + conn.on("error", async (err) => { + if (hardKill) clearTimeout(hardKill); + signal?.removeEventListener("abort", onAbort); + await cleanupFiles(); + reject(subprocessSpawnFailed(scriptName, err)); + }); + + conn.on("close", async () => { + rl.close(); + if (hardKill) clearTimeout(hardKill); + signal?.removeEventListener("abort", onAbort); + + const text = await Bun.file(outputPath) + .text() + .catch(() => ""); + await cleanupFiles(); + + if (cancelled) return reject(subprocessCancelled(scriptName, "SIGTERM")); + if (exitCode && exitCode !== 0) return reject(subprocessFailed(scriptName, exitCode)); + // No EXIT line => the child died mid-flight (crash/OOM/signal). + if (exitCode === null) return reject(subprocessKilled(scriptName, "UNKNOWN")); + if (text) { + try { + return resolve(JSON.parse(text)); + } catch (e) { + console.error(`Unreadable output from ${scriptName}`); + console.error(e); + return reject(malformedResult(scriptName)); + } + } + console.warn("No data returned from pythonland"); + return reject(emptyResult(scriptName)); + }); + }), + async (err) => { + await rm(inputPath).catch(() => {}); + await rm(outputPath).catch(() => {}); + throw err; + } + ); +} + /** Run a python script Each script will be run in its own thread because @@ -61,6 +255,10 @@ export const run = async ( throw subprocessSpawnFailed(scriptName, error); } + if (USE_FORK_SERVER) { + return runForked(scriptName, port, inputPath, outputPath, onLog, onEvent, signal); + } + return new Promise((resolve, reject) => { const proc = spawn( "poetry", diff --git a/services/entry.py b/services/entry.py index 238f476e..daf8f597 100644 --- a/services/entry.py +++ b/services/entry.py @@ -9,17 +9,15 @@ load_dotenv() -# Langfuse: init after load_dotenv so env vars are available, before any Anthropic client is created from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor from opentelemetry.instrumentation.threading import ThreadingInstrumentor -AnthropicInstrumentor().instrument() -ThreadingInstrumentor().instrument() - from langfuse import Langfuse from langfuse.span_filter import is_default_export_span from langfuse_util import mask_secrets +langfuse = None # set by bootstrap() + def _should_export_span(span): """Drop spans marked as tracing-disabled (user has not opted in).""" @@ -29,20 +27,6 @@ def _should_export_span(span): return is_default_export_span(span) -langfuse = Langfuse( - should_export_span=_should_export_span, - mask=mask_secrets, - release=os.getenv("APOLLO_VERSION", "unknown"), -) - -env = os.getenv('ENVIRONMENT', 'unknown') -trace_rates = { - 'development': 1, - 'staging': 0.05, - 'production': 0.03, - 'unknown': 0.0, - } - def _scrub_event(event: dict, _hint: dict) -> dict: """Mask keys in what Sentry is about to send. @@ -54,17 +38,41 @@ def _scrub_event(event: dict, _hint: dict) -> dict: return mask_secrets(event) -sentry_sdk.init( - dsn=os.getenv('SENTRY_DSN'), - environment=env, - sample_rate=1.0, - traces_sample_rate=trace_rates.get(env, 0.0), - enable_tracing=True, - auto_enabling_integrations=False, - before_send=_scrub_event, - # before_send covers error events only, and tracing is on. - before_send_transaction=_scrub_event, -) +def bootstrap() -> None: + global langfuse + + AnthropicInstrumentor().instrument() + ThreadingInstrumentor().instrument() + + langfuse = Langfuse( + should_export_span=_should_export_span, + mask=mask_secrets, + release=os.getenv("APOLLO_VERSION", "unknown"), + ) + + env = os.getenv("ENVIRONMENT", "unknown") + trace_rates = { + "development": 1, + "staging": 0.05, + "production": 0.03, + "unknown": 0.0, + } + + sentry_sdk.init( + dsn=os.getenv("SENTRY_DSN"), + environment=env, + sample_rate=1.0, + traces_sample_rate=trace_rates.get(env, 0.0), + enable_tracing=True, + auto_enabling_integrations=False, + before_send=_scrub_event, + # before_send covers error events only, and tracing is on. + before_send_transaction=_scrub_event, + ) + + +if os.environ.get("APOLLO_FORK_PRELOAD") != "1": + bootstrap() # At or above this the failure is ours; below it the caller sent something we # correctly refused. @@ -156,7 +164,8 @@ def call( code=500, message=str(e), type="INTERNAL_ERROR", ).to_dict() - langfuse.flush() + if langfuse is not None: + langfuse.flush() return _finish(result, output_path) diff --git a/services/fork_server.py b/services/fork_server.py new file mode 100644 index 00000000..9aecdce7 --- /dev/null +++ b/services/fork_server.py @@ -0,0 +1,109 @@ +import os +import sys + +# macOS aborts a forked child that touches an objc framework unless this is set +# before interpreter start, so re-exec once with it in the env. No-op on Linux. +# [To be removed] +if sys.platform == "darwin" and os.environ.get("OBJC_DISABLE_INITIALIZE_FORK_SAFETY") != "YES": + os.environ["OBJC_DISABLE_INITIALIZE_FORK_SAFETY"] = "YES" + os.execvpe(sys.executable, [sys.executable, *sys.argv], os.environ) + +# Tell entry.py to import its heavy modules but defer bootstrap() to each child. +os.environ["APOLLO_FORK_PRELOAD"] = "1" + +import argparse # noqa: E402 +import importlib # noqa: E402 +import json # noqa: E402 +import signal # noqa: E402 +import socket # noqa: E402 +import traceback # noqa: E402 + +import entry # noqa: E402 + + +def _preload_services() -> list[str]: + """Import an opt-in allowlist (APOLLO_FORK_PRELOAD_SERVICES) so children + inherit their deps. Not every service: some do I/O at import and would hang + the parent; entry.call() imports the rest on demand in the child.""" + names = [n.strip() for n in os.environ.get("APOLLO_FORK_PRELOAD_SERVICES", "").split(",") if n.strip()] + loaded = [] + for name in names: + try: + importlib.import_module(f"{name}.{name}") + loaded.append(name) + except Exception as e: + print(f"fork_server: could not preload {name}: {e}", file=sys.stderr, flush=True) + return loaded + + +def _run_child(conn: socket.socket, req: dict) -> None: + os.dup2(conn.fileno(), 1) + os.dup2(conn.fileno(), 2) + sys.stdout = os.fdopen(1, "w", buffering=1) + sys.stderr = os.fdopen(2, "w", buffering=1) + + sys.stdout.write(f"PID:{os.getpid()}\n") # first, so the bridge can cancel this child + sys.stdout.flush() + + code = 0 + try: + entry.bootstrap() + entry.call( + service=req["service"], + input_path=req.get("input"), + output_path=req.get("output"), + apollo_port=req.get("port"), + ) + except Exception: + traceback.print_exc() + code = 1 + finally: + sys.stdout.write(f"EXIT:{code}\n") + sys.stdout.flush() + os._exit(code) + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--socket", required=True) + args = ap.parse_args() + + signal.signal(signal.SIGCHLD, signal.SIG_IGN) # auto-reap children + + loaded = _preload_services() + + if os.path.exists(args.socket): + os.unlink(args.socket) + srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv.bind(args.socket) + srv.listen(128) + print(f"READY services={len(loaded)}", flush=True) # handshake the bridge waits for + + while True: + try: + conn, _ = srv.accept() + except InterruptedError: + continue + line = b"" + while not line.endswith(b"\n"): + chunk = conn.recv(65536) + if not chunk: + break + line += chunk + if not line: + conn.close() + continue + try: + req = json.loads(line) + except json.JSONDecodeError: + conn.close() + continue + + if os.fork() == 0: + srv.close() # the child must not hold the listening socket + _run_child(conn, req) # never returns + conn.close() # parent drops its copy; the child owns the reply + + +if __name__ == "__main__": + main() From 907558abdd583fac8d561cd6b3c87d0b9df74d55 Mon Sep 17 00:00:00 2001 From: doc-han Date: Wed, 7 Oct 2026 08:33:59 +0000 Subject: [PATCH 2/9] feat: revert changes to entry file --- services/entry.py | 69 +++++++++++-------------- services/fork_server.py | 109 ---------------------------------------- 2 files changed, 30 insertions(+), 148 deletions(-) delete mode 100644 services/fork_server.py diff --git a/services/entry.py b/services/entry.py index daf8f597..238f476e 100644 --- a/services/entry.py +++ b/services/entry.py @@ -9,15 +9,17 @@ load_dotenv() +# Langfuse: init after load_dotenv so env vars are available, before any Anthropic client is created from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor from opentelemetry.instrumentation.threading import ThreadingInstrumentor +AnthropicInstrumentor().instrument() +ThreadingInstrumentor().instrument() + from langfuse import Langfuse from langfuse.span_filter import is_default_export_span from langfuse_util import mask_secrets -langfuse = None # set by bootstrap() - def _should_export_span(span): """Drop spans marked as tracing-disabled (user has not opted in).""" @@ -27,6 +29,20 @@ def _should_export_span(span): return is_default_export_span(span) +langfuse = Langfuse( + should_export_span=_should_export_span, + mask=mask_secrets, + release=os.getenv("APOLLO_VERSION", "unknown"), +) + +env = os.getenv('ENVIRONMENT', 'unknown') +trace_rates = { + 'development': 1, + 'staging': 0.05, + 'production': 0.03, + 'unknown': 0.0, + } + def _scrub_event(event: dict, _hint: dict) -> dict: """Mask keys in what Sentry is about to send. @@ -38,41 +54,17 @@ def _scrub_event(event: dict, _hint: dict) -> dict: return mask_secrets(event) -def bootstrap() -> None: - global langfuse - - AnthropicInstrumentor().instrument() - ThreadingInstrumentor().instrument() - - langfuse = Langfuse( - should_export_span=_should_export_span, - mask=mask_secrets, - release=os.getenv("APOLLO_VERSION", "unknown"), - ) - - env = os.getenv("ENVIRONMENT", "unknown") - trace_rates = { - "development": 1, - "staging": 0.05, - "production": 0.03, - "unknown": 0.0, - } - - sentry_sdk.init( - dsn=os.getenv("SENTRY_DSN"), - environment=env, - sample_rate=1.0, - traces_sample_rate=trace_rates.get(env, 0.0), - enable_tracing=True, - auto_enabling_integrations=False, - before_send=_scrub_event, - # before_send covers error events only, and tracing is on. - before_send_transaction=_scrub_event, - ) - - -if os.environ.get("APOLLO_FORK_PRELOAD") != "1": - bootstrap() +sentry_sdk.init( + dsn=os.getenv('SENTRY_DSN'), + environment=env, + sample_rate=1.0, + traces_sample_rate=trace_rates.get(env, 0.0), + enable_tracing=True, + auto_enabling_integrations=False, + before_send=_scrub_event, + # before_send covers error events only, and tracing is on. + before_send_transaction=_scrub_event, +) # At or above this the failure is ours; below it the caller sent something we # correctly refused. @@ -164,8 +156,7 @@ def call( code=500, message=str(e), type="INTERNAL_ERROR", ).to_dict() - if langfuse is not None: - langfuse.flush() + langfuse.flush() return _finish(result, output_path) diff --git a/services/fork_server.py b/services/fork_server.py deleted file mode 100644 index 9aecdce7..00000000 --- a/services/fork_server.py +++ /dev/null @@ -1,109 +0,0 @@ -import os -import sys - -# macOS aborts a forked child that touches an objc framework unless this is set -# before interpreter start, so re-exec once with it in the env. No-op on Linux. -# [To be removed] -if sys.platform == "darwin" and os.environ.get("OBJC_DISABLE_INITIALIZE_FORK_SAFETY") != "YES": - os.environ["OBJC_DISABLE_INITIALIZE_FORK_SAFETY"] = "YES" - os.execvpe(sys.executable, [sys.executable, *sys.argv], os.environ) - -# Tell entry.py to import its heavy modules but defer bootstrap() to each child. -os.environ["APOLLO_FORK_PRELOAD"] = "1" - -import argparse # noqa: E402 -import importlib # noqa: E402 -import json # noqa: E402 -import signal # noqa: E402 -import socket # noqa: E402 -import traceback # noqa: E402 - -import entry # noqa: E402 - - -def _preload_services() -> list[str]: - """Import an opt-in allowlist (APOLLO_FORK_PRELOAD_SERVICES) so children - inherit their deps. Not every service: some do I/O at import and would hang - the parent; entry.call() imports the rest on demand in the child.""" - names = [n.strip() for n in os.environ.get("APOLLO_FORK_PRELOAD_SERVICES", "").split(",") if n.strip()] - loaded = [] - for name in names: - try: - importlib.import_module(f"{name}.{name}") - loaded.append(name) - except Exception as e: - print(f"fork_server: could not preload {name}: {e}", file=sys.stderr, flush=True) - return loaded - - -def _run_child(conn: socket.socket, req: dict) -> None: - os.dup2(conn.fileno(), 1) - os.dup2(conn.fileno(), 2) - sys.stdout = os.fdopen(1, "w", buffering=1) - sys.stderr = os.fdopen(2, "w", buffering=1) - - sys.stdout.write(f"PID:{os.getpid()}\n") # first, so the bridge can cancel this child - sys.stdout.flush() - - code = 0 - try: - entry.bootstrap() - entry.call( - service=req["service"], - input_path=req.get("input"), - output_path=req.get("output"), - apollo_port=req.get("port"), - ) - except Exception: - traceback.print_exc() - code = 1 - finally: - sys.stdout.write(f"EXIT:{code}\n") - sys.stdout.flush() - os._exit(code) - - -def main() -> None: - ap = argparse.ArgumentParser() - ap.add_argument("--socket", required=True) - args = ap.parse_args() - - signal.signal(signal.SIGCHLD, signal.SIG_IGN) # auto-reap children - - loaded = _preload_services() - - if os.path.exists(args.socket): - os.unlink(args.socket) - srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - srv.bind(args.socket) - srv.listen(128) - print(f"READY services={len(loaded)}", flush=True) # handshake the bridge waits for - - while True: - try: - conn, _ = srv.accept() - except InterruptedError: - continue - line = b"" - while not line.endswith(b"\n"): - chunk = conn.recv(65536) - if not chunk: - break - line += chunk - if not line: - conn.close() - continue - try: - req = json.loads(line) - except json.JSONDecodeError: - conn.close() - continue - - if os.fork() == 0: - srv.close() # the child must not hold the listening socket - _run_child(conn, req) # never returns - conn.close() # parent drops its copy; the child owns the reply - - -if __name__ == "__main__": - main() From f40a4988508af8dbfd71564120b71fbad58ab97d Mon Sep 17 00:00:00 2001 From: doc-han Date: Wed, 7 Oct 2026 09:06:44 +0000 Subject: [PATCH 3/9] feat: finalize fork bridge & macos deps --- .env.example | 6 ++ platform/src/bridge.fork.ts | 186 ++++++++++++++++++++++++++++++++++ platform/src/bridge.ts | 197 +----------------------------------- services/entry.fork.py | 129 +++++++++++++++++++++++ 4 files changed, 323 insertions(+), 195 deletions(-) create mode 100644 platform/src/bridge.fork.ts create mode 100644 services/entry.fork.py diff --git a/.env.example b/.env.example index 8f27f4bb..a3f087b1 100644 --- a/.env.example +++ b/.env.example @@ -70,3 +70,9 @@ LANGFUSE_BASE_URL=https://cloud.langfuse.com # ZILLIZ_URI = https://in01-XXXXXXXXXXXXX.aws-us-west-2.vectordb.zillizcloud.com:XXXXX # ZILLIZ_TOKEN =db_admin:password (or ApiKey) +# Warm fork-server (see services/entry.fork.py). Flip to "true" to use it. +# APOLLO_FORK_SERVER=true +# macOS only: required when APOLLO_FORK_SERVER=true, or a forked child that touches +# an Objective-C framework aborts. No-op on Linux, safe to leave set everywhere. +# OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES + diff --git a/platform/src/bridge.fork.ts b/platform/src/bridge.fork.ts new file mode 100644 index 00000000..c3c0e7ea --- /dev/null +++ b/platform/src/bridge.fork.ts @@ -0,0 +1,186 @@ +import readline from "node:readline"; +import net from "node:net"; +import path from "node:path"; +import { spawn, type ChildProcess } from "node:child_process"; +import { rm } from "node:fs/promises"; +import { getInternalToken } from "./auth/internal-token"; +import { + emptyResult, + malformedResult, + subprocessCancelled, + subprocessFailed, + subprocessKilled, + subprocessSpawnFailed, +} from "./util/errors"; +import pkg from "../../package.json"; + +// Only these lines are forwarded to the caller (mirrors bridge.ts). +const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/; + +export const USE_FORK_SERVER = + (process.env.APOLLO_FORK_SERVER ?? "").toLowerCase() === "true"; + +const onLines = (stream: NodeJS.ReadableStream, fn: (line: string) => void) => + readline.createInterface({ input: stream, crlfDelay: Infinity }).on("line", fn); + +// --- the warm master: spawned once, forks a child per request ---------------- + +type Master = { proc: ChildProcess; socketPath: string }; +let master: Promise | null = null; +let masterProc: ChildProcess | null = null; + +// Singleton; a failed start clears the slot so the next request retries. +function forkMaster(): Promise { + return (master ??= startMaster().catch((err) => { + master = null; + throw err; + })); +} + +function startMaster(): Promise { + const socketPath = path.resolve(`tmp/fork_server-${process.pid}.sock`); + const proc = spawn( + "poetry", + ["run", "python", "services/entry.fork.py", "--socket", socketPath], + { + env: { + ...process.env, + APOLLO_INTERNAL_TOKEN: getInternalToken(), + APOLLO_VERSION: pkg.version, + }, + } + ); + masterProc = proc; + onLines(proc.stderr!, (line) => console.error(`fork-server: ${line}`)); + + return new Promise((resolve, reject) => { + const fail = (err: Error) => (clearTimeout(timer), reject(subprocessSpawnFailed("fork_server", err))); + const timer = setTimeout(() => fail(new Error("not READY in time")), 60_000); + + onLines(proc.stdout!, (line) => { + console.log(`fork-server: ${line}`); + if (line.startsWith("READY")) { + clearTimeout(timer); + resolve({ proc, socketPath }); + } + }); + proc.on("error", fail); + proc.on("exit", (code, sig) => { + clearTimeout(timer); + console.error(`fork-server exited (code=${code} signal=${sig})`); + master = masterProc = null; // next request starts a fresh one + rm(socketPath).catch(() => {}); + }); + }); +} + +// Don't leave an orphaned master when the server process goes. +process.once("exit", () => masterProc?.kill("SIGTERM")); + +// --- per-request: ask the master to fork a child, stream it back ------------- + +export async function runForked( + scriptName: string, + port: number, + inputPath: string, + outputPath: string, + onLog?: (str: string) => void, + onEvent?: (type: string, payload: any) => void, + signal?: AbortSignal +): Promise { + const cleanup = () => + Promise.all([rm(inputPath).catch(() => {}), rm(outputPath).catch(() => {})]); + + let active: Master; + try { + active = await forkMaster(); + } catch (err) { + await cleanup(); + throw err; + } + + return new Promise((resolve, reject) => { + const conn = net.createConnection({ path: active.socketPath }); + let childPid: number | null = null; + let exitCode: number | null = null; + let cancelled = false; + let hardKill: ReturnType | undefined; + + const killChild = () => { + if (childPid == null) return; // killed once the PID: line arrives + try { + process.kill(childPid, "SIGTERM"); + } catch {} + hardKill = setTimeout(() => { + try { + process.kill(childPid!, "SIGKILL"); + } catch {} + }, 5_000); + hardKill.unref?.(); + }; + + const onAbort = () => { + cancelled = true; + console.warn(`cancelling ${scriptName}: client went away`); + killChild(); + conn.destroy(); + }; + if (signal) { + if (signal.aborted) onAbort(); + else signal.addEventListener("abort", onAbort, { once: true }); + } + + conn.on("connect", () => + conn.write(JSON.stringify({ service: scriptName, input: inputPath, output: outputPath, port }) + "\n") + ); + + // The child's stream: PID:/EXIT: control lines around the service's own stdout. + onLines(conn, (line) => { + if (line.startsWith("PID:")) { + childPid = Number(line.slice(4)) || null; + if (cancelled) killChild(); + } else if (line.startsWith("EXIT:")) { + exitCode = Number(line.slice(5)); + } else if (LOG_LINE.test(line)) { + console.log(line); + onLog?.(line); + } else if (line.startsWith("EVENT:")) { + const [, type, ...rest] = line.split(":"); + let payload: any = rest.join(":"); + try { + payload = JSON.parse(payload); + } catch { + // not json, forward the raw string + } + onEvent?.(type, payload); + } + }); + + const settle = async (finish: () => void) => { + if (hardKill) clearTimeout(hardKill); + signal?.removeEventListener("abort", onAbort); + await cleanup(); + finish(); + }; + + conn.on("error", (err) => + settle(() => reject(subprocessSpawnFailed(scriptName, err))) + ); + + conn.on("close", async () => { + const text = await Bun.file(outputPath).text().catch(() => ""); + await settle(() => { + if (cancelled) return reject(subprocessCancelled(scriptName, "SIGTERM")); + // No EXIT line => the child died mid-flight (crash/OOM/signal). + if (exitCode === null) return reject(subprocessKilled(scriptName, "UNKNOWN")); + if (exitCode !== 0) return reject(subprocessFailed(scriptName, exitCode)); + if (!text) return reject(emptyResult(scriptName)); + try { + resolve(JSON.parse(text)); + } catch { + reject(malformedResult(scriptName)); + } + }); + }); + }); +} diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index edceebec..011dbeff 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -1,7 +1,6 @@ import readline from "node:readline"; -import net from "node:net"; import path from "node:path"; -import { spawn, type ChildProcess } from "node:child_process"; +import { spawn } from "node:child_process"; import { chmod, rm } from "node:fs/promises"; import { getInternalToken } from "./auth/internal-token"; import { @@ -13,204 +12,12 @@ import { subprocessSpawnFailed, } from "./util/errors"; import pkg from "../../package.json"; +import { USE_FORK_SERVER, runForked } from "./bridge.fork"; // A line a service logged on purpose, as opposed to whatever else lands on a // stream. Only these are forwarded to the caller. const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/; -const USE_FORK_SERVER = ["1", "true", "yes"].includes( - (process.env.APOLLO_FORK_SERVER ?? "").toLowerCase() -); - -interface ForkMaster { - proc: ChildProcess; - socketPath: string; -} -let forkMaster: Promise | null = null; -let forkMasterProc: ChildProcess | null = null; - -function startForkMaster(): Promise { - const socketPath = path.resolve(`tmp/fork_server-${process.pid}.sock`); - const proc = spawn( - "poetry", - ["run", "python", "services/fork_server.py", "--socket", socketPath], - { - env: { - ...process.env, - APOLLO_INTERNAL_TOKEN: getInternalToken(), - APOLLO_VERSION: pkg.version, - OBJC_DISABLE_INITIALIZE_FORK_SAFETY: "YES", - }, - } - ); - forkMasterProc = proc; - - return new Promise((resolve, reject) => { - const timeout = setTimeout( - () => reject(subprocessSpawnFailed("fork_server", new Error("not READY in time"))), - 60_000 - ); - readline.createInterface({ input: proc.stdout!, crlfDelay: Infinity }).on("line", (line) => { - console.log(`fork-server: ${line}`); - if (line.startsWith("READY")) { - clearTimeout(timeout); - resolve({ proc, socketPath }); - } - }); - readline - .createInterface({ input: proc.stderr!, crlfDelay: Infinity }) - .on("line", (line) => console.error(`fork-server: ${line}`)); - - proc.on("error", (err) => { - clearTimeout(timeout); - reject(subprocessSpawnFailed("fork_server", err)); - }); - proc.on("exit", (code, sig) => { - clearTimeout(timeout); - console.error(`fork-server exited (code=${code} signal=${sig})`); - forkMaster = null; // next request starts a fresh master - forkMasterProc = null; - rm(socketPath).catch(() => {}); - }); - }); -} - -function getForkMaster(): Promise { - if (!forkMaster) { - forkMaster = startForkMaster().catch((err) => { - forkMaster = null; - throw err; - }); - } - return forkMaster; -} - -process.once("exit", () => forkMasterProc?.kill("SIGTERM")); - -function runForked( - scriptName: string, - port: number, - inputPath: string, - outputPath: string, - onLog?: (str: string) => void, - onEvent?: (type: string, payload: any) => void, - signal?: AbortSignal -): Promise { - return getForkMaster().then( - (master) => - new Promise((resolve, reject) => { - let childPid: number | null = null; - let exitCode: number | null = null; - let cancelled = false; - let hardKill: ReturnType | undefined; - - const cleanupFiles = async () => { - try { - await rm(inputPath); - await rm(outputPath); - } catch (e) { - console.error("Error removing temporary files"); - console.error(e); - } - }; - - const killChild = () => { - if (childPid == null) return; // killed once the PID: line arrives - try { - process.kill(childPid, "SIGTERM"); - } catch {} - hardKill = setTimeout(() => { - try { - process.kill(childPid!, "SIGKILL"); - } catch {} - }, 5_000); - hardKill.unref?.(); - }; - - const onAbort = () => { - cancelled = true; - console.warn(`cancelling ${scriptName}: client went away`); - killChild(); - conn.destroy(); - }; - - const conn = net.createConnection({ path: master.socketPath }); - conn.on("connect", () => { - conn.write( - JSON.stringify({ service: scriptName, input: inputPath, output: outputPath, port }) + "\n" - ); - }); - - // The child's stream: the service's own stdout, wrapped in PID:/EXIT:. - const rl = readline.createInterface({ input: conn, crlfDelay: Infinity }); - rl.on("line", (line) => { - if (line.startsWith("PID:")) { - childPid = Number(line.slice(4)) || null; - if (cancelled) killChild(); - } else if (line.startsWith("EXIT:")) { - exitCode = Number(line.slice(5)); - } else if (LOG_LINE.test(line)) { - console.log(line); - onLog?.(line); - } else if (/^(EVENT)\:/.test(line)) { - const [_prefix, type, ...payload] = line.split(":"); - let processedPayload: any = payload.join(":"); - try { - processedPayload = JSON.parse(processedPayload); - } catch (e) { - // No json, no problem - } - onEvent?.(type, processedPayload); - } - }); - - if (signal) { - if (signal.aborted) onAbort(); - else signal.addEventListener("abort", onAbort, { once: true }); - } - - conn.on("error", async (err) => { - if (hardKill) clearTimeout(hardKill); - signal?.removeEventListener("abort", onAbort); - await cleanupFiles(); - reject(subprocessSpawnFailed(scriptName, err)); - }); - - conn.on("close", async () => { - rl.close(); - if (hardKill) clearTimeout(hardKill); - signal?.removeEventListener("abort", onAbort); - - const text = await Bun.file(outputPath) - .text() - .catch(() => ""); - await cleanupFiles(); - - if (cancelled) return reject(subprocessCancelled(scriptName, "SIGTERM")); - if (exitCode && exitCode !== 0) return reject(subprocessFailed(scriptName, exitCode)); - // No EXIT line => the child died mid-flight (crash/OOM/signal). - if (exitCode === null) return reject(subprocessKilled(scriptName, "UNKNOWN")); - if (text) { - try { - return resolve(JSON.parse(text)); - } catch (e) { - console.error(`Unreadable output from ${scriptName}`); - console.error(e); - return reject(malformedResult(scriptName)); - } - } - console.warn("No data returned from pythonland"); - return reject(emptyResult(scriptName)); - }); - }), - async (err) => { - await rm(inputPath).catch(() => {}); - await rm(outputPath).catch(() => {}); - throw err; - } - ); -} - /** Run a python script Each script will be run in its own thread because diff --git a/services/entry.fork.py b/services/entry.fork.py new file mode 100644 index 00000000..1211c783 --- /dev/null +++ b/services/entry.fork.py @@ -0,0 +1,129 @@ +"""Fork entry for the Python side (used when APOLLO_FORK_SERVER=true; see +platform/src/bridge.fork.ts). Run as a script, never imported. + +Pays the heavy imports once, then forks a fresh child per request. Each child +inherits the imports via copy-on-write and imports entry.py AFTER forking, so +entry.py's own init runs in the child where it's fork-safe - entry.py is left +untouched. The parent only imports and never inits, so it stays forkable. + +macOS: set OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES in your env (see .env.example), +or a forked child that touches an Objective-C framework aborts. No-op on Linux. + +Protocol (NDJSON over a unix socket, one connection per request): + bridge -> master : {"service","input","output","port"} + child -> bridge : PID: then the service's stdout then EXIT: +""" + +import argparse +import importlib +import json +import os +import signal +import socket +import sys +import traceback + +# Warm entry.py's heavy deps so forked children inherit them (copy-on-write). +# Mirrors entry.py's imports; a missing one is just imported per-child - slower, +# never wrong. +PRELOAD = ( + "sentry_sdk", + "anthropic", + "langfuse", + "langfuse_util", + "util", + "opentelemetry.instrumentation.anthropic", + "opentelemetry.instrumentation.threading", +) + + +def preload() -> int: + """Warm the shared deps, plus any opt-in services. Services are opt-in + (APOLLO_FORK_PRELOAD_SERVICES) because some do I/O at import and would hang + the parent; the child imports the rest on demand.""" + for module in PRELOAD: + importlib.import_module(module) + + names = [s.strip() for s in os.environ.get("APOLLO_FORK_PRELOAD_SERVICES", "").split(",") if s.strip()] + loaded = 0 + for name in names: + try: + importlib.import_module(f"{name}.{name}") + loaded += 1 + except Exception as e: + print(f"entry.fork: could not preload {name}: {e}", file=sys.stderr, flush=True) + return loaded + + +def run_child(conn: socket.socket, req: dict) -> None: + """In the forked child: stream through the socket, run one job, exit.""" + # Route stdout/stderr to the socket so the service's logging reaches the bridge. + os.dup2(conn.fileno(), 1) + os.dup2(conn.fileno(), 2) + sys.stdout = os.fdopen(1, "w", buffering=1) + sys.stderr = os.fdopen(2, "w", buffering=1) + + print(f"PID:{os.getpid()}") # first, so the bridge can cancel this child + + code = 0 + try: + import entry # post-fork: entry.py's init runs here (fork-safe), deps warm + + entry.call( + service=req["service"], + input_path=req.get("input"), + output_path=req.get("output"), + apollo_port=req.get("port"), + ) + except Exception: + traceback.print_exc() + code = 1 + finally: + print(f"EXIT:{code}") + os._exit(code) + + +def read_request(conn: socket.socket) -> dict | None: + """Read one newline-terminated JSON request; None if there isn't a usable one.""" + line = b"" + while not line.endswith(b"\n"): + chunk = conn.recv(65536) + if not chunk: + break + line += chunk + try: + return json.loads(line) if line else None + except json.JSONDecodeError: + return None + + +def serve(sock_path: str) -> None: + signal.signal(signal.SIGCHLD, signal.SIG_IGN) # auto-reap children + count = preload() + + if os.path.exists(sock_path): + os.unlink(sock_path) + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(sock_path) + server.listen(128) + print(f"READY services={count}", flush=True) # handshake the bridge waits for + + while True: + try: + conn, _ = server.accept() + except InterruptedError: + continue + req = read_request(conn) + if req is None: + conn.close() + continue + if os.fork() == 0: + server.close() # the child must not hold the listening socket + run_child(conn, req) # never returns + conn.close() # parent drops its copy; the child owns the reply + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--socket", required=True) + serve(parser.parse_args().socket) From 03a108147bf2998e2dbb6e7cbc6c1d52d76ee9eb Mon Sep 17 00:00:00 2001 From: doc-han Date: Wed, 7 Oct 2026 11:35:47 +0000 Subject: [PATCH 4/9] chore: update changelog --- .env.example | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.env.example b/.env.example index a3f087b1..5e945f07 100644 --- a/.env.example +++ b/.env.example @@ -70,9 +70,9 @@ LANGFUSE_BASE_URL=https://cloud.langfuse.com # ZILLIZ_URI = https://in01-XXXXXXXXXXXXX.aws-us-west-2.vectordb.zillizcloud.com:XXXXX # ZILLIZ_TOKEN =db_admin:password (or ApiKey) -# Warm fork-server (see services/entry.fork.py). Flip to "true" to use it. +# Turn this on to use the fork server mechanism # APOLLO_FORK_SERVER=true -# macOS only: required when APOLLO_FORK_SERVER=true, or a forked child that touches -# an Objective-C framework aborts. No-op on Linux, safe to leave set everywhere. + +# macOS only: required when APOLLO_FORK_SERVER=true # OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES From fb82a745d4078486f1f296021da972e1cc35300e Mon Sep 17 00:00:00 2001 From: doc-han Date: Fri, 9 Oct 2026 08:47:47 +0000 Subject: [PATCH 5/9] fix: resolve issues --- platform/src/bridge.fork.ts | 2 +- services/entry.fork.py | 19 ++++++++++++++++--- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/platform/src/bridge.fork.ts b/platform/src/bridge.fork.ts index c3c0e7ea..70177b79 100644 --- a/platform/src/bridge.fork.ts +++ b/platform/src/bridge.fork.ts @@ -66,10 +66,10 @@ function startMaster(): Promise { }); proc.on("error", fail); proc.on("exit", (code, sig) => { - clearTimeout(timer); console.error(`fork-server exited (code=${code} signal=${sig})`); master = masterProc = null; // next request starts a fresh one rm(socketPath).catch(() => {}); + fail(new Error(`fork-server exited before ready (code=${code} signal=${sig})`)); }); }); } diff --git a/services/entry.fork.py b/services/entry.fork.py index 1211c783..9c4fef70 100644 --- a/services/entry.fork.py +++ b/services/entry.fork.py @@ -23,6 +23,8 @@ import sys import traceback +import sentry_sdk # for the per-child service tag + flush (see run_child) + # Warm entry.py's heavy deps so forked children inherit them (copy-on-write). # Mirrors entry.py's imports; a missing one is just imported per-child - slower, # never wrong. @@ -57,11 +59,11 @@ def preload() -> int: def run_child(conn: socket.socket, req: dict) -> None: """In the forked child: stream through the socket, run one job, exit.""" - # Route stdout/stderr to the socket so the service's logging reaches the bridge. + # Route stdout to the socket so the service's logging reaches the bridge. + # Leave stderr on fd 2 (the master's stderr pipe) so Python tracebacks still + # reach the server logs, the way the spawn path forwards them. os.dup2(conn.fileno(), 1) - os.dup2(conn.fileno(), 2) sys.stdout = os.fdopen(1, "w", buffering=1) - sys.stderr = os.fdopen(2, "w", buffering=1) print(f"PID:{os.getpid()}") # first, so the bridge can cancel this child @@ -69,6 +71,7 @@ def run_child(conn: socket.socket, req: dict) -> None: try: import entry # post-fork: entry.py's init runs here (fork-safe), deps warm + sentry_sdk.set_tag("service", req["service"]) # entry.main() tags it; call() doesn't entry.call( service=req["service"], input_path=req.get("input"), @@ -79,6 +82,16 @@ def run_child(conn: socket.socket, req: dict) -> None: traceback.print_exc() code = 1 finally: + try: + sentry_sdk.flush(timeout=2) + except Exception: + pass + langfuse = getattr(sys.modules.get("entry"), "langfuse", None) + if langfuse is not None: + try: + langfuse.flush() + except Exception: + pass print(f"EXIT:{code}") os._exit(code) From e7dab29c37503066f666d8aeebabd0e40de1ba83 Mon Sep 17 00:00:00 2001 From: doc-han Date: Mon, 12 Oct 2026 09:24:40 +0000 Subject: [PATCH 6/9] feat: move imports to yaml --- services/entry.fork.py | 19 +++++++------------ services/fork_preload.yaml | 22 ++++++++++++++++++++++ 2 files changed, 29 insertions(+), 12 deletions(-) create mode 100644 services/fork_preload.yaml diff --git a/services/entry.fork.py b/services/entry.fork.py index 9c4fef70..c83cbf71 100644 --- a/services/entry.fork.py +++ b/services/entry.fork.py @@ -22,28 +22,23 @@ import socket import sys import traceback +from pathlib import Path import sentry_sdk # for the per-child service tag + flush (see run_child) +import yaml # Warm entry.py's heavy deps so forked children inherit them (copy-on-write). -# Mirrors entry.py's imports; a missing one is just imported per-child - slower, -# never wrong. -PRELOAD = ( - "sentry_sdk", - "anthropic", - "langfuse", - "langfuse_util", - "util", - "opentelemetry.instrumentation.anthropic", - "opentelemetry.instrumentation.threading", -) +# See fork_preload.yaml for the list and why each module is there. +PRELOAD_PATH = Path(__file__).parent / "fork_preload.yaml" def preload() -> int: """Warm the shared deps, plus any opt-in services. Services are opt-in (APOLLO_FORK_PRELOAD_SERVICES) because some do I/O at import and would hang the parent; the child imports the rest on demand.""" - for module in PRELOAD: + with PRELOAD_PATH.open() as f: + modules = yaml.safe_load(f)["preload"] + for module in modules: importlib.import_module(module) names = [s.strip() for s in os.environ.get("APOLLO_FORK_PRELOAD_SERVICES", "").split(",") if s.strip()] diff --git a/services/fork_preload.yaml b/services/fork_preload.yaml new file mode 100644 index 00000000..7aac853a --- /dev/null +++ b/services/fork_preload.yaml @@ -0,0 +1,22 @@ +# Modules the fork master (entry.fork.py) imports once at startup, before +# accepting any requests. A forked child inherits these via copy-on-write, so +# the cost of importing them is paid once per server lifetime instead of once +# per request. A module missing from this list isn't an error - the child +# just imports it itself on demand, which is slower but never wrong. + +preload: + # Mirrors entry.py's own top-level imports (Sentry init, Langfuse, otel). + - sentry_sdk + - anthropic + - langfuse + - langfuse_util + - util + - opentelemetry.instrumentation.anthropic + - opentelemetry.instrumentation.threading + + # global_chat is ~95% of fork traffic; warm its RAG path (search_docsite) + # so the common case doesn't pay for these on the first request. + - pinecone + - openai + - langchain_openai + - langchain_pinecone From cb34f293606ad935ffed5d5a77fe6d7545d50f4e Mon Sep 17 00:00:00 2001 From: doc-han Date: Mon, 12 Oct 2026 09:26:25 +0000 Subject: [PATCH 7/9] chore: share log lines --- platform/src/bridge.fork.ts | 4 +--- platform/src/bridge.ts | 5 +---- platform/src/util/log-line.ts | 4 ++++ 3 files changed, 6 insertions(+), 7 deletions(-) create mode 100644 platform/src/util/log-line.ts diff --git a/platform/src/bridge.fork.ts b/platform/src/bridge.fork.ts index 70177b79..a6d34b14 100644 --- a/platform/src/bridge.fork.ts +++ b/platform/src/bridge.fork.ts @@ -12,11 +12,9 @@ import { subprocessKilled, subprocessSpawnFailed, } from "./util/errors"; +import { LOG_LINE } from "./util/log-line"; import pkg from "../../package.json"; -// Only these lines are forwarded to the caller (mirrors bridge.ts). -const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/; - export const USE_FORK_SERVER = (process.env.APOLLO_FORK_SERVER ?? "").toLowerCase() === "true"; diff --git a/platform/src/bridge.ts b/platform/src/bridge.ts index 011dbeff..bd79ad6e 100644 --- a/platform/src/bridge.ts +++ b/platform/src/bridge.ts @@ -13,10 +13,7 @@ import { } from "./util/errors"; import pkg from "../../package.json"; import { USE_FORK_SERVER, runForked } from "./bridge.fork"; - -// A line a service logged on purpose, as opposed to whatever else lands on a -// stream. Only these are forwarded to the caller. -const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/; +import { LOG_LINE } from "./util/log-line"; /** Run a python script diff --git a/platform/src/util/log-line.ts b/platform/src/util/log-line.ts new file mode 100644 index 00000000..b3c69ff7 --- /dev/null +++ b/platform/src/util/log-line.ts @@ -0,0 +1,4 @@ +// A line a service logged on purpose, as opposed to whatever else lands on a +// stream. Only these are forwarded to the caller. Shared by bridge.ts and +// bridge.fork.ts. +export const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/; From 3b56857dcdd33fddfd304479d36d53866fcff2c0 Mon Sep 17 00:00:00 2001 From: doc-han Date: Mon, 12 Oct 2026 10:59:14 +0000 Subject: [PATCH 8/9] chore: remove log --- platform/src/bridge.fork.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/platform/src/bridge.fork.ts b/platform/src/bridge.fork.ts index a6d34b14..6427d7be 100644 --- a/platform/src/bridge.fork.ts +++ b/platform/src/bridge.fork.ts @@ -56,7 +56,6 @@ function startMaster(): Promise { const timer = setTimeout(() => fail(new Error("not READY in time")), 60_000); onLines(proc.stdout!, (line) => { - console.log(`fork-server: ${line}`); if (line.startsWith("READY")) { clearTimeout(timer); resolve({ proc, socketPath }); From 8ff4d376fc3b36f6853cc6f26e742c1217d627d1 Mon Sep 17 00:00:00 2001 From: doc-han Date: Tue, 13 Oct 2026 10:18:53 +0000 Subject: [PATCH 9/9] chore: update changeset --- .changeset/spicy-vans-wait.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/spicy-vans-wait.md diff --git a/.changeset/spicy-vans-wait.md b/.changeset/spicy-vans-wait.md new file mode 100644 index 00000000..20e45000 --- /dev/null +++ b/.changeset/spicy-vans-wait.md @@ -0,0 +1,5 @@ +--- +"apollo": patch +--- + +fork python processes to reduce startup time