fix(cl2k): stop concurrent Drive uploads creating duplicate files

backend/util/cl2k/gdrive_upload.py

@@ -14,10 +14,22 @@
 import shutil
 import subprocess
 import tempfile
+import threading
 import uuid
+from collections import defaultdict
 from shutil import which
 from typing import Any, List, Optional
 
+# Serialises writes per Drive folder: concurrent uploads both list "absent" and
+# both create, and Drive allows duplicate names. Thread lock — single process only.
+_FOLDER_LOCKS_GUARD = threading.Lock()
+_FOLDER_LOCKS: "defaultdict[str, threading.Lock]" = defaultdict(threading.Lock)
+
+
+def _folder_lock(folder_id: str) -> threading.Lock:
+    with _FOLDER_LOCKS_GUARD:
+        return _FOLDER_LOCKS[folder_id]
+
 
 def _rclone_path() -> str:
     env = os.getenv("RCLONE_PATH")
@@ -88,11 +100,18 @@ def _oauth_args(sync_cfg: Any) -> List[str]:
     token = (token or "").strip()
     if "access_token" not in token and "refresh_token" not in token:
         return []
+    client_id = getattr(sync_cfg, "client_id", "") or ""
+    client_secret = getattr(sync_cfg, "client_secret", "") or ""
+    # A leading "-" would reach rclone as an option, not data. Real Google
+    # credentials never start with one.
+    _reject_unsafe(client_id, "gdrive_client_id")
+    _reject_unsafe(client_secret, "gdrive_client_secret")
+    _reject_unsafe(token, "gdrive_token")
     return [
         "--drive-client-id",
-        getattr(sync_cfg, "client_id", "") or "",
+        client_id,
         "--drive-client-secret",
-        getattr(sync_cfg, "client_secret", "") or "",
+        client_secret,
         "--drive-token",
         token,
     ]
