fix(runtime): supervise optional cache flush after effect drain

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

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<void>> {
+  // 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<string, unknown>;
+    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<void>): Promise<void> {
+  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<void> {
+  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<void> {
+  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<void>((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<void>(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();
+}

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<void>((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<string, unknown>;
-      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));
 });
 

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 =')

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: "

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);
+  });
+}