"""Control-commanded happy-path transfer execution.""" from __future__ import annotations import hashlib import json import shutil import stat import threading import time import uuid from pathlib import Path, PurePosixPath from typing import Callable, Iterable 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, ) 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 JobCancelled(JobExecutionError): 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, verification_timeout: float = 30 * 60, free_space_reserve_bytes: int = 1024 * 1024 * 1024, ): 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 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 ) -> 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, event_callback: Callable[[control_pb2.JobEvent], None] | None = None, ) -> 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, }: if event_callback is not None: for event in replay: event_callback(event) 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 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, 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 ) 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=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 ), 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 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) 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: 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 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, 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) 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, 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, progress) return None if step == job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER: self._wait_for_syncthing(definition, progress) return None if step == job_pb2.JOB_STEP_KIND_TARGET_MATERIALIZE: self._target_materialize(definition, progress) return None if step == job_pb2.JOB_STEP_KIND_QB_VERIFY: return self._qb_verify(definition, progress) if step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP: self._cleanup_staging(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 _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): 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 route_root = self.route_path(definition.transfer.route_id) # Staging is normally zero-copy when qB's data root and the paired # Syncthing route share a filesystem. Do not reserve the complete # logical payload in that case: FileMaterializer will use link(2), # which consumes only directory/inode metadata. Retain the metainfo # allowance and reserve, and account for any source files that really # must fall back to a data-copy path. self._require_space( route_root, self._copy_required_bytes( self.qb_root, route_root, ( (entry.target_canonical_path, entry.logical_bytes) for entry in manifest.files ), ) + len(resource.metainfo_bytes), ) stage_transfer( manifest, source_root=self.qb_root, sync_root=route_root, 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, 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: 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 # ready marker and manifest have arrived atomically as a set. pass time.sleep(self.poll_interval) 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)) # The target can likewise hardlink an arrived Syncthing payload into # qB's content root when those directories share a filesystem. self._require_space( self.qb_root, self._copy_required_bytes( published.job_directory, self.qb_root, ( (entry.payload_relative_path, entry.logical_bytes) for entry in published.manifest.files ), ), ) 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, 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)) 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, ) resource = self.qbittorrent.get_resource(info_hash) if resource is None: raise JobExecutionError( "added qBittorrent resource did not become visible" ) qb_torrent_id = resource.summary.qb_torrent_id 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( qb_torrent_id, selected, total_files ) self.qbittorrent.recheck_and_wait( qb_torrent_id, 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(qb_torrent_id) 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 _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: qb_torrent_id = present.summary.qb_torrent_id if baseline.get("present"): selected = baseline.get("selected_file_indices") if isinstance(selected, list) and selected: self.qbittorrent.set_selection( qb_torrent_id, selected, len(present.files) ) if baseline.get("stopped"): self.qbittorrent.stop(qb_torrent_id) else: self.qbittorrent.start(qb_torrent_id) else: self.qbittorrent.delete_entry(qb_torrent_id) 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" ) @staticmethod def _copy_required_bytes( source_root: Path, destination_root: Path, files: Iterable[tuple[str, int]], ) -> int: """Return logical bytes that cannot be materialized by hardlink. A hardlink is possible only for regular files on the destination filesystem. Conservatively charge a file when it cannot be inspected; the normal materializer will then provide the precise integrity error. """ destination_device = destination_root.stat().st_dev required = 0 for relative_path, logical_bytes in files: relative = PurePosixPath(relative_path) source = source_root.joinpath(*relative.parts) try: metadata = source.stat(follow_symlinks=False) except OSError: required += logical_bytes continue if ( not stat.S_ISREG(metadata.st_mode) or metadata.st_dev != destination_device ): required += logical_bytes return required 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 = ( 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 ( definition.operation == job_pb2.JOB_OPERATION_EVICT_CACHE and definition.WhichOneof("spec") == "eviction" ): return raise JobExecutionError("job definition operation/spec 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, 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) 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): 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 _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 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 if isinstance(error, EvictionError): return common_pb2.ERROR_CODE_PRECONDITION_FAILED return common_pb2.ERROR_CODE_INTERNAL