From 884aed99216bb2a53896fecd73698be9264367c4 Mon Sep 17 00:00:00 2001 From: Cabbagec Date: Thu, 23 Jul 2026 07:21:19 +0000 Subject: [PATCH] feat: execute durable archive transfers --- e2e/README.md | 7 + e2e/control/compose.yaml | 2 +- e2e/scenarios/archive_happy.py | 168 +++++++++ e2e/scripts/archive-happy.sh | 90 +++++ e2e/scripts/up.sh | 3 + src/archive_clients/daemon.py | 213 ++++++++++- src/archive_clients/jobs.py | 557 +++++++++++++++++++++++++++++ src/archive_clients/qbittorrent.py | 84 ++++- src/archive_clients/resources.py | 3 +- src/archive_clients/state.py | 107 ++++++ src/archive_clients/transfer.py | 44 +++ tests/test_daemon.py | 2 +- tests/test_jobs.py | 255 +++++++++++++ tests/test_qbittorrent.py | 91 +++++ 14 files changed, 1618 insertions(+), 8 deletions(-) create mode 100755 e2e/scenarios/archive_happy.py create mode 100755 e2e/scripts/archive-happy.sh create mode 100644 src/archive_clients/jobs.py create mode 100644 tests/test_jobs.py diff --git a/e2e/README.md b/e2e/README.md index 03f5df3..c8378ad 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -26,6 +26,13 @@ control worktree. `E2E_WAIT_SECONDS` overrides the 360-second assertion deadline. `down.sh` removes only the five exact Compose projects and the labelled E2E network; it intentionally retains all bind-mounted runtime state. +Set `E2E_RUN_TRANSFER=1` to follow route verification with a real +cache-1 → archive-1 transfer followed by archive-1 → cache-2 unarchive. The +scenario creates a deterministic one-file torrent in the isolated cache +qBittorrent, submits both jobs through the test HTTP adapter, waits for all five +durable steps in each direction, and verifies target qBittorrent selections, +retained source entries, and target file digests. + The Syncthing 2.1.2 and LinuxServer qBittorrent multi-platform image indexes are digest-pinned. Runtime secrets are generated with mode 0600 and ignored by Git. The qBittorrent test config limits its authentication bypass to the diff --git a/e2e/control/compose.yaml b/e2e/control/compose.yaml index 54c02bb..bb1f79d 100644 --- a/e2e/control/compose.yaml +++ b/e2e/control/compose.yaml @@ -8,6 +8,7 @@ services: user: "1001:1001" volumes: - ./config/config.json:/etc/archive-control/config.json:ro + - ../scenarios:/e2e/scenarios:ro - ./secrets:/run/secrets:ro - ./runtime/state:/var/lib/archive-control - ./runtime/backups:/var/backups/archive-control @@ -26,4 +27,3 @@ networks: archive-control-e2e: external: true name: archive-control-e2e - diff --git a/e2e/scenarios/archive_happy.py b/e2e/scenarios/archive_happy.py new file mode 100755 index 0000000..ec4961c --- /dev/null +++ b/e2e/scenarios/archive_happy.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +"""Drive one real transfer through the test HTTP adapter.""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +import urllib.parse +import urllib.request +import uuid +from datetime import datetime, timezone + + +CONTROL = "http://127.0.0.1:18081/test/v1" +INFO_HASH = "330be0cb7c2201135a2de63b28e77993745ff688" +LOGICAL_BYTES = 31 + + +def get_json(url: str): + with urllib.request.urlopen(url, timeout=5) as response: + return json.load(response) + + +def post_json(url: str, value: object): + encoded = json.dumps(value, separators=(",", ":")).encode() + request = urllib.request.Request( + url, + data=encoded, + method="POST", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(request, timeout=5) as response: + return json.load(response) + + +def qb_json(endpoint: str, path: str, **parameters: str): + query = urllib.parse.urlencode(parameters) + suffix = f"?{query}" if query else "" + return get_json(f"{endpoint}{path}{suffix}") + + +def wait_for(predicate, timeout: float, detail: str): + deadline = time.monotonic() + timeout + last = None + while time.monotonic() < deadline: + try: + last = predicate() + if last: + return last + except (OSError, ValueError, KeyError, IndexError) as exc: + last = f"{type(exc).__name__}: {exc}" + time.sleep(0.5) + raise RuntimeError(f"timed out waiting for {detail}; last={last!r}") + + +def qb_resource_is_complete(endpoint: str): + records = qb_json(endpoint, "/torrents/info", hashes=INFO_HASH) + if len(records) != 1 or float(records[0].get("progress", 0)) < 1: + return None + files = qb_json(endpoint, "/torrents/files", hash=INFO_HASH) + if ( + len(files) != 1 + or float(files[0].get("progress", 0)) < 1 + or int(files[0].get("priority", 0)) <= 0 + ): + return None + return records[0] + + +def ready_route(cache_client: str, archive_client: str): + for route in get_json(f"{CONTROL}/routes"): + if ( + route["cache_client_id"] == cache_client + and route["archive_client_id"] == archive_client + and route["state"] == "ready" + ): + return route + return None + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument( + "--operation", choices=("archive", "unarchive"), required=True + ) + parser.add_argument("--source-client", required=True) + parser.add_argument("--target-client", required=True) + parser.add_argument("--source-qb", required=True) + parser.add_argument("--target-qb", required=True) + parser.add_argument("--cache-client", required=True) + parser.add_argument("--archive-client", required=True) + args = parser.parse_args() + + wait_for( + lambda: qb_resource_is_complete(args.source_qb), + 60, + "source qBittorrent verification", + ) + route = wait_for( + lambda: ready_route(args.cache_client, args.archive_client), + 60, + f"{args.cache_client}/{args.archive_client} route", + ) + job_id = str(uuid.uuid4()) + now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + definition = { + "jobId": job_id, + "idempotencyKey": str(uuid.uuid4()), + "operation": ( + "JOB_OPERATION_ARCHIVE" + if args.operation == "archive" + else "JOB_OPERATION_UNARCHIVE" + ), + "resourceId": {"infoHashV1Hex": INFO_HASH}, + "resourceDisplayName": "fixture.bin", + "createdAt": now, + "transfer": { + "sourceClientId": args.source_client, + "targetClientId": args.target_client, + "routeId": route["route_id"], + "requestedFiles": {"ranges": [{"first": 0, "last": 0}]}, + "transferDeltaFiles": {"ranges": [{"first": 0, "last": 0}]}, + "requestedLogicalBytes": str(LOGICAL_BYTES), + "transferDeltaLogicalBytes": str(LOGICAL_BYTES), + }, + } + created = post_json(f"{CONTROL}/jobs", definition) + if created["job_id"] != job_id: + raise RuntimeError("control returned the wrong job") + + def finished_job(): + job = get_json(f"{CONTROL}/jobs/{job_id}") + if job["state"] == "JOB_STATE_FAILED": + raise RuntimeError(f"{args.operation} job failed: {job}") + if job["state"] == "JOB_STATE_SUCCEEDED" and job["committed"]: + return job + return None + + job = wait_for(finished_job, 180, f"committed {args.operation} job") + wait_for( + lambda: qb_resource_is_complete(args.target_qb), + 30, + "target qBittorrent placement", + ) + if qb_resource_is_complete(args.source_qb) is None: + raise RuntimeError( + f"{args.operation} transfer did not retain its source placement" + ) + print(json.dumps({ + "operation": args.operation, + "job_id": job_id, + "state": job["state"], + "committed": job["committed"], + "route_id": route["route_id"], + "source_retained": True, + "target_verified": True, + }, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: + print(f"transfer happy-path failed: {exc}", file=sys.stderr) + raise diff --git a/e2e/scripts/archive-happy.sh b/e2e/scripts/archive-happy.sh new file mode 100755 index 0000000..085ac26 --- /dev/null +++ b/e2e/scripts/archive-happy.sh @@ -0,0 +1,90 @@ +#!/usr/bin/env bash +set -euo pipefail + +source "$(dirname "$0")/lib.sh" + +info_hash=330be0cb7c2201135a2de63b28e77993745ff688 +torrent_base64=ZDQ6aW5mb2Q2Omxlbmd0aGkzMWU0Om5hbWUxMTpmaXh0dXJlLmJpbjEyOnBpZWNlIGxlbmd0aGkxNjM4NGU2OnBpZWNlczIwOlFlB05MtIUU4oem2MNz5LJW/GI6ZWU= + +wait_qb_absent() { + local node=$1 + local deadline=$((SECONDS + 30)) + while (( SECONDS < deadline )); do + local records + records=$(compose_node "$node" exec -T qbittorrent curl -fsS \ + "http://127.0.0.1:8080/api/v2/torrents/info?hashes=$info_hash") + if [[ "$records" == "[]" ]]; then + return + fi + sleep 1 + done + printf 'timed out waiting for fixture removal on %s\n' "$node" >&2 + exit 1 +} + +for node in cache-1 archive-1; do + compose_node "$node" exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/delete \ + --data-urlencode "hashes=$info_hash" \ + --data-urlencode "deleteFiles=true" >/dev/null || true + wait_qb_absent "$node" +done + +compose_node cache-1 exec -T qbittorrent /bin/sh -c \ + 'printf "archive-control-e2e-happy-path\n" > /downloads/fixture.bin' +compose_node cache-1 exec -T qbittorrent /bin/sh -c \ + "printf '%s' '$torrent_base64' | base64 -d > /tmp/fixture.torrent" +compose_node cache-1 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/add \ + -F torrents=@/tmp/fixture.torrent \ + -F savepath=/downloads \ + -F stopped=true >/dev/null +compose_node cache-1 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/recheck \ + --data-urlencode "hashes=$info_hash" >/dev/null + +compose_control exec -T control \ + python /e2e/scenarios/archive_happy.py \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-1 \ + --source-qb http://qb-cache-1:8080/api/v2 \ + --target-qb http://qb-archive-1:8080/api/v2 \ + --cache-client cache-1 \ + --archive-client archive-1 + +target_digest=$(compose_node archive-1 exec -T qbittorrent \ + sha256sum /downloads/fixture.bin | awk '{print $1}') +if [[ "$target_digest" != \ + "2cb91dd48809bd418a6a54a8f90a9009eb9eef9259ab46052bc1fe108a317ecb" ]]; then + printf 'archive target content digest mismatch: %s\n' \ + "$target_digest" >&2 + exit 1 +fi +printf 'archive filesystem content verified: sha256=%s\n' "$target_digest" + +compose_node cache-2 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/delete \ + --data-urlencode "hashes=$info_hash" \ + --data-urlencode "deleteFiles=true" >/dev/null || true +wait_qb_absent cache-2 + +compose_control exec -T control \ + python /e2e/scenarios/archive_happy.py \ + --operation unarchive \ + --source-client archive-1 \ + --target-client cache-2 \ + --source-qb http://qb-archive-1:8080/api/v2 \ + --target-qb http://qb-cache-2:8080/api/v2 \ + --cache-client cache-2 \ + --archive-client archive-1 + +cache_digest=$(compose_node cache-2 exec -T qbittorrent \ + sha256sum /downloads/fixture.bin | awk '{print $1}') +if [[ "$cache_digest" != \ + "2cb91dd48809bd418a6a54a8f90a9009eb9eef9259ab46052bc1fe108a317ecb" ]]; then + printf 'unarchive target content digest mismatch: %s\n' \ + "$cache_digest" >&2 + exit 1 +fi +printf 'unarchive filesystem content verified: sha256=%s\n' "$cache_digest" diff --git a/e2e/scripts/up.sh b/e2e/scripts/up.sh index 5a0cfcc..ddb5d16 100755 --- a/e2e/scripts/up.sh +++ b/e2e/scripts/up.sh @@ -21,3 +21,6 @@ for node in "${E2E_NODES[@]}"; do compose_node "$node" up -d --build done "$E2E_ROOT/scripts/wait-routes.sh" +if [[ "${E2E_RUN_TRANSFER:-0}" == "1" ]]; then + "$E2E_ROOT/scripts/archive-happy.sh" +fi diff --git a/src/archive_clients/daemon.py b/src/archive_clients/daemon.py index 727f6ea..ece3a29 100644 --- a/src/archive_clients/daemon.py +++ b/src/archive_clients/daemon.py @@ -7,6 +7,7 @@ import logging import random import time import uuid +from pathlib import Path, PurePosixPath from typing import Any from websockets.asyncio.client import connect @@ -14,6 +15,7 @@ from websockets.asyncio.client import connect from archive_clients.backup import SQLiteBackupManager from archive_clients.config import ClientConfig from archive_clients.inventory import InventoryService +from archive_clients.jobs import ClientJobExecutor, JobExecutionError from archive_clients.locking import DatabaseLease from archive_clients.probes import FilesystemProbe from archive_clients.protocol import ( @@ -73,12 +75,40 @@ class ArchiveClientDaemon: if healthy_syncthing and probes[1].writable else None ) + self._known_route_paths: dict[str, Path] = {} + for probe in service_probes: + if probe.service != "syncthing": + continue + for route in probe.routes: + api_path = ( + config.syncthing.api_root + / PurePosixPath(route.local_relative_path) + ).as_posix() + self._known_route_paths[route.route_id] = ( + config.syncthing.roots.api_to_local(api_path) + ) self.store = ClientStore(config.state_db) self.backups = SQLiteBackupManager( config.state_db, config.backup_dir, config.backup ) self._lease = DatabaseLease(config.state_db) self._active_route_commands: set[str] = set() + self._active_job_commands: set[str] = set() + self._job_execution_lock = asyncio.Lock() + self.jobs = ( + ClientJobExecutor( + client_id=config.client_id, + qbittorrent=resource_reader, + store=self.store, + qb_root=config.qbittorrent.local_root, + qb_api_root=config.qbittorrent.api_root, + route_path=self._route_path, + syncthing_transport=self.routes.transport, + sparse_supported=all(probe.sparse_files for probe in probes), + ) + if resource_reader is not None and self.routes is not None + else None + ) async def run(self) -> None: await asyncio.to_thread(self._lease.acquire) @@ -169,7 +199,7 @@ class ArchiveClientDaemon: writer = asyncio.create_task(self._writer(websocket, outbound)) command_tasks: set[asyncio.Task[None]] = set() try: - await self._resume_route_commands(outbound, command_tasks) + await self._resume_commands(outbound, command_tasks) async for frame in websocket: await self._handle(decode(frame), outbound, command_tasks) finally: @@ -382,12 +412,25 @@ class ArchiveClientDaemon: outbound, command_tasks, ) + elif ( + accepted is not None + and accepted_for_execution + and command.WhichOneof("payload") in {"assign_job", "execute_step"} + and self.jobs is not None + ): + self._schedule_job_command( + command, + envelope.message_id, + outbound, + command_tasks, + ) - async def _resume_route_commands( + async def _resume_commands( self, outbound: asyncio.Queue[str], command_tasks: set[asyncio.Task[None]], ) -> None: + job_commands: list[control_pb2.Command] = [] for row in await asyncio.to_thread(self.store.list_accepted_commands): acknowledgement = decode_message( str(row["acknowledgement_json"]), control_pb2.CommandAck() @@ -401,6 +444,120 @@ class ArchiveClientDaemon: self._schedule_route_command( command, "", outbound, command_tasks ) + elif ( + command.WhichOneof("payload") in {"assign_job", "execute_step"} + and self.jobs is not None + ): + job_commands.append(command) + if job_commands: + task = asyncio.create_task( + self._resume_job_commands(job_commands, outbound), + name="resume-job-commands", + ) + command_tasks.add(task) + task.add_done_callback( + lambda completed: self._command_finished( + completed, command_tasks + ) + ) + + async def _resume_route_commands( + self, + outbound: asyncio.Queue[str], + command_tasks: set[asyncio.Task[None]], + ) -> None: + """Backward-compatible test hook.""" + + await self._resume_commands(outbound, command_tasks) + + def _schedule_job_command( + self, + command: control_pb2.Command, + correlation_id: str, + outbound: asyncio.Queue[str], + command_tasks: set[asyncio.Task[None]], + ) -> None: + if command.command_id in self._active_job_commands: + return + self._active_job_commands.add(command.command_id) + task = asyncio.create_task( + self._execute_job_command(command, correlation_id, outbound), + name=f"job-command-{command.command_id}", + ) + command_tasks.add(task) + task.add_done_callback( + lambda completed: self._job_command_finished( + command.command_id, completed, command_tasks + ) + ) + + async def _resume_job_commands( + self, + commands: list[control_pb2.Command], + outbound: asyncio.Queue[str], + ) -> None: + for command in commands: + if command.command_id in self._active_job_commands: + continue + self._active_job_commands.add(command.command_id) + try: + await self._execute_job_command(command, "", outbound) + except asyncio.CancelledError: + raise + except Exception as error: + logger.error( + "background_command_failed", + extra={ + "error_type": type(error).__name__, + "error_detail": str(error), + }, + ) + finally: + self._active_job_commands.discard(command.command_id) + + def _job_command_finished( + self, + command_id: str, + task: asyncio.Task[None], + command_tasks: set[asyncio.Task[None]], + ) -> None: + self._active_job_commands.discard(command_id) + self._command_finished(task, command_tasks) + + async def _execute_job_command( + self, + command: control_pb2.Command, + correlation_id: str, + outbound: asyncio.Queue[str], + ) -> None: + async with self._job_execution_lock: + await self._execute_job_command_locked( + command, correlation_id, outbound + ) + + async def _execute_job_command_locked( + self, + command: control_pb2.Command, + correlation_id: str, + outbound: asyncio.Queue[str], + ) -> None: + assert self.jobs is not None + payload = command.WhichOneof("payload") + if payload == "assign_job": + events = await asyncio.to_thread( + self.jobs.assign, command.assign_job + ) + elif payload == "execute_step": + events = await asyncio.to_thread( + self.jobs.execute, command.execute_step + ) + else: + raise JobExecutionError("job command payload is unsupported") + for event in events: + response = new_envelope() + response.correlation_id = correlation_id + response.job_event.CopyFrom(event) + await outbound.put(encode(response)) def _schedule_route_command( self, @@ -495,6 +652,7 @@ class ArchiveClientDaemon: configured = await asyncio.to_thread( self.routes.configure, spec, deadline ) + self._known_route_paths[spec.route_id] = configured.local_path await asyncio.to_thread( self.store.record_route_ownership, command.command_id, @@ -658,7 +816,10 @@ class ArchiveClientDaemon: if error is not None: logger.error( "background_command_failed", - extra={"error_type": type(error).__name__}, + extra={ + "error_type": type(error).__name__, + "error_detail": str(error), + }, ) def _initial_acknowledgement( @@ -717,6 +878,46 @@ class ArchiveClientDaemon: acknowledgement.error.message = "ensure route specification is invalid" else: acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED + elif command.WhichOneof("payload") == "assign_job": + definition = command.assign_job.job + if self.jobs is None: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_UNAVAILABLE + acknowledgement.error.message = "job executor is unavailable" + elif ( + not definition.job_id + or definition.WhichOneof("spec") != "transfer" + or self.config.client_id not in { + definition.transfer.source_client_id, + definition.transfer.target_client_id, + } + ): + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT + acknowledgement.error.message = "job assignment is invalid" + else: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED + elif command.WhichOneof("payload") == "execute_step": + step = command.execute_step + if self.jobs is None: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_UNAVAILABLE + acknowledgement.error.message = "job executor is unavailable" + elif ( + not step.job_id + or step.step not in { + job_pb2.JOB_STEP_KIND_SOURCE_STAGE, + job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER, + job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE, + job_pb2.JOB_STEP_KIND_QB_VERIFY, + job_pb2.JOB_STEP_KIND_STAGING_CLEANUP, + } + ): + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT + acknowledgement.error.message = "job step is invalid" + else: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED else: acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED acknowledgement.error.code = common_pb2.ERROR_CODE_UNSUPPORTED @@ -725,6 +926,12 @@ class ArchiveClientDaemon: ) return acknowledgement + def _route_path(self, route_id: str) -> Path: + try: + return self._known_route_paths[route_id] + except KeyError as exc: + raise JobExecutionError("job route is not configured locally") from exc + def _route_error(exc: Exception) -> tuple[int, bool]: if isinstance(exc, RoutePathConflict): diff --git a/src/archive_clients/jobs.py b/src/archive_clients/jobs.py new file mode 100644 index 0000000..f10c5f3 --- /dev/null +++ b/src/archive_clients/jobs.py @@ -0,0 +1,557 @@ +"""Control-commanded happy-path transfer execution.""" + +from __future__ import annotations + +import hashlib +import time +import uuid +from pathlib import Path, PurePosixPath +from typing import Callable + +from archive_clients.protocol import decode_message, encode_message +from archive_clients.qbittorrent import ( + QBittorrentDownloadAttempt, + QBittorrentReader, +) +from archive_clients.resources import NormalizedResource +from archive_clients.state import ClientStore +from archive_clients.syncthing import SyncthingTransferObserver +from archive_clients.transfer import ( + TransferError, + TransferIntegrityError, + cleanup_transfer, + load_published_transfer, + materialize_transfer, + stage_transfer, +) +from archive_control.v1 import ( + common_pb2, + control_pb2, + job_pb2, + resource_pb2, + transfer_pb2, +) + + +class JobExecutionError(RuntimeError): + pass + + +class ClientJobExecutor: + def __init__( + self, + *, + client_id: str, + qbittorrent: QBittorrentReader, + store: ClientStore, + qb_root: Path, + qb_api_root: PurePosixPath, + route_path: Callable[[str], Path], + syncthing_transport: object, + sparse_supported: bool, + poll_interval: float = 1, + ): + self.client_id = client_id + self.qbittorrent = qbittorrent + self.store = store + self.qb_root = qb_root + self.qb_api_root = qb_api_root + self.route_path = route_path + self.syncthing_transport = syncthing_transport + self.sparse_supported = sparse_supported + self.poll_interval = poll_interval + + def assign( + self, command: control_pb2.AssignJobCommand + ) -> list[control_pb2.JobEvent]: + definition = command.job + self._validate_definition(definition) + replay = self._replay( + definition.job_id, command.expected_last_event_sequence + ) + if replay: + return replay + event = self._event( + definition, + sequence=command.expected_last_event_sequence + 1, + revision=command.expected_job_revision, + event_type=control_pb2.JOB_EVENT_TYPE_ASSIGNED, + state=job_pb2.JOB_STATE_PREPARING, + committed=False, + ) + self._record(definition, event) + return [event] + + def execute( + self, command: control_pb2.ExecuteStepCommand + ) -> list[control_pb2.JobEvent]: + definition = self._definition(command.job_id) + replay = self._replay( + command.job_id, command.expected_last_event_sequence + ) + if replay and replay[-1].type in { + control_pb2.JOB_EVENT_TYPE_STEP_SUCCEEDED, + control_pb2.JOB_EVENT_TYPE_COMMITTED, + control_pb2.JOB_EVENT_TYPE_SUCCEEDED, + control_pb2.JOB_EVENT_TYPE_FAILED, + control_pb2.JOB_EVENT_TYPE_CLEANUP_REQUIRED, + control_pb2.JOB_EVENT_TYPE_CANCELLED, + }: + return replay + started = replay[0] if replay else self._event( + definition, + sequence=command.expected_last_event_sequence + 1, + revision=command.expected_job_revision + 1, + event_type=control_pb2.JOB_EVENT_TYPE_STEP_STARTED, + state=job_pb2.JOB_STATE_RUNNING, + committed=( + self._committed(command.job_id) + or command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP + ), + step=command.step, + step_state=job_pb2.STEP_STATE_RUNNING, + ) + if not replay: + self._record(definition, started) + try: + result = self._execute_step(definition, command.step) + except Exception as error: + cleanup = ( + command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP + and started.committed + ) + failed = self._event( + definition, + sequence=started.sequence + 1, + revision=started.job_revision + 1, + event_type=( + control_pb2.JOB_EVENT_TYPE_CLEANUP_REQUIRED + if cleanup else control_pb2.JOB_EVENT_TYPE_FAILED + ), + state=( + job_pb2.JOB_STATE_CLEANUP_REQUIRED + if cleanup else job_pb2.JOB_STATE_FAILED + ), + committed=started.committed, + step=command.step, + step_state=job_pb2.STEP_STATE_FAILED, + ) + failed.error.code = _job_error_code(error) + failed.error.message = str(error) or type(error).__name__ + failed.error.retryable = cleanup + self._record(definition, failed) + return [started, failed] + sequence = started.sequence + 1 + revision = started.job_revision + 1 + committed = started.committed + event_type = control_pb2.JOB_EVENT_TYPE_STEP_SUCCEEDED + state = job_pb2.JOB_STATE_RUNNING + if command.step == job_pb2.JOB_STEP_KIND_QB_VERIFY: + committed = True + event_type = control_pb2.JOB_EVENT_TYPE_COMMITTED + elif command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP: + event_type = control_pb2.JOB_EVENT_TYPE_SUCCEEDED + state = job_pb2.JOB_STATE_SUCCEEDED + succeeded = self._event( + definition, + sequence=sequence, + revision=revision, + event_type=event_type, + state=state, + committed=committed, + step=command.step, + step_state=job_pb2.STEP_STATE_SUCCEEDED, + ) + if result is not None: + succeeded.observed_placement.CopyFrom(result) + self._record(definition, succeeded) + return [started, succeeded] + + def _execute_step( + self, definition: job_pb2.JobDefinition, step: int + ): + if step == job_pb2.JOB_STEP_KIND_SOURCE_STAGE: + self._source_stage(definition) + return None + if step == job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER: + self._wait_for_syncthing(definition) + return None + if step == job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE: + self._target_materialize(definition) + return None + if step == job_pb2.JOB_STEP_KIND_QB_VERIFY: + return self._qb_verify(definition) + if step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP: + cleanup_transfer(self._job_directory(definition)) + temporary_metainfo = self._metainfo_path(definition.job_id) + if temporary_metainfo.exists(): + temporary_metainfo.unlink() + for directory in ( + temporary_metainfo.parent, + temporary_metainfo.parent.parent, + ): + try: + directory.rmdir() + except OSError: + break + return None + raise JobExecutionError("job step is unsupported") + + def _source_stage(self, definition: job_pb2.JobDefinition) -> None: + if definition.transfer.source_client_id != self.client_id: + raise JobExecutionError("source stage was sent to the wrong client") + resource = self._resource(definition) + indices = _selection_indices(definition.transfer.transfer_delta_files) + by_index = {item.file_index: item for item in resource.files} + if not indices or any(index not in by_index for index in indices): + raise JobExecutionError("transfer delta file selection is invalid") + if any( + not by_index[index].selected + or by_index[index].completed_bytes != by_index[index].logical_bytes + for index in indices + ): + raise JobExecutionError("transfer delta contains unavailable files") + manifest = transfer_pb2.TransferManifest( + manifest_version=1, + job_id=definition.job_id, + source_client_id=definition.transfer.source_client_id, + target_client_id=definition.transfer.target_client_id, + route_id=definition.transfer.route_id, + ) + manifest.resource_id.CopyFrom(definition.resource_id) + manifest.requested_files.CopyFrom(definition.transfer.requested_files) + manifest.target_baseline_files.CopyFrom( + definition.transfer.target_baseline_files + ) + manifest.transfer_delta_files.CopyFrom( + definition.transfer.transfer_delta_files + ) + manifest.source_fingerprint.CopyFrom( + definition.transfer.source_fingerprint + ) + if definition.transfer.HasField("target_baseline_fingerprint"): + manifest.target_baseline_fingerprint.CopyFrom( + definition.transfer.target_baseline_fingerprint + ) + manifest.created_at.GetCurrentTime() + for index in indices: + item = by_index[index] + manifest.files.add( + file_index=index, + payload_relative_path=f"payload/{item.canonical_path}", + target_canonical_path=item.canonical_path, + logical_bytes=item.logical_bytes, + ) + metainfo_path = self._metainfo_path(definition.job_id) + metainfo_path.parent.mkdir(parents=True, exist_ok=True) + if not metainfo_path.exists(): + metainfo_path.write_bytes(resource.metainfo_bytes) + artifact = manifest.artifacts.add( + kind=transfer_pb2.ARTIFACT_KIND_TORRENT_FILE, + payload_relative_path="metainfo/source.torrent", + logical_bytes=len(resource.metainfo_bytes), + format="application/x-bittorrent", + sha256_hex=hashlib.sha256(resource.metainfo_bytes).hexdigest(), + ) + del artifact + stage_transfer( + manifest, + source_root=self.qb_root, + sync_root=self.route_path(definition.transfer.route_id), + store=self.store, + artifact_sources={"metainfo/source.torrent": metainfo_path}, + sparse_supported=self.sparse_supported, + ) + self._observer(definition).rescan() + + def _wait_for_syncthing(self, definition: job_pb2.JobDefinition) -> None: + if definition.transfer.target_client_id != self.client_id: + raise JobExecutionError( + "Syncthing completion was sent to the wrong client" + ) + observer = self._observer(definition) + while True: + try: + if observer.status().complete: + return + except (FileNotFoundError, TransferIntegrityError): + # Syncthing can expose the per-job directory before the + # ready marker and manifest have arrived atomically as a set. + pass + time.sleep(self.poll_interval) + + def _target_materialize(self, definition: job_pb2.JobDefinition) -> None: + if definition.transfer.target_client_id != self.client_id: + raise JobExecutionError( + "target materialization was sent to the wrong client" + ) + published = load_published_transfer(self._job_directory(definition)) + materialize_transfer( + published, + target_root=self.qb_root, + store=self.store, + sparse_supported=self.sparse_supported, + ) + + def _qb_verify(self, definition: job_pb2.JobDefinition): + if definition.transfer.target_client_id != self.client_id: + raise JobExecutionError("qB verification was sent to the wrong client") + published = load_published_transfer(self._job_directory(definition)) + metainfo_path = ( + published.job_directory / "metainfo" / "source.torrent" + ) + info_hash = ( + definition.resource_id.info_hash_v1_hex + or definition.resource_id.info_hash_v2_hex + ) + resource = self.qbittorrent.get_resource(info_hash) + should_start = ( + resource is None + or resource.summary.runtime_state + != resource_pb2.TORRENT_RUNTIME_STATE_STOPPED + ) + if resource is None: + self.qbittorrent.add_stopped_with_retry( + metainfo_path.read_bytes(), + self.qb_api_root.as_posix(), + info_hash, + ) + target_union = _union_selection( + definition.transfer.target_baseline_files, + definition.transfer.transfer_delta_files, + ) + selected = _selection_indices(target_union) + total_files = len( + decode_message_metainfo(metainfo_path).files + ) + self.qbittorrent.set_selection(info_hash, selected, total_files) + self.qbittorrent.recheck_and_wait( + info_hash, selected, timeout=30 * 60, poll_interval=self.poll_interval + ) + if should_start: + self.qbittorrent.start(info_hash) + verified = self.qbittorrent.get_resource(info_hash) + if verified is None: + raise JobExecutionError( + "verified qBittorrent resource disappeared" + ) + placement = resource_pb2.Placement( + client_id=self.client_id, + state=resource_pb2.PLACEMENT_STATE_PRESENT, + generation=( + definition.transfer.expected_target_placement_generation + 1 + if definition.transfer.HasField( + "expected_target_placement_generation" + ) + else 1 + ), + created_by_job_id=definition.job_id, + ) + placement.resource_id.CopyFrom(definition.resource_id) + placement.verified_files.CopyFrom(target_union) + metainfo = decode_message_metainfo(metainfo_path) + placement.verified_logical_bytes = sum( + metainfo.files[index].logical_bytes for index in selected + ) + placement.fingerprint.CopyFrom( + _resource_fingerprint(verified, self.client_id) + ) + placement.verified_at.GetCurrentTime() + return placement + + def _observer( + self, definition: job_pb2.JobDefinition + ) -> SyncthingTransferObserver: + return SyncthingTransferObserver( + self.syncthing_transport, + definition.transfer.route_id, + f".archive-control/jobs/{definition.job_id}", + self._job_directory(definition), + ) + + def _job_directory(self, definition: job_pb2.JobDefinition) -> Path: + return ( + self.route_path(definition.transfer.route_id) + / ".archive-control" / "jobs" / definition.job_id + ) + + def _metainfo_path(self, job_id: str) -> Path: + return self.store.database.parent / "artifacts" / job_id / "source.torrent" + + def _resource(self, definition: job_pb2.JobDefinition) -> NormalizedResource: + for info_hash in ( + definition.resource_id.info_hash_v1_hex, + definition.resource_id.info_hash_v2_hex, + ): + if info_hash: + resource = self.qbittorrent.get_resource(info_hash) + if resource is not None: + return resource + raise JobExecutionError("source resource is not present") + + def _definition(self, job_id: str) -> job_pb2.JobDefinition: + rows = self.store.job_snapshot_rows([job_id]) + if not rows: + raise JobExecutionError("job is not assigned") + return decode_message( + str(rows[0]["definition_json"]), job_pb2.JobDefinition() + ) + + def _committed(self, job_id: str) -> bool: + rows = self.store.job_snapshot_rows([job_id]) + return bool(rows and rows[0]["committed"]) + + def _replay( + self, job_id: str, after_sequence: int + ) -> list[control_pb2.JobEvent]: + return [ + decode_message(str(row["event_json"]), control_pb2.JobEvent()) + for row in self.store.job_event_rows(job_id, after_sequence) + ] + + def _record( + self, + definition: job_pb2.JobDefinition, + event: control_pb2.JobEvent, + ) -> None: + self.store.record_job_event( + job_id=definition.job_id, + definition_json=encode_message(definition), + event_id=event.event_id, + event_json=encode_message(event), + state=job_pb2.JobState.Name(event.state), + revision=event.job_revision, + sequence=event.sequence, + committed=event.committed, + ) + + @staticmethod + def _event( + definition: job_pb2.JobDefinition, + *, + sequence: int, + revision: int, + event_type: int, + state: int, + committed: bool, + step: int = job_pb2.JOB_STEP_KIND_UNSPECIFIED, + step_state: int = job_pb2.STEP_STATE_UNSPECIFIED, + ) -> control_pb2.JobEvent: + event = control_pb2.JobEvent( + event_id=str(uuid.uuid4()), + job_id=definition.job_id, + sequence=sequence, + job_revision=revision, + type=event_type, + state=state, + committed=committed, + ) + event.occurred_at.GetCurrentTime() + if step != job_pb2.JOB_STEP_KIND_UNSPECIFIED: + event.progress.step = step + event.progress.state = step_state + event.progress.fraction_complete = ( + 1 if step_state == job_pb2.STEP_STATE_SUCCEEDED else 0 + ) + event.progress.overall_fraction_complete = _overall(step, step_state) + event.progress.display_step_number = _step_number(step) + event.progress.display_step_total = 5 + event.progress.last_progress_at.CopyFrom(event.occurred_at) + return event + + @staticmethod + def _validate_definition(definition: job_pb2.JobDefinition) -> None: + if ( + not definition.job_id + or definition.operation not in { + job_pb2.JOB_OPERATION_ARCHIVE, + job_pb2.JOB_OPERATION_UNARCHIVE, + } + or definition.WhichOneof("spec") != "transfer" + ): + raise JobExecutionError("transfer job definition is invalid") + + +def _selection_indices(selection) -> list[int]: + result: list[int] = [] + previous = -1 + for item in selection.ranges: + if item.first > item.last or item.first <= previous: + raise JobExecutionError("selection ranges are invalid") + result.extend(range(item.first, item.last + 1)) + previous = item.last + return result + + +def _union_selection(*selections) -> resource_pb2.SelectionSet: + indices = sorted({ + index + for selection in selections + for index in _selection_indices(selection) + }) + result = resource_pb2.SelectionSet() + if not indices: + return result + first = previous = indices[0] + for index in indices[1:]: + if index == previous + 1: + previous = index + continue + result.ranges.add(first=first, last=previous) + first = previous = index + result.ranges.add(first=first, last=previous) + return result + + +def _step_number(step: int) -> int: + return { + job_pb2.JOB_STEP_KIND_SOURCE_STAGE: 1, + job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER: 2, + job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE: 3, + job_pb2.JOB_STEP_KIND_QB_VERIFY: 4, + job_pb2.JOB_STEP_KIND_STAGING_CLEANUP: 5, + }.get(step, 0) + + +def _overall(step: int, state: int) -> float: + number = _step_number(step) + return max(0, number - (0 if state == job_pb2.STEP_STATE_SUCCEEDED else 1)) / 5 + + +def decode_message_metainfo(path: Path): + from archive_clients.bencode import decode_metainfo + + return decode_metainfo(path.read_bytes()) + + +def _resource_fingerprint( + resource: NormalizedResource, client_id: str +) -> resource_pb2.ResourceStateFingerprint: + summary = resource.summary + fingerprint = resource_pb2.ResourceStateFingerprint( + client_id=client_id, + qb_torrent_id=summary.qb_torrent_id, + content_revision=summary.content_revision, + runtime_state=summary.runtime_state, + ) + fingerprint.resource_id.CopyFrom(summary.resource_id) + fingerprint.selected_files.CopyFrom(summary.selected_files) + fingerprint.selected_complete_files.CopyFrom( + summary.selected_complete_files + ) + fingerprint.observed_at.CopyFrom(summary.observed_at) + return fingerprint + + +def _job_error_code(error: Exception) -> int: + if isinstance(error, QBittorrentDownloadAttempt): + return common_pb2.ERROR_CODE_MANUAL_INTERVENTION_REQUIRED + if isinstance(error, TransferIntegrityError): + return common_pb2.ERROR_CODE_INTEGRITY_CHECK_FAILED + if isinstance(error, PermissionError): + return common_pb2.ERROR_CODE_PERMISSION_DENIED + if isinstance(error, JobExecutionError): + return common_pb2.ERROR_CODE_PRECONDITION_FAILED + if isinstance(error, TransferError): + return common_pb2.ERROR_CODE_PATH_CONFLICT + return common_pb2.ERROR_CODE_INTERNAL diff --git a/src/archive_clients/qbittorrent.py b/src/archive_clients/qbittorrent.py index 2c1f831..1a5d8fb 100644 --- a/src/archive_clients/qbittorrent.py +++ b/src/archive_clients/qbittorrent.py @@ -126,6 +126,37 @@ class QBittorrentReader: ) _require_mutation_success(response, "add torrent") + def add_stopped_with_retry( + self, + metainfo: bytes, + save_path: str, + torrent_hash: str, + *, + max_attempts: int = 3, + initial_delay: float = 1, + ) -> None: + if max_attempts < 1 or initial_delay < 0: + raise QBittorrentError("torrent add retry values are invalid") + delay = initial_delay + last_error: QBittorrentError | None = None + for attempt in range(1, max_attempts + 1): + try: + self.add_stopped(metainfo, save_path) + self.wait_until_present( + torrent_hash, timeout=15, poll_interval=0.1 + ) + return + except QBittorrentError as error: + last_error = error + if self._torrent_is_present(torrent_hash): + return + if attempt == max_attempts: + break + time.sleep(delay) + delay *= 2 + assert last_error is not None + raise last_error + def set_selection( self, torrent_hash: str, @@ -141,7 +172,9 @@ class QBittorrentReader: "/api/v2/torrents/filePrio", { "hash": torrent_hash, - "id": f"0-{total_file_count - 1}", + "id": "|".join( + str(index) for index in range(total_file_count) + ), "priority": "0", }, ) @@ -166,6 +199,53 @@ class QBittorrentReader: "/api/v2/torrents/pause", {"hashes": torrent_hash} ) + def start(self, torrent_hash: str) -> None: + try: + self._post_form( + "/api/v2/torrents/start", {"hashes": torrent_hash} + ) + except QBittorrentHttpError as exc: + if exc.status != 404: + raise + self._post_form( + "/api/v2/torrents/resume", {"hashes": torrent_hash} + ) + + def wait_until_present( + self, + torrent_hash: str, + *, + timeout: float, + poll_interval: float = 0.1, + ) -> None: + if timeout <= 0 or poll_interval < 0: + raise QBittorrentError("torrent lookup timing values are invalid") + deadline = time.monotonic() + timeout + while True: + if self._torrent_is_present(torrent_hash): + return + if time.monotonic() >= deadline: + raise QBittorrentError( + "qBittorrent did not expose the added torrent in time" + ) + time.sleep( + min(poll_interval, max(0, deadline - time.monotonic())) + ) + + def _torrent_is_present(self, torrent_hash: str) -> bool: + torrents = self._json( + "/api/v2/torrents/info", {"hashes": torrent_hash} + ) + if not isinstance(torrents, list): + raise QBittorrentError( + "qBittorrent lookup response is invalid" + ) + return any( + isinstance(item, dict) + and str(item.get("hash", "")).lower() == torrent_hash.lower() + for item in torrents + ) + def delete_entry(self, torrent_hash: str) -> None: self._post_form( "/api/v2/torrents/delete", @@ -430,7 +510,7 @@ class QBittorrentReader: def _is_download_state(state: str) -> bool: - return state not in {"checkingDL"} and ( + return state not in {"checkingDL", "stoppedDL", "pausedDL"} and ( state.endswith("DL") or state in {"downloading", "metaDL", "forcedMetaDL"} ) diff --git a/src/archive_clients/resources.py b/src/archive_clients/resources.py index 8b5dadb..1dcd54a 100644 --- a/src/archive_clients/resources.py +++ b/src/archive_clients/resources.py @@ -22,6 +22,7 @@ class NormalizedResource: summary: resource_pb2.ResourceSummary files: tuple[resource_pb2.TorrentFile, ...] metainfo: Metainfo + metainfo_bytes: bytes = b"" def build_content_tree( @@ -162,7 +163,7 @@ def normalize_resource( revision_data, sort_keys=True, separators=(",", ":"), ).encode("utf-8")).hexdigest() summary.observed_at.FromDatetime(observed_at) - return NormalizedResource(summary, tuple(files), metainfo) + return NormalizedResource(summary, tuple(files), metainfo, metainfo_bytes) def _set_selection(target: Any, indices: list[int]) -> None: diff --git a/src/archive_clients/state.py b/src/archive_clients/state.py index 358cf6e..6b71dd8 100644 --- a/src/archive_clients/state.py +++ b/src/archive_clients/state.py @@ -268,6 +268,113 @@ class ClientStore: ).fetchall() return [dict(row) for row in rows] + def record_job_event( + self, + *, + job_id: str, + definition_json: str, + event_id: str, + event_json: str, + state: str, + revision: int, + sequence: int, + committed: bool, + ) -> dict[str, object]: + definition = _canonical(json.loads(definition_json)) + event = _canonical(json.loads(event_json)) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing_event = connection.execute( + """ + SELECT job_id, sequence, event_json + FROM events WHERE event_id = ? + """, + (event_id,), + ).fetchone() + if existing_event: + if ( + existing_event["job_id"] != job_id + or existing_event["sequence"] != sequence + or existing_event["event_json"] != event + ): + raise JobConflict( + "job event ID was reused with different content" + ) + return dict(existing_event) + existing_sequence = connection.execute( + """ + SELECT event_id, event_json FROM events + WHERE job_id = ? AND sequence = ? + """, + (job_id, sequence), + ).fetchone() + if existing_sequence: + if existing_sequence["event_json"] != event: + raise JobConflict( + "job event sequence has conflicting content" + ) + return dict(existing_sequence) + job = connection.execute( + """ + SELECT definition_json, revision, last_event_sequence, + committed + FROM jobs WHERE job_id = ? + """, + (job_id,), + ).fetchone() + if job and job["definition_json"] != definition: + raise JobConflict("job definition is immutable") + if job and ( + revision < job["revision"] + or sequence <= job["last_event_sequence"] + or (job["committed"] and not committed) + ): + raise JobConflict("job event cursor cannot move backwards") + connection.execute( + """ + INSERT INTO jobs ( + job_id, definition_json, state, revision, + last_event_sequence, committed + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(job_id) DO UPDATE SET + state = excluded.state, + revision = excluded.revision, + last_event_sequence = excluded.last_event_sequence, + committed = excluded.committed + """, + ( + job_id, definition, state, revision, sequence, + int(committed), + ), + ) + connection.execute( + """ + INSERT INTO events ( + event_id, job_id, sequence, event_json + ) VALUES (?, ?, ?, ?) + """, + (event_id, job_id, sequence, event), + ) + row = connection.execute( + "SELECT * FROM events WHERE event_id = ?", (event_id,) + ).fetchone() + return dict(row) + + def job_event_rows( + self, job_id: str, after_sequence: int = 0 + ) -> list[dict[str, object]]: + with self._connect() as connection: + rows = connection.execute( + """ + SELECT event_id, job_id, sequence, event_json + FROM events + WHERE job_id = ? AND sequence > ? + ORDER BY sequence + """, + (job_id, after_sequence), + ).fetchall() + return [dict(row) for row in rows] + def begin_route_attempt( self, command_id: str, diff --git a/src/archive_clients/transfer.py b/src/archive_clients/transfer.py index 29d49d0..82c1a37 100644 --- a/src/archive_clients/transfer.py +++ b/src/archive_clients/transfer.py @@ -387,6 +387,50 @@ def materialize_transfer( return result_manifest +def cleanup_transfer(job_directory: Path) -> bool: + """Remove only manifest-declared files and known job metadata.""" + + try: + published = load_published_transfer(job_directory) + except FileNotFoundError: + return False + paths = [ + job_directory / _relative_path(entry.payload_relative_path) + for entry in published.manifest.files + ] + paths.extend( + job_directory / _relative_path(artifact.payload_relative_path) + for artifact in published.manifest.artifacts + ) + paths.extend((published.ready_path, published.manifest_path)) + directories: set[Path] = set() + for path in paths: + try: + metadata = path.lstat() + except FileNotFoundError: + continue + if not stat.S_ISREG(metadata.st_mode): + raise TransferIntegrityError( + "job-owned cleanup path is not a regular file" + ) + path.unlink() + parent = path.parent + while parent != job_directory.parent: + directories.add(parent) + if parent == job_directory: + break + parent = parent.parent + for directory in sorted( + directories, key=lambda item: len(item.parts), reverse=True + ): + try: + directory.rmdir() + except OSError as exc: + if exc.errno not in {errno.ENOTEMPTY, errno.ENOENT}: + raise + return True + + def canonical_message_json(message: object) -> bytes: value = json_format.MessageToDict( message, diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 39c9a23..4115705 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -344,7 +344,7 @@ class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): observed["rejected"], ( control_pb2.COMMAND_ACK_STATUS_REJECTED, - common_pb2.ERROR_CODE_UNSUPPORTED, + common_pb2.ERROR_CODE_INVALID_ARGUMENT, ), ) self.assertEqual( diff --git a/tests/test_jobs.py b/tests/test_jobs.py new file mode 100644 index 0000000..3dc9c8b --- /dev/null +++ b/tests/test_jobs.py @@ -0,0 +1,255 @@ +import hashlib +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch +from uuid import uuid4 + +from archive_clients.bencode import encode +from archive_clients.jobs import ClientJobExecutor, JobExecutionError +from archive_clients.resources import normalize_resource +from archive_clients.state import ClientStore +from archive_control.v1 import control_pb2, job_pb2 + + +class CompleteSyncthing: + def __init__(self): + self.posts = [] + + def get_json(self, path): + if path.startswith("/rest/db/completion?"): + return {"completion": 100} + if path.startswith("/rest/db/need?"): + return {"progress": [], "queued": [], "rest": []} + raise AssertionError(path) + + def post(self, path): + self.posts.append(path) + + +class ClientJobHappyPathTests(unittest.TestCase): + def test_archive_and_unarchive_five_step_execution(self): + for operation in ( + job_pb2.JOB_OPERATION_ARCHIVE, + job_pb2.JOB_OPERATION_UNARCHIVE, + ): + with self.subTest(operation=operation), tempfile.TemporaryDirectory() as directory: + self._run_transfer(Path(directory), operation) + + def _run_transfer(self, root: Path, operation: int): + source_root = root / "source" + target_root = root / "target" + route_root = root / "route" + source_root.mkdir() + target_root.mkdir() + route_root.mkdir() + content = b"archive-control-happy-path" + (source_root / "fixture.bin").write_bytes(content) + info = { + b"length": len(content), + b"name": b"fixture.bin", + b"piece length": 16384, + b"pieces": hashlib.sha1(content).digest(), + } + metainfo = encode({b"info": info}) + torrent_hash = hashlib.sha1(encode(info)).hexdigest() + resource = normalize_resource( + { + "hash": torrent_hash, + "name": "fixture.bin", + "state": "uploading", + }, + [{ + "index": 0, + "name": "fixture.bin", + "size": len(content), + "completed": len(content), + "priority": 1, + }], + metainfo, + ) + source_id, target_id = ( + ("cache-1", "archive-1") + if operation == job_pb2.JOB_OPERATION_ARCHIVE + else ("archive-1", "cache-1") + ) + definition = job_pb2.JobDefinition( + job_id=str(uuid4()), + idempotency_key=str(uuid4()), + operation=operation, + resource_display_name="fixture.bin", + transfer={ + "source_client_id": source_id, + "target_client_id": target_id, + "route_id": "route-1", + "requested_logical_bytes": len(content), + "transfer_delta_logical_bytes": len(content), + }, + ) + definition.resource_id.info_hash_v1_hex = torrent_hash + definition.created_at.GetCurrentTime() + definition.transfer.requested_files.ranges.add(first=0, last=0) + definition.transfer.transfer_delta_files.ranges.add(first=0, last=0) + + source_store = ClientStore(root / "source.db") + target_store = ClientStore(root / "target.db") + source_store.initialize() + target_store.initialize() + source_qb = Mock() + source_qb.get_resource.return_value = resource + target_qb = Mock() + target_qb.get_resource.side_effect = [None, resource] + syncthing = CompleteSyncthing() + source = ClientJobExecutor( + client_id=source_id, + qbittorrent=source_qb, + store=source_store, + qb_root=source_root, + qb_api_root=Path("/downloads"), + route_path=lambda _: route_root, + syncthing_transport=syncthing, + sparse_supported=True, + poll_interval=0, + ) + target = ClientJobExecutor( + client_id=target_id, + qbittorrent=target_qb, + store=target_store, + qb_root=target_root, + qb_api_root=Path("/downloads"), + route_path=lambda _: route_root, + syncthing_transport=syncthing, + sparse_supported=True, + poll_interval=0, + ) + + source_assigned = source.assign(control_pb2.AssignJobCommand( + job=definition, + expected_job_revision=1, + expected_last_event_sequence=0, + )) + target_assigned = target.assign(control_pb2.AssignJobCommand( + job=definition, + expected_job_revision=1, + expected_last_event_sequence=1, + )) + self.assertEqual(source_assigned[0].sequence, 1) + self.assertEqual(target_assigned[0].sequence, 2) + + cursor_revision = 1 + cursor_sequence = 2 + pipeline = ( + (source, job_pb2.JOB_STEP_KIND_SOURCE_STAGE), + (target, job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER), + (target, job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE), + (target, job_pb2.JOB_STEP_KIND_QB_VERIFY), + (source, job_pb2.JOB_STEP_KIND_STAGING_CLEANUP), + ) + final = None + for executor, step in pipeline: + events = executor.execute(control_pb2.ExecuteStepCommand( + job_id=definition.job_id, + expected_job_revision=cursor_revision, + expected_last_event_sequence=cursor_sequence, + step=step, + attempt=1, + )) + self.assertEqual( + [event.sequence for event in events], + [cursor_sequence + 1, cursor_sequence + 2], + ) + cursor_sequence = events[-1].sequence + cursor_revision = events[-1].job_revision + final = events[-1] + + self.assertEqual((target_root / "fixture.bin").read_bytes(), content) + target_qb.add_stopped_with_retry.assert_called_once_with( + metainfo, "/downloads", torrent_hash + ) + target_qb.set_selection.assert_called_once_with(torrent_hash, [0], 1) + target_qb.recheck_and_wait.assert_called_once() + target_qb.start.assert_called_once_with(torrent_hash) + self.assertIsNotNone(final) + self.assertTrue(final.committed) + self.assertEqual(final.type, control_pb2.JOB_EVENT_TYPE_SUCCEEDED) + self.assertFalse( + (route_root / ".archive-control/jobs" / definition.job_id).exists() + ) + + replay = source.execute(control_pb2.ExecuteStepCommand( + job_id=definition.job_id, + expected_job_revision=cursor_revision - 2, + expected_last_event_sequence=cursor_sequence - 2, + step=job_pb2.JOB_STEP_KIND_STAGING_CLEANUP, + attempt=1, + )) + self.assertEqual(replay[-1].event_id, final.event_id) + + def test_step_failure_is_durable_and_reports_clear_reason(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ClientStore(root / "client.db") + store.initialize() + definition = job_pb2.JobDefinition( + job_id=str(uuid4()), + idempotency_key=str(uuid4()), + operation=job_pb2.JOB_OPERATION_ARCHIVE, + resource_display_name="fixture", + transfer={ + "source_client_id": "cache-1", + "target_client_id": "archive-1", + "route_id": "route-1", + }, + ) + definition.resource_id.info_hash_v1_hex = "a" * 40 + definition.created_at.GetCurrentTime() + executor = ClientJobExecutor( + client_id="cache-1", + qbittorrent=Mock(), + store=store, + qb_root=root, + qb_api_root=Path("/downloads"), + route_path=lambda _: root, + syncthing_transport=Mock(), + sparse_supported=True, + poll_interval=0, + ) + executor.assign(control_pb2.AssignJobCommand( + job=definition, + expected_job_revision=1, + expected_last_event_sequence=0, + )) + command = control_pb2.ExecuteStepCommand( + job_id=definition.job_id, + expected_job_revision=1, + expected_last_event_sequence=1, + step=job_pb2.JOB_STEP_KIND_SOURCE_STAGE, + attempt=1, + ) + with patch.object( + executor, + "_execute_step", + side_effect=JobExecutionError( + "partfile cannot be handled safely" + ), + ): + events = executor.execute(command) + + self.assertEqual( + [event.type for event in events], + [ + control_pb2.JOB_EVENT_TYPE_STEP_STARTED, + control_pb2.JOB_EVENT_TYPE_FAILED, + ], + ) + self.assertEqual(events[-1].state, job_pb2.JOB_STATE_FAILED) + self.assertIn("partfile", events[-1].error.message) + self.assertEqual( + executor.execute(command)[-1].event_id, + events[-1].event_id, + ) + self.assertEqual(store.list_active_job_cursors(), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_qbittorrent.py b/tests/test_qbittorrent.py index f963d85..041d70a 100644 --- a/tests/test_qbittorrent.py +++ b/tests/test_qbittorrent.py @@ -108,6 +108,11 @@ class QBittorrentReaderTests(unittest.TestCase): b"", # stop before recheck b'{"total_downloaded":0}', b"", # recheck + json.dumps([{ + "hash": torrent_hash, "state": "stoppedDL", + }]).encode(), + b'{"total_downloaded":0}', + b'[{"index":0,"progress":0},{"index":1,"progress":0}]', json.dumps([{ "hash": torrent_hash, "state": "checkingUP", }]).encode(), @@ -118,6 +123,7 @@ class QBittorrentReaderTests(unittest.TestCase): }]).encode(), b'{"total_downloaded":0}', b'[{"index":0,"progress":1},{"index":1,"progress":0}]', + b"", # start after successful recheck b"", # entry-only delete ] with tempfile.TemporaryDirectory() as directory: @@ -140,6 +146,7 @@ class QBittorrentReaderTests(unittest.TestCase): result = adapter.recheck_and_wait( torrent_hash, [0], timeout=1, poll_interval=0 ) + adapter.start(torrent_hash) adapter.delete_entry(torrent_hash) self.assertEqual(result.final_state, "stoppedUP") @@ -155,6 +162,7 @@ class QBittorrentReaderTests(unittest.TestCase): for call in calls[2:] if getattr(call, "data", None) ] + self.assertIn("id=0%7C1", form_bodies[0]) self.assertIn("priority=0", form_bodies[0]) self.assertIn("priority=1", form_bodies[1]) self.assertIn("deleteFiles=false", form_bodies[-1]) @@ -182,6 +190,89 @@ class QBittorrentReaderTests(unittest.TestCase): self.assertTrue(opener.calls[-1].full_url.endswith("/torrents/pause")) + def test_start_falls_back_to_qbittorrent_4_resume_endpoint(self): + missing = error.HTTPError( + "http://qb/api/v2/torrents/start", 404, "not found", {}, None + ) + responses = [b"Ok.", missing, b""] + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + password = root / "password" + password.write_text("secret", encoding="utf-8") + os.chmod(password, 0o600) + config = ServiceConfig( + "http://qb", PurePosixPath("/downloads"), root, + username="admin", password_file=password, + ) + opener = _Opener(responses) + with patch( + "archive_clients.qbittorrent.request.build_opener", + return_value=opener, + ): + QBittorrentReader(config).start("a" * 40) + + self.assertTrue(opener.calls[-1].full_url.endswith("/torrents/resume")) + + def test_wait_until_present_polls_until_added_torrent_is_visible(self): + torrent_hash = "a" * 40 + responses = [ + b"Ok.", + b"[]", + json.dumps([{"hash": torrent_hash}]).encode(), + ] + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + password = root / "password" + password.write_text("secret", encoding="utf-8") + os.chmod(password, 0o600) + config = ServiceConfig( + "http://qb", PurePosixPath("/downloads"), root, + username="admin", password_file=password, + ) + opener = _Opener(responses) + with patch( + "archive_clients.qbittorrent.request.build_opener", + return_value=opener, + ): + QBittorrentReader(config).wait_until_present( + torrent_hash, timeout=1, poll_interval=0 + ) + + self.assertEqual(len(opener.calls), 3) + + def test_stopped_add_retries_while_recently_deleted_hash_is_busy(self): + torrent_hash = "a" * 40 + responses = [ + b"Ok.", + b"Fails.", + b"[]", + b"Ok.", + json.dumps([{"hash": torrent_hash}]).encode(), + ] + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + password = root / "password" + password.write_text("secret", encoding="utf-8") + os.chmod(password, 0o600) + config = ServiceConfig( + "http://qb", PurePosixPath("/downloads"), root, + username="admin", password_file=password, + ) + opener = _Opener(responses) + with patch( + "archive_clients.qbittorrent.request.build_opener", + return_value=opener, + ): + QBittorrentReader(config).add_stopped_with_retry( + b"torrent", + "/downloads", + torrent_hash, + max_attempts=2, + initial_delay=0, + ) + + self.assertEqual(len(opener.calls), 5) + if __name__ == "__main__": unittest.main()