From 736ba7d960a2b1d60576f466b9b12bf0b7d61302 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 01:47:17 +0800 Subject: [PATCH] fix(runtime): retain cache-free supervision after flush failure Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/development/testing-and-quality.md | 25 +++++++----- .../effect_runtime_compile_cache.ts | 39 ++++++++++--------- .../runtime_retirement_diagnostics.py | 6 ++- .../test_effect_runtime_integration.py | 14 +++++++ .../effect_runtime_compile_cache.test.ts | 10 ++++- 5 files changed, 63 insertions(+), 31 deletions(-) diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index c51dd851e9..ed8d24fd27 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -409,20 +409,27 @@ Effect 后台进程；相同源码指纹可能让源码环境与已安装版本 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 +the real Node cache before removing the locator. A blocked or failed flush +keeps that cache-disabled worker responsible for the same fenced, identity-checked +locator cleanup and termination of the already settled process. Node invokes +the calling environment's [native exit hooks even on self-`SIGKILL`](https://github.com/nodejs/node/blob/v24.21.0/src/node_process_methods.cc#L199-L205), so terminating +the worker and killing from the main thread can re-enter cache persistence. +The regression checks real cache files, not just the exit signal. If no supervisor +can start, locator cleanup and main-thread termination remain best effort: that +fallback cannot promise to bypass the native cache hook. Worker retirement 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， +再清理定位文件。写盘阻塞或失败时，保留已就绪且关闭缓存的 worker，由它复用 +同一锁和精确身份校验清理定位文件，并终止已结算的进程。Node 自发 `SIGKILL` +也先执行调用环境的原生退出钩子，因此先停 worker 再由主线程终止会重入缓存写盘； +回归检查实际缓存文件，不能只检查退出信号。监督 worker 无法启动时，定位文件 +清理和主线程终止仍尽力执行，该回退不能保证绕过原生缓存钩子。Worker 退休可能 +丢弃可选缓存条目，不允许放弃已接纳的业务操作。关闭缓存时无需 worker， 继续普通清理。一秒额度不涵盖定位文件锁、存储等待或操作系统调度；故障注入不能 代替原生平台资格验证，V8 CPU 样本也不测量内核 IO。 diff --git a/loopx/control_plane/effect_runtime_compile_cache.ts b/loopx/control_plane/effect_runtime_compile_cache.ts index 209ac27a8d..1d1e4ab5b8 100644 --- a/loopx/control_plane/effect_runtime_compile_cache.ts +++ b/loopx/control_plane/effect_runtime_compile_cache.ts @@ -43,13 +43,14 @@ async function locatorRetirement(locator: RuntimeLocator): Promise<() => Promise }); } -async function exitWithoutCache(retire: () => Promise): Promise { +async function exitAfterRetirement(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. + // Node runs the calling environment's native exit hooks even for self-kill. + // Only our cache-disabled worker bypasses the main environment's cache. + // The main-thread fallback remains best effort if no supervisor can start. + // A worker shares this PID and cannot target a later process reusing it. process.kill(process.pid, "SIGKILL"); } } @@ -60,7 +61,7 @@ async function superviseCache(data: RetirementWatchdog): Promise { parentPort!.once("message", () => { setTimeout(() => { if (Atomics.compareExchange(state, 0, FLUSHING, RETIRING) === FLUSHING) { - void exitWithoutCache(retire); + void exitAfterRetirement(retire); } }, CACHE_FLUSH_BUDGET_MS); }); @@ -89,30 +90,30 @@ export async function retireRuntime(locator: RuntimeLocator): Promise { watchdog!.once("exit", () => { clearTimeout(timer); reject(new Error("cache supervisor exited")); }); }); } catch { - // Cache persistence is optional even when worker startup is unavailable. + // There is no cache-free environment to terminate from in this fallback. + // Still retire the locator; native exit-time cache flushing is best effort. void watchdog?.terminate(); - await exitWithoutCache(retire); + await exitAfterRetirement(retire); return; } watchdog.postMessage(null); + let flushed = false; try { flushCompileCache(); + flushed = true; } catch { - if (Atomics.compareExchange(state, 0, FLUSHING, FLUSHED) === FLUSHING) { - await watchdog.terminate(); - await exitWithoutCache(retire); - return; - } + // Keep the ready worker armed. Main-thread self-kill would run the native + // flush again, outside the optional-cache allowance we just supervised. } - 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); + if (flushed && Atomics.compareExchange(state, 0, FLUSHING, FLUSHED) === FLUSHING) { + await watchdog.terminate(); + await retire(); return; } - await watchdog.terminate(); - await retire(); + // The worker owns failed or overdue persistence and fenced cleanup. Never + // terminate it while it holds the locator lock, including a late flush return. + await new Promise(resolve => watchdog.once("exit", () => resolve())); + await exitAfterRetirement(retire); } function privateDirectory(stat: Stats): boolean { diff --git a/tests/control_plane/runtime_retirement_diagnostics.py b/tests/control_plane/runtime_retirement_diagnostics.py index 573c07abf4..31717253a2 100644 --- a/tests/control_plane/runtime_retirement_diagnostics.py +++ b/tests/control_plane/runtime_retirement_diagnostics.py @@ -12,7 +12,7 @@ EVENTS = frozenset({"shutdown_requested", "close", "flush_enter", "flush_done", "locator_read_done", "locator_read_error", "locator_unlink_enter", "locator_unlink_done", "locator_unlink_error", "cleanup_finally", "lock_attempt", "lock_acquired", "lock_acquire_error", "lock_release_error", - "lock_release_result", "request_error"}) + "lock_release_result", "request_error", "kill_enter", "kill_returned"}) CODES = frozenset({"EACCES", "EPERM", "ENOENT", "EEXIST", "EIO", "EBUSY", "ENOTEMPTY", "EBADF", "ENOSPC", "EMFILE", "ENFILE", "ETIMEDOUT", "EINTR", "ENOTDIR", "EROFS", "EDQUOT", "EINVAL", "EISDIR", "ENAMETOOLONG", "mutation_lock_timeout"}) @@ -52,6 +52,8 @@ def instrument_copied_runtime(root: Path) -> None: text = _replace(cache.read_text(), 'const FLUSHING =', TRACE + '\nconst FLUSHING =') for before, after in ( ('flushCompileCache();', 'fixtureTrace("flush_enter"); flushCompileCache(); fixtureTrace("flush_done");'), + ('process.kill(process.pid, "SIGKILL");', + 'fixtureTrace("kill_enter", {main_thread: isMainThread}); process.kill(process.pid, "SIGKILL"); fixtureTrace("kill_returned");'), ('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', @@ -135,6 +137,8 @@ def trace_events(raw: bytes) -> list[dict]: item[key] = number if type(value.get("released")) is bool: item["released"] = value["released"] + if type(value.get("main_thread")) is bool: + item["main_thread"] = value["main_thread"] if "code" in value: item["code"] = value["code"] if isinstance(value["code"], str) and value["code"] in CODES else None if "syscall" in value: diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index 42e737ed7f..707aba5b74 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -713,6 +713,16 @@ def test_runtime_retirement_trace_filters_private_stderr() -> None: assert trace_events(raw) == [{"event": "lock_release_error", "pid": 12, "ms": 1, "code": "EPERM"}] +@pytest.mark.parametrize("main_thread", [True, False, "private", 1]) +def test_runtime_retirement_trace_preserves_only_boolean_thread_identity(main_thread) -> None: + payload = {"event": "kill_enter", "main_thread": main_thread, "token": "private"} + raw = ("loopx_fixture_retirement:" + json.dumps(payload) + "\n").encode() + expected = {"event": "kill_enter"} + if type(main_thread) is bool: + expected["main_thread"] = main_thread + assert trace_events(raw) == [expected] + + @pytest.mark.parametrize("details,expected", [ ({"code": "EISDIR", "syscall": "read", "errno": -21}, {"code": "EISDIR", "syscall": "read", "errno": -21}), @@ -1481,6 +1491,10 @@ def write() -> dict[str, object]: 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 + if cache_fault: + assert not any(path.is_file() for path in (runtime_dir / "compile-cache").rglob("*")), ( + "failed optional persistence must not re-enter the main environment's native exit flush" + ) assert capsys.readouterr().out == "", "healthy drains must stay quiet" except Exception as error: try: 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 da5123a736..9a81b41af5 100644 --- a/tests/control_plane_ts/effect_runtime_compile_cache.test.ts +++ b/tests/control_plane_ts/effect_runtime_compile_cache.test.ts @@ -204,13 +204,17 @@ for (const fault of ["healthy", "blocked", "throws", "worker", "disabled"]) { } else { // Windows reports TerminateProcess as a numeric exit; POSIX names SIGKILL. assert.ok(result.signal === "SIGKILL" || (process.platform === "win32" && result.code !== 0)); + if (fault !== "worker") { + assert.deepEqual(cacheFiles(join(root, "compile-cache")), [], + "failed optional persistence must not re-enter the main environment's native exit flush"); + } } }); } -for (const fault of ["blocked", "late"]) { +for (const fault of ["blocked", "late", "throws"]) { test(`cache watchdog respects the locator fence and replacement after ${fault} flush`, { timeout: 10_000 }, async t => { - const { infoPath, start } = retirementFixture(t, fault); + const { root, 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(); @@ -228,5 +232,7 @@ for (const fault of ["blocked", "late"]) { await run.exited; assert.equal(run.output(), ""); assert.deepEqual(JSON.parse(readFileSync(infoPath, "utf8")), replacement); + assert.deepEqual(cacheFiles(join(root, "compile-cache")), [], + "fenced retirement must still bypass the main environment's native cache flush"); }); }