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
This commit is contained in:
Andrey Epifancev
parent
b411b7e128
commit
cb8c71fc26
@ -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` остаётся при ошибке линковки, если очистка выключена. Корень не создаётся
|
||||
(нет корня — `FileNotFoundError`)."""
|
||||
(повтор задания, R-17): ничего не пишется, `.part` снимается только при включённой очистке. Корень не
|
||||
создаётся (нет корня — `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): оставшийся от прошлого задания мог быть жёсткой ссылкой на уже опубликованный
|
||||
# файл (удаление после публикации подавляется, R-8), запись «поверх» испортила бы файл, который читает ИС
|
||||
part.unlink(missing_ok=True)
|
||||
fd = os.open(part, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
|
||||
if target.exists():
|
||||
# R-17: `.part` от прошлой публикации может быть жёсткой ссылкой на этот файл (удаление после публикации
|
||||
# подавляется, R-8, OQ-144) — писать и даже удалять его незачем, файл уже доставлен
|
||||
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)
|
||||
|
||||
@ -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") and (hf.input_dir / "a.pdf").exists():
|
||||
if str(path).endswith(".part"):
|
||||
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:
|
||||
hf = LocalHotFolder(tmp_path)
|
||||
@pytest.mark.parametrize("cleanup", [False, True])
|
||||
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` после публикации оставило жёсткую ссылку
|
||||
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
|
||||
os.link(published, stale)
|
||||
hf.put("job.pdf", b"different data")
|
||||
assert published.read_bytes() == b"first"
|
||||
assert stale.exists() is (not cleanup) # очистка снимает `.part`, иначе он остаётся как был
|
||||
if not cleanup:
|
||||
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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user