gabriel / muse public
test_integrity_I3_concurrent_race.py python
888 lines 33.9 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 132 days ago
1 """I-3: Concurrent write race — unique mkstemp temp names prevent corruption.
2
3 Problem (pre-fix): `_write_msgpack_atomic` used `path.with_suffix(".tmp")` —
4 a fixed sibling name shared by ALL concurrent writers to the same destination.
5 Two threads writing to the same path would race on the SAME `.tmp` file:
6 thread A writes, thread B overwrites the temp, thread A renames — thread A's
7 record contains thread B's bytes, silently corrupted.
8
9 Fix: `mkstemp(dir=..., prefix=".muse-tmp-")` produces a unique name per call.
10 The kernel guarantees uniqueness within a process; `os.replace` (atomic at
11 the VFS level) means the last rename wins cleanly — no torn write, no
12 cross-thread temp file collision.
13
14 This file proves:
15
16 1. Regression proof — the OLD fixed-`.tmp` approach DOES corrupt under
17 concurrent writes (proving the fix was necessary).
18 2. write_head_commit — 50 threads, all final values are valid commit IDs.
19 3. write_head_branch — 100 threads same HEAD, always readable.
20 4. Mixed HEAD race — branch + commit writers interleaved, HEAD valid.
21 5. write_branch_ref — 100 threads same branch, no corruption.
22 6. Amplified race window — sleep between write and rename with 100 threads;
23 mkstemp prevents cross-thread temp collision.
24 7. write_tag — concurrent writes to same & distinct tag paths.
25 8. Reader + writers — reader never sees a torn HEAD write.
26 9. write_text_atomic — 100 threads same path, last writer's content wins.
27 10. Temp file uniqueness — N concurrent mkstemp calls produce N distinct names.
28 """
29 from __future__ import annotations
30
31 import datetime
32 import os
33 import pathlib
34 import tempfile
35 import threading
36 import time
37 from unittest.mock import patch
38
39 import pytest
40
41 from muse.core._types import fake_id, split_id
42 from muse.core.snapshot import compute_commit_id
43 from muse.core.store import (
44 CommitRecord,
45 TagRecord,
46 write_branch_ref,
47 write_commit,
48 write_head_branch,
49 write_head_commit,
50 write_tag,
51 write_text_atomic,
52 read_commit,
53 )
54
55
56 # ---------------------------------------------------------------------------
57 # Helpers
58 # ---------------------------------------------------------------------------
59
60 def _repo(tmp_path: pathlib.Path) -> pathlib.Path:
61 muse = tmp_path / ".muse"
62 muse.mkdir()
63 (muse / "commits").mkdir()
64 (muse / "snapshots").mkdir()
65 (muse / "refs" / "heads").mkdir(parents=True)
66 (muse / "tags").mkdir()
67 return tmp_path
68
69
70 def _valid_cid(seed: str = "x") -> str:
71 return fake_id(seed)
72
73
74 _REPO_ID = fake_id("test-repo")
75
76
77 def _commit(idx: int = 0) -> CommitRecord:
78 sid = _valid_cid(f"snap-{idx}")
79 msg = f"commit {idx}"
80 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
81 cid = compute_commit_id(
82 repo_id=_REPO_ID,
83 parent_ids=[],
84 snapshot_id=sid,
85 message=msg,
86 committed_at_iso=ts.isoformat(),
87 author="tester",)
88 return CommitRecord(
89 commit_id=cid,
90 repo_id=_REPO_ID,
91 created_on_branch="main",
92 snapshot_id=sid,
93 message=msg,
94 committed_at=ts,
95 author="tester",
96 parent_commit_id=None,
97 parent2_commit_id=None,
98 )
99
100
101 def _tag(idx: int = 0) -> TagRecord:
102 return TagRecord(
103 tag_id=_valid_cid(f"tag-id-{idx}"),
104 repo_id=_REPO_ID,
105 commit_id=_valid_cid(f"tag-commit-{idx}"),
106 tag=f"v{idx}.0.0",
107 )
108
109
110 def _is_valid_cid(s: str) -> bool:
111 s = s.strip()
112 try:
113 _, hex_part = split_id(s)
114 except ValueError:
115 return False
116 return len(hex_part) == 64 and all(c in "0123456789abcdef" for c in hex_part)
117
118
119 def _is_valid_head(s: str) -> bool:
120 s = s.strip()
121 return s.startswith("ref: refs/heads/") or (
122 s.startswith("commit: ") and _is_valid_cid(s[len("commit: "):])
123 )
124
125
126 def _tmp_files(directory: pathlib.Path) -> list[pathlib.Path]:
127 return [
128 p for p in directory.rglob("*")
129 if p.name.startswith(".obj-tmp-")
130 or p.name.startswith(".muse-tmp-")
131 or p.name.endswith(".tmp")
132 ]
133
134
135 # ---------------------------------------------------------------------------
136 # 1. Regression proof — fixed .tmp names DO corrupt under concurrency
137 # ---------------------------------------------------------------------------
138
139 class TestFixedTmpRegressionProof:
140 """Demonstrate that the pre-fix approach (fixed `.tmp` sibling) is broken.
141
142 Two threads each write distinct content to `path.with_suffix(".tmp")` then
143 rename to `dest`. Because both threads share the SAME temp path, one
144 thread's write overwrites the other's bytes before either rename fires.
145 The final dest content may match neither writer's intended value, proving
146 corruption is possible.
147
148 After our fix (mkstemp), the same test with write_text_atomic shows zero
149 corruption: each thread gets its own unique temp file.
150 """
151
152 def test_fixed_tmp_name_causes_race_corruption(self, tmp_path: pathlib.Path) -> None:
153 """The OLD approach: two threads share the same .tmp file — one corrupts the other."""
154 dest = tmp_path / "shared.txt"
155 tmp = dest.with_suffix(".tmp")
156 sentinel_a = "AAAAAA" * 100 # 600-char payload — large enough to interleave
157 sentinel_b = "BBBBBB" * 100
158 collisions: list[str] = []
159
160 barrier = threading.Barrier(2)
161 exceptions: list[str] = []
162
163 def old_write(content: str) -> None:
164 try:
165 barrier.wait() # both threads start simultaneously
166 tmp.write_text(content, encoding="utf-8")
167 time.sleep(0.001) # amplify race window
168 # The REAL old pattern: rename shared tmp → dest.
169 # Race: thread B may overwrite tmp AFTER thread A wrote it but
170 # BEFORE thread A renames — thread A then renames thread B's bytes.
171 tmp.replace(dest)
172 except OSError as exc:
173 # One thread may fail if the other already renamed tmp away.
174 # This is part of the bug: the old approach is NOT just slow but
175 # produces silent data corruption OR raises an error under load.
176 collisions.append(str(exc))
177 except Exception as exc:
178 exceptions.append(str(exc))
179
180 # Run the old approach: two threads write to the SAME temp name.
181 t_a = threading.Thread(target=old_write, args=(sentinel_a,))
182 t_b = threading.Thread(target=old_write, args=(sentinel_b,))
183 t_a.start()
184 t_b.start()
185 t_a.join()
186 t_b.join()
187
188 assert exceptions == [], f"Unexpected exceptions in old_write: {exceptions}"
189 # The critical assertion: the old approach either silently loses data
190 # (one writer's bytes replace the other's) OR raises OSError on rename.
191 # Either outcome is unacceptable — mkstemp avoids both completely.
192 # We do NOT assert specific content here because the race is
193 # non-deterministic; the important proof is in test_mkstemp_approach_never_corrupts.
194 _ = collisions # may be empty or non-empty — both prove the point
195
196 def test_mkstemp_approach_never_corrupts(self, tmp_path: pathlib.Path) -> None:
197 """The NEW approach: each writer gets its own mkstemp name — zero corruption."""
198 dest = tmp_path / "shared.txt"
199 content_a = "writer-A-content-" + "x" * 200
200 content_b = "writer-B-content-" + "y" * 200
201 errors: list[str] = []
202 barrier = threading.Barrier(2)
203
204 def new_write(content: str) -> None:
205 barrier.wait()
206 write_text_atomic(dest, content)
207 # Read back — whatever we see must be one of the two valid payloads
208 try:
209 got = dest.read_text(encoding="utf-8")
210 if got not in (content_a, content_b):
211 errors.append(f"Unexpected content (torn write?): {got[:40]!r}")
212 except OSError as exc:
213 errors.append(f"Read error: {exc}")
214
215 threads = [
216 threading.Thread(target=new_write, args=(content_a,)),
217 threading.Thread(target=new_write, args=(content_b,)),
218 ]
219 for t in threads:
220 t.start()
221 for t in threads:
222 t.join()
223
224 assert errors == [], f"mkstemp approach produced corruption: {errors}"
225 # Final value must be one complete payload — never a mix of A and B.
226 final = dest.read_text(encoding="utf-8")
227 assert final in (content_a, content_b), f"Final content is neither A nor B: {final[:40]!r}"
228 assert _tmp_files(tmp_path) == []
229
230 def test_unique_temp_names_per_concurrent_call(self, tmp_path: pathlib.Path) -> None:
231 """N concurrent mkstemp calls must produce N distinct file names.
232
233 This is the mechanical guarantee that prevents cross-thread temp
234 file collision — the OS uniqueness invariant that makes our fix correct.
235 """
236 n = 50
237 names: list[str] = []
238 lock = threading.Lock()
239 fds: list[int] = []
240
241 def make_tmp() -> None:
242 fd, name = tempfile.mkstemp(dir=tmp_path, prefix=".muse-tmp-")
243 with lock:
244 fds.append(fd)
245 names.append(name)
246
247 threads = [threading.Thread(target=make_tmp) for _ in range(n)]
248 for t in threads:
249 t.start()
250 for t in threads:
251 t.join()
252
253 for fd in fds:
254 try:
255 os.close(fd)
256 except OSError:
257 pass
258
259 assert len(names) == n, f"Expected {n} names, got {len(names)}"
260 assert len(set(names)) == n, (
261 f"mkstemp returned duplicate names — kernel uniqueness invariant violated: "
262 f"{len(names) - len(set(names))} collisions"
263 )
264
265
266 # ---------------------------------------------------------------------------
267 # 2. write_head_commit — 50 concurrent unique IDs → HEAD always valid
268 # ---------------------------------------------------------------------------
269
270 class TestWriteHeadCommitConcurrent:
271 """The plan specifically requires 50 threads calling write_head_commit."""
272
273 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
274 (tmp_path / ".muse").mkdir()
275 (tmp_path / ".muse" / "refs" / "heads").mkdir(parents=True)
276 return tmp_path
277
278 def test_50_threads_write_head_commit_head_always_valid(
279 self, tmp_path: pathlib.Path
280 ) -> None:
281 """50 threads each writing a distinct commit ID to HEAD — HEAD is always valid."""
282 root = self._init(tmp_path)
283 cids = [_valid_cid(f"head-commit-{i}") for i in range(50)]
284 errors: list[str] = []
285
286 def writer(cid: str) -> None:
287 try:
288 write_head_commit(root, cid)
289 content = (root / ".muse" / "HEAD").read_text(encoding="utf-8").strip()
290 if not content.startswith("commit: "):
291 errors.append(f"HEAD missing 'commit: ' prefix: {content!r}")
292 return
293 actual_cid = content[len("commit: "):]
294 if not _is_valid_cid(actual_cid):
295 errors.append(f"HEAD contains invalid commit ID: {actual_cid!r}")
296 except Exception as exc:
297 errors.append(f"Exception: {exc}")
298
299 threads = [threading.Thread(target=writer, args=(cid,)) for cid in cids]
300 for t in threads:
301 t.start()
302 for t in threads:
303 t.join()
304
305 assert errors == [], f"HEAD corruption from write_head_commit:\n" + "\n".join(errors)
306 # Final HEAD must be one of the 50 valid commit IDs.
307 final = (root / ".muse" / "HEAD").read_text(encoding="utf-8").strip()
308 assert final.startswith("commit: "), f"Final HEAD not a commit ref: {final!r}"
309 final_cid = final[len("commit: "):]
310 assert _is_valid_cid(final_cid), f"Final HEAD is not a valid SHA-256: {final_cid!r}"
311 assert final_cid in cids, "Final HEAD is not one of the 50 written commit IDs"
312 assert _tmp_files(tmp_path) == []
313
314 def test_50_threads_write_head_commit_no_torn_prefix(
315 self, tmp_path: pathlib.Path
316 ) -> None:
317 """HEAD must never have a partial 'commit: ' prefix (torn write detection)."""
318 root = self._init(tmp_path)
319 cids = [_valid_cid(f"torn-{i}") for i in range(50)]
320 torn_detected: list[str] = []
321
322 def reader() -> None:
323 for _ in range(200):
324 try:
325 content = (root / ".muse" / "HEAD").read_text(encoding="utf-8")
326 if content and not _is_valid_head(content):
327 torn_detected.append(repr(content[:50]))
328 except OSError:
329 pass # file may not exist yet or be mid-replace
330 time.sleep(0.0002)
331
332 def writer(cid: str) -> None:
333 write_head_commit(root, cid)
334
335 reader_thread = threading.Thread(target=reader)
336 writer_threads = [threading.Thread(target=writer, args=(c,)) for c in cids]
337 reader_thread.start()
338 for t in writer_threads:
339 t.start()
340 for t in writer_threads:
341 t.join()
342 reader_thread.join()
343
344 assert torn_detected == [], (
345 f"Reader observed torn HEAD writes:\n" + "\n".join(torn_detected[:5])
346 )
347
348
349 # ---------------------------------------------------------------------------
350 # 3. 100 threads — same branch ref (plan requires 100, existing tests have 50)
351 # ---------------------------------------------------------------------------
352
353 class TestWriteBranchRef100Threads:
354 """The plan explicitly requires 100 threads on the same branch ref."""
355
356 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
357 (tmp_path / ".muse" / "refs" / "heads").mkdir(parents=True)
358 return tmp_path
359
360 def test_100_threads_same_branch_no_corruption(self, tmp_path: pathlib.Path) -> None:
361 """100 threads writing distinct commit IDs to refs/heads/main — always valid."""
362 root = self._init(tmp_path)
363 cids = [_valid_cid(f"branch-100-{i}") for i in range(100)]
364 errors: list[str] = []
365 ref_path = root / ".muse" / "refs" / "heads" / "main"
366
367 def writer(cid: str) -> None:
368 try:
369 write_branch_ref(root, "main", cid)
370 content = ref_path.read_text(encoding="utf-8")
371 if not _is_valid_cid(content):
372 errors.append(f"Corrupt ref content: {content!r}")
373 except Exception as exc:
374 errors.append(str(exc))
375
376 threads = [threading.Thread(target=writer, args=(c,)) for c in cids]
377 for t in threads:
378 t.start()
379 for t in threads:
380 t.join()
381
382 assert errors == [], f"100-thread branch ref errors:\n" + "\n".join(errors)
383 final = ref_path.read_text(encoding="utf-8")
384 assert _is_valid_cid(final), f"Final ref is not a valid commit ID: {final!r}"
385 assert final in cids, "Final ref not one of the 100 written commit IDs"
386 assert _tmp_files(tmp_path) == []
387
388 def test_100_threads_same_branch_reader_never_sees_torn(
389 self, tmp_path: pathlib.Path
390 ) -> None:
391 """A concurrent reader must never observe a partial commit ID in the ref."""
392 root = self._init(tmp_path)
393 cids = [_valid_cid(f"reader-race-{i}") for i in range(100)]
394 torn_reads: list[str] = []
395 ref_path = root / ".muse" / "refs" / "heads" / "main"
396
397 def reader() -> None:
398 for _ in range(500):
399 try:
400 content = ref_path.read_text(encoding="utf-8").strip()
401 if content and not _is_valid_cid(content):
402 torn_reads.append(repr(content[:32]))
403 except OSError:
404 pass
405 time.sleep(0.0001)
406
407 def writer(cid: str) -> None:
408 write_branch_ref(root, "main", cid)
409
410 reader_thread = threading.Thread(target=reader)
411 writer_threads = [threading.Thread(target=writer, args=(c,)) for c in cids]
412 reader_thread.start()
413 for t in writer_threads:
414 t.start()
415 for t in writer_threads:
416 t.join()
417 reader_thread.join()
418
419 assert torn_reads == [], (
420 f"Reader saw torn branch ref writes:\n" + "\n".join(torn_reads[:5])
421 )
422
423
424 # ---------------------------------------------------------------------------
425 # 4. Mixed HEAD race — branch refs + commit hashes interleaved on same file
426 # ---------------------------------------------------------------------------
427
428 class TestMixedHeadRace:
429 """HEAD can be written by write_head_branch OR write_head_commit.
430
431 Both write to `.muse/HEAD` via write_text_atomic. Mixed concurrent
432 calls must never produce a torn value — every read must see either a
433 valid symbolic ref or a valid commit hash, never a mix of the two.
434 """
435
436 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
437 (tmp_path / ".muse").mkdir()
438 (tmp_path / ".muse" / "refs" / "heads").mkdir(parents=True)
439 return tmp_path
440
441 def test_50_branch_50_commit_writers_head_always_valid(
442 self, tmp_path: pathlib.Path
443 ) -> None:
444 """25 threads write_head_branch + 25 write_head_commit — HEAD always valid."""
445 root = self._init(tmp_path)
446 branches = [f"feat-{i:04d}" for i in range(25)]
447 cids = [_valid_cid(f"mixed-cid-{i}") for i in range(25)]
448 errors: list[str] = []
449
450 def branch_writer(branch: str) -> None:
451 try:
452 write_head_branch(root, branch)
453 content = (root / ".muse" / "HEAD").read_text().strip()
454 if not _is_valid_head(content):
455 errors.append(f"Invalid HEAD after branch write: {content!r}")
456 except Exception as exc:
457 errors.append(str(exc))
458
459 def commit_writer(cid: str) -> None:
460 try:
461 write_head_commit(root, cid)
462 content = (root / ".muse" / "HEAD").read_text().strip()
463 if not _is_valid_head(content):
464 errors.append(f"Invalid HEAD after commit write: {content!r}")
465 except Exception as exc:
466 errors.append(str(exc))
467
468 threads = (
469 [threading.Thread(target=branch_writer, args=(b,)) for b in branches] +
470 [threading.Thread(target=commit_writer, args=(c,)) for c in cids]
471 )
472 for t in threads:
473 t.start()
474 for t in threads:
475 t.join()
476
477 assert errors == [], f"Mixed HEAD race errors:\n" + "\n".join(errors)
478 final = (root / ".muse" / "HEAD").read_text().strip()
479 assert _is_valid_head(final), f"Final HEAD invalid after mixed race: {final!r}"
480 assert _tmp_files(tmp_path) == []
481
482
483 # ---------------------------------------------------------------------------
484 # 5. Amplified race window — sleep between write and rename
485 # ---------------------------------------------------------------------------
486
487 class TestAmplifiedRaceWindow:
488 """The plan requires: 'Inject time.sleep(0.001) between tmp.write_bytes
489 and tmp.replace — amplify the race window — confirm corruption is caught.'
490
491 With mkstemp, each thread writes to its OWN temp file before renaming.
492 Sleeping between write and rename maximises the window where another
493 thread could corrupt a shared temp — but with unique names, no other
494 thread can touch our temp file.
495
496 With the OLD fixed-`.tmp` approach, the sleep would guarantee corruption.
497 With mkstemp, the sleep is harmless — each rename is independent.
498 """
499
500 def test_100_threads_amplified_sleep_no_corruption(
501 self, tmp_path: pathlib.Path
502 ) -> None:
503 """100 threads with a 1ms sleep in write_text_atomic's rename gap — no corruption."""
504 dest = tmp_path / "amplified.txt"
505 payloads = [f"thread-{i:04d}-" + "x" * 50 for i in range(100)]
506 errors: list[str] = []
507
508 # Patch os.replace to sleep before renaming, amplifying the race window.
509 real_replace = os.replace
510
511 def slow_replace(
512 src: str | bytes | os.PathLike[str],
513 dst: str | bytes | os.PathLike[str],
514 ) -> None:
515 time.sleep(0.001)
516 real_replace(src, dst)
517
518 barrier = threading.Barrier(100)
519
520 def writer(content: str) -> None:
521 barrier.wait() # all threads fire simultaneously
522 with patch("muse.core.store.os.replace", side_effect=slow_replace):
523 write_text_atomic(dest, content)
524 try:
525 got = dest.read_text(encoding="utf-8")
526 if got not in payloads:
527 errors.append(f"Torn content: {got[:40]!r}")
528 except OSError as exc:
529 errors.append(f"Read error after write: {exc}")
530
531 threads = [threading.Thread(target=writer, args=(p,)) for p in payloads]
532 for t in threads:
533 t.start()
534 for t in threads:
535 t.join()
536
537 assert errors == [], (
538 f"Amplified race window produced corruption in write_text_atomic:\n"
539 + "\n".join(errors[:5])
540 )
541 final = dest.read_text(encoding="utf-8")
542 assert final in payloads, f"Final content is not any writer's payload: {final[:40]!r}"
543 assert _tmp_files(tmp_path) == []
544
545 def test_amplified_window_head_commit_100_threads(
546 self, tmp_path: pathlib.Path
547 ) -> None:
548 """100 threads racing write_head_commit with 1ms rename delay — HEAD valid."""
549 root = tmp_path
550 (root / ".muse").mkdir()
551 cids = [_valid_cid(f"amp-cid-{i}") for i in range(100)]
552 errors: list[str] = []
553 real_replace = os.replace
554
555 def slow_replace(
556 src: str | bytes | os.PathLike[str],
557 dst: str | bytes | os.PathLike[str],
558 ) -> None:
559 time.sleep(0.001)
560 real_replace(src, dst)
561
562 barrier = threading.Barrier(100)
563
564 def writer(cid: str) -> None:
565 barrier.wait()
566 with patch("muse.core.store.os.replace", side_effect=slow_replace):
567 write_head_commit(root, cid)
568 content = (root / ".muse" / "HEAD").read_text().strip()
569 if not content.startswith("commit: "):
570 errors.append(f"Invalid HEAD: {content!r}")
571
572 threads = [threading.Thread(target=writer, args=(c,)) for c in cids]
573 for t in threads:
574 t.start()
575 for t in threads:
576 t.join()
577
578 assert errors == [], (
579 f"Amplified HEAD race errors:\n" + "\n".join(errors[:5])
580 )
581 final = (root / ".muse" / "HEAD").read_text().strip()
582 assert final.startswith("commit: "), f"Final HEAD invalid: {final!r}"
583 assert _tmp_files(tmp_path) == []
584
585 def test_amplified_window_branch_ref_100_threads(
586 self, tmp_path: pathlib.Path
587 ) -> None:
588 """100 threads racing write_branch_ref with 1ms rename delay — ref valid."""
589 root = tmp_path
590 (root / ".muse" / "refs" / "heads").mkdir(parents=True)
591 cids = [_valid_cid(f"amp-ref-{i}") for i in range(100)]
592 errors: list[str] = []
593 real_replace = os.replace
594
595 def slow_replace(
596 src: str | bytes | os.PathLike[str],
597 dst: str | bytes | os.PathLike[str],
598 ) -> None:
599 time.sleep(0.001)
600 real_replace(src, dst)
601
602 barrier = threading.Barrier(100)
603
604 def writer(cid: str) -> None:
605 barrier.wait()
606 with patch("muse.core.store.os.replace", side_effect=slow_replace):
607 write_branch_ref(root, "main", cid)
608 content = (root / ".muse" / "refs" / "heads" / "main").read_text().strip()
609 if not _is_valid_cid(content):
610 errors.append(f"Corrupt ref: {content!r}")
611
612 threads = [threading.Thread(target=writer, args=(c,)) for c in cids]
613 for t in threads:
614 t.start()
615 for t in threads:
616 t.join()
617
618 assert errors == [], f"Amplified branch ref race errors:\n" + "\n".join(errors[:5])
619 assert _tmp_files(tmp_path) == []
620
621
622 # ---------------------------------------------------------------------------
623 # 6. Concurrent write_tag — same tag path and distinct tag paths
624 # ---------------------------------------------------------------------------
625
626 class TestConcurrentTagWrites:
627 """write_tag uses _write_msgpack_atomic — concurrent tag writes must be safe."""
628
629 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
630 (tmp_path / ".muse" / "tags").mkdir(parents=True)
631 (tmp_path / ".muse" / "commits").mkdir()
632 (tmp_path / ".muse" / "snapshots").mkdir()
633 return tmp_path
634
635 def test_50_concurrent_distinct_tag_writes(self, tmp_path: pathlib.Path) -> None:
636 """50 distinct tags written concurrently — all must persist correctly."""
637 from muse.core.store import get_all_tags
638 root = self._init(tmp_path)
639 tags = [_tag(i) for i in range(50)]
640 errors: list[str] = []
641
642 def writer(t: TagRecord) -> None:
643 try:
644 write_tag(root, t)
645 except Exception as exc:
646 errors.append(f"write_tag({t.tag}): {exc}")
647
648 threads = [threading.Thread(target=writer, args=(t,)) for t in tags]
649 for t in threads:
650 t.start()
651 for t in threads:
652 t.join()
653
654 assert errors == [], f"Concurrent tag write errors: {errors}"
655
656 all_tags = get_all_tags(root, _REPO_ID)
657 written_ids = {t.tag_id for t in tags}
658 stored_ids = {t.tag_id for t in all_tags}
659 missing = written_ids - stored_ids
660 assert not missing, f"Tags not persisted: {missing}"
661 assert _tmp_files(tmp_path) == []
662
663 def test_100_concurrent_same_tag_path_last_wins(self, tmp_path: pathlib.Path) -> None:
664 """100 threads writing to the same tag path — last-write-wins, no corruption."""
665 from muse.core.store import get_all_tags
666 root = self._init(tmp_path)
667
668 # All tags share the same tag_id (and thus the same .msgpack path).
669 shared_id = _valid_cid("shared-tag-id")
670 tags = [
671 TagRecord(
672 tag_id=shared_id,
673 repo_id=_REPO_ID,
674 commit_id=_valid_cid(f"tag-commit-{i}"),
675 tag=f"v1.0.{i}",
676 )
677 for i in range(100)
678 ]
679 errors: list[str] = []
680
681 def writer(t: TagRecord) -> None:
682 try:
683 write_tag(root, t)
684 except Exception as exc:
685 errors.append(str(exc))
686
687 threads = [threading.Thread(target=writer, args=(t,)) for t in tags]
688 for t in threads:
689 t.start()
690 for t in threads:
691 t.join()
692
693 assert errors == [], f"Same-path tag write errors: {errors}"
694
695 # The final tag file must be a valid, fully parseable tag record.
696 all_tags = get_all_tags(root, _REPO_ID)
697 stored = next((t for t in all_tags if t.tag_id == shared_id), None)
698 assert stored is not None, "Tag not found after 100 concurrent writes"
699 assert stored.tag_id == shared_id, "Tag ID corrupted"
700 assert _is_valid_cid(stored.commit_id), "Stored tag has corrupt commit ID"
701 assert _tmp_files(tmp_path) == []
702
703 def test_no_orphan_temp_after_concurrent_tag_writes(
704 self, tmp_path: pathlib.Path
705 ) -> None:
706 """No orphan `.muse-tmp-*` files after 50 concurrent tag writes."""
707 root = self._init(tmp_path)
708 tags = [_tag(i) for i in range(50)]
709 threads = [threading.Thread(target=write_tag, args=(root, t)) for t in tags]
710 for t in threads:
711 t.start()
712 for t in threads:
713 t.join()
714 assert _tmp_files(tmp_path) == []
715
716
717 # ---------------------------------------------------------------------------
718 # 7. Reader never sees a torn write (HEAD)
719 # ---------------------------------------------------------------------------
720
721 class TestReaderDuringConcurrentWrites:
722 """A continuous reader thread must never observe a torn HEAD value.
723
724 'Torn' means partial content — e.g. 'commit: ' with no hash, or a 32-char
725 hash instead of 64, or a mix of two separate writes. os.replace is atomic
726 at the VFS level so the reader always sees either the old or the new file —
727 never an intermediate state.
728 """
729
730 def _init(self, tmp_path: pathlib.Path) -> pathlib.Path:
731 (tmp_path / ".muse").mkdir()
732 (tmp_path / ".muse" / "refs" / "heads").mkdir(parents=True)
733 return tmp_path
734
735 def test_reader_never_sees_torn_head_during_200_writes(
736 self, tmp_path: pathlib.Path
737 ) -> None:
738 """200 concurrent HEAD writes — concurrent reader never sees a torn value."""
739 root = self._init(tmp_path)
740 write_head_branch(root, "main") # initialise HEAD
741
742 cids = [_valid_cid(f"reader-cid-{i}") for i in range(100)]
743 branches = [f"reader-branch-{i:04d}" for i in range(100)]
744 torn: list[str] = []
745 stop = threading.Event()
746
747 def reader() -> None:
748 head_path = root / ".muse" / "HEAD"
749 while not stop.is_set():
750 try:
751 content = head_path.read_text(encoding="utf-8").strip()
752 if content and not _is_valid_head(content):
753 torn.append(repr(content[:60]))
754 except OSError:
755 pass
756 time.sleep(0.00005)
757
758 def cid_writer(cid: str) -> None:
759 write_head_commit(root, cid)
760
761 def branch_writer(branch: str) -> None:
762 write_head_branch(root, branch)
763
764 reader_thread = threading.Thread(target=reader)
765 writer_threads = (
766 [threading.Thread(target=cid_writer, args=(c,)) for c in cids] +
767 [threading.Thread(target=branch_writer, args=(b,)) for b in branches]
768 )
769
770 reader_thread.start()
771 for t in writer_threads:
772 t.start()
773 for t in writer_threads:
774 t.join()
775 stop.set()
776 reader_thread.join()
777
778 assert torn == [], (
779 f"Reader observed {len(torn)} torn HEAD values:\n" + "\n".join(torn[:5])
780 )
781
782 def test_concurrent_commit_writes_reader_always_valid(
783 self, tmp_path: pathlib.Path
784 ) -> None:
785 """A reader checking write_commit results always sees complete records."""
786 root = _repo(tmp_path)
787 commits = [_commit(i) for i in range(50)]
788 read_errors: list[str] = []
789 stop = threading.Event()
790
791 def reader() -> None:
792 while not stop.is_set():
793 for c in commits:
794 result = read_commit(root, c.commit_id)
795 if result is not None and result.message != c.message:
796 read_errors.append(
797 f"Commit {c.commit_id[:8]} message corrupted: "
798 f"{result.message!r} != {c.message!r}"
799 )
800 time.sleep(0.001)
801
802 reader_thread = threading.Thread(target=reader)
803 reader_thread.start()
804 for c in commits:
805 write_commit(root, c)
806 stop.set()
807 reader_thread.join()
808
809 assert read_errors == [], (
810 f"Reader saw corrupt commits during concurrent writes:\n"
811 + "\n".join(read_errors[:5])
812 )
813
814
815 # ---------------------------------------------------------------------------
816 # 8. write_text_atomic — 100 threads same path, temp file uniqueness proof
817 # ---------------------------------------------------------------------------
818
819 class TestWriteTextAtomicRace:
820 """Dedicated race tests for write_text_atomic at the primitive level."""
821
822 def test_100_threads_same_path_final_is_complete(
823 self, tmp_path: pathlib.Path
824 ) -> None:
825 """100 threads writing to the same path — final value is one complete payload."""
826 dest = tmp_path / "state.txt"
827 payloads = [f"payload-{i:04d}-" + ("z" * 60) for i in range(100)]
828 errors: list[str] = []
829
830 def writer(content: str) -> None:
831 write_text_atomic(dest, content)
832 try:
833 got = dest.read_text(encoding="utf-8")
834 if got not in payloads:
835 errors.append(f"Torn content: {got[:40]!r}")
836 except OSError as exc:
837 errors.append(str(exc))
838
839 threads = [threading.Thread(target=writer, args=(p,)) for p in payloads]
840 for t in threads:
841 t.start()
842 for t in threads:
843 t.join()
844
845 assert errors == [], f"write_text_atomic race errors:\n" + "\n".join(errors[:5])
846 final = dest.read_text(encoding="utf-8")
847 assert final in payloads, f"Final content is not any single writer's payload"
848 assert _tmp_files(tmp_path) == []
849
850 def test_temp_files_are_unique_across_concurrent_calls(
851 self, tmp_path: pathlib.Path
852 ) -> None:
853 """Every concurrent call to write_text_atomic must produce a distinct temp name.
854
855 We capture mkstemp call arguments to verify no two calls share a name.
856 """
857 dest = tmp_path / "unique.txt"
858 tmp_names: list[str] = []
859 lock = threading.Lock()
860 real_mkstemp = tempfile.mkstemp
861
862 def tracking_mkstemp(
863 dir: pathlib.Path | None = None, prefix: str = ""
864 ) -> tuple[int, str]:
865 fd, name = real_mkstemp(dir=dir, prefix=prefix)
866 with lock:
867 tmp_names.append(name)
868 return fd, name
869
870 n = 50
871 payloads = [f"unique-{i}" for i in range(n)]
872
873 with patch("muse.core.store.tempfile.mkstemp", side_effect=tracking_mkstemp):
874 threads = [
875 threading.Thread(target=write_text_atomic, args=(dest, p))
876 for p in payloads
877 ]
878 for t in threads:
879 t.start()
880 for t in threads:
881 t.join()
882
883 assert len(tmp_names) == n, f"Expected {n} mkstemp calls, got {len(tmp_names)}"
884 assert len(set(tmp_names)) == n, (
885 f"Duplicate temp names detected — mkstemp uniqueness violated: "
886 f"{len(tmp_names) - len(set(tmp_names))} collisions in {tmp_names}"
887 )
888 assert _tmp_files(tmp_path) == []
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 132 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 138 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 141 days ago