Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions examples/coordination/sqlite-capacity-report.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,44 @@ export interface CapacityScanInput {
skipped_private_files: number;
}

/**
* Split a reader command emitted by the quota context-capture navigation the
* same way its generator quotes it. The producer builds commands with POSIX
* `shlex.join`, so a quoted path — including a temporary directory containing
* spaces or an apostrophe — is exactly one argument, never split on its own
* whitespace. Unbalanced quoting fails loudly instead of mis-parsing a
* reference path into several arguments.
*/
export function splitShellCommand(command: string): string[] {
const argv: string[] = [];
let current = "";
let started = false;
let quote: "'" | "\"" | null = null;
const characters = Array.from(command);
for (let index = 0; index < characters.length; index++) {
const character = characters[index]!;
if (quote === "'") {
if (character === "'") quote = null;
else current += character;
} else if (quote === "\"") {
if (character === "\"") quote = null;
else if (character === "\\" && index + 1 < characters.length &&
"\"\\$`".includes(characters[index + 1]!)) current += characters[++index]!;
else current += character;
} else if (character === "'" || character === "\"") {
quote = character; started = true;
} else if (character === " " || character === "\t") {
if (started || current !== "") argv.push(current);
current = ""; started = false;
} else {
current += character; started = true;
}
}
if (quote !== null) throw new Error(`unbalanced ${quote} in reader command`);
if (started || current !== "") argv.push(current);
return argv;
}

