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
This commit is contained in:
Andrey Epifancev
parent
c1288d08fb
commit
b956d5746f
@ -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 — запасной путь
|
||||
# через replace с проверкой назначения (окно гонки крошечное, повтор задания с тем же job_id редок)
|
||||
if err.errno in (errno.EPERM, errno.EXDEV, errno.ENOTSUP) and not target.exists():
|
||||
os.replace(part, target)
|
||||
# ФС без жёстких ссылок (некоторые тома Docker/сетевые монтирования): запасной путь через replace с
|
||||
# проверкой назначения (окно гонки крошечное, повтор задания с тем же job_id редок). Только для самой
|
||||
# линковки: ошибка записи выше поднимается как есть
|
||||
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):
|
||||
|
||||
@ -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`."""
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user