From afd394e6c402525d7b770e53ba1a828ab4ee032f Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Fri, 4 Sep 2026 13:08:20 -0400 Subject: [PATCH 1/7] fix: silent errors --- src/archivist/__main__.py | 43 ++++++++++++++++++++++++++ src/archivist/core/archive_job.py | 44 +++++++++++++++++++++++++++ src/archivist/storage/storage_disk.py | 30 +++++++++++++++--- src/archivist/tasks/archive.py | 11 +++++++ 4 files changed, 123 insertions(+), 5 deletions(-) diff --git a/src/archivist/__main__.py b/src/archivist/__main__.py index 1c7bb7e..7607cd5 100644 --- a/src/archivist/__main__.py +++ b/src/archivist/__main__.py @@ -193,6 +193,49 @@ def resend_callback(ctx, manifest_id): session.close() +@main.command() +@click.option("--manifest-id", type=str, required=True, help="The manifest whose archive should be stored again") +@click.pass_context +def requeue_archive(ctx, manifest_id): + """Return a completed archive to the pending queue so its files are stored again. + + `resend-callback` only resets callback bookkeeping on a row that is already + complete; this re-drives the storage itself, for an archive that was marked + complete without its files actually being there. + + Clearing `consumed` is enough for the running server's `dequeue()` to pick + the row up on its next worker iteration, so no restart is needed. Entries + already present at the right size are skipped, so a partial archive resumes + instead of copying everything again. + """ + + from archivist.database import get_session + from archivist.orm.archive import Archive + + session = get_session() + try: + item = session.query(Archive).filter_by(manifest_id=manifest_id).first() + if item is None: + click.secho("[ERROR]", fg="red", nl=False, err=True) + click.echo(f" No archive for manifest {manifest_id}", err=True) + sys.exit(1) + + # Deliberately not Archive.requeue(): that increments `retries` and + # leaves `completed` set, which is right for crash recovery and wrong + # for an operator re-drive. + item.consumed = False + item.consumed_time = None + item.completed = False + item.completed_time = None + item.failed = False + item.retries = 0 + session.commit() + + click.echo(f"Archive {item.id} ({manifest_id}) requeued; the server will pick it up shortly.") + finally: + session.close() + + @main.command() @click.pass_context def start_server(ctx): diff --git a/src/archivist/core/archive_job.py b/src/archivist/core/archive_job.py index 91e8448..03e6162 100644 --- a/src/archivist/core/archive_job.py +++ b/src/archivist/core/archive_job.py @@ -4,6 +4,8 @@ from pathlib import Path +from loguru import logger + from .models import ManifestRequest @@ -61,6 +63,45 @@ def manifest_id(self): def __repr__(self): return f", , , " + def _expand_directory(self, src_path, entry): + """Expand a directory manifest entry into one triple per contained file. + + The Librarian sends a directory's `size` as the sum of the files it + holds, which says nothing about any individual file. Storage checks + completeness per destination path, so a directory entry can never be + verified as it stands. Walking it gives each file its own size, which + makes verification meaningful and -- because already-correct files are + skipped -- makes a partial copy resume rather than start over. + + The declared aggregate is logged as a cross-check rather than enforced. + It is the only Librarian-supplied guarantee here -- the per-file sizes + come from disk -- but how the Librarian computes a directory's size is + not pinned down by the contract, and a hard gate on an off-by-a-little + definition would fail every directory manifest. Warn, and let the + per-file checks do the real work. + """ + expanded = [] + total = 0 + + for path in sorted(src_path.rglob("*")): + # Follows symlinks, as `shutil.copy2` does when storing, so verify + # and store agree on what a symlinked entry weighs. A broken + # symlink is skipped here and shows up as an aggregate mismatch. + if not path.is_file(): + continue + size = path.stat().st_size + total += size + expanded.append((path, Path(self._archive_root) / path.relative_to(self._local_root), size)) + + if total != entry["size"]: + logger.warning( + f"Directory {src_path} holds {total} bytes across {len(expanded)} files, but the " + f"manifest declares {entry['size']} (difference {total - entry['size']}). Archiving " + "what is on disk." + ) + + return expanded + def create_archive(self): """Create the archive based on the manifest or paths.""" # Placeholder for archive creation logic @@ -68,6 +109,9 @@ def create_archive(self): src_dst_paths = [] for entry in self._manifest["store_files"]: src_path = Path(entry["instance_path"]) + if src_path.is_dir(): + src_dst_paths.extend(self._expand_directory(src_path, entry)) + continue dst_path = Path(self._archive_root) / src_path.relative_to(self._local_root) src_dst_paths.append((src_path, dst_path, entry["size"])) return src_dst_paths diff --git a/src/archivist/storage/storage_disk.py b/src/archivist/storage/storage_disk.py index 2d41bc3..7f0c67b 100644 --- a/src/archivist/storage/storage_disk.py +++ b/src/archivist/storage/storage_disk.py @@ -44,15 +44,35 @@ def _entry_satisfied(dst_path: Path, expected_size: int) -> bool: return dst_path.is_file() and dst_path.stat().st_size == expected_size def _copy_files(self, file_list: list[tuple[Path, Path, int]]) -> None: + """Copy every entry, then raise if any of them failed. + + One unreadable file does not abandon the rest of the manifest: a + manifest can be a terabyte across hundreds of entries, and stopping at + the first error throws away every copy that would have succeeded after + it. Whatever did copy stays on disk and is skipped on the next run; + the raise is what stops the archive being reported as complete. + """ + errors: list[tuple[Path, Exception]] = [] + for src_path, dst_path, expected_size in file_list: if self._entry_satisfied(dst_path, expected_size): logger.debug(f"Skipping already-present {dst_path} (size {expected_size})") continue - os.makedirs(dst_path.parent, exist_ok=True) - if os.path.isdir(src_path): - shutil.copytree(src_path, dst_path, dirs_exist_ok=True) - else: - shutil.copy2(src_path, dst_path) + try: + os.makedirs(dst_path.parent, exist_ok=True) + if os.path.isdir(src_path): + shutil.copytree(src_path, dst_path, dirs_exist_ok=True) + else: + shutil.copy2(src_path, dst_path) + except OSError as err: + logger.exception(f"Failed to copy {src_path} -> {dst_path}") + errors.append((src_path, err)) + + if errors: + first_path, first_error = errors[0] + raise RuntimeError( + f"{len(errors)} of {len(file_list)} entries failed to copy; first was {first_path}: {first_error!r}" + ) def _store(self, archive: ArchiveJob) -> "StorageDisk": """Copy the archive's files into a per-manifest subdirectory of archive_root.""" diff --git a/src/archivist/tasks/archive.py b/src/archivist/tasks/archive.py index 4aa4321..1abde43 100644 --- a/src/archivist/tasks/archive.py +++ b/src/archivist/tasks/archive.py @@ -85,6 +85,17 @@ def process_status_queue(session_maker: Callable[[], Session] = get_session) -> ) if item is None: return True + + # `done()` is True for a future that raised, so the exception + # has to be read explicitly. Without this a failed or partial + # copy is reported to the Librarian as a success, and the + # error is discarded rather than logged. + store_error = storage_task.future.exception() + if store_error is not None: + loguru.logger.error(f"Archive {storage_task._archive.manifest_id} failed to store: {store_error!r}") + item.fail(session) + return True + try: item.archive_path = storage_task._archive.archive_root if item.manifest.librarian_name == get_settings().cli_librarian_name: From 255d99efd946a91d8af9e51c9072052614e82a91 Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Fri, 4 Sep 2026 13:36:29 -0400 Subject: [PATCH 2/7] performance improvements --- src/archivist/core/archive_job.py | 10 +++++++++- src/archivist/server.py | 19 +++++++++++++++++-- src/archivist/settings.py | 9 +++++++++ src/archivist/tasks/archive.py | 5 ++++- tests/conftest.py | 6 +++--- 5 files changed, 42 insertions(+), 7 deletions(-) diff --git a/src/archivist/core/archive_job.py b/src/archivist/core/archive_job.py index 03e6162..7d9b7c8 100644 --- a/src/archivist/core/archive_job.py +++ b/src/archivist/core/archive_job.py @@ -61,7 +61,15 @@ def manifest_id(self): return self._manifest_id def __repr__(self): - return f", , , " + # Summarise the manifest rather than dumping it: a production manifest + # runs to hundreds of entries, and a repr that interpolates all of them + # turns any incidental log line into megabytes of string building. + entries = self._manifest.get("store_files", []) if self._manifest else [] + return ( + f"" + ) def _expand_directory(self, src_path, entry): """Expand a directory manifest entry into one triple per contained file. diff --git a/src/archivist/server.py b/src/archivist/server.py index f4a09df..7a2a29f 100644 --- a/src/archivist/server.py +++ b/src/archivist/server.py @@ -15,6 +15,12 @@ _archive_stop_event = threading.Event() +# Idle backoff for the archive and status workers. Both used to spin with no +# sleep, which pegs a core and -- worse -- holds the GIL against the storage +# threads doing the actual copying, throttling every archive in flight. +_WORKER_IDLE_SECONDS = 1.0 + + def _archive_worker_loop(): from loguru import logger @@ -22,9 +28,14 @@ def _archive_worker_loop(): while not _archive_stop_event.is_set(): try: - start_archive(librarian_name=server_settings.name) - except Exception as ex: # noqa: BLE001 + did_work = start_archive(librarian_name=server_settings.name) + except Exception: # noqa: BLE001 logger.exception("Archive worker iteration failed") + did_work = False + # Only pause when the queue was empty, so a backlog still drains at + # full speed. + if not did_work: + _archive_stop_event.wait(_WORKER_IDLE_SECONDS) def _status_worker_loop(): @@ -37,6 +48,10 @@ def _status_worker_loop(): process_status_queue() except Exception: # noqa: BLE001 logger.exception("Archive worker iteration failed") + # Always pause: this loop polls futures that take minutes to hours, and + # it reports "handled" even when it merely re-queued an unfinished one, + # so there is no busy case worth spinning for. + _archive_stop_event.wait(_WORKER_IDLE_SECONDS) def _callback_worker_loop(): diff --git a/src/archivist/settings.py b/src/archivist/settings.py index 83cf72a..b7dc46b 100644 --- a/src/archivist/settings.py +++ b/src/archivist/settings.py @@ -209,6 +209,15 @@ def get_settings() -> "Settings": """ global _settings + # Return the cached object rather than re-reading the config file. The + # background workers call this on every loop iteration, and rebuilding a + # BaseSettings means a disk read plus a full pydantic-settings + # construction -- pure Python, holding the GIL, starving the storage + # threads doing the actual copying. Tests reset `_settings` to None to + # force a reload. + if _settings is not None: + return _settings + try_paths = [ os.environ.get("ARCHIVIST_CONFIG_PATH", None), ] diff --git a/src/archivist/tasks/archive.py b/src/archivist/tasks/archive.py index 1abde43..61c2d4e 100644 --- a/src/archivist/tasks/archive.py +++ b/src/archivist/tasks/archive.py @@ -74,7 +74,10 @@ def process_status_queue(session_maker: Callable[[], Session] = get_session) -> return False try: - loguru.logger.debug(f"Processing status queue for storage task: {storage_task._archive}") + # Not an f-string: this runs on every iteration of a hot loop, and an + # f-string is built whether or not DEBUG is enabled. loguru defers the + # formatting until it knows the record will be emitted. + loguru.logger.debug("Processing status queue for storage task: {}", storage_task._archive.manifest_id) if storage_task.future.done(): loguru.logger.debug(f"Storage task for archive {storage_task._archive.manifest_id} future completed.") session = session_maker() diff --git a/tests/conftest.py b/tests/conftest.py index a291196..cc812dc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -62,9 +62,9 @@ def settings(tmp_path, archive_root, local_root) -> Settings: """A Settings instance backed by a throwaway on-disk sqlite database. Callback settings belong here rather than being assigned onto the object - later: `get_settings()` re-reads the config file on every call and builds - a fresh Settings, so a mutated instance is invisible to anything that - calls it again. + later, so the fixture describes one complete configuration. The autouse + reset clears the `get_settings()` cache around every test, so nothing + leaks between them. """ return Settings( From 09cf86cdaee21af78a4d52915f04a2b56afa81ce Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Fri, 4 Sep 2026 13:56:10 -0400 Subject: [PATCH 3/7] fix: concurrent copies --- src/archivist/server.py | 14 +++--- src/archivist/settings.py | 17 ++++++++ src/archivist/storage/storage_disk.py | 62 ++++++++++++++++++++------- 3 files changed, 69 insertions(+), 24 deletions(-) diff --git a/src/archivist/server.py b/src/archivist/server.py index 7a2a29f..3a68116 100644 --- a/src/archivist/server.py +++ b/src/archivist/server.py @@ -15,17 +15,13 @@ _archive_stop_event = threading.Event() -# Idle backoff for the archive and status workers. Both used to spin with no -# sleep, which pegs a core and -- worse -- holds the GIL against the storage -# threads doing the actual copying, throttling every archive in flight. -_WORKER_IDLE_SECONDS = 1.0 - - def _archive_worker_loop(): from loguru import logger + from .settings import get_settings from .tasks.archive import start_archive + poll_interval = get_settings().worker_poll_interval_seconds while not _archive_stop_event.is_set(): try: did_work = start_archive(librarian_name=server_settings.name) @@ -35,14 +31,16 @@ def _archive_worker_loop(): # Only pause when the queue was empty, so a backlog still drains at # full speed. if not did_work: - _archive_stop_event.wait(_WORKER_IDLE_SECONDS) + _archive_stop_event.wait(poll_interval) def _status_worker_loop(): from loguru import logger + from .settings import get_settings from .tasks.archive import process_status_queue + poll_interval = get_settings().worker_poll_interval_seconds while not _archive_stop_event.is_set(): try: process_status_queue() @@ -51,7 +49,7 @@ def _status_worker_loop(): # Always pause: this loop polls futures that take minutes to hours, and # it reports "handled" even when it merely re-queued an unfinished one, # so there is no busy case worth spinning for. - _archive_stop_event.wait(_WORKER_IDLE_SECONDS) + _archive_stop_event.wait(poll_interval) def _callback_worker_loop(): diff --git a/src/archivist/settings.py b/src/archivist/settings.py index b7dc46b..ffc3446 100644 --- a/src/archivist/settings.py +++ b/src/archivist/settings.py @@ -126,8 +126,25 @@ class Settings(BaseSettings): checksum_timeout: datetime.timedelta = datetime.timedelta(days=1) # Number of background threads used to perform filesystem storage copies. + # This sizes the pool that runs whole archive *jobs*, not the files within + # one job -- see copy_threads for that. storage_threads: int = 4 + # Concurrent file copies within a single archive job. Copying to network + # storage is latency-bound rather than bandwidth-bound: each file costs a + # round trip to open, write, commit and stat, and a single stream spends + # most of its time waiting. Concurrent streams hide that latency. Set to 1 + # for serial copying, which is the right choice for local disk where + # concurrency only causes seeking. + copy_threads: int = 8 + + # How long the archive and status workers idle between polls. These loops + # are pure Python, so spinning without a pause holds the GIL against the + # storage threads doing the copying and throttles every archive in flight. + # Lower it for snappier pickup of new manifests, raise it to leave more + # room for large transfers. + worker_poll_interval_seconds: float = 1.0 + # Reserved librarian name used by the CLI archive path. Jobs submitted # under this name never trigger a Librarian callback. cli_librarian_name: str = "__cli__" diff --git a/src/archivist/storage/storage_disk.py b/src/archivist/storage/storage_disk.py index 7f0c67b..ddb098a 100644 --- a/src/archivist/storage/storage_disk.py +++ b/src/archivist/storage/storage_disk.py @@ -1,6 +1,6 @@ import os import shutil -from concurrent.futures import Future, ThreadPoolExecutor +from concurrent.futures import Future, ThreadPoolExecutor, as_completed from pathlib import Path from loguru import logger @@ -43,30 +43,60 @@ def _entry_satisfied(dst_path: Path, expected_size: int) -> bool: """True if dst_path exists as a regular file of the expected size.""" return dst_path.is_file() and dst_path.stat().st_size == expected_size + def _copy_one(self, src_path: Path, dst_path: Path, expected_size: int) -> bool: + """Copy a single entry; True if copied, False if already satisfied. + + Raises whatever the filesystem raises -- the caller collects failures. + """ + if self._entry_satisfied(dst_path, expected_size): + logger.debug("Skipping already-present {} (size {})", dst_path, expected_size) + return False + + os.makedirs(dst_path.parent, exist_ok=True) + if os.path.isdir(src_path): + shutil.copytree(src_path, dst_path, dirs_exist_ok=True) + else: + shutil.copy2(src_path, dst_path) + return True + def _copy_files(self, file_list: list[tuple[Path, Path, int]]) -> None: """Copy every entry, then raise if any of them failed. + Entries are copied concurrently. Archive storage is typically remote + (NFS), where a file's cost is dominated by round trips to open, write, + commit and stat it rather than by bandwidth, so a single stream spends + most of its time waiting. `settings.copy_threads` streams overlap that + latency; set it to 1 for local disk, where concurrency only causes + seeking. + One unreadable file does not abandon the rest of the manifest: a manifest can be a terabyte across hundreds of entries, and stopping at the first error throws away every copy that would have succeeded after it. Whatever did copy stays on disk and is skipped on the next run; the raise is what stops the archive being reported as complete. """ - errors: list[tuple[Path, Exception]] = [] - - for src_path, dst_path, expected_size in file_list: - if self._entry_satisfied(dst_path, expected_size): - logger.debug(f"Skipping already-present {dst_path} (size {expected_size})") - continue - try: - os.makedirs(dst_path.parent, exist_ok=True) - if os.path.isdir(src_path): - shutil.copytree(src_path, dst_path, dirs_exist_ok=True) - else: - shutil.copy2(src_path, dst_path) - except OSError as err: - logger.exception(f"Failed to copy {src_path} -> {dst_path}") - errors.append((src_path, err)) + errors: list[tuple[Path, BaseException]] = [] + workers = self._settings.copy_threads if self._settings else 8 + + if workers <= 1: + for src_path, dst_path, expected_size in file_list: + try: + self._copy_one(src_path, dst_path, expected_size) + except OSError as err: + logger.exception(f"Failed to copy {src_path} -> {dst_path}") + errors.append((src_path, err)) + else: + # A pool of its own rather than StorageDisk._executor: this method + # is itself running on that shared pool, so taking slots from it + # for the children deadlocks once enough archives are in flight. + with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="copy") as pool: + futures = {pool.submit(self._copy_one, src, dst, size): (src, dst) for src, dst, size in file_list} + for future in as_completed(futures): + src_path, dst_path = futures[future] + error = future.exception() + if error is not None: + logger.error(f"Failed to copy {src_path} -> {dst_path}: {error!r}") + errors.append((src_path, error)) if errors: first_path, first_error = errors[0] From c76b2d383463bff6b9482e448adf49240da76607 Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Fri, 4 Sep 2026 16:54:22 -0400 Subject: [PATCH 4/7] fix: timestamps --- src/archivist/storage/storage_disk.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/archivist/storage/storage_disk.py b/src/archivist/storage/storage_disk.py index ddb098a..a809608 100644 --- a/src/archivist/storage/storage_disk.py +++ b/src/archivist/storage/storage_disk.py @@ -98,12 +98,38 @@ def _copy_files(self, file_list: list[tuple[Path, Path, int]]) -> None: logger.error(f"Failed to copy {src_path} -> {dst_path}: {error!r}") errors.append((src_path, error)) + self._restore_directory_times(file_list) + if errors: first_path, first_error = errors[0] raise RuntimeError( f"{len(errors)} of {len(file_list)} entries failed to copy; first was {first_path}: {first_error!r}" ) + @staticmethod + def _restore_directory_times(file_list: list[tuple[Path, Path, int]]) -> None: + """Copy each destination directory's timestamps from its source. + + `copy2` preserves a file's own mtime, but the directories holding them + are created by `os.makedirs`, which stamps them with the time of the + copy. This has to run after every file is written, because writing into + a directory updates that directory's mtime again. + + Deepest first, so a parent is stamped after the children that were + created inside it. + + Failures here are logged and not collected as errors: the archived data + is intact either way, and losing a directory's mtime is not worth + failing a terabyte transfer over. + """ + pairs = {file_dst.parent: file_src.parent for file_src, file_dst, _ in file_list} + + for dst_dir in sorted(pairs, key=lambda path: len(path.parts), reverse=True): + try: + shutil.copystat(pairs[dst_dir], dst_dir) + except OSError as err: + logger.warning(f"Could not restore timestamps on {dst_dir}: {err!r}") + def _store(self, archive: ArchiveJob) -> "StorageDisk": """Copy the archive's files into a per-manifest subdirectory of archive_root.""" self._archive = archive From 518ee55608582092d44da8be757198c0c66bd033 Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Fri, 4 Sep 2026 17:18:38 -0400 Subject: [PATCH 5/7] fix:delayed port opening --- src/archivist/server.py | 19 ++++++++++--- src/archivist/tasks/archive.py | 52 +++++++++++++++++++++++++++++----- tests/test_server.py | 8 +++++- 3 files changed, 67 insertions(+), 12 deletions(-) diff --git a/src/archivist/server.py b/src/archivist/server.py index 3a68116..a0794fb 100644 --- a/src/archivist/server.py +++ b/src/archivist/server.py @@ -8,7 +8,7 @@ from fastapi import FastAPI -from archivist.tasks.archive import reconcile_orphaned_archives +from archivist.tasks.archive import find_orphaned_archives, reconcile_orphaned_archives from .settings import server_settings @@ -81,9 +81,20 @@ async def startup_shutdown_server(app: FastAPI): logger.info("Archivist server starting up") - reconcile_orphaned_archives() - - thread_pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="worker") + # Only the snapshot happens here -- one indexed query. Verification stats + # every file of every unfinished archive, which is minutes over network + # storage, and uvicorn does not bind its socket until this handler returns. + # Doing it inline would mean the port stays closed, and the Librarian's + # POSTs are refused rather than queued, for the whole of that window. + orphan_ids = find_orphaned_archives() + if orphan_ids: + logger.info(f"Found {len(orphan_ids)} archive(s) to reconcile; verifying in the background.") + + thread_pool = ThreadPoolExecutor(max_workers=5, thread_name_prefix="worker") + # Safe to run alongside the workers because the orphan set was captured + # before any of them started: a manifest arriving from here on becomes + # `consumed` under a worker and can never be mistaken for an orphan. + thread_pool.submit(reconcile_orphaned_archives, orphan_ids) thread_pool.submit(_archive_worker_loop) thread_pool.submit(_status_worker_loop) thread_pool.submit(_callback_worker_loop) diff --git a/src/archivist/tasks/archive.py b/src/archivist/tasks/archive.py index 61c2d4e..205fa4e 100644 --- a/src/archivist/tasks/archive.py +++ b/src/archivist/tasks/archive.py @@ -126,23 +126,61 @@ def process_status_queue(session_maker: Callable[[], Session] = get_session) -> return True +def find_orphaned_archives(session_maker: Callable[[], Session] = get_session) -> list[int]: + """ + Return the ids of archives orphaned by a crash: consumed but never + completed. + + Split from the reconciliation itself because this part is a single indexed + query and the reconciliation is not. Taking the snapshot before any worker + starts is what makes it safe to reconcile in the background: a manifest + that arrives afterwards becomes `consumed=True` the moment a worker picks + it up, and would otherwise look exactly like an orphan to a concurrent + scan -- which would requeue a job that is already being copied. + """ + session = session_maker() + try: + return [item.id for item in session.query(Archive).filter_by(consumed=True, completed=False).all()] + finally: + session.close() + + def reconcile_orphaned_archives( + orphan_ids: list[int] | None = None, session_maker: Callable[[], Session] = get_session, ) -> None: """ - On startup, re-drive archive jobs orphaned by a crash. + Re-drive archive jobs orphaned by a crash. A row that was consumed but never completed lost its in-memory Future when the process died. For each such row: verify the destination against the manifest; if already satisfied, complete it; otherwise requeue it (up to settings.max_archive_retries) so the normal worker loop re-runs the now idempotent store(), or fail it once the retry cap is hit. + + Verification stats every file of every unfinished archive, so this is slow + -- minutes on a large manifest over network storage. It therefore runs on a + worker thread rather than blocking startup, working from `orphan_ids` + captured by `find_orphaned_archives` before the workers began. Passing + nothing falls back to finding them here, which is only safe when no worker + is running. """ settings = get_settings() - session = session_maker() - try: - orphans = session.query(Archive).filter_by(consumed=True, completed=False).all() - for item in orphans: + if orphan_ids is None: + orphan_ids = find_orphaned_archives(session_maker) + + if not orphan_ids: + return + + # A session per orphan rather than one for the whole pass: this can run for + # minutes, and a transaction held open that long blocks vacuum and shows up + # as idle-in-transaction. + for orphan_id in orphan_ids: + session = session_maker() + try: + item = session.query(Archive).filter_by(id=orphan_id).first() + if item is None: + continue try: manifest = { "store_files": [ @@ -170,5 +208,5 @@ def reconcile_orphaned_archives( loguru.logger.info(f"Requeued orphan {item.id} (retry {item.retries}).") except Exception: # noqa: BLE001 loguru.logger.exception(f"Failed to reconcile orphan {getattr(item, 'manifest_id', '?')}; skipping.") - finally: - session.close() + finally: + session.close() diff --git a/tests/test_server.py b/tests/test_server.py index 0525720..976c508 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -70,10 +70,16 @@ async def _run(): mock.patch.object(server_module, "_archive_worker_loop", lambda: started.append("archive")), mock.patch.object(server_module, "_status_worker_loop", lambda: started.append("status")), mock.patch.object(server_module, "_callback_worker_loop", lambda: started.append("callback")), + mock.patch.object(server_module, "find_orphaned_archives", return_value=[7, 9]) as mock_find, mock.patch.object(server_module, "reconcile_orphaned_archives") as mock_reconcile, ): asyncio.run(_run()) - mock_reconcile.assert_called_once() + # The snapshot is taken inline, before any worker starts: that ordering is + # what stops a newly-dequeued manifest being mistaken for an orphan. + mock_find.assert_called_once() + # Reconciliation runs on a worker thread, but the lifespan's shutdown waits + # for the pool, so by here it has run -- with the ids captured up front. + mock_reconcile.assert_called_once_with([7, 9]) assert sorted(started) == ["archive", "callback", "status"] assert server_module._archive_stop_event.is_set() From 9f0b214a358502f9182d91b39b44639a42fb039f Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Tue, 15 Sep 2026 13:34:47 +0300 Subject: [PATCH 6/7] fix: retrying single file if it fails --- src/archivist/settings.py | 7 +++++ src/archivist/storage/storage_disk.py | 27 ++++++++++++++---- tests/conftest.py | 1 + tests/test_storage.py | 41 +++++++++++++++++++++++++++ 4 files changed, 70 insertions(+), 6 deletions(-) diff --git a/src/archivist/settings.py b/src/archivist/settings.py index ffc3446..03b661d 100644 --- a/src/archivist/settings.py +++ b/src/archivist/settings.py @@ -138,6 +138,13 @@ class Settings(BaseSettings): # concurrency only causes seeking. copy_threads: int = 8 + # Attempts per file before a copy error counts as a failure. Minimum 1, which means no retries. + copy_attempts: int = 3 + + # Delay before retrying a failed file copy, doubling each attempt + # (5s, then 10s). + copy_retry_base_seconds: float = 5.0 + # How long the archive and status workers idle between polls. These loops # are pure Python, so spinning without a pause holds the GIL against the # storage threads doing the copying and throttles every archive in flight. diff --git a/src/archivist/storage/storage_disk.py b/src/archivist/storage/storage_disk.py index a809608..c863a11 100644 --- a/src/archivist/storage/storage_disk.py +++ b/src/archivist/storage/storage_disk.py @@ -1,5 +1,6 @@ import os import shutil +import time from concurrent.futures import Future, ThreadPoolExecutor, as_completed from pathlib import Path @@ -45,19 +46,33 @@ def _entry_satisfied(dst_path: Path, expected_size: int) -> bool: def _copy_one(self, src_path: Path, dst_path: Path, expected_size: int) -> bool: """Copy a single entry; True if copied, False if already satisfied. + In case of a NFS error, retry to copy the file a few times and then fail. Raises whatever the filesystem raises -- the caller collects failures. """ + if self._entry_satisfied(dst_path, expected_size): logger.debug("Skipping already-present {} (size {})", dst_path, expected_size) return False - os.makedirs(dst_path.parent, exist_ok=True) - if os.path.isdir(src_path): - shutil.copytree(src_path, dst_path, dirs_exist_ok=True) - else: - shutil.copy2(src_path, dst_path) - return True + for attempt in range(1, self._settings.copy_attempts + 1): + try: + os.makedirs(dst_path.parent, exist_ok=True) + # Inside the retry: isdir() returns False when stat itself + # errors, so a flaky stat can send a file down the wrong path. + if os.path.isdir(src_path): + shutil.copytree(src_path, dst_path, dirs_exist_ok=True) + else: + shutil.copy2(src_path, dst_path) + return True + except OSError as err: + if attempt == self._settings.copy_attempts: + raise + delay = self._settings.copy_retry_base_seconds * 2 ** (attempt - 1) + logger.warning( + f"Copy of {src_path} failed on attempt {attempt}/{self._settings.copy_attempts} ({err!r}); retrying in {delay:.0f}s" + ) + time.sleep(delay) def _copy_files(self, file_list: list[tuple[Path, Path, int]]) -> None: """Copy every entry, then raise if any of them failed. diff --git a/tests/conftest.py b/tests/conftest.py index cc812dc..8021ac4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -87,6 +87,7 @@ def settings(tmp_path, archive_root, local_root) -> Settings: "__cli__": ClientConfig(username=CLI_CREDENTIALS[0], password=CLI_CREDENTIALS[1]), }, callback_poll_interval_seconds=0.01, # keep the callback worker's idle wait out of test runtime + copy_retry_base_seconds=0, # keep copy retry backoff out of test runtime ) diff --git a/tests/test_storage.py b/tests/test_storage.py index 408fc55..528ab49 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -1,5 +1,8 @@ # Copyright (c) 2025-2026 Simons Observatory. # Full license can be found in the top level "LICENSE" file. +import shutil +from unittest import mock + import pytest from conftest import make_manifest_entry @@ -78,6 +81,44 @@ def test_disk_store_archive_directory_trees(use_settings, local_root, archive_ro assert (archive_root / "adir" / "inner.txt").read_text() == "inner-content" +def test_disk_store_retries_transient_copy_errors(use_settings, local_root, archive_root): + """Network storage returns EIO under load for files that read fine moments later.""" + + source = local_root / "file.txt" + source.write_text("content") + real_copy2 = shutil.copy2 + calls = [] + + def _flaky_copy2(src, dst): + calls.append(src) + if len(calls) == 1: + raise OSError(5, "Input/output error") + return real_copy2(src, dst) + + storage = StorageDisk(settings=use_settings) + with mock.patch("archivist.storage.storage_disk.shutil.copy2", side_effect=_flaky_copy2): + storage.store(_archive_for(source, local_root, archive_root)).future.result(timeout=5) + + assert len(calls) == 2 + assert (archive_root / "file.txt").read_text() == "content" + + +def test_disk_store_fails_after_copy_attempts_exhausted(use_settings, local_root, archive_root): + source = local_root / "file.txt" + source.write_text("content") + + storage = StorageDisk(settings=use_settings) + with mock.patch( + "archivist.storage.storage_disk.shutil.copy2", side_effect=OSError(5, "Input/output error") + ) as copy2: + # The errno has to survive into the manifest's failure message: it is + # what tells a transient storage error apart from a real problem. + with pytest.raises(RuntimeError, match="Input/output error"): + storage.store(_archive_for(source, local_root, archive_root)).future.result(timeout=5) + + assert copy2.call_count == use_settings.copy_attempts + + def test_disk_verify_archive(use_settings, local_root, archive_root): source = local_root / "file.txt" source.write_text("content") # 7 bytes From 73a569de21c2e6c93ddac2a40d31fdf1c103a213 Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Tue, 15 Sep 2026 17:57:18 +0300 Subject: [PATCH 7/7] fix: coverage, adding tests --- src/archivist/__init__.py | 1 - src/archivist/api.py | 16 -------- src/archivist/config.py | 3 -- src/archivist/core/registry.py | 67 ---------------------------------- tests/test_cli.py | 34 +++++++++++++++++ tests/test_storage.py | 19 ++++++++++ tests/test_tasks.py | 9 +++++ 7 files changed, 62 insertions(+), 87 deletions(-) delete mode 100644 src/archivist/api.py delete mode 100644 src/archivist/config.py delete mode 100644 src/archivist/core/registry.py diff --git a/src/archivist/__init__.py b/src/archivist/__init__.py index 5f632e4..27b573f 100644 --- a/src/archivist/__init__.py +++ b/src/archivist/__init__.py @@ -5,5 +5,4 @@ from ._version import __version__ from .core.archive_job import ArchiveJob -from .core.registry import RegistryLibrarian, RegistrySqlite from .storage import StorageDisk, StorageHPSS diff --git a/src/archivist/api.py b/src/archivist/api.py deleted file mode 100644 index 73c42d8..0000000 --- a/src/archivist/api.py +++ /dev/null @@ -1,16 +0,0 @@ -class Archivist: - def __init__(self, config): - self.config = config - self.archivist = None - - def start_server(self): - # Initialize the archivist with the provided configuration - self.archivist = self.config.get("archivist", {}) - - def archive(self, source_path, dest_path): - # Implementation for archiving a file - pass - - def extract(self, archive, dest_path): - # Implementation for extracting an archive - pass diff --git a/src/archivist/config.py b/src/archivist/config.py deleted file mode 100644 index dfe602f..0000000 --- a/src/archivist/config.py +++ /dev/null @@ -1,3 +0,0 @@ -class ArchivistConfiguration: - def __init__(self, config): - self.config = config diff --git a/src/archivist/core/registry.py b/src/archivist/core/registry.py deleted file mode 100644 index bb20b5a..0000000 --- a/src/archivist/core/registry.py +++ /dev/null @@ -1,67 +0,0 @@ -# Copyright (c) 2025-2026 Simons Observatory. -# Full license can be found in the top level "LICENSE" file. -"""Classes for working with archive registries.""" - - -class Registry: - """Base class representing a registry of archives.""" - - def __init__(self): - pass - - def _register(self, archive, storage_info): - raise NotImplementedError("Fell through to base class") - - def register(self, archive, storage_info): - """Store an archive in the registry. - - Args: - archive (Archive): The archive to register. - storage_info (dict): Storage-system specific metadata. - - Returns: - (None) - - """ - self._register(archive, storage_info) - - def _remove(self, archive_name): - raise NotImplementedError("Fell through to base class") - - def remove(self, archive_name): - """Remove an archive from the registry. - - Args: - archive_name (str): The name of the archive to remove. - - Returns: - (None) - - """ - self._remove(archive_name) - - -class RegistrySqlite(Registry): - """Simple class which registers archives in an SQLite DB.""" - - def __init__(self): - super().__init__() - - def _register(self, archive, storage_info): - pass - - def _remove(self, archive_name): - pass - - -class RegistryLibrarian(Registry): - """Store archive metadata with with the librarian.""" - - def __init__(self): - super().__init__() - - def _register(self, archive, storage_info): - pass - - def _remove(self, archive_name): - pass diff --git a/tests/test_cli.py b/tests/test_cli.py index 4c5a363..fde5fbe 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -8,6 +8,7 @@ is click/requests plumbing, so that's all this covers. """ +import datetime import hashlib import os from pathlib import Path @@ -166,6 +167,39 @@ def test_resend_callback_only_touches_completed_archives(config_path, db_session assert item.callback_next_retry is None +def test_requeue_archive_returns_a_finished_archive_to_the_queue(config_path, db_session, archive_root): + """Operator re-drive for an archive that finished without all of its files landing.""" + + now = datetime.datetime.now(datetime.UTC) + item = make_archive_item(db_session, id="m1", archive_root=str(archive_root)) + item.consumed = True + item.consumed_time = now + item.completed = True + item.completed_time = now + item.failed = True + item.retries = 2 + db_session.commit() + + result = CliRunner().invoke(main, ["-c", str(config_path), "requeue-archive", "--manifest-id", "m1"]) + + assert result.exit_code == 0 + assert "requeued" in result.output + + db_session.expire_all() + assert not item.consumed and item.consumed_time is None + assert not item.completed and item.completed_time is None + assert not item.failed + # Reset, unlike crash recovery's requeue: an operator re-drive starts the retry budget over. + assert item.retries == 0 + + +def test_requeue_archive_rejects_an_unknown_manifest(config_path, db_session): + result = CliRunner().invoke(main, ["-c", str(config_path), "requeue-archive", "--manifest-id", "nope"]) + + assert result.exit_code == 1 + assert "No archive for manifest nope" in result.output + + def test_start_server_hands_the_settings_to_uvicorn(config_path, settings): with mock.patch("uvicorn.run") as mock_run: result = CliRunner().invoke(main, ["-c", str(config_path), "start-server"]) diff --git a/tests/test_storage.py b/tests/test_storage.py index 528ab49..a1a2c46 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -119,6 +119,25 @@ def test_disk_store_fails_after_copy_attempts_exhausted(use_settings, local_root assert copy2.call_count == use_settings.copy_attempts +@pytest.mark.parametrize("existing,copied", [("content", False), ("con", True)], ids=["complete", "truncated"]) +def test_disk_store_skips_files_already_present_at_the_expected_size( + use_settings, local_root, archive_root, existing, copied +): + """What makes a requeue or restart resume, rather than copy a whole manifest again.""" + + source = local_root / "file.txt" + source.write_text("content") # 7 bytes + # A truncated file is what a crash mid-copy leaves behind; it must not count as done. + (archive_root / "file.txt").write_text(existing) + + storage = StorageDisk(settings=use_settings) + with mock.patch("archivist.storage.storage_disk.shutil.copy2", wraps=shutil.copy2) as copy2: + storage.store(_archive_for(source, local_root, archive_root, size=7)).future.result(timeout=5) + + assert copy2.called is copied + assert (archive_root / "file.txt").read_text() == "content" + + def test_disk_verify_archive(use_settings, local_root, archive_root): source = local_root / "file.txt" source.write_text("content") # 7 bytes diff --git a/tests/test_tasks.py b/tests/test_tasks.py index b05a661..4a96f02 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -116,6 +116,15 @@ def test_process_status_queue_marks_rows_completed_or_failed(db_session, enqueue assert item.completed assert item.failed + # A copy that raised still leaves its future `done()`; it must fail, not complete. + enqueue_manifest(source, id="copy-raised") + with mock.patch("archivist.storage.storage_disk.shutil.copy2", side_effect=OSError(5, "Input/output error")): + start_archive(librarian_name="test-librarian") + item = _drain_until_completed(db_session, "copy-raised") + + assert item.completed + assert item.failed + def test_completion_records_whether_a_callback_is_owed(db_session, use_settings, enqueue_manifest, local_root): """Completing a job decides the callback: real librarians are owed one, CLI jobs are not."""