diff --git a/src/netmover/mover/engine.py b/src/netmover/mover/engine.py index 1213899..85ca126 100644 --- a/src/netmover/mover/engine.py +++ b/src/netmover/mover/engine.py @@ -13,6 +13,7 @@ Reihenfolge pro Lauf: from __future__ import annotations +import dataclasses import errno import os import stat @@ -39,7 +40,7 @@ from netmover.mover.destinations import ( ) from netmover.mover.ledger import FailureInfo, Ledger 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 log = get_logger(__name__) @@ -222,26 +223,32 @@ class Engine: else: 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: self._sleep(rule.stability_seconds) token.check() stable: list[Candidate] = [] for c in pending: - st = current_state(c.path) - if st is None: + before, st = first[c.relpath], current_state(c.path) + if before is None or st is None: continue # inzwischen verschwunden – nichts zu tun - if st != c.state: + if st != before or st == BUSY: result.add( ItemResult( c.relpath, ItemStatus.SKIPPED, - size=st.size, - message="wird noch geschrieben (Größe/Zeit geändert)", + size=max(st.size, 0), + message="wird noch geschrieben (Größe/Zeit geändert)" + if st != BUSY + else "Datei ist gerade in Benutzung", ) ) 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) for c in stable: diff --git a/src/netmover/mover/scan.py b/src/netmover/mover/scan.py index ca57c2c..eb9f882 100644 --- a/src/netmover/mover/scan.py +++ b/src/netmover/mover/scan.py @@ -10,6 +10,7 @@ from __future__ import annotations +import errno import fnmatch import os import re @@ -100,11 +101,31 @@ def scan( 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: + """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: - st = os.stat(path, follow_symlinks=False) + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) except FileNotFoundError: 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): return None return FileState(st.st_size, st.st_mtime_ns) diff --git a/tests/test_engine.py b/tests/test_engine.py index 106d8f9..a2e50f3 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -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 bool(links) is not supported 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()