From 249773887ce8c0903208f8f7308a099914010105 Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Sat, 31 Oct 2026 17:43:41 +0800 Subject: [PATCH 1/2] perf(effect-runtime): reuse a private source digest across processes Every short CLI process re-opened all packaged TS/JSON sources to compute the managed runtime fingerprint. On hosts that scan each open this cost 0.15-0.55 s per heartbeat, quota or status command. On POSIX, record the computed digest in the existing private per-user runtime namespace, keyed by release root, interpreter, schema and the complete source stat snapshot. A later process with the same snapshot reuses it; a key mismatch, symlink, shared namespace or untrusted record falls back to the unchanged full-byte hash and freshness recheck. Sources changed within two seconds are hashed but not recorded, and the memo never creates the namespace. Other platforms keep full reads. The bilingual TypeScript migration RFC records the trade-off and the residual metadata-preserving replacement risk. Signed-off-by: lusendong.6789 Co-authored-by: TRAE CLI --- .../typescript-control-plane-migration-v0.md | 14 +- ...script-control-plane-migration-v0.zh-CN.md | 5 +- loopx/control_plane/effect_runtime.py | 114 +++++++++++++++- .../test_runtime_source_read_batching.py | 123 ++++++++++++++++++ 4 files changed, 249 insertions(+), 7 deletions(-) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 77d425f6c5..11916426e9 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -301,10 +301,16 @@ Canonical summaries prepare recovery history only for role groups that contain a resume condition. Succession still uses complete active and archived lineage. Role summaries reuse request-local lineage facts and the succession owner's existing context fields. That owner validates and indexes the complete graph -before materializing requested evaluations. Source fingerprinting still reads -every admitted byte and rechecks freshness; sequential reads avoid starting a -thread pool in each process. Healthy read commands use canonical registry/path -owners and defer mutation error imports until needed. +before materializing requested evaluations. Source fingerprinting hashes every +admitted byte and rechecks freshness; sequential reads avoid a thread pool. On +POSIX, a private per-user digest keyed by release, interpreter and complete +source metadata lets later processes skip reopening unchanged sources. Only +sources unchanged for two seconds are recorded; a key mismatch or untrusted +record rereads every byte. A different file matching size, mtime and ctime, +such as a retargeted symlink, or a filesystem that does not maintain ctime can +hide a change; other hosts reread in every process. +Healthy read commands use canonical registry/path owners and defer mutation +error imports until needed. Exact detail queries push selection into the existing source projection after complete source validation. Queries that report unfiltered counts retain the complete first projection. diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index daf0138f69..7fbae148e1 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -246,7 +246,10 @@ Canonical 摘要仅在某角色组包含恢复条件时准备恢复历史。 继任计算仍使用完整的活跃与归档血缘。 角色摘要复用一次请求内的血缘事实和继任 owner 的既有上下文字段。 该 owner 先校验并索引完整图，再生成需要返回的评估。 -源码指纹仍读取每个已准入字节并复查新鲜度；顺序读取省去每个进程的线程池创建。 +源码指纹对每个已准入字节求哈希并复查新鲜度；顺序读取省去线程池创建。 +在 POSIX 上，按 release、解释器与完整源码元数据建键的私有用户级摘要，使后续进程无需重新打开未变化的源码。 +仅记录两秒内未变化的源码；键不匹配或记录不可信时重读每个字节。 +size、mtime 与 ctime 均相同的另一文件（例如重新指向的 symlink）或不维护 ctime 的文件系统可能掩盖修改；其他平台每个进程仍全量重读。 正常读取直接使用注册表和路径的 canonical owner，发生错误时才导入相关错误类型。 精确详情查询在完整来源校验后，将筛选下推到既有来源投影。 需报告未筛选数量的查询仍保留完整的首次投影。 diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 873fd6f0be..f3f911d0dc 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -86,6 +86,11 @@ CANONICAL_AUTHORITY_READ_TIMEOUT_SECONDS = 15.0 _RUNTIME_SOURCE_SUFFIXES = frozenset({".json", ".ts"}) _RuntimeSourceSnapshot = tuple[tuple[str, int, int, int], ...] +_SOURCE_DIGEST_SCHEMA_VERSION = "loopx_effect_runtime_source_digest_v1" +_SOURCE_DIGEST_MAX_BYTES = 4096 +# A coarse ctime clock can give a rewrite the timestamp it already had. Sources +# changed this recently are hashed but not recorded for other processes. +_SOURCE_DIGEST_SETTLE_NS = 2_000_000_000 def source_number_tokens(value: object) -> list[list[Any]]: @@ -347,11 +352,113 @@ def _runtime_source_files(root: Path | None = None) -> tuple[str, ...]: return tuple(relative for relative, *_metadata in _runtime_source_snapshot(root)) +def _source_digest_location( + root: str, + snapshot: _RuntimeSourceSnapshot, +) -> tuple[Path, str] | None: + """Name one private digest per release and interpreter, keyed by metadata. + + Only POSIX ctime cannot be set by a writer, so other hosts keep full reads. + """ + + if os.name != "posix": + return None + try: + directory = _runtime_dir() + except OSError: + return None + identity = hashlib.sha256() + for part in (_SOURCE_DIGEST_SCHEMA_VERSION, sys.executable, str(sys.version_info[:3])): + identity.update(part.encode("utf-8") + b"\0") + identity.update(os.fsencode(os.path.normcase(root)) + b"\0") + path = directory / f"source-digest-{identity.hexdigest()[:32]}.json" + identity.update(json.dumps(snapshot, separators=(",", ":")).encode("ascii")) + return path, identity.hexdigest() + + +def _private_directory(path: Path) -> bool: + try: + metadata = os.lstat(path) + except OSError: + return False + return (stat.S_ISDIR(metadata.st_mode) and metadata.st_uid == os.getuid() + and not metadata.st_mode & 0o077) + + +def _read_source_digest(path: Path, key: str) -> str | None: + if not _private_directory(path.parent): + return None + try: + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + except OSError: + return None + try: + with os.fdopen(descriptor, "rb") as file: + metadata = os.fstat(file.fileno()) + if (not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.getuid() + or metadata.st_mode & 0o077): + return None + payload = json.loads(file.read(_SOURCE_DIGEST_MAX_BYTES + 1)) + except (OSError, ValueError, RecursionError): + return None + if (not isinstance(payload, dict) + or payload.get("schema_version") != _SOURCE_DIGEST_SCHEMA_VERSION + or payload.get("key") != key): + return None + fingerprint = payload.get("fingerprint") + if not isinstance(fingerprint, str) or BARE_SHA256_PATTERN.fullmatch(fingerprint) is None: + return None + return fingerprint + + +def _write_source_digest( + path: Path, + key: str, + fingerprint: str, + snapshot: _RuntimeSourceSnapshot, +) -> None: + """Best-effort publication into an existing private runtime namespace.""" + + settled_before = time.time_ns() - _SOURCE_DIGEST_SETTLE_NS + if any(max(mtime, ctime) > settled_before for _name, mtime, ctime, _size in snapshot): + return + if not _private_directory(path.parent): + return + data = json.dumps({ + "schema_version": _SOURCE_DIGEST_SCHEMA_VERSION, + "key": key, + "fingerprint": fingerprint, + }, separators=(",", ":")).encode("ascii") + temporary = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp") + try: + descriptor = os.open( + temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600, + ) + except OSError: + return + try: + with os.fdopen(descriptor, "wb") as file: + file.write(data) + os.replace(temporary, path) + except OSError: + try: + temporary.unlink() + except OSError: + pass + + @lru_cache(maxsize=8) def _runtime_fingerprint_for_snapshot( root: str, snapshot: _RuntimeSourceSnapshot, ) -> str: + # A fresh CLI process would otherwise open every source; on hosts that scan + # each open this dominates short commands. The record is only a memo. + location = _source_digest_location(root, snapshot) + if location is not None: + recorded = _read_source_digest(*location) + if recorded is not None: + return recorded digest = hashlib.sha256() # The daemon's fixed Python adapter context must follow the interpreter # selected by this caller, even when two environments share source files. @@ -360,7 +467,7 @@ def _runtime_fingerprint_for_snapshot( source_root = Path(root) # Lazy imports and Python bridges resolve from this physical release. Equal # bytes in another checkout must not reuse a loader whose release may retire. - # Resolved aliases still share one owner; source bytes remain fully checked. + # Resolved aliases still share one owner; the digest covers every source byte. digest.update(b"loopx_effect_runtime_source_instance_v1\0") digest.update(os.fsencode(os.path.normcase(root))) digest.update(b"\0") @@ -377,7 +484,10 @@ def _runtime_fingerprint_for_snapshot( current_snapshot = _runtime_source_snapshot(source_root) if current_snapshot != snapshot: raise _RuntimeSourceChanged(current_snapshot) - return digest.hexdigest() + fingerprint = digest.hexdigest() + if location is not None: + _write_source_digest(*location, fingerprint, snapshot) + return fingerprint def _runtime_fingerprint() -> str: diff --git a/tests/control_plane/test_runtime_source_read_batching.py b/tests/control_plane/test_runtime_source_read_batching.py index 0e1d2ba91a..013edc19a8 100644 --- a/tests/control_plane/test_runtime_source_read_batching.py +++ b/tests/control_plane/test_runtime_source_read_batching.py @@ -1,6 +1,7 @@ from __future__ import annotations import hashlib +import json import sys import os from pathlib import Path @@ -80,3 +81,125 @@ def read(path): monkeypatch.setattr(Path, "read_bytes", read) assert runtime._runtime_fingerprint() == serial_fingerprint(tmp_path) + + +posix_only = pytest.mark.skipif(os.name != "posix", reason="POSIX ctime keys the memo") + + +def _private_namespace(tmp_path, monkeypatch, *, settle_ns=0): + namespace = tmp_path / "runtime" + namespace.mkdir(mode=0o700) + monkeypatch.setattr(runtime, "_runtime_dir", lambda: namespace) + monkeypatch.setattr(runtime, "_SOURCE_DIGEST_SETTLE_NS", settle_ns) + return namespace + + +def _fresh_process_fingerprint(): + runtime._runtime_fingerprint_for_snapshot.cache_clear() + return runtime._runtime_fingerprint() + + +def _count_source_reads(monkeypatch): + original, reads = runtime.iter_binary_file_reads, [] + + def counted(paths, **kwargs): + for read in original(paths, **kwargs): + reads.append(read.path.name) + yield read + + monkeypatch.setattr(runtime, "iter_binary_file_reads", counted) + return reads + + +@posix_only +def test_fresh_process_reuses_the_private_digest_until_sources_change(tmp_path, monkeypatch): + root = tmp_path / "release" + (root / "nested").mkdir(parents=True) + (root / "a.ts").write_bytes(b"export {};\n") + (root / "nested/b.json").write_bytes(b"{}") + monkeypatch.setattr(runtime, "_control_plane_root", lambda: root) + namespace = _private_namespace(tmp_path, monkeypatch) + reads = _count_source_reads(monkeypatch) + + first = _fresh_process_fingerprint() + assert first == serial_fingerprint(root) and len(reads) == 2 + [record] = namespace.iterdir() + # Locator retirement and fixture cleanup own only runtime-*.json names. + assert record.name.startswith("source-digest-") and not record.match("runtime-*.json") + assert record.stat().st_mode & 0o777 == 0o600 + assert _fresh_process_fingerprint() == first and len(reads) == 2 + + (root / "a.ts").write_bytes(b"export const edited = true;\n") + edited = _fresh_process_fingerprint() + assert edited == serial_fingerprint(root) != first and len(reads) == 4 + (root / "c.ts").write_bytes(b"export {};\n") + added = _fresh_process_fingerprint() + assert added == serial_fingerprint(root) != edited + (root / "nested/b.json").unlink() + removed = _fresh_process_fingerprint() + assert removed == serial_fingerprint(root) != added + assert [path.name for path in namespace.iterdir()] == [record.name] + + copy = tmp_path / "copy" + copy.mkdir() + for name in ("a.ts", "c.ts"): + (copy / name).write_bytes((root / name).read_bytes()) + monkeypatch.setattr(runtime, "_control_plane_root", lambda: copy) + assert _fresh_process_fingerprint() == serial_fingerprint(copy) != removed + + +@posix_only +@pytest.mark.parametrize("defect", ["foreign_mode", "symlink", "shared_namespace", + "corrupt", "stale_key", "invalid_digest"]) +def test_untrusted_digest_record_is_recomputed(tmp_path, monkeypatch, defect): + root = tmp_path / "release" + root.mkdir() + (root / "a.ts").write_bytes(b"export {};\n") + monkeypatch.setattr(runtime, "_control_plane_root", lambda: root) + namespace = _private_namespace(tmp_path, monkeypatch) + path, key = runtime._source_digest_location( + os.fspath(root.resolve()), runtime._runtime_source_snapshot(root)) + forged = {"schema_version": runtime._SOURCE_DIGEST_SCHEMA_VERSION, + "key": "0" * 64 if defect == "stale_key" else key, + "fingerprint": "F" * 64 if defect == "invalid_digest" else "0" * 64} + target = tmp_path / "elsewhere.json" if defect == "symlink" else path + target.write_text("{" if defect == "corrupt" else json.dumps(forged)) + target.chmod(0o644 if defect == "foreign_mode" else 0o600) + if defect == "symlink": + path.symlink_to(target) + if defect == "shared_namespace": + namespace.chmod(0o755) + reads = _count_source_reads(monkeypatch) + + assert _fresh_process_fingerprint() == serial_fingerprint(root) + assert reads == ["a.ts"] + + +@posix_only +def test_unsettled_or_failed_hash_records_no_digest(tmp_path, monkeypatch): + root = tmp_path / "release" + root.mkdir() + for name in ("a.ts", "b.ts"): + (root / name).write_bytes(b"export {};\n") + monkeypatch.setattr(runtime, "_control_plane_root", lambda: root) + namespace = _private_namespace(tmp_path, monkeypatch, settle_ns=3600 * 10**9) + assert _fresh_process_fingerprint() == serial_fingerprint(root) + assert list(namespace.iterdir()) == [] + + monkeypatch.setattr(runtime, "_SOURCE_DIGEST_SETTLE_NS", 0) + original = Path.read_bytes + + def read(path): + if path.name == "b.ts": + raise PermissionError("denied") + return original(path) + + monkeypatch.setattr(Path, "read_bytes", read) + with pytest.raises(PermissionError): + _fresh_process_fingerprint() + assert list(namespace.iterdir()) == [] + + namespace.rmdir() + monkeypatch.setattr(Path, "read_bytes", original) + assert _fresh_process_fingerprint() == serial_fingerprint(root) + assert not namespace.exists(), "the memo never creates the runtime namespace" From 6506e22e06c063f4cf2d8e6b10bf942636f20bb0 Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Sat, 31 Oct 2026 23:25:59 +0800 Subject: [PATCH 2/2] fix(effect-runtime): recheck sources before adopting a digest hit A persisted digest matched the caller's snapshot but skipped the full hash's freshness recheck, so a source edited between the snapshot and the record read could return the previous fingerprint. Recheck before adopting the record so the retry path observes the new topology, and add a deterministic race regression. Signed-off-by: lusendong.6789 Co-authored-by: TRAE CLI --- loopx/control_plane/effect_runtime.py | 10 ++++++- .../test_runtime_source_read_batching.py | 28 +++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index f3f911d0dc..27c769a7e5 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -447,6 +447,12 @@ def _write_source_digest( pass +def _require_unchanged_snapshot(root: Path, snapshot: _RuntimeSourceSnapshot) -> None: + current_snapshot = _runtime_source_snapshot(root) + if current_snapshot != snapshot: + raise _RuntimeSourceChanged(current_snapshot) + + @lru_cache(maxsize=8) def _runtime_fingerprint_for_snapshot( root: str, @@ -454,17 +460,19 @@ def _runtime_fingerprint_for_snapshot( ) -> str: # A fresh CLI process would otherwise open every source; on hosts that scan # each open this dominates short commands. The record is only a memo. + source_root = Path(root) location = _source_digest_location(root, snapshot) if location is not None: recorded = _read_source_digest(*location) if recorded is not None: + # A hit adopts the record only under the full hash's freshness check. + _require_unchanged_snapshot(source_root, snapshot) return recorded digest = hashlib.sha256() # The daemon's fixed Python adapter context must follow the interpreter # selected by this caller, even when two environments share source files. digest.update(sys.executable.encode("utf-8")) digest.update(str(sys.version_info[:3]).encode("ascii")) - source_root = Path(root) # Lazy imports and Python bridges resolve from this physical release. Equal # bytes in another checkout must not reuse a loader whose release may retire. # Resolved aliases still share one owner; the digest covers every source byte. diff --git a/tests/control_plane/test_runtime_source_read_batching.py b/tests/control_plane/test_runtime_source_read_batching.py index 013edc19a8..434e58bb22 100644 --- a/tests/control_plane/test_runtime_source_read_batching.py +++ b/tests/control_plane/test_runtime_source_read_batching.py @@ -148,6 +148,34 @@ def test_fresh_process_reuses_the_private_digest_until_sources_change(tmp_path, assert _fresh_process_fingerprint() == serial_fingerprint(copy) != removed +@posix_only +def test_source_change_before_a_digest_hit_is_not_adopted(tmp_path, monkeypatch): + root = tmp_path / "release" + root.mkdir() + edited = root / "a.ts" + edited.write_bytes(b"export {};\n") + (root / "b.json").write_bytes(b"{}") + monkeypatch.setattr(runtime, "_control_plane_root", lambda: root) + namespace = _private_namespace(tmp_path, monkeypatch) + first = _fresh_process_fingerprint() + assert first == serial_fingerprint(root) and len(list(namespace.iterdir())) == 1 + original, hits = runtime._read_source_digest, [] + + def edit_after_snapshot(path, key): + # The caller's snapshot is already taken; the stored key still matches it. + recorded = original(path, key) + if not hits: + edited.write_bytes(b"export const edited = true;\n") + hits.append(recorded) + return recorded + + monkeypatch.setattr(runtime, "_read_source_digest", edit_after_snapshot) + reads = _count_source_reads(monkeypatch) + + assert _fresh_process_fingerprint() == serial_fingerprint(root) != first + assert hits[0] == first and hits[1:] == [None] and reads == ["a.ts", "b.json"] + + @posix_only @pytest.mark.parametrize("defect", ["foreign_mode", "symlink", "shared_namespace", "corrupt", "stale_key", "invalid_digest"])