fix(mounts): gescheiterte Start-Wiederherstellung ebenfalls minütlich neu versuchen
Bootet die VM schneller als das NAS (Stromausfall), käme die Freigabe sonst nie zurück. Ein Scheduler-Job prüft fällige Neuversuche jede Minute, auch für Anbindungen ohne Regel. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
ea50eacc9d
commit
1bf0fae2fc
@@ -58,7 +58,9 @@ class NetMover:
|
|||||||
)
|
)
|
||||||
self.runner = RuleRunner(self.session_factory, self.mounts)
|
self.runner = RuleRunner(self.session_factory, self.mounts)
|
||||||
self.runs = RunService(self.session_factory, self.runner, self.notifier)
|
self.runs = RunService(self.session_factory, self.runner, self.notifier)
|
||||||
self.scheduler = RuleScheduler(self.session_factory, self.runs)
|
self.scheduler = RuleScheduler(
|
||||||
|
self.session_factory, self.runs, retry_mounts=self.mounts.retry_pending
|
||||||
|
)
|
||||||
self.started_at: datetime = utcnow()
|
self.started_at: datetime = utcnow()
|
||||||
self.restore_results: list[RestoreResult] = []
|
self.restore_results: list[RestoreResult] = []
|
||||||
|
|
||||||
|
|||||||
@@ -344,6 +344,8 @@ class MountManager:
|
|||||||
self.mount(mount_id)
|
self.mount(mount_id)
|
||||||
results.append(RestoreResult(name, True, "eingehängt"))
|
results.append(RestoreResult(name, True, "eingehängt"))
|
||||||
except (MountError, ValidationError, LookupError) as exc:
|
except (MountError, ValidationError, LookupError) as exc:
|
||||||
|
# z. B. Stromausfall: die VM bootet schneller als das NAS
|
||||||
|
self._retry_at[mount_id] = self._clock() + self.retry_interval
|
||||||
results.append(RestoreResult(name, False, str(exc)))
|
results.append(RestoreResult(name, False, str(exc)))
|
||||||
self._notify("mount.restore_failed", name, str(exc))
|
self._notify("mount.restore_failed", name, str(exc))
|
||||||
for r in results:
|
for r in results:
|
||||||
@@ -597,6 +599,35 @@ class MountManager:
|
|||||||
self._notify("mount.error", m.name, msg)
|
self._notify("mount.error", m.name, msg)
|
||||||
raise MountUnavailable(msg)
|
raise MountUnavailable(msg)
|
||||||
|
|
||||||
|
def retry_pending(self) -> list[str]:
|
||||||
|
"""Fällige Neuversuche ausführen (Scheduler-Job, jede Minute) – auch für
|
||||||
|
Anbindungen, die gerade keine Regel benutzt. Liefert die zurückgeholten Namen."""
|
||||||
|
recovered: list[str] = []
|
||||||
|
for mount_id in list(self._retry_at):
|
||||||
|
if self._retry_at.get(mount_id, float("inf")) > self._clock():
|
||||||
|
continue
|
||||||
|
with self._lock(mount_id):
|
||||||
|
if mount_id not in self._retry_at:
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
m, cred = self._load(mount_id)
|
||||||
|
except LookupError:
|
||||||
|
self._retry_at.pop(mount_id, None)
|
||||||
|
continue
|
||||||
|
if not (m.enabled and m.auto_mount and m.type.is_kernel_mount):
|
||||||
|
self._retry_at.pop(mount_id, None)
|
||||||
|
continue
|
||||||
|
mp = self.mountpoint_for(m.slug)
|
||||||
|
if self.backend.is_mounted(mp):
|
||||||
|
self._retry_at.pop(mount_id, None)
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
self._retry_mount(mount_id, m, cred, mp)
|
||||||
|
recovered.append(m.name)
|
||||||
|
except MountUnavailable:
|
||||||
|
pass
|
||||||
|
return recovered
|
||||||
|
|
||||||
def _retry_mount(self, mount_id: int, m: Mount, cred: CredentialData | None, mp: Path) -> None:
|
def _retry_mount(self, mount_id: int, m: Mount, cred: CredentialData | None, mp: Path) -> None:
|
||||||
"""Ein Versuch, einen von NetMover selbst ausgehängten Mount zurückzuholen –
|
"""Ein Versuch, einen von NetMover selbst ausgehängten Mount zurückzuholen –
|
||||||
höchstens einmal pro ``retry_interval``, damit kein Lauf alle 10 s hängt."""
|
höchstens einmal pro ``retry_interval``, damit kein Lauf alle 10 s hängt."""
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import random
|
import random
|
||||||
|
from collections.abc import Callable
|
||||||
from concurrent.futures import ThreadPoolExecutor
|
from concurrent.futures import ThreadPoolExecutor
|
||||||
from datetime import UTC, datetime, timedelta
|
from datetime import UTC, datetime, timedelta
|
||||||
|
|
||||||
@@ -31,6 +32,7 @@ log = get_logger(__name__)
|
|||||||
MAX_START_OFFSET = 10.0
|
MAX_START_OFFSET = 10.0
|
||||||
HEARTBEAT_SECONDS = 30
|
HEARTBEAT_SECONDS = 30
|
||||||
CLEANUP_SECONDS = 3600
|
CLEANUP_SECONDS = 3600
|
||||||
|
RETRY_SECONDS = 60
|
||||||
|
|
||||||
|
|
||||||
def _job_id(rule_id: int) -> str:
|
def _job_id(rule_id: int) -> str:
|
||||||
@@ -43,11 +45,13 @@ class RuleScheduler:
|
|||||||
session_factory: sessionmaker[Session],
|
session_factory: sessionmaker[Session],
|
||||||
service: RunService,
|
service: RunService,
|
||||||
*,
|
*,
|
||||||
|
retry_mounts: Callable[[], object] | None = None,
|
||||||
rng: random.Random | None = None,
|
rng: random.Random | None = None,
|
||||||
workers: int = 8,
|
workers: int = 8,
|
||||||
) -> None:
|
) -> None:
|
||||||
self.session_factory = session_factory
|
self.session_factory = session_factory
|
||||||
self.service = service
|
self.service = service
|
||||||
|
self._retry_mounts = retry_mounts
|
||||||
self._rng = rng or random.Random() # noqa: S311 – nur Startversatz
|
self._rng = rng or random.Random() # noqa: S311 – nur Startversatz
|
||||||
self._executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="nm-run")
|
self._executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="nm-run")
|
||||||
self.scheduler = AsyncIOScheduler(
|
self.scheduler = AsyncIOScheduler(
|
||||||
@@ -70,6 +74,10 @@ class RuleScheduler:
|
|||||||
IntervalTrigger(seconds=CLEANUP_SECONDS, start_date=utcnow() + timedelta(minutes=1)),
|
IntervalTrigger(seconds=CLEANUP_SECONDS, start_date=utcnow() + timedelta(minutes=1)),
|
||||||
id="cleanup",
|
id="cleanup",
|
||||||
)
|
)
|
||||||
|
if self._retry_mounts is not None:
|
||||||
|
self.scheduler.add_job(
|
||||||
|
self._retry, IntervalTrigger(seconds=RETRY_SECONDS), id="mount-retry"
|
||||||
|
)
|
||||||
self.sync()
|
self.sync()
|
||||||
event(log, "scheduler.started", rules=len(self._intervals))
|
event(log, "scheduler.started", rules=len(self._intervals))
|
||||||
|
|
||||||
@@ -141,6 +149,11 @@ class RuleScheduler:
|
|||||||
self._executor, self.service.execute, rule_id, RunTrigger.MANUAL, dry_run
|
self._executor, self.service.execute, rule_id, RunTrigger.MANUAL, dry_run
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def _retry(self) -> None:
|
||||||
|
if self._retry_mounts is not None:
|
||||||
|
loop = asyncio.get_running_loop()
|
||||||
|
await loop.run_in_executor(self._executor, self._retry_mounts)
|
||||||
|
|
||||||
async def _cleanup(self) -> None:
|
async def _cleanup(self) -> None:
|
||||||
loop = asyncio.get_running_loop()
|
loop = asyncio.get_running_loop()
|
||||||
await loop.run_in_executor(self._executor, cleanup, self.session_factory)
|
await loop.run_in_executor(self._executor, cleanup, self.session_factory)
|
||||||
|
|||||||
@@ -485,3 +485,41 @@ def test_hint_text_is_not_redacted() -> None:
|
|||||||
out = _redact_output(msg, ["falsch"])
|
out = _redact_output(msg, ["falsch"])
|
||||||
assert out.startswith("mount error(13): Permission denied ••••••")
|
assert out.startswith("mount error(13): Permission denied ••••••")
|
||||||
assert out.endswith("→ Benutzername/Passwort falsch oder …")
|
assert out.endswith("→ Benutzername/Passwort falsch oder …")
|
||||||
|
|
||||||
|
|
||||||
|
def test_restore_failure_at_boot_is_retried_by_background_job(
|
||||||
|
retry_manager: MountManager,
|
||||||
|
fake_backend: FakeMountBackend,
|
||||||
|
make_mount: MountFactory,
|
||||||
|
session_factory: sessionmaker[Session],
|
||||||
|
notifications: list[tuple[str, str, str]],
|
||||||
|
clock: Clock,
|
||||||
|
) -> None:
|
||||||
|
"""Stromausfall: die VM bootet schneller als das NAS."""
|
||||||
|
fake_backend.export("nas", "scans")
|
||||||
|
mid = make_mount() # keine Regel benutzt die Anbindung
|
||||||
|
fake_backend.fail_next_mount()
|
||||||
|
assert [r.ok for r in retry_manager.restore_all()] == [False]
|
||||||
|
assert retry_manager.retry_pending() == [] # noch nicht fällig
|
||||||
|
clock.now += 60
|
||||||
|
assert retry_manager.retry_pending() == ["scans"]
|
||||||
|
assert fake_backend.is_mounted(retry_manager.mountpoint_for("scans"))
|
||||||
|
assert _state(session_factory, mid).state is MountState.MOUNTED
|
||||||
|
assert notifications[-1][0] == "mount.recovered"
|
||||||
|
clock.now += 60
|
||||||
|
assert retry_manager.retry_pending() == [] # erledigt
|
||||||
|
|
||||||
|
|
||||||
|
def test_manual_unmount_stops_retrying(
|
||||||
|
retry_manager: MountManager,
|
||||||
|
fake_backend: FakeMountBackend,
|
||||||
|
make_mount: MountFactory,
|
||||||
|
clock: Clock,
|
||||||
|
) -> None:
|
||||||
|
mid = make_mount() # nicht exportiert → Wiederherstellung scheitert
|
||||||
|
retry_manager.restore_all()
|
||||||
|
retry_manager.unmount(mid)
|
||||||
|
fake_backend.export("nas", "scans")
|
||||||
|
clock.now += 120
|
||||||
|
assert retry_manager.retry_pending() == []
|
||||||
|
assert not fake_backend.mounted
|
||||||
|
|||||||
Reference in New Issue
Block a user