object_store.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
33 days ago
| 1 | """Canonical content-addressed object store for the Muse VCS. |
| 2 | |
| 3 | All Muse commands that read or write blobs — ``muse commit``, ``muse read-tree``, |
| 4 | ``muse reset`` — go through this module exclusively. No command may implement |
| 5 | its own path logic or copy its own blobs. |
| 6 | |
| 7 | Layout |
| 8 | ------ |
| 9 | Objects are stored under ``<repo_root>/.muse/objects/`` using a sharded directory |
| 10 | layout that mirrors Git's loose-object format:: |
| 11 | |
| 12 | .muse/objects/<prefix>/<remainder> |
| 13 | |
| 14 | The prefix length is configurable via ``[limits] shard_prefix_length`` in |
| 15 | ``.muse/config.toml``: |
| 16 | |
| 17 | * **2** (default, 256 shards) — matches Git's layout; safe up to ~25 million |
| 18 | objects before any shard exceeds 100 000 entries on ext4. |
| 19 | * **4** (65 536 shards) — use for repos exceeding 10 million objects to keep |
| 20 | per-shard counts below 1 000 at Linux-kernel scale (17 million objects). |
| 21 | |
| 22 | When switching from 2-char to 4-char sharding, existing objects are still |
| 23 | found via a fallback lookup at the 2-char path so no migration is required. |
| 24 | |
| 25 | File permissions |
| 26 | ---------------- |
| 27 | Every object file is written with mode ``0o444`` (read-only for all users). |
| 28 | Content-addressing makes objects immutable by definition; the read-only mode |
| 29 | expresses that constraint at the OS level, preventing accidental truncation by |
| 30 | any tool that opens the file for writing. |
| 31 | |
| 32 | Startup cleanup |
| 33 | --------------- |
| 34 | A call to :func:`cleanup_stale_object_temps` removes any ``.obj-tmp-*`` files |
| 35 | left behind by a previous SIGKILL. :func:`objects_dir` calls this automatically |
| 36 | on the first access so callers never need to invoke it directly. |
| 37 | |
| 38 | This module is the single source of truth for all local object I/O. |
| 39 | The store is append-only: writing the same object twice is always a no-op. |
| 40 | """ |
| 41 | |
| 42 | from __future__ import annotations |
| 43 | |
| 44 | import fcntl |
| 45 | import functools |
| 46 | import hashlib |
| 47 | import logging |
| 48 | import os |
| 49 | import pathlib |
| 50 | import shutil |
| 51 | import sys |
| 52 | import tempfile |
| 53 | import time |
| 54 | |
| 55 | from muse.core.validation import ( |
| 56 | MAX_FILE_BYTES, |
| 57 | MAX_OBJECT_WRITE_BYTES, |
| 58 | assert_not_symlink, |
| 59 | assert_write_inside_repo, |
| 60 | validate_object_id, |
| 61 | ) |
| 62 | |
| 63 | logger = logging.getLogger(__name__) |
| 64 | |
| 65 | _OBJECTS_DIR = "objects" |
| 66 | _DEFAULT_SHARD_PREFIX_LEN: int = 2 |
| 67 | _VALID_SHARD_PREFIX_LENS: frozenset[int] = frozenset({2, 4}) |
| 68 | # Mode applied to every newly-written object: read-only for owner, group, other. |
| 69 | # Content-addressed objects are immutable; this enforces that at the OS level. |
| 70 | _OBJECT_MODE: int = 0o444 |
| 71 | |
| 72 | # Minimum age (seconds) before cleanup_stale_object_temps will delete a temp |
| 73 | # file. This prevents a concurrent process that just started up from deleting |
| 74 | # temp files that belong to in-progress writes from another process. A write |
| 75 | # that cannot complete within 60 seconds is effectively stalled and its temp |
| 76 | # file is genuinely stale. |
| 77 | _CLEANUP_MIN_AGE_SECS: float = 60.0 |
| 78 | |
| 79 | |
| 80 | def _fsync_fd(fd: int) -> None: |
| 81 | """Flush *fd* to durable storage, using the fastest safe syscall available. |
| 82 | |
| 83 | Flushes OS page-cache writes to durable storage before the caller performs |
| 84 | ``os.replace``. Without it, a power loss between the kernel accepting the |
| 85 | rename syscall and flushing the page cache can leave a zero-byte or |
| 86 | partially-written file at the destination even though ``os.replace`` |
| 87 | returned successfully (observed on APFS, ext4, btrfs). |
| 88 | |
| 89 | On macOS, ``os.fsync()`` maps to ``fcntl(F_FULLFSYNC)`` which forces a |
| 90 | full hardware flush — ~4–5 ms per call on APFS. ``F_BARRIERFSYNC`` (85) |
| 91 | provides the same crash-safety guarantee (write barrier to APFS journal) |
| 92 | at ~0.1 ms per call. We prefer it on macOS and fall through to |
| 93 | ``os.fsync()`` only if the kernel rejects it. |
| 94 | |
| 95 | On other platforms ``os.fsync()`` is used directly; on virtual filesystems |
| 96 | (tmpfs, ramfs) where fsync is a no-op it is silently skipped. |
| 97 | """ |
| 98 | try: |
| 99 | if sys.platform == "darwin": |
| 100 | # F_BARRIERFSYNC (85) ensures APFS crash safety without a full |
| 101 | # hardware flush — ~50× faster than F_FULLFSYNC on APFS SSDs. |
| 102 | try: |
| 103 | fcntl.fcntl(fd, 85) # F_BARRIERFSYNC |
| 104 | return |
| 105 | except OSError: |
| 106 | pass # kernel too old or FS doesn't support it — fall through |
| 107 | os.fsync(fd) |
| 108 | except OSError: |
| 109 | pass # best-effort — unsupported on some virtual filesystems |
| 110 | |
| 111 | |
| 112 | def _fsync_path(path: pathlib.Path) -> None: |
| 113 | """``fsync`` a file by path when no file descriptor is available. |
| 114 | |
| 115 | Opens the file read-only solely to obtain an fd, syncs, and closes. |
| 116 | Used after ``shutil.copy2`` which does not expose an fd. |
| 117 | """ |
| 118 | try: |
| 119 | with path.open("rb") as fh: |
| 120 | os.fsync(fh.fileno()) |
| 121 | except OSError: |
| 122 | pass # best-effort — unsupported on some virtual filesystems |
| 123 | |
| 124 | |
| 125 | @functools.lru_cache(maxsize=16) |
| 126 | def _cached_shard_prefix_len(repo_root_str: str) -> int: |
| 127 | """LRU-cached shard prefix length reader. |
| 128 | |
| 129 | The shard prefix length is effectively static for the lifetime of a |
| 130 | process: it is set once in ``.muse/config.toml`` and almost never |
| 131 | changed. Caching eliminates redundant TOML file reads on every call to |
| 132 | :func:`object_path` and :func:`_object_path_with_fallback` — both called |
| 133 | at least twice per :func:`write_object` invocation. |
| 134 | |
| 135 | Cache capacity of 16 covers all realistic multi-repo scenarios in a single |
| 136 | agent process. Values that fail validation fall through to the default and |
| 137 | are never cached with an invalid value. |
| 138 | """ |
| 139 | from muse.cli.config import get_limit |
| 140 | raw = get_limit("shard_prefix_length", pathlib.Path(repo_root_str)) |
| 141 | return raw if raw in _VALID_SHARD_PREFIX_LENS else _DEFAULT_SHARD_PREFIX_LEN |
| 142 | |
| 143 | |
| 144 | def _shard_prefix_len(repo_root: pathlib.Path) -> int: |
| 145 | """Return the configured shard prefix length (2 or 4) for *repo_root*. |
| 146 | |
| 147 | Reads ``[limits] shard_prefix_length`` from ``.muse/config.toml``. |
| 148 | Defaults to ``2`` when absent. Values outside ``{2, 4}`` are ignored |
| 149 | and the default is used. |
| 150 | |
| 151 | Results are cached per repo root for the lifetime of the process — see |
| 152 | :func:`_cached_shard_prefix_len`. |
| 153 | """ |
| 154 | return _cached_shard_prefix_len(str(repo_root)) |
| 155 | |
| 156 | |
| 157 | # --------------------------------------------------------------------------- |
| 158 | # Shard directory validation cache |
| 159 | # --------------------------------------------------------------------------- |
| 160 | |
| 161 | # Set of shard directories that have been created AND path-traversal-validated |
| 162 | # at least once this process run. Key: str(shard_dir). |
| 163 | # |
| 164 | # What is amortised (happens only on the FIRST write per shard): |
| 165 | # • assert_write_inside_repo — expensive resolve() chain (~6 lstat calls) |
| 166 | # • mkdir(parents=True) — stat checks on each parent component |
| 167 | # |
| 168 | # What is NOT amortised (happens on EVERY write): |
| 169 | # • assert_not_symlink(shard_dir) — single lstat, essential TOCTOU guard |
| 170 | # |
| 171 | # Safety rationale: |
| 172 | # assert_write_inside_repo guards against a symlink placed at shard_dir |
| 173 | # BEFORE the first write. assert_not_symlink (below, in write_object) guards |
| 174 | # against a concurrent swap AFTER the first write — because it fires just |
| 175 | # before mkstemp, the race window between the check and the actual write is |
| 176 | # nanoseconds, not milliseconds. Both guards must be in place; only the |
| 177 | # expensive path-resolution check is amortised. |
| 178 | _created_object_shards: set[str] = set() |
| 179 | |
| 180 | |
| 181 | def _ensure_object_shard(repo_root: pathlib.Path, shard_dir: pathlib.Path) -> None: |
| 182 | """Create and path-traversal-validate *shard_dir*, amortised per process run. |
| 183 | |
| 184 | On the first call for a given *shard_dir* this process run: |
| 185 | 1. Validates *shard_dir* resolves inside *repo_root* (path-traversal guard, |
| 186 | prevents a pre-placed symlink from redirecting writes outside the repo). |
| 187 | 2. Creates the directory with ``mkdir -p``. |
| 188 | 3. Records the shard as created so future calls skip steps 1 and 2. |
| 189 | |
| 190 | Subsequent calls for the same *shard_dir* return immediately — the |
| 191 | ``resolve()`` chain and parent ``stat`` calls are not repeated. |
| 192 | |
| 193 | **The caller is still required to call** ``assert_not_symlink(shard_dir)`` |
| 194 | after this function returns. That per-write check catches concurrent |
| 195 | symlink-swap attacks (attacker removes the real shard dir and replaces it |
| 196 | with a symlink between the path-traversal validation and the ``mkstemp`` |
| 197 | call). |
| 198 | |
| 199 | Args: |
| 200 | repo_root: Repository root (parent of ``.muse/``). |
| 201 | shard_dir: The shard subdirectory (e.g. ``.muse/objects/ab/``). |
| 202 | """ |
| 203 | shard_str = str(shard_dir) |
| 204 | if shard_str in _created_object_shards: |
| 205 | return |
| 206 | # First write to this shard this process run: path-traversal check + mkdir. |
| 207 | assert_write_inside_repo(repo_root, shard_dir) |
| 208 | shard_dir.mkdir(parents=True, exist_ok=True) |
| 209 | _created_object_shards.add(shard_str) |
| 210 | |
| 211 | |
| 212 | def cleanup_stale_object_temps(repo_root: pathlib.Path) -> int: |
| 213 | """Remove stale ``.obj-tmp-*`` files left by a previous SIGKILL. |
| 214 | |
| 215 | A SIGKILL between :func:`write_object`'s ``mkstemp`` call and the |
| 216 | ``os.replace`` rename leaves a partial object temp file in the shard |
| 217 | directory. These files are safe to delete because the object was never |
| 218 | committed to the store (the rename never happened). |
| 219 | |
| 220 | **Age gate**: only files older than :data:`_CLEANUP_MIN_AGE_SECS` seconds |
| 221 | are deleted. This prevents a newly-started process from deleting temp files |
| 222 | that belong to in-progress writes from another concurrently-running process |
| 223 | — a genuine risk when multiple agents start simultaneously and one calls |
| 224 | ``require_repo()`` while another is mid-write. A write that cannot complete |
| 225 | within 60 seconds is treated as genuinely stalled. |
| 226 | |
| 227 | This function is called automatically by :func:`objects_dir` on first |
| 228 | access and by :func:`muse.core.repo.require_repo` at startup. |
| 229 | |
| 230 | Args: |
| 231 | repo_root: Root of the Muse repository. |
| 232 | |
| 233 | Returns: |
| 234 | Number of stale temp files removed. |
| 235 | """ |
| 236 | store = repo_root / ".muse" / _OBJECTS_DIR |
| 237 | if not store.exists(): |
| 238 | return 0 |
| 239 | now = time.time() |
| 240 | removed = 0 |
| 241 | for shard in store.iterdir(): |
| 242 | # Skip symlinked shard directories — never delete files inside an |
| 243 | # attacker-controlled location that was swapped in via symlink. |
| 244 | if not shard.is_dir() or shard.is_symlink(): |
| 245 | continue |
| 246 | for f in shard.iterdir(): |
| 247 | if not (f.name.startswith(".obj-tmp-") or f.name.startswith(".restore-tmp-")): |
| 248 | continue |
| 249 | try: |
| 250 | age = now - f.stat().st_mtime |
| 251 | except OSError: |
| 252 | continue # file disappeared between iterdir and stat — skip |
| 253 | if age < _CLEANUP_MIN_AGE_SECS: |
| 254 | logger.debug( |
| 255 | "⏳ Skipping recent temp %s (age %.1fs < %ss — may be in-progress write)", |
| 256 | f.name, age, _CLEANUP_MIN_AGE_SECS, |
| 257 | ) |
| 258 | continue |
| 259 | try: |
| 260 | f.unlink() |
| 261 | removed += 1 |
| 262 | logger.warning( |
| 263 | "⚠️ Removed stale object temp %s (left by prior crash)", f |
| 264 | ) |
| 265 | except OSError as exc: |
| 266 | logger.warning("⚠️ Could not remove stale temp %s: %s", f, exc) |
| 267 | return removed |
| 268 | |
| 269 | |
| 270 | def objects_dir(repo_root: pathlib.Path) -> pathlib.Path: |
| 271 | """Return the path to the local object store root directory. |
| 272 | |
| 273 | The store lives at ``<repo_root>/.muse/objects/``. Shard subdirectories |
| 274 | are created lazily by :func:`write_object` and :func:`write_object_from_path`. |
| 275 | |
| 276 | Stale ``.obj-tmp-*`` files from a prior SIGKILL are swept by |
| 277 | :func:`muse.core.repo.require_repo` at every command startup via |
| 278 | :func:`cleanup_stale_object_temps`. Callers do not need to invoke the |
| 279 | cleanup function directly. |
| 280 | |
| 281 | Args: |
| 282 | repo_root: Root of the Muse repository (the directory containing |
| 283 | ``.muse/``). |
| 284 | |
| 285 | Returns: |
| 286 | Absolute path to the objects directory (may not yet exist). |
| 287 | """ |
| 288 | return repo_root / ".muse" / _OBJECTS_DIR |
| 289 | |
| 290 | |
| 291 | def object_path( |
| 292 | repo_root: pathlib.Path, |
| 293 | object_id: str, |
| 294 | prefix_len: int | None = None, |
| 295 | ) -> pathlib.Path: |
| 296 | """Return the canonical on-disk path for a single object. |
| 297 | |
| 298 | Objects are sharded by the first *prefix_len* hex characters of their |
| 299 | SHA-256 digest:: |
| 300 | |
| 301 | .muse/objects/<prefix>/<remainder> |
| 302 | |
| 303 | When *prefix_len* is ``None`` (the default) the value is read from |
| 304 | ``[limits] shard_prefix_length`` in ``.muse/config.toml`` (default ``2``). |
| 305 | |
| 306 | Args: |
| 307 | repo_root: Root of the Muse repository. |
| 308 | object_id: SHA-256 hex digest of the object's content (64 chars). |
| 309 | prefix_len: Shard prefix length override (2 or 4). ``None`` reads |
| 310 | from config. |
| 311 | |
| 312 | Returns: |
| 313 | Absolute path to the object file (may not yet exist). |
| 314 | |
| 315 | Raises: |
| 316 | ValueError: If *object_id* is not exactly 64 lowercase hex characters. |
| 317 | """ |
| 318 | validate_object_id(object_id) |
| 319 | plen = prefix_len if prefix_len is not None else _shard_prefix_len(repo_root) |
| 320 | return objects_dir(repo_root) / object_id[:plen] / object_id[plen:] |
| 321 | |
| 322 | |
| 323 | def _object_path_with_fallback( |
| 324 | repo_root: pathlib.Path, object_id: str |
| 325 | ) -> pathlib.Path: |
| 326 | """Return the path where *object_id* actually lives, with migration fallback. |
| 327 | |
| 328 | When the configured prefix length is 4 but the object was written with a |
| 329 | 2-char prefix (legacy), this function returns the 2-char path so callers |
| 330 | transparently read existing objects after a sharding upgrade. |
| 331 | |
| 332 | Args: |
| 333 | repo_root: Root of the Muse repository. |
| 334 | object_id: SHA-256 hex digest (64 chars). |
| 335 | |
| 336 | Returns: |
| 337 | The path that exists on disk, or the configured-prefix path if neither |
| 338 | exists (callers can treat a non-existent return as "object absent"). |
| 339 | """ |
| 340 | plen = _shard_prefix_len(repo_root) |
| 341 | primary = object_path(repo_root, object_id, prefix_len=plen) |
| 342 | if primary.exists(): |
| 343 | return primary |
| 344 | # Fallback: if primary uses 4-char prefix, also check 2-char (migration). |
| 345 | if plen != _DEFAULT_SHARD_PREFIX_LEN: |
| 346 | fallback = object_path(repo_root, object_id, prefix_len=_DEFAULT_SHARD_PREFIX_LEN) |
| 347 | if fallback.exists(): |
| 348 | return fallback |
| 349 | return primary |
| 350 | |
| 351 | |
| 352 | def has_object(repo_root: pathlib.Path, object_id: str) -> bool: |
| 353 | """Return ``True`` if *object_id* is present in the local store. |
| 354 | |
| 355 | Cheaper than :func:`read_object` when the caller only needs to check |
| 356 | existence (e.g. to pre-flight a hard reset before touching the working |
| 357 | tree). |
| 358 | |
| 359 | Checks the configured shard path first; if not found and the configured |
| 360 | prefix is 4, also checks the legacy 2-char path for migration compatibility. |
| 361 | |
| 362 | Args: |
| 363 | repo_root: Root of the Muse repository. |
| 364 | object_id: SHA-256 hex digest to check. |
| 365 | """ |
| 366 | return _object_path_with_fallback(repo_root, object_id).exists() |
| 367 | |
| 368 | |
| 369 | def write_object(repo_root: pathlib.Path, object_id: str, content: bytes) -> bool: |
| 370 | """Write *content* to the local object store under *object_id*. |
| 371 | |
| 372 | If the object already exists (same ID = same content, content-addressed) |
| 373 | the write is skipped and ``False`` is returned. Returns ``True`` when a |
| 374 | new object was written. |
| 375 | |
| 376 | The shard directory is created on first write. Subsequent writes for the |
| 377 | same ``object_id`` are no-ops — they never overwrite existing content. |
| 378 | |
| 379 | The content hash is verified against *object_id* before writing to prevent |
| 380 | corrupt or malicious blobs from entering the store. |
| 381 | |
| 382 | Writes are atomic: content is written to a temp file then renamed, |
| 383 | so a crash mid-write never leaves a partial object. |
| 384 | |
| 385 | Args: |
| 386 | repo_root: Root of the Muse repository. |
| 387 | object_id: SHA-256 hex digest that identifies this object (64 chars). |
| 388 | content: Raw bytes to persist. |
| 389 | |
| 390 | Returns: |
| 391 | ``True`` if the object was newly written, ``False`` if it already |
| 392 | existed (idempotent). |
| 393 | |
| 394 | Raises: |
| 395 | ValueError: If *object_id* is not a valid 64-char hex string, or if |
| 396 | the hash of *content* does not match *object_id*. |
| 397 | """ |
| 398 | validate_object_id(object_id) |
| 399 | |
| 400 | # Size cap: reject blobs that exceed the per-object write ceiling before |
| 401 | # doing any I/O. This mirrors the read-time MAX_FILE_BYTES ceiling in |
| 402 | # read_object so the same limit applies at both boundaries. |
| 403 | if len(content) > MAX_OBJECT_WRITE_BYTES: |
| 404 | raise ValueError( |
| 405 | f"Object {object_id[:8]}… is {len(content):,} bytes, exceeding the " |
| 406 | f"{MAX_OBJECT_WRITE_BYTES // (1024 * 1024)} MiB per-object write limit." |
| 407 | ) |
| 408 | |
| 409 | actual = hashlib.sha256(content).hexdigest() |
| 410 | if actual != object_id: |
| 411 | raise ValueError( |
| 412 | f"Content integrity failure: expected object {object_id[:8]}…, " |
| 413 | f"got {actual[:8]}…" |
| 414 | ) |
| 415 | |
| 416 | dest = object_path(repo_root, object_id) |
| 417 | # Also check legacy 2-char path when configured prefix differs. |
| 418 | if _object_path_with_fallback(repo_root, object_id).exists(): |
| 419 | logger.debug("⚠️ Object %s already in store — skipped", object_id[:8]) |
| 420 | return False |
| 421 | |
| 422 | # Amortised path-traversal check + mkdir. The first write per shard calls |
| 423 | # assert_write_inside_repo (expensive resolve chain) and mkdir; subsequent |
| 424 | # writes to the same shard skip both. |
| 425 | _ensure_object_shard(repo_root, dest.parent) |
| 426 | |
| 427 | # Per-write symlink guard: assert the shard is STILL a real directory. |
| 428 | # This catches a concurrent attacker who removes the real shard and |
| 429 | # replaces it with a symlink after our first-write validation. The check |
| 430 | # fires just before mkstemp so the race window is nanoseconds wide. |
| 431 | assert_not_symlink(dest.parent, label=f"object shard directory ({dest.parent.name}/)") |
| 432 | |
| 433 | fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".obj-tmp-") |
| 434 | tmp = pathlib.Path(tmp_str) |
| 435 | try: |
| 436 | with os.fdopen(fd, "wb") as fh: |
| 437 | fh.write(content) |
| 438 | fh.flush() |
| 439 | # Set read-only via fd *before* closing — path-based os.chmod after |
| 440 | # fh.close() is vulnerable to cleanup_stale_object_temps unlinking |
| 441 | # the temp file between the close and the chmod call. os.fchmod |
| 442 | # uses the open fd (immune to concurrent unlink of the path). |
| 443 | os.fchmod(fh.fileno(), _OBJECT_MODE) |
| 444 | _fsync_fd(fh.fileno()) |
| 445 | os.replace(tmp, dest) |
| 446 | except Exception: |
| 447 | tmp.unlink(missing_ok=True) |
| 448 | raise |
| 449 | |
| 450 | logger.debug("✅ Stored object %s (%d bytes)", object_id[:8], len(content)) |
| 451 | return True |
| 452 | |
| 453 | |
| 454 | def write_object_from_path( |
| 455 | repo_root: pathlib.Path, |
| 456 | object_id: str, |
| 457 | src: pathlib.Path, |
| 458 | ) -> bool: |
| 459 | """Copy *src* into the object store without loading it into memory. |
| 460 | |
| 461 | Preferred over :func:`write_object` for large blobs (dense MIDI renders, |
| 462 | audio previews) because ``shutil.copy2`` delegates to the OS copy |
| 463 | mechanism, keeping the interpreter heap clean. |
| 464 | |
| 465 | Idempotent: if the object already exists it is never overwritten. |
| 466 | |
| 467 | The source file's hash is verified against *object_id* before writing. |
| 468 | Writes are atomic (temp file + rename). |
| 469 | |
| 470 | Args: |
| 471 | repo_root: Root of the Muse repository. |
| 472 | object_id: SHA-256 hex digest of *src*'s content (64 chars). |
| 473 | src: Absolute path of the source file to store. |
| 474 | |
| 475 | Returns: |
| 476 | ``True`` if the object was newly written, ``False`` if it already |
| 477 | existed (idempotent). |
| 478 | |
| 479 | Raises: |
| 480 | ValueError: If *object_id* is invalid or the file's hash does not match. |
| 481 | """ |
| 482 | validate_object_id(object_id) |
| 483 | |
| 484 | dest = object_path(repo_root, object_id) |
| 485 | # Also check legacy 2-char path when configured prefix differs. |
| 486 | if _object_path_with_fallback(repo_root, object_id).exists(): |
| 487 | logger.debug("⚠️ Object %s already in store — skipped", object_id[:8]) |
| 488 | return False |
| 489 | |
| 490 | # Guard BEFORE any filesystem modification. |
| 491 | assert_write_inside_repo(repo_root, dest) |
| 492 | |
| 493 | # Size cap: stat the source file before reading it. This avoids loading |
| 494 | # an oversized blob into the hash loop and then discovering it's too large. |
| 495 | src_size = src.stat().st_size |
| 496 | if src_size > MAX_OBJECT_WRITE_BYTES: |
| 497 | raise ValueError( |
| 498 | f"Source file {src.name!r} is {src_size:,} bytes, exceeding the " |
| 499 | f"{MAX_OBJECT_WRITE_BYTES // (1024 * 1024)} MiB per-object write limit." |
| 500 | ) |
| 501 | |
| 502 | # Verify hash before writing. |
| 503 | h = hashlib.sha256() |
| 504 | with src.open("rb") as fh: |
| 505 | for chunk in iter(lambda: fh.read(65536), b""): |
| 506 | h.update(chunk) |
| 507 | actual = h.hexdigest() |
| 508 | if actual != object_id: |
| 509 | raise ValueError( |
| 510 | f"Content integrity failure for {src}: expected {object_id[:8]}…, " |
| 511 | f"got {actual[:8]}…" |
| 512 | ) |
| 513 | |
| 514 | dest.parent.mkdir(parents=True, exist_ok=True) |
| 515 | |
| 516 | # Second guard after mkdir: catch symlink swap between the resolve check |
| 517 | # and mkdir completing. |
| 518 | assert_not_symlink(dest.parent, label=f"object shard directory ({dest.parent.name}/)") |
| 519 | |
| 520 | fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".obj-tmp-") |
| 521 | tmp = pathlib.Path(tmp_str) |
| 522 | try: |
| 523 | os.close(fd) |
| 524 | shutil.copy2(src, tmp) |
| 525 | # Re-open tmp for an fd so we can fchmod and fsync via descriptor rather |
| 526 | # than path. Path-based chmod after close is vulnerable to |
| 527 | # cleanup_stale_object_temps unlinking the temp between the close and |
| 528 | # the chmod call. |
| 529 | with tmp.open("rb") as fh: |
| 530 | os.fchmod(fh.fileno(), _OBJECT_MODE) |
| 531 | _fsync_fd(fh.fileno()) |
| 532 | os.replace(tmp, dest) |
| 533 | except Exception: |
| 534 | tmp.unlink(missing_ok=True) |
| 535 | raise |
| 536 | |
| 537 | logger.debug("✅ Stored object %s (%s)", object_id[:8], src.name) |
| 538 | return True |
| 539 | |
| 540 | |
| 541 | def read_object(repo_root: pathlib.Path, object_id: str) -> bytes | None: |
| 542 | """Read and return the raw bytes for *object_id* from the local store. |
| 543 | |
| 544 | Every read re-verifies the SHA-256 digest of the returned bytes against |
| 545 | *object_id*. If the file has been silently corrupted (disk error, cosmic |
| 546 | ray, filesystem bug, or deliberate tamper) this function raises |
| 547 | ``OSError`` rather than returning bad data. Silent corruption is the |
| 548 | worst possible failure mode for a version-control system. |
| 549 | |
| 550 | Returns ``None`` when the object is not present in the store so callers |
| 551 | can produce a user-facing error rather than raising ``FileNotFoundError``. |
| 552 | |
| 553 | Args: |
| 554 | repo_root: Root of the Muse repository. |
| 555 | object_id: SHA-256 hex digest of the desired object. |
| 556 | |
| 557 | Returns: |
| 558 | Raw bytes, or ``None`` when the object is absent from the store. |
| 559 | |
| 560 | Raises: |
| 561 | ValueError: If *object_id* is not a valid 64-char hex string. |
| 562 | OSError: If the object file exceeds MAX_FILE_BYTES or fails the |
| 563 | SHA-256 integrity check. |
| 564 | """ |
| 565 | validate_object_id(object_id) |
| 566 | dest = _object_path_with_fallback(repo_root, object_id) |
| 567 | if not dest.exists(): |
| 568 | logger.debug("⚠️ Object %s not found in local store", object_id[:8]) |
| 569 | return None |
| 570 | size = dest.stat().st_size |
| 571 | if size > MAX_FILE_BYTES: |
| 572 | raise OSError( |
| 573 | f"Object {object_id[:8]} is {size} bytes, exceeding the " |
| 574 | f"{MAX_FILE_BYTES // (1024 * 1024)} MiB read limit." |
| 575 | ) |
| 576 | # Stream-hash while reading so memory usage stays constant for large blobs. |
| 577 | h = hashlib.sha256() |
| 578 | chunks: list[bytes] = [] |
| 579 | with dest.open("rb") as fh: |
| 580 | for chunk in iter(lambda: fh.read(65536), b""): |
| 581 | h.update(chunk) |
| 582 | chunks.append(chunk) |
| 583 | actual = h.hexdigest() |
| 584 | if actual != object_id: |
| 585 | logger.critical( |
| 586 | "❌ Object %s failed integrity check — store may be corrupt " |
| 587 | "(expected %s, got %s). Do not trust any data from this object.", |
| 588 | object_id[:8], |
| 589 | object_id[:16], |
| 590 | actual[:16], |
| 591 | ) |
| 592 | raise OSError( |
| 593 | f"Object {object_id[:8]}… failed SHA-256 integrity check. " |
| 594 | f"Expected digest starting {object_id[:16]}, got {actual[:16]}. " |
| 595 | "The object store may be corrupt. Run `muse plumbing verify-pack` " |
| 596 | "to audit the full store." |
| 597 | ) |
| 598 | return b"".join(chunks) |
| 599 | |
| 600 | |
| 601 | def restore_object( |
| 602 | repo_root: pathlib.Path, |
| 603 | object_id: str, |
| 604 | dest: pathlib.Path, |
| 605 | ) -> bool: |
| 606 | """Copy an object from the store to *dest*, preserving the destination inode. |
| 607 | |
| 608 | When *dest* already contains the correct content this function returns |
| 609 | immediately **without any write**. Skipping the write keeps the inode |
| 610 | number and mtime unchanged, which is critical for editors (Cursor, VS Code, |
| 611 | Vim, …) that watch open buffers via inode-based filesystem events. A |
| 612 | spurious ``os.replace`` (rename syscall) replaces the inode even when the |
| 613 | bytes are identical; the editor was watching the old inode and goes blind, |
| 614 | leaving a permanently stale buffer that shows the previous file content |
| 615 | until the user closes and reopens the tab. |
| 616 | |
| 617 | When *dest* does not yet exist or contains different content, the write is |
| 618 | done atomically: content is staged to a sibling temp file and renamed into |
| 619 | place via :func:`os.replace` so a partially-written destination is never |
| 620 | visible to other processes. |
| 621 | |
| 622 | Creates parent directories of *dest* if they do not exist. |
| 623 | |
| 624 | The caller is responsible for ensuring *dest* is within a safe base |
| 625 | directory (use :func:`muse.core.validation.contain_path` before calling). |
| 626 | |
| 627 | Args: |
| 628 | repo_root: Root of the Muse repository. |
| 629 | object_id: SHA-256 hex digest of the desired object (64 chars). |
| 630 | dest: Absolute path to write the restored file. |
| 631 | |
| 632 | Returns: |
| 633 | ``True`` on success, ``False`` if the object is not in the store. |
| 634 | |
| 635 | Raises: |
| 636 | ValueError: If *object_id* is not a valid 64-char hex string. |
| 637 | OSError: If the atomic write fails after the object was found. |
| 638 | """ |
| 639 | validate_object_id(object_id) |
| 640 | src = _object_path_with_fallback(repo_root, object_id) |
| 641 | if not src.exists(): |
| 642 | # The empty-content object (SHA-256("") = e3b0c44…) is deterministic — |
| 643 | # its content is always b"". Repos pushed before the server-side |
| 644 | # empty-object fix may be missing it from the store. Synthesize it |
| 645 | # inline so apply_manifest does not fail, and persist it into the store |
| 646 | # for consistency so future calls find it normally. |
| 647 | if object_id == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855": |
| 648 | write_object(repo_root, object_id, b"") |
| 649 | src = _object_path_with_fallback(repo_root, object_id) |
| 650 | if not src.exists(): |
| 651 | return False |
| 652 | else: |
| 653 | logger.debug( |
| 654 | "⚠️ Object %s not found in local store — cannot restore", object_id[:8] |
| 655 | ) |
| 656 | return False |
| 657 | |
| 658 | # Re-verify the source object's hash before restoring it to the working |
| 659 | # tree. The in-store file is 0o444 (immutable) but disk errors or cosmic |
| 660 | # rays can corrupt it silently after it was written. Detecting corruption |
| 661 | # here prevents restoring bad bytes to the working tree without error. |
| 662 | size = src.stat().st_size |
| 663 | if size > MAX_FILE_BYTES: |
| 664 | raise OSError( |
| 665 | f"Object {object_id[:8]} is {size:,} bytes, exceeding the " |
| 666 | f"{MAX_FILE_BYTES // (1024 * 1024)} MiB read limit." |
| 667 | ) |
| 668 | h = hashlib.sha256() |
| 669 | with src.open("rb") as fh: |
| 670 | for chunk in iter(lambda: fh.read(65536), b""): |
| 671 | h.update(chunk) |
| 672 | actual = h.hexdigest() |
| 673 | if actual != object_id: |
| 674 | logger.critical( |
| 675 | "❌ Object %s failed integrity check during restore " |
| 676 | "(expected %s, got %s). Refusing to restore corrupt data.", |
| 677 | object_id[:8], object_id[:16], actual[:16], |
| 678 | ) |
| 679 | raise OSError( |
| 680 | f"Object {object_id[:8]}… failed SHA-256 integrity check during restore. " |
| 681 | f"Expected digest starting {object_id[:16]}, got {actual[:16]}. " |
| 682 | "Run `muse plumbing verify-pack` to audit the store." |
| 683 | ) |
| 684 | |
| 685 | # Idempotent check-before-write: if dest already contains object_id's |
| 686 | # bytes, skip the rename entirely. Preserving the inode keeps any editor |
| 687 | # watching the file alive — an atomic rename (os.replace) always produces a |
| 688 | # new inode, silently blinding inode-based watchers even for identical |
| 689 | # content. Size is checked first as a cheap pre-filter; hashing is only |
| 690 | # done when sizes match, amortising the cost over the common mismatch case. |
| 691 | if dest.exists(): |
| 692 | try: |
| 693 | if dest.stat().st_size == size: |
| 694 | h_dest = hashlib.sha256() |
| 695 | with dest.open("rb") as fh: |
| 696 | for chunk in iter(lambda: fh.read(65536), b""): |
| 697 | h_dest.update(chunk) |
| 698 | if h_dest.hexdigest() == object_id: |
| 699 | logger.debug( |
| 700 | "✅ Object %s already at dest — skipping restore (inode preserved)", |
| 701 | object_id[:8], |
| 702 | ) |
| 703 | return True |
| 704 | except OSError: |
| 705 | pass # dest unreadable or raced away — fall through to the write path |
| 706 | |
| 707 | dest.parent.mkdir(parents=True, exist_ok=True) |
| 708 | fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".restore-tmp-") |
| 709 | tmp = pathlib.Path(tmp_str) |
| 710 | try: |
| 711 | os.close(fd) |
| 712 | shutil.copy2(src, tmp) |
| 713 | # Stored objects are 0o444 (immutable); shutil.copy2 would propagate that |
| 714 | # mode to the restored working-tree file, making it read-only and |
| 715 | # uneditable. Working-tree files must be writable by their owner. |
| 716 | os.chmod(tmp, 0o644) |
| 717 | # shutil.copy2 copies the source mtime, which for object-store files is |
| 718 | # the time the object was originally committed — potentially days or weeks |
| 719 | # in the past. After os.replace the destination would carry that old |
| 720 | # mtime, causing editors (Cursor, VS Code, Vim) to see "new mtime < |
| 721 | # cached mtime" and decide the file was not modified, keeping the stale |
| 722 | # buffer open. Resetting to "now" guarantees the destination always has |
| 723 | # a fresh timestamp, so file-system watchers reliably trigger a buffer |
| 724 | # reload for any content change. |
| 725 | os.utime(tmp, None) |
| 726 | _fsync_path(tmp) |
| 727 | os.replace(tmp, dest) |
| 728 | except Exception: |
| 729 | tmp.unlink(missing_ok=True) |
| 730 | raise |
| 731 | logger.debug("✅ Restored object %s → %s", object_id[:8], dest) |
| 732 | return True |
File History
2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
33 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
33 days ago