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, 43 insertions, 5 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py
index 89b0441..a4f9cc9 100644
--- a/packages/meshbay-node/src/meshbay_node/roots.py
+++ b/packages/meshbay-node/src/meshbay_node/roots.py
@@ -30,6 +30,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import os
 import re
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
@@ -52,25 +53,62 @@ SAFE_UPLOAD_NAME = re.compile(
     re.UNICODE)
 
 
-def _free_name(directory: Path, filename: str) -> str:
+def _free_name(directory: Path, filename: str,
+               taken: frozenset[str] | set[str] = frozenset()) -> str:
     """
     `filename`, or the first "name (n).ext" that is not taken.
 
-    Never returns the name of a file that exists, so an upload cannot replace
-    one — the property the per-user quarantine used to provide (C5a).
+    Never returns the name of a file that exists, nor one in `taken` — names
+    uploads in flight will publish under — so an upload cannot replace a file
+    or another upload (C5a).
     """
-    if not (directory / filename).exists():
+    def free(name: str) -> bool:
+        return name not in taken and not (directory / name).exists()
+
+    if free(filename):
         return filename
     stem, dot, ext = filename.rpartition(".")
     if not dot:
         stem, ext = filename, ""
     for n in range(2, 1000):
         candidate = f"{stem} ({n}){dot}{ext}"
-        if not (directory / candidate).exists():
+        if free(candidate):
             return candidate
     raise FileExistsError(filename)
 
 
+def publish_upload(part: Path, directory: Path, stored_name: str, filename: str,
+                   taken: frozenset[str] | set[str] = frozenset()) -> str:
+    """
+    Move a finished `.part` to its name without ever replacing a file. The name
+    it was published under, which may not be `stored_name`.
+
+    A rename replaces whatever is at the target, and the target can appear
+    while the upload runs — the operator copying a file in, another group's
+    upload into a shared folder. A hard link refuses an existing target, so it
+    is the publication; where the filesystem has none (FAT, exFAT, some network
+    shares), the existence check and the rename are as close as it gets. A
+    taken name moves on to the next free one rather than failing the upload.
+    """
+    name = stored_name
+    for _ in range(8):
+        target = directory / name
+        try:
+            os.link(part, target)
+        except FileExistsError:
+            name = _free_name(directory, filename, taken)
+            continue
+        except OSError:
+            if target.exists():
+                name = _free_name(directory, filename, taken)
+                continue
+            part.rename(target)
+            return name
+        part.unlink()
+        return name
+    raise FileExistsError(stored_name)
+
+
 def safe_subdir(roots: RootSet, rel: str) -> Path | None:
     """
     Resolve a client-supplied directory inside one of the group's roots, or refuse.