Enhancement: Archive Retention and Verification

.gitignore

@@ -9,4 +9,6 @@ api_keys.json
 
 venv
 
-vite.config.ts.timestamp*
+vite.config.ts.timestamp*
+
+data_examples/*

TODO.md

@@ -1,23 +1,28 @@
 # TODO
 
 ## Scale and Performance
-- [ ] Add configurable limits for archive jobs (for example max files, max bytes, max runtime) and enforce them in backend checks.
-- [ ] Optimise for large drives and high file counts (streaming, batching, incremental processing or splitting into multiple RO-Crates). Max file size for ActiveScale is 50TB.
-- [ ] Reassess background execution approach; replace FastAPI `BackgroundTasks` with a durable queue (such as Celery or RQ) if reliability requirements increase.
+- [ ] Add configurable limits for archive jobs (for example max files, max bytes, max runtime) and enforce them in backend checks. Especially once we know the limit of whatever the prod infrastructure will be.
+- [ ] Reassess background execution approach; replace FastAPI `BackgroundTasks` with a durable queue (such as Celery or RQ) if reliability requirements increase. OR make it so only one archiving job can be run at a time to avoid concurrency issues with the current implementation.
 
-## Large Archive Datamodel and API (Draft)
-- [x] Extend `ArchiveSubmission` schema to track archive transport metadata (`archive_part_count`, byte sizes, object prefix, manifest key, and ordered part keys).
-- [x] Extend `JobStage` vocabulary for chunked archive workflow (`packaging`, `uploading`, `writing_manifest`) while retaining legacy stage values for compatibility.
-- [x] Extend `GET /api/v1/submission` response payload with new archive transport fields.
-- [x] Implement chunked archive writer (single logical tar split into ordered parts below ActiveScale object limit).
-- [x] Upload each part as a separate object under a deterministic prefix and persist uploaded part keys incrementally.
-- [x] Write sidecar archive manifest file (`archive-manifest.json`) during packaging with part ordering, per-part checksum, and total byte count.
-- [x] Upload sidecar archive manifest object to ActiveScale alongside uploaded parts.
-- [x] Replace single-object upload call in archive worker with chunked upload pipeline.
-- [x] Add retry/resume support to skip already uploaded parts and continue from persisted metadata.
-- [x] Add archive retrieval/reassembly utility using persisted part ordering from manifest.
-- [x] Add integration tests for chunked upload success, interrupted upload resume, and manifest integrity checks.
 
 ## Quality and Validation
+- [x] Improved object and Tar integrity checks (e.g. validate checksums of uploaded parts, verify manifest integrity, and ensure reassembled archive matches original input).
+- [ ] Add minimum metadata validation for the archive job (required fields are documented in the Wiki). This should be checked upfront in the API request validation so a job fails fast if there is missing metadata. Allow force flag to override this check if needed.
+- [ ] Validate generated RO-Crate output against the profile (follows on from the previous item). This should be done after the RO-Crate is generated, but before the archive is uploaded. If the RO-Crate fails validation, the job should fail and log an error message indicating what was invalid.
+- [ ] Ensure temp files are cleaned up properly from disk after a job completes or fails. Consider sdelete for permanent deletion of data.
 - [ ] Improve end-to-end tests that cover submission -> manifest -> RO-Crate build -> upload flow.
-- [ ] Validate generated RO-Crate output against the profile as part of CI.
+- [ ] Question: should custom s3 metadata be added on every uploaded object/prat of the archive? Currently just the archive manifest object has the metadata added.
+
+## Features and Enhancements
+- [x] Set retention policies on the stored archive objects to prevent accidental deletion, and ensure long-term preservation of the archived data.
+- [ ] Convert this project to use uv and replace linting etc with Ruff
+- [ ] The ProjectDB now stores data retention and classification metadata (Storage Properties table). The archive submission workflow should be updated to use this metadata from ProjectDB instead of requiring it to be provided in the API request. If values are provided in the request, they should override the ProjectDB values (but not alter the values in the ProjectDB). If the retention and classification are incorect in the projectDB they can be updated via the ProjectDB API, but driveoff s not responsible for updating ProjectDB values.
+- [ ] Add workflow for deleting the original drive data after successful archive. Key steps would be: flagging the source data as ready for deletion, running a separate cleanup job that verifies the archive integrity, and verifies the object exists before deleting, and handling any edge cases (e.g. what if the archive is corrupted?). It would also need to retain a copy of the archive manifest, and location of the stored archive, in the research drive (the drives/views/shares themselves will not be deleted). This may be a completely separate workflow from the archiving process, OR could be triggered from the archive submission API endpoint by adding an additional parameter to indicate whether deletion should be performed after archiving - design decision needed.
+- [ ] Notifications module to send slack messages to admins when jobs complete or fail
+
+## Infrastructure and Deployment
+- [ ] Add scripts/playbook for setting up the VM this will run on. Initial thought is to use Ansible for configuration management to set up the application and its dependencies. This will help automate the deployment process and ensure consistency across environments. Key steps would include: set squid proxy variables, configure git, clone repo, install Python, install Poetry (or uv if we switch), install dependencies, setup api keys file for driveoff (expects allowed keys to be in api_keys.json file), set up environment variables for configuration, and setup a process manager (e.g. systemd or supervisor) to run the application as a service and ensure it restarts on failure (for initial phase we may just run the fastapi manually though).
+- [ ] Setup autofs for auto mount and unmount of drives (if we will use linux)
+- [ ] Setup secrets management service - investigate options (Barbican, AWS Parameter Store, Hashicorp Vault, etc.).
+- [ ] Following on from the above item, set up a better solution for storing and managing API keys for the driveoff service. Currently, the allowed keys are stored in a JSON file, but a more secure and manageable solution should be implemented (e.g. using a secrets management service or encrypted storage).
+- [ ] Setup a monitoring tool and external log aggregation - investigate options for logging to a durable store (e.g. file, database, or logging service) instead of just stdout for better traceability and debugging.

modes/.env.development

@@ -1,10 +1,10 @@
 VITE_API_BASE_URL=http://localhost:8000
 CORS_ALLOW_HOST='["http://localhost:5173"]'
-# SMB_DRIVE_BASE_PATH=//files.auckland.ac.nz/research # Unifiles
 SMB_DRIVE_BASE_PATH=//test-2.d02pvaststg01.uoa.auckland.ac.nz
 SMB_LINUX_MOUNT_BASE_PATH=/mnt
 ACTIVESCALE_HOSTNAME=asi.s3.prod.nz
 ACTIVESCALE_REGION=us-east-1
 LOG_LEVEL=INFO
 PROJECTDB_BASE_URL=https://apis.test.auckland.ac.nz/proxy-eresearch-project
-ARCHIVE_CHUNK_SIZE_BYTES=54975581388798 # 50 TB
+ARCHIVE_CHUNK_SIZE_BYTES=54975581388798 # 50 TB
+ACTIVESCALE_RETENTION_OVERRIDE_DAYS=1

modes/.env.production

@@ -1,5 +1,5 @@
 CORS_ALLOW_HOST='["https://uoa-eresearch.github.io"]'
-SMB_DRIVE_BASE_PATH=//files.auckland.ac.nz/research
+SMB_DRIVE_BASE_PATH=
 SMB_LINUX_MOUNT_BASE_PATH=/mnt
 ACTIVESCALE_HOSTNAME=asi.s3.prod.nz
 ACTIVESCALE_REGION=us-east-1

pyproject.toml

@@ -8,7 +8,7 @@ package-mode = false
 
 [tool.poetry.dependencies]
 python = "^3.14"
-fastapi = {extras = ["standard"], version = "^0.136.3"}
+fastapi = {extras = ["standard"], version = "^0.137.1"}
 sqlmodel = "^0.0.22"
 sqlalchemy = "^2.0.36"
 bagit = "^1.8.1"
@@ -80,3 +80,4 @@ fix = [
        "black-fix"
        ]
 coverage-report = "python -m coverage report -m"
+dev = "fastapi dev src/api/main.py"

src/config.py

@@ -63,6 +63,14 @@ class Settings(BaseSettings):
     activescale_restore_poll_interval_seconds: int = 60
     # Maximum total time to wait for a restore to complete, in seconds (default 24 h)
     activescale_restore_poll_max_seconds: int = 86400
+    # Object retention (object lock COMPLIANCE mode) - (default True).
+    # Set to False in TEST environments so objects can be deleted quickly.
+    activescale_enable_object_retention: bool = True
+    # Fallback retention in years when retention_period_years is not set on a submission.
+    activescale_default_retention_years: int = 10
+    # Override: when set, use this many days as the retention period instead of the
+    # years-based calculation. Intended for TEST environments so objects expire quickly.
+    activescale_retention_override_days: int | None = None
 
     model_config = SettingsConfigDict(env_file=get_env_file(), extra="ignore")
 

src/models/common.py

@@ -1,7 +1,7 @@
 """Classes common to other models."""
 
 import re
-from datetime import datetime
+from datetime import datetime, timezone
 from enum import Enum
 from typing import Annotated
 
@@ -22,14 +22,30 @@ def validate_resdrive_name(drive_name: str) -> str:
     return drive_name
 
 
+def calculate_retention_end_datetime(
+    start_date: datetime, retention_years: int
+) -> datetime:
+    """Return the UTC-aware datetime on which retained data may be deleted.
+
+    Args:
+        start_date: The starting datetime (will be treated as UTC if naive).
+        retention_years: Number of full years to add.
+    """
+    if start_date.tzinfo is None:
+        start_date = start_date.replace(tzinfo=timezone.utc)
+    return start_date + relativedelta(years=retention_years)
+
+
 def calculate_retention_end_date(start_date: datetime, retention_years: int) -> str:
     """Return the date on which retained data may be deleted.
 
     Args:
         start_date: The starting date (typically project end date or today).
         retention_years: Number of full years to add.
     """
-    return (start_date + relativedelta(years=retention_years)).strftime("%Y-%m-%d")
+    return calculate_retention_end_datetime(start_date, retention_years).strftime(
+        "%Y-%m-%d"
+    )
 
 
 ResearchDriveName = Annotated[str, AfterValidator(validate_resdrive_name)]

src/packaging/archive_chunks.py

@@ -166,6 +166,96 @@ def _finalize_current_part(self) -> None:
         self._current_size = 0
 
 
+class _ChainReader:
+    """Read sequentially across an ordered list of part files without loading them into memory.
+
+    Presents a file-like ``read()`` interface so the concatenated byte stream
+    can be passed directly to :func:`tarfile.open` without first assembling a
+    single file on disk.
+    """
+
+    def __init__(self, parts: list[ArchivePartInfo], parts_dir: Path) -> None:
+        self._paths = [
+            parts_dir / p.file_name for p in sorted(parts, key=lambda p: p.index)
+        ]
+        self._file_index = 0
+        self._current_fp: BinaryIO | None = None
+
+    def read(self, size: int = -1) -> bytes:
+        """Read up to *size* bytes across part boundaries, or all remaining bytes if -1."""
+        if size == 0:
+            return b""
+
+        buf = bytearray()
+        remaining = size  # -1 means read everything
+
+        while True:
+            if self._current_fp is None:
+                if self._file_index >= len(self._paths):
+                    break
+                self._current_fp = (
+                    open(  # noqa: SIM115  # pylint: disable=consider-using-with
+                        self._paths[self._file_index], "rb"
+                    )
+                )
+                self._file_index += 1
+
+            chunk = self._current_fp.read(remaining if remaining != -1 else -1)
+            if chunk:
+                buf.extend(chunk)
+                if remaining != -1:
+                    remaining -= len(chunk)
+                    if remaining == 0:
+                        break
+            else:
+                # Current file exhausted — move to next
+                self._current_fp.close()
+                self._current_fp = None
+
+        return bytes(buf)
+
+    def close(self) -> None:
+        """Close any open file handle."""
+        if self._current_fp is not None:
+            self._current_fp.close()
+            self._current_fp = None
+
+    def __enter__(self) -> _ChainReader:
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        self.close()
+
+
+def verify_tar_parts_stream(parts: list[ArchivePartInfo], parts_dir: Path) -> None:
+    """Verify the integrity of a chunked tar.gz archive by streaming all parts.
+
+    Chains the ordered part files into a single logical byte stream and passes
+    it to :func:`tarfile.open` in streaming read mode (``r|gz``).  Iterating
+    :meth:`~tarfile.TarFile.getmembers` forces full decompression and gzip CRC
+    validation without writing anything to disk.
+
+    Raises:
+        FileNotFoundError: If any part file is missing.
+        tarfile.TarError: If the gzip stream is corrupt or the tar structure is invalid.
+    """
+    for part in parts:
+        part_path = parts_dir / part.file_name
+        if not part_path.exists():
+            raise FileNotFoundError(f"Archive part file not found: {part_path}")
+
+    with _ChainReader(parts, parts_dir) as chain:
+        with tarfile.open(fileobj=cast(BinaryIO, chain), mode="r|gz") as tar:
+            member_count = 0
+            for _ in tar:
+                member_count += 1
+
+    if member_count == 0:
+        raise tarfile.TarError(
+            "Tar stream contained no members — archive may be empty or corrupt"
+        )
+
+
 def build_chunked_tar_archive(
     source_dir: Path,
     output_dir: Path,