export function sameCapacityScan(left: CapacityScanInput | undefined, right: CapacityScanInput | undefined): boolean {
return !!left && !!right && /^[a-f0-9]{64}$/.test(left.sha256) &&
Number.isSafeInteger(left.files) && left.files > 0 &&
Expand Down
65 changes: 55 additions & 10 deletions examples/coordination/sqlite-capacity.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@ import {selectLocalSqliteAuthority} from "../../loopx/control_plane/coordination
import {engageLegacyCoordinationWriterFence} from "../../loopx/control_plane/coordination/legacy_writer_fence.ts";
import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts";
import {authorityProjectionFixture} from "../../tests/control_plane_ts/authority_projection_fixture.ts";
import {CAPACITY_PROFILES, capacityLedger, latency, type CapacityAxis, type CapacityProfileId,
type QualificationRow} from "./sqlite-capacity-report.ts";
import {CAPACITY_PROFILES, capacityLedger, latency, splitShellCommand,
type CapacityAxis, type CapacityProfileId, type QualificationRow} from "./sqlite-capacity-report.ts";

const {values: options} = parseArgs({options: {
profile: {type: "string", default: "rehearsal"}, output: {type: "string"},
Expand Down Expand Up @@ -148,23 +148,54 @@ async function measureAxis(count: number): Promise<CapacityAxis> {
let phase = "setup", managedRuntimeUsed = false;
const stopRuntime = () => {
if (!managedRuntimeUsed) return;
// Cleanup deletes the runtime's whole temporary namespace, so the stop is
// anchored to observed process exit: a shutdown receipt alone let the
// runtime's exit-time lock retirement mkdir the axis directory back after
// cleanup_verified was already recorded. Reuse the disposable-owner stop
// interface and require its result to prove the exit before any delete.
const child = spawnSync(options.python!, ["-c",
"from loopx.control_plane.effect_runtime import effect_runtime_result; effect_runtime_result('runtime.shutdown', {}, retry_safe=False)"],
"import json; from loopx.control_plane.effect_runtime import restart_effect_runtime; "
+ "print(json.dumps(restart_effect_runtime(wait_for_exit=True)))"],
{cwd: repository, env: environment, encoding: "utf8", timeout: 30000});
assert.equal(child.status, 0, "isolated Effect runtime shutdown failed");
assert.equal((JSON.parse(child.stdout) as Record<string, unknown>).stopped, true,
"isolated Effect runtime did not exit before cleanup");
managedRuntimeUsed = false;
};
const cli = (args: string[], samples: number[]) => {
phase = `cold_cli_${args[0] === "todo" ? "mutation" : args[0]}`;
const start = performance.now(); managedRuntimeUsed = true;
const child = spawnSync(options.python!, ["-m", "loopx.cli", "--registry", registry, "--format", "json", ...args],
const run = (file: string, argv: string[]) => spawnSync(file, argv,
{cwd: repository, env: environment, encoding: "utf8", timeout: 60000, maxBuffer: 4 * 1024 ** 2});
const elapsed = performance.now() - start;
const child = run(options.python!, ["-m", "loopx.cli", "--registry", registry, "--format", "json", ...args]);
if (child.status !== 0) process.stderr.write(`${phase}: process exited ${child.status ?? "without an exit code"}\n`);
assert.equal(child.status, 0, `cold CLI ${args[0]} failed`);
const result = JSON.parse(child.stdout) as Record<string, unknown>;
assert.notEqual(result.ok, false, `cold CLI ${args[0]} rejected`);
samples.push(elapsed); stopRuntime(); return result;
if (result.schema_version === "context_capture_navigation_v1") {
// `quota should-run` delivers its payload through a context capture:
// follow the returned reader commands page by page until eof, then parse
// the reassembled original JSON. The sample keeps the whole navigation
// cost, because a cold reader cannot act before the payload arrives.
let follow = splitShellCommand(String(result.command ?? result.first_read_command ?? ""));
assert.ok(follow.length > 0, `cold CLI ${args[0]} navigation returned no read command`);
let payload = "";
for (let pages = 0; ; pages++) {
assert.ok(pages < 256, `cold CLI ${args[0]} context read exceeded the page guard`);
const read = run(follow[0]!, follow.slice(1));
if (read.status !== 0) process.stderr.write(`${phase}: context read exited ${read.status ?? "without an exit code"}\n`);
assert.equal(read.status, 0, `cold CLI ${args[0]} context read failed`);
const page = JSON.parse(read.stdout) as Record<string, unknown>;
assert.equal(page.ok, true, `cold CLI ${args[0]} context read rejected`);
payload += typeof page.content === "string" ? page.content : "";
if (page.eof === true) break;
const next = page.next_command;
assert.ok(typeof next === "string" && next.length > 0, `cold CLI ${args[0]} context page lacks the next command before eof`);
follow = splitShellCommand(next);
}
samples.push(performance.now() - start); stopRuntime(); return JSON.parse(payload) as Record<string, unknown>;
}
samples.push(performance.now() - start); stopRuntime(); return result;
};
try {
assert.equal((await selectLocalSqliteAuthority(runtime, goal, true)).ok, true);
Expand Down Expand Up @@ -382,12 +413,26 @@ async function measureAxis(count: number): Promise<CapacityAxis> {
}
axis.status = "passed";
} catch (error) {
axis.failure = `${phase}: ${error instanceof Error ? error.name : "unknown failure"}`;
axis.failure = `${phase}: ${error instanceof Error ? `${error.name}: ${error.message.slice(0, 600)}` : "unknown failure"}`;
} finally {
try { stopRuntime(); } catch { axis.status = "failed"; axis.failure = "isolated_runtime_cleanup_failed"; }
// A stop that cannot prove observed process exit fails the axis, but the
// first recorded failure (the measurement error) keeps precedence.
try { stopRuntime(); }
catch { axis.status = "failed"; if (!axis.failure) axis.failure = "isolated_runtime_cleanup_failed"; }
axis.resource_peak_rss_bytes = process.resourceUsage().maxRSS * 1024;
try { rmSync(root, {recursive: true, force: true}); axis.cleanup_verified = !existsSync(root); }
catch { axis.status = "failed"; axis.failure = "temporary_database_cleanup_failed"; }
// A just-stopped managed runtime can leave a grandchild briefly holding the
// temp root (for example its working directory), which makes an immediate
// recursive delete raise while a retry a moment later succeeds. Retry a few
// times before declaring the disposable database uncleaned.
for (let attempt = 0; ; attempt++) {
try { rmSync(root, {recursive: true, force: true}); } catch { /* retried below */ }
axis.cleanup_verified = !existsSync(root);

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] 先等待原进程退出再确认目录清理。将真实退役 mkdir 延后 500 毫秒后,两个轴均报 cleanup_verified=true,已收到关闭回执的原进程随后分别重建目录。同一延迟下使用 restart_effect_runtime(wait_for_exit=True) 后,两组目录均无残留。请复用该接口,并保留超时及首个测量错误。

if (axis.cleanup_verified || attempt >= 4) break;
Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 250);
}
if (!axis.cleanup_verified && axis.status !== "failed") {
axis.status = "failed"; axis.failure = "temporary_database_cleanup_failed";
}
}
return axis;
}
36 changes: 35 additions & 1 deletion tests/control_plane_ts/sqlite_capacity.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,47 @@ import {tmpdir} from "node:os";
import {performance} from "node:perf_hooks";
import {fileURLToPath} from "node:url";
import test from "node:test";
import {capacityLedger, latency, sameCapacityScan, type CapacityAxis} from "../../examples/coordination/sqlite-capacity-report.ts";
import {capacityLedger, latency, sameCapacityScan, splitShellCommand,
type CapacityAxis} from "../../examples/coordination/sqlite-capacity-report.ts";

test("latency has explicit nearest-rank tails and rejects absent or invalid samples", () => {
assert.deepEqual(latency([5, 1, 4, 2, 3]), {n: 5, p50_ms: 3, p95_ms: 5, p99_ms: 5});
for (const samples of [[], [NaN], [Infinity], [-1]]) assert.throws(() => latency(samples));
});

test("reader commands split exactly as shlex.join quotes them", () => {
assert.deepEqual(splitShellCommand("python3 -m loopx.cli --format json quota context-read --offset 0"),
["python3", "-m", "loopx.cli", "--format", "json", "quota", "context-read", "--offset", "0"]);
// shlex.join single-quotes a path containing spaces: it stays one argument.
assert.deepEqual(splitShellCommand(
`python3 -m loopx.cli quota context-read --reference-file '${"/tmp/space dir/capacity.reader.json"}' --offset 0`),
["python3", "-m", "loopx.cli", "quota", "context-read", "--reference-file",
"/tmp/space dir/capacity.reader.json", "--offset", "0"]);
// An apostrophe inside the path uses the POSIX '"'"' escape sequence.
assert.deepEqual(splitShellCommand(
`python3 -m loopx.cli quota context-read --reference-file '/tmp/loop'"'"'x y/capacity.reader.json' --offset 0`),
["python3", "-m", "loopx.cli", "quota", "context-read", "--reference-file",
"/tmp/loop'x y/capacity.reader.json", "--offset", "0"]);
// Double-quoted arguments and escaped quotes inside them survive too.
assert.deepEqual(splitShellCommand(`python3 "a b" "say \\"hi\\"" c`), ["python3", "a b", 'say "hi"', "c"]);
// A mis-quoted command fails loudly instead of splitting a reference path.
assert.throws(() => splitShellCommand("python3 --reference-file '/tmp/space dir/x"));
});

test("cleanup is anchored to observed runtime exit and keeps the first failure", async () => {
// The stop lives in the runner script itself: a shutdown receipt alone let
// the retiring runtime's delayed lock mkdir recreate the axis directories
// after cleanup_verified was recorded. The stop must reuse the disposable
// owner's observed-exit interface, check its result, and never let a stop
// failure overwrite the first recorded measurement error.
const source = await readFile(
fileURLToPath(new URL("../../examples/coordination/sqlite-capacity.ts", import.meta.url)), "utf8");
assert.match(source, /restart_effect_runtime\(wait_for_exit=True\)/);
assert.doesNotMatch(source, /runtime\.shutdown/);
assert.match(source, /\.stopped, true,\s*\n?\s*"isolated Effect runtime did not exit before cleanup"/);
assert.match(source, /if \(!axis\.failure\) axis\.failure = "isolated_runtime_cleanup_failed"/);
});

function axis(count: number, payloadBytes = 65536): CapacityAxis {
const sample = (n: number) => ({n, p50_ms: 1, p95_ms: 2, p99_ms: 3});
return {target_commits: count, completed_commits: count, projection_json_bytes: payloadBytes,
Expand Down
Loading