From 1f084e41824031fe719d45fc6006e4edc0a67641 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:09:05 +0800 Subject: [PATCH 1/2] fix(runtime): retry contended Windows lock creation within budget Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/development/testing-and-quality.md | 11 +++ loopx/control_plane/effect_runtime_io.ts | 34 ++++++--- .../effect_runtime_io.test.ts | 71 +++++++++++++++++++ 3 files changed, 105 insertions(+), 11 deletions(-) diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index 279516e9e6..b4c5f484dc 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -426,6 +426,17 @@ Effect 服务先排空已接纳的 handler，再退出。启用编译缓存时 继续普通清理。一秒额度不涵盖定位文件锁、存储等待或操作系统调度；故障注入不能 代替原生平台资格验证，V8 CPU 样本也不测量内核 IO。 +Windows mutation-lock acquisition retries `EPERM` from exclusive creation only +after observing contention, within the existing lock timeout. This covers a +[pending-delete pathname](https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-deletefilew#remarks). +Initial permission denial, errors after creation and other platforms retain +their failure behavior; persistent denial preserves its original error. +Injected filesystem failures complement actual Windows execution, not replace it. + +Windows 文件锁只在已观察到争用后，对独占创建返回的 `EPERM` 使用原锁等待额度重试。 +初次权限拒绝、创建后的错误和其它平台仍直接失败；持续拒绝保留原错误。 +故障注入验证重试边界，原生 Windows 运行仍须单独验收。 + Smoke cleanup owns every valid runtime locator in its newly created private temporary namespace, including runtimes loaded by temporary base checkouts. It uses the existing authenticated shutdown protocol, then observes actual diff --git a/loopx/control_plane/effect_runtime_io.ts b/loopx/control_plane/effect_runtime_io.ts index a8f4b14dc4..33da3223b0 100644 --- a/loopx/control_plane/effect_runtime_io.ts +++ b/loopx/control_plane/effect_runtime_io.ts @@ -321,9 +321,12 @@ export async function acquireFileMutationLock( const token = randomUUID(); const deadline = Date.now() + timeoutMs; let retriedAfterReclaim = false; + let contended = false; while (true) { + let opened = false; try { const handle = await open(lockPath, "wx", 0o600); + opened = true; let identity: CreatedFileIdentity | null = null; try { try { @@ -364,17 +367,26 @@ export async function acquireFileMutationLock( } return { targetPath, lockPath, token }; } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; - const reclaimed = await reclaimStaleMutationLock(lockPath); - // Removing a dead owner is progress, not waiting for a live owner. Allow - // one immediate acquisition even for a zero-wait caller. Bound the retry - // so repeated replacement cannot extend that caller's lock budget. - if (reclaimed && !retriedAfterReclaim) { - retriedAfterReclaim = true; - continue; - } - if (Date.now() >= deadline) { - throw new EffectRuntimeLockTimeoutError(); + const code = (error as NodeJS.ErrnoException).code; + // Windows may deny exclusive creation while a contended lock is pending + // deletion. Retry only the unopened pathname within the original budget; + // initial permission errors and failures after creation remain fatal. + if (!opened && contended && process.platform === "win32" && code === "EPERM") { + if (Date.now() >= deadline) throw error; + } else { + if (code !== "EEXIST") throw error; + if (!opened) contended = true; + const reclaimed = await reclaimStaleMutationLock(lockPath); + // Removing a dead owner is progress, not waiting for a live owner. Allow + // one immediate acquisition even for a zero-wait caller. Bound the retry + // so repeated replacement cannot extend that caller's lock budget. + if (reclaimed && !retriedAfterReclaim) { + retriedAfterReclaim = true; + continue; + } + if (Date.now() >= deadline) { + throw new EffectRuntimeLockTimeoutError(); + } } await new Promise((resolve) => setTimeout( diff --git a/tests/control_plane_ts/effect_runtime_io.test.ts b/tests/control_plane_ts/effect_runtime_io.test.ts index 46fd3fb1f6..31bcc94441 100644 --- a/tests/control_plane_ts/effect_runtime_io.test.ts +++ b/tests/control_plane_ts/effect_runtime_io.test.ts @@ -20,6 +20,77 @@ async function workspace(t: TestContext): Promise { return root; } +for (const scenario of [ + {name: "Windows replacement open recovers", platform: "win32", contended: true, persistent: false, writeFault: false, timeout: 500, code: "EPERM", expected: null}, + {name: "Windows initial permission denial fails", platform: "win32", contended: false, persistent: true, writeFault: false, timeout: 500, code: "EPERM", expected: "EPERM"}, + {name: "Windows persistent replacement denial keeps its error", platform: "win32", contended: true, persistent: true, writeFault: false, timeout: 250, code: "EPERM", expected: "EPERM"}, + {name: "non-Windows replacement denial fails", platform: "linux", contended: true, persistent: false, writeFault: false, timeout: 500, code: "EPERM", expected: "EPERM"}, + {name: "unrelated Windows open failure fails", platform: "win32", contended: true, persistent: false, writeFault: false, timeout: 500, code: "EIO", expected: "EIO"}, + {name: "Windows failure after creation is not retried", platform: "win32", contended: true, persistent: false, writeFault: true, timeout: 500, code: "EPERM", expected: "EPERM"}, + {name: "zero-wait live contention does not gain a retry", platform: "win32", contended: true, persistent: false, writeFault: false, timeout: 0, code: "EPERM", expected: "mutation_lock_timeout"}, +]) { + test(scenario.name, async t => { + const root = await workspace(t), target = join(root, "state"); + // Only the selected open/write failure and platform predicate are injected. + // All lock bytes, handles, identities, live-owner checks and release use + // the production filesystem owner in an isolated child. + const script = ` + import assert from "node:assert/strict"; + import * as fs from "node:fs/promises"; + import {mock} from "node:test"; + const [module, target, scenario] = JSON.parse(process.argv[1]); + const path = target + ".ts-effect.lock"; + let attempts = 0; + if (scenario.contended) { + await fs.writeFile(path, JSON.stringify({pid: process.pid, token: "live-holder"})); + } + const {default: ignored, ...namedExports} = fs; + mock.module("node:fs/promises", {namedExports: {...namedExports, + open: async (file, flags, ...args) => { + if (file !== path || flags !== "wx") return fs.open(file, flags, ...args); + attempts++; + if (scenario.contended && attempts === 1) return fs.open(file, flags, ...args); + if (scenario.contended && attempts === 2) await fs.rm(path); + const fault = Object.assign(new Error("controlled filesystem failure"), + {code: scenario.code, syscall: scenario.writeFault ? "write" : "open"}); + if (!scenario.writeFault && (scenario.persistent || attempts === 2)) throw fault; + const handle = await fs.open(file, flags, ...args); + if (scenario.writeFault) handle.writeFile = async () => { throw fault; }; + return handle; + }, + }}); + const io = await import(module); + Object.defineProperty(process, "platform", {value: scenario.platform}); + let code = null; + const started = Date.now(); // Match the acquisition deadline's clock. + try { + const lock = await io.acquireFileMutationLock(target, process.pid, scenario.timeout); + assert.equal((await io.mutationLockOwner(target)).token, lock.token); + assert.equal(await io.releaseFileMutationLock(target, lock.token), true); + await assert.rejects(fs.stat(path), {code: "ENOENT"}); + } catch (error) { code = error.code ?? String(error); } + if (scenario.writeFault) await assert.rejects(fs.stat(path), {code: "ENOENT"}); + console.log(JSON.stringify({code, attempts, elapsed: Date.now() - started})); + `; + const result = spawnSync(process.execPath, [ + "--no-warnings", "--experimental-test-module-mocks", "--experimental-strip-types", + "--input-type=module", "-e", script, + JSON.stringify([new URL("../../loopx/control_plane/effect_runtime_io.ts", import.meta.url).href, + target, scenario]), + ], {encoding: "utf8", timeout: 10_000}); + assert.equal(result.status, 0, result.stderr); + const observed = JSON.parse(result.stdout.trim()); + assert.equal(observed.code, scenario.expected); + if (scenario.expected === null) assert.equal(observed.attempts, 3); + else if (!scenario.contended || scenario.timeout === 0) assert.equal(observed.attempts, 1); + else if (scenario.persistent) { + assert.ok(observed.attempts > 2); + assert.ok(observed.elapsed >= scenario.timeout); + assert.ok(observed.elapsed < 2_000, "permission denial must keep a bounded wait"); + } else assert.equal(observed.attempts, 2); + }); +} + test("JSONL append starts a new row after an unterminated tail", async t => { const root = await workspace(t); const path = join(root, "index.jsonl"); From ef46fdc835522978be214d9e3517c2eb66b63262 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:09:06 +0800 Subject: [PATCH 2/2] test(runtime): stop the recorded replacement during fixture cleanup Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_effect_runtime_integration.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index 830594738e..42e737ed7f 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -637,10 +637,12 @@ def test_identical_release_does_not_reuse_a_retired_lazy_loader( effect_runtime.restart_effect_runtime() +@pytest.mark.parametrize("cleanup_delay", [0, 0.6], ids=["serving", "idle-retired"]) def test_runtime_decode_change_rotates_identity_and_starts_replacement( tmp_path: Path, monkeypatch, request, + cleanup_delay: float, ) -> None: staged_root, captures = observe_copied_runtime(tmp_path, monkeypatch, request) runtime_dir = tmp_path / "runtime" @@ -661,9 +663,20 @@ def test_runtime_decode_change_rotates_identity_and_starts_replacement( assert original_info != replacement_info assert replacement_info.exists() + replacement_locator = effect_runtime._read_info( + replacement_info, fingerprint=replacement_fingerprint, + ) + assert replacement_locator is not None + runtime_count = len(captures) + time.sleep(cleanup_delay) shutdown_started = time.monotonic() - effect_runtime.effect_runtime_result("runtime.shutdown", {}, retry_safe=False) + # Cleanup owns this exact process. Rediscovery can race its 250 ms idle + # retirement or start a third runtime merely to shut that new one down. + effect_runtime._stop_effect_runtime( + replacement_info, replacement_locator, timeout=2, wait_for_process_exit=True, + ) shutdown_acknowledged = time.monotonic() + assert len(captures) == runtime_count, "cleanup must not start another runtime" deadline = time.monotonic() + 2 while ( original_info.exists() or replacement_info.exists()