"""Control-commanded happy-path transfer execution.""" from __future__ import annotations import hashlib import json import logging import os 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 RouteSetupError, SyncthingTransferObserver from archive_clients.transfer import ( TransferError, TransferIntegrityError, cleanup_partial_transfer, 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 logger = logging.getLogger(__name__) 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 = 32 * 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 ) step_replay = [ event for event in replay if event.progress.step == command.step ] if step_replay and step_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 if step_replay: started = step_replay[0] cursor = step_replay[-1] emitted = list(step_replay) else: previous = replay[-1] if replay else None started = self._event( definition, sequence=( previous.sequence + 1 if previous is not None else command.expected_last_event_sequence + 1 ), revision=( previous.job_revision + 1 if previous is not None else 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, ) self._record(definition, started) cursor = started emitted = [started] if event_callback is not None: for event in emitted: event_callback(event) # A newly-started step must publish its first observation. In # particular, a sparse Syncthing temporary file may retain the same # allocated-block count for a long time while data is still needed; # suppressing that first observation made a healthy transfer look # permanently stalled to the control daemon. speed_sample: list[float | int | None] = [None, 0] def progress( fraction: float, bytes_complete: int = 0, bytes_total: int = 0, detail: str = "", ) -> None: nonlocal cursor now = time.monotonic() previous_time = speed_sample[0] elapsed = ( now - float(previous_time) if previous_time is not None else 0.0 ) if previous_time is not None and 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", ), ) try: self._observer(definition).rescan() except RouteSetupError as exc: # A manual Syncthing scan may take longer than the bounded REST # request timeout for a large newly linked file. The folder watch # and the following transfer observation still converge, so do # not roll back an otherwise durable staged transfer. logger.warning("syncthing_rescan_deferred", extra={ "job_id": definition.job_id, "error_type": type(exc).__name__, }) 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, ) cleanup_partial_transfer( job_directory, job_id=definition.job_id, 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 or _mount_id(source) != _mount_id(destination_root) ): 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, RouteSetupError): return common_pb2.ERROR_CODE_UNAVAILABLE 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 def _mount_id(path: Path) -> str | None: """Return Linux's effective mount ID for a path when procfs is available. Bind mounts can share ``st_dev`` while still rejecting ``link(2)`` with ``EXDEV``. Mount IDs distinguish that case without creating probe files inside a Syncthing folder. """ try: target = os.path.realpath(path) best: tuple[int, str] | None = None with open("/proc/self/mountinfo", encoding="utf-8") as source: for line in source: fields = line.rstrip("\n").split(" ") if len(fields) < 5: continue mountpoint = _unescape_mount_path(fields[4]) if target != mountpoint and not target.startswith( mountpoint.rstrip("/") + "/" ): continue candidate = (len(mountpoint), fields[0]) if best is None or candidate[0] > best[0]: best = candidate return None if best is None else best[1] except OSError: return None def _unescape_mount_path(value: str) -> str: return ( value.replace("\\040", " ") .replace("\\011", "\t") .replace("\\012", "\n") .replace("\\134", "\\") )