b728fdd627
pm imported runtime_state's private helpers (_lock, _atomic_bytes, _bytes, _digest) at a dozen sites while runtime_state imports pm.environments at module top. The primitives are pm's: move them into the stdlib-only pm.filesystem as lock_fd, durable_write_bytes, read_bytes_or_none and file_digest, and repoint every pm caller. runtime_state keeps the private names only as import aliases: it still calls them through its own globals, and pre-PM updaters load them by these names mid-swap (tests/compat/old_updater_surface.json). Boot-subset test fixtures now copy pm/filesystem.py, since runtime_state imports it at process boot; worker-injection tests patch the name pm.publication now reads.
307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""The lockfile (versions + hashes, machine-written) and the installed-state
|
|
file (what is actually on this machine).
|
|
|
|
lock.json: {"schema": 1, "packages": {name: {"version": ..., "artifacts": {target: {"url": ..., "sha256": ...}}}}}
|
|
facts.json: {"schema": 1, "packages": {name: {"entry": ..., "version": ..., "env": ..., "stamp": ...}}}
|
|
|
|
A target's value is one {"url", "sha256"} object, or a LIST of them when the
|
|
package is split across several archives that must land in one directory
|
|
(llama.cpp's engine zip and its cudart zip: Windows resolves a DLL from the
|
|
loading executable's own directory). Both shapes read back as a list.
|
|
|
|
lock.json artifacts carry the RESOLVED url beside the hash: the lockfile is
|
|
the complete machine interface (nix reads it as pure data), and the python
|
|
url templates are consulted only at `pm lock --bump` time. The "any" target
|
|
key covers target-independent artifacts (npm's tarball).
|
|
|
|
facts.json env values hold {{store}} templates so a CI-built file adopts onto
|
|
any machine by substitution.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
|
|
SCHEMA = 1
|
|
STORE_TOKEN = "{{store}}"
|
|
|
|
|
|
def _read(path: Path, *, strict: bool = False) -> dict:
|
|
try:
|
|
text = path.read_text(encoding="utf-8-sig")
|
|
except FileNotFoundError:
|
|
return {"schema": SCHEMA, "packages": {}}
|
|
except OSError:
|
|
if strict:
|
|
raise
|
|
return {"schema": SCHEMA, "packages": {}}
|
|
try:
|
|
data = json.loads(text)
|
|
if isinstance(data, dict) and data.get("schema") == SCHEMA and isinstance(data.get("packages"), dict):
|
|
return data
|
|
except ValueError:
|
|
pass
|
|
if strict:
|
|
raise ValueError(f"cannot read recorded package state: {path}")
|
|
if text.strip():
|
|
# An unparsable-but-nonempty state file is evidence, not garbage:
|
|
# the next _write would silently discard every installed-state
|
|
# record. Keep the bytes for post-mortem.
|
|
try:
|
|
from pm.paths import store_root
|
|
if path.parent == store_root() and (store_root().parent / "manifest.json").is_file():
|
|
import logging
|
|
logging.getLogger(__name__).warning("invalid shipped state file: %s", path)
|
|
else:
|
|
path.with_suffix(".corrupt").write_text(text, encoding="utf-8")
|
|
except OSError:
|
|
pass
|
|
return {"schema": SCHEMA, "packages": {}}
|
|
|
|
|
|
def _write(path: Path, data: dict) -> None:
|
|
from pm.filesystem import durable_write_bytes
|
|
durable_write_bytes(path, (json.dumps(data, indent=2, sort_keys=True) + "\n").encode("utf-8"))
|
|
|
|
|
|
class StaleLockRow(RuntimeError):
|
|
"""A concurrent editor changed a row that this writer plans to publish."""
|
|
|
|
|
|
class Lockfile:
|
|
"""Snapshot of lock.json with conflict-checked publication of changed pins."""
|
|
|
|
def __init__(self, path: Path):
|
|
self.path = Path(path).resolve()
|
|
self._packages = _read(self.path)["packages"]
|
|
self._base = deepcopy(self._packages)
|
|
self._touched: set[str] = set()
|
|
|
|
def version(self, name: str) -> str | None:
|
|
return (self._packages.get(name) or {}).get("version")
|
|
|
|
def artifacts(self, name: str, target: str) -> list[dict]:
|
|
"""Every archive this target needs, in extraction order. One
|
|
artifact and a list of them are the same thing here — the single
|
|
form is just the common case written short."""
|
|
artifacts = (self._packages.get(name) or {}).get("artifacts") or {}
|
|
found = artifacts.get(target)
|
|
if found is None:
|
|
found = artifacts.get("any")
|
|
if found is None:
|
|
return []
|
|
return list(found) if isinstance(found, list) else [found]
|
|
|
|
def pinned_artifacts(self, name: str) -> dict:
|
|
"""Return the complete pin table, including targets without discovery."""
|
|
return deepcopy((self._packages.get(name) or {}).get("artifacts") or {})
|
|
|
|
def names(self) -> list[str]:
|
|
return sorted(self._packages)
|
|
|
|
def set_pin(self, name: str, version: str, artifacts: dict[str, dict]) -> None:
|
|
self._packages[name] = {"version": version, "artifacts": deepcopy(artifacts)}
|
|
self._touched.add(name)
|
|
|
|
def save(self) -> None:
|
|
"""Merge touched rows under the file's lock, or leave all rows unchanged."""
|
|
from pm.filesystem import lock_fd
|
|
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd = os.open(self.path.with_name(f".{self.path.name}.lock"), os.O_CREAT | os.O_RDWR, 0o600)
|
|
try:
|
|
lock_fd(fd, wait=True)
|
|
current = _read(self.path, strict=True)["packages"]
|
|
updated = dict(current)
|
|
for name in self._touched:
|
|
intended = self._packages[name]
|
|
if current.get(name) != self._base.get(name) and current.get(name) != intended:
|
|
raise StaleLockRow(f"{name} changed in {self.path}; read the lockfile again and retry")
|
|
updated[name] = intended
|
|
if updated != current or not self.path.exists():
|
|
_write(self.path, {"schema": SCHEMA, "packages": updated})
|
|
self._packages = updated
|
|
self._base = deepcopy(updated)
|
|
self._touched.clear()
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def termux_docker_digest() -> str:
|
|
"""The pinned termux-docker image digest (pm package `termux-docker`).
|
|
|
|
Single reader for every consumer -- the build/deb scripts, the builder
|
|
image script, and the release workflow -- so a lock-schema change lands
|
|
once.
|
|
"""
|
|
from pm.paths import lockfile_path
|
|
|
|
return Lockfile(lockfile_path()).version("termux-docker")
|
|
|
|
|
|
class Facts:
|
|
"""The installed-state file. Written only by pm."""
|
|
|
|
def __init__(self, path: Path, *, strict: bool = False):
|
|
self.path = path
|
|
self._packages = _read(path, strict=strict)["packages"]
|
|
|
|
def reload(self) -> None:
|
|
self._packages = _read(self.path, strict=True)["packages"]
|
|
|
|
def get(self, name: str) -> dict | None:
|
|
return self._packages.get(name)
|
|
|
|
def refresh_digests(self, store_root: Path) -> int:
|
|
"""Publish all tool digests after packaging finishes changing their bytes."""
|
|
from pm.store import tree_digest
|
|
|
|
packages = _read(self.path, strict=True)["packages"]
|
|
root = store_root.resolve()
|
|
count = 0
|
|
for name, fact in packages.items():
|
|
if not isinstance(fact, dict):
|
|
raise ValueError(f"invalid payload fact: {name}")
|
|
if "entry" not in fact and "stamp" in fact:
|
|
continue
|
|
entry_name = fact.get("entry")
|
|
artifacts = fact.get("artifacts")
|
|
if (not isinstance(entry_name, str) or entry_name in ("", ".", "..")
|
|
or any(c in entry_name for c in "/\\:")
|
|
or not isinstance(fact.get("version"), str) or not fact["version"]
|
|
or not isinstance(fact.get("target"), str) or not fact["target"]
|
|
or not isinstance(artifacts, list) or not artifacts
|
|
or any(not isinstance(sha, str) or not re.fullmatch(r"[a-f0-9]{64}", sha)
|
|
for sha in artifacts)):
|
|
raise ValueError(f"incomplete payload tool: {name}")
|
|
entry = root / entry_name
|
|
if not entry.is_dir() or not entry.resolve().is_relative_to(root):
|
|
raise ValueError(f"incomplete payload tool: {name}")
|
|
fact["digest"] = tree_digest(entry)
|
|
count += 1
|
|
if not count:
|
|
raise ValueError(f"payload facts carry no tool entries: {self.path}")
|
|
_write(self.path, {"schema": SCHEMA, "packages": packages})
|
|
self._packages = packages
|
|
return count
|
|
|
|
def installed(
|
|
self,
|
|
name: str,
|
|
expected_version: str | None,
|
|
store_root: Path,
|
|
expected_identity=None,
|
|
) -> bool:
|
|
"""``expected_identity`` is (target, tuple(artifact sha256s)) — the
|
|
identity the lock currently pins. When given, the fact must record
|
|
that exact identity: a same-version re-pin, a different target, or
|
|
a fact written before identity existed all count as NOT installed
|
|
and force a reinstall. ``None`` keeps the version+path check."""
|
|
fact = self._packages.get(name)
|
|
if not fact or "entry" not in fact:
|
|
# State facts (``record_state``: venv stamp + extras) share this
|
|
# file with tool facts but own no store entry.
|
|
return False
|
|
if expected_version is not None and fact.get("version") != expected_version:
|
|
return False
|
|
if expected_identity is not None:
|
|
target, shas = expected_identity
|
|
if "target" not in fact or "artifacts" not in fact:
|
|
# Legacy fact: pre-dates digest-bound identity. Not
|
|
# vouchable — force one reinstall.
|
|
return False
|
|
if (fact["target"], tuple(fact["artifacts"])) != (target, tuple(shas)):
|
|
return False
|
|
return (store_root / fact["entry"]).exists()
|
|
|
|
def env_for(self, name: str, store_root: Path) -> dict:
|
|
fact = self._packages.get(name) or {}
|
|
return _resolve(fact.get("env", {}), store_root)
|
|
|
|
def _merge_and_write(self, name: str, fact: dict) -> None:
|
|
"""Read-modify-write against disk so concurrent installs of
|
|
different packages never clobber each other."""
|
|
on_disk = _read(self.path, strict=True)["packages"]
|
|
for key, value in self._packages.items():
|
|
on_disk.setdefault(key, value)
|
|
self._packages = on_disk
|
|
self._packages[name] = fact
|
|
_write(self.path, {"schema": SCHEMA, "packages": self._packages})
|
|
|
|
def record(
|
|
self,
|
|
name: str,
|
|
version: str,
|
|
entry: str,
|
|
env: dict,
|
|
store_root: Path,
|
|
target: str | None = None,
|
|
artifacts: list[str] | None = None,
|
|
digest: str | None = None,
|
|
) -> None:
|
|
"""``target``/``artifacts`` record the identity this install came
|
|
from (work item 1); ``digest`` is the realized tree digest of the
|
|
published entry (work item 2). All three are additive — schema
|
|
stays 1 and older facts without them still read back."""
|
|
fact: dict = {
|
|
"entry": entry,
|
|
"version": version,
|
|
"env": _templatize(env, store_root),
|
|
}
|
|
if target is not None:
|
|
fact["target"] = target
|
|
if artifacts is not None:
|
|
fact["artifacts"] = list(artifacts)
|
|
if digest is not None:
|
|
fact["digest"] = digest
|
|
self._merge_and_write(name, fact)
|
|
|
|
def record_state(
|
|
self, name: str, stamp: str, extras: list[str], *,
|
|
environment: Path | None = None, resolved_lock: Path | None = None,
|
|
) -> None:
|
|
"""Commit a verified state package and its selected environment atomically."""
|
|
fact = {"stamp": stamp, "extras": extras}
|
|
if environment is not None:
|
|
fact["environment"] = str(environment.resolve())
|
|
if resolved_lock is not None:
|
|
fact["resolved_lock"] = str(resolved_lock.resolve())
|
|
self._merge_and_write(name, fact)
|
|
|
|
def retain(self, names: set[str]) -> None:
|
|
"""Drop unselected package facts from an exclusively owned staged store."""
|
|
packages = _read(self.path, strict=True)["packages"]
|
|
retained = {name: fact for name, fact in packages.items() if name in names}
|
|
if retained != packages:
|
|
_write(self.path, {"schema": SCHEMA, "packages": retained})
|
|
self._packages = retained
|
|
|
|
def entries_in_use(self) -> set[str]:
|
|
return {f["entry"] for f in self._packages.values() if "entry" in f}
|
|
|
|
|
|
def _templatize(env: dict, store_root: Path) -> dict:
|
|
root = str(store_root)
|
|
out = {}
|
|
for key, value in env.items():
|
|
if isinstance(value, list):
|
|
out[key] = [str(v).replace(root, STORE_TOKEN) for v in value]
|
|
else:
|
|
out[key] = str(value).replace(root, STORE_TOKEN)
|
|
return out
|
|
|
|
|
|
def _resolve(env: dict, store_root: Path) -> dict:
|
|
root = str(store_root)
|
|
out = {}
|
|
for key, value in env.items():
|
|
if isinstance(value, list):
|
|
out[key] = [str(v).replace(STORE_TOKEN, root) for v in value]
|
|
else:
|
|
out[key] = str(value).replace(STORE_TOKEN, root)
|
|
return out
|