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, 24 insertions, 6 deletions

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}")