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>

3 files changed, 87 insertions, 11 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.
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py
index 1c6d1ce..f1ed139 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py
@@ -8,7 +8,7 @@ from pathlib import Path
 from meshbay_common.protocol import UPLOAD_PROBE_INDEX, file_upload_ack_wire, file_upload_payload
 
 from meshbay_node import uploads as uploads_mod
-from meshbay_node.roots import SAFE_UPLOAD_NAME, RootSet, _free_name, off_disk
+from meshbay_node.roots import SAFE_UPLOAD_NAME, RootSet, _free_name, off_disk, publish_upload
 from meshbay_node.transport.webrtc.disk import _append_chunk
 from meshbay_node.transport.webrtc.limits import LEASE_NONE, LEASE_QUEUED
 
@@ -306,9 +306,13 @@ class UploadMixin:
         # A shared directory means two people can send the same name. Refusing the
         # second is safe but silly — everyone's camera produces IMG_1234.jpg — so
         # a free name is found instead. Never a replacement.
+        # Names other uploads into this directory will publish under are taken
+        # too: none of them is on disk yet.
+        reserved = uploads.reserved_names(rel_dir)
         stored_name = (state.stored_name if state
-                       else await off_disk(roots, _free_name, target_dir, filename))
-        tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}"
+                       else await off_disk(roots, _free_name, target_dir, filename, reserved))
+        tmp_path = (state.part_path if state and state.part_path
+                    else target_dir / uploads_mod.part_name(stored_name))
         final_path = target_dir / stored_name
 
         if chunk_index == UPLOAD_PROBE_INDEX:
@@ -361,6 +365,22 @@ class UploadMixin:
         await off_disk(roots, _append_chunk, tmp_path, chunk_bytes, chunk_index == 0)
         uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes))
 
+        last = chunk_index + 1 >= total_chunks
+        if last:
+            # Published before the last ack, so the ack names the file as it is
+            # on disk: publication never replaces a file, and may have had to
+            # take another free name for this one.
+            uploads.drop(user_id, rel_dir, filename)
+            try:
+                stored_name = await off_disk(roots, publish_upload, tmp_path, target_dir,
+                                             stored_name, filename,
+                                             uploads.reserved_names(rel_dir))
+            except OSError as e:
+                log.warning("Upload %s could not be published: %s", stored_name, e)
+                _refuse("The file could not be stored", "store_failed")
+                return
+            final_path = target_dir / stored_name
+
         self._send(file_upload_ack_wire(
             gek, self._group_id or "",
             upload_id=upload_id,
@@ -372,9 +392,7 @@ class UploadMixin:
             dir=rel_dir,
         ))
 
-        if chunk_index + 1 >= total_chunks:
-            uploads.drop(user_id, rel_dir, filename)
-            await off_disk(roots, tmp_path.rename, final_path)
+        if last:
             log.info("Upload complete: %s (%d chunks, %d bytes)",
                      stored_name, total_chunks, state.bytes)
             self._audit("file_upload", f"{rel_dir}/{stored_name}")
diff --git a/packages/meshbay-node/src/meshbay_node/uploads.py b/packages/meshbay-node/src/meshbay_node/uploads.py
index dd31e1e..92b854e 100644
--- a/packages/meshbay-node/src/meshbay_node/uploads.py
+++ b/packages/meshbay-node/src/meshbay_node/uploads.py
@@ -24,6 +24,7 @@ build first.
 
 from __future__ import annotations
 
+import secrets
 import time
 from collections.abc import Iterable
 from dataclasses import dataclass, field
@@ -34,6 +35,16 @@ from pathlib import Path
 # recognise one, and a second spelling of it would be a bug nobody could see.
 PART_SUFFIX = ".part"
 
+
+def part_name(stored_name: str) -> str:
+    """The `.part` one upload writes: its final name, a tag of its own, `.part`.
+
+    Its own, because the final name alone is shared: two uploads that settled
+    on one name — two groups hosting one folder, each with its own lock —
+    would write one file, the second truncating the first.
+    """
+    return f"{stored_name}.{secrets.token_hex(4)}{PART_SUFFIX}"
+
 # How long a `.part` with no upload behind it is kept before it is deleted.
 #
 # Generous on purpose. The cost of waiting is disk; the cost of being wrong is
@@ -107,6 +118,15 @@ class PartialUploads:
     def drop(self, user_id: str, rel_dir: str, filename: str) -> Partial | None:
         return self._by_key.pop((user_id, rel_dir, filename), None)
 
+    def reserved_names(self, rel_dir: str) -> set[str]:
+        """The final names uploads in flight into `rel_dir` will take.
+
+        None of them exists on disk yet, so a name check that looked only at
+        the directory would hand the same name to a second upload.
+        """
+        return {state.stored_name for (_u, d, _f), state in self._by_key.items()
+                if d == rel_dir}
+
     def __len__(self) -> int:
         return len(self._by_key)