fix(node): an upload never replaces a file, nor shares a part with another

A name an upload in flight will take is reserved; each upload writes its own
`name.<tag>.part`; the finished file is published by a hard link, which
refuses an existing target, and takes the next free name if one appeared
meanwhile — the last ack names it. Two members sending one name at once wrote
one part and published it twice; a file copied in during an upload was
replaced (F-09).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 64 insertions, 1 deletions

diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py
index 80ab454..138ee3b 100644
--- a/packages/meshbay-node/tests/test_partial_uploads.py
+++ b/packages/meshbay-node/tests/test_partial_uploads.py
@@ -17,6 +17,7 @@ may inherit — or overwrite the position of — the other's.
 """
 
 import os
+import re
 import time
 import types
 from pathlib import Path
@@ -362,7 +363,7 @@ async def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path):
                                        total_chunks=2))
     live = ctx["partial_uploads"].live_paths()
     assert len(live) == 1
-    assert next(iter(live)).name == "film.mkv.part"
+    assert re.fullmatch(r"film\.mkv\.[0-9a-f]{8}\.part", next(iter(live)).name)
     assert next(iter(live)).exists()
 
 
@@ -489,3 +490,65 @@ async def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path):
     assert _errors(peer) == []
     assert slots.leases["up-1"].used is True, (
         "the node still believes nobody took this slot up, and will reclaim it")
+
+
+
+# ── never replacing a file ──────────────────────────────────────────────────
+
+async def test_two_members_sending_one_name_at_once_get_two_files(tmp_path):
+    """
+    Both chose a free name at chunk 0, and the free name was the same: the
+    final file did not exist yet, only the first one's `.part`. They then wrote
+    one `.part`, the second truncating the first, and published it twice.
+    """
+    ctx = _group_ctx(tmp_path)
+    alice, bob = _peer(ctx, "alice"), _peer(ctx, "bob")
+    for who, data in ((alice, b"hers-1"), (bob, b"his-1")):
+        await who._do_file_upload(sealed_upload(who, filename="IMG_1234.jpg", data=data,
+                                                chunk_index=0, total_chunks=2))
+    for who, data in ((alice, b"hers-2"), (bob, b"his-2")):
+        await who._do_file_upload(sealed_upload(who, filename="IMG_1234.jpg", data=data,
+                                                chunk_index=1, total_chunks=2))
+    assert _errors(alice) == [] and _errors(bob) == []
+
+    root = ctx["roots"].roots[0].path
+    assert (root / "IMG_1234.jpg").read_bytes() == b"hers-1hers-2"
+    assert (root / "IMG_1234 (2).jpg").read_bytes() == b"his-1his-2"
+    assert _acks(bob, ctx)[-1]["stored_as"] == "IMG_1234 (2).jpg"
+
+
+async def test_a_file_that_appears_during_an_upload_is_not_replaced(tmp_path):
+    """The operator copies a file in under the same name while a member's
+    upload is running. The rename at the end used to replace it."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    await peer._do_file_upload(sealed_upload(peer, filename="film.mkv", data=b"up-1",
+                                             chunk_index=0, total_chunks=2))
+    root = ctx["roots"].roots[0].path
+    (root / "film.mkv").write_bytes(b"the operator's")
+    await peer._do_file_upload(sealed_upload(peer, filename="film.mkv", data=b"up-2",
+                                             chunk_index=1, total_chunks=2))
+
+    assert _errors(peer) == []
+    assert (root / "film.mkv").read_bytes() == b"the operator's"
+    assert (root / "film (2).mkv").read_bytes() == b"up-1up-2"
+    assert _acks(peer, ctx)[-1]["stored_as"] == "film (2).mkv"
+    assert not list(root.glob("*.part")), "the part was left behind"
+
+
+def test_without_hard_links_a_file_is_still_not_replaced(tmp_path, monkeypatch):
+    """FAT, exFAT and some network shares have no hard links."""
+    from meshbay_node import roots as roots_mod
+
+    def no_links(*a, **k):
+        raise PermissionError("operation not permitted")
+    monkeypatch.setattr(roots_mod.os, "link", no_links)
+    (tmp_path / "a.txt").write_bytes(b"there first")
+    part = tmp_path / "a.txt.0123abcd.part"
+    part.write_bytes(b"uploaded")
+
+    name = roots_mod.publish_upload(part, tmp_path, "a.txt", "a.txt")
+    assert name == "a (2).txt"
+    assert (tmp_path / "a.txt").read_bytes() == b"there first"
+    assert (tmp_path / "a (2).txt").read_bytes() == b"uploaded"
+    assert not part.exists()