diff --git a/e2e/README.md b/e2e/README.md index c8378ad..ac03dc9 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -27,11 +27,12 @@ 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. +cache-1 → archive-1 transfer, covered cache-1 eviction, and +archive-1 → cache-2 unarchive. The scenario creates a deterministic one-file +torrent in the isolated cache qBittorrent, uses fresh preview revisions for all +jobs, waits for the durable five/three/five-step flows, and verifies target +qBittorrent selections, archive retention, safe cache removal, 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 diff --git a/e2e/scenarios/archive_happy.py b/e2e/scenarios/archive_happy.py index ec4961c..34d677b 100755 --- a/e2e/scenarios/archive_happy.py +++ b/e2e/scenarios/archive_happy.py @@ -9,8 +9,6 @@ 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" @@ -103,30 +101,18 @@ def main() -> int: 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) + preview = post_json(f"{CONTROL}/jobs/preview", { + "operation": args.operation, + "source_client_id": args.source_client, + "target_client_id": args.target_client, + "resource_id": {"info_hash_v1_hex": INFO_HASH}, + }) + definition = preview["definition"] + job_id = definition["job_id"] + created = post_json(f"{CONTROL}/jobs", { + "preview_revision": preview["preview_revision"], + "definition": definition, + }) if created["job_id"] != job_id: raise RuntimeError("control returned the wrong job") diff --git a/e2e/scenarios/evict_happy.py b/e2e/scenarios/evict_happy.py new file mode 100644 index 0000000..674f74a --- /dev/null +++ b/e2e/scenarios/evict_happy.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +"""Drive a covered cache eviction through the bot-free HTTP adapter.""" + +from __future__ import annotations + +import json +import sys +import time +import urllib.request + + +CONTROL = "http://127.0.0.1:18081/test/v1" +INFO_HASH = "330be0cb7c2201135a2de63b28e77993745ff688" + + +def get_json(url: str): + with urllib.request.urlopen(url, timeout=35) as response: + return json.load(response) + + +def post_json(url: str, value: object): + request = urllib.request.Request( + url, + data=json.dumps(value, separators=(",", ":")).encode(), + method="POST", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(request, timeout=35) as response: + return json.load(response) + + +def main() -> int: + preview = post_json(f"{CONTROL}/jobs/preview", { + "operation": "evict_cache", + "cache_client_id": "cache-1", + "resource_id": {"info_hash_v1_hex": INFO_HASH}, + }) + definition = preview["definition"] + job_id = definition["job_id"] + post_json(f"{CONTROL}/jobs", { + "preview_revision": preview["preview_revision"], + "definition": definition, + }) + deadline = time.monotonic() + 180 + while time.monotonic() < deadline: + job = get_json(f"{CONTROL}/jobs/{job_id}") + if job["state"] == "JOB_STATE_SUCCEEDED" and job["committed"]: + print(json.dumps({ + "job_id": job_id, + "state": job["state"], + "committed": job["committed"], + }, sort_keys=True)) + return 0 + if job["state"] in {"JOB_STATE_FAILED", "JOB_STATE_CANCELLED"}: + raise RuntimeError(f"eviction did not succeed: {job}") + time.sleep(0.5) + raise RuntimeError("timed out waiting for cache eviction") + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: + print(f"eviction happy-path failed: {exc}", file=sys.stderr) + raise diff --git a/e2e/scripts/archive-happy.sh b/e2e/scripts/archive-happy.sh index 085ac26..8d1b25f 100755 --- a/e2e/scripts/archive-happy.sh +++ b/e2e/scripts/archive-happy.sh @@ -63,6 +63,16 @@ if [[ "$target_digest" != \ fi printf 'archive filesystem content verified: sha256=%s\n' "$target_digest" +compose_control exec -T control \ + python /e2e/scenarios/evict_happy.py +wait_qb_absent cache-1 +if compose_node cache-1 exec -T qbittorrent \ + test -e /downloads/fixture.bin; then + printf 'eviction left the unshared cache file behind\n' >&2 + exit 1 +fi +printf 'covered cache eviction verified\n' + 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" \ diff --git a/examples/client.toml b/examples/client.toml index d27382c..4625b6f 100644 --- a/examples/client.toml +++ b/examples/client.toml @@ -13,6 +13,16 @@ reconnect_max = "60s" reconnect_reset_after = "60s" reconnect_jitter = true +[jobs] +# Mark a job stalled after it has made no progress for this long. +stall_after = "30m" +# Upper bound for qBittorrent's stopped full recheck. +verification_timeout = "30m" +# Poll qBittorrent and Syncthing at this interval while a step is active. +poll_interval = "1s" +# Space that must remain free after a worst-case copy fallback. +free_space_reserve_bytes = 1073741824 + [backup] interval = "6h" recent = 12 diff --git a/src/archive_clients/config.py b/src/archive_clients/config.py index bdb39b3..0377aad 100644 --- a/src/archive_clients/config.py +++ b/src/archive_clients/config.py @@ -80,6 +80,9 @@ class ConnectionConfig: @dataclass(frozen=True) class JobsConfig: stall_after: float = 30 * 60 + verification_timeout: float = 30 * 60 + poll_interval: float = 1 + free_space_reserve_bytes: int = 1024 * 1024 * 1024 @dataclass(frozen=True) @@ -217,8 +220,27 @@ def _connection(value: Any) -> ConnectionConfig: def _jobs(value: Any) -> JobsConfig: if not isinstance(value, dict): raise ConfigError("jobs must be a table") - _keys(value, {"stall_after"}, "jobs") - return JobsConfig(stall_after=_duration(value.get("stall_after", "30m"))) + _keys( + value, + { + "stall_after", + "verification_timeout", + "poll_interval", + "free_space_reserve_bytes", + }, + "jobs", + ) + return JobsConfig( + stall_after=_duration(value.get("stall_after", "30m")), + verification_timeout=_duration( + value.get("verification_timeout", "30m") + ), + poll_interval=_duration(value.get("poll_interval", "1s")), + free_space_reserve_bytes=_positive_int( + value.get("free_space_reserve_bytes", 1024 * 1024 * 1024), + "jobs.free_space_reserve_bytes", + ), + ) def _backup(value: Any) -> BackupConfig: diff --git a/src/archive_clients/daemon.py b/src/archive_clients/daemon.py index ece3a29..b480c19 100644 --- a/src/archive_clients/daemon.py +++ b/src/archive_clients/daemon.py @@ -105,6 +105,11 @@ class ArchiveClientDaemon: route_path=self._route_path, syncthing_transport=self.routes.transport, sparse_supported=all(probe.sparse_files for probe in probes), + poll_interval=config.jobs.poll_interval, + verification_timeout=config.jobs.verification_timeout, + free_space_reserve_bytes=( + config.jobs.free_space_reserve_bytes + ), ) if resource_reader is not None and self.routes is not None else None @@ -415,9 +420,13 @@ class ArchiveClientDaemon: elif ( accepted is not None and accepted_for_execution - and command.WhichOneof("payload") in {"assign_job", "execute_step"} + and command.WhichOneof("payload") in { + "assign_job", "execute_step", "cancel_job" + } and self.jobs is not None ): + if command.WhichOneof("payload") == "cancel_job": + self.jobs.request_cancel(command.cancel_job.job_id) self._schedule_job_command( command, envelope.message_id, @@ -445,9 +454,13 @@ class ArchiveClientDaemon: command, "", outbound, command_tasks ) elif ( - command.WhichOneof("payload") in {"assign_job", "execute_step"} + command.WhichOneof("payload") in { + "assign_job", "execute_step", "cancel_job" + } and self.jobs is not None ): + if command.WhichOneof("payload") == "cancel_job": + self.jobs.request_cancel(command.cancel_job.job_id) job_commands.append(command) if job_commands: task = asyncio.create_task( @@ -543,17 +556,39 @@ class ArchiveClientDaemon: ) -> None: assert self.jobs is not None payload = command.WhichOneof("payload") + streamed = False if payload == "assign_job": events = await asyncio.to_thread( self.jobs.assign, command.assign_job ) elif payload == "execute_step": + loop = asyncio.get_running_loop() + + def emit(event): + response = new_envelope() + response.correlation_id = correlation_id + response.job_event.CopyFrom(event) + future = asyncio.run_coroutine_threadsafe( + outbound.put(encode(response)), loop + ) + try: + future.result(timeout=30) + except Exception: + # The event is already durable in the client DB and will + # be replayed after reconnect. + pass + events = await asyncio.to_thread( - self.jobs.execute, command.execute_step + self.jobs.execute, command.execute_step, emit + ) + streamed = True + elif payload == "cancel_job": + events = await asyncio.to_thread( + self.jobs.cancel, command.cancel_job ) else: raise JobExecutionError("job command payload is unsupported") - for event in events: + for event in (() if streamed else events): response = new_envelope() response.correlation_id = correlation_id response.job_event.CopyFrom(event) @@ -880,17 +915,25 @@ class ArchiveClientDaemon: acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED elif command.WhichOneof("payload") == "assign_job": definition = command.assign_job.job + specification = definition.WhichOneof("spec") + assigned_to_client = ( + specification == "transfer" + and self.config.client_id + in { + definition.transfer.source_client_id, + definition.transfer.target_client_id, + } + ) or ( + specification == "eviction" + and self.config.client_id == definition.eviction.cache_client_id + ) 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, - } + or not assigned_to_client ): acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT @@ -911,6 +954,9 @@ class ArchiveClientDaemon: job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE, job_pb2.JOB_STEP_KIND_QB_VERIFY, job_pb2.JOB_STEP_KIND_STAGING_CLEANUP, + job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE, + job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY, + job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK, } ): acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED diff --git a/src/archive_clients/eviction.py b/src/archive_clients/eviction.py new file mode 100644 index 0000000..faeda58 --- /dev/null +++ b/src/archive_clients/eviction.py @@ -0,0 +1,242 @@ +"""Idempotent cache eviction without recursive or qB data deletion.""" + +from __future__ import annotations + +import errno +import hashlib +import os +import stat +from pathlib import Path, PurePosixPath +from typing import Iterable + +from archive_clients.qbittorrent import QBittorrentReader +from archive_clients.resources import NormalizedResource +from archive_clients.state import ClientStore + + +class EvictionError(RuntimeError): + pass + + +def verify_and_snapshot( + *, + job_id: str, + resource: NormalizedResource, + selected_indices: Iterable[int], + qb_root: Path, + store: ClientStore, +) -> dict[str, object]: + selected = set(selected_indices) + by_index = {item.file_index: item for item in resource.files} + if not selected or any(index not in by_index for index in selected): + raise EvictionError("eviction file selection is invalid") + files = [] + for index in sorted(selected): + item = by_index[index] + if not item.selected or item.completed_bytes != item.logical_bytes: + raise EvictionError( + f"cache file {index} is not selected and complete" + ) + relative = _relative(item.canonical_path) + path = qb_root.joinpath(*relative.parts) + try: + metadata = path.lstat() + except FileNotFoundError as exc: + raise EvictionError( + f"cache file disappeared before eviction: {relative}" + ) from exc + if not stat.S_ISREG(metadata.st_mode): + raise EvictionError( + f"cache path is not a regular file: {relative}" + ) + if metadata.st_size != item.logical_bytes: + raise EvictionError( + f"cache file size changed before eviction: {relative}" + ) + files.append( + { + "file_index": index, + "path": relative.as_posix(), + "logical_bytes": item.logical_bytes, + "device": metadata.st_dev, + "inode": metadata.st_ino, + "mtime_ns": metadata.st_mtime_ns, + "ctime_ns": metadata.st_ctime_ns, + "sha256": _sha256(path), + } + ) + snapshot = {"files": files} + return store.put_job_artifact(job_id, "eviction-snapshot", snapshot)["value"] + + +def remove_qb_entry( + *, + job_id: str, + torrent_hash: str, + qbittorrent: QBittorrentReader, + store: ClientStore, +) -> None: + marker = store.get_job_artifact(job_id, "qb-entry-removed") + if marker is not None: + return + if qbittorrent.get_resource(torrent_hash) is not None: + qbittorrent.delete_entry(torrent_hash) + store.put_job_artifact( + job_id, "qb-entry-removed", {"torrent_hash": torrent_hash} + ) + + +def safe_unlink( + *, + job_id: str, + qb_root: Path, + qbittorrent: QBittorrentReader, + store: ClientStore, +) -> dict[str, object]: + completed = store.get_job_artifact(job_id, "eviction-unlinked") + if completed is not None: + return completed["value"] + snapshot_row = store.get_job_artifact(job_id, "eviction-snapshot") + if snapshot_row is None: + raise EvictionError("eviction snapshot is missing") + snapshot = snapshot_row["value"] + files = snapshot.get("files") + if not isinstance(files, list): + raise EvictionError("eviction snapshot is invalid") + + referenced = _remaining_paths(qbittorrent.list_resources()) + removed: list[str] = [] + retained: list[dict[str, str]] = [] + directories: set[Path] = set() + for record in files: + if not isinstance(record, dict) or not isinstance(record.get("path"), str): + raise EvictionError("eviction file record is invalid") + relative = _relative(record["path"]) + if relative.as_posix() in referenced: + retained.append({"path": relative.as_posix(), "reason": "shared"}) + continue + path = qb_root.joinpath(*relative.parts) + try: + metadata = path.lstat() + except FileNotFoundError: + removed.append(relative.as_posix()) + continue + if not stat.S_ISREG(metadata.st_mode): + retained.append( + {"path": relative.as_posix(), "reason": "type-changed"} + ) + continue + if ( + metadata.st_dev != record.get("device") + or metadata.st_ino != record.get("inode") + or metadata.st_size != record.get("logical_bytes") + or metadata.st_mtime_ns != record.get("mtime_ns") + or metadata.st_ctime_ns != record.get("ctime_ns") + or _sha256(path) != record.get("sha256") + ): + retained.append( + {"path": relative.as_posix(), "reason": "identity-changed"} + ) + continue + path.unlink() + removed.append(relative.as_posix()) + parent = path.parent + while parent != qb_root: + directories.add(parent) + parent = parent.parent + + removed_directories: list[str] = [] + for directory in sorted( + directories, key=lambda item: len(item.parts), reverse=True + ): + try: + directory.rmdir() + removed_directories.append( + directory.relative_to(qb_root).as_posix() + ) + except OSError as exc: + if exc.errno not in {errno.ENOTEMPTY, errno.ENOENT}: + raise + result = { + "removed_files": removed, + "retained_files": retained, + "removed_directories": removed_directories, + } + return store.put_job_artifact(job_id, "eviction-unlinked", result)["value"] + + +def compensate_materialized_files( + *, job_id: str, qb_root: Path, store: ClientStore +) -> list[str]: + """Remove only target files this job created and whose inode still matches.""" + + removed: list[str] = [] + directories: set[Path] = set() + for row in store.file_operation_rows(job_id): + if row["state"] != "completed" or row["result_json"] is None: + continue + import json + + intent = json.loads(str(row["intent_json"])) + result = json.loads(str(row["result_json"])) + if result.get("destination_preexisted"): + continue + relative_value = intent.get("destination") + if not isinstance(relative_value, str): + continue + relative = _relative(relative_value) + path = qb_root.joinpath(*relative.parts) + try: + metadata = path.lstat() + except FileNotFoundError: + continue + if ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_dev != result.get("destination_device") + or metadata.st_ino != result.get("destination_inode") + ): + continue + path.unlink() + removed.append(relative.as_posix()) + parent = path.parent + while parent != qb_root: + directories.add(parent) + 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 removed + + +def _remaining_paths(resources: Iterable[NormalizedResource]) -> set[str]: + return { + item.canonical_path + for resource in resources + for item in resource.files + } + + +def _relative(value: str) -> PurePosixPath: + candidate = PurePosixPath(value) + if ( + not value + or candidate.is_absolute() + or ".." in candidate.parts + or "." in candidate.parts + or "\\" in value + ): + raise EvictionError("eviction path is unsafe") + return candidate + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + while chunk := source.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() diff --git a/src/archive_clients/jobs.py b/src/archive_clients/jobs.py index f10c5f3..40feec3 100644 --- a/src/archive_clients/jobs.py +++ b/src/archive_clients/jobs.py @@ -3,12 +3,22 @@ from __future__ import annotations import hashlib +import json +import shutil +import threading 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.eviction import ( + EvictionError, + compensate_materialized_files, + remove_qb_entry, + safe_unlink, + verify_and_snapshot, +) from archive_clients.qbittorrent import ( QBittorrentDownloadAttempt, QBittorrentReader, @@ -37,6 +47,10 @@ class JobExecutionError(RuntimeError): pass +class JobCancelled(JobExecutionError): + pass + + class ClientJobExecutor: def __init__( self, @@ -50,6 +64,8 @@ class ClientJobExecutor: syncthing_transport: object, sparse_supported: bool, poll_interval: float = 1, + verification_timeout: float = 30 * 60, + free_space_reserve_bytes: int = 1024 * 1024 * 1024, ): self.client_id = client_id self.qbittorrent = qbittorrent @@ -60,6 +76,12 @@ class ClientJobExecutor: self.syncthing_transport = syncthing_transport self.sparse_supported = sparse_supported self.poll_interval = poll_interval + self.verification_timeout = verification_timeout + self.free_space_reserve_bytes = free_space_reserve_bytes + self._cancel_events: dict[str, threading.Event] = {} + + def request_cancel(self, job_id: str) -> None: + self._cancel_events.setdefault(job_id, threading.Event()).set() def assign( self, command: control_pb2.AssignJobCommand @@ -83,7 +105,9 @@ class ClientJobExecutor: return [event] def execute( - self, command: control_pb2.ExecuteStepCommand + self, + command: control_pb2.ExecuteStepCommand, + event_callback: Callable[[control_pb2.JobEvent], None] | None = None, ) -> list[control_pb2.JobEvent]: definition = self._definition(command.job_id) replay = self._replay( @@ -97,6 +121,9 @@ class ClientJobExecutor: control_pb2.JOB_EVENT_TYPE_CLEANUP_REQUIRED, control_pb2.JOB_EVENT_TYPE_CANCELLED, }: + if event_callback is not None: + for event in replay: + event_callback(event) return replay started = replay[0] if replay else self._event( definition, @@ -107,23 +134,124 @@ class ClientJobExecutor: committed=( self._committed(command.job_id) or command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP + or ( + command.step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK + and self.store.get_job_artifact( + command.job_id, "qb-entry-removed" + ) is not None + ) ), step=command.step, step_state=job_pb2.STEP_STATE_RUNNING, ) if not replay: self._record(definition, started) + if event_callback is not None: + event_callback(started) + emitted = [started] + cursor = started + speed_sample = [time.monotonic(), 0] + + def progress( + fraction: float, + bytes_complete: int = 0, + bytes_total: int = 0, + detail: str = "", + ) -> None: + nonlocal cursor + now = time.monotonic() + elapsed = now - float(speed_sample[0]) + if fraction < 1 and elapsed < 1: + return + speed = ( + max(0, bytes_complete - int(speed_sample[1])) / elapsed + if elapsed > 0 else 0 + ) + event = self._event( + definition, + sequence=cursor.sequence + 1, + revision=cursor.job_revision + 1, + event_type=control_pb2.JOB_EVENT_TYPE_PROGRESS, + state=job_pb2.JOB_STATE_RUNNING, + committed=cursor.committed, + step=command.step, + step_state=job_pb2.STEP_STATE_RUNNING, + ) + event.progress.fraction_complete = max( + 0.0, min(float(fraction), 1.0) + ) + event.progress.bytes_complete = max(0, bytes_complete) + event.progress.bytes_total = max(0, bytes_total) + event.progress.approximate_bytes_per_second = speed + event.progress.detail = detail + completed_steps = max(0, _step_number(command.step) - 1) + event.progress.overall_fraction_complete = ( + completed_steps + event.progress.fraction_complete + ) / event.progress.display_step_total + self._record(definition, event) + emitted.append(event) + cursor = event + speed_sample[:] = [now, bytes_complete] + if event_callback is not None: + event_callback(event) try: - result = self._execute_step(definition, command.step) + result = self._execute_step(definition, command.step, progress) + except JobCancelled as error: + cancelling = self._event( + definition, + sequence=cursor.sequence + 1, + revision=cursor.job_revision + 1, + event_type=control_pb2.JOB_EVENT_TYPE_CANCELLING, + state=job_pb2.JOB_STATE_CANCELLING, + committed=started.committed, + step=command.step, + step_state=job_pb2.STEP_STATE_CANCELLED, + ) + cancelling.error.code = common_pb2.ERROR_CODE_CANCELLED + cancelling.error.message = str(error) + self._record(definition, cancelling) + emitted.append(cancelling) + if event_callback is not None: + event_callback(cancelling) + return emitted except Exception as error: + rollback_error: Exception | None = None + if ( + definition.WhichOneof("spec") == "transfer" + and not started.committed + ): + try: + if ( + self.client_id == definition.transfer.target_client_id + and command.step in { + job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE, + job_pb2.JOB_STEP_KIND_QB_VERIFY, + } + ): + self._rollback_target(definition) + elif ( + self.client_id == definition.transfer.source_client_id + and command.step == job_pb2.JOB_STEP_KIND_SOURCE_STAGE + ): + self._cleanup_staging(definition) + except Exception as compensation_error: + rollback_error = compensation_error cleanup = ( - command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP - and started.committed + ( + command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP + and started.committed + ) + or ( + command.step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK + and self.store.get_job_artifact( + definition.job_id, "qb-entry-removed" + ) is not None + ) ) failed = self._event( definition, - sequence=started.sequence + 1, - revision=started.job_revision + 1, + sequence=cursor.sequence + 1, + revision=cursor.job_revision + 1, event_type=( control_pb2.JOB_EVENT_TYPE_CLEANUP_REQUIRED if cleanup else control_pb2.JOB_EVENT_TYPE_FAILED @@ -139,11 +267,23 @@ class ClientJobExecutor: failed.error.code = _job_error_code(error) failed.error.message = str(error) or type(error).__name__ failed.error.retryable = cleanup + if rollback_error is not None: + failed.error.code = ( + common_pb2.ERROR_CODE_MANUAL_INTERVENTION_REQUIRED + ) + failed.error.message = ( + f"{failed.error.message}; compensation also failed: " + f"{rollback_error}" + ) + failed.error.retryable = False self._record(definition, failed) - return [started, failed] - sequence = started.sequence + 1 - revision = started.job_revision + 1 - committed = started.committed + emitted.append(failed) + if event_callback is not None: + event_callback(failed) + return emitted + sequence = cursor.sequence + 1 + revision = cursor.job_revision + 1 + committed = cursor.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: @@ -152,6 +292,10 @@ class ClientJobExecutor: elif command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP: event_type = control_pb2.JOB_EVENT_TYPE_SUCCEEDED state = job_pb2.JOB_STATE_SUCCEEDED + elif command.step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK: + committed = True + event_type = control_pb2.JOB_EVENT_TYPE_SUCCEEDED + state = job_pb2.JOB_STATE_SUCCEEDED succeeded = self._event( definition, sequence=sequence, @@ -165,24 +309,91 @@ class ClientJobExecutor: if result is not None: succeeded.observed_placement.CopyFrom(result) self._record(definition, succeeded) - return [started, succeeded] + emitted.append(succeeded) + if event_callback is not None: + event_callback(succeeded) + return emitted + + def cancel( + self, command: control_pb2.CancelJobCommand + ) -> list[control_pb2.JobEvent]: + definition = self._definition(command.job_id) + self.request_cancel(command.job_id) + replay = self._replay( + command.job_id, command.expected_last_event_sequence + ) + if replay: + return replay + committed = self._committed(command.job_id) + if definition.WhichOneof("spec") == "eviction": + removed = self.store.get_job_artifact( + definition.job_id, "qb-entry-removed" + ) + placement = None + if removed is not None: + placement = self._execute_eviction_step( + definition, job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK + ) + committed = True + event_type = control_pb2.JOB_EVENT_TYPE_CANCELLED + state = job_pb2.JOB_STATE_CANCELLED + elif committed: + if definition.transfer.source_client_id == self.client_id: + self._cleanup_staging(definition) + event_type = control_pb2.JOB_EVENT_TYPE_CANCELLED + state = job_pb2.JOB_STATE_CANCELLED + placement = None + elif definition.transfer.source_client_id == self.client_id: + self._cleanup_staging(definition) + temporary = self._metainfo_path(definition.job_id) + if temporary.exists(): + temporary.unlink() + event_type = control_pb2.JOB_EVENT_TYPE_ROLLBACK_SUCCEEDED + state = job_pb2.JOB_STATE_ROLLING_BACK + placement = None + else: + self._rollback_target(definition) + self._cleanup_staging(definition) + event_type = control_pb2.JOB_EVENT_TYPE_CANCELLED + state = job_pb2.JOB_STATE_CANCELLED + placement = None + event = self._event( + definition, + sequence=command.expected_last_event_sequence + 1, + revision=command.expected_job_revision + 1, + event_type=event_type, + state=state, + committed=committed, + step=job_pb2.JOB_STEP_KIND_ROLLBACK, + step_state=job_pb2.STEP_STATE_SUCCEEDED, + ) + if placement is not None: + event.observed_placement.CopyFrom(placement) + self._record(definition, event) + return [event] def _execute_step( - self, definition: job_pb2.JobDefinition, step: int + self, + definition: job_pb2.JobDefinition, + step: int, + progress: Callable[[float, int, int, str], None], ): + self._raise_if_cancelled(definition.job_id) + if definition.WhichOneof("spec") == "eviction": + return self._execute_eviction_step(definition, step) if step == job_pb2.JOB_STEP_KIND_SOURCE_STAGE: - self._source_stage(definition) + self._source_stage(definition, progress) return None if step == job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER: - self._wait_for_syncthing(definition) + self._wait_for_syncthing(definition, progress) return None if step == job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE: - self._target_materialize(definition) + self._target_materialize(definition, progress) return None if step == job_pb2.JOB_STEP_KIND_QB_VERIFY: - return self._qb_verify(definition) + return self._qb_verify(definition, progress) if step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP: - cleanup_transfer(self._job_directory(definition)) + self._cleanup_staging(definition) temporary_metainfo = self._metainfo_path(definition.job_id) if temporary_metainfo.exists(): temporary_metainfo.unlink() @@ -197,10 +408,86 @@ class ClientJobExecutor: return None raise JobExecutionError("job step is unsupported") - def _source_stage(self, definition: job_pb2.JobDefinition) -> None: + def _execute_eviction_step( + self, definition: job_pb2.JobDefinition, step: int + ): + if definition.eviction.cache_client_id != self.client_id: + raise JobExecutionError("eviction step was sent to the wrong client") + info_hash = _info_hash(definition) + if step == job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE: + covered = { + index + for proof in definition.eviction.archive_coverage + for index in _selection_indices(proof.covered_files) + } + requested = set( + _selection_indices(definition.eviction.files_to_evict) + ) + if not requested or requested - covered: + raise JobExecutionError( + "archive coverage no longer covers the cache selection" + ) + resource = self._resource(definition) + current = _resource_fingerprint(resource, self.client_id) + if not _fingerprint_matches( + current, definition.eviction.cache_fingerprint + ): + raise JobExecutionError( + "cache resource changed after eviction confirmation" + ) + verify_and_snapshot( + job_id=definition.job_id, + resource=resource, + selected_indices=requested, + qb_root=self.qb_root, + store=self.store, + ) + return None + if step == job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY: + remove_qb_entry( + job_id=definition.job_id, + torrent_hash=info_hash, + qbittorrent=self.qbittorrent, + store=self.store, + ) + return None + if step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK: + safe_unlink( + job_id=definition.job_id, + qb_root=self.qb_root, + qbittorrent=self.qbittorrent, + store=self.store, + ) + placement = resource_pb2.Placement( + client_id=self.client_id, + state=resource_pb2.PLACEMENT_STATE_ABSENT, + generation=definition.eviction.cache_placement_generation + 1, + created_by_job_id=definition.job_id, + ) + placement.resource_id.CopyFrom(definition.resource_id) + placement.fingerprint.CopyFrom( + definition.eviction.cache_fingerprint + ) + placement.verified_at.GetCurrentTime() + return placement + raise JobExecutionError("eviction job step is unsupported") + + def _source_stage( + self, + definition: job_pb2.JobDefinition, + progress: Callable[[float, int, int, str], None], + ) -> None: if definition.transfer.source_client_id != self.client_id: raise JobExecutionError("source stage was sent to the wrong client") resource = self._resource(definition) + if not _fingerprint_matches( + _resource_fingerprint(resource, self.client_id), + definition.transfer.source_fingerprint, + ): + raise JobExecutionError( + "source resource changed after job confirmation" + ) + self._reject_unsafe_partfile(definition, resource) 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): @@ -254,6 +541,11 @@ class ClientJobExecutor: sha256_hex=hashlib.sha256(resource.metainfo_bytes).hexdigest(), ) del artifact + self._require_space( + self.route_path(definition.transfer.route_id), + definition.transfer.transfer_delta_logical_bytes + + len(resource.metainfo_bytes), + ) stage_transfer( manifest, source_root=self.qb_root, @@ -261,18 +553,43 @@ class ClientJobExecutor: store=self.store, artifact_sources={"metainfo/source.torrent": metainfo_path}, sparse_supported=self.sparse_supported, + cancel_check=lambda: self._raise_if_cancelled( + definition.job_id + ), + progress=lambda completed, total: progress( + completed / total if total else 1, + completed, + total, + f"staged {completed} of {total} bytes", + ), ) self._observer(definition).rescan() - def _wait_for_syncthing(self, definition: job_pb2.JobDefinition) -> None: + def _wait_for_syncthing( + self, + definition: job_pb2.JobDefinition, + progress: Callable[[float, int, int, str], None], + ) -> None: if definition.transfer.target_client_id != self.client_id: raise JobExecutionError( "Syncthing completion was sent to the wrong client" ) observer = self._observer(definition) + last_reported: tuple[int, int] | None = None while True: + self._raise_if_cancelled(definition.job_id) try: - if observer.status().complete: + status = observer.status() + sample = (status.bytes_complete, status.bytes_total) + if sample != last_reported: + progress( + status.fraction_complete, + status.bytes_complete, + status.bytes_total, + f"{status.needed_items} Syncthing items still needed", + ) + last_reported = sample + if status.complete: return except (FileNotFoundError, TransferIntegrityError): # Syncthing can expose the per-job directory before the @@ -280,20 +597,73 @@ class ClientJobExecutor: pass time.sleep(self.poll_interval) - def _target_materialize(self, definition: job_pb2.JobDefinition) -> None: + def _target_materialize( + self, + definition: job_pb2.JobDefinition, + progress: Callable[[float, int, int, str], None], + ) -> 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)) + self._require_space( + self.qb_root, + definition.transfer.transfer_delta_logical_bytes, + ) + info_hash = _info_hash(definition) + resource = self.qbittorrent.get_resource(info_hash) + if resource is not None: + self._reject_unsafe_partfile(definition, resource) + if definition.transfer.HasField("target_baseline_fingerprint"): + if resource is None or not _fingerprint_matches( + _resource_fingerprint(resource, self.client_id), + definition.transfer.target_baseline_fingerprint, + ): + raise JobExecutionError( + "target resource changed after job confirmation" + ) + elif resource is not None: + raise JobExecutionError( + "target resource appeared after job confirmation" + ) + baseline = { + "present": resource is not None, + "selected_file_indices": ( + _selection_indices(resource.summary.selected_files) + if resource is not None else [] + ), + "stopped": ( + resource is not None + and resource.summary.runtime_state + == resource_pb2.TORRENT_RUNTIME_STATE_STOPPED + ), + "total_file_count": len(published.manifest.files), + } + self.store.put_job_artifact( + definition.job_id, "target-baseline", baseline + ) materialize_transfer( published, target_root=self.qb_root, store=self.store, sparse_supported=self.sparse_supported, + cancel_check=lambda: self._raise_if_cancelled( + definition.job_id + ), + progress=lambda completed, total: progress( + completed / total if total else 1, + completed, + total, + f"materialized {completed} of {total} bytes", + ), ) - def _qb_verify(self, definition: job_pb2.JobDefinition): + def _qb_verify( + self, + definition: job_pb2.JobDefinition, + progress: Callable[[float, int, int, str], None], + ): 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)) @@ -326,7 +696,16 @@ class ClientJobExecutor: ) 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 + info_hash, + selected, + timeout=self.verification_timeout, + poll_interval=self.poll_interval, + cancel_check=lambda: self._raise_if_cancelled( + definition.job_id + ), + progress_callback=lambda fraction: progress( + fraction, 0, 0, "qBittorrent stopped recheck" + ), ) if should_start: self.qbittorrent.start(info_hash) @@ -359,6 +738,102 @@ class ClientJobExecutor: placement.verified_at.GetCurrentTime() return placement + def _rollback_target(self, definition: job_pb2.JobDefinition) -> None: + baseline_row = self.store.get_job_artifact( + definition.job_id, "target-baseline" + ) + if baseline_row is None: + compensate_materialized_files( + job_id=definition.job_id, + qb_root=self.qb_root, + store=self.store, + ) + return + baseline = baseline_row["value"] + info_hash = _info_hash(definition) + present = self.qbittorrent.get_resource(info_hash) + if present is not None: + if baseline.get("present"): + selected = baseline.get("selected_file_indices") + if isinstance(selected, list) and selected: + self.qbittorrent.set_selection( + info_hash, selected, len(present.files) + ) + if baseline.get("stopped"): + self.qbittorrent.stop(info_hash) + else: + self.qbittorrent.start(info_hash) + else: + self.qbittorrent.delete_entry(info_hash) + compensate_materialized_files( + job_id=definition.job_id, + qb_root=self.qb_root, + store=self.store, + ) + + def _cleanup_staging(self, definition: job_pb2.JobDefinition) -> None: + job_directory = self._job_directory(definition) + try: + if cleanup_transfer(job_directory): + return + except FileNotFoundError: + pass + if not job_directory.is_dir(): + return + compensate_materialized_files( + job_id=definition.job_id, + qb_root=job_directory, + store=self.store, + ) + for name in ("ready.json", "manifest.json"): + candidate = job_directory / name + if candidate.is_file() and not candidate.is_symlink(): + candidate.unlink() + try: + job_directory.rmdir() + except OSError: + pass + + def _reject_unsafe_partfile( + self, + definition: job_pb2.JobDefinition, + resource: NormalizedResource, + ) -> None: + roots = {self.qb_root} + for item in resource.files: + candidate = self.qb_root / PurePosixPath(item.canonical_path).parts[0] + roots.add(candidate if candidate.is_dir() else candidate.parent) + hashes = { + value.lower() for value in ( + definition.resource_id.info_hash_v1_hex, + definition.resource_id.info_hash_v2_hex, + ) if value + } + for root in roots: + if not root.is_dir(): + continue + for candidate in root.glob("*.parts"): + lowered = candidate.name.lower() + if lowered == ".parts" or any(value in lowered for value in hashes): + raise JobExecutionError( + "resource uses a qBittorrent partfile that cannot be " + "safely transferred by this client" + ) + + def _raise_if_cancelled(self, job_id: str) -> None: + event = self._cancel_events.get(job_id) + if event is not None and event.is_set(): + raise JobCancelled("job cancellation requested") + + def _require_space(self, root: Path, required_bytes: int) -> None: + available = shutil.disk_usage(root).free + required = required_bytes + self.free_space_reserve_bytes + if available < required: + raise JobExecutionError( + f"insufficient free space: {available} bytes available, " + f"{required} bytes required including reserve" + ) + def _observer( self, definition: job_pb2.JobDefinition ) -> SyncthingTransferObserver: @@ -455,21 +930,29 @@ class ClientJobExecutor: ) 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.display_step_total = ( + 3 + if definition.operation == job_pb2.JOB_OPERATION_EVICT_CACHE + else 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: + raise JobExecutionError("job definition has no ID") + if definition.operation in { + job_pb2.JOB_OPERATION_ARCHIVE, + job_pb2.JOB_OPERATION_UNARCHIVE, + } and definition.WhichOneof("spec") == "transfer": + return 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" + definition.operation == job_pb2.JOB_OPERATION_EVICT_CACHE + and definition.WhichOneof("spec") == "eviction" ): - raise JobExecutionError("transfer job definition is invalid") + return + raise JobExecutionError("job definition operation/spec is invalid") def _selection_indices(selection) -> list[int]: @@ -510,12 +993,27 @@ def _step_number(step: int) -> int: job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE: 3, job_pb2.JOB_STEP_KIND_QB_VERIFY: 4, job_pb2.JOB_STEP_KIND_STAGING_CLEANUP: 5, + job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE: 1, + job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY: 2, + job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK: 3, + job_pb2.JOB_STEP_KIND_ROLLBACK: 1, }.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 + total = ( + 3 + if step in { + job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE, + job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY, + job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK, + } + else 5 + ) + return max( + 0, number - (0 if state == job_pb2.STEP_STATE_SUCCEEDED else 1) + ) / total def decode_message_metainfo(path: Path): @@ -543,6 +1041,38 @@ def _resource_fingerprint( return fingerprint +def _info_hash(definition: job_pb2.JobDefinition) -> str: + value = ( + definition.resource_id.info_hash_v1_hex + or definition.resource_id.info_hash_v2_hex + ) + if not value: + raise JobExecutionError("job resource identity has no info hash") + return value + + +def _fingerprint_matches( + current: resource_pb2.ResourceStateFingerprint, + expected: resource_pb2.ResourceStateFingerprint, +) -> bool: + left = resource_pb2.ResourceStateFingerprint() + right = resource_pb2.ResourceStateFingerprint() + left.CopyFrom(current) + right.CopyFrom(expected) + left.ClearField("observed_at") + right.ClearField("observed_at") + for value in (left, right): + if value.runtime_state in { + resource_pb2.TORRENT_RUNTIME_STATE_QUEUED, + resource_pb2.TORRENT_RUNTIME_STATE_SEEDING, + resource_pb2.TORRENT_RUNTIME_STATE_STALLED, + }: + value.runtime_state = ( + resource_pb2.TORRENT_RUNTIME_STATE_SEEDING + ) + return left == right + + def _job_error_code(error: Exception) -> int: if isinstance(error, QBittorrentDownloadAttempt): return common_pb2.ERROR_CODE_MANUAL_INTERVENTION_REQUIRED @@ -554,4 +1084,6 @@ def _job_error_code(error: Exception) -> int: return common_pb2.ERROR_CODE_PRECONDITION_FAILED if isinstance(error, TransferError): return common_pb2.ERROR_CODE_PATH_CONFLICT + if isinstance(error, EvictionError): + return common_pb2.ERROR_CODE_PRECONDITION_FAILED return common_pb2.ERROR_CODE_INTERNAL diff --git a/src/archive_clients/qbittorrent.py b/src/archive_clients/qbittorrent.py index 1a5d8fb..84f7460 100644 --- a/src/archive_clients/qbittorrent.py +++ b/src/archive_clients/qbittorrent.py @@ -10,6 +10,7 @@ from dataclasses import dataclass from http.cookiejar import CookieJar from pathlib import PurePosixPath from typing import Any +from typing import Callable from urllib import error, parse, request from archive_clients.config import ServiceConfig @@ -259,6 +260,8 @@ class QBittorrentReader: *, timeout: float, poll_interval: float = 1, + cancel_check: Callable[[], None] | None = None, + progress_callback: Callable[[float], None] | None = None, ) -> RecheckResult: selected = tuple(sorted(set(selected_file_indices))) if not selected: @@ -271,7 +274,9 @@ class QBittorrentReader: "/api/v2/torrents/recheck", {"hashes": torrent_hash} ) deadline = time.monotonic() + timeout + cancel_check = cancel_check or (lambda: None) while True: + cancel_check() record = self._torrent_record(torrent_hash) state = record.get("state") if not isinstance(state, str): @@ -295,6 +300,13 @@ class QBittorrentReader: item.get("index"): item.get("progress") for item in files } + if progress_callback is not None: + progress_callback( + sum( + max(0.0, min(float(progress.get(index, 0)), 1.0)) + for index in selected + ) / len(selected) + ) if all( isinstance(progress.get(index), (int, float)) and float(progress[index]) >= 1 diff --git a/src/archive_clients/state.py b/src/archive_clients/state.py index 6b71dd8..c9c6671 100644 --- a/src/archive_clients/state.py +++ b/src/archive_clients/state.py @@ -11,7 +11,7 @@ from dataclasses import dataclass from pathlib import Path -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 3 class CommandConflict(RuntimeError): @@ -58,6 +58,10 @@ class ClientStore: if version == 1: connection.executescript(_SCHEMA_V2) connection.execute("PRAGMA user_version = 2") + version = 2 + if version == 2: + connection.executescript(_SCHEMA_V3) + connection.execute("PRAGMA user_version = 3") if connection.execute("PRAGMA foreign_key_check").fetchall(): raise RuntimeError("client database foreign-key check failed") if not existed: @@ -268,6 +272,57 @@ class ClientStore: ).fetchall() return [dict(row) for row in rows] + def put_job_artifact( + self, job_id: str, kind: str, value: dict[str, object] + ) -> dict[str, object]: + encoded = _canonical(value) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + """ + SELECT value_json FROM job_artifacts + WHERE job_id = ? AND kind = ? + """, + (job_id, kind), + ).fetchone() + if existing is not None and existing["value_json"] != encoded: + raise JobConflict(f"job artifact {kind} is immutable") + connection.execute( + """ + INSERT INTO job_artifacts (job_id, kind, value_json) + VALUES (?, ?, ?) + ON CONFLICT(job_id, kind) DO NOTHING + """, + (job_id, kind, encoded), + ) + row = connection.execute( + """ + SELECT job_id, kind, value_json FROM job_artifacts + WHERE job_id = ? AND kind = ? + """, + (job_id, kind), + ).fetchone() + result = dict(row) + result["value"] = json.loads(str(result.pop("value_json"))) + return result + + def get_job_artifact( + self, job_id: str, kind: str + ) -> dict[str, object] | None: + with self._connect() as connection: + row = connection.execute( + """ + SELECT job_id, kind, value_json FROM job_artifacts + WHERE job_id = ? AND kind = ? + """, + (job_id, kind), + ).fetchone() + if row is None: + return None + result = dict(row) + result["value"] = json.loads(str(result.pop("value_json"))) + return result + def record_job_event( self, *, @@ -567,3 +622,13 @@ CREATE TABLE route_attempt_updates ( PRIMARY KEY(command_id, sequence) ); """ + + +_SCHEMA_V3 = """ +CREATE TABLE job_artifacts ( + job_id TEXT NOT NULL REFERENCES jobs(job_id), + kind TEXT NOT NULL, + value_json TEXT NOT NULL, + PRIMARY KEY(job_id, kind) +); +""" diff --git a/src/archive_clients/transfer.py b/src/archive_clients/transfer.py index 82c1a37..3678265 100644 --- a/src/archive_clients/transfer.py +++ b/src/archive_clients/transfer.py @@ -11,7 +11,7 @@ import stat import uuid from dataclasses import dataclass from pathlib import Path, PurePosixPath -from typing import Mapping +from typing import Callable, Mapping from google.protobuf import json_format @@ -79,6 +79,7 @@ class FileMaterializer: allow_hardlink: bool = True, allow_reflink: bool = True, copy_chunk_bytes: int = _COPY_CHUNK_BYTES, + cancel_check: Callable[[], None] | None = None, ): if copy_chunk_bytes < 1: raise ValueError("copy_chunk_bytes must be positive") @@ -86,6 +87,7 @@ class FileMaterializer: self.allow_hardlink = allow_hardlink self.allow_reflink = allow_reflink self.copy_chunk_bytes = copy_chunk_bytes + self.cancel_check = cancel_check or (lambda: None) def materialize( self, @@ -100,6 +102,7 @@ class FileMaterializer: allow_preexisting_reuse: bool, sparse_supported: bool, ) -> MaterializedFile: + self.cancel_check() source_relative = _relative_path(source_relative_path) destination_relative = _relative_path(destination_relative_path) source = _existing_regular_file(source_root, source_relative) @@ -184,6 +187,7 @@ class FileMaterializer: destination, operation_id, self.copy_chunk_bytes, + self.cancel_check, ) method = transfer_pb2.MATERIALIZATION_METHOD_COPY @@ -233,6 +237,8 @@ def stage_transfer( allow_hardlink: bool = True, allow_reflink: bool = True, sparse_supported: bool = True, + cancel_check: Callable[[], None] | None = None, + progress: Callable[[int, int], None] | None = None, ) -> PublishedTransfer: """Publish a complete isolated transfer namespace and ready marker.""" @@ -244,7 +250,12 @@ def stage_transfer( store, allow_hardlink=allow_hardlink, allow_reflink=allow_reflink, + cancel_check=cancel_check, ) + total_bytes = sum(entry.logical_bytes for entry in staged.files) + sum( + artifact.logical_bytes for artifact in staged.artifacts + ) + completed_bytes = 0 for entry in staged.files: result = materializer.materialize( job_id=staged.job_id, @@ -263,6 +274,9 @@ def stage_transfer( entry.allocated_bytes = result.allocated_bytes entry.sparse = result.sparse entry.source_staging_method = result.method + completed_bytes += entry.logical_bytes + if progress is not None: + progress(completed_bytes, total_bytes) artifact_sources = artifact_sources or {} for artifact in staged.artifacts: @@ -287,6 +301,9 @@ def stage_transfer( ) artifact.allocated_bytes = result.allocated_bytes artifact.sparse = result.sparse + completed_bytes += artifact.logical_bytes + if progress is not None: + progress(completed_bytes, total_bytes) digest = _sha256_file( _existing_regular_file( job_directory, _relative_path(artifact.payload_relative_path) @@ -356,6 +373,8 @@ def materialize_transfer( allow_hardlink: bool = True, allow_reflink: bool = True, sparse_supported: bool = True, + cancel_check: Callable[[], None] | None = None, + progress: Callable[[int, int], None] | None = None, ) -> transfer_pb2.TransferManifest: """Materialize a verified published payload into a target content root.""" @@ -366,7 +385,10 @@ def materialize_transfer( store, allow_hardlink=allow_hardlink, allow_reflink=allow_reflink, + cancel_check=cancel_check, ) + total_bytes = sum(entry.logical_bytes for entry in result_manifest.files) + completed_bytes = 0 for entry in result_manifest.files: result = materializer.materialize( job_id=result_manifest.job_id, @@ -384,6 +406,9 @@ def materialize_transfer( ) entry.target_method = result.method entry.target_preexisted = result.destination_preexisted + completed_bytes += entry.logical_bytes + if progress is not None: + progress(completed_bytes, total_bytes) return result_manifest @@ -455,20 +480,35 @@ def _validate_manifest(manifest: transfer_pb2.TransferManifest) -> None: raise TransferIntegrityError("manifest creation timestamp is required") indices: set[int] = set() payload_paths: set[str] = set() + folded_payload_paths: set[str] = set() + target_paths: set[str] = set() + folded_target_paths: set[str] = set() for entry in manifest.files: if entry.file_index in indices: raise TransferIntegrityError("manifest contains duplicate file indices") indices.add(entry.file_index) payload = _relative_path(entry.payload_relative_path).as_posix() - _relative_path(entry.target_canonical_path) + target = _relative_path(entry.target_canonical_path).as_posix() if payload in payload_paths: raise TransferIntegrityError("manifest contains duplicate payload paths") + if payload.casefold() in folded_payload_paths: + raise TransferIntegrityError("manifest contains case-colliding payload paths") + if target in target_paths: + raise TransferIntegrityError("manifest contains duplicate target paths") + if target.casefold() in folded_target_paths: + raise TransferIntegrityError("manifest contains case-colliding target paths") payload_paths.add(payload) + folded_payload_paths.add(payload.casefold()) + target_paths.add(target) + folded_target_paths.add(target.casefold()) for artifact in manifest.artifacts: payload = _relative_path(artifact.payload_relative_path).as_posix() if payload in payload_paths: raise TransferIntegrityError("manifest contains duplicate payload paths") + if payload.casefold() in folded_payload_paths: + raise TransferIntegrityError("manifest contains case-colliding payload paths") payload_paths.add(payload) + folded_payload_paths.add(payload.casefold()) def _relative_path(value: str) -> PurePosixPath: @@ -569,6 +609,7 @@ def _sparse_copy( destination: Path, operation_id: str, chunk_bytes: int, + cancel_check: Callable[[], None], ) -> None: temporary = _temporary_path(destination, operation_id) source_fd = os.open(source, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) @@ -580,7 +621,13 @@ def _sparse_copy( os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), stat.S_IMODE(source_stat.st_mode), ) - _copy_extents(source_fd, destination_fd, source_stat.st_size, chunk_bytes) + _copy_extents( + source_fd, + destination_fd, + source_stat.st_size, + chunk_bytes, + cancel_check, + ) os.ftruncate(destination_fd, source_stat.st_size) os.fsync(destination_fd) _publish_temporary(temporary, destination) @@ -596,6 +643,7 @@ def _copy_extents( destination_fd: int, size: int, chunk_bytes: int, + cancel_check: Callable[[], None], ) -> None: if size == 0: return @@ -613,6 +661,7 @@ def _copy_extents( os.lseek(destination_fd, data_offset, os.SEEK_SET) remaining = min(hole_offset, size) - data_offset while remaining: + cancel_check() data = os.read(source_fd, min(chunk_bytes, remaining)) if not data: raise TransferIntegrityError("source ended during sparse copy") @@ -626,6 +675,7 @@ def _copy_extents( os.lseek(destination_fd, 0, os.SEEK_SET) remaining = size while remaining: + cancel_check() data = os.read(source_fd, min(chunk_bytes, remaining)) if not data: raise TransferIntegrityError("source ended during copy") diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 4115705..67d3f25 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -22,6 +22,51 @@ from archive_control.v1 import ( class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): + async def test_eviction_assignment_and_steps_are_admitted(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + token = root / "token" + token.write_text("shared-secret", encoding="utf-8") + os.chmod(token, 0o600) + service = ServiceConfig( + "http://local", PurePosixPath("/api"), root + ) + config = ClientConfig( + "cache-1", "Cache 1", "cache", "ws://control", token, + root / "state.db", root / "backups", service, service, + ) + probe = FilesystemProbe(root, True, True, True, True, True) + daemon = ArchiveClientDaemon(config, [probe, probe], []) + daemon.jobs = Mock() + + assign = control_pb2.Command(command_id=str(uuid4())) + definition = assign.assign_job.job + definition.job_id = str(uuid4()) + definition.eviction.cache_client_id = "cache-1" + acknowledgement = daemon._initial_acknowledgement( + assign, set(), False + ) + self.assertEqual( + acknowledgement.status, + control_pb2.COMMAND_ACK_STATUS_ACCEPTED, + ) + + for step_kind in ( + job_pb2.JOB_STEP_KIND_VERIFY_ARCHIVE_COVERAGE, + job_pb2.JOB_STEP_KIND_QB_REMOVE_ENTRY, + job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK, + ): + execute = control_pb2.Command(command_id=str(uuid4())) + execute.execute_step.job_id = definition.job_id + execute.execute_step.step = step_kind + acknowledgement = daemon._initial_acknowledgement( + execute, set(), False + ) + self.assertEqual( + acknowledgement.status, + control_pb2.COMMAND_ACK_STATUS_ACCEPTED, + ) + async def test_ensure_route_is_durable_and_duplicate_replays_updates(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) diff --git a/tests/test_eviction.py b/tests/test_eviction.py new file mode 100644 index 0000000..138d5f1 --- /dev/null +++ b/tests/test_eviction.py @@ -0,0 +1,181 @@ +import tempfile +import unittest +from pathlib import Path + +from archive_clients.eviction import ( + remove_qb_entry, + safe_unlink, + verify_and_snapshot, +) +from archive_clients.resources import NormalizedResource +from archive_clients.state import ClientStore +from archive_control.v1 import resource_pb2 + + +def normalized(torrent_hash, paths): + summary = resource_pb2.ResourceSummary( + qb_torrent_id=torrent_hash, + display_name="resource", + total_file_count=len(paths), + canonical_paths=True, + ) + summary.resource_id.info_hash_v1_hex = torrent_hash + files = [] + for index, path in enumerate(paths): + size = len(path) + 10 + files.append(resource_pb2.TorrentFile( + file_index=index, + canonical_path=path, + logical_bytes=size, + completed_bytes=size, + selected=True, + )) + summary.selected_files.ranges.add(first=index, last=index) + summary.selected_complete_files.ranges.add(first=index, last=index) + return NormalizedResource(summary, tuple(files), None) + + +class FakeQB: + def __init__(self, evicted, remaining): + self.evicted = evicted + self.remaining = remaining + self.deleted = [] + + def get_resource(self, torrent_hash): + return ( + self.evicted + if self.evicted + and self.evicted.summary.qb_torrent_id == torrent_hash + else None + ) + + def delete_entry(self, torrent_hash): + self.deleted.append(torrent_hash) + self.evicted = None + + def list_resources(self): + return list(self.remaining) + + +class EvictionTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) / "qb" + self.root.mkdir() + self.store = ClientStore(Path(self.temp.name) / "state.db") + self.store.initialize() + self.store.save_job( + "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING", 1, 0, False + ) + + def tearDown(self): + self.temp.cleanup() + + def test_shared_paths_and_unknown_files_survive(self): + evicted = normalized("a" * 40, ["tree/shared.bin", "tree/owned.bin"]) + other = normalized("b" * 40, ["tree/shared.bin"]) + for item in evicted.files: + path = self.root / item.canonical_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"x" * item.logical_bytes) + unknown = self.root / "tree" / "from-another-app.txt" + unknown.write_text("keep") + qb = FakeQB(evicted, [other]) + + verify_and_snapshot( + job_id="job-1", + resource=evicted, + selected_indices=[0, 1], + qb_root=self.root, + store=self.store, + ) + remove_qb_entry( + job_id="job-1", + torrent_hash="a" * 40, + qbittorrent=qb, + store=self.store, + ) + result = safe_unlink( + job_id="job-1", + qb_root=self.root, + qbittorrent=qb, + store=self.store, + ) + self.assertTrue((self.root / "tree/shared.bin").exists()) + self.assertFalse((self.root / "tree/owned.bin").exists()) + self.assertTrue(unknown.exists()) + self.assertTrue((self.root / "tree").is_dir()) + self.assertEqual(qb.deleted, ["a" * 40]) + self.assertEqual( + result["retained_files"], + [{"path": "tree/shared.bin", "reason": "shared"}], + ) + + def test_changed_inode_is_not_unlinked(self): + evicted = normalized("a" * 40, ["tree/owned.bin"]) + path = self.root / "tree/owned.bin" + path.parent.mkdir(parents=True) + path.write_bytes(b"x" * evicted.files[0].logical_bytes) + qb = FakeQB(evicted, []) + verify_and_snapshot( + job_id="job-1", + resource=evicted, + selected_indices=[0], + qb_root=self.root, + store=self.store, + ) + path.unlink() + path.write_bytes(b"y" * evicted.files[0].logical_bytes) + remove_qb_entry( + job_id="job-1", + torrent_hash="a" * 40, + qbittorrent=qb, + store=self.store, + ) + result = safe_unlink( + job_id="job-1", + qb_root=self.root, + qbittorrent=qb, + store=self.store, + ) + self.assertTrue(path.exists()) + self.assertEqual( + result["retained_files"][0]["reason"], "identity-changed" + ) + + def test_retry_is_idempotent(self): + evicted = normalized("a" * 40, ["owned.bin"]) + path = self.root / "owned.bin" + path.write_bytes(b"x" * evicted.files[0].logical_bytes) + qb = FakeQB(evicted, []) + verify_and_snapshot( + job_id="job-1", + resource=evicted, + selected_indices=[0], + qb_root=self.root, + store=self.store, + ) + remove_qb_entry( + job_id="job-1", + torrent_hash="a" * 40, + qbittorrent=qb, + store=self.store, + ) + first = safe_unlink( + job_id="job-1", + qb_root=self.root, + qbittorrent=qb, + store=self.store, + ) + second = safe_unlink( + job_id="job-1", + qb_root=self.root, + qbittorrent=qb, + store=self.store, + ) + self.assertEqual(first, second) + self.assertEqual(qb.deleted, ["a" * 40]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_jobs.py b/tests/test_jobs.py index 3dc9c8b..baa8736 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -6,7 +6,11 @@ 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.jobs import ( + ClientJobExecutor, + JobExecutionError, + _resource_fingerprint, +) from archive_clients.resources import normalize_resource from archive_clients.state import ClientStore from archive_control.v1 import control_pb2, job_pb2 @@ -90,6 +94,9 @@ class ClientJobHappyPathTests(unittest.TestCase): 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) + definition.transfer.source_fingerprint.CopyFrom( + _resource_fingerprint(resource, source_id) + ) source_store = ClientStore(root / "source.db") target_store = ClientStore(root / "target.db") @@ -98,7 +105,7 @@ class ClientJobHappyPathTests(unittest.TestCase): source_qb = Mock() source_qb.get_resource.return_value = resource target_qb = Mock() - target_qb.get_resource.side_effect = [None, resource] + target_qb.get_resource.side_effect = [None, None, resource] syncthing = CompleteSyncthing() source = ClientJobExecutor( client_id=source_id, @@ -156,7 +163,10 @@ class ClientJobHappyPathTests(unittest.TestCase): )) self.assertEqual( [event.sequence for event in events], - [cursor_sequence + 1, cursor_sequence + 2], + list(range( + cursor_sequence + 1, + cursor_sequence + len(events) + 1, + )), ) cursor_sequence = events[-1].sequence cursor_revision = events[-1].job_revision @@ -250,6 +260,54 @@ class ClientJobHappyPathTests(unittest.TestCase): ) self.assertEqual(store.list_active_job_cursors(), []) + def test_active_source_cancellation_reports_durable_rollback(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 / "route", + 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, + )) + events = executor.cancel(control_pb2.CancelJobCommand( + job_id=definition.job_id, + expected_job_revision=2, + expected_last_event_sequence=1, + reason="test", + )) + self.assertEqual( + events[-1].type, + control_pb2.JOB_EVENT_TYPE_ROLLBACK_SUCCEEDED, + ) + self.assertEqual( + events[-1].state, job_pb2.JOB_STATE_ROLLING_BACK + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_state.py b/tests/test_state.py index b90812f..8b6c78f 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -120,6 +120,28 @@ class ClientStoreTests(unittest.TestCase): "operation-1", '{"method":2}' ) + def test_job_artifacts_are_immutable_and_survive_restart(self): + with tempfile.TemporaryDirectory() as directory: + database = Path(directory) / "state.db" + store = ClientStore(database) + store.initialize() + store.save_job( + "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING", + 1, 0, False, + ) + store.put_job_artifact( + "job-1", "baseline", {"selected": [1, 3]} + ) + reopened = ClientStore(database) + self.assertEqual( + reopened.get_job_artifact("job-1", "baseline")["value"], + {"selected": [1, 3]}, + ) + with self.assertRaises(JobConflict): + reopened.put_job_artifact( + "job-1", "baseline", {"selected": [2]} + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_transfer.py b/tests/test_transfer.py index afcf855..f8ffc03 100644 --- a/tests/test_transfer.py +++ b/tests/test_transfer.py @@ -9,6 +9,7 @@ from uuid import uuid4 from archive_clients.state import ClientStore from archive_clients.transfer import ( FileMaterializer, + TransferIntegrityError, canonical_message_json, load_published_transfer, materialize_transfer, @@ -212,6 +213,22 @@ class TransferHappyPathTests(unittest.TestCase): self.assertEqual(first, second) self.assertEqual(first, canonical_message_json(manifest)) + def test_manifest_rejects_case_colliding_target_paths(self): + manifest = self._manifest() + manifest.files[1].target_canonical_path = "album/ONE.bin" + with self.assertRaisesRegex( + TransferIntegrityError, "case-colliding target" + ): + stage_transfer( + manifest, + source_root=self.source, + sync_root=self.sync, + store=self.store, + artifact_sources={ + "metainfo/source.torrent": self.metainfo + }, + ) + def _manifest(self, job_id=None): manifest = transfer_pb2.TransferManifest( manifest_version=1,