@@ -143,6 +162,7 @@ def upload_file(
     non-zero rclone exit so the caller can record the failure.
     """
     _reject_unsafe_id(folder_id, "gdrive_folder_id")
+    _reject_unsafe(local_path, "local_path")
     auth = _upload_auth_args(sync_cfg)
     if not auth:
         raise RuntimeError(
@@ -163,12 +183,72 @@ def upload_file(
         "-v",
         *auth,
     ]
-    result = subprocess.run(cmd, check=False, capture_output=True, text=True)
-    if result.returncode != 0:
-        raise RuntimeError(f"rclone copy failed: {_rclone_error_detail(result.stderr)}")
+    # Serialised per folder: see _FOLDER_LOCKS. rclone replaces a same-named file
+    # correctly on its own — it only duplicates when two runs overlap.
+    with _folder_lock(folder_id):
+        result = subprocess.run(cmd, check=False, capture_output=True, text=True)
+        if result.returncode != 0:
+            raise RuntimeError(
+                f"rclone copy failed: {_rclone_error_detail(result.stderr)}"
+            )
+        _reap_duplicates(os.path.basename(local_path), folder_id, sync_cfg, logger)
     logger.debug(f"uploaded {os.path.basename(local_path)} to drive {folder_id}")
 
 
+# rclone filter metacharacters. CL2K names carry `{tmdb-…}`, and `{}` is
+# alternation — unescaped it matches the wrong thing.
+_FILTER_META = re.compile(r"([\\*?\[\]{}])")
+
+
+def _filter_literal(name: str) -> str:
+    """``name`` as an rclone filter that matches it and nothing else."""
+    return "/" + _FILTER_META.sub(r"\\\1", name)
+
+
+def _reap_duplicates(name: str, folder_id: str, sync_cfg: Any, logger) -> None:
+    """Collapse same-named copies of ``name``, keeping the newest — ours.
+
+    Scoped to ``name``: an upload of one file must never delete another file's
+    duplicates. Only runs when a duplicate of what we just wrote exists, so the
+    normal path issues no destructive command. Never raises — the upload
+    succeeded, and failing to tidy up must not report it as failed.
+    """
+    try:
+        names = list_files(folder_id, sync_cfg, logger, strict=True)
+        if sum(1 for n in names if n == name) < 2:
+            return
+        logger.warning(
+            f"CL2K drive {folder_id}: {name} exists more than once — keeping the newest"
+        )
+        auth = _upload_auth_args(sync_cfg)
+        rclone = _rclone_path()
+        result = subprocess.run(
+            [
+                rclone,
+                "dedupe",
+                "--dedupe-mode",
+                "newest",
+                "posters:",
+                "--drive-root-folder-id",
+                folder_id,
+                "--drive-use-trash=false",
+                # dedupe honours filters from rclone 1.61.
+                "--include",
+                _filter_literal(name),
+                *auth,
+            ],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        if result.returncode != 0:
+            logger.warning(
+                f"CL2K drive dedupe failed: {_rclone_error_detail(result.stderr)}"
+            )
+    except Exception as exc:
+        logger.warning(f"CL2K drive duplicate check failed for {name}: {exc}")
+
+
 def move_file(
     old_name: str,
     new_name: str,
@@ -207,7 +287,10 @@ def move_file(
         "-v",
         *auth,
     ]
-    result = subprocess.run(cmd, check=False, capture_output=True, text=True)
+    # Upload's lock: this is the folder's other writer, and a rename onto a name
+    # an upload is creating duplicates the same way.
+    with _folder_lock(folder_id):
+        result = subprocess.run(cmd, check=False, capture_output=True, text=True)
     if result.returncode != 0:
         raise RuntimeError(
             f"rclone moveto failed: {_rclone_error_detail(result.stderr)}"

tests/test_cl2k_gdrive_upload_race.py

@@ -0,0 +1,255 @@
+"""Concurrent uploads to one Drive folder must not create duplicate names.
+
+rclone decides create-vs-replace by listing the destination first, and Drive
+permits two files sharing a name — so two overlapping uploads both see "absent"
+and both create. Reproduced against the real Drive before this fix: sequential
+uploads left 1 copy, concurrent uploads left 2.
+
+The fake below models exactly that: a listing, then a create-or-replace, with a
+window between them.
+"""
+
+import threading
+import time
+import types
+
+import pytest
+
+from backend.util.cl2k import gdrive_upload as gd
+
+
+class FakeDrive:
+    """A folder where create-vs-replace is decided by a listing, with a real
+    gap between the check and the write — the shape that duplicates."""
+
+    def __init__(self, gap=0.05):
+        self.files = []  # names; a list, not a set — Drive allows duplicates
+        self.gap = gap
+        self._mutex = threading.Lock()
+
+    def rclone_copy(self, name):
+        present = name in self.files  # check
+        time.sleep(self.gap)  # ...window...
+        with self._mutex:
+            if not present:
+                self.files.append(name)  # create (duplicates if two got here)
+
+    def count(self, name):
+        return sum(1 for n in self.files if n == name)
+
+
+def _install(monkeypatch, drive, name="poster.png"):
+    """Point upload_file's subprocess at the fake, keeping the real locking."""
+    monkeypatch.setattr(gd, "_rclone_path", lambda: "/usr/bin/rclone")
+    monkeypatch.setattr(gd, "_ensure_remote", lambda _r: None)
+    monkeypatch.setattr(gd, "_upload_auth_args", lambda _c: ["--drive-token", "x"])
+    monkeypatch.setattr(
+        gd, "list_files", lambda fid, cfg, log, strict=False: list(drive.files)
+    )
+
+    def fake_run(cmd, **kw):
+        if "copy" in cmd:
+            drive.rclone_copy(name)
+        return types.SimpleNamespace(returncode=0, stdout="", stderr="")
+
+    monkeypatch.setattr(gd.subprocess, "run", fake_run)
+
+
+def _logger():
+    noop = lambda *a, **k: None  # noqa: E731
+    return types.SimpleNamespace(debug=noop, info=noop, warning=noop, error=noop)
+
+
+def _upload_twice_concurrently(folder="folder-A", path="/tmp/poster.png"):
+    barrier = threading.Barrier(2)
+
+    def worker():
+        barrier.wait()  # maximise overlap
+        gd.upload_file(path, folder, object(), _logger())
+
+    threads = [threading.Thread(target=worker) for _ in range(2)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+
+def test_the_fake_reproduces_the_duplicate_without_the_lock(monkeypatch):
+    """Guard the guard: if this ever passes, the fake stopped modelling the bug
+    and the test below would prove nothing."""
+    drive = FakeDrive()
+    _install(monkeypatch, drive)
+    monkeypatch.setattr(gd, "_folder_lock", lambda _f: threading.Lock())  # useless lock
+    _upload_twice_concurrently()
+    assert drive.count("poster.png") == 2
+
+
+def test_concurrent_uploads_to_one_folder_leave_a_single_copy(monkeypatch):
+    drive = FakeDrive()
+    _install(monkeypatch, drive)
+    _upload_twice_concurrently()
+    assert drive.count("poster.png") == 1
+
+
+def test_different_folders_still_upload_in_parallel(monkeypatch):
+    """The lock is per folder, so unrelated destinations must not serialise."""
+    assert gd._folder_lock("folder-A") is gd._folder_lock("folder-A")
+    assert gd._folder_lock("folder-A") is not gd._folder_lock("folder-B")
+
+
+def test_a_preexisting_duplicate_is_reaped(monkeypatch):
+    drive = FakeDrive()
+    drive.files = ["poster.png", "poster.png"]  # e.g. written before this fix
+    _install(monkeypatch, drive)
+    deduped = []
+
+    def fake_run(cmd, **kw):
+        if "dedupe" in cmd:
+            deduped.append(cmd)
+            drive.files = ["poster.png"]
+        elif "copy" in cmd:
+            drive.rclone_copy("poster.png")
+        return types.SimpleNamespace(returncode=0, stdout="", stderr="")
+
+    monkeypatch.setattr(gd.subprocess, "run", fake_run)
+    gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger())
+    assert deduped, "an existing duplicate should trigger a dedupe"
+    assert "newest" in deduped[0], "must keep the newest, i.e. the copy just uploaded"
+    assert drive.count("poster.png") == 1
+
+
+def test_no_duplicate_means_no_destructive_command(monkeypatch):
+    """The normal path must never run dedupe — it deletes."""
+    drive = FakeDrive()
+    _install(monkeypatch, drive)
+    seen = []
+
+    def fake_run(cmd, **kw):
+        seen.append(cmd)
+        if "copy" in cmd:
+            drive.rclone_copy("poster.png")
+        return types.SimpleNamespace(returncode=0, stdout="", stderr="")
+
+    monkeypatch.setattr(gd.subprocess, "run", fake_run)
+    gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger())
+    assert not any("dedupe" in c for c in seen)
+
+
+def test_a_rename_serialises_against_an_upload_to_the_same_folder(monkeypatch):
+    """move_file is the folder's other writer (the poster healer renames while
+    the maker uploads), so it has to take the same lock or the convention only
+    covers half the writers."""
+    drive = FakeDrive()
+    _install(monkeypatch, drive)
+    order = []
+
+    def fake_run(cmd, **kw):
+        op = "moveto" if "moveto" in cmd else "copy"
+        order.append("%s-start" % op)
+        time.sleep(0.05)
+        order.append("%s-end" % op)
+        if op == "copy":
+            drive.rclone_copy("poster.png")
+        return types.SimpleNamespace(returncode=0, stdout="", stderr="")
+
+    monkeypatch.setattr(gd.subprocess, "run", fake_run)
+    barrier = threading.Barrier(2)
+
+    def up():
+        barrier.wait()
+        gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger())
+
+    def mv():
+        barrier.wait()
+        gd.move_file("old.png", "poster.png", "folder-A", object(), _logger())
+
+    ts = [threading.Thread(target=up), threading.Thread(target=mv)]
+    for t in ts:
+        t.start()
+    for t in ts:
+        t.join()
+    # Never interleaved: each write finishes before the other starts.
+    assert order[0].endswith("-start") and order[1].endswith("-end")
+    assert order[1].split("-")[0] == order[0].split("-")[0]
+
+
+def test_a_failed_duplicate_check_does_not_fail_the_upload(monkeypatch):
+    drive = FakeDrive()
+    _install(monkeypatch, drive)
+    monkeypatch.setattr(
+        gd, "list_files", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("api down"))
+    )
+    gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger())
+    assert drive.count("poster.png") == 1
+
+
+def test_every_value_reaching_the_rclone_argv_is_validated(monkeypatch):
+    """CodeQL flags this call site as an uncontrolled command line. subprocess is
+    list-form (no shell), so the residual risk is argument injection: a value
+    starting with "-" is read by rclone as an option. folder_id and the SA path
+    were already validated; local_path and the OAuth trio were not."""
+    drive = FakeDrive()
+    _install(monkeypatch, drive)
+    with pytest.raises(ValueError, match="local_path"):
+        gd.upload_file("-X=evil", "folder-A", object(), _logger())
+
+
+def test_an_option_shaped_oauth_value_is_refused(monkeypatch):
+    cfg = types.SimpleNamespace(
+        token='{"access_token": "a"}', client_id="-oops", client_secret=""
+    )
+    with pytest.raises(ValueError, match="gdrive_client_id"):
+        gd._oauth_args(cfg)
+
+
+def test_real_google_credentials_still_pass(monkeypatch):
+    """The validator must not reject legitimate config: ids are numeric-led,
+    secrets are GOCSPX-…, and the token is JSON."""
+    cfg = types.SimpleNamespace(
+        token='{"access_token": "ya29.x", "refresh_token": "1//y"}',
+        client_id="123456789-abc.apps.googleusercontent.com",
+        client_secret="GOCSPX-abcdef123456",
+    )
+    args = gd._oauth_args(cfg)
+    assert "--drive-token" in args and "GOCSPX-abcdef123456" in args
+
+
+def test_the_reap_is_scoped_to_the_uploaded_filename(monkeypatch):
+    """Uploading one file must never collapse another file's duplicates. Proven
+    against the real Drive too: staging duplicates of two names and reaping one
+    left the other at 2."""
+    drive = FakeDrive()
+    drive.files = ["poster.png", "poster.png", "other.png", "other.png"]
+    _install(monkeypatch, drive)
+    seen = []
+
+    def fake_run(cmd, **kw):
+        seen.append(cmd)
+        if "dedupe" in cmd:
+            # Actually APPLY the filter rather than assuming it. A fake that
+            # collapses a hard-coded name passes even when the filter is wrong,
+            # which proves nothing about the thing under test.
+            pattern = cmd[cmd.index("--include") + 1]
+            target = pattern.lstrip("/").replace("\\", "")
+            while drive.files.count(target) > 1:
+                drive.files.remove(target)
+        elif "copy" in cmd:
+            drive.rclone_copy("poster.png")
+        return types.SimpleNamespace(returncode=0, stdout="", stderr="")
+
+    monkeypatch.setattr(gd.subprocess, "run", fake_run)
+    gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger())
+
+    dedupe = [c for c in seen if "dedupe" in c]
+    assert dedupe, "a duplicate of the uploaded name should trigger a dedupe"
+    assert dedupe[0][dedupe[0].index("--include") + 1] == "/poster.png"
+    assert drive.count("poster.png") == 1, "our own duplicate must be collapsed"
+    assert drive.count("other.png") == 2, "another file's duplicates must survive"
+
+
+def test_filter_escapes_the_metacharacters_in_real_filenames(monkeypatch):
+    """CL2K names carry `{tmdb-…}`, and `{}` is alternation in an rclone filter —
+    unescaped, the pattern matches the wrong thing or nothing at all."""
+    got = gd._filter_literal("Cassadaga (2011) {tmdb-92398} - logo.png")
+    assert got == r"/Cassadaga (2011) \{tmdb-92398\} - logo.png"
+    assert gd._filter_literal("a[b]*c?d") == r"/a\[b\]\*c\?d"