fix(runtime): retain the cache watchdog after flush failures

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。
 

loopx/control_plane/effect_runtime_compile_cache.ts

@@ -43,13 +43,14 @@ async function locatorRetirement(locator: RuntimeLocator): Promise<() => Promise
   });
 }
 
-async function exitWithoutCache(retire: () => Promise<void>): Promise<void> {
+async function exitAfterRetirement(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.
+    // 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<void> {
   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<void> {
       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<void>(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<void>(resolve => watchdog.once("exit", () => resolve()));
+  await exitAfterRetirement(retire);
 }
 
 function privateDirectory(stat: Stats): boolean {

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:

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:

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