fix(hotfolder): .part создаётся заново (O_EXCL), fallback os.replace только вокруг os.link (№32)

Старый .part, оставшийся жёсткой ссылкой на опубликованный файл, больше не перезаписывается на месте; ошибка записи
поднимается как есть, а не как FileNotFoundError от os.replace; EOPNOTSUPP добавлен к списку ФС без жёстких ссылок.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2

api/src/gateway/hotfolder.py

@@ -29,6 +29,8 @@ _ILLEGAL_CHARS = frozenset(':<>"|?*') | frozenset(chr(code) for code in range(0x
 _FORBIDDEN_CHARS = _ILLEGAL_CHARS | frozenset(_PATH_SEPARATORS)
 # Расширение входящего файла — суффикс после последней точки, в нижнем регистре, 1–16 символов ASCII [0-9a-z]
 _EXTENSION_RE = re.compile(r"[0-9a-z]{1,16}")
+# errno `os.link`, означающие «ФС не умеет жёсткие ссылки» (запасной путь записи — `os.replace`)
+_NO_HARD_LINKS = (errno.EPERM, errno.EXDEV, errno.ENOTSUP, errno.EOPNOTSUPP)
 
 
 def validate_name(name: str) -> None:
@@ -211,17 +213,27 @@ class LocalHotFolder:
         self.input_dir.mkdir(exist_ok=True)
         part = self.tmp_dir / f".{name}.part"
         target = self.input_dir / name
+        # `.part` создаётся заново (№32): оставшийся от прошлого задания мог быть жёсткой ссылкой на уже опубликованный
+        # файл (удаление после публикации подавляется, R-8), запись «поверх» испортила бы файл, который читает ИС
+        part.unlink(missing_ok=True)
+        fd = os.open(part, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+        try:
+            with os.fdopen(fd, "wb") as handle:
+                handle.write(data)
+        except BaseException:
+            self._drop_part(part)
+            raise
         try:
-            part.write_bytes(data)
             os.link(part, target)  # без замены: существующее назначение даёт FileExistsError
         except FileExistsError:
             self._drop_part(part)
             return
         except OSError as err:
-            # ФС без жёстких ссылок (некоторые тома Docker/сетевые монтирования): EPERM/EXDEV/ENOTSUP — запасной путь
+            # ФС без жёстких ссылок (некоторые тома Docker/сетевые монтирования): запасной путь через replace с
-            # через replace с проверкой назначения (окно гонки крошечное, повтор задания с тем же job_id редок)
+            # проверкой назначения (окно гонки крошечное, повтор задания с тем же job_id редок). Только для самой
-            if err.errno in (errno.EPERM, errno.EXDEV, errno.ENOTSUP) and not target.exists():
+            # линковки: ошибка записи выше поднимается как есть
-                os.replace(part, target)
+            if err.errno in _NO_HARD_LINKS and not target.exists():
+                self._replace_part(part, target)
                 return
             self._drop_part(part)
             raise
@@ -233,6 +245,13 @@ class LocalHotFolder:
         with contextlib.suppress(OSError):
             part.unlink(missing_ok=True)
 
+    def _replace_part(self, part: Path, target: Path) -> None:
+        try:
+            os.replace(part, target)
+        except BaseException:
+            self._drop_part(part)
+            raise
+
     def _drop_part(self, part: Path) -> None:
         if self.cleanup_enabled:
             with contextlib.suppress(OSError):

api/tests/unit/test_local_hotfolder.py

@@ -4,6 +4,7 @@
 с готовностью по стабилизации (settle, часы подменены), проверка здоровья, валидация имён.
 """
 
+import errno
 import os
 import uuid
 from pathlib import Path
@@ -321,7 +322,8 @@ class TestPut:
 
         def unlink_with_failure(path: Any) -> None:
             unlink_calls.append(path)
-            if str(path).endswith(".part"):
+            # перед записью `.part` создаётся заново (№32) — отказ удаления только после публикации файла (R-8)
+            if str(path).endswith(".part") and (hf.input_dir / "a.pdf").exists():
                 raise PermissionError("удаление запрещено")
             real_unlink(path)
 
@@ -341,6 +343,8 @@ class TestPut:
             raise OSError("диск недоступен")
 
         def unlink_fails(self: Path, missing_ok: bool = False) -> None:
+            if missing_ok and not self.exists():  # как настоящий unlink: нет файла — нечего запрещать (№32)
+                return
             raise PermissionError("удаление .part запрещено")
 
         monkeypatch.setattr("src.gateway.hotfolder.os.link", boom)
@@ -349,6 +353,107 @@ class TestPut:
             hf.put("a.pdf", b"data")
 
 
+class TestPartIsCreatedFresh:
+    """Решение №32: `.part` создаётся заново (удаление старого + `O_EXCL`), fallback — только вокруг `os.link`."""
+
+    def test_stale_part_hard_linked_to_published_file_is_not_truncated(self, tmp_path: Path) -> None:
+        hf = LocalHotFolder(tmp_path)
+        hf.put("job.pdf", b"first")
+        published = hf.input_dir / "job.pdf"
+        stale = hf.tmp_dir / ".job.pdf.part"
+        os.link(published, stale)  # подавленное удаление `.part` после публикации оставило жёсткую ссылку
+        assert published.stat().st_nlink == 2
+        hf.put("job.pdf", b"different data")  # повтор задания: назначение есть, файл считается доставленным
+        assert published.read_bytes() == b"first"  # опубликованный файл не изменился
+        assert stale.read_bytes() == b"different data"  # `.part` — новый файл, а не запись в ту же ссылку
+        assert published.stat().st_nlink == 1
+
+    def test_stale_part_is_replaced_before_publication(self, tmp_path: Path) -> None:
+        hf = LocalHotFolder(tmp_path)
+        hf.tmp_dir.mkdir()
+        (hf.tmp_dir / ".job.pdf.part").write_bytes(b"old and longer content")
+        hf.put("job.pdf", b"new")
+        assert (hf.input_dir / "job.pdf").read_bytes() == b"new"
+        assert not (hf.tmp_dir / ".job.pdf.part").exists()
+
+    def test_part_is_opened_exclusively(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        hf = LocalHotFolder(tmp_path)
+        flags: list[int] = []
+        real_open = os.open
+
+        def spy(path: Any, flag: int, mode: int = 0o777, **kwargs: Any) -> int:
+            flags.append(flag)
+            return real_open(path, flag, mode, **kwargs)
+
+        monkeypatch.setattr("src.gateway.hotfolder.os.open", spy)
+        hf.put("job.pdf", b"x")
+        assert flags and flags[0] & os.O_EXCL and flags[0] & os.O_CREAT
+
+    def test_write_error_propagates_as_itself(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        """EPERM записи не превращается в `FileNotFoundError` от `os.replace` несуществующего `.part`."""
+        hf = LocalHotFolder(tmp_path, cleanup=True)
+
+        class FailingFile:
+            def __enter__(self) -> "FailingFile":
+                return self
+
+            def __exit__(self, *exc: object) -> None:
+                return None
+
+            def write(self, data: bytes) -> int:
+                raise OSError(errno.EPERM, "запись запрещена")
+
+        monkeypatch.setattr("src.gateway.hotfolder.os.fdopen", lambda fd, mode: (os.close(fd), FailingFile())[1])
+        with pytest.raises(OSError, match="запись запрещена") as caught:
+            hf.put("job.pdf", b"data")
+        assert caught.value.errno == errno.EPERM and not isinstance(caught.value, FileNotFoundError)
+        assert list(hf.input_dir.iterdir()) == [] and list(hf.tmp_dir.iterdir()) == []
+
+    @pytest.mark.parametrize("code", [errno.EPERM, errno.EXDEV, errno.ENOTSUP, errno.EOPNOTSUPP])
+    def test_fallback_to_replace_when_link_unsupported(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, code: int
+    ) -> None:
+        hf = LocalHotFolder(tmp_path)
+
+        def no_links(src: Any, dst: Any) -> None:
+            raise OSError(code, "жёсткие ссылки не поддерживаются")
+
+        monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links)
+        hf.put("job.pdf", b"data")
+        assert (hf.input_dir / "job.pdf").read_bytes() == b"data"
+        assert list(hf.tmp_dir.iterdir()) == []  # `.part` стал назначением
+
+    def test_no_fallback_when_destination_exists(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        hf = LocalHotFolder(tmp_path, cleanup=True)
+        hf.input_dir.mkdir()
+        (hf.input_dir / "job.pdf").write_bytes(b"first")
+
+        def no_links(src: Any, dst: Any) -> None:
+            raise OSError(errno.EPERM, "жёсткие ссылки не поддерживаются")
+
+        monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links)
+        with pytest.raises(OSError, match="жёсткие ссылки"):
+            hf.put("job.pdf", b"second")
+        assert (hf.input_dir / "job.pdf").read_bytes() == b"first"
+
+    def test_replace_failure_after_link_fallback_propagates(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        hf = LocalHotFolder(tmp_path, cleanup=True)
+
+        def no_links(src: Any, dst: Any) -> None:
+            raise OSError(errno.EPERM, "жёсткие ссылки не поддерживаются")
+
+        def broken_replace(src: Any, dst: Any) -> None:
+            raise OSError(errno.EIO, "замена не удалась")
+
+        monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links)
+        monkeypatch.setattr("src.gateway.hotfolder.os.replace", broken_replace)
+        with pytest.raises(OSError, match="замена не удалась"):
+            hf.put("job.pdf", b"data")
+        assert list(hf.tmp_dir.iterdir()) == []  # при очистке `.part` снят
+
+
 class TestPoll:
     """`poll` — все `*.json` каталога `<output>/<job_id>/` по имени, когда набор настоялся `settle_seconds`."""