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..27c769a7e5 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,20 +352,130 @@ 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 + + +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, 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. + 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; 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 +492,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..434e58bb22 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,153 @@ 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 +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"]) +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"