fix(hotfolder): повтор задания при существующем назначении не трогает .part (R-17)

LocalHotFolder.put сначала проверяет назначение: если файл во входящем каталоге уже есть, он считается доставленным,
ничего не пишется, а .part снимается только при включённой очистке. Иначе удаление старого .part подавляется (остаток
прерванной записи не роняет задание); если он остался, усечение разрешено только при st_nlink == 1, ссылка на чужой файл
не трогается и исключение поднимается как есть.

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

api/src/gateway/hotfolder.py

@@ -206,17 +206,23 @@ class LocalHotFolder:
     def put(self, name: str, data: bytes) -> None:
         """Атомарная запись без замены: `.<name>.part` в `tmp_dir`, затем жёсткая ссылка в `input_dir/<name>` и
         удаление `.part` как часть успешной публикации (R-8). Назначение уже есть — файл считается доставленным
-        (повтор задания), `.part` остаётся при ошибке линковки, если очистка выключена. Корень не создаётся
+        (повтор задания, R-17): ничего не пишется, `.part` снимается только при включённой очистке. Корень не
-        (нет корня — `FileNotFoundError`)."""
+        создаётся (нет корня — `FileNotFoundError`)."""
         validate_name(name)
         self.tmp_dir.mkdir(exist_ok=True)
         self.input_dir.mkdir(exist_ok=True)
         part = self.tmp_dir / f".{name}.part"
         target = self.input_dir / name
-        # `.part` создаётся заново (№32): оставшийся от прошлого задания мог быть жёсткой ссылкой на уже опубликованный
+        if target.exists():
-        # файл (удаление после публикации подавляется, R-8), запись «поверх» испортила бы файл, который читает ИС
+            # R-17: `.part` от прошлой публикации может быть жёсткой ссылкой на этот файл (удаление после публикации
-        part.unlink(missing_ok=True)
+            # подавляется, R-8, OQ-144) — писать и даже удалять его незачем, файл уже доставлен
-        fd = os.open(part, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+            self._drop_part(part)
+            return
+        # `.part` создаётся заново (№32): остаток от прерванной записи не должен ронять задание, поэтому ошибка его
+        # удаления подавляется; если он всё же остался, `_open_part` не даст записать «поверх» ссылки на чужой файл
+        with contextlib.suppress(OSError):
+            part.unlink(missing_ok=True)
+        fd = self._open_part(part)
         try:
             with os.fdopen(fd, "wb") as handle:
                 handle.write(data)
@@ -245,6 +251,18 @@ class LocalHotFolder:
         with contextlib.suppress(OSError):
             part.unlink(missing_ok=True)
 
+    @staticmethod
+    def _open_part(part: Path) -> int:
+        """Открыть `.part` на запись: новый файл (`O_EXCL`). Если старый не удалился (`FileExistsError`) — усечь его
+        только когда это обычный файл без других ссылок (`st_nlink == 1`); ссылку на опубликованный файл не трогаем,
+        исключение поднимается как есть (R-17)."""
+        try:
+            return os.open(part, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+        except FileExistsError:
+            if os.lstat(part).st_nlink != 1:
+                raise
+            return os.open(part, os.O_WRONLY | os.O_TRUNC)
+
     def _replace_part(self, part: Path, target: Path) -> None:
         try:
             os.replace(part, target)

api/tests/unit/test_local_hotfolder.py

@@ -63,7 +63,8 @@ class TestPutWithoutReplace:
         hf.put("job.pdf", b"first")
         hf.put("job.pdf", b"second")
         assert (hf.input_dir / "job.pdf").read_bytes() == b"first"
-        assert (hf.tmp_dir / ".job.pdf.part").read_bytes() == b"second"  # .part остаётся: удаление только под флагом
+        # R-17: назначение есть — повтор ничего не пишет; прежнего `.part` после успешной публикации нет
+        assert not (hf.tmp_dir / ".job.pdf.part").exists()
 
     def test_existing_destination_part_removed_when_cleanup_enabled(self, tmp_path: Path) -> None:
         hf = LocalHotFolder(tmp_path, cleanup=True)
@@ -322,8 +323,7 @@ class TestPut:
 
         def unlink_with_failure(path: Any) -> None:
             unlink_calls.append(path)
-            # перед записью `.part` создаётся заново (№32) — отказ удаления только после публикации файла (R-8)
+            if str(path).endswith(".part"):
-            if str(path).endswith(".part") and (hf.input_dir / "a.pdf").exists():
                 raise PermissionError("удаление запрещено")
             real_unlink(path)
 
@@ -343,8 +343,6 @@ 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)
@@ -356,17 +354,69 @@ class TestPut:
 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:
+    @pytest.mark.parametrize("cleanup", [False, True])
-        hf = LocalHotFolder(tmp_path)
+    def test_existing_destination_is_not_touched_by_repeat(self, tmp_path: Path, cleanup: bool) -> None:
+        """R-17: назначение есть, `.part` — жёсткая ссылка на него: байты не меняются, put не падает."""
+        hf = LocalHotFolder(tmp_path, cleanup=cleanup)
         hf.put("job.pdf", b"first")
         published = hf.input_dir / "job.pdf"
         stale = hf.tmp_dir / ".job.pdf.part"
-        os.link(published, stale)  # подавленное удаление `.part` после публикации оставило жёсткую ссылку
+        os.link(published, stale)
-        assert published.stat().st_nlink == 2
+        hf.put("job.pdf", b"different data")
-        hf.put("job.pdf", b"different data")  # повтор задания: назначение есть, файл считается доставленным
+        assert published.read_bytes() == b"first"
-        assert published.read_bytes() == b"first"  # опубликованный файл не изменился
+        assert stale.exists() is (not cleanup)  # очистка снимает `.part`, иначе он остаётся как был
-        assert stale.read_bytes() == b"different data"  # `.part` — новый файл, а не запись в ту же ссылку
+        if not cleanup:
-        assert published.stat().st_nlink == 1
+            assert stale.read_bytes() == b"first" and published.stat().st_nlink == 2
+
+    def test_existing_destination_with_undeletable_part_does_not_fail(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        hf = LocalHotFolder(tmp_path)  # очистка выключена: удаление `.part` вообще не пытается
+        hf.put("job.pdf", b"first")
+        os.link(hf.input_dir / "job.pdf", hf.tmp_dir / ".job.pdf.part")
+
+        def deny(self: Path, missing_ok: bool = False) -> None:
+            raise PermissionError("удаление запрещено")
+
+        monkeypatch.setattr(Path, "unlink", deny)
+        hf.put("job.pdf", b"second")
+        assert (hf.input_dir / "job.pdf").read_bytes() == b"first"
+
+    def test_undeletable_stale_part_with_single_link_is_truncated(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Назначения нет, старый `.part` не удаляется, но это обычный файл (nlink == 1): put успешен."""
+        hf = LocalHotFolder(tmp_path)
+        hf.tmp_dir.mkdir()
+        stale = hf.tmp_dir / ".job.pdf.part"
+        stale.write_bytes(b"old and longer content")
+
+        def deny(self: Path, missing_ok: bool = False) -> None:
+            raise PermissionError("удаление запрещено")
+
+        monkeypatch.setattr(Path, "unlink", deny)
+        hf.put("job.pdf", b"new")
+        assert (hf.input_dir / "job.pdf").read_bytes() == b"new"
+
+    def test_undeletable_stale_part_with_other_links_is_not_written(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Назначения нет, старый `.part` не удаляется и имеет nlink == 2 (синтетически): запись не идёт, ошибка не
+        маскируется, второй файл не изменён."""
+        hf = LocalHotFolder(tmp_path)
+        hf.tmp_dir.mkdir()
+        other = tmp_path / "other.bin"
+        other.write_bytes(b"precious")
+        os.link(other, hf.tmp_dir / ".job.pdf.part")
+
+        def deny(self: Path, missing_ok: bool = False) -> None:
+            raise PermissionError("удаление запрещено")
+
+        monkeypatch.setattr(Path, "unlink", deny)
+        with pytest.raises(FileExistsError):
+            hf.put("job.pdf", b"new")
+        assert other.read_bytes() == b"precious"
+        assert not (hf.input_dir / "job.pdf").exists()
 
     def test_stale_part_is_replaced_before_publication(self, tmp_path: Path) -> None:
         hf = LocalHotFolder(tmp_path)
@@ -423,12 +473,15 @@ class TestPartIsCreatedFresh:
         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:
+    def test_no_fallback_when_destination_appeared_during_link(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Назначение появилось между проверкой и линковкой, а `os.link` дал EPERM: `os.replace` не затирает."""
         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:
+            Path(dst).write_bytes(b"first")  # файл успел появиться (параллельный повтор того же задания)
             raise OSError(errno.EPERM, "жёсткие ссылки не поддерживаются")
 
         monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links)