From e25332a90b86e0198a4a8b3e1abe4488b0b6f4bc Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 07:39:18 +0000
Subject: [PATCH 01/11] test(api): upload robustness over a real TCP socket

A uvicorn subprocess on 127.0.0.1 drives the volume upload routes the way
the phone reaches them: raw-socket PUTs that abort mid-body or stay open
while a retry to the same signed URL lands, a completed-PUT replay, a PUT
after the URL expires, wrong Content-MD5 / body / length, eight concurrent
PUTs racing two open event long-polls, and the storage scope rule that an
Authorization header is never sent and never required.

3 of 7 fail on alpha 4bcbedba: an aborted PUT leaks its temp file, a PUT
that lands while an earlier attempt still writes 500s the first attempt,
and a short body leaks its temp file. Fixes follow in the next commit.
---
 backend/server/api/tests/test_uploads_tcp.py | 512 +++++++++++++++++++
 1 file changed, 512 insertions(+)
 create mode 100644 backend/server/api/tests/test_uploads_tcp.py

diff --git a/backend/server/api/tests/test_uploads_tcp.py b/backend/server/api/tests/test_uploads_tcp.py
new file mode 100644
index 00000000..bad5a528
--- /dev/null
+++ b/backend/server/api/tests/test_uploads_tcp.py
@@ -0,0 +1,512 @@
+"""Upload volume routes over a real TCP socket, the way the phone reaches them.
+
+Every other upload test runs through ``TestClient``, which never drops a
+connection and never lets uvicorn's event loop multiplex requests, so it cannot
+see what an interrupted PUT, a retry on the same signed URL, or concurrent PUTs
+racing open event long-polls do to the volume. ``CaptureUploader`` retries a
+dropped PUT on the *same* signed URL (a transport error is ``Transient``, only a
+403/410 goes back to register for a fresh one), resends sealed bytes once on a
+400, and honors ``Retry-After`` on a 429, so the server side of those rules is
+what these tests pin:
+
+- an aborted PUT commits nothing and leaves no temp file behind;
+- a PUT that arrives while an earlier attempt is still writing cannot corrupt
+  the committed object (`prepare_tmp` used to hand both attempts the same
+  `{key}.uploading` path — the Greptile finding on #247 about parallel uploads);
+- a replay of a completed PUT succeeds;
+- a PUT after the token expired is 410 and a fresh URL commits;
+- a wrong Content-MD5, a wrong body, a short body and a wrong Content-Length
+  are refused with nothing committed;
+- 8 concurrent PUTs alongside 2 open event long-polls all finish inside a
+  bound, with 429s retried the way the app retries them;
+- an Authorization header on a storage PUT is ignored, never required: the
+  app's `ScopedCaptureHTTP` refuses to send one (signed headers only).
+
+The server runs as a subprocess bound to 127.0.0.1 on an ephemeral port with
+memory state and local volume storage; no cloud service, no real data.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import socket
+import struct
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+from urllib.parse import urlparse
+
+import httpx
+import pytest
+
+from server.api.keys import upload_key
+from server.common.backends.storage_volume import mint_token
+
+TOKEN = "d" * 40
+SECRET = "h" * 32
+BACKEND_DIR = Path(__file__).resolve().parents[3]
+
+RUNNER = """
+import json, sys
+import uvicorn
+from server.api.app import create_app
+from server.common.config import Settings
+
+settings = Settings(**json.loads(sys.argv[1]))
+app = create_app(settings=settings)
+uvicorn.run(app, host="127.0.0.1", port=int(sys.argv[2]), log_level="warning", access_log=False)
+"""
+
+
+class Server:
+    """One create_app under uvicorn on 127.0.0.1, plus its volume root."""
+
+    def __init__(self, tmp_path: Path) -> None:
+        self.root = tmp_path / "vol"
+        port_socket = socket.socket()
+        port_socket.bind(("127.0.0.1", 0))
+        self.port = port_socket.getsockname()[1]
+        port_socket.close()
+        settings = {
+            "SMP_ENVIRONMENT": "dev",
+            "HSS_AUTH_MODE": "static",
+            "CAPTURE_API_TOKENS": f"device:{TOKEN}",
+            "UPLOAD_URL_HMAC_SECRET": SECRET,
+            "STORAGE_BACKEND": "volume",
+            "STATE_BACKEND": "memory",
+            "VOLUME_ROOT": str(self.root),
+            "CAPTURE_API_PUBLIC_URL": f"http://127.0.0.1:{self.port}",
+            "HSS_INGEST_HOOKS": "null",
+            "VISION_EXTERNAL_CALLS": "on",
+            "GIT_SHA": "0123456789abcdef",
+            "SMP_SERVICE_NAME": "capture-api",
+        }
+        self.process = subprocess.Popen(
+            [sys.executable, "-c", RUNNER, json.dumps(settings), str(self.port)],
+            cwd=str(BACKEND_DIR),
+            env={"PYTHONPATH": str(BACKEND_DIR), "PATH": "/usr/bin:/bin"},
+        )
+        self.client = httpx.Client(
+            base_url=f"http://127.0.0.1:{self.port}",
+            headers={"Authorization": f"Bearer {TOKEN}"},
+            timeout=httpx.Timeout(15.0),
+            limits=httpx.Limits(max_connections=16),
+        )
+        deadline = time.monotonic() + 30
+        last_error: Exception | None = None
+        while time.monotonic() < deadline:
+            try:
+                if self.client.get("/v1/healthz").status_code == 200:
+                    return
+            except httpx.TransportError as error:
+                last_error = error
+            time.sleep(0.1)
+        self.close()
+        raise RuntimeError(f"the test server never became healthy: {last_error}")
+
+    def close(self) -> None:
+        self.client.close()
+        self.process.terminate()
+        try:
+            self.process.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            self.process.kill()
+            self.process.wait(timeout=5)
+
+
+@pytest.fixture
+def server(tmp_path):
+    live = Server(tmp_path)
+    yield live
+    live.close()
+
+
+def _md5_b64(data: bytes) -> str:
+    return base64.b64encode(hashlib.md5(data).digest()).decode()
+
+
+def _pattern(size: int, seed: int = 0) -> bytes:
+    return bytes((index * 31 + seed * 7 + 11) % 251 for index in range(size))
+
+
+def _create_capture(server: Server, packet_id: str) -> str:
+    created = server.client.post(
+        "/v1/captures",
+        json={
+            "packetId": packet_id,
+            "formatVersion": "0.4",
+            "tier": "photo_only",
+            "flow": "photo",
+            "device": {"model": "iPhone", "systemVersion": "18", "appVersion": "1"},
+        },
+        headers={"Idempotency-Key": packet_id},
+    )
+    assert created.status_code == 201, created.text
+    return created.json()["captureId"]
+
+
+def _register(server: Server, capture_id: str, path: str, data: bytes) -> dict:
+    """The app's register call; returns the signed PUT target it would use."""
+    registered = server.client.post(
+        f"/v1/captures/{capture_id}/files",
+        json={
+            "files": [
+                {
+                    "path": path,
+                    "bytes": len(data),
+                    "sha256": hashlib.sha256(data).hexdigest(),
+                    "md5": _md5_b64(data),
+                    "role": "still",
+                    "contentType": "image/jpeg",
+                }
+            ]
+        },
+    )
+    assert registered.status_code == 200, registered.text
+    entry = registered.json()["files"][0]
+    assert entry["path"] == path and entry["state"] == "pending"
+    target = entry["upload"]
+    assert target["method"] == "PUT" and target["url"]
+    return target
+
+
+def _put(server: Server, target: dict, data: bytes, **extra_headers: str) -> httpx.Response:
+    headers = dict(target["headers"])
+    headers.update(extra_headers)
+    return server.client.put(target["url"], content=data, headers=headers)
+
+
+def _commit(server: Server, capture_id: str, path: str, data: bytes) -> dict:
+    committed = server.client.post(
+        f"/v1/captures/{capture_id}/files:commit",
+        json={"files": [{"path": path, "sha256": hashlib.sha256(data).hexdigest()}]},
+    )
+    assert committed.status_code == 200, committed.text
+    return committed.json()
+
+
+def _stored(server: Server, capture_id: str, path: str) -> bytes:
+    object_path = server.root / "uploads" / capture_id / path
+    assert object_path.is_file(), f"no committed object at {object_path}"
+    return object_path.read_bytes()
+
+
+def _leftover_tmp(server: Server) -> list[str]:
+    return [str(path) for path in server.root.rglob("*") if path.is_file() and path.name.endswith(".uploading")]
+
+
+class RawPut:
+    """A PUT sent by hand, so the test can stop mid-body or replay it while open."""
+
+    def __init__(self, port: int, url: str, content_type: str, md5_b64: str, total: int) -> None:
+        self.path = urlparse(url).path
+        self.total = total
+        self.sock = socket.create_connection(("127.0.0.1", port), timeout=10)
+        self.head = (
+            f"PUT {self.path} HTTP/1.1\r\n"
+            f"Host: 127.0.0.1:{port}\r\n"
+            f"Content-Type: {content_type}\r\n"
+            f"Content-MD5: {md5_b64}\r\n"
+            f"Content-Length: {total}\r\n"
+            "Connection: close\r\n\r\n"
+        ).encode()
+
+    def send_head(self) -> None:
+        self.sock.sendall(self.head)
+
+    def send_bytes(self, data: bytes) -> None:
+        self.sock.sendall(data)
+
+    def abort(self) -> None:
+        """Close with an RST, as a phone does when it gives up on a stalled PUT."""
+        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+        self.sock.close()
+
+    def response(self) -> tuple[int, bytes]:
+        """Send nothing more (the peer half-closes) and read the whole response."""
+        try:
+            self.sock.shutdown(socket.SHUT_WR)
+        except OSError:
+            pass
+        chunks: list[bytes] = []
+        while True:
+            try:
+                chunk = self.sock.recv(65536)
+            except OSError:
+                break
+            if not chunk:
+                break
+            chunks.append(chunk)
+        self.sock.close()
+        raw = b"".join(chunks)
+        head, _, body = raw.partition(b"\r\n\r\n")
+        status = int(head.split(b"\r\n")[0].split()[1])
+        return status, body
+
+
+def _wait_for_tmp_to_settle(server: Server, seconds: float = 5.0) -> list[str]:
+    """Give the server a moment to notice a dropped connection, then report leftovers."""
+    deadline = time.monotonic() + seconds
+    leftover = _leftover_tmp(server)
+    while leftover and time.monotonic() < deadline:
+        time.sleep(0.1)
+        leftover = _leftover_tmp(server)
+    return leftover
+
+
+def test_aborted_put_commits_nothing_and_a_retry_to_the_same_url_succeeds(server: Server) -> None:
+    capture_id = _create_capture(server, "tcp-abort-1")
+    data = _pattern(256 * 1024)
+    target = _register(server, capture_id, "photos/a.jpg", data)
+    md5 = _md5_b64(data)
+
+    aborted = RawPut(server.port, target["url"], "image/jpeg", md5, len(data))
+    aborted.send_head()
+    aborted.send_bytes(data[: len(data) // 2])
+    # Let the handler start and settle into its write loop first, so the abort
+    # cannot win a race against request dispatch.
+    time.sleep(0.5)
+    aborted.abort()
+
+    leftover = _wait_for_tmp_to_settle(server)
+    assert leftover == [], f"an aborted PUT left temp files behind: {leftover}"
+    answer = _commit(server, capture_id, "photos/a.jpg", data)
+    assert answer["notFound"] == ["photos/a.jpg"], "an aborted PUT must not commit anything"
+
+    # The app's rule: a transport error is transient, so the same signed URL is retried.
+    retried = _put(server, target, data)
+    assert retried.status_code == 200, retried.text
+    answer = _commit(server, capture_id, "photos/a.jpg", data)
+    assert answer["committed"] == ["photos/a.jpg"]
+    assert _stored(server, capture_id, "photos/a.jpg") == data
+
+
+def test_a_put_while_an_earlier_attempt_is_still_writing_never_corrupts_the_object(server: Server) -> None:
+    """The retry for a dropped PUT reuses the signed URL while the first attempt's
+    handler can still be draining its body. Both attempts carry the same sealed
+    bytes, so whichever commits, the object must be exactly those bytes."""
+    capture_id = _create_capture(server, "tcp-overlap-1")
+    data = _pattern(256 * 1024, seed=3)
+    target = _register(server, capture_id, "photos/b.jpg", data)
+    md5 = _md5_b64(data)
+
+    first = RawPut(server.port, target["url"], "image/jpeg", md5, len(data))
+    first.send_head()
+    first.send_bytes(data[: len(data) // 2])
+    # Let the first attempt's handler create its temp file and settle into its
+    # write loop, so the overlap below is deterministic.
+    time.sleep(0.5)
+
+    second = _put(server, target, data)
+    assert second.status_code == 200, second.text
+
+    first.send_bytes(data[len(data) // 2 :])
+    status, body = first.response()
+    assert status == 200, f"the still-open first attempt must commit cleanly, got {status}: {body[:200]!r}"
+
+    answer = _commit(server, capture_id, "photos/b.jpg", data)
+    assert answer["committed"] == ["photos/b.jpg"], answer
+    assert _stored(server, capture_id, "photos/b.jpg") == data, "the committed object is not the sealed bytes"
+    assert _leftover_tmp(server) == []
+
+
+def test_replaying_a_completed_put_succeeds(server: Server) -> None:
+    capture_id = _create_capture(server, "tcp-replay-1")
+    data = _pattern(64 * 1024, seed=5)
+    target = _register(server, capture_id, "photos/c.jpg", data)
+
+    first = _put(server, target, data)
+    assert first.status_code == 200, first.text
+    replay = _put(server, target, data)
+    assert replay.status_code == 200, replay.text
+
+    answer = _commit(server, capture_id, "photos/c.jpg", data)
+    assert answer["committed"] == ["photos/c.jpg"]
+    assert _stored(server, capture_id, "photos/c.jpg") == data
+
+
+def test_put_after_the_url_expires_is_410_and_a_wrong_op_token_is_403(server: Server) -> None:
+    capture_id = _create_capture(server, "tcp-expiry-1")
+    data = _pattern(32 * 1024, seed=7)
+    path = "photos/d.jpg"
+    target = _register(server, capture_id, path, data)
+    key = upload_key(capture_id, path)
+
+    expired = mint_token(
+        {"op": "put", "key": key, "md5": _md5_b64(data), "ct": "image/jpeg", "exp": int(time.time()) - 30},
+        SECRET.encode(),
+    )
+    late = server.client.put(
+        f"/v1/uploads/{expired}", content=data, headers={"Content-Type": "image/jpeg", "Content-MD5": _md5_b64(data)}
+    )
+    assert late.status_code == 410, late.text
+    assert late.json()["errors"][0]["code"] == "token_expired"
+
+    get_token = mint_token({"op": "get", "key": key, "md5": "", "ct": "", "exp": int(time.time()) + 300}, SECRET.encode())
+    wrong_op = server.client.put(
+        f"/v1/uploads/{get_token}",
+        content=data,
+        headers={"Content-Type": "image/jpeg", "Content-MD5": _md5_b64(data)},
+    )
+    assert wrong_op.status_code == 403, wrong_op.text
+
+    answer = _commit(server, capture_id, path, data)
+    assert answer["notFound"] == [path], "an expired-URL PUT must not commit anything"
+
+    # The app's rule: a spent URL goes back to register, which mints a fresh one.
+    # (A re-register inside the same second returns a byte-identical token — the
+    # mint is deterministic at second granularity — which the app PUTs again.)
+    fresh = _register(server, capture_id, path, data)
+    retried = _put(server, fresh, data)
+    assert retried.status_code == 200, retried.text
+    assert _commit(server, capture_id, path, data)["committed"] == [path]
+    assert _stored(server, capture_id, path) == data
+
+
+def test_wrong_md5_wrong_body_short_body_and_wrong_length_are_refused(server: Server) -> None:
+    capture_id = _create_capture(server, "tcp-digests-1")
+    data = _pattern(48 * 1024, seed=9)
+    path = "photos/e.jpg"
+    target = _register(server, capture_id, path, data)
+    md5 = _md5_b64(data)
+
+    wrong_header = _put(server, target, data, **{"Content-MD5": _md5_b64(b"other bytes")})
+    assert wrong_header.status_code == 400, wrong_header.text
+    assert wrong_header.json()["errors"][0]["code"] == "bad_digest"
+
+    wrong_content_type = server.client.put(
+        target["url"], content=data, headers={"Content-Type": "image/png", "Content-MD5": md5}
+    )
+    assert wrong_content_type.status_code == 400, wrong_content_type.text
+
+    wrong_body = _put(server, target, data[:-1] + bytes([data[-1] ^ 0xFF]))
+    assert wrong_body.status_code == 400, wrong_body.text
+    assert wrong_body.text == "BadDigest"
+
+    short = RawPut(server.port, target["url"], "image/jpeg", md5, len(data))
+    short.send_head()
+    short.send_bytes(data[: len(data) - 1])
+    # A body that ends before its declared length is a disconnect on the wire, so
+    # the server may be unable to answer at all; if an answer is readable it must
+    # not be a success. Either way nothing may commit.
+    try:
+        status, _body = short.response()
+        assert status // 100 != 2, f"a short body must not be accepted, got {status}"
+    except (OSError, IndexError, ValueError):
+        pass  # the server dropped the connection instead of answering
+
+    wrong_length = RawPut(server.port, target["url"], "image/jpeg", md5, len(data) - 1)
+    wrong_length.send_head()
+    wrong_length.send_bytes(data[:-1])
+    status, body = wrong_length.response()
+    assert status == 400, f"a declared length that does not match the registered object must be refused, got {status}"
+    assert json.loads(body)["errors"][0]["pointer"] == "/headers/content-length"
+
+    assert _leftover_tmp(server) == [], "a refused PUT must not leave temp files"
+    answer = _commit(server, capture_id, path, data)
+    assert answer["notFound"] == [path], "none of the refused PUTs may commit anything"
+
+
+def test_eight_concurrent_puts_with_two_open_event_polls_all_finish(server: Server) -> None:
+    capture_id = _create_capture(server, "tcp-burst-1")
+    files = {f"photos/f{index}.jpg": _pattern(32 * 1024, seed=index) for index in range(8)}
+    batch = [
+        {
+            "path": path,
+            "bytes": len(data),
+            "sha256": hashlib.sha256(data).hexdigest(),
+            "md5": _md5_b64(data),
+            "role": "still",
+            "contentType": "image/jpeg",
+        }
+        for path, data in files.items()
+    ]
+    registered = server.client.post(f"/v1/captures/{capture_id}/files", json={"files": batch})
+    assert registered.status_code == 200, registered.text
+    targets = {entry["path"]: entry["upload"] for entry in registered.json()["files"]}
+
+    poll_results: dict[int, tuple[float, int]] = {}
+
+    def poll(name: int) -> None:
+        started = time.monotonic()
+        answer = server.client.get(f"/v1/captures/{capture_id}/events", params={"after": 0, "wait": 10})
+        poll_results[name] = (time.monotonic() - started, answer.status_code)
+
+    polls = [threading.Thread(target=poll, args=(name,)) for name in range(2)]
+    for thread in polls:
+        thread.start()
+
+    put_results: dict[str, tuple[float, int | None, str]] = {}
+    lock = threading.Lock()
+
+    def put(path: str) -> None:
+        data = files[path]
+        target = targets[path]
+        started = time.monotonic()
+        # The app's rules: a 429 waits Retry-After and retries; a dropped
+        # connection retries the same URL. Both bounded.
+        last_status: int | None = None
+        for _attempt in range(6):
+            try:
+                answer = _put(server, target, data)
+            except httpx.TransportError:
+                time.sleep(0.2)
+                continue
+            if answer.status_code == 429:
+                time.sleep(min(float(answer.headers.get("Retry-After", "1")), 2))
+                continue
+            last_status = answer.status_code
+            break
+        with lock:
+            put_results[path] = (time.monotonic() - started, last_status, path)
+
+    threads = [threading.Thread(target=put, args=(path,), name=path) for path in files]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=40)
+        assert not thread.is_alive(), f"the PUT of {thread.name} never finished"
+    for thread in polls:
+        thread.join(timeout=25)
+        assert not thread.is_alive(), "an event long-poll stalled past its bound"
+
+    for path, (elapsed, status, _name) in put_results.items():
+        assert status == 200, f"{path} never committed: last status {status}"
+        assert elapsed < 25, f"{path} took {elapsed:.1f}s"
+    for elapsed, status in poll_results.values():
+        assert status == 200
+        assert elapsed < 15, f"a long-poll took {elapsed:.1f}s, longer than its wait"
+
+    commit_body = {"files": [{"path": path, "sha256": hashlib.sha256(data).hexdigest()} for path, data in files.items()]}
+    committed = server.client.post(f"/v1/captures/{capture_id}/files:commit", json=commit_body)
+    assert committed.status_code == 200, committed.text
+    answer = committed.json()
+    assert answer["committed"] == sorted(files), answer
+    assert not answer["notFound"] and not answer["mismatch"]
+    for path, data in files.items():
+        assert _stored(server, capture_id, path) == data
+    assert _leftover_tmp(server) == []
+
+
+def test_a_storage_put_ignores_an_authorization_header(server: Server) -> None:
+    """`ScopedCaptureHTTP.upload` refuses to send Authorization on a storage PUT,
+    so the storage boundary must never require one. A header that leaks in
+    anyway is ignored: it grants nothing and blocks nothing."""
+    capture_id = _create_capture(server, "tcp-auth-1")
+    data = _pattern(24 * 1024, seed=11)
+    path = "photos/g.jpg"
+    target = _register(server, capture_id, path, data)
+
+    with_token = _put(server, target, data, **{"Authorization": f"Bearer {TOKEN}"})
+    assert with_token.status_code == 200, with_token.text
+    assert _commit(server, capture_id, path, data)["committed"] == [path]
+
+    second = _register(server, capture_id, "photos/g2.jpg", data)
+    with_garbage = _put(server, second, data, **{"Authorization": "Basic !!!not-a-credential"})
+    assert with_garbage.status_code == 200, with_garbage.text
+    assert _commit(server, capture_id, "photos/g2.jpg", data)["committed"] == ["photos/g2.jpg"]
+    assert _stored(server, capture_id, path) == data

From eabd57debe84bc8860a74863092c089dd2468f11 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 07:56:32 +0000
Subject: [PATCH 02/11] fix(api): make volume PUT attempts collision-free and
 disconnect-clean

prepare_tmp now returns a unique path per attempt under .uploads/singles/
(a prefix list_prefix excludes), so a retried PUT to the same signed URL
can no longer truncate the file under an earlier attempt's still-open
handle or 500 its commit on a vanished temp path. Each attempt commits
its own complete bytes; the last commit wins, which is the replay rule.

The single-put route now catches starlette ClientDisconnect, discards
the attempt's temp file, and answers 400 BadDigest, so an aborted or
short PUT leaves no temp litter and the retry starts clean.

Both failures reproduced by the previous commit's TCP suite on alpha
4bcbedba; the suite is 7/7 green with these fixes.

Co-authored-by: Obvious <obvious@obvious.ai>
---
 backend/server/api/routes/uploads.py          |  9 +++++++++
 .../server/common/backends/storage_volume.py  | 19 +++++++++++++++++++
 2 files changed, 28 insertions(+)

diff --git a/backend/server/api/routes/uploads.py b/backend/server/api/routes/uploads.py
index 64eeb112..09e987c2 100644
--- a/backend/server/api/routes/uploads.py
+++ b/backend/server/api/routes/uploads.py
@@ -8,6 +8,7 @@
 
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse, Response
+from starlette.requests import ClientDisconnect
 
 from server.api.deps import get_deps
 from server.api.errors import fail
@@ -145,6 +146,14 @@ async def put_upload(token: str, request: Request):
             if tmp is not None:
                 deps.storage.discard_tmp(tmp)
             return Response(content=b"BadDigest", status_code=400, media_type="text/plain")
+        except ClientDisconnect:
+            # The phone dropped the connection mid-body — it does this when a
+            # retried PUT opens while the old attempt still drains. Nothing is
+            # committed and the temp file goes away, so the retry starts clean
+            # instead of leaving `{key}.uploading` litter in the volume.
+            if tmp is not None:
+                deps.storage.discard_tmp(tmp)
+            return Response(content=b"BadDigest", status_code=400, media_type="text/plain")
         finally:
             deps.upload_slots.release()
 
diff --git a/backend/server/common/backends/storage_volume.py b/backend/server/common/backends/storage_volume.py
index 0ef222ce..3f76f4c4 100644
--- a/backend/server/common/backends/storage_volume.py
+++ b/backend/server/common/backends/storage_volume.py
@@ -7,6 +7,7 @@
 import hmac
 import json
 import time
+import uuid
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -72,6 +73,24 @@ def copy(self, src: str, dst: str, *, if_generation_match: str | None = None):
                 raise
             raise PreconditionFailed(src) from exc
 
+    def prepare_tmp(self, key: str) -> Path:
+        """One temp file per PUT attempt, not one per key.
+
+        The inherited `{key}.uploading` name collides exactly when the phone
+        retries a dropped PUT on the same signed URL while the first attempt's
+        handler can still be draining its body: the retry's open truncated the
+        file under the first attempt's handle, and whichever attempt committed
+        second either moved the other's bytes or crashed on a vanished temp
+        path (a 500 on valid bytes). A unique name under `.uploads/` — a prefix
+        `list_prefix` already excludes — makes every attempt commit its own
+        complete bytes; the last commit wins, which is the replay rule.
+        """
+
+        self._uploads.mkdir(parents=True, exist_ok=True)
+        singles = self._uploads / "singles"
+        singles.mkdir(parents=True, exist_ok=True)
+        return singles / f"{uuid.uuid4().hex}.uploading"
+
     def signed_put_url(self, key: str, content_type: str, md5_b64: str, ttl_s: int) -> SignedUrl:
         exp, stamp = self._expires(ttl_s)
         token = mint_token({"op": "put", "key": key, "md5": md5_b64, "ct": content_type, "exp": exp}, self.hmac_secret)

From e1f6e983f00b25baf30c9cf3e0b4d29cfef60515 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 07:56:48 +0000
Subject: [PATCH 03/11] test(api): lint and format pass on the TCP suite

Co-authored-by: Obvious <obvious@obvious.ai>
---
 backend/server/api/tests/test_uploads_tcp.py | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/backend/server/api/tests/test_uploads_tcp.py b/backend/server/api/tests/test_uploads_tcp.py
index bad5a528..7a1c4691 100644
--- a/backend/server/api/tests/test_uploads_tcp.py
+++ b/backend/server/api/tests/test_uploads_tcp.py
@@ -333,7 +333,7 @@ def test_put_after_the_url_expires_is_410_and_a_wrong_op_token_is_403(server: Se
     capture_id = _create_capture(server, "tcp-expiry-1")
     data = _pattern(32 * 1024, seed=7)
     path = "photos/d.jpg"
-    target = _register(server, capture_id, path, data)
+    _register(server, capture_id, path, data)
     key = upload_key(capture_id, path)
 
     expired = mint_token(
@@ -346,7 +346,9 @@ def test_put_after_the_url_expires_is_410_and_a_wrong_op_token_is_403(server: Se
     assert late.status_code == 410, late.text
     assert late.json()["errors"][0]["code"] == "token_expired"
 
-    get_token = mint_token({"op": "get", "key": key, "md5": "", "ct": "", "exp": int(time.time()) + 300}, SECRET.encode())
+    get_token = mint_token(
+        {"op": "get", "key": key, "md5": "", "ct": "", "exp": int(time.time()) + 300}, SECRET.encode()
+    )
     wrong_op = server.client.put(
         f"/v1/uploads/{get_token}",
         content=data,
@@ -481,7 +483,9 @@ def put(path: str) -> None:
         assert status == 200
         assert elapsed < 15, f"a long-poll took {elapsed:.1f}s, longer than its wait"
 
-    commit_body = {"files": [{"path": path, "sha256": hashlib.sha256(data).hexdigest()} for path, data in files.items()]}
+    commit_body = {
+        "files": [{"path": path, "sha256": hashlib.sha256(data).hexdigest()} for path, data in files.items()]
+    }
     committed = server.client.post(f"/v1/captures/{capture_id}/files:commit", json=commit_body)
     assert committed.status_code == 200, committed.text
     answer = committed.json()

From 62d2a8011a6c1a30002a348e9533f3a7bfa89cd3 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 11:43:45 +0000
Subject: [PATCH 04/11] test(api): two-writer commit and singles-temp sweep
 regressions

The volume path hands every PUT attempt its own temp file, so an old
upload still in flight and a re-registered upload can both reach
commit_tmp. A gate holds the first committer at the metadata lock until
a second commit has fully completed, reproducing the A-rename,
B-rename, B-metadata, A-metadata interleave deterministically; the
sweep test covers the .uploads/singles temporaries the uploads walk
never visits. Both fail on the current head.
---
 .../api/tests/test_storage_and_sweep.py       | 145 ++++++++++++++++++
 1 file changed, 145 insertions(+)

diff --git a/backend/server/api/tests/test_storage_and_sweep.py b/backend/server/api/tests/test_storage_and_sweep.py
index 533aca1d..30ca91f9 100644
--- a/backend/server/api/tests/test_storage_and_sweep.py
+++ b/backend/server/api/tests/test_storage_and_sweep.py
@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import base64
+import hashlib
 import json
 import os
+import threading
 import time
 from datetime import datetime, timedelta
 from urllib.parse import urlparse
@@ -144,3 +147,145 @@ def test_volume_sweeper_ages_uploads(client, tmp_path):
     _ = urlparse
     _ = StageRecord
     _ = tmp_path
+
+
+def _md5_b64(data: bytes) -> str:
+    return base64.b64encode(hashlib.md5(data).digest()).decode()
+
+
+class _GatedLock:
+    """Lets every acquire through except the first commit's, which it holds.
+
+    The hold happens before the real lock is taken, so the held writer blocks
+    nothing else. Reentrant acquires from the committing thread itself pass
+    through untouched.
+    """
+
+    def __init__(self, real: threading.RLock, gate: _HoldFirstCommitter) -> None:
+        self._real = real
+        self._gate = gate
+        self._owner: int | None = None
+        self._depth = 0
+
+    def acquire(self, *args, **kwargs) -> bool:
+        me = threading.get_ident()
+        if self._depth and self._owner == me:
+            self._depth += 1
+            return self._real.acquire(*args, **kwargs)
+        gate = self._gate
+        if getattr(gate.tl, "committing", None) == gate.key and not gate.held.is_set():
+            gate.held.set()
+            deadline = time.monotonic() + gate.timeout
+            while gate.completed == 0 and time.monotonic() < deadline:
+                time.sleep(0.005)
+        got = self._real.acquire(*args, **kwargs)
+        if got:
+            self._owner = me
+            self._depth += 1
+        return got
+
+    def release(self) -> None:
+        self._depth -= 1
+        if not self._depth:
+            self._owner = None
+        self._real.release()
+
+    def __enter__(self) -> _GatedLock:
+        self.acquire()
+        return self
+
+    def __exit__(self, *_exc: object) -> None:
+        self.release()
+
+
+class _HoldFirstCommitter:
+    """Deterministic scheduler for two overlapping commits on one key.
+
+    The first committer is held at the metadata lock until a second commit has
+    fully completed — the interleave an old in-flight upload racing a
+    re-register produces when commits run on more than one thread. Holding the
+    first committer *after* its final-file replace but *before* its metadata
+    write is only possible when the replace sits outside the lock; once the
+    two are serialized, the hold lands before the replace and both writers
+    commit atomically.
+    """
+
+    def __init__(self, key: str, timeout: float = 10.0) -> None:
+        self.key = key
+        self.timeout = timeout
+        self.completed = 0
+        self.held = threading.Event()
+        self.tl = threading.local()
+
+    def install(self, storage) -> None:
+        original = type(storage).commit_tmp
+
+        def commit_tmp(key, tmp, content_type, md5_b64, nbytes, _storage=storage, _original=original):
+            self.tl.committing = key
+            try:
+                _original(_storage, key, tmp, content_type, md5_b64, nbytes)
+            finally:
+                if key == self.key:
+                    self.completed += 1
+                self.tl.committing = None
+
+        storage.commit_tmp = commit_tmp
+        storage._lock = _GatedLock(storage._lock, self)
+
+
+def test_volume_commit_tmp_leaves_bytes_and_metadata_agreeing_under_two_writers(tmp_path):
+    """Two overlapping commits for one key must not mix bytes and metadata.
+
+    The volume path hands every PUT attempt its own temp file, so an old
+    upload still in flight and a re-registered upload can both reach
+    commit_tmp. Whichever commits last must win the bytes AND the metadata
+    together; head() reporting one writer's digest and size while the volume
+    holds the other's bytes makes files:commit mark a valid upload as a
+    mismatch.
+    """
+    storage = VolumeStorage(tmp_path / "vol", "http://testserver", b"secret")
+    key = "uploads/cap/f.jpg"
+    old = b"old upload bytes still in flight"
+    new = b"re-registered replacement bytes!"
+    gate = _HoldFirstCommitter(key)
+    gate.install(storage)
+
+    def commit(data: bytes) -> None:
+        tmp = storage.prepare_tmp(key)
+        tmp.write_bytes(data)
+        storage.commit_tmp(key, tmp, "image/jpeg", _md5_b64(data), len(data))
+
+    first = threading.Thread(target=commit, args=(old,))
+    first.start()
+    assert gate.held.wait(5), "the first committer never reached the metadata lock"
+    second = threading.Thread(target=commit, args=(new,))
+    second.start()
+    first.join(10)
+    second.join(10)
+    assert not first.is_alive() and not second.is_alive(), "a committer deadlocked"
+
+    stored = storage.get_bytes(key)
+    meta = storage.head(key)
+    assert stored in (old, new)
+    assert meta.bytes == len(stored)
+    assert meta.md5_b64 == _md5_b64(stored), "head() reports metadata from another writer than the committed bytes"
+
+
+def test_volume_sweeper_ages_stale_single_attempt_temps(client):
+    """Per-attempt upload temporaries live outside the uploads walk — in
+    .uploads/singles — so the sweeper must visit them there. A process killed
+    between prepare_tmp and commit_tmp otherwise strands the file forever."""
+    root = client.app_state.settings.VOLUME_ROOT
+    singles = os.path.join(root, ".uploads", "singles")
+    os.makedirs(singles, exist_ok=True)
+    stale = os.path.join(singles, "stale-attempt.uploading")
+    live = os.path.join(singles, "live-attempt.uploading")
+    open(stale, "wb").write(b"abandoned attempt")
+    open(live, "wb").write(b"attempt still settling")
+    stale_time = time.time() - 8 * 86400
+    live_time = time.time() - 6 * 86400
+    os.utime(stale, (stale_time, stale_time))
+    os.utime(live, (live_time, live_time))
+    ingest_sweep(client.app_state)
+    assert not os.path.exists(stale), "a stale per-attempt temporary must be swept"
+    assert os.path.exists(live), "a fresh per-attempt temporary must survive"

From 9389d0de5e1b97b2ec4cfdc4ae2b80be98e33084 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 11:43:45 +0000
Subject: [PATCH 05/11] fix(common): commit_tmp replaces the final file under
 the metadata lock

Renaming the temp file into place before taking the metadata lock let
two commits for one key interleave so the volume held one writer's
bytes while head() reported the other's digest and size, and
files:commit then marked a valid upload as a mismatch. Under the lock
every commit is atomic: the last committer wins the bytes and the
metadata together, which is the replay rule.
---
 backend/server/common/backends/storage_local.py | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/backend/server/common/backends/storage_local.py b/backend/server/common/backends/storage_local.py
index 47d3a9cf..a90727b9 100644
--- a/backend/server/common/backends/storage_local.py
+++ b/backend/server/common/backends/storage_local.py
@@ -93,10 +93,17 @@ def discard_tmp(self, tmp: Path) -> None:
             tmp.unlink()
 
     def commit_tmp(self, key: str, tmp: Path, content_type: str, md5_b64: str, nbytes: int) -> None:
+        # The rename and the metadata write happen under one lock. Replacing the
+        # final file before taking the lock lets two commits for one key
+        # interleave as A-rename, B-rename, B-metadata, A-metadata: the volume
+        # ends up with B's bytes while head() reports A's digest and size, and
+        # files:commit then marks a valid upload as a mismatch. Under the lock
+        # every commit is atomic — the last committer wins the bytes and the
+        # metadata together, which is the replay rule.
         final = self._path(key)
-        final.parent.mkdir(parents=True, exist_ok=True)
-        tmp.replace(final)
         with self._lock:
+            final.parent.mkdir(parents=True, exist_ok=True)
+            tmp.replace(final)
             self._meta[key] = ObjectMeta(
                 bytes=nbytes,
                 md5_b64=md5_b64,

From c90e562943145beb424b18e9ff45b9a740fe6ac5 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 11:43:45 +0000
Subject: [PATCH 06/11] fix(api): sweep stale per-attempt upload temporaries

VolumeStorage.prepare_tmp parks one temporary per PUT attempt under
.uploads/singles, outside the uploads walk. Sweep them on the same
age rule, so a process killed between prepare_tmp and commit_tmp
strands nothing in the volume.
---
 backend/server/api/sweeper.py | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/backend/server/api/sweeper.py b/backend/server/api/sweeper.py
index 2b7a91c6..9e60347d 100644
--- a/backend/server/api/sweeper.py
+++ b/backend/server/api/sweeper.py
@@ -61,6 +61,16 @@ def _sweep_volume(root: Path) -> int:
             if path.is_file() and path.stat().st_mtime < now - 7 * 86400:
                 path.unlink()
                 deleted += 1
+    # VolumeStorage.prepare_tmp parks one temporary per PUT attempt under
+    # .uploads/singles, outside the uploads walk above. Sweep them on the same
+    # age rule, so a process killed between prepare_tmp and commit_tmp strands
+    # nothing in the volume.
+    singles = root / ".uploads" / "singles"
+    if singles.exists():
+        for path in singles.glob("*.uploading"):
+            if path.is_file() and path.stat().st_mtime < now - 7 * 86400:
+                path.unlink()
+                deleted += 1
     for path in root.rglob("*"):
         if path.is_dir() and path.name.endswith(".parts") and path.stat().st_mtime < now - 86400:
             shutil.rmtree(path, ignore_errors=True)

From 007d5d16d64f18f87ea83bcf9c4cd2247a323240 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 11:43:45 +0000
Subject: [PATCH 07/11] test(api): overlapping re-registered PUTs keep bytes
 and metadata agreeing

Drives the re-register race over a real socket: an old PUT held
mid-body while a full PUT for the same key commits, then the old
attempt finishes. Both must get 200 and the volume's metadata index
must agree with the committed bytes, whichever attempt won.
---
 backend/server/api/tests/test_uploads_tcp.py | 37 ++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/backend/server/api/tests/test_uploads_tcp.py b/backend/server/api/tests/test_uploads_tcp.py
index 7a1c4691..40465646 100644
--- a/backend/server/api/tests/test_uploads_tcp.py
+++ b/backend/server/api/tests/test_uploads_tcp.py
@@ -514,3 +514,40 @@ def test_a_storage_put_ignores_an_authorization_header(server: Server) -> None:
     assert with_garbage.status_code == 200, with_garbage.text
     assert _commit(server, capture_id, "photos/g2.jpg", data)["committed"] == ["photos/g2.jpg"]
     assert _stored(server, capture_id, path) == data
+
+
+def test_overlapping_puts_after_a_re_register_leave_bytes_and_metadata_agreeing(server: Server) -> None:
+    """A re-register mints a fresh URL for the same key while the old upload may
+    still be in flight. Both attempts get 200 and each commits its own complete
+    bytes; whichever commits last must win the bytes and the metadata together,
+    so the volume never holds one writer's bytes under another writer's
+    digest/size (that state would make files:commit mark a valid upload as a
+    mismatch). Commits run on the event-loop thread today, so this pins the
+    end-to-end invariant; the deterministic interleaving regression for the
+    storage method itself lives in test_storage_and_sweep.py."""
+    capture_id = _create_capture(server, "tcp-rereg-1")
+    old = _pattern(256 * 1024, seed=7)
+    new = _pattern(256 * 1024, seed=8)
+    old_target = _register(server, capture_id, "photos/d.jpg", old)
+    new_target = _register(server, capture_id, "photos/d.jpg", new)
+    assert old_target["url"] != new_target["url"], "a re-register must mint a fresh URL"
+
+    first = RawPut(server.port, old_target["url"], "image/jpeg", _md5_b64(old), len(old))
+    first.send_head()
+    first.send_bytes(old[: len(old) // 2])
+    # Let the first attempt's handler create its temp file and settle into its
+    # write loop, so the re-registered PUT below lands while it is still open.
+    time.sleep(0.5)
+    second = _put(server, new_target, new)
+    assert second.status_code == 200, second.text
+    first.send_bytes(old[len(old) // 2 :])
+    status, body = first.response()
+    assert status == 200, f"the still-open first attempt must commit cleanly, got {status}: {body[:200]!r}"
+
+    stored = _stored(server, capture_id, "photos/d.jpg")
+    assert stored in (old, new)
+    index = json.loads((server.root / ".hss-meta.json").read_text())
+    meta = index["objects"][upload_key(capture_id, "photos/d.jpg")]
+    assert meta["bytes"] == len(stored)
+    assert meta["md5_b64"] == _md5_b64(stored), "the metadata index disagrees with the committed bytes"
+    assert _leftover_tmp(server) == []

From 5fab0a83e0afc35d29c57c85ae18d5b8e18009a9 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 17:54:01 +0000
Subject: [PATCH 08/11] test(api): a dropped local PUT attempt must not take
 the retry's temp file

---
 .../api/tests/test_storage_and_sweep.py       | 28 ++++++-
 backend/server/api/tests/test_uploads_tcp.py  | 80 ++++++++++++++++++-
 2 files changed, 105 insertions(+), 3 deletions(-)

diff --git a/backend/server/api/tests/test_storage_and_sweep.py b/backend/server/api/tests/test_storage_and_sweep.py
index 30ca91f9..1e909e97 100644
--- a/backend/server/api/tests/test_storage_and_sweep.py
+++ b/backend/server/api/tests/test_storage_and_sweep.py
@@ -12,7 +12,7 @@
 from urllib.parse import urlparse
 
 from server.api.keys import final_key, upload_key
-from server.api.sweeper import ingest_sweep
+from server.api.sweeper import _sweep_volume, ingest_sweep
 from server.api.tests.conftest import create_body, photo_packet, sha256_hex
 from server.common.backends.storage_gcs import GcsStorage
 from server.common.backends.storage_local import LocalFsStorage
@@ -233,6 +233,32 @@ def commit_tmp(key, tmp, content_type, md5_b64, nbytes, _storage=storage, _origi
         storage._lock = _GatedLock(storage._lock, self)
 
 
+def test_local_sweeper_ages_stale_single_attempt_temps(tmp_path):
+    """LocalFsStorage mints the same per-attempt temporaries as VolumeStorage —
+    one file under .uploads/singles per PUT attempt, outside the uploads walk —
+    so the sweeper's `*.uploading` pattern covers both backends. The old
+    per-key `{key}.uploading` name sat beside the object under root and gave
+    two overlapping attempts a single shared file."""
+    root = tmp_path / "local"
+    storage = LocalFsStorage(root)
+    key = "uploads/cap/local.jpg"
+    first = storage.prepare_tmp(key)
+    second = storage.prepare_tmp(key)
+    assert first.parent == root / ".uploads" / "singles", "a local temp must land where the sweeper sweeps"
+    assert first.name.endswith(".uploading") and second.name.endswith(".uploading")
+    assert first != second, "two attempts for one key must not share a temp file"
+    assert storage.list_prefix("uploads/") == [], "a temp file must not list as an object"
+
+    stale = root / ".uploads" / "singles" / "killed-attempt.uploading"
+    stale.write_bytes(b"process died between prepare_tmp and commit_tmp")
+    stale_time = time.time() - 8 * 86400
+    os.utime(stale, (stale_time, stale_time))
+    deleted = _sweep_volume(root)
+    assert deleted >= 1, "the sweep must count the stale local temporary"
+    assert not stale.exists(), "a stale local per-attempt temporary must be swept"
+    assert first.exists() and second.exists(), "a fresh per-attempt temporary must survive"
+
+
 def test_volume_commit_tmp_leaves_bytes_and_metadata_agreeing_under_two_writers(tmp_path):
     """Two overlapping commits for one key must not mix bytes and metadata.
 
diff --git a/backend/server/api/tests/test_uploads_tcp.py b/backend/server/api/tests/test_uploads_tcp.py
index 40465646..88711f76 100644
--- a/backend/server/api/tests/test_uploads_tcp.py
+++ b/backend/server/api/tests/test_uploads_tcp.py
@@ -31,6 +31,7 @@
 import base64
 import hashlib
 import json
+import os
 import socket
 import struct
 import subprocess
@@ -44,6 +45,7 @@
 import pytest
 
 from server.api.keys import upload_key
+from server.api.sweeper import _sweep_volume
 from server.common.backends.storage_volume import mint_token
 
 TOKEN = "d" * 40
@@ -65,7 +67,7 @@
 class Server:
     """One create_app under uvicorn on 127.0.0.1, plus its volume root."""
 
-    def __init__(self, tmp_path: Path) -> None:
+    def __init__(self, tmp_path: Path, storage_backend: str = "volume") -> None:
         self.root = tmp_path / "vol"
         port_socket = socket.socket()
         port_socket.bind(("127.0.0.1", 0))
@@ -76,7 +78,7 @@ def __init__(self, tmp_path: Path) -> None:
             "HSS_AUTH_MODE": "static",
             "CAPTURE_API_TOKENS": f"device:{TOKEN}",
             "UPLOAD_URL_HMAC_SECRET": SECRET,
-            "STORAGE_BACKEND": "volume",
+            "STORAGE_BACKEND": storage_backend,
             "STATE_BACKEND": "memory",
             "VOLUME_ROOT": str(self.root),
             "CAPTURE_API_PUBLIC_URL": f"http://127.0.0.1:{self.port}",
@@ -125,6 +127,14 @@ def server(tmp_path):
     live.close()
 
 
+@pytest.fixture
+def local_server(tmp_path):
+    """The same one-app-under-uvicorn harness, on STORAGE_BACKEND=local."""
+    live = Server(tmp_path, storage_backend="local")
+    yield live
+    live.close()
+
+
 def _md5_b64(data: bytes) -> str:
     return base64.b64encode(hashlib.md5(data).digest()).decode()
 
@@ -314,6 +324,72 @@ def test_a_put_while_an_earlier_attempt_is_still_writing_never_corrupts_the_obje
     assert _leftover_tmp(server) == []
 
 
+def test_a_dropped_attempt_on_local_storage_leaves_the_retry_clean(local_server: Server) -> None:
+    """The same overlap the volume suite pins, on the local backend.
+
+    `LocalFsStorage.prepare_tmp` used to hand every attempt for a key the same
+    `{key}.uploading` path: the retry's open truncated the file under the
+    still-draining first attempt's handle, the dropped attempt's cleanup
+    unlinked the retry's temp, and the retry's commit_tmp failed with a 500 on
+    valid bytes. Both backends now mint one temp per attempt under
+    `.uploads/singles`, so a dropped attempt discards only its own file and
+    the sweeper's `*.uploading` pattern covers the location for both.
+    """
+    capture_id = _create_capture(local_server, "tcp-local-abort-1")
+    data = _pattern(256 * 1024, seed=13)
+    path = "photos/h.jpg"
+    _register(local_server, capture_id, path, data)
+    # Local storage signs PUTs as file:// URIs, so the test mints the token URL
+    # the route serves directly — which is also exactly the app's retry rule:
+    # both attempts reuse the one signed URL.
+    token = mint_token(
+        {
+            "op": "put",
+            "key": upload_key(capture_id, path),
+            "md5": _md5_b64(data),
+            "ct": "image/jpeg",
+            "exp": int(time.time()) + 600,
+        },
+        SECRET.encode(),
+    )
+    url = f"/v1/uploads/{token}"
+
+    first = RawPut(local_server.port, url, "image/jpeg", _md5_b64(data), len(data))
+    first.send_head()
+    first.send_bytes(data[: len(data) // 2])
+    # Let the first attempt's handler create its temp file and settle into its
+    # write loop, so the retry below opens while it is still writing.
+    time.sleep(0.5)
+
+    second = RawPut(local_server.port, url, "image/jpeg", _md5_b64(data), len(data))
+    second.send_head()
+    # Let the retry's handler create ITS temp file too, then drop the first
+    # attempt mid-body, exactly as a phone does when it gives up on a stall.
+    time.sleep(0.5)
+    first.abort()
+
+    status, body = second.response()
+    assert status == 200, f"the retry must commit cleanly after the first attempt drops, got {status}: {body[:200]!r}"
+
+    answer = _commit(local_server, capture_id, path, data)
+    assert answer["committed"] == [path], answer
+    stored = _stored(local_server, capture_id, path)
+    assert stored == data, "the committed object is not the retry's bytes"
+    stored_md5 = base64.b64encode(hashlib.md5(stored).digest()).decode()
+    assert stored_md5 == _md5_b64(data), "the object's md5 does not match the retry's bytes"
+
+    # The sweeper must cover where the local backend parks per-attempt temps:
+    # a process killed between prepare_tmp and commit_tmp strands a file there.
+    stranded = local_server.root / ".uploads" / "singles" / "killed-process.uploading"
+    stranded.parent.mkdir(parents=True, exist_ok=True)
+    stranded.write_bytes(b"a process died between prepare_tmp and commit_tmp")
+    stale = time.time() - 8 * 86400
+    os.utime(stranded, (stale, stale))
+    _sweep_volume(local_server.root)
+    assert not stranded.exists(), "the sweeper must age out stranded local per-attempt temps"
+    assert _leftover_tmp(local_server) == [], "no temp file may remain after the sweeper runs"
+
+
 def test_replaying_a_completed_put_succeeds(server: Server) -> None:
     capture_id = _create_capture(server, "tcp-replay-1")
     data = _pattern(64 * 1024, seed=5)

From a76dd7099c67031cfa705bed69afda2e8d986f7a Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 17:55:54 +0000
Subject: [PATCH 09/11] fix(storage): per-attempt upload temporaries for the
 local backend too

LocalFsStorage.prepare_tmp returned the one shared {key}.uploading path, so
when an older PUT attempt disconnected while a retry was still writing, the
disconnect cleanup unlinked the retry's temp and the retry's commit_tmp
failed with a 500 on valid bytes. VolumeStorage already overrode this with
one temp per attempt; the hoisted implementation in LocalFsStorage gives
both backends the same naming under .uploads/singles, which is the pattern
the sweeper already ages out.
---
 backend/server/api/routes/uploads.py          |  4 ++--
 backend/server/api/sweeper.py                 |  8 +++----
 .../server/common/backends/storage_local.py   | 23 ++++++++++++++++---
 .../server/common/backends/storage_volume.py  | 19 ---------------
 4 files changed, 26 insertions(+), 28 deletions(-)

diff --git a/backend/server/api/routes/uploads.py b/backend/server/api/routes/uploads.py
index 09e987c2..e64c9562 100644
--- a/backend/server/api/routes/uploads.py
+++ b/backend/server/api/routes/uploads.py
@@ -149,8 +149,8 @@ async def put_upload(token: str, request: Request):
         except ClientDisconnect:
             # The phone dropped the connection mid-body — it does this when a
             # retried PUT opens while the old attempt still drains. Nothing is
-            # committed and the temp file goes away, so the retry starts clean
-            # instead of leaving `{key}.uploading` litter in the volume.
+            # committed and only this attempt's temp file goes away, so the
+            # retry starts clean on its own per-attempt temp.
             if tmp is not None:
                 deps.storage.discard_tmp(tmp)
             return Response(content=b"BadDigest", status_code=400, media_type="text/plain")
diff --git a/backend/server/api/sweeper.py b/backend/server/api/sweeper.py
index 9e60347d..9342c77b 100644
--- a/backend/server/api/sweeper.py
+++ b/backend/server/api/sweeper.py
@@ -61,10 +61,10 @@ def _sweep_volume(root: Path) -> int:
             if path.is_file() and path.stat().st_mtime < now - 7 * 86400:
                 path.unlink()
                 deleted += 1
-    # VolumeStorage.prepare_tmp parks one temporary per PUT attempt under
-    # .uploads/singles, outside the uploads walk above. Sweep them on the same
-    # age rule, so a process killed between prepare_tmp and commit_tmp strands
-    # nothing in the volume.
+    # Both storage backends park one temporary per PUT attempt under
+    # .uploads/singles — LocalFsStorage owns the naming, VolumeStorage inherits
+    # it — outside the uploads walk above. Sweep them on the same age rule, so
+    # a process killed between prepare_tmp and commit_tmp strands nothing.
     singles = root / ".uploads" / "singles"
     if singles.exists():
         for path in singles.glob("*.uploading"):
diff --git a/backend/server/common/backends/storage_local.py b/backend/server/common/backends/storage_local.py
index a90727b9..b1402692 100644
--- a/backend/server/common/backends/storage_local.py
+++ b/backend/server/common/backends/storage_local.py
@@ -7,6 +7,7 @@
 import json
 import re
 import threading
+import uuid
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
@@ -84,9 +85,25 @@ def delete_prefix(self, prefix: str) -> int:
         return len(keys)
 
     def prepare_tmp(self, key: str) -> Path:
-        tmp = self._path(f"{key}.uploading")
-        tmp.parent.mkdir(parents=True, exist_ok=True)
-        return tmp
+        """One temp file per PUT attempt, not one per key.
+
+        The per-key `{key}.uploading` name collides exactly when the phone
+        retries a dropped PUT on the same signed URL while the first attempt's
+        handler can still be draining its body: the retry's open truncated the
+        file under the first attempt's handle, and the dropped attempt's
+        cleanup unlinked the retry's temp, so the retry's commit_tmp failed
+        with a 500 on valid bytes. A unique name under `.uploads/` — a prefix
+        `list_prefix` already excludes — gives every attempt its own complete
+        bytes; the last commit wins, which is the replay rule. Parking them in
+        `.uploads/singles` also puts them where the sweeper ages stranded
+        files out. VolumeStorage PUTs go through the same handler and inherit
+        this naming, so the sweep covers both backends.
+        """
+
+        self._uploads.mkdir(parents=True, exist_ok=True)
+        singles = self._uploads / "singles"
+        singles.mkdir(parents=True, exist_ok=True)
+        return singles / f"{uuid.uuid4().hex}.uploading"
 
     def discard_tmp(self, tmp: Path) -> None:
         if tmp.is_file():
diff --git a/backend/server/common/backends/storage_volume.py b/backend/server/common/backends/storage_volume.py
index 3f76f4c4..0ef222ce 100644
--- a/backend/server/common/backends/storage_volume.py
+++ b/backend/server/common/backends/storage_volume.py
@@ -7,7 +7,6 @@
 import hmac
 import json
 import time
-import uuid
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -73,24 +72,6 @@ def copy(self, src: str, dst: str, *, if_generation_match: str | None = None):
                 raise
             raise PreconditionFailed(src) from exc
 
-    def prepare_tmp(self, key: str) -> Path:
-        """One temp file per PUT attempt, not one per key.
-
-        The inherited `{key}.uploading` name collides exactly when the phone
-        retries a dropped PUT on the same signed URL while the first attempt's
-        handler can still be draining its body: the retry's open truncated the
-        file under the first attempt's handle, and whichever attempt committed
-        second either moved the other's bytes or crashed on a vanished temp
-        path (a 500 on valid bytes). A unique name under `.uploads/` — a prefix
-        `list_prefix` already excludes — makes every attempt commit its own
-        complete bytes; the last commit wins, which is the replay rule.
-        """
-
-        self._uploads.mkdir(parents=True, exist_ok=True)
-        singles = self._uploads / "singles"
-        singles.mkdir(parents=True, exist_ok=True)
-        return singles / f"{uuid.uuid4().hex}.uploading"
-
     def signed_put_url(self, key: str, content_type: str, md5_b64: str, ttl_s: int) -> SignedUrl:
         exp, stamp = self._expires(ttl_s)
         token = mint_token({"op": "put", "key": key, "md5": md5_b64, "ct": content_type, "exp": exp}, self.hmac_secret)

From a16df4814e7a7ccf27751c1a293304cd063642ea Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 17:57:18 +0000
Subject: [PATCH 10/11] test(api): give the local sweeper test in-flight temp
 files to survive

---
 backend/server/api/tests/test_storage_and_sweep.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/backend/server/api/tests/test_storage_and_sweep.py b/backend/server/api/tests/test_storage_and_sweep.py
index 1e909e97..fd05975c 100644
--- a/backend/server/api/tests/test_storage_and_sweep.py
+++ b/backend/server/api/tests/test_storage_and_sweep.py
@@ -244,6 +244,9 @@ def test_local_sweeper_ages_stale_single_attempt_temps(tmp_path):
     key = "uploads/cap/local.jpg"
     first = storage.prepare_tmp(key)
     second = storage.prepare_tmp(key)
+    # prepare_tmp mints the path; an in-flight attempt's file exists under it.
+    first.write_bytes(b"first attempt in flight")
+    second.write_bytes(b"second attempt in flight")
     assert first.parent == root / ".uploads" / "singles", "a local temp must land where the sweeper sweeps"
     assert first.name.endswith(".uploading") and second.name.endswith(".uploading")
     assert first != second, "two attempts for one key must not share a temp file"

From 04dc12ba5b538923486e7cfbae7c967abaab9a50 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 18:19:52 +0000
Subject: [PATCH 11/11] test(api): the retried local PUT finishes its body
 before reading its answer

---
 backend/server/api/tests/test_uploads_tcp.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/backend/server/api/tests/test_uploads_tcp.py b/backend/server/api/tests/test_uploads_tcp.py
index 88711f76..6e80c2ec 100644
--- a/backend/server/api/tests/test_uploads_tcp.py
+++ b/backend/server/api/tests/test_uploads_tcp.py
@@ -368,6 +368,8 @@ def test_a_dropped_attempt_on_local_storage_leaves_the_retry_clean(local_server:
     time.sleep(0.5)
     first.abort()
 
+    # The retry finishes its body after the first attempt is gone.
+    second.send_bytes(data)
     status, body = second.response()
     assert status == 200, f"the retry must commit cleanly after the first attempt drops, got {status}: {body[:200]!r}"