From ee05c5e98b15ce7b90445150fb46bfc35d338835 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 15 Oct 2026 05:40:34 +0800 Subject: [PATCH] fix(runtime): supervise optional cache flush after draining effects Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/development/testing-and-quality.md | 20 +++ .../effect_runtime_compile_cache.ts | 121 +++++++++++++++++- loopx/control_plane/effect_runtime_server.ts | 23 +--- .../runtime_retirement_diagnostics.py | 22 ++-- .../test_effect_runtime_integration.py | 25 +++- .../effect_runtime_compile_cache.test.ts | 94 +++++++++++++- 6 files changed, 273 insertions(+), 32 deletions(-) diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index f2eb06c6b7..0c8baefac1 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -386,6 +386,26 @@ Effect 后台进程；相同源码指纹可能让源码环境与已安装版本 回执、作废受污染耗时并重新采样；超时不能当作服务端已停止，也不能因此重复恢复。 恢复成功不证明并发管理操作公平性或自然时间 soak 已合格。 +The Effect server drains admitted handlers before retiring. With compilation +caching enabled, its final synchronous cache flush has a one-second allowance, +supervised by a worker in the same process. Healthy completion still persists +the real Node cache before removing the locator. A blocked/failed flush or +unavailable supervisor instead performs the same fenced, identity-checked +locator cleanup and forcefully exits the already settled process, bypassing +Node's automatic cache flush at exit. This may discard optional cache entries; +it never authorizes abandoning an admitted effect. Cache-disabled execution +uses ordinary cleanup without a worker. The allowance does not bound locator +lock/storage waits or OS scheduling. Native platform qualification remains +separate from injected-failure coverage; V8 CPU samples do not measure kernel IO. + +Effect 服务先排空已接纳的 handler，再退出。启用编译缓存时，同进程 worker +监督最终同步写盘，给缓存一秒处理时间；正常完成仍先持久化真实 Node 缓存， +再清理定位文件。写盘阻塞、失败或监督 worker 不可用时，复用同一锁和精确身份 +校验清理定位文件，然后强制终止已结算的进程，绕过 Node 退出时的自动缓存写盘。 +这可能丢弃可选缓存条目，不允许放弃已接纳的业务操作。关闭缓存时无需 worker， +继续普通清理。一秒额度不涵盖定位文件锁、存储等待或操作系统调度；故障注入不能 +代替原生平台资格验证，V8 CPU 样本也不测量内核 IO。 + Keep a deterministic, public-safe production-scale fixture beside the focused cases. Its envelope should cover realistic role/status distributions, multi-agent claims, user gates and standing decisions, current and retired diff --git a/loopx/control_plane/effect_runtime_compile_cache.ts b/loopx/control_plane/effect_runtime_compile_cache.ts index a11922b663..209ac27a8d 100644 --- a/loopx/control_plane/effect_runtime_compile_cache.ts +++ b/loopx/control_plane/effect_runtime_compile_cache.ts @@ -1,6 +1,119 @@ import { lstatSync, mkdirSync, type Stats } from "node:fs"; -import { enableCompileCache, getCompileCacheDir } from "node:module"; +import { readFile, rm } from "node:fs/promises"; +import { enableCompileCache, flushCompileCache, getCompileCacheDir } from "node:module"; import { dirname, isAbsolute, join } from "node:path"; +import { isMainThread, parentPort, Worker, workerData } from "node:worker_threads"; + +// Local synchronization states, not a persisted or caller-facing protocol. +const FLUSHING = 0; +const FLUSHED = 1; +const RETIRING = 2; +const CACHE_FLUSH_BUDGET_MS = 1_000; + +interface RuntimeLocator { + infoPath: string; + token: string; + pid: number; + fingerprint: string; +} + +interface RetirementWatchdog { + kind: "compile_cache_retirement"; + locator: RuntimeLocator; + state: SharedArrayBuffer; +} + +async function locatorRetirement(locator: RuntimeLocator): Promise<() => Promise> { + // Load before flushing so normal process exit cannot write newly compiled + // helpers back into a namespace its caller has already removed. + const { withFileMutationLock } = await import("./effect_runtime_io.ts"); + return () => withFileMutationLock(locator.infoPath, async () => { + let published: Record; + try { + published = JSON.parse(await readFile(locator.infoPath, "utf8")); + } catch { + return; + } + // Both retirement paths share the original fence and exact identity check. + // An old runtime must never erase a replacement server's locator. + if (published.token === locator.token && published.pid === locator.pid && + published.fingerprint === locator.fingerprint) { + await rm(locator.infoPath, { force: true }); + } + }); +} + +async function exitWithoutCache(retire: () => Promise): Promise { + try { + await retire(); + } finally { + // process.exit() invokes Node's synchronous cache flush again. Only after + // drain and fenced cleanup may we bypass that optional exit hook. A worker + // shares this PID, so this cannot target a later process that reused a PID. + process.kill(process.pid, "SIGKILL"); + } +} + +async function superviseCache(data: RetirementWatchdog): Promise { + const retire = await locatorRetirement(data.locator); + const state = new Int32Array(data.state); + parentPort!.once("message", () => { + setTimeout(() => { + if (Atomics.compareExchange(state, 0, FLUSHING, RETIRING) === FLUSHING) { + void exitWithoutCache(retire); + } + }, CACHE_FLUSH_BUDGET_MS); + }); + parentPort!.postMessage(null); +} + +// Called only after the listener closes and every admitted effect settles. +export async function retireRuntime(locator: RuntimeLocator): Promise { + const retire = await locatorRetirement(locator); + if (getCompileCacheDir() === undefined) { + await retire(); + return; + } + const state = new Int32Array(new SharedArrayBuffer(4)); + let watchdog: Worker | undefined; + try { + watchdog = new Worker(new URL(import.meta.url), { + workerData: { kind: "compile_cache_retirement", locator, state: state.buffer } satisfies RetirementWatchdog, + execArgv: ["--no-warnings", "--experimental-strip-types"], + env: { ...process.env, NODE_DISABLE_COMPILE_CACHE: "1" }, + }); + await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("cache supervisor unavailable")), CACHE_FLUSH_BUDGET_MS); + watchdog!.once("message", () => { clearTimeout(timer); resolve(); }); + watchdog!.once("error", error => { clearTimeout(timer); reject(error); }); + watchdog!.once("exit", () => { clearTimeout(timer); reject(new Error("cache supervisor exited")); }); + }); + } catch { + // Cache persistence is optional even when worker startup is unavailable. + void watchdog?.terminate(); + await exitWithoutCache(retire); + return; + } + watchdog.postMessage(null); + try { + flushCompileCache(); + } catch { + if (Atomics.compareExchange(state, 0, FLUSHING, FLUSHED) === FLUSHING) { + await watchdog.terminate(); + await exitWithoutCache(retire); + return; + } + } + if (Atomics.compareExchange(state, 0, FLUSHING, FLUSHED) === RETIRING) { + // The watchdog owns fenced cleanup now. Never terminate it while it holds + // the locator lock, even if the native flush finishes just after its timer. + await new Promise(resolve => watchdog.once("exit", () => resolve())); + await exitWithoutCache(retire); + return; + } + await watchdog.terminate(); + await retire(); +} function privateDirectory(stat: Stats): boolean { return stat.isDirectory() && !stat.isSymbolicLink() && @@ -32,4 +145,8 @@ function enableRuntimeCompileCache(): void { } } -enableRuntimeCompileCache(); +if (!isMainThread && workerData?.kind === "compile_cache_retirement") { + void superviseCache(workerData as RetirementWatchdog); +} else { + enableRuntimeCompileCache(); +} diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index 2b90e2b5f8..301909bf0e 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -1,7 +1,6 @@ import { writeSync } from "node:fs"; import { createServer, type Socket } from "node:net"; -import { flushCompileCache } from "node:module"; -import { readFile, rm, type FileHandle } from "node:fs/promises"; +import { type FileHandle } from "node:fs/promises"; import type { JsonObject } from "./effect_program.ts"; import { @@ -13,6 +12,7 @@ import { effectRuntimeErrorPayload, } from "./effect_runtime_errors.ts"; import { atomicWriteJson, withFileMutationLock } from "./effect_runtime_io.ts"; +import { retireRuntime } from "./effect_runtime_compile_cache.ts"; import { sqliteRuntimeIdentity } from "./coordination/sqlite_runtime.ts"; import {openPrivateResponseSink, readPrivateJsonSnapshot, writePrivateResponse} from "./effect_runtime_snapshot.ts"; import { @@ -269,24 +269,7 @@ server.on("close", () => { if (pendingRequests > 0) { await new Promise((resolve) => { resolveDrain = resolve; }); } - // Retiring the locator lets a fixture remove the private namespace. Flush - // compiled code first, so Node's exit does not recreate a cleaned cache. - // This optional optimization never decides whether shutdown is permitted. - flushCompileCache(); - await withFileMutationLock(infoPath, async () => { - let published: Record; - try { - published = JSON.parse(await readFile(infoPath, "utf8")); - } catch { - return; - } - // A timed-out client may have published a replacement server. The old - // server must never erase that server's locator when it finally exits. - if (published.token === token && published.pid === process.pid && - published.fingerprint === fingerprint) { - await rm(infoPath, { force: true }); - } - }); + await retireRuntime({ infoPath, token, pid: process.pid, fingerprint }); })().finally(() => process.exit(0)); }); diff --git a/tests/control_plane/runtime_retirement_diagnostics.py b/tests/control_plane/runtime_retirement_diagnostics.py index 096a869575..573c07abf4 100644 --- a/tests/control_plane/runtime_retirement_diagnostics.py +++ b/tests/control_plane/runtime_retirement_diagnostics.py @@ -43,19 +43,25 @@ def instrument_copied_runtime(root: Path) -> None: for before, after in ( ('shutdownRequested = true;', 'shutdownRequested = true; fixtureTrace("shutdown_requested");'), ('server.on("close", () => {', 'server.on("close", () => { fixtureTrace("close", {pending: pendingRequests});'), - ('flushCompileCache();', 'fixtureTrace("flush_enter"); flushCompileCache(); fixtureTrace("flush_done");'), - ('published = JSON.parse(await readFile(infoPath, "utf8"));', - 'published = JSON.parse(await readFile(infoPath, "utf8")); fixtureTrace("locator_read_done");'), - ('} catch {\n return;\n }\n // A timed-out client', - '} catch (error) {\n fixtureTrace("locator_read_error", {code: fixtureCode(error)});\n return;\n }\n // A timed-out client'), - ('await rm(infoPath, { force: true });', - 'fixtureTrace("locator_unlink_enter");\n try { await rm(infoPath, { force: true }); fixtureTrace("locator_unlink_done"); }\n' - ' catch (error) { fixtureTrace("locator_unlink_error", {code: fixtureCode(error)}); throw error; }'), ('})().finally(() => process.exit(0));', '})().finally(() => { fixtureTrace("cleanup_finally"); process.exit(0); });'), ): text = _replace(text, before, after) server.write_text(text) + cache = root / "effect_runtime_compile_cache.ts" + text = _replace(cache.read_text(), 'const FLUSHING =', TRACE + '\nconst FLUSHING =') + for before, after in ( + ('flushCompileCache();', 'fixtureTrace("flush_enter"); flushCompileCache(); fixtureTrace("flush_done");'), + ('published = JSON.parse(await readFile(locator.infoPath, "utf8"));', + 'published = JSON.parse(await readFile(locator.infoPath, "utf8")); fixtureTrace("locator_read_done");'), + ('} catch {\n return;\n }\n // Both retirement paths', + '} catch (error) {\n fixtureTrace("locator_read_error", {code: fixtureCode(error)});\n return;\n }\n // Both retirement paths'), + ('await rm(locator.infoPath, { force: true });', + 'fixtureTrace("locator_unlink_enter");\n try { await rm(locator.infoPath, { force: true }); fixtureTrace("locator_unlink_done"); }\n' + ' catch (error) { fixtureTrace("locator_unlink_error", {code: fixtureCode(error)}); throw error; }'), + ): + text = _replace(text, before, after) + cache.write_text(text) io = root / "effect_runtime_io.ts" text = io.read_text() text = _replace(text, 'const MUTATION_LOCK_TIMEOUT_MS =', TRACE + '\nconst MUTATION_LOCK_TIMEOUT_MS =') diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index 4a8ec22b06..4fd5ecc676 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -1332,11 +1332,27 @@ def test_managed_runtime_releases_memory_after_idle_timeout( @pytest.mark.parametrize("shutdown", [False, True], ids=["idle", "shutdown"]) @pytest.mark.parametrize("disconnect", [False, True], ids=["connected", "timed-out"]) +@pytest.mark.parametrize("cache_fault", [None, "blocked", "throws"], ids=["cache", "blocked-cache", "failed-cache"]) def test_runtime_drains_admitted_write_before_exit( tmp_path: Path, monkeypatch, request, capsys, shutdown: bool, disconnect: bool, + cache_fault: str | None, ) -> None: """Socket lifetime must not define the lifetime of an accepted effect.""" - _, captures = observe_copied_runtime(tmp_path, monkeypatch, request) + root, captures = observe_copied_runtime(tmp_path, monkeypatch, request) + if cache_fault: + # Exercise cache failure even under an outer coverage/disabled-cache run, + # using only this fixture's private cache rather than an inherited one. + monkeypatch.delenv("NODE_DISABLE_COMPILE_CACHE", raising=False) + monkeypatch.delenv("NODE_COMPILE_CACHE", raising=False) + monkeypatch.setenv("NODE_V8_COVERAGE", "") + # Fault only the optional cache call in this disposable real server. + # The accepted write, result transport and locator locks stay real. + cache = root / "effect_runtime_compile_cache.ts" + source = cache.read_text() + assert source.count("flushCompileCache();") == 1 + fault = ('Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 60_000);' + if cache_fault == "blocked" else 'throw new Error("fixture cache failure");') + cache.write_text(source.replace("flushCompileCache();", fault, 1)) runtime_dir = tmp_path / "runtime" monkeypatch.setattr(effect_runtime, "_runtime_dir", lambda: runtime_dir) monkeypatch.setenv("LOOPX_EFFECT_RUNTIME_IDLE_MS", "250") @@ -1406,6 +1422,12 @@ def write() -> dict[str, object]: while info_path.exists() and time.monotonic() < deadline: time.sleep(0.025) assert not info_path.exists(), "settled runtime must still retire when idle or stopped" + while time.monotonic() < deadline: + effect_runtime._reap_exited_runtime_child(info) + if not effect_runtime._pid_is_alive(original["pid"]): + break + time.sleep(0.025) + assert not effect_runtime._pid_is_alive(original["pid"]), "retirement must release the runtime process" assert json.loads(journal_path.read_text(encoding="utf-8")) == journal assert capsys.readouterr().out == "", "healthy drains must stay quiet" except Exception as error: @@ -1459,6 +1481,7 @@ def failed_observation(_): with pytest.raises(effect_runtime.EffectRuntimePermanentIOError): test_runtime_drains_admitted_write_before_exit( tmp_path, monkeypatch, request, capsys, shutdown=True, disconnect=False, + cache_fault=None, ) output = capsys.readouterr().out prefix = "runtime drain diagnostic: " diff --git a/tests/control_plane_ts/effect_runtime_compile_cache.test.ts b/tests/control_plane_ts/effect_runtime_compile_cache.test.ts index a1ecadc25f..da5123a736 100644 --- a/tests/control_plane_ts/effect_runtime_compile_cache.test.ts +++ b/tests/control_plane_ts/effect_runtime_compile_cache.test.ts @@ -1,11 +1,12 @@ import assert from "node:assert/strict"; -import { spawnSync } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; import { chmodSync, lstatSync, mkdtempSync, mkdirSync, readdirSync, readFileSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import test, { type TestContext } from "node:test"; import { pathToFileURL } from "node:url"; +import { setTimeout as delay } from "node:timers/promises"; const preloadSource = new URL( "../../loopx/control_plane/effect_runtime_compile_cache.ts", import.meta.url, @@ -138,3 +139,94 @@ for (const targetMode of [0o755, 0o700, undefined]) { } }); } + +function retirementFixture(t: TestContext, fault: string, disabled = false) { + const { root } = fixture(t); + const infoPath = join(root, "runtime.json"); + const script = join(root, "retire.ts"); + writeFileSync(script, ` + import module from 'node:module'; + import threads from 'node:worker_threads'; + import { writeFileSync } from 'node:fs'; + const input = JSON.parse(process.argv[2]); + if (!input.disabled) module.enableCompileCache(input.cache); + if ((module.getCompileCacheDir() !== undefined) !== !input.disabled) throw new Error('fixture cache state'); + if (input.fault === 'blocked' || input.fault === 'late') { + module.flushCompileCache = () => Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, + input.fault === 'late' ? 1_100 : 60_000); + } else if (input.fault === 'throws' || input.disabled) { + module.flushCompileCache = () => { throw new Error('fixture cache failure'); }; + } + if (input.fault === 'worker' || input.disabled) { + threads.Worker = class { constructor() { throw new Error('fixture worker unavailable'); } }; + } + module.syncBuiltinESMExports(); + const { retireRuntime } = await import(input.preload); + const locator = { infoPath: input.infoPath, token: 'original', pid: process.pid, fingerprint: 'fixture' }; + writeFileSync(input.infoPath, JSON.stringify(locator)); + await retireRuntime(locator); + process.exit(0); + `); + function start() { + const env: NodeJS.ProcessEnv = { ...process.env, NODE_V8_COVERAGE: "" }; + delete env.NODE_COMPILE_CACHE; + delete env.NODE_DISABLE_COMPILE_CACHE; + if (disabled) env.NODE_DISABLE_COMPILE_CACHE = "1"; + const child = spawn(process.execPath, ["--no-warnings", "--experimental-strip-types", script, + JSON.stringify({ preload: preloadSource.href, infoPath, cache: join(root, "compile-cache"), fault, disabled })], { + env, + stdio: ["ignore", "pipe", "pipe"], + }); + t.after(() => { if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); }); + let output = ""; + child.stdout.on("data", chunk => { output += chunk; }); + child.stderr.on("data", chunk => { output += chunk; }); + const exited = new Promise<{ code: number | null; signal: string | null }>((resolve, reject) => { + child.once("error", reject); + child.once("close", (code, signal) => resolve({ code, signal })); + }); + return { child, exited, output: () => output }; + } + return { root, infoPath, start }; +} + +for (const fault of ["healthy", "blocked", "throws", "worker", "disabled"]) { + test(`settled retirement remains available with ${fault} cache persistence`, { timeout: 10_000 }, async t => { + const { root, infoPath, start } = retirementFixture(t, fault, fault === "disabled"); + const run = start(); + const result = await run.exited; + assert.equal(run.output(), ""); + assert.throws(() => lstatSync(infoPath), { code: "ENOENT" }); + if (fault === "healthy" || fault === "disabled") { + assert.equal(result.code, 0); + if (fault === "healthy") assert.ok(cacheFiles(join(root, "compile-cache")).length > 0); + else assert.ok(!readdirSync(root).includes("compile-cache")); + } else { + // Windows reports TerminateProcess as a numeric exit; POSIX names SIGKILL. + assert.ok(result.signal === "SIGKILL" || (process.platform === "win32" && result.code !== 0)); + } + }); +} + +for (const fault of ["blocked", "late"]) { + test(`cache watchdog respects the locator fence and replacement after ${fault} flush`, { timeout: 10_000 }, async t => { + const { infoPath, start } = retirementFixture(t, fault); + const lock = infoPath + ".ts-effect.lock"; + writeFileSync(lock, JSON.stringify({ pid: process.pid, token: "fixture-locator-holder" })); + const run = start(); + const deadline = Date.now() + 3_000; + while (Date.now() < deadline) { + try { JSON.parse(readFileSync(infoPath, "utf8")); break; } catch { await delay(10); } + } + assert.equal(JSON.parse(readFileSync(infoPath, "utf8")).pid, run.child.pid); + await delay(2_000); // Both the watchdog deadline and late native return pass. + assert.equal(run.child.exitCode, null, "retirement must wait for the existing locator fence"); + assert.equal(run.child.signalCode, null); + const replacement = { token: "replacement", pid: process.pid, fingerprint: "fixture" }; + writeFileSync(infoPath, JSON.stringify(replacement)); + rmSync(lock); + await run.exited; + assert.equal(run.output(), ""); + assert.deepEqual(JSON.parse(readFileSync(infoPath, "utf8")), replacement); + }); +}