fix(mover): Stabilität mit frischen Attributen (open + fstat)
CIFS lieferte per stat() sekundenlang eine veraltete Größe einer noch wachsenden Datei; erst open() erzwingt frische Attribute. Gesperrte Dateien gelten als in Benutzung. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
48a52c5290
commit
8276ce2afb
@@ -13,6 +13,7 @@ Reihenfolge pro Lauf:
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import dataclasses
|
||||||
import errno
|
import errno
|
||||||
import os
|
import os
|
||||||
import stat
|
import stat
|
||||||
@@ -39,7 +40,7 @@ from netmover.mover.destinations import (
|
|||||||
)
|
)
|
||||||
from netmover.mover.ledger import FailureInfo, Ledger
|
from netmover.mover.ledger import FailureInfo, Ledger
|
||||||
from netmover.mover.naming import collision_candidates
|
from netmover.mover.naming import collision_candidates
|
||||||
from netmover.mover.scan import Candidate, FileState, Matcher, current_state, scan
|
from netmover.mover.scan import BUSY, Candidate, FileState, Matcher, current_state, scan
|
||||||
from netmover.timeouts import Cancelled, CancelToken
|
from netmover.timeouts import Cancelled, CancelToken
|
||||||
|
|
||||||
log = get_logger(__name__)
|
log = get_logger(__name__)
|
||||||
@@ -222,26 +223,32 @@ class Engine:
|
|||||||
else:
|
else:
|
||||||
pending.append(c)
|
pending.append(c)
|
||||||
|
|
||||||
# Stabilität (Fallstrick 1): zweite Messung nach dem Stabilitätsfenster.
|
# Stabilität (Fallstrick 1): zwei frische Messungen (öffnen + fstat, nicht
|
||||||
|
# der Verzeichnis-Cache) im Abstand des Stabilitätsfensters.
|
||||||
|
first: dict[str, FileState | None] = {c.relpath: current_state(c.path) for c in pending}
|
||||||
if pending and rule.stability_seconds > 0:
|
if pending and rule.stability_seconds > 0:
|
||||||
self._sleep(rule.stability_seconds)
|
self._sleep(rule.stability_seconds)
|
||||||
token.check()
|
token.check()
|
||||||
stable: list[Candidate] = []
|
stable: list[Candidate] = []
|
||||||
for c in pending:
|
for c in pending:
|
||||||
st = current_state(c.path)
|
before, st = first[c.relpath], current_state(c.path)
|
||||||
if st is None:
|
if before is None or st is None:
|
||||||
continue # inzwischen verschwunden – nichts zu tun
|
continue # inzwischen verschwunden – nichts zu tun
|
||||||
if st != c.state:
|
if st != before or st == BUSY:
|
||||||
result.add(
|
result.add(
|
||||||
ItemResult(
|
ItemResult(
|
||||||
c.relpath,
|
c.relpath,
|
||||||
ItemStatus.SKIPPED,
|
ItemStatus.SKIPPED,
|
||||||
size=st.size,
|
size=max(st.size, 0),
|
||||||
message="wird noch geschrieben (Größe/Zeit geändert)",
|
message="wird noch geschrieben (Größe/Zeit geändert)"
|
||||||
|
if st != BUSY
|
||||||
|
else "Datei ist gerade in Benutzung",
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
continue
|
continue
|
||||||
stable.append(c)
|
# ab hier gilt der frisch gemessene Zustand (für Prüfung vor dem Umbenennen
|
||||||
|
# und für die Kopier-Buchführung)
|
||||||
|
stable.append(dataclasses.replace(c, state=st))
|
||||||
|
|
||||||
failures = self.ledger.failures(rule.id)
|
failures = self.ledger.failures(rule.id)
|
||||||
for c in stable:
|
for c in stable:
|
||||||
|
|||||||
@@ -10,6 +10,7 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import errno
|
||||||
import fnmatch
|
import fnmatch
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
@@ -100,11 +101,31 @@ def scan(
|
|||||||
yield from walk(root, "")
|
yield from walk(root, "")
|
||||||
|
|
||||||
|
|
||||||
|
# Datei ist geöffnet, aber gerade nicht lesbar (z. B. exklusiv vom Schreiber
|
||||||
|
# gesperrt): gilt als „noch nicht fertig“ – gleicht nie einem echten Zustand.
|
||||||
|
BUSY = FileState(-1, -1)
|
||||||
|
|
||||||
|
|
||||||
def current_state(path: Path) -> FileState | None:
|
def current_state(path: Path) -> FileState | None:
|
||||||
|
"""Aktuelle Größe und mtime – **frisch vom Server**.
|
||||||
|
|
||||||
|
``stat()`` beantworten CIFS- und NFS-Clients aus ihrem Attribut-Cache; im
|
||||||
|
Praxistest lieferte CIFS so sekundenlang eine veraltete Größe, während die
|
||||||
|
Datei noch wuchs. Erst ``open()`` erzwingt frische Attribute (NFS:
|
||||||
|
Close-to-open-Konsistenz, SMB: Antwort auf CREATE). Darum öffnen + ``fstat``.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
st = os.stat(path, follow_symlinks=False)
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
return None
|
return None
|
||||||
|
except OSError as exc:
|
||||||
|
if exc.errno in (errno.ELOOP, errno.ENOTDIR):
|
||||||
|
return None # Symlink oder kein Pfad mehr
|
||||||
|
return BUSY
|
||||||
|
try:
|
||||||
|
st = os.fstat(fd)
|
||||||
|
finally:
|
||||||
|
os.close(fd)
|
||||||
if not stat.S_ISREG(st.st_mode):
|
if not stat.S_ISREG(st.st_mode):
|
||||||
return None
|
return None
|
||||||
return FileState(st.st_size, st.st_mtime_ns)
|
return FileState(st.st_size, st.st_mtime_ns)
|
||||||
|
|||||||
@@ -730,3 +730,52 @@ def test_final_step_is_a_real_rename_when_possible(
|
|||||||
assert (dst / "a.pdf").read_text() == "alt" and (dst / "a-1.pdf").read_text() == "neu"
|
assert (dst / "a.pdf").read_text() == "alt" and (dst / "a-1.pdf").read_text() == "neu"
|
||||||
assert bool(links) is not supported
|
assert bool(links) is not supported
|
||||||
assert [p.name for p in dst.iterdir() if p.name.endswith(".part")] == []
|
assert [p.name for p in dst.iterdir() if p.name.endswith(".part")] == []
|
||||||
|
|
||||||
|
|
||||||
|
# -- Praxisbefund: veralteter Attribut-Cache (CIFS/NFS) -------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_stability_uses_fresh_attributes_not_stat_cache(
|
||||||
|
engine: Engine,
|
||||||
|
rule_id: int,
|
||||||
|
src: Path,
|
||||||
|
dst: Path,
|
||||||
|
hooks: list[Callable[[], None]],
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""Auf dem Testhost lieferte CIFS per stat() sekundenlang eine veraltete Größe,
|
||||||
|
während die Datei wuchs. Simuliert: os.stat liefert für die Datei immer den
|
||||||
|
alten Stand (Cache), nur öffnen + fstat sieht die Wahrheit."""
|
||||||
|
growing = write(src / "langsam.pdf", "a" * 10)
|
||||||
|
stale = os.stat(growing)
|
||||||
|
real_stat = os.stat
|
||||||
|
|
||||||
|
def cached_stat(path: object, *a: object, **kw: object) -> os.stat_result:
|
||||||
|
if str(path) == str(growing):
|
||||||
|
return stale
|
||||||
|
return real_stat(path, *a, **kw) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
monkeypatch.setattr(os, "stat", cached_stat)
|
||||||
|
hooks.append(lambda: growing.write_text("a" * 20))
|
||||||
|
result = run(engine, snap(rule_id), src, dst)
|
||||||
|
assert statuses(result) == {"langsam.pdf": ItemStatus.SKIPPED}
|
||||||
|
assert "wird noch geschrieben" in (result.items[0].message or "")
|
||||||
|
assert growing.exists() and not (dst / "langsam.pdf").exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_locked_file_counts_as_in_use(
|
||||||
|
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||||||
|
) -> None:
|
||||||
|
f = write(src / "gesperrt.pdf")
|
||||||
|
real_open = os.open
|
||||||
|
|
||||||
|
def locked(path: object, flags: int, *a: object, **kw: object) -> int:
|
||||||
|
if str(path) == str(f):
|
||||||
|
raise PermissionError(errno.EACCES, "Permission denied", str(path))
|
||||||
|
return real_open(path, flags, *a, **kw) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
monkeypatch.setattr("netmover.mover.scan.os.open", locked)
|
||||||
|
result = run(engine, snap(rule_id), src, dst)
|
||||||
|
assert statuses(result) == {"gesperrt.pdf": ItemStatus.SKIPPED}
|
||||||
|
assert result.items[0].message == "Datei ist gerade in Benutzung"
|
||||||
|
assert f.exists()
|
||||||
|
|||||||
Reference in New Issue
Block a user