diff --git a/loopx/control_plane/runtime/checkpoint_artifacts.ts b/loopx/control_plane/runtime/checkpoint_artifacts.ts index 66a6d02067..2712159c7b 100644 --- a/loopx/control_plane/runtime/checkpoint_artifacts.ts +++ b/loopx/control_plane/runtime/checkpoint_artifacts.ts @@ -134,8 +134,25 @@ function unchanged(a: unknown, b: unknown, message = "checkpoint source moved; r } async function absentDestination(path: string): Promise { await physical(dirname(path)); - try {await lstat(path); throw new Error("checkpoint destination already exists");} - catch (error) {if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;} + try {await lstat(path);} + catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return; + throw error; + } + throw new EffectRuntimeConflictError("checkpoint destination already exists"); +} +async function publishCheckpointDirectory(staging: string, destination: string): Promise { + try {await publishNewDirectory(staging, destination);} + catch (error) { + const code = (error as NodeJS.ErrnoException).code; + const permissionFailure = code === "EACCES" || code === "EPERM"; + // A directory rename can report a permission error for an occupied target. + // Confirm the entry exists before changing the meaning of that error. + const occupied = code === "EEXIST" || code === "ENOTEMPTY" || + (permissionFailure && await lstat(destination).then(() => true, () => false)); + if (occupied) throw new EffectRuntimeConflictError("checkpoint destination already exists"); + throw error; + } } async function sqliteFile(path: string): Promise { const handle = await open(path, "r"); @@ -427,7 +444,7 @@ export async function checkpointArtifacts(value: unknown): Promise { await copyComponent(root, join(staging, "unit"), false); await verify(join(staging, "unit"), request.expected_sha256); withinDeadline(); - await publishNewDirectory(join(staging, "unit"), destination); + await publishCheckpointDirectory(join(staging, "unit"), destination); return {...verification, materialized: true, activation_performed: false}; } const sources = components(request.components); @@ -462,7 +479,7 @@ export async function checkpointArtifacts(value: unknown): Promise { await verify(staging, manifest.sha256); unchanged(before, await witness(sources)); withinDeadline(); - await publishNewDirectory(staging, destination); + await publishCheckpointDirectory(staging, destination); return {...await verify(destination, manifest.sha256), source_stability_verified: true}; } finally {await rm(staging, {recursive: true, force: true});} } diff --git a/tests/control_plane/test_checkpoint_artifact_transport.py b/tests/control_plane/test_checkpoint_artifact_transport.py index 0de71f48e3..295381e207 100644 --- a/tests/control_plane/test_checkpoint_artifact_transport.py +++ b/tests/control_plane/test_checkpoint_artifact_transport.py @@ -9,6 +9,7 @@ from loopx.control_plane.effect_runtime import ( MAX_REQUEST_BYTES, MAX_RESPONSE_BYTES, + EffectRuntimeConflict, EffectRuntimeRejected, EffectRuntimeRemoteError, effect_runtime_result, @@ -68,6 +69,12 @@ def test_large_member_transport_with_explicit_rpc_budget_preserves_inventory(tmp assert sorted(p.name for p in (destination / "components/research").iterdir()) == [ row["path"] for row in expected[1:]] member = destination / "components/research" / expected[-1]["path"] + with pytest.raises(EffectRuntimeConflict, match="^checkpoint destination already exists$") as rejected: + effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "materialize", "root": str(unit), "expected_sha256": captured["sha256"], + "destination": str(destination), + }, timeout=LARGE_ARTIFACT_RPC_SECONDS, retry_safe=False, large_local_snapshot=True) + assert rejected.value.diagnostic_code == "state_conflict" assert member.read_bytes() == content member.unlink() with pytest.raises(EffectRuntimeRemoteError): diff --git a/tests/control_plane_ts/checkpoint_artifacts.test.ts b/tests/control_plane_ts/checkpoint_artifacts.test.ts index 1c466387e4..97ad736ab8 100644 --- a/tests/control_plane_ts/checkpoint_artifacts.test.ts +++ b/tests/control_plane_ts/checkpoint_artifacts.test.ts @@ -7,7 +7,7 @@ import {chmod, mkdir, mkdtemp, readFile, readdir, realpath, rm, stat, symlink, w import {tmpdir} from "node:os"; import {join} from "node:path"; import {checkpointArtifacts} from "../../loopx/control_plane/runtime/checkpoint_artifacts.ts"; -import {EffectRuntimeRequestError} from "../../loopx/control_plane/effect_runtime_errors.ts"; +import {EffectRuntimeConflictError, EffectRuntimeRequestError} from "../../loopx/control_plane/effect_runtime_errors.ts"; import {sqliteAuthorityRuntime} from "../../loopx/control_plane/coordination/sqlite_runtime.ts"; async function fixture(t: test.TestContext) { @@ -40,7 +40,9 @@ test("capture, verify and inert materialization preserve executable code and emp assert.deepEqual(await readdir(join(root, "restored/components/code/empty")), []); assert.deepEqual(await readFile(join(root, "restored/checkpoint.json")), await readFile(join(root, "captured/checkpoint.json"))); await assert.rejects(checkpointArtifacts({action: "materialize", root: request.destination, - expected_sha256: captured.sha256, destination: join(root, "restored")}), /already exists/); + expected_sha256: captured.sha256, destination: join(root, "restored")}), { + name: "EffectRuntimeConflictError", code: "state_conflict", message: "checkpoint destination already exists", + }); }); test("a read-only participant's copied mode survives the durability flush and materialization", async t => { @@ -179,7 +181,9 @@ test("concurrent capture reserves one destination and cannot replace an existing const {root, request} = await fixture(t); const results = await Promise.allSettled([checkpointArtifacts(request), checkpointArtifacts(request)]); assert.equal(results.filter(r => r.status === "fulfilled").length, 1); - assert.equal(results.filter(r => r.status === "rejected").length, 1); + const rejected = results.find(result => result.status === "rejected"); + assert.ok(rejected?.reason instanceof EffectRuntimeConflictError, rejected?.reason); + assert.equal(rejected.reason.code, "state_conflict"); assert.deepEqual((await readdir(root)).sort(), ["captured", "code", "conversation"]); const bytes = await readFile(join(root, "captured/checkpoint.json")); await assert.rejects(checkpointArtifacts(request), /already exists/);