Fixing bugs

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

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):

src/archivist/core/archive_job.py

@@ -4,6 +4,8 @@
 
 from pathlib import Path
 
+from loguru import logger
+
 from .models import ManifestRequest
 
 
@@ -59,7 +61,54 @@ def manifest_id(self):
         return self._manifest_id
 
     def __repr__(self):
-        return f"<Archive manifest_id={self._manifest_id} type={self._type}>, <manifest={self._manifest}>, <local_root={self._local_root}>, <archive_root={self._archive_root}>"
+        # 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"<ArchiveJob manifest_id={self._manifest_id} type={self._type} "
+            f"entries={len(entries)} local_root={self._local_root} "
+            f"archive_root={self._archive_root}>"
+        )
+
+    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."""
@@ -68,6 +117,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

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
 
@@ -18,25 +18,38 @@
 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:
-            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(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()
         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(poll_interval)
 
 
 def _callback_worker_loop():
@@ -68,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)

src/archivist/settings.py

@@ -126,8 +126,32 @@ 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
+
+    # 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.
+    # 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__"
@@ -209,6 +233,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),
     ]