feat: Mover-Engine mit atomarem Schreiben, Stabilität und Quarantäne
Lokale/Kernel-Mount- und SFTP-Ziele schreiben in .part, fsyncen, prüfen Größe und Quelle und benennen ohne Überschreiben um; erst dann wird die Quelle gelöscht. Stabilitätsfenster, Platzprüfung, Metadaten, Berechtigungen, Kopiermodus mit Gedächtnis, Quarantäne. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Florian Wilde
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
c3a42ce6f5
commit
5633d3057a
@@ -0,0 +1,489 @@
|
||||
"""Ziele: lokales Verzeichnis bzw. Kernel-Mount, oder SFTP.
|
||||
|
||||
Fallstrick 3 – nie eine wachsende Datei unter ihrem Endnamen, nie beide Kopien
|
||||
verlieren. Ablauf für jede Datei:
|
||||
|
||||
1. in ``<ziel>/.netmover-<uuid>.part`` schreiben, ``flush`` + ``fsync``,
|
||||
Größe prüfen,
|
||||
2. Metadaten setzen (mtime, Modus, Eigentümer – Fehler sind nur Hinweise),
|
||||
3. prüfen, dass sich die Quelle seit der Stabilitätsmessung nicht verändert hat,
|
||||
4. atomar auf den Endnamen umbenennen – ohne zu überschreiben (``link`` bzw.
|
||||
SFTP-``rename``), außer „überschreiben“ ist ausdrücklich eingestellt,
|
||||
5. erst danach löscht der Aufrufer die Quelle.
|
||||
|
||||
Scheitert irgendein Schritt vor 4, wird nur die ``.part``-Datei entfernt; die
|
||||
Quelle ist unangetastet. Bricht der Prozess hart ab, bleibt höchstens eine
|
||||
``.part``-Datei zurück, die :meth:`Destination.prepare` später aufräumt.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import errno
|
||||
import os
|
||||
import posixpath
|
||||
import stat
|
||||
import time
|
||||
import uuid
|
||||
from abc import ABC, abstractmethod
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
from netmover.models import CollisionPolicy
|
||||
from netmover.mounting.sftp import SftpSession
|
||||
from netmover.mover.naming import collision_candidates
|
||||
from netmover.timeouts import CancelToken
|
||||
|
||||
CHUNK = 1024 * 1024
|
||||
STALE_PART_SECONDS = 3600
|
||||
PART_PREFIX = ".netmover-"
|
||||
PART_SUFFIX = ".part"
|
||||
|
||||
FsKind = Literal["local", "cifs", "nfs", "sftp"]
|
||||
|
||||
# Fehler, bei denen ``link`` nicht geht und wir auf „prüfen + rename“ ausweichen.
|
||||
_NO_LINK_ERRNOS = frozenset(
|
||||
{errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EMLINK, errno.EXDEV, errno.ENOSYS}
|
||||
)
|
||||
|
||||
|
||||
class TransferError(Exception):
|
||||
"""Übertragung dieser Datei gescheitert (Meldung im Klartext)."""
|
||||
|
||||
|
||||
class SourceChanged(TransferError):
|
||||
"""Quelle hat sich seit der Stabilitätsmessung verändert – später erneut versuchen."""
|
||||
|
||||
|
||||
class CollisionSkip(Exception):
|
||||
"""Ziel existiert und die Regel sagt „überspringen“."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MetaOptions:
|
||||
atime_ns: int
|
||||
mtime_ns: int
|
||||
mode: int | None # Rechte-Bits, None = nicht setzen
|
||||
owner: tuple[int, int] | None # None = nicht setzen
|
||||
|
||||
|
||||
@dataclass
|
||||
class CommitResult:
|
||||
final_rel: str
|
||||
notes: list[str] = field(default_factory=list)
|
||||
source_consumed: bool = False # True: Quelle wurde per rename bereits „verschoben“
|
||||
|
||||
|
||||
Verify = Callable[[], None]
|
||||
|
||||
|
||||
class Destination(ABC):
|
||||
kind: FsKind
|
||||
|
||||
@abstractmethod
|
||||
def describe(self, rel: str) -> str:
|
||||
"""Lesbarer Zielpfad für Protokoll und Probelauf."""
|
||||
|
||||
@abstractmethod
|
||||
def prepare(self) -> None:
|
||||
"""Vor dem Lauf: Zielordner prüfen, alte ``.part``-Reste entfernen."""
|
||||
|
||||
@abstractmethod
|
||||
def exists(self, rel: str) -> bool: ...
|
||||
|
||||
@abstractmethod
|
||||
def free_bytes(self) -> int | None: ...
|
||||
|
||||
def close(self) -> None: # noqa: B027 – absichtlich optional
|
||||
pass
|
||||
|
||||
# -- Primitive der Unterklassen -------------------------------------------------
|
||||
|
||||
@abstractmethod
|
||||
def _ensure_parent(self, rel: str) -> None: ...
|
||||
|
||||
@abstractmethod
|
||||
def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]: ...
|
||||
|
||||
@abstractmethod
|
||||
def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]: ...
|
||||
|
||||
@abstractmethod
|
||||
def _link_noreplace(self, tmp: str, final: str) -> None:
|
||||
"""Atomar umbenennen; ``FileExistsError``, wenn ``final`` existiert."""
|
||||
|
||||
@abstractmethod
|
||||
def _replace(self, tmp: str, final: str) -> list[str]: ...
|
||||
|
||||
@abstractmethod
|
||||
def _remove(self, rel: str) -> None: ...
|
||||
|
||||
def _sync_dir(self, rel_dir: str) -> None: # noqa: B027
|
||||
pass
|
||||
|
||||
# -- gemeinsamer Ablauf ---------------------------------------------------------
|
||||
|
||||
def plan(self, rel: str, policy: CollisionPolicy) -> str | None:
|
||||
"""Voraussichtlicher Endname (Probelauf / Vorabprüfung); None = überspringen."""
|
||||
if policy is CollisionPolicy.OVERWRITE:
|
||||
return rel
|
||||
if policy is CollisionPolicy.SKIP:
|
||||
return None if self.exists(rel) else rel
|
||||
for cand in collision_candidates(rel):
|
||||
if not self.exists(cand):
|
||||
return cand
|
||||
raise TransferError(f"Kein freier Dateiname für {rel}")
|
||||
|
||||
def commit(
|
||||
self,
|
||||
src: Path,
|
||||
size: int,
|
||||
rel: str,
|
||||
policy: CollisionPolicy,
|
||||
meta: MetaOptions,
|
||||
verify: Verify,
|
||||
token: CancelToken,
|
||||
) -> CommitResult:
|
||||
self._ensure_parent(rel)
|
||||
tmp = posixpath.join(
|
||||
posixpath.dirname(rel), f"{PART_PREFIX}{uuid.uuid4().hex}{PART_SUFFIX}"
|
||||
)
|
||||
try:
|
||||
notes = self._upload(src, tmp, size, token)
|
||||
notes += self._apply_meta(tmp, meta)
|
||||
verify()
|
||||
token.check()
|
||||
final, more = self._finalize(tmp, rel, policy)
|
||||
notes += more
|
||||
except BaseException:
|
||||
with contextlib.suppress(Exception):
|
||||
self._remove(tmp)
|
||||
raise
|
||||
self._sync_dir(posixpath.dirname(final))
|
||||
return CommitResult(final, notes)
|
||||
|
||||
def move_same_device(
|
||||
self, src: Path, rel: str, policy: CollisionPolicy, meta: MetaOptions
|
||||
) -> CommitResult | None:
|
||||
"""Schneller Weg ohne Kopie, wenn Quelle und Ziel auf demselben Gerät liegen."""
|
||||
return None
|
||||
|
||||
def _finalize(self, tmp: str, rel: str, policy: CollisionPolicy) -> tuple[str, list[str]]:
|
||||
if policy is CollisionPolicy.OVERWRITE:
|
||||
return rel, self._replace(tmp, rel)
|
||||
candidates = collision_candidates(rel) if policy is CollisionPolicy.RENAME else iter([rel])
|
||||
for cand in candidates:
|
||||
if self.exists(cand):
|
||||
if policy is CollisionPolicy.SKIP:
|
||||
raise CollisionSkip
|
||||
continue
|
||||
try:
|
||||
self._link_noreplace(tmp, cand)
|
||||
except FileExistsError:
|
||||
# Wettlauf: jemand anderes hat den Namen gerade belegt
|
||||
if policy is CollisionPolicy.SKIP:
|
||||
raise CollisionSkip from None
|
||||
continue
|
||||
return cand, []
|
||||
raise TransferError(f"Kein freier Dateiname für {rel}")
|
||||
|
||||
|
||||
class LocalDestination(Destination):
|
||||
"""Lokales Verzeichnis oder Verzeichnis auf einem (geprüften) Kernel-Mount."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
root: Path,
|
||||
kind: Literal["local", "cifs", "nfs"] = "local",
|
||||
*,
|
||||
create_root: bool = False,
|
||||
) -> None:
|
||||
self.root = root
|
||||
self.kind = kind
|
||||
self.create_root = create_root
|
||||
|
||||
def _p(self, rel: str) -> Path:
|
||||
return self.root / rel
|
||||
|
||||
def describe(self, rel: str) -> str:
|
||||
return str(self._p(rel))
|
||||
|
||||
def prepare(self) -> None:
|
||||
if not self.root.is_dir():
|
||||
if not self.create_root:
|
||||
raise TransferError(f"Zielordner {self.root} existiert nicht")
|
||||
self.root.mkdir(mode=0o755, parents=True, exist_ok=True)
|
||||
cutoff = time.time() - STALE_PART_SECONDS
|
||||
for entry in os.scandir(self.root):
|
||||
if entry.name.startswith(PART_PREFIX) and entry.name.endswith(PART_SUFFIX):
|
||||
with contextlib.suppress(OSError):
|
||||
if entry.stat(follow_symlinks=False).st_mtime < cutoff:
|
||||
os.unlink(entry.path)
|
||||
|
||||
def exists(self, rel: str) -> bool:
|
||||
return os.path.lexists(self._p(rel))
|
||||
|
||||
def free_bytes(self) -> int | None:
|
||||
st = os.statvfs(self.root)
|
||||
return st.f_bavail * st.f_frsize
|
||||
|
||||
def _ensure_parent(self, rel: str) -> None:
|
||||
self._p(rel).parent.mkdir(mode=0o755, parents=True, exist_ok=True)
|
||||
|
||||
def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]:
|
||||
written = 0
|
||||
fd = os.open(self._p(tmp), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||||
try:
|
||||
with src.open("rb") as fin:
|
||||
while chunk := fin.read(CHUNK):
|
||||
token.check()
|
||||
view = memoryview(chunk)
|
||||
while view:
|
||||
n = os.write(fd, view)
|
||||
view = view[n:]
|
||||
written += len(chunk)
|
||||
if written != size:
|
||||
raise SourceChanged(
|
||||
"Quelle hat sich während der Übertragung geändert "
|
||||
f"({written} statt {size} Bytes)"
|
||||
)
|
||||
os.fsync(fd)
|
||||
if os.fstat(fd).st_size != size:
|
||||
raise TransferError("Größe der Zieldatei stimmt nach dem Schreiben nicht")
|
||||
finally:
|
||||
os.close(fd)
|
||||
return []
|
||||
|
||||
def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]:
|
||||
notes: list[str] = []
|
||||
path = self._p(rel)
|
||||
if self.kind != "cifs": # auf CIFS bestimmen die Mount-Optionen Eigentümer/Rechte
|
||||
if meta.owner is not None:
|
||||
try:
|
||||
os.chown(path, *meta.owner)
|
||||
except PermissionError:
|
||||
notes.append(
|
||||
f"Eigentümer {meta.owner[0]}:{meta.owner[1]} nicht gesetzt – keine "
|
||||
"Berechtigung (läuft NetMover als root? Root-Squash?)"
|
||||
)
|
||||
except OSError as exc:
|
||||
notes.append(f"Eigentümer nicht gesetzt: {exc.strerror}")
|
||||
if meta.mode is not None:
|
||||
try:
|
||||
os.chmod(path, meta.mode)
|
||||
except OSError as exc:
|
||||
notes.append(f"Modus {meta.mode:04o} nicht gesetzt: {exc.strerror}")
|
||||
try:
|
||||
os.utime(path, ns=(meta.atime_ns, meta.mtime_ns))
|
||||
except OSError as exc:
|
||||
notes.append(f"Änderungszeit nicht übernommen: {exc.strerror}")
|
||||
return notes
|
||||
|
||||
def _link_noreplace(self, tmp: str, final: str) -> None:
|
||||
src, dst = self._p(tmp), self._p(final)
|
||||
try:
|
||||
os.link(src, dst)
|
||||
except FileExistsError:
|
||||
raise
|
||||
except OSError as exc:
|
||||
if exc.errno not in _NO_LINK_ERRNOS:
|
||||
raise
|
||||
# Dateisystem ohne Hardlinks (häufig CIFS): prüfen + rename.
|
||||
# Kleines Wettlauf-Fenster, aber nie eine halbe Datei unter dem Endnamen.
|
||||
if os.path.lexists(dst):
|
||||
raise FileExistsError(errno.EEXIST, "exists", str(dst)) from None
|
||||
os.rename(src, dst)
|
||||
return
|
||||
with contextlib.suppress(OSError):
|
||||
os.unlink(src) # Rest wird sonst später als alte .part-Datei aufgeräumt
|
||||
|
||||
def _replace(self, tmp: str, final: str) -> list[str]:
|
||||
os.replace(self._p(tmp), self._p(final))
|
||||
return []
|
||||
|
||||
def _remove(self, rel: str) -> None:
|
||||
with contextlib.suppress(FileNotFoundError):
|
||||
os.unlink(self._p(rel))
|
||||
|
||||
def _sync_dir(self, rel_dir: str) -> None:
|
||||
with contextlib.suppress(OSError):
|
||||
fd = os.open(self._p(rel_dir) if rel_dir else self.root, os.O_RDONLY | os.O_DIRECTORY)
|
||||
try:
|
||||
os.fsync(fd)
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
def move_same_device(
|
||||
self, src: Path, rel: str, policy: CollisionPolicy, meta: MetaOptions
|
||||
) -> CommitResult | None:
|
||||
try:
|
||||
same = os.stat(src).st_dev == os.stat(self.root).st_dev
|
||||
except OSError:
|
||||
return None
|
||||
if not same:
|
||||
return None
|
||||
self._ensure_parent(rel)
|
||||
consumed = True
|
||||
if policy is CollisionPolicy.OVERWRITE:
|
||||
os.replace(src, self._p(rel))
|
||||
final = rel
|
||||
else:
|
||||
final = ""
|
||||
candidates = (
|
||||
collision_candidates(rel) if policy is CollisionPolicy.RENAME else iter([rel])
|
||||
)
|
||||
for cand in candidates:
|
||||
if self.exists(cand):
|
||||
if policy is CollisionPolicy.SKIP:
|
||||
raise CollisionSkip
|
||||
continue
|
||||
try:
|
||||
os.link(src, self._p(cand))
|
||||
except FileExistsError:
|
||||
if policy is CollisionPolicy.SKIP:
|
||||
raise CollisionSkip from None
|
||||
continue
|
||||
except OSError as exc:
|
||||
if exc.errno not in _NO_LINK_ERRNOS:
|
||||
raise
|
||||
if self.exists(cand):
|
||||
continue
|
||||
os.rename(src, self._p(cand))
|
||||
final = cand
|
||||
break
|
||||
# Ziel ist vollständig da; den alten Namen entfernt der Aufrufer –
|
||||
# mit derselben Fehlerbehandlung wie nach einer Kopie.
|
||||
consumed = False
|
||||
final = cand
|
||||
break
|
||||
if not final:
|
||||
raise TransferError(f"Kein freier Dateiname für {rel}")
|
||||
notes = self._apply_meta(final, meta)
|
||||
self._sync_dir(posixpath.dirname(final))
|
||||
return CommitResult(final, notes, source_consumed=consumed)
|
||||
|
||||
|
||||
class SftpDestination(Destination):
|
||||
kind: FsKind = "sftp"
|
||||
|
||||
def __init__(self, session: SftpSession, base: str, label: str = "") -> None:
|
||||
self.session = session
|
||||
self.client = session.client
|
||||
self.base = base
|
||||
self.label = label
|
||||
|
||||
def _p(self, rel: str) -> str:
|
||||
return posixpath.join(self.base, rel) if rel else self.base
|
||||
|
||||
def describe(self, rel: str) -> str:
|
||||
prefix = f"{self.label}:" if self.label else "sftp:"
|
||||
return prefix + self._p(rel)
|
||||
|
||||
def close(self) -> None:
|
||||
self.session.close()
|
||||
|
||||
def prepare(self) -> None:
|
||||
try:
|
||||
attrs = self.client.stat(self.base)
|
||||
except OSError as exc:
|
||||
raise TransferError(f"SFTP-Zielordner {self.base}: {exc}") from None
|
||||
if attrs.st_mode is None or not stat.S_ISDIR(attrs.st_mode):
|
||||
raise TransferError(f"SFTP-Ziel {self.base} ist kein Verzeichnis")
|
||||
cutoff = time.time() - STALE_PART_SECONDS
|
||||
for a in self.client.listdir_attr(self.base):
|
||||
is_part = a.filename.startswith(PART_PREFIX) and a.filename.endswith(PART_SUFFIX)
|
||||
if is_part and a.st_mtime is not None and a.st_mtime < cutoff:
|
||||
with contextlib.suppress(OSError):
|
||||
self.client.remove(self._p(a.filename))
|
||||
|
||||
def exists(self, rel: str) -> bool:
|
||||
try:
|
||||
self.client.lstat(self._p(rel))
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
return True
|
||||
|
||||
def free_bytes(self) -> int | None:
|
||||
usage = self.session.statvfs(self.base)
|
||||
return usage[1] if usage else None
|
||||
|
||||
def _ensure_parent(self, rel: str) -> None:
|
||||
parent = posixpath.dirname(rel)
|
||||
if not parent:
|
||||
return
|
||||
current = self.base
|
||||
for part in parent.split("/"):
|
||||
current = posixpath.join(current, part)
|
||||
try:
|
||||
self.client.stat(current)
|
||||
except FileNotFoundError:
|
||||
self.client.mkdir(current, mode=0o755)
|
||||
|
||||
def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]:
|
||||
notes: list[str] = []
|
||||
written = 0
|
||||
with self.client.open(self._p(tmp), "wb") as fh, src.open("rb") as fin:
|
||||
fh.set_pipelined(True)
|
||||
while chunk := fin.read(CHUNK // 4):
|
||||
token.check()
|
||||
fh.write(chunk)
|
||||
written += len(chunk)
|
||||
fh.flush()
|
||||
if written != size:
|
||||
raise SourceChanged(
|
||||
"Quelle hat sich während der Übertragung geändert "
|
||||
f"({written} statt {size} Bytes)"
|
||||
)
|
||||
if not self.session.fsync(fh):
|
||||
notes.append("SFTP-Server unterstützt kein fsync")
|
||||
remote = self.client.stat(self._p(tmp)).st_size
|
||||
if remote != size:
|
||||
raise TransferError(f"Zieldatei hat {remote} statt {size} Bytes")
|
||||
return notes
|
||||
|
||||
def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]:
|
||||
notes: list[str] = []
|
||||
path = self._p(rel)
|
||||
if meta.owner is not None:
|
||||
try:
|
||||
self.client.chown(path, *meta.owner)
|
||||
except OSError as exc:
|
||||
notes.append(f"Eigentümer per SFTP nicht gesetzt: {exc}")
|
||||
if meta.mode is not None:
|
||||
try:
|
||||
self.client.chmod(path, meta.mode)
|
||||
except OSError as exc:
|
||||
notes.append(f"Modus per SFTP nicht gesetzt: {exc}")
|
||||
try:
|
||||
self.client.utime(path, (meta.atime_ns / 1e9, meta.mtime_ns / 1e9))
|
||||
except OSError as exc:
|
||||
notes.append(f"Änderungszeit nicht übernommen: {exc}")
|
||||
return notes
|
||||
|
||||
def _link_noreplace(self, tmp: str, final: str) -> None:
|
||||
if self.exists(final):
|
||||
raise FileExistsError(errno.EEXIST, "exists", final)
|
||||
try:
|
||||
# SFTPv3-rename überschreibt nicht (OpenSSH: link + unlink).
|
||||
self.client.rename(self._p(tmp), self._p(final))
|
||||
except OSError:
|
||||
if self.exists(final):
|
||||
raise FileExistsError(errno.EEXIST, "exists", final) from None
|
||||
raise
|
||||
|
||||
def _replace(self, tmp: str, final: str) -> list[str]:
|
||||
try:
|
||||
self.client.posix_rename(self._p(tmp), self._p(final))
|
||||
except OSError:
|
||||
# Server ohne posix-rename: löschen + umbenennen (kurz kein Ziel, nie halb)
|
||||
with contextlib.suppress(FileNotFoundError):
|
||||
self.client.remove(self._p(final))
|
||||
self.client.rename(self._p(tmp), self._p(final))
|
||||
return ["Server kann nicht atomar überschreiben – Ziel kurz nicht vorhanden"]
|
||||
return []
|
||||
|
||||
def _remove(self, rel: str) -> None:
|
||||
with contextlib.suppress(FileNotFoundError):
|
||||
self.client.remove(self._p(rel))
|
||||
@@ -0,0 +1,434 @@
|
||||
"""Ablauf eines Regellaufs – ohne Scheduler, ohne Datenbank-Protokoll.
|
||||
|
||||
Reihenfolge pro Lauf:
|
||||
|
||||
1. Ziel vorbereiten, Quelle scannen (Filter, versteckte/unfertige Dateien raus),
|
||||
2. Kopier-Buchführung: bereits übertragene Dateiversionen still auslassen,
|
||||
3. Größe und Mindestalter prüfen,
|
||||
4. Stabilität: nach dem Stabilitätsfenster erneut messen – nur Dateien mit
|
||||
unveränderter Größe **und** mtime gehen weiter (Fallstrick 1),
|
||||
5. pro Datei: ggf. Quarantäne, Kollisionsregel, Platzprüfung (Größe + 5 %,
|
||||
Fallstrick 14), atomare Übertragung, erst dann Quelle löschen.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import os
|
||||
import stat
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from netmover.logging_setup import event, get_logger
|
||||
from netmover.models import (
|
||||
CollisionPolicy,
|
||||
ItemStatus,
|
||||
PatternType,
|
||||
RuleAction,
|
||||
RunStatus,
|
||||
)
|
||||
from netmover.mover.destinations import (
|
||||
CollisionSkip,
|
||||
Destination,
|
||||
LocalDestination,
|
||||
MetaOptions,
|
||||
SourceChanged,
|
||||
TransferError,
|
||||
)
|
||||
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.timeouts import Cancelled, CancelToken
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
||||
SPACE_RESERVE = 1.05
|
||||
# Fehler, die auf eine gestörte Anbindung hindeuten: Lauf abbrechen statt jede
|
||||
# weitere Datei scheitern zu lassen – der nächste Lauf prüft und hängt neu ein.
|
||||
MOUNT_ERRNOS = frozenset({errno.ESTALE, errno.EIO, errno.ENOTCONN, errno.EHOSTDOWN})
|
||||
|
||||
|
||||
class RunAborted(Exception):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RuleSnapshot:
|
||||
id: int
|
||||
name: str
|
||||
action: RuleAction = RuleAction.MOVE
|
||||
on_collision: CollisionPolicy = CollisionPolicy.RENAME
|
||||
pattern: str = "*"
|
||||
pattern_type: PatternType = PatternType.GLOB
|
||||
recursive: bool = False
|
||||
stability_seconds: int = 5
|
||||
min_age_seconds: int = 0
|
||||
max_size_mb: int | None = None
|
||||
set_owner: tuple[int, int] | None = None
|
||||
set_mode: int | None = None
|
||||
quarantine_after: int = 3
|
||||
dst_root_squash: bool | None = None # nur NFS-Ziele: Ergebnis des Schreibtests
|
||||
|
||||
|
||||
@dataclass
|
||||
class ItemResult:
|
||||
src: str
|
||||
status: ItemStatus
|
||||
dst: str | None = None
|
||||
size: int | None = None
|
||||
message: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class RunResult:
|
||||
status: RunStatus = RunStatus.OK
|
||||
dry_run: bool = False
|
||||
seen: int = 0
|
||||
moved: int = 0
|
||||
skipped: int = 0
|
||||
failed: int = 0
|
||||
message: str | None = None
|
||||
items: list[ItemResult] = field(default_factory=list)
|
||||
|
||||
def add(self, item: ItemResult) -> None:
|
||||
self.items.append(item)
|
||||
if item.status in (ItemStatus.MOVED, ItemStatus.COPIED, ItemStatus.PLANNED):
|
||||
self.moved += 1
|
||||
elif item.status in (ItemStatus.SKIPPED, ItemStatus.QUARANTINED):
|
||||
self.skipped += 1
|
||||
else:
|
||||
self.failed += 1
|
||||
|
||||
|
||||
def _describe_os_error(exc: OSError) -> str:
|
||||
where = f" ({exc.filename})" if exc.filename else ""
|
||||
return f"{exc.strerror or exc}{where}"
|
||||
|
||||
|
||||
def _mb(n: int) -> str:
|
||||
return f"{n / 1024 / 1024:.1f} MB"
|
||||
|
||||
|
||||
class Engine:
|
||||
def __init__(
|
||||
self,
|
||||
ledger: Ledger,
|
||||
*,
|
||||
sleep: Callable[[float], None] = time.sleep,
|
||||
clock: Callable[[], float] = time.time,
|
||||
warn_once: Callable[[str], bool] | None = None,
|
||||
) -> None:
|
||||
self.ledger = ledger
|
||||
self._sleep = sleep
|
||||
self._clock = clock
|
||||
self._warned: set[str] = set()
|
||||
self._warn_once = warn_once or self._default_warn_once
|
||||
|
||||
def _default_warn_once(self, key: str) -> bool:
|
||||
if key in self._warned:
|
||||
return False
|
||||
self._warned.add(key)
|
||||
return True
|
||||
|
||||
# -- Lauf -------------------------------------------------------------------------
|
||||
|
||||
def execute(
|
||||
self,
|
||||
rule: RuleSnapshot,
|
||||
src_root: Path,
|
||||
dest: Destination,
|
||||
token: CancelToken,
|
||||
*,
|
||||
dry_run: bool = False,
|
||||
quarantine_root: Path | None = None,
|
||||
) -> RunResult:
|
||||
result = RunResult(dry_run=dry_run)
|
||||
try:
|
||||
self._execute(rule, src_root, dest, token, result, dry_run, quarantine_root)
|
||||
except Cancelled:
|
||||
result.status = RunStatus.ERROR
|
||||
result.message = "Lauf abgebrochen"
|
||||
except RunAborted as exc:
|
||||
result.status = RunStatus.ERROR
|
||||
result.message = str(exc)
|
||||
except TransferError as exc:
|
||||
result.status = RunStatus.ERROR
|
||||
result.message = str(exc)
|
||||
except OSError as exc:
|
||||
result.status = RunStatus.ERROR
|
||||
result.message = _describe_os_error(exc)
|
||||
else:
|
||||
if result.failed:
|
||||
result.status = RunStatus.PARTIAL
|
||||
return result
|
||||
|
||||
def _execute(
|
||||
self,
|
||||
rule: RuleSnapshot,
|
||||
src_root: Path,
|
||||
dest: Destination,
|
||||
token: CancelToken,
|
||||
result: RunResult,
|
||||
dry_run: bool,
|
||||
quarantine_root: Path | None,
|
||||
) -> None:
|
||||
if not src_root.is_dir():
|
||||
raise RunAborted(f"Quellordner {src_root} existiert nicht")
|
||||
if not dry_run:
|
||||
dest.prepare()
|
||||
token.check()
|
||||
|
||||
exclude: set[Path] = set()
|
||||
if quarantine_root is not None:
|
||||
exclude.add(quarantine_root.resolve())
|
||||
if isinstance(dest, LocalDestination):
|
||||
exclude.add(dest.root.resolve()) # Ziel innerhalb der Quelle nie erneut scannen
|
||||
matcher = Matcher(rule.pattern, rule.pattern_type)
|
||||
candidates = list(scan(src_root, matcher, rule.recursive, frozenset(exclude)))
|
||||
result.seen = len(candidates)
|
||||
token.check()
|
||||
|
||||
# Kopier-Buchführung (Fallstrick 11). Gilt auch für „verschieben“, falls
|
||||
# dort einmal die Quelle nicht gelöscht werden konnte.
|
||||
done = self.ledger.copied_keys(rule.id)
|
||||
if not dry_run:
|
||||
self.ledger.prune_copied(rule.id, {c.relpath for c in candidates})
|
||||
fresh = [c for c in candidates if (c.relpath, c.state.size, c.state.mtime_ns) not in done]
|
||||
unchanged = len(candidates) - len(fresh)
|
||||
|
||||
pending: list[Candidate] = []
|
||||
now = self._clock()
|
||||
for c in fresh:
|
||||
if rule.max_size_mb is not None and c.state.size > rule.max_size_mb * 1024 * 1024:
|
||||
result.add(
|
||||
ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.SKIPPED,
|
||||
size=c.state.size,
|
||||
message=f"größer als {rule.max_size_mb} MB",
|
||||
)
|
||||
)
|
||||
elif rule.min_age_seconds and now - c.state.mtime_ns / 1e9 < rule.min_age_seconds:
|
||||
result.add(
|
||||
ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.SKIPPED,
|
||||
size=c.state.size,
|
||||
message=f"jünger als {rule.min_age_seconds} s",
|
||||
)
|
||||
)
|
||||
else:
|
||||
pending.append(c)
|
||||
|
||||
# Stabilität (Fallstrick 1): zweite Messung nach dem Stabilitätsfenster.
|
||||
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:
|
||||
continue # inzwischen verschwunden – nichts zu tun
|
||||
if st != c.state:
|
||||
result.add(
|
||||
ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.SKIPPED,
|
||||
size=st.size,
|
||||
message="wird noch geschrieben (Größe/Zeit geändert)",
|
||||
)
|
||||
)
|
||||
continue
|
||||
stable.append(c)
|
||||
|
||||
failures = self.ledger.failures(rule.id)
|
||||
for c in stable:
|
||||
token.check()
|
||||
result.add(self._process(rule, c, dest, token, dry_run, quarantine_root, failures))
|
||||
|
||||
if unchanged:
|
||||
result.message = f"{unchanged} bereits übertragen"
|
||||
|
||||
# -- einzelne Datei ------------------------------------------------------------------
|
||||
|
||||
def _meta(
|
||||
self, rule: RuleSnapshot, path: Path, dest: Destination
|
||||
) -> tuple[MetaOptions, list[str]]:
|
||||
st = os.stat(path)
|
||||
notes: list[str] = []
|
||||
owner = rule.set_owner
|
||||
if owner is not None:
|
||||
if dest.kind == "cifs":
|
||||
owner = None
|
||||
if self._warn_once(f"{rule.id}:cifs-chown"):
|
||||
notes.append(
|
||||
"CIFS-Ziel: Eigentümer bestimmen die Mount-Optionen, chown übersprungen"
|
||||
)
|
||||
elif dest.kind == "nfs" and rule.dst_root_squash is not False:
|
||||
owner = None
|
||||
if self._warn_once(f"{rule.id}:nfs-chown"):
|
||||
reason = "Root-Squash aktiv" if rule.dst_root_squash else "Schreibtest fehlt"
|
||||
notes.append(f"NFS-Ziel: chown übersprungen ({reason})")
|
||||
mode = rule.set_mode if rule.set_mode is not None else stat.S_IMODE(st.st_mode)
|
||||
return MetaOptions(st.st_atime_ns, st.st_mtime_ns, mode, owner), notes
|
||||
|
||||
def _process(
|
||||
self,
|
||||
rule: RuleSnapshot,
|
||||
c: Candidate,
|
||||
dest: Destination,
|
||||
token: CancelToken,
|
||||
dry_run: bool,
|
||||
quarantine_root: Path | None,
|
||||
failures: dict[str, FailureInfo],
|
||||
) -> ItemResult:
|
||||
size = c.state.size
|
||||
failure = failures.get(c.relpath)
|
||||
|
||||
# Quarantäne für wiederholt scheiternde Dateien
|
||||
if (
|
||||
quarantine_root is not None
|
||||
and failure is not None
|
||||
and failure.attempts >= rule.quarantine_after
|
||||
and (failure.size, failure.mtime_ns) == (c.state.size, c.state.mtime_ns)
|
||||
):
|
||||
if dry_run:
|
||||
return ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.PLANNED,
|
||||
str(quarantine_root / c.relpath),
|
||||
size,
|
||||
"würde in Quarantäne verschoben",
|
||||
)
|
||||
return self._quarantine(rule, c, quarantine_root, failure.attempts)
|
||||
|
||||
try:
|
||||
planned = dest.plan(c.relpath, rule.on_collision)
|
||||
except OSError as exc:
|
||||
return self._fail(rule, c, exc, dry_run)
|
||||
if planned is None:
|
||||
return ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.SKIPPED,
|
||||
dest.describe(c.relpath),
|
||||
size,
|
||||
"existiert bereits im Ziel",
|
||||
)
|
||||
if dry_run:
|
||||
msg = (
|
||||
"würde überschreiben"
|
||||
if (rule.on_collision is CollisionPolicy.OVERWRITE and dest.exists(planned))
|
||||
else None
|
||||
)
|
||||
verb = "kopieren" if rule.action is RuleAction.COPY else "verschieben"
|
||||
return ItemResult(
|
||||
c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg or f"würde {verb}"
|
||||
)
|
||||
|
||||
try:
|
||||
free = dest.free_bytes()
|
||||
except OSError:
|
||||
free = None
|
||||
if free is not None and free < size * SPACE_RESERVE:
|
||||
return ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.FAILED,
|
||||
dest.describe(planned),
|
||||
size,
|
||||
f"zu wenig Platz am Ziel (frei {_mb(free)}, "
|
||||
f"benötigt {_mb(int(size * SPACE_RESERVE))})",
|
||||
)
|
||||
|
||||
def verify() -> None:
|
||||
if current_state(c.path) != c.state:
|
||||
raise SourceChanged("Quelle hat sich während der Übertragung geändert")
|
||||
|
||||
try:
|
||||
meta, notes = self._meta(rule, c.path, dest)
|
||||
res = None
|
||||
if rule.action is RuleAction.MOVE:
|
||||
res = dest.move_same_device(c.path, c.relpath, rule.on_collision, meta)
|
||||
if res is None:
|
||||
res = dest.commit(c.path, size, c.relpath, rule.on_collision, meta, verify, token)
|
||||
except CollisionSkip:
|
||||
return ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.SKIPPED,
|
||||
dest.describe(c.relpath),
|
||||
size,
|
||||
"existiert bereits im Ziel",
|
||||
)
|
||||
except SourceChanged as exc:
|
||||
return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, str(exc))
|
||||
except (TransferError, OSError) as exc:
|
||||
return self._fail(rule, c, exc, dry_run)
|
||||
|
||||
notes += res.notes
|
||||
target = dest.describe(res.final_rel)
|
||||
if c.relpath in failures:
|
||||
self.ledger.clear_failure(rule.id, c.relpath)
|
||||
|
||||
if rule.action is RuleAction.COPY:
|
||||
self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target)
|
||||
return ItemResult(c.relpath, ItemStatus.COPIED, target, size, "; ".join(notes) or None)
|
||||
|
||||
if not res.source_consumed:
|
||||
try:
|
||||
os.unlink(c.path)
|
||||
except OSError as exc:
|
||||
# Ziel ist vollständig – Buchführung verhindert Duplikate im nächsten Lauf.
|
||||
self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target)
|
||||
return ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.FAILED,
|
||||
target,
|
||||
size,
|
||||
f"Ziel geschrieben, aber Quelle nicht löschbar: {_describe_os_error(exc)} – "
|
||||
"Datei wird nicht erneut übertragen",
|
||||
)
|
||||
return ItemResult(c.relpath, ItemStatus.MOVED, target, size, "; ".join(notes) or None)
|
||||
|
||||
def _fail(self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool) -> ItemResult:
|
||||
if isinstance(exc, OSError) and exc.errno in MOUNT_ERRNOS:
|
||||
raise RunAborted(f"Anbindung gestört: {_describe_os_error(exc)}") from exc
|
||||
msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc)
|
||||
if not dry_run:
|
||||
attempts = self.ledger.record_failure(
|
||||
rule.id, c.relpath, c.state.size, c.state.mtime_ns, msg
|
||||
)
|
||||
msg = f"{msg} (Versuch {attempts})"
|
||||
event(log, "file.failed", level=30, rule=rule.name, file=c.relpath, error=msg)
|
||||
return ItemResult(c.relpath, ItemStatus.FAILED, None, c.state.size, msg)
|
||||
|
||||
def _quarantine(
|
||||
self, rule: RuleSnapshot, c: Candidate, root: Path, attempts: int
|
||||
) -> ItemResult:
|
||||
try:
|
||||
for cand in collision_candidates(c.relpath):
|
||||
target = root / cand
|
||||
if os.path.lexists(target):
|
||||
continue
|
||||
target.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
|
||||
os.rename(c.path, target)
|
||||
break
|
||||
else:
|
||||
raise TransferError("kein freier Name im Quarantäneordner")
|
||||
except (OSError, TransferError) as exc:
|
||||
msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc)
|
||||
return ItemResult(
|
||||
c.relpath, ItemStatus.FAILED, None, c.state.size, f"Quarantäne gescheitert: {msg}"
|
||||
)
|
||||
self.ledger.clear_failure(rule.id, c.relpath)
|
||||
event(log, "file.quarantined", level=30, rule=rule.name, file=c.relpath)
|
||||
return ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.QUARANTINED,
|
||||
str(target),
|
||||
c.state.size,
|
||||
f"nach {attempts} Fehlversuchen in Quarantäne verschoben",
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["Engine", "FileState", "ItemResult", "RuleSnapshot", "RunAborted", "RunResult"]
|
||||
Reference in New Issue
Block a user