gabriel / muse public
object_store.py python
942 lines 37.8 KB
Raw
sha256:fe844c2411edd1cec3d4c847f36a96c6ccd4e3d7d1a715106d2ecd64216bf94f fix: bare object detection and read recovery; rm adapter files Sonnet 4.6 minor ⚠ breaking 44 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 with an algorithm subdirectory::
11
12 .muse/objects/sha256/<prefix>/<remainder>
13
14 The algorithm directory (``sha256/``) makes the on-disk layout self-describing.
15 The prefix length is configurable via ``[limits] shard_prefix_length`` in
16 ``.muse/config.toml``:
17
18 * **2** (default, 256 shards) — matches Git's layout; safe up to ~25 million
19 objects before any shard exceeds 100 000 entries on ext4.
20 * **4** (65 536 shards) — use for repos exceeding 10 million objects to keep
21 per-shard counts below 1 000 at Linux-kernel scale (17 million objects).
22
23 Legacy layout (no algo dir) is transparently supported via fallback — repos
24 written before this layout change are read without any migration.
25
26 When switching from 2-char to 4-char sharding, existing objects are still
27 found via a fallback lookup at the 2-char path so no migration is required.
28
29 File permissions
30 ----------------
31 Every object file is written with mode ``0o444`` (read-only for all users).
32 Content-addressing makes objects immutable by definition; the read-only mode
33 expresses that constraint at the OS level, preventing accidental truncation by
34 any tool that opens the file for writing.
35
36 Startup cleanup
37 ---------------
38 A call to :func:`cleanup_stale_object_temps` removes any ``.obj-tmp-*`` files
39 left behind by a previous SIGKILL. :func:`objects_dir` calls this automatically
40 on the first access so callers never need to invoke it directly.
41
42 This module is the single source of truth for all local object I/O.
43 The store is append-only: writing the same object twice is always a no-op.
44 """
45
46 import fcntl
47 import functools
48 import hashlib
49 import logging
50 import os
51 import pathlib
52 import sys
53 import tempfile
54 import time
55 from collections.abc import Iterator
56
57 from muse.core.types import DEFAULT_HASH_ALGO, blob_id, hash_file, long_id, short_id, split_id
58 from muse.core.paths import objects_dir as _objects_dir
59 from muse.core.validation import (
60 MAX_FILE_BYTES,
61 MAX_OBJECT_WRITE_BYTES,
62 assert_not_symlink,
63 assert_write_inside_repo,
64 validate_object_id,
65 )
66
67 logger = logging.getLogger(__name__)
68 _DEFAULT_SHARD_PREFIX_LEN: int = 2
69 _VALID_SHARD_PREFIX_LENS: frozenset[int] = frozenset({2, 4})
70 # Mode applied to every newly-written object: read-only for owner, group, other.
71 # Content-addressed objects are immutable; this enforces that at the OS level.
72 _OBJECT_MODE: int = 0o444
73 # Hex character set used by iter_stored_objects to reject stray files.
74 _HEX_CHARS: frozenset[str] = frozenset("0123456789abcdef")
75
76 # Minimum age (seconds) before cleanup_stale_object_temps will delete a temp
77 # file. This prevents a concurrent process that just started up from deleting
78 # temp files that belong to in-progress writes from another process. A write
79 # that cannot complete within 60 seconds is effectively stalled and its temp
80 # file is genuinely stale.
81 _CLEANUP_MIN_AGE_SECS: float = 60.0
82
83 def _fsync_fd(fd: int) -> None:
84 """Flush *fd* to durable storage, using the fastest safe syscall available.
85
86 Flushes OS page-cache writes to durable storage before the caller performs
87 ``os.replace``. Without it, a power loss between the kernel accepting the
88 rename syscall and flushing the page cache can leave a zero-byte or
89 partially-written file at the destination even though ``os.replace``
90 returned successfully (observed on APFS, ext4, btrfs).
91
92 On macOS, ``os.fsync()`` maps to ``fcntl(F_FULLFSYNC)`` which forces a
93 full hardware flush — ~4–5 ms per call on APFS. ``F_BARRIERFSYNC`` (85)
94 provides the same crash-safety guarantee (write barrier to APFS journal)
95 at ~0.1 ms per call. We prefer it on macOS and fall through to
96 ``os.fsync()`` only if the kernel rejects it.
97
98 On other platforms ``os.fsync()`` is used directly; on virtual filesystems
99 (tmpfs, ramfs) where fsync is a no-op it is silently skipped.
100 """
101 try:
102 if sys.platform == "darwin":
103 # F_BARRIERFSYNC (85) ensures APFS crash safety without a full
104 # hardware flush — ~50× faster than F_FULLFSYNC on APFS SSDs.
105 try:
106 fcntl.fcntl(fd, 85) # F_BARRIERFSYNC
107 return
108 except OSError:
109 pass # kernel too old or FS doesn't support it — fall through
110 os.fsync(fd)
111 except OSError:
112 pass # best-effort — unsupported on some virtual filesystems
113
114 def _fsync_path(path: pathlib.Path) -> None:
115 """``fsync`` a file by path when no file descriptor is available.
116
117 Opens the file read-only solely to obtain an fd, syncs, and closes.
118 """
119 try:
120 with path.open("rb") as fh:
121 os.fsync(fh.fileno())
122 except OSError:
123 pass # best-effort — unsupported on some virtual filesystems
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 def _shard_prefix_len(repo_root: pathlib.Path) -> int:
144 """Return the configured shard prefix length (2 or 4) for *repo_root*.
145
146 Reads ``[limits] shard_prefix_length`` from ``.muse/config.toml``.
147 Defaults to ``2`` when absent. Values outside ``{2, 4}`` are ignored
148 and the default is used.
149
150 Results are cached per repo root for the lifetime of the process — see
151 :func:`_cached_shard_prefix_len`.
152 """
153 return _cached_shard_prefix_len(str(repo_root))
154
155 # ---------------------------------------------------------------------------
156 # Shard directory validation cache
157 # ---------------------------------------------------------------------------
158
159 # Set of shard directories that have been created AND path-traversal-validated
160 # at least once this process run. Key: str(shard_dir).
161 #
162 # What is amortised (happens only on the FIRST write per shard):
163 # • assert_write_inside_repo — expensive resolve() chain (~6 lstat calls)
164 # • mkdir(parents=True) — stat checks on each parent component
165 #
166 # What is NOT amortised (happens on EVERY write):
167 # • assert_not_symlink(shard_dir) — single lstat, essential TOCTOU guard
168 #
169 # Safety rationale:
170 # assert_write_inside_repo guards against a symlink placed at shard_dir
171 # BEFORE the first write. assert_not_symlink (below, in write_object) guards
172 # against a concurrent swap AFTER the first write — because it fires just
173 # before mkstemp, the race window between the check and the actual write is
174 # nanoseconds, not milliseconds. Both guards must be in place; only the
175 # expensive path-resolution check is amortised.
176 _created_object_shards: set[str] = set()
177
178 def _ensure_object_shard(repo_root: pathlib.Path, shard_dir: pathlib.Path) -> None:
179 """Create and path-traversal-validate *shard_dir*, amortised per process run.
180
181 On the first call for a given *shard_dir* this process run:
182 1. Validates *shard_dir* resolves inside *repo_root* (path-traversal guard,
183 prevents a pre-placed symlink from redirecting writes outside the repo).
184 2. Creates the directory with ``mkdir -p``.
185 3. Records the shard as created so future calls skip steps 1 and 2.
186
187 Subsequent calls for the same *shard_dir* return immediately — the
188 ``resolve()`` chain and parent ``stat`` calls are not repeated.
189
190 **The caller is still required to call** ``assert_not_symlink(shard_dir)``
191 after this function returns. That per-write check catches concurrent
192 symlink-swap attacks (attacker removes the real shard dir and replaces it
193 with a symlink between the path-traversal validation and the ``mkstemp``
194 call).
195
196 Args:
197 repo_root: Repository root (parent of ``.muse/``).
198 shard_dir: The shard subdirectory (e.g. ``.muse/objects/sha256/ab/``).
199 """
200 shard_str = str(shard_dir)
201 if shard_str in _created_object_shards:
202 return
203 # First write to this shard this process run: path-traversal check + mkdir.
204 assert_write_inside_repo(repo_root, shard_dir)
205 shard_dir.mkdir(parents=True, exist_ok=True)
206 _created_object_shards.add(shard_str)
207
208 def cleanup_stale_object_temps(repo_root: pathlib.Path) -> int:
209 """Remove stale ``.obj-tmp-*`` files left by a previous SIGKILL.
210
211 A SIGKILL between :func:`write_object`'s ``mkstemp`` call and the
212 ``os.replace`` rename leaves a partial object temp file in the shard
213 directory. These files are safe to delete because the object was never
214 committed to the store (the rename never happened).
215
216 **Age gate**: only files older than :data:`_CLEANUP_MIN_AGE_SECS` seconds
217 are deleted. This prevents a newly-started process from deleting temp files
218 that belong to in-progress writes from another concurrently-running process
219 — a genuine risk when multiple agents start simultaneously and one calls
220 ``require_repo()`` while another is mid-write. A write that cannot complete
221 within 60 seconds is treated as genuinely stalled.
222
223 This function is called automatically by :func:`objects_dir` on first
224 access and by :func:`muse.core.repo.require_repo` at startup.
225
226 Args:
227 repo_root: Root of the Muse repository.
228
229 Returns:
230 Number of stale temp files removed.
231 """
232 store = _objects_dir(repo_root)
233 if not store.exists():
234 return 0
235 now = time.time()
236 removed = 0
237
238 def _clean_shard(shard: pathlib.Path) -> int:
239 count = 0
240 for f in shard.iterdir():
241 if not (f.name.startswith(".obj-tmp-") or f.name.startswith(".restore-tmp-")):
242 continue
243 try:
244 age = now - f.stat().st_mtime
245 except OSError:
246 continue # file disappeared between iterdir and stat — skip
247 if age < _CLEANUP_MIN_AGE_SECS:
248 logger.debug(
249 "⏳ Skipping recent temp %s (age %.1fs < %ss — may be in-progress write)",
250 f.name, age, _CLEANUP_MIN_AGE_SECS,
251 )
252 continue
253 try:
254 f.unlink()
255 count += 1
256 logger.warning(
257 "⚠️ Removed stale object temp %s (left by prior crash)", f
258 )
259 except OSError as exc:
260 logger.warning("⚠️ Could not remove stale temp %s: %s", f, exc)
261 return count
262
263 algo_dir = store / DEFAULT_HASH_ALGO
264 if not algo_dir.exists():
265 return removed
266 for shard in algo_dir.iterdir():
267 if not shard.is_dir() or shard.is_symlink():
268 continue
269 removed += _clean_shard(shard)
270
271 return removed
272
273 def objects_dir(repo_root: pathlib.Path) -> pathlib.Path:
274 """Return the path to the local object store root directory.
275
276 The store lives at ``<repo_root>/.muse/objects/``. Shard subdirectories
277 are created lazily by :func:`write_object` and :func:`write_object_from_path`.
278
279 Stale ``.obj-tmp-*`` files from a prior SIGKILL are swept by
280 :func:`muse.core.repo.require_repo` at every command startup via
281 :func:`cleanup_stale_object_temps`. Callers do not need to invoke the
282 cleanup function directly.
283
284 Args:
285 repo_root: Root of the Muse repository (the directory containing
286 ``.muse/``).
287
288 Returns:
289 Absolute path to the objects directory (may not yet exist).
290 """
291 return _objects_dir(repo_root)
292
293 def objects_algo_dir(repo_root: pathlib.Path) -> pathlib.Path:
294 """Return the algorithm-prefixed subdirectory inside the object store.
295
296 All objects are stored under ``<objects_dir>/<algo>/``. Tests and tools
297 that need to create or inspect shard directories should use this function
298 rather than hardcoding ``"sha256"``.
299
300 Args:
301 repo_root: Root of the Muse repository.
302
303 Returns:
304 Absolute path to the algorithm subdirectory (e.g. ``.muse/objects/sha256/``).
305 """
306 return objects_dir(repo_root) / DEFAULT_HASH_ALGO
307
308 def object_path(
309 repo_root: pathlib.Path,
310 object_id: str,
311 prefix_len: int | None = None,
312 *,
313 objects_base: pathlib.Path | None = None,
314 ) -> pathlib.Path:
315 """Return the canonical on-disk path for a single object.
316
317 Objects are sharded by the first *prefix_len* hex characters of their
318 SHA-256 digest, nested under the algorithm subdirectory::
319
320 <objects_base>/sha256/<prefix>/<remainder>
321
322 The default *objects_base* is ``<repo_root>/.muse/objects/`` (client layout).
323 Pass ``objects_base`` explicitly to use a different store root — for example,
324 the server-side bare repo layout uses ``<repo_root>/objects/`` (no ``.muse/``
325 wrapper) and passes that path here directly. The sharding logic is identical
326 regardless of which base is used.
327
328 When *prefix_len* is ``None`` (the default) the value is read from
329 ``[limits] shard_prefix_length`` in ``.muse/config.toml`` (default ``2``).
330
331 Args:
332 repo_root: Root of the repository (used to locate config and the
333 default *objects_base*).
334 object_id: Prefixed SHA-256 object ID (``sha256:<64hex>``).
335 prefix_len: Shard prefix length override (2 or 4). ``None`` reads
336 from config.
337 objects_base: Override for the objects directory. Defaults to
338 ``<repo_root>/.muse/objects/`` (client layout).
339
340 Returns:
341 Absolute path to the object file (may not yet exist).
342
343 Raises:
344 ValueError: If *object_id* is not a valid prefixed SHA-256 object ID.
345 """
346 validate_object_id(object_id)
347 _, hex_id = split_id(object_id)
348 plen = prefix_len if prefix_len is not None else _shard_prefix_len(repo_root)
349 base = objects_base if objects_base is not None else objects_dir(repo_root)
350 return base / DEFAULT_HASH_ALGO / hex_id[:plen] / hex_id[plen:]
351
352 def _object_path_with_fallback(
353 repo_root: pathlib.Path, object_id: str
354 ) -> pathlib.Path:
355 """Return the path where *object_id* actually lives, with shard-prefix fallback.
356
357 Checks in preference order:
358
359 1. Canonical layout with configured shard prefix —
360 ``objects/sha256/<plen_shard>/<rest>``
361 2. If the configured prefix is 4-char, also checks the 2-char shard prefix
362 (migration path when upgrading from 2-char to 4-char sharding).
363
364 Args:
365 repo_root: Root of the Muse repository.
366 object_id: Prefixed SHA-256 object ID (``sha256:<64hex>``).
367
368 Returns:
369 The path that exists on disk, or the primary path if the object is
370 absent (callers treat a non-existent return as "absent").
371 """
372 validate_object_id(object_id)
373 _, hex_id = split_id(object_id)
374 plen = _shard_prefix_len(repo_root)
375 store = objects_dir(repo_root)
376
377 primary = store / DEFAULT_HASH_ALGO / hex_id[:plen] / hex_id[plen:]
378 if primary.exists():
379 return primary
380
381 # Shard-prefix migration: if configured for 4-char, also check 2-char.
382 if plen != _DEFAULT_SHARD_PREFIX_LEN:
383 alt = store / DEFAULT_HASH_ALGO / hex_id[:_DEFAULT_SHARD_PREFIX_LEN] / hex_id[_DEFAULT_SHARD_PREFIX_LEN:]
384 if alt.exists():
385 return alt
386
387 return primary
388
389 def iter_stored_objects(
390 repo_root: pathlib.Path,
391 *,
392 skip_symlinks: bool = True,
393 ) -> Iterator[tuple[str, pathlib.Path]]:
394 """Yield ``(prefixed_object_id, path)`` for every valid object in the store.
395
396 This is the single canonical walker for the object store. It walks only
397 ``objects/sha256/<shard>/<rest>`` — the current layout after the
398 algo-prefix migration.
399
400 Only real files (not symlinks, not directories) whose shard + filename form
401 a valid 64-char lowercase hex string are yielded. Stray files (``.DS_Store``,
402 editor temporaries, etc.) are silently skipped.
403
404 Args:
405 repo_root: Root of the Muse repository.
406 skip_symlinks: When ``True`` (default), symlinked shard dirs and
407 symlinked object files are skipped. Set ``False``
408 only in tests that need to verify the guard.
409
410 Yields:
411 ``(object_id, path)`` where *object_id* is the canonical
412 ``sha256:<64hex>`` string and *path* is the absolute on-disk path.
413 """
414 store = objects_dir(repo_root)
415 if not store.exists():
416 return
417
418 seen: set[str] = set()
419
420 def _is_hex(s: str) -> bool:
421 return bool(s) and all(c in _HEX_CHARS for c in s)
422
423 def _walk_shard(shard_dir: pathlib.Path) -> Iterator[tuple[str, pathlib.Path]]:
424 prefix = shard_dir.name
425 if not _is_hex(prefix):
426 return
427 for obj_file in shard_dir.iterdir():
428 if skip_symlinks and obj_file.is_symlink():
429 continue
430 if not obj_file.is_file():
431 continue
432 rest = obj_file.name
433 if not _is_hex(rest):
434 continue
435 hex_id = prefix + rest
436 if len(hex_id) != 64:
437 continue
438 oid = long_id(hex_id)
439 if oid not in seen:
440 seen.add(oid)
441 yield oid, obj_file
442
443 algo_dir = store / DEFAULT_HASH_ALGO
444 if not algo_dir.exists() or not algo_dir.is_dir():
445 return
446 if skip_symlinks and algo_dir.is_symlink():
447 return
448 for shard_dir in algo_dir.iterdir():
449 if skip_symlinks and shard_dir.is_symlink():
450 continue
451 if not shard_dir.is_dir():
452 continue
453 yield from _walk_shard(shard_dir)
454
455 def has_object(repo_root: pathlib.Path, object_id: str) -> bool:
456 """Return ``True`` if *object_id* is present in the local store.
457
458 Cheaper than :func:`read_object` when the caller only needs to check
459 existence (e.g. to pre-flight a hard reset before touching the working
460 tree).
461
462 Checks loose objects first (canonical layout + legacy fallback), then
463 falls through to the MPack local pack store.
464
465 Args:
466 repo_root: Root of the Muse repository.
467 object_id: Prefixed SHA-256 object ID to check.
468 """
469 if _object_path_with_fallback(repo_root, object_id).exists():
470 return True
471 from muse.core.pack_store import has_object_in_packs
472 return has_object_in_packs(repo_root, object_id)
473
474 def write_object(repo_root: pathlib.Path, object_id: str, content: bytes) -> bool:
475 """Write *content* to the local object store under *object_id*.
476
477 If the object already exists (same ID = same content, content-addressed)
478 the write is skipped and ``False`` is returned. Returns ``True`` when a
479 new object was written.
480
481 The shard directory is created on first write. Subsequent writes for the
482 same ``object_id`` are no-ops — they never overwrite existing content.
483
484 The content hash is verified against *object_id* before writing to prevent
485 corrupt or malicious blobs from entering the store.
486
487 Writes are atomic: content is written to a temp file then renamed,
488 so a crash mid-write never leaves a partial object.
489
490 Args:
491 repo_root: Root of the Muse repository.
492 object_id: Prefixed SHA-256 object ID that identifies this object.
493 content: Raw bytes to persist.
494
495 Returns:
496 ``True`` if the object was newly written, ``False`` if it already
497 existed (idempotent).
498
499 Raises:
500 ValueError: If *object_id* is not a valid prefixed object ID, or if
501 the hash of *content* does not match *object_id*.
502 """
503 validate_object_id(object_id)
504
505 # Size cap: reject blobs that exceed the per-object write ceiling before
506 # doing any I/O. This mirrors the read-time MAX_FILE_BYTES ceiling in
507 # read_object so the same limit applies at both boundaries.
508 if len(content) > MAX_OBJECT_WRITE_BYTES:
509 raise ValueError(
510 f"Object {object_id} is {len(content):,} bytes, exceeding the "
511 f"{MAX_OBJECT_WRITE_BYTES // (1024 * 1024)} MiB per-object write limit."
512 )
513
514 from muse.core.ids import hash_blob as _hash_blob
515 actual = _hash_blob(content)
516 if actual != object_id:
517 raise ValueError(
518 f"Content integrity failure: expected object {object_id} "
519 f"got {actual}"
520 )
521
522 dest = object_path(repo_root, object_id)
523 # Also check legacy paths when configured prefix differs.
524 if _object_path_with_fallback(repo_root, object_id).exists():
525 logger.debug("⚠️ Object %s already in store — skipped", short_id(object_id))
526 return False
527
528 # Amortised path-traversal check + mkdir. The first write per shard calls
529 # assert_write_inside_repo (expensive resolve chain) and mkdir; subsequent
530 # writes to the same shard skip both.
531 _ensure_object_shard(repo_root, dest.parent)
532
533 # Per-write symlink guard: assert the shard is STILL a real directory.
534 # This catches a concurrent attacker who removes the real shard and
535 # replaces it with a symlink after our first-write validation. The check
536 # fires just before mkstemp so the race window is nanoseconds wide.
537 assert_not_symlink(dest.parent, label=f"object shard directory ({dest.parent.name}/)")
538
539 # Store the muse-idiomatic format: "<type> <size>\0<payload>".
540 # The hash of this full string equals object_id (verified above via hash_blob).
541 blob_header = f"blob {len(content)}\0".encode()
542 stored_bytes = blob_header + content
543
544 fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".obj-tmp-")
545 tmp = pathlib.Path(tmp_str)
546 try:
547 with os.fdopen(fd, "wb") as fh:
548 fh.write(stored_bytes)
549 fh.flush()
550 # Set read-only via fd *before* closing — path-based os.chmod after
551 # fh.close() is vulnerable to cleanup_stale_object_temps unlinking
552 # the temp file between the close and the chmod call. os.fchmod
553 # uses the open fd (immune to concurrent unlink of the path).
554 os.fchmod(fh.fileno(), _OBJECT_MODE)
555 _fsync_fd(fh.fileno())
556 os.replace(tmp, dest)
557 except Exception:
558 tmp.unlink(missing_ok=True)
559 raise
560
561 logger.debug("✅ Stored object %s (%d bytes)", short_id(object_id), len(content))
562 return True
563
564 def write_object_from_path(
565 repo_root: pathlib.Path,
566 object_id: str,
567 src: pathlib.Path,
568 ) -> bool:
569 """Copy *src* into the object store without loading it into memory.
570
571 Preferred over :func:`write_object` for large blobs (dense MIDI renders,
572 audio previews) because ``pathlib.Path.copy`` delegates to the OS copy
573 mechanism (sendfile / copy_file_range), keeping the interpreter heap clean.
574
575 Idempotent: if the object already exists it is never overwritten.
576
577 The source file's hash is verified against *object_id* before writing.
578 Writes are atomic (temp file + rename).
579
580 Args:
581 repo_root: Root of the Muse repository.
582 object_id: Prefixed SHA-256 object ID of *src*'s content.
583 src: Absolute path of the source file to store.
584
585 Returns:
586 ``True`` if the object was newly written, ``False`` if it already
587 existed (idempotent).
588
589 Raises:
590 ValueError: If *object_id* is invalid or the file's hash does not match.
591 """
592 validate_object_id(object_id)
593
594 dest = object_path(repo_root, object_id)
595 # Also check legacy paths when configured prefix differs.
596 if _object_path_with_fallback(repo_root, object_id).exists():
597 logger.debug("⚠️ Object %s already in store — skipped", short_id(object_id))
598 return False
599
600 # Guard BEFORE any filesystem modification.
601 assert_write_inside_repo(repo_root, dest)
602
603 # Size cap: stat the source file before reading it. This avoids loading
604 # an oversized blob into the hash loop and then discovering it's too large.
605 src_size = src.stat().st_size
606 if src_size > MAX_OBJECT_WRITE_BYTES:
607 raise ValueError(
608 f"Source file {src.name!r} is {src_size:,} bytes, exceeding the "
609 f"{MAX_OBJECT_WRITE_BYTES // (1024 * 1024)} MiB per-object write limit."
610 )
611
612 # Verify hash before writing.
613 actual = hash_file(src)
614 if actual != object_id:
615 raise ValueError(
616 f"Content integrity failure for {src}: expected {object_id} "
617 f"got {actual}"
618 )
619
620 dest.parent.mkdir(parents=True, exist_ok=True)
621
622 # Second guard after mkdir: catch symlink swap between the resolve check
623 # and mkdir completing.
624 assert_not_symlink(dest.parent, label=f"object shard directory ({dest.parent.name}/)")
625
626 # Write the muse blob format: "blob <size>\0<content>".
627 # Matches write_object's on-disk layout exactly — the two functions must
628 # produce structurally identical object files.
629 blob_header = f"blob {src_size}\0".encode()
630 fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".obj-tmp-")
631 tmp = pathlib.Path(tmp_str)
632 try:
633 with os.fdopen(fd, "wb") as fh:
634 fh.write(blob_header)
635 with src.open("rb") as src_fh:
636 for chunk in iter(lambda: src_fh.read(65536), b""):
637 fh.write(chunk)
638 os.fchmod(fh.fileno(), _OBJECT_MODE)
639 _fsync_fd(fh.fileno())
640 os.replace(tmp, dest)
641 except Exception:
642 tmp.unlink(missing_ok=True)
643 raise
644
645 logger.debug("✅ Stored object %s (%s)", short_id(object_id), src.name)
646 return True
647
648 def read_object(repo_root: pathlib.Path, object_id: str) -> bytes | None:
649 """Read and return the raw bytes for *object_id* from the local store.
650
651 Every read re-verifies the SHA-256 digest of the returned bytes against
652 *object_id*. If the file has been silently corrupted (disk error, cosmic
653 ray, filesystem bug, or deliberate tamper) this function raises
654 ``OSError`` rather than returning bad data. Silent corruption is the
655 worst possible failure mode for a version-control system.
656
657 Returns ``None`` when the object is not present in the store so callers
658 can produce a user-facing error rather than raising ``FileNotFoundError``.
659
660 Args:
661 repo_root: Root of the Muse repository.
662 object_id: Prefixed SHA-256 object ID of the desired object.
663
664 Returns:
665 Raw bytes, or ``None`` when the object is absent from the store.
666
667 Raises:
668 ValueError: If *object_id* is not a valid prefixed object ID.
669 OSError: If the object file exceeds MAX_FILE_BYTES or fails the
670 SHA-256 integrity check.
671 """
672 validate_object_id(object_id)
673 dest = _object_path_with_fallback(repo_root, object_id)
674 if not dest.exists():
675 from muse.core.pack_store import read_object_from_packs
676 return read_object_from_packs(repo_root, object_id)
677 size = dest.stat().st_size
678 if size > MAX_FILE_BYTES:
679 raise OSError(
680 f"Object {object_id} is {size} bytes, exceeding the "
681 f"{MAX_FILE_BYTES // (1024 * 1024)} MiB read limit."
682 )
683 # Stream-hash while reading so memory usage stays constant for large blobs.
684 h = hashlib.sha256()
685 chunks: list[bytes] = []
686 with dest.open("rb") as fh:
687 for chunk in iter(lambda: fh.read(65536), b""):
688 h.update(chunk)
689 chunks.append(chunk)
690 actual = long_id(h.hexdigest())
691 if actual != object_id:
692 # Recovery: bare objects written without the typed header have a hash of
693 # their raw content rather than hash("blob <size>\0" + content).
694 # If blob_id(raw) matches the expected object_id the content is intact.
695 data_raw = b"".join(chunks)
696 from muse.core.ids import hash_blob as _hash_blob
697 if _hash_blob(data_raw) == object_id:
698 logger.warning(
699 "⚠️ Object %s has no typed header (bare format). "
700 "Content intact. Run `muse code migrate` to normalise.",
701 short_id(object_id),
702 )
703 return data_raw
704 logger.critical(
705 "❌ Object %s failed integrity check — store may be corrupt "
706 "(expected %s, got %s). Do not trust any data from this object.",
707 object_id, object_id, actual,
708 )
709 raise OSError(
710 f"Object {object_id} failed SHA-256 integrity check. "
711 f"Expected {object_id}, got {actual}. "
712 "The object store may be corrupt. Run `muse verify-pack` "
713 "to audit the full store."
714 )
715 data = b"".join(chunks)
716 # Strip muse object header ("blob <size>\0", "commit <size>\0", etc.)
717 # if present. The header is valid when the type token and declared size
718 # both match — this avoids misidentifying raw blobs that happen to contain
719 # an early null byte.
720 null_idx = data.find(b"\0")
721 if null_idx > 0:
722 header = data[:null_idx].decode(errors="replace")
723 parts = header.split(" ", 1)
724 if len(parts) == 2 and parts[1].isdigit():
725 payload = data[null_idx + 1:]
726 if len(payload) == int(parts[1]):
727 return payload
728 return data
729
730 # ---------------------------------------------------------------------------
731 # Muse object store — idiomatic format
732 #
733 # On-disk layout mirrors the approach used by Git:
734 # "<type> <size>\0<payload>"
735 #
736 # The full string (header + payload) is hashed to produce the object ID,
737 # so the type is part of the object's identity — two objects with identical
738 # payload bytes but different types will never share an ID.
739 # ---------------------------------------------------------------------------
740
741 def write_muse_object(
742 repo_root: pathlib.Path,
743 type_str: str,
744 payload: bytes,
745 ) -> str:
746 """Write a Muse object to the unified store.
747
748 Prepends ``"<type_str> <len>\\0"`` to *payload*, hashes the result to
749 produce the object ID, and writes to ``objects/<algo>/<2>/<62>``.
750
751 Idempotent — returns the object ID without writing if already present.
752
753 Args:
754 repo_root: Repository root.
755 type_str: Object type — ``"blob"``, ``"snapshot"``, or ``"commit"``.
756 payload: Raw bytes to store.
757
758 Returns:
759 The prefixed SHA-256 object ID.
760 """
761 header = f"{type_str} {len(payload)}\0".encode()
762 full = header + payload
763 object_id = long_id(hashlib.sha256(full).hexdigest())
764 path = object_path(repo_root, object_id)
765 if path.exists():
766 return object_id
767 path.parent.mkdir(parents=True, exist_ok=True)
768 path.write_bytes(full)
769 return object_id
770
771
772 def read_muse_object(
773 repo_root: pathlib.Path,
774 object_id: str,
775 ) -> tuple[str, bytes] | None:
776 """Read a Muse object from the unified store.
777
778 Returns ``(type_str, payload)`` or ``None`` if the object is absent or
779 malformed (callers treat malformed the same as missing).
780
781 Args:
782 repo_root: Repository root.
783 object_id: Prefixed SHA-256 object ID.
784 """
785 path = object_path(repo_root, object_id)
786 if not path.exists():
787 return None
788 data = path.read_bytes()
789 try:
790 null_idx = data.index(b"\0")
791 type_str = data[:null_idx].decode().split(" ", 1)[0]
792 return type_str, data[null_idx + 1:]
793 except (ValueError, UnicodeDecodeError):
794 return None
795
796
797 def restore_object(
798 repo_root: pathlib.Path,
799 object_id: str,
800 dest: pathlib.Path,
801 ) -> bool:
802 """Copy an object from the store to *dest*, preserving the destination inode.
803
804 When *dest* already contains the correct content this function returns
805 immediately **without any write**. Skipping the write keeps the inode
806 number and mtime unchanged, which is critical for editors (Cursor, VS Code,
807 Vim, …) that watch open buffers via inode-based filesystem events. A
808 spurious ``os.replace`` (rename syscall) replaces the inode even when the
809 bytes are identical; the editor was watching the old inode and goes blind,
810 leaving a permanently stale buffer that shows the previous file content
811 until the user closes and reopens the tab.
812
813 When *dest* does not yet exist or contains different content, the write is
814 done atomically: content is staged to a sibling temp file and renamed into
815 place via :func:`os.replace` so a partially-written destination is never
816 visible to other processes.
817
818 Creates parent directories of *dest* if they do not exist.
819
820 The caller is responsible for ensuring *dest* is within a safe base
821 directory (use :func:`muse.core.validation.contain_path` before calling).
822
823 Args:
824 repo_root: Root of the Muse repository.
825 object_id: Prefixed SHA-256 object ID of the desired object.
826 dest: Absolute path to write the restored file.
827
828 Returns:
829 ``True`` on success, ``False`` if the object is not in the store.
830
831 Raises:
832 ValueError: If *object_id* is not a valid prefixed object ID.
833 OSError: If the atomic write fails after the object was found.
834 """
835 validate_object_id(object_id)
836 src = _object_path_with_fallback(repo_root, object_id)
837 if not src.exists():
838 # The empty-content object (SHA-256("") = e3b0c44…) is deterministic —
839 # its content is always b"". Repos pushed before the server-side
840 # empty-object fix may be missing it from the store. Synthesize it
841 # inline so apply_manifest does not fail, and persist it into the store
842 # for consistency so future calls find it normally.
843 if object_id == blob_id(b""):
844 write_object(repo_root, object_id, b"")
845 src = _object_path_with_fallback(repo_root, object_id)
846 if not src.exists():
847 return False
848 else:
849 # Object may be in a pack file (Phase 2+). Extract it to the loose
850 # store so the path-based copy path below can proceed normally.
851 from muse.core.pack_store import read_object_from_packs
852 packed_bytes = read_object_from_packs(repo_root, object_id)
853 if packed_bytes is None:
854 logger.debug(
855 "⚠️ Object %s not found in local store or packs — cannot restore",
856 short_id(object_id),
857 )
858 return False
859 write_object(repo_root, object_id, packed_bytes)
860 src = _object_path_with_fallback(repo_root, object_id)
861 if not src.exists():
862 return False
863
864 # Re-verify the source object's hash before restoring it to the working
865 # tree. The in-store file is 0o444 (immutable) but disk errors or cosmic
866 # rays can corrupt it silently after it was written. Detecting corruption
867 # here prevents restoring bad bytes to the working tree without error.
868 size = src.stat().st_size
869 if size > MAX_FILE_BYTES:
870 raise OSError(
871 f"Object {object_id} is {size:,} bytes, exceeding the "
872 f"{MAX_FILE_BYTES // (1024 * 1024)} MiB read limit."
873 )
874 # Hash the raw object file bytes to verify integrity. Object files store
875 # the full muse blob format ("blob <size>\0<content>"), so the raw bytes
876 # ARE the hash input — do NOT call hash_file() here because that function
877 # now computes the blob-prefix hash for working-tree files (raw content).
878 _h = hashlib.sha256()
879 with src.open("rb") as _fh:
880 for _chunk in iter(lambda: _fh.read(65536), b""):
881 _h.update(_chunk)
882 actual = long_id(_h.hexdigest())
883 if actual != object_id:
884 logger.critical(
885 "❌ Object %s failed integrity check during restore"
886 "(expected %s, got %s). Refusing to restore corrupt data.",
887 object_id, object_id, actual,
888 )
889 raise OSError(
890 f"Object {object_id} failed SHA-256 integrity check during restore. "
891 f"Expected {object_id}, got {actual}. "
892 "Run `muse verify-pack` to audit the store."
893 )
894
895 # Read the raw object file and strip the muse header to get the actual
896 # content bytes ("blob <size>\0<content>" → "<content>").
897 raw = src.read_bytes()
898 null_idx = raw.find(b"\0")
899 if null_idx > 0:
900 header = raw[:null_idx].decode(errors="replace")
901 parts = header.split(" ", 1)
902 if len(parts) == 2 and parts[1].isdigit():
903 payload = raw[null_idx + 1:]
904 if len(payload) == int(parts[1]):
905 content = payload
906 else:
907 content = raw
908 else:
909 content = raw
910 else:
911 content = raw
912
913 # Idempotent check-before-write: if dest already contains the same bytes,
914 # skip the rename entirely. Preserving the inode keeps any editor
915 # watching the file alive — an atomic rename (os.replace) always produces a
916 # new inode, silently blinding inode-based watchers even for identical
917 # content. Size is checked first as a cheap pre-filter.
918 if dest.exists():
919 try:
920 if dest.stat().st_size == len(content) and dest.read_bytes() == content:
921 logger.debug(
922 "✅ Object %s already at dest — skipping restore (inode preserved)",
923 short_id(object_id),
924 )
925 return True
926 except OSError:
927 pass # dest unreadable or raced away — fall through to the write path
928
929 dest.parent.mkdir(parents=True, exist_ok=True)
930 fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".restore-tmp-")
931 tmp = pathlib.Path(tmp_str)
932 try:
933 with os.fdopen(fd, "wb") as fh:
934 fh.write(content)
935 os.chmod(tmp, 0o644)
936 _fsync_path(tmp)
937 os.replace(tmp, dest)
938 except Exception:
939 tmp.unlink(missing_ok=True)
940 raise
941 logger.debug("✅ Restored object %s → %s", short_id(object_id), dest)
942 return True
File History 1 commit
sha256:fe844c2411edd1cec3d4c847f36a96c6ccd4e3d7d1a715106d2ecd64216bf94f fix: bare object detection and read recovery; rm adapter files Sonnet 4.6 minor ⚠ 44 days ago