fix(file-sync): sweep stale sync-back dirs too and tighten the stale window to 6 h
The salvaged commit only swept `hermes-sync-back-*.tar`. The extraction staging dir (`tempfile.TemporaryDirectory(prefix="hermes-sync-back-")`) is leaked by the same hard kill, so the sweep now reclaims both shapes and returns the count. The window drops from 24 h to 6 h (the reporter's value): archives appear every few minutes on a busy gateway and a live transfer is never hours old. Tests trimmed to two invariants: the sweep touches only stale prefixed entries, and a real sync_back reclaims a leaked archive while writing its own tar under the identifiable prefix.
This commit is contained in:
@@ -5,6 +5,8 @@ import logging
|
|||||||
import os
|
import os
|
||||||
import signal
|
import signal
|
||||||
import tarfile
|
import tarfile
|
||||||
|
import tempfile
|
||||||
|
import time
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import MagicMock, patch
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
@@ -14,11 +16,12 @@ fcntl = pytest.importorskip("fcntl")
|
|||||||
|
|
||||||
from tools.environments.file_sync import (
|
from tools.environments.file_sync import (
|
||||||
FileSyncManager,
|
FileSyncManager,
|
||||||
_cleanup_stale_sync_back_tars,
|
_cleanup_stale_sync_back_temp,
|
||||||
_sha256_file,
|
_sha256_file,
|
||||||
_SYNC_BACK_BACKOFF,
|
_SYNC_BACK_BACKOFF,
|
||||||
_SYNC_BACK_MAX_RETRIES,
|
_SYNC_BACK_MAX_RETRIES,
|
||||||
_SYNC_BACK_STALE_TAR_SECONDS,
|
_SYNC_BACK_STALE_SECONDS,
|
||||||
|
_SYNC_BACK_TEMP_PREFIX,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -91,47 +94,52 @@ def _make_manager(
|
|||||||
return mgr
|
return mgr
|
||||||
|
|
||||||
|
|
||||||
class TestStaleSyncBackTarCleanup:
|
class TestStaleSyncBackTempCleanup:
|
||||||
"""Hard-killed sync-back archives are cleaned up conservatively."""
|
"""Sync-back temp entries leaked by a hard kill are reclaimed by the next sync-back (#110812)."""
|
||||||
|
|
||||||
def test_removes_only_stale_hermes_sync_back_tars(self, tmp_path, monkeypatch):
|
def test_removes_only_stale_prefixed_entries(self, tmp_path, monkeypatch):
|
||||||
stale = tmp_path / "hermes-sync-back-stale.tar"
|
stale_tar = tmp_path / "hermes-sync-back-stale.tar"
|
||||||
|
stale_dir = tmp_path / "hermes-sync-back-stale-staging"
|
||||||
recent = tmp_path / "hermes-sync-back-recent.tar"
|
recent = tmp_path / "hermes-sync-back-recent.tar"
|
||||||
unrelated = tmp_path / "other-process.tar"
|
unrelated = tmp_path / "other-process.tar"
|
||||||
for path in (stale, recent, unrelated):
|
for path in (stale_tar, recent, unrelated):
|
||||||
path.write_bytes(b"tar")
|
path.write_bytes(b"tar")
|
||||||
|
stale_dir.mkdir()
|
||||||
|
(stale_dir / "root").mkdir()
|
||||||
now = 10_000.0
|
now = 10_000.0
|
||||||
os.utime(stale, (now - _SYNC_BACK_STALE_TAR_SECONDS - 1,) * 2)
|
for path in (stale_tar, stale_dir):
|
||||||
os.utime(recent, (now - _SYNC_BACK_STALE_TAR_SECONDS + 1,) * 2)
|
os.utime(path, (now - _SYNC_BACK_STALE_SECONDS - 1,) * 2)
|
||||||
|
os.utime(recent, (now - _SYNC_BACK_STALE_SECONDS + 1,) * 2)
|
||||||
monkeypatch.setattr("tools.environments.file_sync.time.time", lambda: now)
|
monkeypatch.setattr("tools.environments.file_sync.time.time", lambda: now)
|
||||||
|
|
||||||
_cleanup_stale_sync_back_tars(tmp_path)
|
assert _cleanup_stale_sync_back_temp(tmp_path) == 2
|
||||||
|
|
||||||
assert not stale.exists()
|
assert not stale_tar.exists()
|
||||||
|
assert not stale_dir.exists()
|
||||||
assert recent.exists()
|
assert recent.exists()
|
||||||
assert unrelated.exists()
|
assert unrelated.exists()
|
||||||
|
|
||||||
def test_preserves_active_tar_when_stat_or_unlink_fails(self, tmp_path, monkeypatch):
|
def test_sync_back_sweeps_leaked_entry_and_uses_identifiable_tar(self, tmp_path, monkeypatch):
|
||||||
active = tmp_path / "hermes-sync-back-active.tar"
|
tmp_root = tmp_path / "tmproot"
|
||||||
stale = tmp_path / "hermes-sync-back-stale.tar"
|
tmp_root.mkdir()
|
||||||
active.write_bytes(b"tar")
|
monkeypatch.setattr(tempfile, "tempdir", str(tmp_root))
|
||||||
stale.write_bytes(b"tar")
|
leaked = tmp_root / "hermes-sync-back-leaked.tar"
|
||||||
now = 10_000.0
|
leaked.write_bytes(b"x" * 1024)
|
||||||
os.utime(stale, (now - _SYNC_BACK_STALE_TAR_SECONDS - 1,) * 2)
|
old = time.time() - _SYNC_BACK_STALE_SECONDS - 60
|
||||||
monkeypatch.setattr("tools.environments.file_sync.time.time", lambda: now)
|
os.utime(leaked, (old, old))
|
||||||
original_unlink = Path.unlink
|
|
||||||
|
|
||||||
def fail_for_stale(path, *args, **kwargs):
|
seen = {}
|
||||||
if path == stale:
|
|
||||||
raise OSError("busy")
|
|
||||||
return original_unlink(path, *args, **kwargs)
|
|
||||||
|
|
||||||
monkeypatch.setattr(Path, "unlink", fail_for_stale)
|
def download(dest: Path):
|
||||||
|
seen["tar"] = dest
|
||||||
|
_make_tar({"root/.hermes/x.txt": b"hi"}, dest)
|
||||||
|
|
||||||
_cleanup_stale_sync_back_tars(tmp_path)
|
mgr = _make_manager(tmp_path, bulk_download_fn=download)
|
||||||
|
mgr.sync_back()
|
||||||
|
|
||||||
assert active.exists()
|
assert seen["tar"].name.startswith(_SYNC_BACK_TEMP_PREFIX)
|
||||||
assert stale.exists()
|
assert not leaked.exists()
|
||||||
|
assert list(tmp_root.iterdir()) == []
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -48,32 +48,41 @@ GetFilesFn = Callable[[], list[tuple[str, str]]] # () -> [(host_path, remote_pa
|
|||||||
_SYNC_BACK_MAX_RETRIES = 3
|
_SYNC_BACK_MAX_RETRIES = 3
|
||||||
_SYNC_BACK_BACKOFF = (2, 4, 8) # seconds between retries
|
_SYNC_BACK_BACKOFF = (2, 4, 8) # seconds between retries
|
||||||
_SYNC_BACK_MAX_BYTES = 2 * 1024 * 1024 * 1024 # 2 GiB — refuse to extract larger tars
|
_SYNC_BACK_MAX_BYTES = 2 * 1024 * 1024 * 1024 # 2 GiB — refuse to extract larger tars
|
||||||
_SYNC_BACK_TEMP_TAR_PREFIX = "hermes-sync-back-"
|
_SYNC_BACK_TEMP_PREFIX = "hermes-sync-back-"
|
||||||
_SYNC_BACK_STALE_TAR_SECONDS = 24 * 60 * 60
|
# A sync-back temp entry (the downloaded tar or the extraction staging dir) is only leaked by
|
||||||
|
# a hard kill (SIGKILL/OOM/power loss — the ``finally`` never runs), so anything older than
|
||||||
|
# this is safe to reclaim; a live transfer is hours younger than the cutoff.
|
||||||
|
_SYNC_BACK_STALE_SECONDS = 6 * 60 * 60
|
||||||
|
|
||||||
|
|
||||||
def _cleanup_stale_sync_back_tars(temp_dir: Path | None = None) -> None:
|
def _cleanup_stale_sync_back_temp(temp_dir: Path | None = None) -> int:
|
||||||
"""Remove old sync-back archives left behind when a process is hard-killed.
|
"""Remove sync-back tars and staging dirs left behind by a hard-killed process.
|
||||||
|
|
||||||
Only archives created with this module's dedicated prefix are considered. A
|
Only entries carrying this module's prefix and older than ``_SYNC_BACK_STALE_SECONDS``
|
||||||
cleanup race or permission error must not prevent the current sync-back.
|
are touched. Returns the number of entries removed; a permission error or a race with
|
||||||
|
another sync-back must not prevent the current one.
|
||||||
"""
|
"""
|
||||||
directory = temp_dir or Path(tempfile.gettempdir())
|
directory = temp_dir or Path(tempfile.gettempdir())
|
||||||
cutoff = time.time() - _SYNC_BACK_STALE_TAR_SECONDS
|
cutoff = time.time() - _SYNC_BACK_STALE_SECONDS
|
||||||
|
removed = 0
|
||||||
try:
|
try:
|
||||||
candidates = directory.glob(f"{_SYNC_BACK_TEMP_TAR_PREFIX}*.tar")
|
candidates = list(directory.glob(f"{_SYNC_BACK_TEMP_PREFIX}*"))
|
||||||
for candidate in candidates:
|
|
||||||
try:
|
|
||||||
if candidate.is_symlink() or not candidate.is_file():
|
|
||||||
continue
|
|
||||||
if candidate.stat().st_mtime >= cutoff:
|
|
||||||
continue
|
|
||||||
candidate.unlink()
|
|
||||||
logger.debug("sync_back: removed stale temporary archive %s", candidate)
|
|
||||||
except OSError:
|
|
||||||
logger.debug("sync_back: could not remove stale temporary archive %s", candidate)
|
|
||||||
except OSError:
|
except OSError:
|
||||||
logger.debug("sync_back: could not scan temporary directory %s", directory)
|
logger.debug("sync_back: could not scan temporary directory %s", directory)
|
||||||
|
return 0
|
||||||
|
for candidate in candidates:
|
||||||
|
try:
|
||||||
|
if candidate.is_symlink() or candidate.lstat().st_mtime >= cutoff:
|
||||||
|
continue
|
||||||
|
if candidate.is_dir():
|
||||||
|
shutil.rmtree(candidate)
|
||||||
|
else:
|
||||||
|
candidate.unlink()
|
||||||
|
removed += 1
|
||||||
|
logger.debug("sync_back: removed stale temporary entry %s", candidate)
|
||||||
|
except OSError:
|
||||||
|
logger.debug("sync_back: could not remove stale temporary entry %s", candidate)
|
||||||
|
return removed
|
||||||
|
|
||||||
|
|
||||||
def iter_sync_files(container_base: str = "/root/.hermes") -> list[tuple[str, str]]:
|
def iter_sync_files(container_base: str = "/root/.hermes") -> list[tuple[str, str]]:
|
||||||
@@ -337,13 +346,13 @@ class FileSyncManager:
|
|||||||
except Exception:
|
except Exception:
|
||||||
file_mapping = []
|
file_mapping = []
|
||||||
|
|
||||||
# A hard kill bypasses the finally below. Reclaim only old archives using
|
# A hard kill bypasses the finally below. Reclaim only old entries carrying our
|
||||||
# our dedicated prefix before allocating another full-tree download.
|
# prefix before allocating another full-tree download.
|
||||||
_cleanup_stale_sync_back_tars()
|
_cleanup_stale_sync_back_temp()
|
||||||
|
|
||||||
# mkstemp + close: NamedTemporaryFile keeps an exclusive handle on Windows, so the
|
# mkstemp + close: NamedTemporaryFile keeps an exclusive handle on Windows, so the
|
||||||
# backend's open(dest, "wb") / write_bytes on the same path raised PermissionError.
|
# backend's open(dest, "wb") / write_bytes on the same path raised PermissionError.
|
||||||
fd, tar_path = tempfile.mkstemp(prefix=_SYNC_BACK_TEMP_TAR_PREFIX, suffix=".tar")
|
fd, tar_path = tempfile.mkstemp(prefix=_SYNC_BACK_TEMP_PREFIX, suffix=".tar")
|
||||||
os.close(fd)
|
os.close(fd)
|
||||||
try:
|
try:
|
||||||
self._bulk_download_fn(Path(tar_path))
|
self._bulk_download_fn(Path(tar_path))
|
||||||
@@ -359,7 +368,7 @@ class FileSyncManager:
|
|||||||
tar_size, _SYNC_BACK_MAX_BYTES)
|
tar_size, _SYNC_BACK_MAX_BYTES)
|
||||||
return
|
return
|
||||||
|
|
||||||
with tempfile.TemporaryDirectory(prefix="hermes-sync-back-") as staging:
|
with tempfile.TemporaryDirectory(prefix=_SYNC_BACK_TEMP_PREFIX) as staging:
|
||||||
with tarfile.open(tar_path) as tar:
|
with tarfile.open(tar_path) as tar:
|
||||||
tar.extractall(staging, filter="data")
|
tar.extractall(staging, filter="data")
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user