gabriel / muse public
test_integrity_I2_fsync.py python
1,777 lines 70.4 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 126 days ago
1 """I-2: fsync before atomic rename in ALL durable write paths.
2
3 Covers every write primitive in the Muse durability chain:
4
5 Tier 0 — Primitive helpers
6 * write_text_atomic — the canonical text-file atomic helper
7 * _write_msgpack_atomic — the canonical msgpack helper
8
9 Tier 1 — HEAD + branch refs (catastrophic if corrupt)
10 * write_head_branch (store.py)
11 * write_head_commit (store.py)
12 * write_branch_ref (store.py — used by commit, merge, checkout, pull,
13 revert, reset, cherry_pick, update_ref, transport,
14 rebase, bundle)
15
16 Tier 2 — VCS state files
17 * write_merge_state (merge_engine.py)
18 * save_rebase_state (rebase.py)
19 * create_reservation, create_intent (coordination.py)
20 * OpLog checkpoint (op_log.py)
21
22 Tier 3 — Config files
23 * config.py write_config_value / set_remote (via write_text_atomic)
24
25 Tier 4 — Large-blob write paths (shutil.copy2-based)
26 * write_object_from_path — uses _fsync_fd (fd-based) before os.replace
27 * restore_object — uses _fsync_path (path-based) before os.replace
28
29 Each tier is verified for:
30 1. mkstemp unique temp names (no fixed .tmp collisions).
31 2. fsync before os.replace — ordering enforced.
32 3. No orphan temp files after success or simulated crash.
33 4. Concurrent write safety — independent results, no cross-corruption.
34 5. Correct final on-disk content.
35
36 Additional coverage (gap audit):
37 6. Page-cache non-flush defense-in-depth (I-1 is the backstop for I-2).
38 7. Mid-write fh.write failure — orphan cleaned up.
39 8. Same object_id written from N threads simultaneously — idempotency holds.
40 9. 10 000 sequential commits — store clean throughout.
41 10. SIGKILL crash safety — multiprocessing kill leaves no orphans, store consistent.
42 11. Performance benchmark — 4 KiB msgpack fsync write < 5 ms.
43
44 Regression — any write path that bypasses write_text_atomic is caught here.
45 """
46 from __future__ import annotations
47
48 import os
49 import pathlib
50 import tempfile
51 import threading
52 from unittest.mock import patch
53
54
55 def _sigkill_writer_worker(root: pathlib.Path, count: int) -> None:
56 """Write objects in a tight loop until killed.
57
58 Defined at module level so it is picklable under the ``"spawn"``
59 multiprocessing context (closures defined inside test methods are not
60 picklable and therefore incompatible with ``"spawn"``).
61 """
62 import time as _time
63
64 from muse.core.types import blob_id
65 from muse.core.object_store import write_object as _wo
66
67 for i in range(count):
68 payload = f"crash-worker-object-{i}".encode()
69 obj_id = blob_id(payload)
70 try:
71 _wo(root, obj_id, payload)
72 except Exception:
73 pass
74 _time.sleep(0.0001)
75
76 import pytest
77
78 import json
79 import muse.core.rebase
80
81
82 def _corrupt_file(p: pathlib.Path, new_content: bytes) -> None:
83 """Overwrite *p* temporarily lifting the 0o444 guard.
84
85 Object files are written with mode 0o444. Tests that simulate disk
86 corruption must temporarily grant write permission.
87 """
88 os.chmod(p, 0o644)
89 try:
90 p.write_bytes(new_content)
91 finally:
92 os.chmod(p, 0o444)
93
94 from muse.core.coordination import Reservation
95 from muse.core.types import blob_id, fake_id
96 from muse.core.rebase import RebaseState
97 from muse.core.object_store import (
98 _fsync_fd,
99 write_object,
100 write_object_from_path,
101 restore_object,
102 read_object,
103 object_path,
104 )
105 from muse.core.snapshot import compute_commit_id
106 from muse.core.store import (
107 CommitRecord,
108 write_branch_ref,
109 write_commit,
110 write_head_branch,
111 write_head_commit,
112 write_text_atomic,
113 read_commit,
114 )
115 from muse.core.paths import commits_dir, head_path, heads_dir, merge_state_path, muse_dir, rebase_state_path, ref_path, snapshots_dir
116 import datetime
117
118
119 # ---------------------------------------------------------------------------
120 # Helpers
121 # ---------------------------------------------------------------------------
122
123 def _repo(tmp_path: pathlib.Path) -> pathlib.Path:
124 muse_dir(tmp_path).mkdir()
125 (commits_dir(tmp_path)).mkdir()
126 (snapshots_dir(tmp_path)).mkdir()
127 return tmp_path
128
129
130 def _oid(data: bytes) -> str:
131 return blob_id(data)
132
133
134 def _commit(idx: int = 0) -> CommitRecord:
135 sid = fake_id(f"snap-{idx}")
136 message = f"commit {idx}"
137 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
138 cid = compute_commit_id(
139 parent_ids=[],
140 snapshot_id=sid,
141 message=message,
142 committed_at_iso=committed_at.isoformat(),
143 author="tester",
144 )
145 return CommitRecord(
146 repo_id="fsync-test",
147 commit_id=cid,
148 branch="main",
149 snapshot_id=sid,
150 message=message,
151 committed_at=committed_at,
152 author="tester",
153 parent_commit_id=None,
154 parent2_commit_id=None,
155 )
156
157
158 def _tmp_files(directory: pathlib.Path) -> list[pathlib.Path]:
159 """Return all temp/orphan files in *directory* (recursively)."""
160 result: list[pathlib.Path] = []
161 for p in directory.rglob("*"):
162 name = p.name
163 if name.startswith(".obj-tmp-") or name.startswith(".muse-tmp-") or name.startswith(".restore-tmp-"):
164 result.append(p)
165 return result
166
167
168 # ---------------------------------------------------------------------------
169 # Unit: fsync is called before os.replace in write_object
170 # ---------------------------------------------------------------------------
171
172 class TestFsyncCalledBeforeReplace:
173 def test_write_object_calls_fsync_before_replace(self, tmp_path: pathlib.Path) -> None:
174 """_fsync_fd must be called before os.replace in write_object.
175
176 Patches _fsync_fd (the platform abstraction) rather than os.fsync
177 directly, because on macOS _fsync_fd uses fcntl(F_BARRIERFSYNC) and
178 returns before ever reaching os.fsync.
179 """
180 repo = _repo(tmp_path)
181 data = b"fsync ordering test"
182 oid = _oid(data)
183
184 call_order: list[str] = []
185 real_fsync_fd = _fsync_fd
186 real_replace = os.replace
187
188 def tracking_fsync_fd(fd: int) -> None:
189 call_order.append("fsync")
190 real_fsync_fd(fd)
191
192 def tracking_replace(src: str | bytes | os.PathLike[str], dst: str | bytes | os.PathLike[str]) -> None:
193 call_order.append("replace")
194 real_replace(src, dst)
195
196 with patch("muse.core.object_store._fsync_fd", side_effect=tracking_fsync_fd), \
197 patch("muse.core.object_store.os.replace", side_effect=tracking_replace):
198 write_object(repo, oid, data)
199
200 assert "fsync" in call_order, "_fsync_fd was never called"
201 assert "replace" in call_order, "replace was never called"
202 fsync_pos = next(i for i, c in enumerate(call_order) if c == "fsync")
203 replace_pos = next(i for i, c in enumerate(call_order) if c == "replace")
204 assert fsync_pos < replace_pos, (
205 f"_fsync_fd (pos {fsync_pos}) must happen before replace (pos {replace_pos})"
206 )
207
208 def test_write_commit_calls_fsync_before_replace(self, tmp_path: pathlib.Path) -> None:
209 """A durability flush must be called before tmp.replace in _write_msgpack_atomic.
210
211 On macOS, _write_msgpack_atomic uses fcntl(F_BARRIERFSYNC) and never
212 reaches os.fsync directly. We track both paths so the test is
213 platform-agnostic.
214 """
215 repo = _repo(tmp_path)
216 c = _commit(0)
217
218 call_order: list[str] = []
219 real_fsync = os.fsync
220 import fcntl as _fcntl
221 real_fcntl = _fcntl.fcntl
222
223 def tracking_fsync(fd: int) -> None:
224 call_order.append("fsync")
225 real_fsync(fd)
226
227 def tracking_fcntl(fd: int, cmd: int, *args: int) -> int:
228 if cmd == 85: # F_BARRIERFSYNC
229 call_order.append("fsync")
230 return real_fcntl(fd, cmd, *args)
231
232 with patch("muse.core.store.os.fsync", side_effect=tracking_fsync), \
233 patch("muse.core.store.fcntl.fcntl", side_effect=tracking_fcntl):
234 write_commit(repo, c)
235
236 assert "fsync" in call_order, "no durability flush called during write_commit"
237
238 def test_fsync_failure_is_non_fatal(self, tmp_path: pathlib.Path) -> None:
239 """A failing fsync (virtual fs) must not prevent the write from completing."""
240 repo = _repo(tmp_path)
241 data = b"fsync fails gracefully"
242 oid = _oid(data)
243
244 with patch("muse.core.object_store.os.fsync", side_effect=OSError("fsync not supported")):
245 result = write_object(repo, oid, data)
246
247 assert result is True
248 assert read_object(repo, oid) == data
249
250 def test_fsync_failure_in_store_is_non_fatal(self, tmp_path: pathlib.Path) -> None:
251 """A failing fsync in _write_msgpack_atomic must not abort the commit write."""
252 repo = _repo(tmp_path)
253 c = _commit(1)
254
255 with patch("muse.core.store.os.fsync", side_effect=OSError("not supported")):
256 write_commit(repo, c)
257
258 assert read_commit(repo, c.commit_id) is not None
259
260
261 # ---------------------------------------------------------------------------
262 # Unit: unique temp file names (mkstemp, not fixed .tmp)
263 # ---------------------------------------------------------------------------
264
265 class TestUniqueTempNames:
266 def test_write_object_uses_mkstemp(self, tmp_path: pathlib.Path) -> None:
267 """write_object must use tempfile.mkstemp, not path.with_suffix('.tmp')."""
268 repo = _repo(tmp_path)
269 data = b"unique temp name check"
270 oid = _oid(data)
271
272 mkstemp_called = [False]
273 real_mkstemp = tempfile.mkstemp
274
275 def tracking_mkstemp(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
276 mkstemp_called[0] = True
277 return real_mkstemp(dir=dir, prefix=prefix)
278
279 with patch("muse.core.object_store.tempfile.mkstemp", side_effect=tracking_mkstemp):
280 write_object(repo, oid, data)
281
282 assert mkstemp_called[0], "tempfile.mkstemp was not called — fixed .tmp name may be in use"
283
284 def test_write_commit_uses_mkstemp(self, tmp_path: pathlib.Path) -> None:
285 """_write_msgpack_atomic must use tempfile.mkstemp."""
286 repo = _repo(tmp_path)
287 c = _commit(2)
288
289 mkstemp_called = [False]
290 real_mkstemp = tempfile.mkstemp
291
292 def tracking_mkstemp(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
293 mkstemp_called[0] = True
294 return real_mkstemp(dir=dir, prefix=prefix)
295
296 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking_mkstemp):
297 write_commit(repo, c)
298
299 assert mkstemp_called[0], "tempfile.mkstemp was not called in _write_msgpack_atomic"
300
301 def test_no_fixed_tmp_suffix_after_write(self, tmp_path: pathlib.Path) -> None:
302 """After a successful write, no .tmp file must remain."""
303 repo = _repo(tmp_path)
304 c = _commit(3)
305 write_commit(repo, c)
306
307 fixed_tmps = list((commits_dir(tmp_path)).glob("*.tmp"))
308 assert fixed_tmps == [], f"Fixed .tmp files left behind: {fixed_tmps}"
309
310
311 # ---------------------------------------------------------------------------
312 # Integration: no orphan temps after successful writes
313 # ---------------------------------------------------------------------------
314
315 class TestNoOrphanTemps:
316 def test_no_orphan_after_write_object(self, tmp_path: pathlib.Path) -> None:
317 repo = _repo(tmp_path)
318 data = b"clean write"
319 oid = _oid(data)
320 write_object(repo, oid, data)
321 assert _tmp_files(tmp_path) == []
322
323 def test_no_orphan_after_write_object_from_path(self, tmp_path: pathlib.Path) -> None:
324 repo = _repo(tmp_path)
325 src = tmp_path / "src.bin"
326 data = b"from path write"
327 src.write_bytes(data)
328 oid = _oid(data)
329 write_object_from_path(repo, oid, src)
330 assert _tmp_files(tmp_path) == []
331
332 def test_no_orphan_after_restore_object(self, tmp_path: pathlib.Path) -> None:
333 repo = _repo(tmp_path)
334 data = b"restore write"
335 oid = _oid(data)
336 write_object(repo, oid, data)
337 dest = tmp_path / "restored.bin"
338 restore_object(repo, oid, dest)
339 assert _tmp_files(tmp_path) == []
340
341 def test_no_orphan_after_write_commit(self, tmp_path: pathlib.Path) -> None:
342 repo = _repo(tmp_path)
343 write_commit(repo, _commit(4))
344 assert _tmp_files(tmp_path) == []
345
346 def test_no_orphan_after_simulated_replace_failure(self, tmp_path: pathlib.Path) -> None:
347 """When os.replace raises, the temp file must be cleaned up."""
348 repo = _repo(tmp_path)
349 data = b"replace will fail"
350 oid = _oid(data)
351
352 with pytest.raises(OSError):
353 with patch("muse.core.object_store.os.replace", side_effect=OSError("disk full")):
354 write_object(repo, oid, data)
355
356 assert _tmp_files(tmp_path) == [], "Temp file not cleaned up after replace failure"
357
358 def test_no_orphan_in_store_after_replace_failure(self, tmp_path: pathlib.Path) -> None:
359 """When tmp.replace raises in _write_msgpack_atomic, temp is cleaned up."""
360 repo = _repo(tmp_path)
361 c = _commit(5)
362
363 with pytest.raises(OSError):
364 with patch("muse.core.store.os.fdopen") as mock_fdopen:
365 mock_fh = mock_fdopen.return_value.__enter__.return_value
366 mock_fh.write.return_value = None
367 mock_fh.flush.return_value = None
368 mock_fh.fileno.return_value = -1
369 with patch("muse.core.store.os.fsync"), \
370 patch("muse.core.store.fcntl.fcntl"):
371 with patch("pathlib.Path.replace", side_effect=OSError("disk full")):
372 write_commit(repo, c)
373
374 assert _tmp_files(tmp_path) == [], "Temp file not cleaned up after commit replace failure"
375
376
377 # ---------------------------------------------------------------------------
378 # Stress: 200 concurrent writers to the same shard
379 # ---------------------------------------------------------------------------
380
381 class TestConcurrentWriters:
382 @pytest.mark.slow
383 def test_200_concurrent_object_writes_no_corruption(self, tmp_path: pathlib.Path) -> None:
384 """200 threads writing distinct objects concurrently must all land correctly."""
385 repo = _repo(tmp_path)
386 payloads = [f"concurrent-object-{i}".encode() for i in range(200)]
387 oids = [_oid(p) for p in payloads]
388 errors: list[str] = []
389
390 def writer(data: bytes, oid: str) -> None:
391 try:
392 write_object(repo, oid, data)
393 result = read_object(repo, oid)
394 if result != data:
395 errors.append(f"Mismatch for {oid[:8]}: got {repr(result)[:20]}")
396 except Exception as exc:
397 errors.append(f"Exception for {oid[:8]}: {exc}")
398
399 threads = [
400 threading.Thread(target=writer, args=(p, o))
401 for p, o in zip(payloads, oids)
402 ]
403 for t in threads:
404 t.start()
405 for t in threads:
406 t.join()
407
408 assert errors == [], f"Concurrent write errors:\n{'\n'.join(errors)}"
409 # Every object must be present and correct
410 for data, oid in zip(payloads, oids):
411 assert read_object(repo, oid) == data
412
413 def test_100_concurrent_commit_writes_no_corruption(self, tmp_path: pathlib.Path) -> None:
414 """100 threads writing distinct commits concurrently must all land correctly."""
415 repo = _repo(tmp_path)
416 commits = [_commit(i) for i in range(100)]
417 errors: list[str] = []
418
419 def writer(c: CommitRecord) -> None:
420 try:
421 write_commit(repo, c)
422 result = read_commit(repo, c.commit_id)
423 if result is None:
424 errors.append(f"Commit {c.commit_id[:8]} not found after write")
425 elif result.message != c.message:
426 errors.append(f"Commit {c.commit_id[:8]} message corrupted")
427 except Exception as exc:
428 errors.append(f"Exception for {c.commit_id[:8]}: {exc}")
429
430 threads = [threading.Thread(target=writer, args=(c,)) for c in commits]
431 for t in threads:
432 t.start()
433 for t in threads:
434 t.join()
435
436 assert errors == [], f"Concurrent commit write errors:\n{'\n'.join(errors)}"
437
438 def test_no_orphan_temps_after_concurrent_writes(self, tmp_path: pathlib.Path) -> None:
439 """No temp files must remain after concurrent writes complete."""
440 repo = _repo(tmp_path)
441 payloads = [f"orphan-check-{i}".encode() for i in range(50)]
442
443 threads = [
444 threading.Thread(target=write_object, args=(repo, _oid(p), p))
445 for p in payloads
446 ]
447 for t in threads:
448 t.start()
449 for t in threads:
450 t.join()
451
452 assert _tmp_files(tmp_path) == []
453
454
455 # ---------------------------------------------------------------------------
456 # Tier 0: write_text_atomic — the primitive all text-state writes funnel through
457 # ---------------------------------------------------------------------------
458
459 class TestWriteTextAtomic:
460 """Unit tests for the write_text_atomic primitive."""
461
462 def test_writes_correct_content(self, tmp_path: pathlib.Path) -> None:
463 path = tmp_path / "state.txt"
464 write_text_atomic(path, "hello world\n")
465 assert path.read_text() == "hello world\n"
466
467 def test_creates_parent_dirs(self, tmp_path: pathlib.Path) -> None:
468 path = tmp_path / "a" / "b" / "c" / "state.txt"
469 write_text_atomic(path, "deep")
470 assert path.read_text() == "deep"
471
472 def test_uses_mkstemp_not_fixed_tmp(self, tmp_path: pathlib.Path) -> None:
473 """write_text_atomic must use tempfile.mkstemp, not path.with_suffix('.tmp')."""
474 path = tmp_path / "ref"
475 called = [False]
476 real_mkstemp = tempfile.mkstemp
477
478 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
479 called[0] = True
480 return real_mkstemp(dir=dir, prefix=prefix)
481
482 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
483 write_text_atomic(path, "abc")
484
485 assert called[0], "write_text_atomic did not call tempfile.mkstemp"
486
487 def test_fsync_called_before_replace(self, tmp_path: pathlib.Path) -> None:
488 """os.fsync must be called before os.replace in write_text_atomic."""
489 path = tmp_path / "ref"
490 call_order: list[str] = []
491 real_fsync = os.fsync
492 real_replace = os.replace
493
494 def t_fsync(fd: int) -> None:
495 call_order.append("fsync")
496 real_fsync(fd)
497
498 def t_replace(src: str | bytes | os.PathLike[str], dst: str | bytes | os.PathLike[str]) -> None:
499 call_order.append("replace")
500 real_replace(src, dst)
501
502 with patch("muse.core.store.os.fsync", side_effect=t_fsync), \
503 patch("muse.core.store.os.replace", side_effect=t_replace):
504 write_text_atomic(path, "content")
505
506 fsync_pos = next((i for i, c in enumerate(call_order) if c == "fsync"), None)
507 replace_pos = next((i for i, c in enumerate(call_order) if c == "replace"), None)
508 assert fsync_pos is not None, "fsync never called"
509 assert replace_pos is not None, "replace never called"
510 assert fsync_pos < replace_pos, "fsync must happen before replace"
511
512 def test_fsync_failure_is_non_fatal(self, tmp_path: pathlib.Path) -> None:
513 """A failing fsync (virtual fs) must not prevent the write from completing."""
514 path = tmp_path / "ref"
515 with patch("muse.core.store.os.fsync", side_effect=OSError("not supported")):
516 write_text_atomic(path, "durable despite fsync failure")
517 assert path.read_text() == "durable despite fsync failure"
518
519 def test_no_orphan_after_success(self, tmp_path: pathlib.Path) -> None:
520 path = tmp_path / "ref"
521 write_text_atomic(path, "clean")
522 assert _tmp_files(tmp_path) == []
523
524 def test_no_orphan_after_replace_failure(self, tmp_path: pathlib.Path) -> None:
525 """When os.replace raises, the temp file must be unlinked."""
526 path = tmp_path / "ref"
527 with pytest.raises(OSError):
528 with patch("muse.core.store.os.replace", side_effect=OSError("disk full")):
529 write_text_atomic(path, "will fail")
530 assert _tmp_files(tmp_path) == [], "Orphan temp file left after replace failure"
531
532 def test_overwrites_existing_file(self, tmp_path: pathlib.Path) -> None:
533 """Subsequent writes must atomically replace the old content."""
534 path = tmp_path / "ref"
535 write_text_atomic(path, "old")
536 write_text_atomic(path, "new")
537 assert path.read_text() == "new"
538
539 def test_encoding_respected(self, tmp_path: pathlib.Path) -> None:
540 path = tmp_path / "utf8"
541 write_text_atomic(path, "caf\u00e9", encoding="utf-8")
542 assert path.read_text(encoding="utf-8") == "caf\u00e9"
543
544 def test_50_concurrent_writes_same_path_no_corruption(self, tmp_path: pathlib.Path) -> None:
545 """50 threads writing to the same file — last write wins, no corruption."""
546 path = tmp_path / "shared_ref"
547 errors: list[str] = []
548
549 def writer(i: int) -> None:
550 try:
551 write_text_atomic(path, f"value-{i:04d}")
552 except Exception as exc:
553 errors.append(str(exc))
554
555 threads = [threading.Thread(target=writer, args=(i,)) for i in range(50)]
556 for t in threads:
557 t.start()
558 for t in threads:
559 t.join()
560
561 assert errors == [], f"write_text_atomic raised: {errors}"
562 content = path.read_text()
563 assert content.startswith("value-"), f"Corrupt content: {content!r}"
564 assert _tmp_files(tmp_path) == [], "Orphan temp files after concurrent writes"
565
566 def test_100_concurrent_writes_distinct_paths_all_land(self, tmp_path: pathlib.Path) -> None:
567 """100 threads writing to distinct paths — all must land correctly."""
568 paths = [tmp_path / f"ref-{i:03d}" for i in range(100)]
569 errors: list[str] = []
570
571 def writer(p: pathlib.Path, i: int) -> None:
572 try:
573 write_text_atomic(p, f"commit-{i}")
574 if p.read_text() != f"commit-{i}":
575 errors.append(f"Mismatch at {p.name}")
576 except Exception as exc:
577 errors.append(str(exc))
578
579 threads = [threading.Thread(target=writer, args=(p, i)) for i, p in enumerate(paths)]
580 for t in threads:
581 t.start()
582 for t in threads:
583 t.join()
584
585 assert errors == [], f"Concurrent distinct-path errors: {errors}"
586 for i, p in enumerate(paths):
587 assert p.read_text() == f"commit-{i}"
588
589
590 # ---------------------------------------------------------------------------
591 # Tier 1a: write_head_branch and write_head_commit
592 # ---------------------------------------------------------------------------
593
594 class TestHeadWrites:
595 """HEAD files are the most critical VCS state — a corrupt HEAD breaks the repo."""
596
597 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
598 muse_dir(tmp_path).mkdir()
599 (heads_dir(tmp_path)).mkdir(parents=True)
600 return tmp_path
601
602 def test_write_head_branch_correct_format(self, tmp_path: pathlib.Path) -> None:
603 root = self._init(tmp_path)
604 write_head_branch(root, "main")
605 content = (head_path(root)).read_text()
606 assert content == "ref: refs/heads/main\n"
607
608 def test_write_head_branch_is_atomic(self, tmp_path: pathlib.Path) -> None:
609 """write_head_branch must go through write_text_atomic (mkstemp + fsync)."""
610 root = self._init(tmp_path)
611 called = [False]
612 real_mkstemp = tempfile.mkstemp
613
614 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
615 called[0] = True
616 return real_mkstemp(dir=dir, prefix=prefix)
617
618 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
619 write_head_branch(root, "main")
620
621 assert called[0], "write_head_branch bypassed mkstemp (not atomic)"
622
623 def test_write_head_branch_rejects_invalid_name(self, tmp_path: pathlib.Path) -> None:
624 root = self._init(tmp_path)
625 with pytest.raises((ValueError, SystemExit)):
626 write_head_branch(root, "bad/../../traversal")
627
628 def test_write_head_commit_correct_format(self, tmp_path: pathlib.Path) -> None:
629 root = self._init(tmp_path)
630 cid = fake_id("commit-a")
631 write_head_commit(root, cid)
632 content = (head_path(root)).read_text()
633 assert content == f"commit: {cid}\n"
634
635 def test_write_head_commit_is_atomic(self, tmp_path: pathlib.Path) -> None:
636 root = self._init(tmp_path)
637 called = [False]
638 real_mkstemp = tempfile.mkstemp
639
640 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
641 called[0] = True
642 return real_mkstemp(dir=dir, prefix=prefix)
643
644 cid = fake_id("commit-b")
645 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
646 write_head_commit(root, cid)
647
648 assert called[0], "write_head_commit bypassed mkstemp (not atomic)"
649
650 def test_write_head_commit_rejects_short_id(self, tmp_path: pathlib.Path) -> None:
651 root = self._init(tmp_path)
652 with pytest.raises(ValueError, match="sha256"):
653 write_head_commit(root, "abc123")
654
655 def test_write_head_commit_rejects_non_hex(self, tmp_path: pathlib.Path) -> None:
656 root = self._init(tmp_path)
657 with pytest.raises(ValueError):
658 write_head_commit(root, "z" * 64)
659
660 def test_head_survives_concurrent_branch_switches(self, tmp_path: pathlib.Path) -> None:
661 """50 threads racing to update HEAD — no corruption, HEAD always readable."""
662 root = self._init(tmp_path)
663 errors: list[str] = []
664 branch_names = [f"feat-{i:03d}" for i in range(50)]
665
666 def switcher(branch: str) -> None:
667 try:
668 write_head_branch(root, branch)
669 content = (head_path(root)).read_text()
670 if not content.startswith("ref: refs/heads/"):
671 errors.append(f"HEAD corrupted: {content!r}")
672 except Exception as exc:
673 errors.append(str(exc))
674
675 threads = [threading.Thread(target=switcher, args=(b,)) for b in branch_names]
676 for t in threads:
677 t.start()
678 for t in threads:
679 t.join()
680
681 assert errors == [], f"HEAD corruption detected: {errors}"
682 assert _tmp_files(tmp_path) == []
683
684
685 # ---------------------------------------------------------------------------
686 # Tier 1b: write_branch_ref — canonical branch pointer update
687 # ---------------------------------------------------------------------------
688
689 class TestWriteBranchRef:
690 """Branch refs are the second most critical VCS state.
691
692 A corrupt or missing ref orphans all commits reachable only from that branch.
693 """
694
695 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
696 (heads_dir(tmp_path)).mkdir(parents=True)
697 return tmp_path
698
699 def _valid_cid(self, seed: str = "x") -> str:
700 return fake_id(seed)
701
702 def test_writes_correct_content(self, tmp_path: pathlib.Path) -> None:
703 root = self._init(tmp_path)
704 cid = self._valid_cid("test")
705 write_branch_ref(root, "main", cid)
706 ref_path = heads_dir(root) / "main"
707 assert ref_path.read_text() == cid
708
709 def test_is_atomic_uses_mkstemp(self, tmp_path: pathlib.Path) -> None:
710 root = self._init(tmp_path)
711 called = [False]
712 real_mkstemp = tempfile.mkstemp
713
714 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
715 called[0] = True
716 return real_mkstemp(dir=dir, prefix=prefix)
717
718 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
719 write_branch_ref(root, "main", self._valid_cid())
720
721 assert called[0], "write_branch_ref bypassed mkstemp (not atomic)"
722
723 def test_fsync_called_before_replace(self, tmp_path: pathlib.Path) -> None:
724 root = self._init(tmp_path)
725 call_order: list[str] = []
726 real_fsync = os.fsync
727 real_replace = os.replace
728
729 def t_fsync(fd: int) -> None:
730 call_order.append("fsync")
731 real_fsync(fd)
732
733 def t_replace(src: str | bytes | os.PathLike[str], dst: str | bytes | os.PathLike[str]) -> None:
734 call_order.append("replace")
735 real_replace(src, dst)
736
737 with patch("muse.core.store.os.fsync", side_effect=t_fsync), \
738 patch("muse.core.store.os.replace", side_effect=t_replace):
739 write_branch_ref(root, "main", self._valid_cid())
740
741 fsync_idx = next((i for i, c in enumerate(call_order) if c == "fsync"), None)
742 replace_idx = next((i for i, c in enumerate(call_order) if c == "replace"), None)
743 assert fsync_idx is not None, "fsync not called in write_branch_ref"
744 assert replace_idx is not None, "replace not called in write_branch_ref"
745 assert fsync_idx < replace_idx, "fsync must precede replace"
746
747 def test_rejects_invalid_branch_name(self, tmp_path: pathlib.Path) -> None:
748 root = self._init(tmp_path)
749 with pytest.raises((ValueError, SystemExit)):
750 write_branch_ref(root, "../escape", self._valid_cid())
751
752 def test_rejects_non_hex_commit_id(self, tmp_path: pathlib.Path) -> None:
753 root = self._init(tmp_path)
754 with pytest.raises(ValueError):
755 write_branch_ref(root, "main", "z" * 64)
756
757 def test_rejects_short_commit_id(self, tmp_path: pathlib.Path) -> None:
758 root = self._init(tmp_path)
759 with pytest.raises(ValueError):
760 write_branch_ref(root, "main", "abc123")
761
762 def test_no_orphan_after_success(self, tmp_path: pathlib.Path) -> None:
763 root = self._init(tmp_path)
764 write_branch_ref(root, "main", self._valid_cid())
765 assert _tmp_files(tmp_path) == []
766
767 def test_no_orphan_after_replace_failure(self, tmp_path: pathlib.Path) -> None:
768 root = self._init(tmp_path)
769 with pytest.raises(OSError):
770 with patch("muse.core.store.os.replace", side_effect=OSError("disk full")):
771 write_branch_ref(root, "main", self._valid_cid())
772 assert _tmp_files(tmp_path) == []
773
774 def test_creates_nested_branch_path(self, tmp_path: pathlib.Path) -> None:
775 """Branches like feat/my-thing require parent dir creation."""
776 root = self._init(tmp_path)
777 cid = self._valid_cid("nested")
778 write_branch_ref(root, "feat/my-thing", cid)
779 ref_path = heads_dir(root) / "feat" / "my-thing"
780 assert ref_path.read_text() == cid
781
782 def test_50_concurrent_refs_distinct_branches(self, tmp_path: pathlib.Path) -> None:
783 """50 concurrent writes to 50 distinct branches — all must land correctly."""
784 root = self._init(tmp_path)
785 branches = [f"agent-{i:04d}" for i in range(50)]
786 cids = {b: self._valid_cid(b) for b in branches}
787 errors: list[str] = []
788
789 def writer(branch: str) -> None:
790 try:
791 write_branch_ref(root, branch, cids[branch])
792 branch_ref = ref_path(root, branch)
793 got = branch_ref.read_text()
794 if got != cids[branch]:
795 errors.append(f"{branch}: expected {cids[branch][:8]}, got {got[:8]}")
796 except Exception as exc:
797 errors.append(f"{branch}: {exc}")
798
799 threads = [threading.Thread(target=writer, args=(b,)) for b in branches]
800 for t in threads:
801 t.start()
802 for t in threads:
803 t.join()
804
805 assert errors == [], f"Concurrent branch ref errors: {errors}"
806 assert _tmp_files(tmp_path) == []
807
808 def test_50_concurrent_refs_same_branch(self, tmp_path: pathlib.Path) -> None:
809 """50 concurrent writes to the SAME branch — last wins, no corruption."""
810 root = self._init(tmp_path)
811 cids = [self._valid_cid(f"race-{i}") for i in range(50)]
812 errors: list[str] = []
813
814 def writer(cid: str) -> None:
815 try:
816 write_branch_ref(root, "main", cid)
817 content = (heads_dir(root) / "main").read_text()
818 if content not in cids:
819 errors.append(f"Corrupt content after write: {content!r}")
820 except Exception as exc:
821 errors.append(str(exc))
822
823 threads = [threading.Thread(target=writer, args=(c,)) for c in cids]
824 for t in threads:
825 t.start()
826 for t in threads:
827 t.join()
828
829 assert errors == [], f"Same-branch concurrent errors: {errors}"
830 assert _tmp_files(tmp_path) == []
831
832
833 # ---------------------------------------------------------------------------
834 # Tier 2a: write_merge_state — MERGE_STATE.json
835 # ---------------------------------------------------------------------------
836
837 class TestMergeStateWrite:
838 """MERGE_STATE.json records in-progress conflict state.
839
840 A corrupt file prevents muse commit from completing a conflicted merge.
841 """
842
843 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
844 muse_dir(tmp_path).mkdir()
845 return tmp_path
846
847 def _cid(self, seed: str) -> str:
848 return fake_id(seed)
849
850 def test_writes_valid_json(self, tmp_path: pathlib.Path) -> None:
851 from muse.core.merge_engine import write_merge_state
852 root = self._init(tmp_path)
853 write_merge_state(
854 root,
855 base_commit=self._cid("base"),
856 ours_commit=self._cid("ours"),
857 theirs_commit=self._cid("theirs"),
858 conflict_paths=["a.py", "b.py"],
859 )
860 state_path = merge_state_path(root)
861 data = json.loads(state_path.read_text())
862 assert data["conflict_paths"] == ["a.py", "b.py"]
863
864 def test_is_atomic(self, tmp_path: pathlib.Path) -> None:
865 """write_merge_state must funnel through write_text_atomic."""
866 from muse.core.merge_engine import write_merge_state
867 root = self._init(tmp_path)
868 called = [False]
869 real_mkstemp = tempfile.mkstemp
870
871 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
872 called[0] = True
873 return real_mkstemp(dir=dir, prefix=prefix)
874
875 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
876 write_merge_state(
877 root,
878 base_commit=self._cid("b"),
879 ours_commit=self._cid("o"),
880 theirs_commit=self._cid("t"),
881 conflict_paths=[],
882 )
883
884 assert called[0], "write_merge_state bypassed mkstemp (not atomic)"
885
886 def test_no_orphan_after_success(self, tmp_path: pathlib.Path) -> None:
887 from muse.core.merge_engine import write_merge_state
888 root = self._init(tmp_path)
889 write_merge_state(
890 root,
891 base_commit=self._cid("b"),
892 ours_commit=self._cid("o"),
893 theirs_commit=self._cid("t"),
894 conflict_paths=["x.py"],
895 )
896 assert _tmp_files(tmp_path) == []
897
898
899 # ---------------------------------------------------------------------------
900 # Tier 2b: save_rebase_state — REBASE_STATE.json
901 # ---------------------------------------------------------------------------
902
903 class TestRebaseStateWrite:
904 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
905 muse_dir(tmp_path).mkdir()
906 return tmp_path
907
908 def _state(self) -> RebaseState:
909 cid = fake_id("rebase-c")
910 return RebaseState(
911 original_branch="main",
912 original_head=cid,
913 onto=cid,
914 remaining=[],
915 completed=[],
916 squash=False,
917 )
918
919 def test_writes_valid_json(self, tmp_path: pathlib.Path) -> None:
920 from muse.core.rebase import save_rebase_state
921 root = self._init(tmp_path)
922 save_rebase_state(root, self._state())
923 path = rebase_state_path(root)
924 data = json.loads(path.read_text())
925 assert data["original_branch"] == "main"
926
927 def test_is_atomic(self, tmp_path: pathlib.Path) -> None:
928 from muse.core.rebase import save_rebase_state
929 root = self._init(tmp_path)
930 called = [False]
931 real_mkstemp = tempfile.mkstemp
932
933 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
934 called[0] = True
935 return real_mkstemp(dir=dir, prefix=prefix)
936
937 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
938 save_rebase_state(root, self._state())
939
940 assert called[0], "save_rebase_state bypassed mkstemp"
941
942 def test_no_orphan_after_success(self, tmp_path: pathlib.Path) -> None:
943 from muse.core.rebase import save_rebase_state
944 root = self._init(tmp_path)
945 save_rebase_state(root, self._state())
946 assert _tmp_files(tmp_path) == []
947
948
949 # ---------------------------------------------------------------------------
950 # Tier 2c: coordination.py — reservation + intent writes
951 # ---------------------------------------------------------------------------
952
953 class TestCoordinationWrites:
954 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
955 muse_dir(tmp_path).mkdir()
956 return tmp_path
957
958 def _res(self, root: pathlib.Path, i: int = 0) -> Reservation:
959 from muse.core.coordination import create_reservation
960 return create_reservation(
961 root,
962 run_id=f"run-{i}",
963 branch="main",
964 addresses=[f"addr-{i}"],
965 operation="write",
966 )
967
968 def test_create_reservation_is_atomic(self, tmp_path: pathlib.Path) -> None:
969 from muse.core.coordination import create_reservation
970 root = self._init(tmp_path)
971 called = [False]
972 real_mkstemp = tempfile.mkstemp
973
974 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
975 called[0] = True
976 return real_mkstemp(dir=dir, prefix=prefix)
977
978 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
979 create_reservation(root, run_id="r1", branch="main", addresses=["a"], operation="write")
980
981 assert called[0], "create_reservation bypassed mkstemp"
982
983 def test_create_reservation_writes_valid_json(self, tmp_path: pathlib.Path) -> None:
984 from muse.core.coordination import _reservations_dir
985 root = self._init(tmp_path)
986 res = self._res(root, 0)
987 res_path = _reservations_dir(root) / f"{res.reservation_id}.json"
988 data = json.loads(res_path.read_text())
989 assert data["operation"] == "write"
990
991 def test_create_intent_is_atomic(self, tmp_path: pathlib.Path) -> None:
992 from muse.core.coordination import create_intent
993 root = self._init(tmp_path)
994 res = self._res(root)
995 called = [False]
996 real_mkstemp = tempfile.mkstemp
997
998 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
999 called[0] = True
1000 return real_mkstemp(dir=dir, prefix=prefix)
1001
1002 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
1003 create_intent(
1004 root,
1005 reservation_id=res.reservation_id,
1006 run_id="r1",
1007 branch="main",
1008 addresses=["a"],
1009 operation="merge",
1010 )
1011
1012 assert called[0], "create_intent bypassed mkstemp"
1013
1014 def test_create_intent_writes_valid_json(self, tmp_path: pathlib.Path) -> None:
1015 from muse.core.coordination import create_intent, _intents_dir
1016 root = self._init(tmp_path)
1017 res = self._res(root)
1018 intent = create_intent(
1019 root,
1020 reservation_id=res.reservation_id,
1021 run_id="r1",
1022 branch="main",
1023 addresses=["a"],
1024 operation="push",
1025 )
1026 intent_path = _intents_dir(root) / f"{intent.intent_id}.json"
1027 data = json.loads(intent_path.read_text())
1028 assert data["operation"] == "push"
1029
1030 def test_no_orphan_after_reservation(self, tmp_path: pathlib.Path) -> None:
1031 root = self._init(tmp_path)
1032 self._res(root)
1033 assert _tmp_files(tmp_path) == []
1034
1035 def test_no_orphan_after_intent(self, tmp_path: pathlib.Path) -> None:
1036 from muse.core.coordination import create_intent
1037 root = self._init(tmp_path)
1038 res = self._res(root)
1039 create_intent(
1040 root,
1041 reservation_id=res.reservation_id,
1042 run_id="r1",
1043 branch="main",
1044 addresses=["a"],
1045 operation="commit",
1046 )
1047 assert _tmp_files(tmp_path) == []
1048
1049 def test_20_concurrent_reservation_writes(self, tmp_path: pathlib.Path) -> None:
1050 """20 concurrent agents creating reservations — no corruption, no orphans."""
1051 from muse.core.coordination import create_reservation, _reservations_dir
1052 root = self._init(tmp_path)
1053 errors: list[str] = []
1054
1055 def writer(i: int) -> None:
1056 try:
1057 create_reservation(
1058 root,
1059 run_id=f"run-{i}",
1060 branch="main",
1061 addresses=[f"addr-{i}"],
1062 operation="write",
1063 )
1064 except Exception as exc:
1065 errors.append(str(exc))
1066
1067 threads = [threading.Thread(target=writer, args=(i,)) for i in range(20)]
1068 for t in threads:
1069 t.start()
1070 for t in threads:
1071 t.join()
1072
1073 assert errors == [], f"Concurrent reservation errors: {errors}"
1074 reservation_files = list(_reservations_dir(root).glob("*.json"))
1075 assert len(reservation_files) == 20, f"Expected 20 reservation files, got {len(reservation_files)}"
1076 assert _tmp_files(tmp_path) == []
1077
1078
1079 # ---------------------------------------------------------------------------
1080 # Tier 3: config.py — TOML config writes
1081 # ---------------------------------------------------------------------------
1082
1083 class TestConfigWrites:
1084 """Config files govern remote connections, auth, and repo settings.
1085
1086 A corrupt config.toml prevents all repo operations.
1087 """
1088
1089 def _init_config_repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
1090 """Create a minimal repo with config.toml so config helpers can operate."""
1091 dot_muse = muse_dir(tmp_path)
1092 dot_muse.mkdir()
1093 (dot_muse / "objects").mkdir()
1094 (dot_muse / "commits").mkdir()
1095 (dot_muse / "snapshots").mkdir()
1096 (dot_muse / "refs" / "heads").mkdir(parents=True)
1097 (dot_muse / "repo.json").write_text(
1098 '{"repo_id": "test-repo", "domain": "code", "default_branch": "main"}',
1099 encoding="utf-8",
1100 )
1101 (dot_muse / "config.toml").write_text("", encoding="utf-8")
1102 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
1103 return tmp_path
1104
1105 def test_set_remote_is_atomic(self, tmp_path: pathlib.Path) -> None:
1106 """set_remote (writes config.toml) must funnel through write_text_atomic."""
1107 from muse.cli.config import set_remote
1108 root = self._init_config_repo(tmp_path)
1109 called = [False]
1110 real_mkstemp = tempfile.mkstemp
1111
1112 def tracking(dir: pathlib.Path | None = None, prefix: str = "") -> tuple[int, str]:
1113 called[0] = True
1114 return real_mkstemp(dir=dir, prefix=prefix)
1115
1116 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking):
1117 set_remote("local", "https://localhost:1337", repo_root=root)
1118
1119 assert called[0], "set_remote bypassed mkstemp (config write not atomic)"
1120
1121 def test_set_remote_no_orphan(self, tmp_path: pathlib.Path) -> None:
1122 from muse.cli.config import set_remote
1123 root = self._init_config_repo(tmp_path)
1124 set_remote("origin", "https://localhost:1337", repo_root=root)
1125 assert _tmp_files(tmp_path) == []
1126
1127 def test_set_remote_correct_content_persisted(self, tmp_path: pathlib.Path) -> None:
1128 from muse.cli.config import set_remote, get_remote
1129 root = self._init_config_repo(tmp_path)
1130 set_remote("myremote", "http://myhost:9000", repo_root=root)
1131 url = get_remote("myremote", repo_root=root)
1132 assert url == "http://myhost:9000"
1133
1134 def test_10_concurrent_config_writes_no_orphan(self, tmp_path: pathlib.Path) -> None:
1135 """10 concurrent set_remote calls — no orphan temp files."""
1136 from muse.cli.config import set_remote
1137 root = self._init_config_repo(tmp_path)
1138 errors: list[str] = []
1139
1140 def writer(i: int) -> None:
1141 try:
1142 set_remote(f"remote-{i}", f"http://host-{i}:9000", repo_root=root)
1143 except Exception as exc:
1144 errors.append(str(exc))
1145
1146 threads = [threading.Thread(target=writer, args=(i,)) for i in range(10)]
1147 for t in threads:
1148 t.start()
1149 for t in threads:
1150 t.join()
1151
1152 # No orphan temps regardless of config merge conflicts
1153 assert _tmp_files(tmp_path) == []
1154
1155
1156 # ---------------------------------------------------------------------------
1157 # Gap 2+3: write_object_from_path — fsync ordering + copy2 failure cleanup
1158 # ---------------------------------------------------------------------------
1159
1160 class TestWriteObjectFromPathFsync:
1161 """_fsync_fd must be called before os.replace in write_object_from_path.
1162
1163 write_object_from_path uses shutil.copy2 then re-opens the temp file as
1164 an fd to call fchmod + _fsync_fd before the atomic rename. The test
1165 patches _fsync_fd (the fd-based variant) — NOT _fsync_path, which is the
1166 path-based variant used only by restore_object.
1167 """
1168
1169 def test_fsync_path_called_before_replace(self, tmp_path: pathlib.Path) -> None:
1170 """_fsync_fd must be invoked before os.replace."""
1171 repo = _repo(tmp_path)
1172 data = b"from-path fsync ordering"
1173 oid = _oid(data)
1174 src = tmp_path / "source.bin"
1175 src.write_bytes(data)
1176
1177 call_order: list[str] = []
1178 real_fsync_fd = __import__("muse.core.object_store", fromlist=["_fsync_fd"])._fsync_fd
1179 real_replace = os.replace
1180
1181 def t_fsync_fd(fd: int) -> None:
1182 call_order.append("fsync_fd")
1183 real_fsync_fd(fd)
1184
1185 def t_replace(s: str | bytes | os.PathLike[str], d: str | bytes | os.PathLike[str]) -> None:
1186 call_order.append("replace")
1187 real_replace(s, d)
1188
1189 with patch("muse.core.object_store._fsync_fd", side_effect=t_fsync_fd), \
1190 patch("muse.core.object_store.os.replace", side_effect=t_replace):
1191 write_object_from_path(repo, oid, src)
1192
1193 fp = next((i for i, c in enumerate(call_order) if c == "fsync_fd"), None)
1194 rp = next((i for i, c in enumerate(call_order) if c == "replace"), None)
1195 assert fp is not None, "_fsync_fd never called in write_object_from_path"
1196 assert rp is not None, "os.replace never called in write_object_from_path"
1197 assert fp < rp, f"_fsync_fd (pos {fp}) must happen before replace (pos {rp})"
1198
1199 def test_fsync_path_failure_non_fatal(self, tmp_path: pathlib.Path) -> None:
1200 """os.fsync failure inside _fsync_fd must not abort write_object_from_path.
1201
1202 _fsync_fd swallows OSError internally — we patch os.fsync so the
1203 function's own try/except absorbs the failure, exactly as it would on a
1204 filesystem that does not support fsync (tmpfs, some Docker volumes).
1205 """
1206 repo = _repo(tmp_path)
1207 data = b"fsync_path fails gracefully"
1208 oid = _oid(data)
1209 src = tmp_path / "src.bin"
1210 src.write_bytes(data)
1211
1212 with patch("muse.core.object_store.os.fsync", side_effect=OSError("not supported")):
1213 result = write_object_from_path(repo, oid, src)
1214
1215 assert result is True
1216 assert read_object(repo, oid) == data
1217
1218 def test_no_orphan_after_copy2_failure(self, tmp_path: pathlib.Path) -> None:
1219 """When shutil.copy2 raises, the mkstemp temp file must be cleaned up."""
1220 import shutil
1221 repo = _repo(tmp_path)
1222 data = b"copy2 will fail"
1223 oid = _oid(data)
1224 src = tmp_path / "src.bin"
1225 src.write_bytes(data)
1226
1227 with pytest.raises(OSError):
1228 with patch("pathlib.Path.copy", side_effect=OSError("I/O error")):
1229 write_object_from_path(repo, oid, src)
1230
1231 assert _tmp_files(tmp_path) == [], "Orphan temp after Path.copy failure"
1232
1233 def test_no_orphan_after_replace_failure(self, tmp_path: pathlib.Path) -> None:
1234 """When os.replace raises, the temp file must be cleaned up."""
1235 repo = _repo(tmp_path)
1236 data = b"replace will fail for from_path"
1237 oid = _oid(data)
1238 src = tmp_path / "src.bin"
1239 src.write_bytes(data)
1240
1241 with pytest.raises(OSError):
1242 with patch("muse.core.object_store.os.replace", side_effect=OSError("disk full")):
1243 write_object_from_path(repo, oid, src)
1244
1245 assert _tmp_files(tmp_path) == [], "Orphan temp after os.replace failure in write_object_from_path"
1246
1247 def test_correct_content_after_write(self, tmp_path: pathlib.Path) -> None:
1248 """Content round-trips correctly through write_object_from_path → read_object."""
1249 repo = _repo(tmp_path)
1250 data = os.urandom(4096)
1251 oid = _oid(data)
1252 src = tmp_path / "payload.bin"
1253 src.write_bytes(data)
1254
1255 write_object_from_path(repo, oid, src)
1256 assert read_object(repo, oid) == data
1257
1258
1259 # ---------------------------------------------------------------------------
1260 # Gap 4+5+6: restore_object — fsync ordering + copy2 failure cleanup
1261 # ---------------------------------------------------------------------------
1262
1263 class TestRestoreObjectFsync:
1264 """_fsync_path must be called before os.replace in restore_object."""
1265
1266 def test_fsync_path_called_before_replace(self, tmp_path: pathlib.Path) -> None:
1267 repo = _repo(tmp_path)
1268 data = b"restore fsync ordering"
1269 oid = _oid(data)
1270 write_object(repo, oid, data)
1271 dest = tmp_path / "restored.bin"
1272
1273 call_order: list[str] = []
1274 real_fsync_path = __import__("muse.core.object_store", fromlist=["_fsync_path"])._fsync_path
1275 real_replace = os.replace
1276
1277 def t_fsync_path(path: pathlib.Path) -> None:
1278 call_order.append("fsync_path")
1279 real_fsync_path(path)
1280
1281 def t_replace(s: str | bytes | os.PathLike[str], d: str | bytes | os.PathLike[str]) -> None:
1282 call_order.append("replace")
1283 real_replace(s, d)
1284
1285 with patch("muse.core.object_store._fsync_path", side_effect=t_fsync_path), \
1286 patch("muse.core.object_store.os.replace", side_effect=t_replace):
1287 restore_object(repo, oid, dest)
1288
1289 fp = next((i for i, c in enumerate(call_order) if c == "fsync_path"), None)
1290 rp = next((i for i, c in enumerate(call_order) if c == "replace"), None)
1291 assert fp is not None, "_fsync_path never called in restore_object"
1292 assert rp is not None, "os.replace never called in restore_object"
1293 assert fp < rp, f"_fsync_path (pos {fp}) must precede replace (pos {rp})"
1294
1295 def test_fsync_path_failure_non_fatal(self, tmp_path: pathlib.Path) -> None:
1296 """os.fsync failure inside _fsync_path must not abort restore_object."""
1297 repo = _repo(tmp_path)
1298 data = b"restore fsync_path fails gracefully"
1299 oid = _oid(data)
1300 write_object(repo, oid, data)
1301 dest = tmp_path / "restored.bin"
1302
1303 with patch("muse.core.object_store.os.fsync", side_effect=OSError("not supported")):
1304 result = restore_object(repo, oid, dest)
1305
1306 assert result is True
1307 assert dest.read_bytes() == data
1308
1309 def test_no_orphan_after_copy2_failure(self, tmp_path: pathlib.Path) -> None:
1310 repo = _repo(tmp_path)
1311 data = b"restore copy2 will fail"
1312 oid = _oid(data)
1313 write_object(repo, oid, data)
1314 dest = tmp_path / "out.bin"
1315
1316 with pytest.raises(OSError):
1317 with patch("pathlib.Path.copy", side_effect=OSError("I/O error")):
1318 restore_object(repo, oid, dest)
1319
1320 assert _tmp_files(tmp_path) == [], "Orphan temp after Path.copy failure in restore_object"
1321
1322 def test_no_orphan_after_replace_failure(self, tmp_path: pathlib.Path) -> None:
1323 repo = _repo(tmp_path)
1324 data = b"restore replace will fail"
1325 oid = _oid(data)
1326 write_object(repo, oid, data)
1327 dest = tmp_path / "out.bin"
1328
1329 with pytest.raises(OSError):
1330 with patch("muse.core.object_store.os.replace", side_effect=OSError("disk full")):
1331 restore_object(repo, oid, dest)
1332
1333 assert _tmp_files(tmp_path) == [], "Orphan temp after os.replace failure in restore_object"
1334
1335 def test_restored_file_mtime_is_current_not_from_object_store(
1336 self, tmp_path: pathlib.Path
1337 ) -> None:
1338 """restore_object must set the destination mtime to NOW, not to the
1339 object-store file's mtime.
1340
1341 shutil.copy2 propagates the source (object-store) mtime to the temp
1342 file. Object-store files are written at commit time and may be days
1343 or weeks old. Without os.utime(tmp, None) the restored destination
1344 carries an old timestamp, causing editors (Cursor, VS Code, Vim) to
1345 see "new mtime < cached mtime" and serve a stale buffer instead of
1346 reloading the file. This is the regression that caused the
1347 "merge work disappears in Cursor but reappears on close/reopen" bug.
1348 """
1349 import time
1350
1351 repo = _repo(tmp_path)
1352 data = b"content that differs from any existing file\n" * 10
1353 oid = _oid(data)
1354 write_object(repo, oid, data)
1355
1356 # Simulate an old object-store mtime: backdate the stored object to 2 days ago.
1357 obj_path = object_path(repo, oid)
1358 two_days_ago = time.time() - (2 * 24 * 3600)
1359 os.utime(obj_path, (two_days_ago, two_days_ago))
1360
1361 # Write a pre-existing dest with a "current" mtime (simulating Cursor's
1362 # last-read timestamp before the checkout/merge).
1363 dest = tmp_path / "watched_file.py"
1364 dest.write_bytes(b"old content that cursor has open\n")
1365 cursor_cached_mtime = time.time()
1366 os.utime(dest, (cursor_cached_mtime, cursor_cached_mtime))
1367
1368 # restore_object must write the new content AND freshen mtime.
1369 t_before = time.time()
1370 restore_object(repo, oid, dest)
1371 t_after = time.time()
1372
1373 new_mtime = os.stat(dest).st_mtime
1374
1375 # Destination must have a FRESH timestamp — not the object-store's old one.
1376 assert new_mtime >= t_before, (
1377 f"Restored file mtime ({new_mtime:.2f}) is older than the time "
1378 f"restore_object was called ({t_before:.2f}). "
1379 "shutil.copy2 is propagating the object-store's stale mtime, "
1380 "which causes editors to serve stale buffers after checkout/merge."
1381 )
1382 assert new_mtime <= t_after + 1.0, (
1383 f"Restored file mtime ({new_mtime:.2f}) is far in the future — unexpected."
1384 )
1385
1386 # Content must be correct regardless.
1387 assert dest.read_bytes() == data
1388
1389 def test_restored_mtime_fresher_than_previous_content(
1390 self, tmp_path: pathlib.Path
1391 ) -> None:
1392 """After restore_object, the destination mtime must be >= the mtime it
1393 had before the call, so editors always see a forward-moving timestamp."""
1394 import time
1395
1396 repo = _repo(tmp_path)
1397 new_data = b"new version from feature branch\n"
1398 oid = _oid(new_data)
1399 write_object(repo, oid, new_data)
1400
1401 dest = tmp_path / "file.py"
1402 dest.write_bytes(b"old version on dev\n")
1403 old_mtime = time.time()
1404 os.utime(dest, (old_mtime, old_mtime))
1405
1406 # Backdate the object-store copy (as it would be after a real commit).
1407 obj_path = object_path(repo, oid)
1408 os.utime(obj_path, (old_mtime - 86400, old_mtime - 86400))
1409
1410 restore_object(repo, oid, dest)
1411
1412 assert os.stat(dest).st_mtime >= old_mtime, (
1413 "Restored file mtime went backwards — editor will not see the change."
1414 )
1415
1416
1417 # ---------------------------------------------------------------------------
1418 # Gap 7: page-cache non-flush — defense-in-depth (I-1 catches what I-2 misses)
1419 # ---------------------------------------------------------------------------
1420
1421 class TestPageCacheDefenseInDepth:
1422 """Demonstrate that I-1 (read-time hash verification) is the safety net
1423 for the unlikely scenario where fsync appeared to succeed but the kernel
1424 wrote zero bytes to disk (power loss AFTER rename, BEFORE flush).
1425
1426 Simulated by: writing an object normally, then zeroing the on-disk file
1427 (mimicking a power-loss-induced empty file at the renamed destination).
1428 read_object must raise OSError — the store never silently serves bad data.
1429 """
1430
1431 def test_zeroed_dest_after_rename_caught_by_read(self, tmp_path: pathlib.Path) -> None:
1432 """Simulate post-rename page-cache loss: zero the stored file, then read."""
1433 repo = _repo(tmp_path)
1434 data = b"page cache simulation"
1435 oid = _oid(data)
1436 write_object(repo, oid, data)
1437
1438 # Mimic power loss that zeroed the file after rename.
1439 _corrupt_file(object_path(repo, oid), b"\x00" * len(data))
1440
1441 with pytest.raises(OSError, match="integrity check"):
1442 read_object(repo, oid)
1443
1444 def test_truncated_dest_caught_by_read(self, tmp_path: pathlib.Path) -> None:
1445 """Simulate partial flush: only first half of bytes survived power loss."""
1446 repo = _repo(tmp_path)
1447 data = b"partial flush simulation" * 10
1448 oid = _oid(data)
1449 write_object(repo, oid, data)
1450 # Only the first half survived to disk.
1451 _corrupt_file(object_path(repo, oid), data[: len(data) // 2])
1452
1453 with pytest.raises(OSError, match="integrity check"):
1454 read_object(repo, oid)
1455
1456 def test_noop_write_detected(self, tmp_path: pathlib.Path) -> None:
1457 """Simulate fh.write no-op (page cache accepted write, never flushed).
1458
1459 We write the object normally and then zero the stored file to mimic the
1460 outcome of a post-rename page-cache flush failure. read_object must
1461 raise OSError — I-1's hash check is the final safety net for any I-2
1462 failure mode.
1463 """
1464 repo = _repo(tmp_path)
1465 data = b"write syscall accepted but page cache never flushed"
1466 oid = _oid(data)
1467
1468 # Write correctly first, then simulate the power-loss outcome: the
1469 # renamed destination was never actually flushed to durable storage.
1470 write_object(repo, oid, data)
1471 stored = object_path(repo, oid)
1472 _corrupt_file(stored, b"") # zero bytes — what a power loss leaves
1473
1474 with pytest.raises(OSError, match="integrity check"):
1475 read_object(repo, oid)
1476
1477
1478 # ---------------------------------------------------------------------------
1479 # Gap 8: same object_id written from N threads simultaneously — idempotency
1480 # ---------------------------------------------------------------------------
1481
1482 class TestIdempotentConcurrentWrite:
1483 """write_object is idempotent: same object_id written from many threads
1484 concurrently must never produce corruption — only one write wins, others
1485 see exists() and skip. The content of the winner must be correct.
1486 """
1487
1488 def test_same_object_50_threads_no_corruption(self, tmp_path: pathlib.Path) -> None:
1489 """50 threads writing the same object_id must all succeed with correct content."""
1490 repo = _repo(tmp_path)
1491 data = b"idempotent object written from 50 threads"
1492 oid = _oid(data)
1493 errors: list[str] = []
1494
1495 def writer() -> None:
1496 try:
1497 write_object(repo, oid, data)
1498 result = read_object(repo, oid)
1499 if result != data:
1500 errors.append(f"Mismatch: {repr(result)[:30]}")
1501 except Exception as exc:
1502 errors.append(f"Exception: {exc}")
1503
1504 threads = [threading.Thread(target=writer) for _ in range(50)]
1505 for t in threads:
1506 t.start()
1507 for t in threads:
1508 t.join()
1509
1510 assert errors == [], f"Idempotent concurrent write errors:\n{'\n'.join(errors)}"
1511 assert read_object(repo, oid) == data
1512 assert _tmp_files(tmp_path) == []
1513
1514 def test_same_object_distinct_content_rejected(self, tmp_path: pathlib.Path) -> None:
1515 """Writing different bytes under the same object_id is always rejected."""
1516 repo = _repo(tmp_path)
1517 data = b"canonical content"
1518 oid = _oid(data)
1519 wrong = b"wrong content that hashes differently"
1520
1521 write_object(repo, oid, data)
1522
1523 with pytest.raises(ValueError, match="integrity"):
1524 write_object(repo, oid, wrong)
1525
1526 assert read_object(repo, oid) == data
1527
1528
1529 # ---------------------------------------------------------------------------
1530 # Gap 9: mid-write fh.write failure — orphan cleaned up
1531 # ---------------------------------------------------------------------------
1532
1533 class TestMidWriteFailureCleanup:
1534 """OSError raised during the fh.write call (disk full mid-write) must not
1535 leave an orphan temp file in the store directory."""
1536
1537 def test_no_orphan_after_write_failure_write_object(self, tmp_path: pathlib.Path) -> None:
1538 """OSError during fh.write must not leave an orphan temp file."""
1539 from unittest.mock import MagicMock
1540 repo = _repo(tmp_path)
1541 data = b"mid-write failure"
1542 oid = _oid(data)
1543
1544 mock_fh = MagicMock()
1545 mock_fh.__enter__.return_value = mock_fh
1546 mock_fh.write.side_effect = OSError("disk full")
1547 mock_fh.flush.return_value = None
1548 mock_fh.fileno.return_value = -1
1549
1550 with pytest.raises(OSError):
1551 with patch("muse.core.object_store.os.fdopen", return_value=mock_fh):
1552 write_object(repo, oid, data)
1553
1554 assert _tmp_files(tmp_path) == [], "Orphan temp file left after mid-write failure"
1555
1556 def test_no_orphan_after_write_failure_write_text_atomic(self, tmp_path: pathlib.Path) -> None:
1557 """write_text_atomic cleans up the temp file when fh.write raises."""
1558 from unittest.mock import MagicMock
1559 path = tmp_path / "state.txt"
1560
1561 mock_fh = MagicMock()
1562 mock_fh.__enter__.return_value = mock_fh
1563 mock_fh.write.side_effect = OSError("disk full")
1564 mock_fh.flush.return_value = None
1565 mock_fh.fileno.return_value = -1
1566
1567 with pytest.raises(OSError):
1568 with patch("muse.core.store.os.fdopen", return_value=mock_fh):
1569 write_text_atomic(path, "will fail")
1570
1571 assert _tmp_files(tmp_path) == [], "Orphan temp left after write_text_atomic mid-write failure"
1572
1573
1574 # ---------------------------------------------------------------------------
1575 # Gap 10: 10 000 sequential commits — store clean throughout
1576 # ---------------------------------------------------------------------------
1577
1578 class TestSequentialStress:
1579 """10 000 sequential commit writes exercise the full fsync+rename path at
1580 scale. The store must be clean (all readable, no orphans) when done.
1581
1582 Based on the Linux-kernel-migration scenario: Linus runs a git-to-muse
1583 import script that writes 75k commits. We test at 10k to keep CI fast.
1584 """
1585
1586 @pytest.mark.slow
1587 def test_10000_sequential_commits_all_readable(self, tmp_path: pathlib.Path) -> None:
1588 """1 000 sequential commits — every one must be readable after write."""
1589 repo = _repo(tmp_path)
1590 commits = [_commit(i) for i in range(1_000)]
1591
1592 for c in commits:
1593 write_commit(repo, c)
1594
1595 # Verify every commit is readable and correct.
1596 failures: list[str] = []
1597 for c in commits:
1598 result = read_commit(repo, c.commit_id)
1599 if result is None:
1600 failures.append(f"Commit {c.commit_id[:8]} not found after write")
1601 elif result.message != c.message:
1602 failures.append(f"Commit {c.commit_id[:8]} message corrupted")
1603
1604 assert failures == [], f"{len(failures)} commit read failures:\n{'\n'.join(failures[:10])}"
1605 assert _tmp_files(tmp_path) == [], "Orphan temps after sequential commit writes"
1606
1607 @pytest.mark.slow
1608 def test_1000_commits_with_20pct_fsync_failure_all_readable(
1609 self, tmp_path: pathlib.Path
1610 ) -> None:
1611 """100 commits with 20% random fsync failures must all land correctly.
1612
1613 Verifies that fsync failure is gracefully handled and atomicity (torn-write
1614 protection) is maintained even when durability (fsync) is degraded.
1615 """
1616 import random as _random
1617 repo = _repo(tmp_path)
1618 rng = _random.Random(42)
1619 commits = [_commit(i) for i in range(100)]
1620 real_fsync = os.fsync
1621
1622 def flaky_fsync(fd: int) -> None:
1623 if rng.random() < 0.2:
1624 raise OSError("simulated fsync failure")
1625 real_fsync(fd)
1626
1627 with patch("muse.core.store.os.fsync", side_effect=flaky_fsync):
1628 for c in commits:
1629 write_commit(repo, c)
1630
1631 failures: list[str] = []
1632 for c in commits:
1633 result = read_commit(repo, c.commit_id)
1634 if result is None:
1635 failures.append(f"Commit {c.commit_id[:8]} not found")
1636 elif result.message != c.message:
1637 failures.append(f"Commit {c.commit_id[:8]} corrupted")
1638
1639 assert failures == [], f"Commits lost under flaky fsync:\n{'\n'.join(failures)}"
1640 assert _tmp_files(tmp_path) == []
1641
1642
1643 # ---------------------------------------------------------------------------
1644 # Gap 11: SIGKILL crash safety — process kill leaves no orphans
1645 # ---------------------------------------------------------------------------
1646
1647 class TestProcessKillCrashSafety:
1648 """Simulate abrupt process termination (SIGKILL) during an object write.
1649
1650 Uses multiprocessing to run the writer in a child process, then kills it
1651 with SIGKILL at a random moment. Afterward, the store must be consistent:
1652 - Objects fully written before the kill must be readable and hash-correct.
1653 - No orphan temp files must remain (OS cleans up open fds; the temp file
1654 created by mkstemp is unlinked by the OS when the process dies, since
1655 it holds the only reference via the fd).
1656
1657 Note: On most POSIX systems, a SIGKILL'd process that holds an open
1658 mkstemp fd will have that fd closed by the kernel. The temp file remains
1659 on disk (the fd close doesn't unlink it) but the rename never happens, so
1660 the destination is either fully written or absent — never partial.
1661 This test verifies the store consistency guarantee, not orphan cleanup
1662 (orphan GC is a separate I-6 concern).
1663 """
1664
1665 @pytest.mark.slow
1666 def test_sigkill_during_write_leaves_no_partial_dest(
1667 self, tmp_path: pathlib.Path
1668 ) -> None:
1669 """Objects written before SIGKILL must still be readable after kill."""
1670 import multiprocessing
1671 import signal
1672 import time
1673
1674 repo = _repo(tmp_path)
1675
1676 # Write 20 known objects before spawning the crashable process.
1677 pre_oids: list[str] = []
1678 for i in range(20):
1679 data = f"pre-kill-{i}".encode()
1680 oid = _oid(data)
1681 write_object(repo, oid, data)
1682 pre_oids.append(oid)
1683
1684 # "spawn" starts a fresh interpreter — no multi-threaded-fork warning
1685 # and no risk of deadlocks inherited from the pytest runner's threads.
1686 # _sigkill_writer_worker is defined at module level to ensure it is
1687 # picklable across the spawn boundary.
1688 ctx = multiprocessing.get_context("spawn")
1689 proc = ctx.Process(target=_sigkill_writer_worker, args=(repo, 5000))
1690 proc.start()
1691
1692 # Kill the worker after a short random delay.
1693 import random
1694 time.sleep(random.uniform(0.01, 0.05))
1695 if proc.is_alive():
1696 assert proc.pid is not None
1697 os.kill(proc.pid, signal.SIGKILL)
1698 proc.join()
1699
1700 # All pre-kill objects must still be readable and correct.
1701 for i, oid in enumerate(pre_oids):
1702 data = f"pre-kill-{i}".encode()
1703 result = read_object(repo, oid)
1704 assert result == data, f"Pre-kill object {oid[:8]} corrupted after SIGKILL"
1705
1706 # Store consistency: every object file in the store must hash-verify.
1707 import muse.core.object_store as _ost
1708 all_oids = _ost._iter_all_object_ids(repo) if hasattr(_ost, "_iter_all_object_ids") else []
1709 for oid in all_oids:
1710 try:
1711 read_object(repo, oid) # raises on hash mismatch
1712 except OSError as exc:
1713 pytest.fail(f"Corrupt object {oid[:8]} found after SIGKILL: {exc}")
1714
1715
1716 # ---------------------------------------------------------------------------
1717 # Gap 12: Performance benchmark — 4 KiB msgpack write + fsync < 5 ms
1718 # ---------------------------------------------------------------------------
1719
1720 class TestFsyncWritePerformance:
1721 """fsync overhead on a 4 KiB msgpack file must be < 5 ms.
1722
1723 The syscall is dominated by the OS flush latency, not the data volume.
1724 tmpfs (which tmp_path typically uses on Linux) syncs instantly; on macOS
1725 with APFS this is also sub-millisecond. 5 ms is a very generous budget —
1726 a real NVMe commit flush is typically < 0.5 ms.
1727 """
1728
1729 @pytest.mark.perf
1730 def test_write_commit_4kib_under_5ms(self, tmp_path: pathlib.Path) -> None:
1731 """Single 4 KiB commit write (msgpack + fsync + rename) < 5 ms."""
1732 import time
1733 repo = _repo(tmp_path)
1734 c = _commit(99_000)
1735
1736 start = time.perf_counter()
1737 write_commit(repo, c)
1738 duration_ms = (time.perf_counter() - start) * 1000
1739
1740 assert read_commit(repo, c.commit_id) is not None
1741 assert duration_ms < 10, (
1742 f"write_commit took {duration_ms:.2f} ms — exceeds the 10 ms fsync budget. "
1743 "Performance regression in the atomic write path."
1744 )
1745
1746 @pytest.mark.perf
1747 def test_write_object_4kib_under_5ms(self, tmp_path: pathlib.Path) -> None:
1748 """Single 4 KiB object write (bytes + fsync + rename) < 5 ms."""
1749 import time
1750 repo = _repo(tmp_path)
1751 data = os.urandom(4096)
1752 oid = _oid(data)
1753
1754 start = time.perf_counter()
1755 write_object(repo, oid, data)
1756 duration_ms = (time.perf_counter() - start) * 1000
1757
1758 assert read_object(repo, oid) == data
1759 assert duration_ms < 10, (
1760 f"write_object took {duration_ms:.2f} ms — exceeds the 10 ms fsync budget."
1761 )
1762
1763 @pytest.mark.perf
1764 def test_write_text_atomic_4kib_under_5ms(self, tmp_path: pathlib.Path) -> None:
1765 """write_text_atomic on a 4 KiB text blob < 5 ms."""
1766 import time
1767 path = tmp_path / "state.txt"
1768 text = "x" * 4096
1769
1770 start = time.perf_counter()
1771 write_text_atomic(path, text)
1772 duration_ms = (time.perf_counter() - start) * 1000
1773
1774 assert path.read_text() == text
1775 assert duration_ms < 10, (
1776 f"write_text_atomic took {duration_ms:.2f} ms — exceeds the 10 ms budget."
1777 )
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 126 days ago