"""Native, verified updater for the public PristonTale EU depot.""" from __future__ import annotations from base64 import b64decode from binascii import Error as BinAsciiError from collections import deque from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait from dataclasses import asdict, dataclass from datetime import datetime import hashlib from http.client import HTTPException import json import os from pathlib import Path, PurePosixPath import re import stat import threading import time from typing import Callable, Iterable, Mapping, Protocol from urllib.error import HTTPError from urllib.parse import quote, urlparse from urllib.request import Request, urlopen from uuid import uuid4 from . import __version__ from .paths import AppPaths from .telemetry import classify_network_failure, current_telemetry CDN_CONFIGURATION_URL = "https://cdn-main.chatplex.io/pristontale-eu/cdn_configuration.json" DEPOT_ID = 1 FALLBACK_SERVERS = ( "cdn-fr-sbg1.chatplex.io/pristontale-eu", "cdn-us-ore1.chatplex.io/pristontale-eu", ) READ_SIZE = 128 * 1024 HASH_READ_SIZE = 1024 * 1024 DOWNLOAD_WORKERS = 8 DOWNLOAD_INITIAL_WORKERS = DOWNLOAD_WORKERS LOW_BANDWIDTH_WORKERS = 2 LOW_BANDWIDTH_BYTES_PER_SECOND = 5_000_000 DOWNLOAD_RETRY_DELAYS_SECONDS = (1.0, 2.0, 4.0, 8.0, 16.0) PROGRESS_UPDATE_INTERVAL_SECONDS = 0.075 ADAPTIVE_EVALUATION_INTERVAL_SECONDS = 1.0 ADAPTIVE_RETRY_WINDOW_SECONDS = 15.0 ADAPTIVE_RETRY_RECOVERY_SECONDS = 5.0 TRANSFER_RATE_WINDOW_SECONDS = 5.0 MAX_CREATE_IF_MISSING_BYTES = 16 * 1024 * 1024 MAX_LINE_ENDING_RECOVERY_BYTES = 1024 * 1024 UNMANAGED_EXECUTABLE_SUFFIXES = frozenset( { ".asi", ".bat", ".cmd", ".com", ".cpl", ".desktop", ".dll", ".drv", ".exe", ".jar", ".js", ".lnk", ".lua", ".msi", ".ocx", ".ps1", ".py", ".reg", ".scr", ".sh", ".so", ".url", ".vbe", ".vbs", ".wsf", } ) ALWAYS_VERIFY_PATHS = frozenset( {"game.exe", "game.dll", "gamese.exe", "gamese.dll"} ) SHA1_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$") CONTENT_RANGE_PATTERN = re.compile(r"^bytes\s+(\d+)-(\d+)/(\d+)$", re.IGNORECASE) DEPOT_LAUNCH_CATALOG_SCHEMA = 1 DEPOT_FILE_CATALOG_SCHEMA = 1 MAX_PREFLIGHT_ISSUES = 32 class DepotError(RuntimeError): """Raised when a depot response or update operation cannot be trusted.""" class _RetryableDownloadError(DepotError): """One transient network or content-integrity failure.""" def __init__( self, message: str, *, discard_partial: bool = False, http_status: int | None = None, ) -> None: super().__init__(message) self.discard_partial = discard_partial self.http_status = http_status @dataclass(frozen=True) class DownloadOptions: """User-selected limits around the adaptive depot scheduler.""" bandwidth_limit_bytes_per_second: int | None = None low_bandwidth: bool = False minimum_workers: int = 1 maximum_workers: int | None = None def __post_init__(self) -> None: if not isinstance(self.low_bandwidth, bool): raise ValueError("low_bandwidth must be a boolean") if ( self.bandwidth_limit_bytes_per_second is not None and ( isinstance(self.bandwidth_limit_bytes_per_second, bool) or self.bandwidth_limit_bytes_per_second <= 0 ) ): raise ValueError("bandwidth limit must be positive or None") if isinstance(self.minimum_workers, bool) or self.minimum_workers <= 0: raise ValueError("minimum download workers must be positive") if self.maximum_workers is not None and ( isinstance(self.maximum_workers, bool) or self.maximum_workers <= 0 ): raise ValueError("maximum download workers must be positive or None") if ( self.maximum_workers is not None and self.minimum_workers > self.maximum_workers ): raise ValueError("minimum download workers exceed the maximum") if self.low_bandwidth and self.minimum_workers > LOW_BANDWIDTH_WORKERS: raise ValueError("low-bandwidth minimum workers exceed its worker cap") @property def effective_bandwidth_limit(self) -> int | None: configured = self.bandwidth_limit_bytes_per_second if not self.low_bandwidth: return configured return min( configured if configured is not None else LOW_BANDWIDTH_BYTES_PER_SECOND, LOW_BANDWIDTH_BYTES_PER_SECOND, ) @property def effective_maximum_workers(self) -> int: configured = self.maximum_workers or DOWNLOAD_WORKERS if self.low_bandwidth: configured = min(configured, LOW_BANDWIDTH_WORKERS) return max(self.minimum_workers, configured) @dataclass(frozen=True) class DownloadTelemetry: """One internal measurement produced while retrieving a depot file.""" kind: str file: str server: str attempt: int latency_seconds: float | None = None bytes_written: int = 0 write_seconds: float = 0.0 reason: str | None = None http_status: int | None = None resumed_bytes: int = 0 DownloadObserver = Callable[[DownloadTelemetry], None] class BandwidthLimiter: """Thread-safe aggregate byte-rate limiter with a one-chunk burst allowance.""" def __init__( self, bytes_per_second: int, *, clock: Callable[[], float] = time.monotonic, sleeper: Callable[[float], None] = time.sleep, ) -> None: if isinstance(bytes_per_second, bool) or bytes_per_second <= 0: raise ValueError("bandwidth limit must be positive") self.bytes_per_second = bytes_per_second self._clock = clock self._sleeper = sleeper self._lock = threading.Lock() self._next_available_at = clock() def consume(self, byte_count: int) -> None: """Wait until this newly received chunk's aggregate reservation begins. The network read happens before this call. Reserving the chunk's *start* keeps all workers under one aggregate rate without charging again for time already spent receiving it. The first chunk is the bucket's burst allowance. """ if byte_count < 0: raise ValueError("byte count must not be negative") if byte_count == 0: return with self._lock: now = self._clock() start = max(now, self._next_available_at) self._next_available_at = start + (byte_count / self.bytes_per_second) delay = start - now if delay > 0: self._sleeper(delay) class AdaptiveConcurrencyController: """Select a bounded worker target from live network, host, and plan data.""" def __init__( self, options: DownloadOptions, *, clock: Callable[[], float] = time.monotonic, cpu_load: Callable[[], float | None] | None = None, ) -> None: self.options = options self._clock = clock self._cpu_load = cpu_load or _normalized_cpu_load self._lock = threading.Lock() self._target = max( options.minimum_workers, min(DOWNLOAD_INITIAL_WORKERS, options.effective_maximum_workers), ) self._last_evaluation = clock() self._transferred_bytes = 0 self._transfer_samples: deque[tuple[float, int]] = deque( [(self._last_evaluation, 0)] ) self._response_samples: deque[tuple[float, float]] = deque() self._retry_samples: deque[float] = deque() self._retry_count = 0 self._handled_retry_count = 0 self._last_retry_at: float | None = None self._latency_seconds: float | None = None self._disk_samples: deque[tuple[float, int, float]] = deque() def record_response(self, latency_seconds: float) -> None: if latency_seconds < 0: return now = self._clock() with self._lock: self._response_samples.append((now, latency_seconds)) self._prune_health_samples(now) if self._latency_seconds is None: self._latency_seconds = latency_seconds else: self._latency_seconds = ( (self._latency_seconds * 0.75) + (latency_seconds * 0.25) ) def record_retry(self) -> None: now = self._clock() with self._lock: self._retry_count += 1 self._retry_samples.append(now) self._last_retry_at = now self._prune_health_samples(now) def record_chunk(self, byte_count: int, write_seconds: float) -> None: if byte_count <= 0: return now = self._clock() with self._lock: self._transferred_bytes += byte_count self._transfer_samples.append((now, self._transferred_bytes)) _prune_transfer_samples(self._transfer_samples, now) self._disk_samples.append((now, byte_count, max(0.0, write_seconds))) self._prune_disk_samples(now) def _prune_health_samples(self, now: float) -> None: cutoff = now - ADAPTIVE_RETRY_WINDOW_SECONDS while self._response_samples and self._response_samples[0][0] < cutoff: self._response_samples.popleft() while self._retry_samples and self._retry_samples[0] < cutoff: self._retry_samples.popleft() def _prune_disk_samples(self, now: float) -> None: cutoff = now - TRANSFER_RATE_WINDOW_SECONDS while self._disk_samples and self._disk_samples[0][0] < cutoff: self._disk_samples.popleft() def target_workers(self, remaining_files: int, remaining_bytes: int) -> int: """Return the next target; active workers are allowed to drain naturally.""" if remaining_files <= 0: return 0 maximum = min(self.options.effective_maximum_workers, remaining_files) minimum = min(self.options.minimum_workers, maximum) with self._lock: self._target = min(maximum, max(minimum, self._target)) now = self._clock() average_file_size = remaining_bytes / max(1, remaining_files) if now - self._last_evaluation < ADAPTIVE_EVALUATION_INTERVAL_SECONDS: return self._target self._last_evaluation = now _prune_transfer_samples(self._transfer_samples, now) self._prune_health_samples(now) self._prune_disk_samples(now) rate = _sample_rate(self._transfer_samples) attempts = len(self._response_samples) + len(self._retry_samples) error_rate = len(self._retry_samples) / attempts if attempts else 0.0 latency = self._latency_seconds disk_bytes = sum(sample[1] for sample in self._disk_samples) disk_seconds = sum(sample[2] for sample in self._disk_samples) disk_rate = ( disk_bytes / disk_seconds if disk_seconds > 0 else None ) try: cpu_load = self._cpu_load() except (OSError, ValueError): cpu_load = None new_retries = self._retry_count - self._handled_retry_count if new_retries: self._handled_retry_count = self._retry_count if error_rate >= 0.12 and self._target > minimum: backoff = min(2, new_retries) self._target = max(minimum, self._target - backoff) return self._target resource_constrained = ( (cpu_load is not None and cpu_load >= 0.90) or ( rate is not None and disk_rate is not None and disk_rate < rate * 0.75 ) ) if resource_constrained and self._target > minimum: self._target -= 1 return self._target limit = self.options.effective_bandwidth_limit limit_saturated = ( limit is not None and rate is not None and rate >= limit * 0.90 ) healthy = ( (cpu_load is None or cpu_load < 0.80) and ( rate is None or disk_rate is None or disk_rate >= rate ) ) retry_recovery_pending = ( self._last_retry_at is not None and now - self._last_retry_at < ADAPTIVE_RETRY_RECOVERY_SECONDS ) if ( healthy and not retry_recovery_pending and not limit_saturated and self._target < maximum ): # Parallel requests hide CDN round-trip time and are especially useful # for this depot's large population of small files. step = 2 if ( (latency is not None and latency >= 0.75) or average_file_size <= 512 * 1024 ) else 1 self._target = min(maximum, self._target + step) return min(maximum, max(minimum, self._target)) @dataclass(frozen=True) class DepotFile: """One exact file in the published depot build.""" path: str build: int size: int sha1: str @dataclass(frozen=True) class CreateIfMissing: """A small publisher-provided file created only when absent.""" path: str content_base64: str @dataclass(frozen=True) class DepotStartOption: """One publisher-advertised realm launch target.""" name: str path: str arguments: str DEFAULT_DEPOT_START_OPTION = DepotStartOption( name="Babel", path="Game.exe", arguments="PTEU_REBORN", ) @dataclass(frozen=True) class DepotSnapshot: """The current, verified shape of the public game depot.""" region: str server: str name: str build: int version: str files: tuple[DepotFile, ...] deleted_files: tuple[str, ...] create_if_missing: tuple[CreateIfMissing, ...] start_options: tuple[DepotStartOption, ...] = (DEFAULT_DEPOT_START_OPTION,) default_start_option: str = DEFAULT_DEPOT_START_OPTION.name @dataclass(frozen=True) class DepotLaunchCatalog: """Small cached subset of a depot manifest needed for offline Play.""" build: int start_options: tuple[DepotStartOption, ...] default_start_option: str @dataclass(frozen=True) class DepotFileCatalogEntry: """One manifest file plus the identity recorded after verified installation.""" path: str size: int sha1: str device: int inode: int mtime_ns: int ctime_ns: int verified: bool @dataclass(frozen=True) class DepotFileCatalog: """Cached manifest used for fast, network-free launch preflight checks.""" build: int files: tuple[DepotFileCatalogEntry, ...] created_at: str @dataclass(frozen=True) class DepotIntegrityIssue: path: str reason: str @dataclass(frozen=True) class DepotPreflightResult: """Outcome of validating cached file identities before starting the game.""" build: int files_total: int verified_files: int unverified_files: int changed_files_hashed: int issue_count: int issues: tuple[DepotIntegrityIssue, ...] duration_ms: int @property def ready(self) -> bool: return self.issue_count == 0 @property def fully_verified(self) -> bool: return self.ready and self.unverified_files == 0 @dataclass(frozen=True) class _ScannedGameEntry: path: Path state: os.stat_result kind: str def launch_catalog_from_snapshot(snapshot: DepotSnapshot) -> DepotLaunchCatalog: return DepotLaunchCatalog( build=snapshot.build, start_options=snapshot.start_options, default_start_option=snapshot.default_start_option, ) def select_depot_start_option( catalog: DepotLaunchCatalog | None, preferred_realm: str | None = None, ) -> DepotStartOption: """Choose a published realm, falling back to Babel when possible.""" current = catalog or DepotLaunchCatalog( build=0, start_options=(DEFAULT_DEPOT_START_OPTION,), default_start_option=DEFAULT_DEPOT_START_OPTION.name, ) if preferred_realm: preferred = preferred_realm.casefold() for option in current.start_options: if option.name.casefold() == preferred: return option for option in current.start_options: if option.name.casefold() == DEFAULT_DEPOT_START_OPTION.name.casefold(): return option for option in current.start_options: if option.name.casefold() == current.default_start_option.casefold(): return option return current.start_options[0] def cache_depot_launch_catalog(path: Path, snapshot: DepotSnapshot) -> None: """Cache only publisher-advertised launch metadata for network-free Play.""" catalog = launch_catalog_from_snapshot(snapshot) _write_json( path, { "schema": DEPOT_LAUNCH_CATALOG_SCHEMA, "build": catalog.build, "default_start_option": catalog.default_start_option, "start_options": [asdict(option) for option in catalog.start_options], }, ) def read_cached_depot_launch_catalog(path: Path) -> DepotLaunchCatalog | None: """Read the last valid launch catalog without performing network I/O.""" try: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): return None if value.get("schema") != DEPOT_LAUNCH_CATALOG_SCHEMA: return None build = _positive_int(value.get("build"), "cached depot build") raw_options = value.get("start_options") if not isinstance(raw_options, list): return None manifest_options = [ { "Name": item.get("name"), "Path": item.get("path"), "Arguments": item.get("arguments"), } if isinstance(item, dict) else item for item in raw_options ] options, default_name = _parse_start_options( manifest_options, value.get("default_start_option"), ) except (OSError, UnicodeError, json.JSONDecodeError, DepotError): return None return DepotLaunchCatalog( build=build, start_options=options, default_start_option=default_name, ) def cache_depot_file_catalog( path: Path, game_directory: Path, snapshot: DepotSnapshot, *, verified_paths: Iterable[str], previous: DepotFileCatalog | None = None, ) -> DepotFileCatalog: """Record identities without rehashing files already verified by the updater.""" root = _prepare_game_root(game_directory, create=False) scanned_files = _scan_game_tree(root) explicitly_verified = {verified_path.casefold() for verified_path in verified_paths} previous_files = ( {entry.path.casefold(): entry for entry in previous.files} if previous is not None else {} ) entries: list[DepotFileCatalogEntry] = [] for file in snapshot.files: scanned = scanned_files.get(file.path) if scanned is None: raise DepotError(f"cannot cache missing depot file identity: {file.path}") state = scanned.state if scanned.kind != "file" or state.st_size != file.size: raise DepotError(f"cannot cache invalid depot file identity: {file.path}") prior = previous_files.get(file.path.casefold()) verified = file.path.casefold() in explicitly_verified if ( not verified and prior is not None and prior.verified and prior.size == file.size and prior.sha1 == file.sha1 and _catalog_identity_matches(prior, state) ): verified = True entries.append(_catalog_entry(file, state, verified=verified)) catalog = DepotFileCatalog( build=snapshot.build, files=tuple(entries), created_at=datetime.now().astimezone().isoformat(), ) _write_json( path, _depot_file_catalog_payload(catalog), compact=True, ) return catalog def read_cached_depot_file_catalog(path: Path) -> DepotFileCatalog | None: """Read and strictly validate the local manifest/identity cache.""" try: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): return None if value.get("schema") != DEPOT_FILE_CATALOG_SCHEMA: return None build = _positive_int(value.get("build"), "cached depot file build") created_at = _string(value.get("created_at"), "cached depot file timestamp") raw_files = value.get("files") if not isinstance(raw_files, list): return None entries: list[DepotFileCatalogEntry] = [] seen: set[str] = set() for raw in raw_files: entry = _parse_depot_file_catalog_entry(raw) if entry.path.casefold() in seen: return None seen.add(entry.path.casefold()) entries.append(entry) except (OSError, UnicodeError, json.JSONDecodeError, DepotError): return None return DepotFileCatalog(build=build, files=tuple(entries), created_at=created_at) def read_current_depot_file_catalog(paths: AppPaths) -> DepotFileCatalog | None: """Return the file catalog only when it matches the last completed sync.""" catalog = read_cached_depot_file_catalog(paths.depot_file_catalog) if catalog is None: return None if _completed_depot_build(paths.native_depot_state) != catalog.build: return None return catalog def preflight_managed_game( paths: AppPaths, *, catalog: DepotFileCatalog | None = None, ) -> DepotPreflightResult | None: """Check all managed paths cheaply and hash only files whose identity changed.""" catalog = catalog or read_current_depot_file_catalog(paths) if catalog is None: return None if _completed_depot_build(paths.native_depot_state) != catalog.build: return None started = time.perf_counter() root = _prepare_game_root(paths.source_game, create=False) if not root.is_dir(): return DepotPreflightResult( build=catalog.build, files_total=len(catalog.files), verified_files=0, unverified_files=sum(not entry.verified for entry in catalog.files), changed_files_hashed=0, issue_count=1, issues=(DepotIntegrityIssue(".", "managed game directory is missing"),), duration_ms=_elapsed_milliseconds(started), ) try: scanned_files = _scan_game_tree(root) except DepotError as exc: return DepotPreflightResult( build=catalog.build, files_total=len(catalog.files), verified_files=0, unverified_files=sum(not entry.verified for entry in catalog.files), changed_files_hashed=0, issue_count=1, issues=(DepotIntegrityIssue(".", str(exc)),), duration_ms=_elapsed_milliseconds(started), ) verified_files = 0 unverified_files = 0 changed_files_hashed = 0 issue_count = 0 issues: list[DepotIntegrityIssue] = [] updated_entries: list[DepotFileCatalogEntry] = [] cache_changed = False for entry in catalog.files: reason = None state: os.stat_result | None = None scanned = scanned_files.get(entry.path) if scanned is None: reason = "missing" elif scanned.kind != "file": reason = scanned.kind elif scanned.state.st_size != entry.size: reason = "size mismatch" else: state = scanned.state updated_entry = entry if reason is None and state is not None: if not entry.verified: unverified_files += 1 elif _catalog_identity_matches(entry, state): verified_files += 1 else: changed_files_hashed += 1 try: assert scanned is not None matches = _sha1(scanned.path).casefold() == entry.sha1 except DepotError as exc: reason = f"cannot verify changed file: {exc}" else: if matches: verified_files += 1 updated_entry = _catalog_entry_from_cached(entry, state) cache_changed = True else: reason = "hash mismatch" if reason is not None: issue_count += 1 if len(issues) < MAX_PREFLIGHT_ISSUES: issues.append(DepotIntegrityIssue(entry.path, reason)) updated_entries.append(updated_entry) if cache_changed: updated_catalog = DepotFileCatalog( build=catalog.build, files=tuple(updated_entries), created_at=datetime.now().astimezone().isoformat(), ) _write_json( paths.depot_file_catalog, _depot_file_catalog_payload(updated_catalog), compact=True, ) return DepotPreflightResult( build=catalog.build, files_total=len(catalog.files), verified_files=verified_files, unverified_files=unverified_files, changed_files_hashed=changed_files_hashed, issue_count=issue_count, issues=tuple(issues), duration_ms=_elapsed_milliseconds(started), ) @dataclass(frozen=True) class PlannedDownload: file: DepotFile reason: str @dataclass(frozen=True) class UpdatePlan: """The safe changes needed to make a local tree match one depot snapshot.""" downloads: tuple[PlannedDownload, ...] deletes: tuple[str, ...] files_total: int bytes_total: int hash_verified_paths: tuple[str, ...] = () @dataclass(frozen=True) class SyncProgress: """A thread-safe progress snapshot emitted by ``sync_managed_game``.""" phase: str files_completed: int files_total: int bytes_completed: int bytes_total: int current_file: str | None bytes_transferred: int = 0 bytes_remaining: int = 0 transfer_rate_bytes_per_second: float | None = None eta_seconds: float | None = None current_server: str | None = None retry_count: int = 0 retry_reason: str | None = None active_workers: int = 0 target_workers: int = 0 @dataclass(frozen=True) class DepotSyncResult: """Persistent result of one native depot synchronization.""" region: str server: str name: str build: int version: str files_total: int files_downloaded: int bytes_downloaded: int files_deleted: int files_created: int files_quarantined: int quarantine_directory: str | None repair: bool dry_run: bool completed_at: str files_planned: int = 0 bytes_planned: int = 0 manifest_duration_ms: int = 0 scan_duration_ms: int = 0 download_duration_ms: int = 0 apply_duration_ms: int = 0 update_duration_ms: int = 0 network_bytes_downloaded: int = 0 download_retry_count: int = 0 download_resume_used: bool = False download_average_mbps: float | None = None mirror_id: str | None = None @dataclass(frozen=True) class DepotDownloadMetrics: """Aggregate network measurements for one depot download phase.""" duration_ms: int = 0 transferred_bytes: int = 0 retry_count: int = 0 resume_used: bool = False resumed_bytes: int = 0 average_mbps: float | None = None mirror_id: str | None = None http_status: int | None = None @dataclass(frozen=True) class QuarantineResult: """Recoverable unmanaged entries removed before authenticating an import.""" files: int directory: str | None ProgressCallback = Callable[[SyncProgress], None] class HttpResponse(Protocol): """The small response surface used by the native depot client.""" headers: Mapping[str, str] def read(self, size: int = -1) -> bytes: ... def close(self) -> None: ... HttpOpener = Callable[..., HttpResponse] class DepotClient: """Small HTTP client for the publisher's documented public depot endpoints.""" def __init__( self, *, timeout_seconds: float = 60.0, opener: HttpOpener = urlopen, retry_delays: tuple[float, ...] = DOWNLOAD_RETRY_DELAYS_SECONDS, sleeper: Callable[[float], None] = time.sleep, download_observer: DownloadObserver | None = None, ) -> None: if timeout_seconds <= 0: raise ValueError("timeout_seconds must be positive") if any(delay < 0 for delay in retry_delays): raise ValueError("retry delays must not be negative") self.timeout_seconds = timeout_seconds self._opener = opener self._retry_delays = retry_delays self._sleeper = sleeper self._candidate_servers: tuple[str, ...] = () self._download_observer = download_observer self._bandwidth_limiter: BandwidthLimiter | None = None @property def download_observer(self) -> DownloadObserver | None: return self._download_observer @download_observer.setter def download_observer(self, observer: DownloadObserver | None) -> None: self._download_observer = observer @property def bandwidth_limiter(self) -> BandwidthLimiter | None: return self._bandwidth_limiter @bandwidth_limiter.setter def bandwidth_limiter(self, limiter: BandwidthLimiter | None) -> None: self._bandwidth_limiter = limiter def fetch_snapshot(self, *, preferred_region: str = "Europe") -> DepotSnapshot: """Fetch the live depot and build manifests, trying trusted regional fallbacks.""" configuration: Mapping[str, object] | None = None configuration_error: DepotError | None = None try: configuration = self._get_json(CDN_CONFIGURATION_URL) except DepotError as exc: configuration_error = exc candidates = _server_candidates(configuration, preferred_region) self._candidate_servers = tuple(server for _region, server in candidates) last_error = configuration_error for region, server in candidates: try: depot_manifest = self._get_json(_depot_manifest_url(server)) build = _positive_int(depot_manifest.get("Build"), "depot Build") build_manifest = self._get_json(_build_manifest_url(server, build)) return _parse_snapshot(region, server, depot_manifest, build_manifest) except DepotError as exc: last_error = exc if last_error is not None: raise DepotError(f"could not load the PristonTale EU depot: {last_error}") raise DepotError("could not load the PristonTale EU depot") def download_file( self, server: str, file: DepotFile, destination: Path, *, progress: Callable[[int], None] | None = None, ) -> None: """Download, resume, and verify one file before replacing its destination. The current publisher CDN normalizes a small number of legacy text files from CRLF to LF while its manifest still hashes the CRLF bytes. A candidate is accepted only when restoring bare LF line endings produces the manifest's exact size and SHA-1. """ servers = _ordered_download_servers(server, self._candidate_servers) temporary = _download_partial_path(destination) last_error: _RetryableDownloadError | None = None attempts = len(self._retry_delays) + 1 for attempt_index in range(attempts): attempt_server = servers[attempt_index % len(servers)] try: self._download_file_once( attempt_server, file, destination, attempt=attempt_index + 1, progress=progress, ) return except _RetryableDownloadError as exc: last_error = exc self._notify_download( DownloadTelemetry( kind="retry", file=file.path, server=attempt_server, attempt=attempt_index + 1, reason=str(exc), http_status=exc.http_status, ) ) if exc.discard_partial: temporary.unlink(missing_ok=True) if attempt_index >= len(self._retry_delays): break self._sleeper(self._retry_delays[attempt_index]) assert last_error is not None resume_hint = ( "; partial data was retained for the next update run" if temporary.is_file() and not temporary.is_symlink() else "" ) raise DepotError( f"download failed for {file.path} after {attempts} attempts: " f"{last_error}{resume_hint}" ) from last_error def _download_file_once( self, server: str, file: DepotFile, destination: Path, *, attempt: int, progress: Callable[[int], None] | None, ) -> None: temporary = _download_partial_path(destination) written = 0 digest = hashlib.sha1(usedforsecurity=False) if temporary.is_symlink(): raise DepotError(f"refusing to use symbolic-link partial file for {file.path}") if temporary.exists(): if not temporary.is_file(): raise DepotError(f"partial download is not a file for {file.path}") try: with temporary.open("rb") as partial_handle: while True: chunk = partial_handle.read(HASH_READ_SIZE) if not chunk: break written += len(chunk) digest.update(chunk) except OSError as exc: raise DepotError(f"cannot inspect partial download for {file.path}: {exc}") from exc if written > file.size: temporary.unlink() written = 0 digest = hashlib.sha1(usedforsecurity=False) elif written == file.size: if digest.hexdigest().casefold() == file.sha1.casefold(): self._notify_download( DownloadTelemetry( kind="resume", file=file.path, server=server, attempt=attempt, resumed_bytes=written, ) ) temporary.replace(destination) if progress is not None: progress(written) return temporary.unlink() written = 0 digest = hashlib.sha1(usedforsecurity=False) elif written and _recover_crlf_manifest_file(temporary, destination, file): self._notify_download( DownloadTelemetry( kind="resume", file=file.path, server=server, attempt=attempt, resumed_bytes=written, ) ) temporary.unlink(missing_ok=True) if progress is not None: progress(file.size) return if progress is not None: progress(written) requested_offset = written if requested_offset: self._notify_download( DownloadTelemetry( kind="resume", file=file.path, server=server, attempt=attempt, resumed_bytes=requested_offset, ) ) headers = _headers() if requested_offset: headers["Range"] = f"bytes={requested_offset}-" url = _file_url(server, file) request = Request(url, headers=headers, method="GET") self._notify_download( DownloadTelemetry( kind="request", file=file.path, server=server, attempt=attempt, ) ) request_started_at = time.monotonic() try: response = self._opener(request, timeout=self.timeout_seconds) except HTTPError as exc: status = exc.code exc.close() if status == 416 and requested_offset: raise _RetryableDownloadError( f"server rejected the saved offset for {file.path}", discard_partial=True, http_status=status, ) from exc raise _RetryableDownloadError( f"HTTP {status} from {server} for {file.path}", http_status=status, ) from exc except (OSError, HTTPException) as exc: raise _RetryableDownloadError( f"could not connect to {server} for {file.path}: {exc}" ) from exc response_status = getattr(response, "status", None) self._notify_download( DownloadTelemetry( kind="response", file=file.path, server=server, attempt=attempt, latency_seconds=max(0.0, time.monotonic() - request_started_at), http_status=(response_status if isinstance(response_status, int) else None), ) ) try: append = False if requested_offset: content_range = response.headers.get("Content-Range") status = getattr(response, "status", None) if status == 206 or content_range is not None: try: range_start, _range_end, range_total = _content_range(content_range) except DepotError as exc: raise _RetryableDownloadError( f"invalid resume response from {server} for {file.path}: {exc}" ) from exc if range_start != requested_offset or range_total != file.size: raise _RetryableDownloadError( f"resume response did not match {file.path}: " f"offset {range_start}/{requested_offset}, size {range_total}/{file.size}" ) append = True else: # A server may ignore Range and send a normal 200 response. In # that case safely restart this file using the full response body. written = 0 digest = hashlib.sha1(usedforsecurity=False) if progress is not None: progress(written) try: content_length = _content_length(response.headers.get("Content-Length")) except DepotError as exc: raise _RetryableDownloadError( f"invalid response length from {server} for {file.path}: {exc}" ) from exc remaining = file.size - written if content_length is not None and content_length > remaining: raise _RetryableDownloadError( f"download exceeds expected size for {file.path}: " f"server declared {content_length}, expected at most {remaining}" ) mode = "ab" if append else "wb" try: with temporary.open(mode) as handle: while True: try: chunk = response.read(READ_SIZE) except (OSError, HTTPException) as exc: raise _RetryableDownloadError( f"could not read {file.path} from {server}: {exc}" ) from exc if not chunk: break written += len(chunk) if written > file.size: raise _RetryableDownloadError( f"download exceeds expected size for {file.path}", discard_partial=True, ) if self._bandwidth_limiter is not None: self._bandwidth_limiter.consume(len(chunk)) digest.update(chunk) write_started_at = time.monotonic() handle.write(chunk) self._notify_download( DownloadTelemetry( kind="chunk", file=file.path, server=server, attempt=attempt, bytes_written=len(chunk), write_seconds=max( 0.0, time.monotonic() - write_started_at, ), ) ) if progress is not None: progress(written) except _RetryableDownloadError: raise except OSError as exc: raise DepotError(f"cannot write partial download for {file.path}: {exc}") from exc if written != file.size or digest.hexdigest().casefold() != file.sha1.casefold(): if _recover_crlf_manifest_file(temporary, destination, file): temporary.unlink(missing_ok=True) if progress is not None: progress(file.size) return if written != file.size: raise _RetryableDownloadError( f"incomplete response from {server} for {file.path}: " f"{written}/{file.size} bytes" ) raise _RetryableDownloadError( f"SHA-1 mismatch for {file.path} from {server}", discard_partial=True, ) temporary.replace(destination) finally: try: response.close() except (OSError, HTTPException): pass def _notify_download(self, event: DownloadTelemetry) -> None: observer = self._download_observer if observer is not None: # noinspection PyBroadException try: observer(event) except Exception: # Telemetry must never weaken verified download behavior. pass def _get_json(self, url: str) -> Mapping[str, object]: request = Request(url, headers=_headers(), method="GET") try: response = self._opener(request, timeout=self.timeout_seconds) except (OSError, HTTPException) as exc: raise DepotError(f"request failed for {url}: {exc}") from exc payload = b"" try: payload = response.read() except (OSError, HTTPException) as exc: raise DepotError(f"could not read {url}: {exc}") from exc finally: try: response.close() except (OSError, HTTPException): pass try: value = json.loads(payload.decode("utf-8-sig")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise DepotError(f"invalid JSON from {url}") from exc if not isinstance(value, dict): raise DepotError(f"unexpected JSON root from {url}") return value def plan_update( game_directory: Path, snapshot: DepotSnapshot, *, repair: bool = False, ) -> UpdatePlan: """Inspect a local game tree without writing to it.""" root = _prepare_game_root(game_directory, create=False) downloads: list[PlannedDownload] = [] hash_verified_paths: list[str] = [] for file in snapshot.files: target = _safe_target(root, file.path) if target.is_symlink(): downloads.append(PlannedDownload(file, "symbolic link")) continue if not target.is_file(): downloads.append(PlannedDownload(file, "missing")) continue try: if target.stat().st_size != file.size: downloads.append(PlannedDownload(file, "size mismatch")) continue except OSError as exc: raise DepotError(f"cannot inspect {file.path}: {exc}") from exc if ( repair or file.build == snapshot.build or file.path.casefold() in ALWAYS_VERIFY_PATHS ): if _sha1(target).casefold() != file.sha1.casefold(): downloads.append(PlannedDownload(file, "hash mismatch")) else: hash_verified_paths.append(file.path) deletes: list[str] = [] for relative_path in snapshot.deleted_files: target = _safe_target(root, relative_path) if target.is_symlink() or target.is_file(): deletes.append(relative_path) elif target.exists(): raise DepotError( f"refusing to delete non-file depot target: {relative_path}" ) return UpdatePlan( downloads=tuple(downloads), deletes=tuple(deletes), files_total=len(snapshot.files), bytes_total=sum(item.file.size for item in downloads), hash_verified_paths=tuple(hash_verified_paths), ) def sync_managed_game( paths: AppPaths, *, repair: bool = False, quarantine_unmanaged: bool = False, dry_run: bool = False, progress: ProgressCallback | None = None, client: DepotClient | None = None, download_options: DownloadOptions | None = None, ) -> DepotSyncResult: """Synchronize the managed game directory using only official public data.""" operation_started = time.perf_counter() operation = ( "repair" if repair else "update" if paths.source_game.is_dir() else "install" ) telemetry_enabled = not dry_run paths.create_base_directories() depot_client = client or DepotClient() previous_file_catalog = read_current_depot_file_catalog(paths) current_version = _completed_depot_version(paths.native_depot_state) manifest_started = time.perf_counter() if operation == "update": _send_depot_telemetry( "update_check_started", metadata={"current_version": current_version}, context={"operation": operation, "stage": "manifest"}, enabled=telemetry_enabled, ) _emit_progress(progress, "manifest", 0, 0, 0, 0, None) try: snapshot = depot_client.fetch_snapshot() except BaseException as exc: _send_depot_telemetry( "download_failed", severity="error", message=str(exc), metadata={ "resource_type": "manifest", "duration_ms": _elapsed_milliseconds(manifest_started), "failure_type": classify_network_failure(exc), }, context={"operation": operation, "stage": "manifest"}, enabled=telemetry_enabled, ) raise manifest_duration_ms = _elapsed_milliseconds(manifest_started) scan_started = time.perf_counter() quarantine = QuarantineResult(0, None) try: if quarantine_unmanaged and not dry_run: quarantine = _quarantine_unmanaged_files(paths, snapshot) plan = plan_update(paths.source_game, snapshot, repair=repair) except BaseException as exc: _send_depot_telemetry( "file_validation_failed", severity="error", message=str(exc), metadata={ "duration_ms": _elapsed_milliseconds(scan_started), "repair": repair, }, context={"operation": operation, "stage": "scan"}, enabled=telemetry_enabled, ) raise scan_duration_ms = _elapsed_milliseconds(scan_started) _emit_progress( progress, "planned", 0, len(plan.downloads), 0, plan.bytes_total, None, ) plan_metadata = { "current_version": current_version, "target_version": snapshot.version, "changed_file_count": len(plan.downloads) + len(plan.deletes), "files_checked": plan.files_total, "files_missing": sum(item.reason == "missing" for item in plan.downloads), "files_invalid": sum(item.reason != "missing" for item in plan.downloads), "files_replaced": len(plan.downloads), "download_bytes": plan.bytes_total, "manifest_duration_ms": manifest_duration_ms, "scan_duration_ms": scan_duration_ms, } if operation == "update": _send_depot_telemetry( "update_available" if plan.downloads or plan.deletes else "update_not_required", metadata=plan_metadata, context={"operation": operation, "stage": "scan"}, enabled=telemetry_enabled, ) elif operation == "repair": _send_depot_telemetry( "repair_scan_completed", metadata=plan_metadata, context={"operation": operation, "stage": "scan"}, enabled=telemetry_enabled, ) if dry_run: return _sync_result( snapshot, plan, 0, 0, 0, repair, True, quarantine=quarantine, manifest_duration_ms=manifest_duration_ms, scan_duration_ms=scan_duration_ms, update_duration_ms=_elapsed_milliseconds(operation_started), ) root = _prepare_game_root(paths.source_game, create=True) download_metrics = _download_plan( depot_client, snapshot, root, plan, progress, download_options or DownloadOptions(), operation=operation, telemetry_enabled=telemetry_enabled, ) apply_started = time.perf_counter() if operation == "update": _send_depot_telemetry( "update_apply_started", metadata={ "target_version": snapshot.version, "changed_file_count": len(plan.downloads) + len(plan.deletes), }, context={"operation": operation, "stage": "apply"}, enabled=telemetry_enabled, ) try: deleted = _apply_deletions(root, plan.deletes) created = _apply_create_if_missing(root, snapshot.create_if_missing) verified_paths = { *plan.hash_verified_paths, *(download.file.path for download in plan.downloads), } cache_depot_file_catalog( paths.depot_file_catalog, root, snapshot, verified_paths=verified_paths, previous=previous_file_catalog, ) cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot) apply_duration_ms = _elapsed_milliseconds(apply_started) update_duration_ms = _elapsed_milliseconds(operation_started) result = _sync_result( snapshot, plan, len(plan.downloads), plan.bytes_total, deleted, repair, False, created=created, quarantine=quarantine, manifest_duration_ms=manifest_duration_ms, scan_duration_ms=scan_duration_ms, apply_duration_ms=apply_duration_ms, update_duration_ms=update_duration_ms, download_metrics=download_metrics, ) _write_json(paths.native_depot_state, asdict(result)) except BaseException as exc: _send_depot_telemetry( "filesystem_error", severity="error", message=str(exc), metadata={ "duration_ms": _elapsed_milliseconds(apply_started), "target_version": snapshot.version, }, context={"operation": operation, "stage": "apply"}, enabled=telemetry_enabled, ) raise if operation == "update": _send_depot_telemetry( "update_apply_completed", metadata={ **plan_metadata, "apply_duration_ms": result.apply_duration_ms, "download_duration_ms": result.download_duration_ms, "update_duration_ms": result.update_duration_ms, "network_bytes_downloaded": result.network_bytes_downloaded, "average_download_mbps": result.download_average_mbps, "retry_count": result.download_retry_count, "resume_used": result.download_resume_used, }, context={"operation": operation, "stage": "apply"}, enabled=telemetry_enabled, ) _emit_progress( progress, "complete", len(plan.downloads), len(plan.downloads), plan.bytes_total, plan.bytes_total, None, ) return result def _quarantine_unmanaged_files( paths: AppPaths, snapshot: DepotSnapshot, ) -> QuarantineResult: """Move non-depot import contents aside without following links.""" root = _prepare_game_root(paths.source_game, create=False) if not root.exists(): return QuarantineResult(0, None) allowed = { file.path for file in snapshot.files } | { entry.path for entry in snapshot.create_if_missing } candidates: list[tuple[Path, str]] = [] def walk_error(error: OSError) -> None: raise DepotError(f"cannot inspect imported game tree: {error}") from error for directory, directory_names, file_names in os.walk( root, topdown=True, followlinks=False, onerror=walk_error, ): parent = Path(directory) for name in tuple(directory_names): path = parent / name try: status = path.lstat() except OSError as exc: raise DepotError(f"cannot inspect imported entry {path}: {exc}") from exc if stat.S_ISDIR(status.st_mode): continue directory_names.remove(name) candidates.append((path, path.relative_to(root).as_posix())) for name in file_names: path = parent / name relative = path.relative_to(root).as_posix() try: status = path.lstat() except OSError as exc: raise DepotError(f"cannot inspect imported entry {path}: {exc}") from exc if not stat.S_ISREG(status.st_mode): candidates.append((path, relative)) continue if relative in allowed: continue if _safe_imported_user_file(path, relative): continue candidates.append((path, relative)) if not candidates: return QuarantineResult(0, None) base = paths.game_import_quarantine.resolve(strict=False) data_root = paths.data.resolve(strict=False) try: base.relative_to(data_root) except ValueError as exc: raise DepotError(f"import quarantine is outside the data root: {base}") from exc stamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S") quarantine = base / f"{stamp}-{uuid4().hex}" files_root = quarantine / "files" files_root.mkdir(parents=True, exist_ok=False) manifest = { "schema": 1, "complete": False, "created_at": datetime.now().astimezone().isoformat(), "source": str(root), "files": [relative for _path, relative in candidates], } _write_json(quarantine / "quarantine.json", manifest) for source, relative in candidates: destination = files_root.joinpath(*PurePosixPath(relative).parts) destination.parent.mkdir(parents=True, exist_ok=True) try: source.replace(destination) except OSError as exc: raise DepotError( f"could not quarantine imported entry {relative}: {exc}" ) from exc manifest["complete"] = True _write_json(quarantine / "quarantine.json", manifest) return QuarantineResult(len(candidates), str(quarantine)) def _safe_imported_user_file(path: Path, relative: str) -> bool: parts = PurePosixPath(relative).parts if not parts or parts[0].casefold() != "save": return False if path.suffix.casefold() in UNMANAGED_EXECUTABLE_SUFFIXES: return False try: with path.open("rb") as handle: prefix = handle.read(4) except OSError as exc: raise DepotError(f"cannot inspect imported user file {path}: {exc}") from exc return not ( prefix.startswith(b"MZ") or prefix.startswith(b"\x7fELF") or prefix.startswith(b"#!") ) def _download_plan( client: DepotClient, depot_snapshot: DepotSnapshot, root: Path, plan: UpdatePlan, progress: ProgressCallback | None, options: DownloadOptions, *, operation: str = "update", telemetry_enabled: bool = True, ) -> DepotDownloadMetrics: if not plan.downloads: return DepotDownloadMetrics(mirror_id=_server_display_name(depot_snapshot.server)) lock = threading.Lock() emit_lock = threading.Lock() current_bytes: dict[str, int] = {} current_servers: dict[str, str] = {} completed_files = 0 completed_bytes = 0 transferred_bytes = 0 retry_count = 0 retry_reason: str | None = None resume_offsets: dict[str, int] = {} mirrors: set[str] = set() last_http_status: int | None = None retry_events_sent = 0 resume_events_sent = 0 active_workers = 0 target_workers = 0 last_progress_emitted_at = 0.0 started_at = time.monotonic() transfer_samples: deque[tuple[float, int]] = deque([(started_at, 0)]) controller = AdaptiveConcurrencyController(options) bandwidth_limit = options.effective_bandwidth_limit limiter = BandwidthLimiter(bandwidth_limit) if bandwidth_limit is not None else None previous_observer = client.download_observer previous_limiter = client.bandwidth_limiter started_event, completed_event = _download_event_names(operation) _send_depot_telemetry( started_event, metadata={ "resource_type": "repair_file" if operation == "repair" else "game", "download_bytes": plan.bytes_total, "file_count": len(plan.downloads), "mirror_id": _server_display_name(depot_snapshot.server), "cdn_region": depot_snapshot.region, }, context={"operation": operation, "stage": "download"}, enabled=telemetry_enabled, ) def emit_download_progress( current_file: str, *, force: bool = False, ) -> None: """Throttle progress I/O without holding the download-state lock. The workers can produce many chunk and file-completion updates per second. The callback writes to a subprocess pipe for the native dashboard, so retaining every event would make that pipe and its UI consumer the bottleneck. """ nonlocal last_progress_emitted_at if progress is None: return with emit_lock: now = time.monotonic() if not force and now - last_progress_emitted_at < PROGRESS_UPDATE_INTERVAL_SECONDS: return last_progress_emitted_at = now with lock: bytes_completed = min( plan.bytes_total, completed_bytes + sum(current_bytes.values()), ) _prune_transfer_samples(transfer_samples, now) transfer_rate = _sample_rate(transfer_samples) bytes_remaining = max(0, plan.bytes_total - bytes_completed) eta_seconds = ( bytes_remaining / transfer_rate if transfer_rate is not None and transfer_rate > 0 else None ) event = SyncProgress( "downloading", completed_files, len(plan.downloads), bytes_completed, plan.bytes_total, current_file, transferred_bytes, bytes_remaining, transfer_rate, eta_seconds, current_servers.get( current_file, _server_display_name(depot_snapshot.server), ), retry_count, retry_reason, active_workers, target_workers, ) progress(event) def observe_download(event: DownloadTelemetry) -> None: nonlocal last_http_status, resume_events_sent, retry_count nonlocal retry_events_sent, retry_reason, transferred_bytes if event.kind == "response" and event.latency_seconds is not None: controller.record_response(event.latency_seconds) elif event.kind == "retry": controller.record_retry() elif event.kind == "chunk": controller.record_chunk(event.bytes_written, event.write_seconds) force = event.kind == "retry" emit_retry = False emit_resume = False if event.kind in {"request", "response", "retry", "resume", "chunk"}: with lock: current_servers[event.file] = _server_display_name(event.server) mirrors.add(_server_display_name(event.server)) if event.kind == "chunk": transferred_bytes += event.bytes_written now = time.monotonic() transfer_samples.append((now, transferred_bytes)) _prune_transfer_samples(transfer_samples, now) if event.kind == "retry": retry_count += 1 retry_reason = event.reason if retry_events_sent < 5: retry_events_sent += 1 emit_retry = True if event.kind == "resume" and event.resumed_bytes > 0: prior_offset = resume_offsets.get(event.file, 0) resume_offsets[event.file] = max(prior_offset, event.resumed_bytes) if prior_offset == 0 and resume_events_sent < 5: resume_events_sent += 1 emit_resume = True if event.http_status is not None: last_http_status = event.http_status if previous_observer is not None: previous_observer(event) if emit_retry: _send_depot_telemetry( "download_retry", severity="warning", message=event.reason, metadata={ "resource_type": ( "repair_file" if operation == "repair" else "game" ), "file": event.file, "attempt": event.attempt, "failure_type": classify_network_failure(event.reason), "http_status": event.http_status, "mirror_id": _server_display_name(event.server), }, context={"operation": operation, "stage": "download_retry"}, enabled=telemetry_enabled, ) if emit_resume: _send_depot_telemetry( "download_resumed", metadata={ "resource_type": ( "repair_file" if operation == "repair" else "game" ), "file": event.file, "resumed_bytes": event.resumed_bytes, "mirror_id": _server_display_name(event.server), }, context={"operation": operation, "stage": "download_resume"}, enabled=telemetry_enabled, ) if event.kind != "chunk": emit_download_progress(event.file, force=force) def download(planned_download: PlannedDownload) -> None: target = _safe_target(root, planned_download.file.path) _ensure_target_parent(root, target) def report_file_bytes(written: int) -> None: with lock: current_bytes[planned_download.file.path] = written emit_download_progress(planned_download.file.path) client.download_file( depot_snapshot.server, planned_download.file, target, progress=report_file_bytes if progress is not None else None, ) def metrics_snapshot() -> DepotDownloadMetrics: with lock: duration_seconds = max(0.0, time.monotonic() - started_at) primary_mirror = _server_display_name(depot_snapshot.server) return DepotDownloadMetrics( duration_ms=round(duration_seconds * 1000), transferred_bytes=transferred_bytes, retry_count=retry_count, resume_used=bool(resume_offsets), resumed_bytes=sum(resume_offsets.values()), average_mbps=_average_megabits_per_second( transferred_bytes, duration_seconds, ), mirror_id=primary_mirror, http_status=last_http_status, ) maximum_workers = min(options.effective_maximum_workers, len(plan.downloads)) # Start large transfers first so they overlap the small-file request workload # instead of becoming a single-connection tail at the end of an install. remaining_items: deque[PlannedDownload] = deque( sorted(plan.downloads, key=lambda item: item.file.size, reverse=True) ) client.download_observer = observe_download client.bandwidth_limiter = limiter try: with ThreadPoolExecutor(max_workers=maximum_workers) as executor: future_items: dict[Future[None], PlannedDownload] = {} def submit_until(worker_target: int) -> None: nonlocal active_workers, target_workers with lock: target_workers = worker_target while remaining_items and len(future_items) < worker_target: next_item = remaining_items.popleft() with lock: active_workers += 1 try: submitted_future = executor.submit(download, next_item) except BaseException: with lock: active_workers -= 1 remaining_items.appendleft(next_item) raise future_items[submitted_future] = next_item try: initial_target = controller.target_workers( len(remaining_items), plan.bytes_total, ) submit_until(initial_target) while future_items: done, _pending = wait( tuple(future_items), timeout=ADAPTIVE_EVALUATION_INTERVAL_SECONDS, return_when=FIRST_COMPLETED, ) completed_items: list[PlannedDownload] = [] for future in done: completed_item = future_items.pop(future) future.result() completed_items.append(completed_item) for completed_item in completed_items: with lock: active_workers -= 1 completed_files += 1 completed_bytes += completed_item.file.size current_bytes.pop(completed_item.file.path, None) is_final_download = completed_files == len(plan.downloads) emit_download_progress( completed_item.file.path, force=is_final_download, ) with lock: current_servers.pop(completed_item.file.path, None) with lock: in_progress_bytes = sum(current_bytes.values()) remaining_file_count = len(remaining_items) + len(future_items) remaining_byte_count = max( 0, plan.bytes_total - completed_bytes - in_progress_bytes, ) next_target = controller.target_workers( remaining_file_count, remaining_byte_count, ) submit_until(next_target) except BaseException: for future in future_items: future.cancel() raise except BaseException as exc: metrics = metrics_snapshot() _send_depot_telemetry( "download_failed", severity="error", message=str(exc), metadata={ "resource_type": "repair_file" if operation == "repair" else "game", "download_bytes": metrics.transferred_bytes, "content_bytes": plan.bytes_total, "duration_ms": metrics.duration_ms, "average_mbps": metrics.average_mbps, "retry_count": metrics.retry_count, "resume_used": metrics.resume_used, "resumed_bytes": metrics.resumed_bytes, "http_status": metrics.http_status, "failure_type": classify_network_failure(exc), "mirror_id": metrics.mirror_id, "mirrors_tried": sorted(mirrors), }, context={"operation": operation, "stage": "download"}, enabled=telemetry_enabled, ) raise finally: client.download_observer = previous_observer client.bandwidth_limiter = previous_limiter metrics = metrics_snapshot() _send_depot_telemetry( completed_event, metadata={ "resource_type": "repair_file" if operation == "repair" else "game", "download_bytes": metrics.transferred_bytes, "content_bytes": plan.bytes_total, "duration_ms": metrics.duration_ms, "average_mbps": metrics.average_mbps, "retry_count": metrics.retry_count, "resume_used": metrics.resume_used, "resumed_bytes": metrics.resumed_bytes, "http_status": metrics.http_status, "mirror_id": metrics.mirror_id, "mirrors_used": sorted(mirrors), "file_count": len(plan.downloads), }, context={"operation": operation, "stage": "download"}, enabled=telemetry_enabled, ) return metrics def _apply_deletions(root: Path, relative_paths: Iterable[str]) -> int: deleted = 0 for relative_path in relative_paths: target = _safe_target(root, relative_path) if target.is_symlink() or target.is_file(): target.unlink() deleted += 1 return deleted def _apply_create_if_missing( root: Path, entries: Iterable[CreateIfMissing], ) -> int: created = 0 for entry in entries: target = _safe_target(root, entry.path) if target.is_symlink(): raise DepotError(f"refusing to follow symbolic link for {entry.path}") if target.exists(): continue content = _decode_create_if_missing(entry) _ensure_target_parent(root, target) temporary = target.with_name(f".{target.name}.{uuid4().hex}.partial") try: with temporary.open("xb") as handle: handle.write(content) handle.flush() os.fsync(handle.fileno()) temporary.replace(target) except BaseException: temporary.unlink(missing_ok=True) raise created += 1 return created def _catalog_entry( file: DepotFile, state: os.stat_result, *, verified: bool, ) -> DepotFileCatalogEntry: return DepotFileCatalogEntry( path=file.path, size=file.size, sha1=file.sha1.casefold(), device=state.st_dev, inode=state.st_ino, mtime_ns=state.st_mtime_ns, ctime_ns=state.st_ctime_ns, verified=verified, ) def _scan_game_tree(root: Path) -> dict[str, _ScannedGameEntry]: """Inventory files without following symlinks or resolving every manifest path.""" entries: dict[str, _ScannedGameEntry] = {} def scan(directory: Path, relative_parent: PurePosixPath) -> None: children = _scan_directory(directory) for child in children: relative = relative_parent / child.name key = relative.as_posix() try: state = child.stat(follow_symlinks=False) except OSError as exc: raise DepotError(f"cannot inspect managed game path {key}: {exc}") from exc path = Path(child.path) if stat.S_ISLNK(state.st_mode): entries[key] = _ScannedGameEntry(path, state, "symbolic link") elif stat.S_ISDIR(state.st_mode): scan(path, relative) elif stat.S_ISREG(state.st_mode): entries[key] = _ScannedGameEntry(path, state, "file") else: entries[key] = _ScannedGameEntry(path, state, "not a regular file") scan(root, PurePosixPath()) return entries def _scan_directory(directory: Path) -> list[os.DirEntry[str]]: try: with os.scandir(directory) as iterator: return list(iterator) except OSError as exc: raise DepotError(f"cannot scan managed game directory {directory}: {exc}") from exc def _catalog_entry_from_cached( entry: DepotFileCatalogEntry, state: os.stat_result, ) -> DepotFileCatalogEntry: return DepotFileCatalogEntry( path=entry.path, size=entry.size, sha1=entry.sha1, device=state.st_dev, inode=state.st_ino, mtime_ns=state.st_mtime_ns, ctime_ns=state.st_ctime_ns, verified=True, ) def _catalog_identity_matches( entry: DepotFileCatalogEntry, state: os.stat_result, ) -> bool: return ( entry.device == state.st_dev and entry.inode == state.st_ino and entry.size == state.st_size and entry.mtime_ns == state.st_mtime_ns and entry.ctime_ns == state.st_ctime_ns ) def _depot_file_catalog_payload(catalog: DepotFileCatalog) -> dict[str, object]: """Use compact rows because this cache contains tens of thousands of files.""" return { "schema": DEPOT_FILE_CATALOG_SCHEMA, "build": catalog.build, "created_at": catalog.created_at, "files": [ [ entry.path, entry.size, entry.sha1, entry.device, entry.inode, entry.mtime_ns, entry.ctime_ns, entry.verified, ] for entry in catalog.files ], } def _parse_depot_file_catalog_entry(value: object) -> DepotFileCatalogEntry: if not isinstance(value, list) or len(value) != 8: raise DepotError("cached depot file entry must be an eight-value row") path = _normalise_relative_path(value[0], "cached depot file path") size = _non_negative_int(value[1], f"cached depot file size for {path}") sha1 = _string(value[2], f"cached depot file hash for {path}").casefold() if not SHA1_PATTERN.fullmatch(sha1): raise DepotError(f"invalid cached SHA-1 hash for {path}") identity: dict[str, int] = {} for field_name, field_value in zip( ("device", "inode", "mtime_ns", "ctime_ns"), value[3:7], strict=True, ): identity[field_name] = _non_negative_int( field_value, f"cached depot file {field_name} for {path}", ) verified = value[7] if not isinstance(verified, bool): raise DepotError(f"cached depot verification flag is invalid for {path}") return DepotFileCatalogEntry( path=path, size=size, sha1=sha1, device=identity["device"], inode=identity["inode"], mtime_ns=identity["mtime_ns"], ctime_ns=identity["ctime_ns"], verified=verified, ) def _completed_depot_build(path: Path) -> int | None: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): return None if not isinstance(value, dict) or value.get("dry_run") is not False: return None build = value.get("build") if isinstance(build, bool) or not isinstance(build, int) or build <= 0: return None return build def _completed_depot_version(path: Path) -> str | None: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): return None if not isinstance(value, dict) or value.get("dry_run") is not False: return None version = value.get("version") return version[:32] if isinstance(version, str) and version else None def _elapsed_milliseconds(started: float) -> int: return max(0, round((time.perf_counter() - started) * 1000)) def _average_megabits_per_second(byte_count: int, duration_seconds: float) -> float | None: if byte_count <= 0 or duration_seconds <= 0: return None return round((byte_count * 8) / duration_seconds / 1_000_000, 3) def _download_event_names(operation: str) -> tuple[str, str]: if operation == "install": return "install_download_started", "install_download_completed" if operation == "update": return "update_download_started", "update_download_completed" return "download_started", "download_completed" def _send_depot_telemetry( event: str, *, severity: str = "info", message: str | None = None, metadata: Mapping[str, object] | None = None, context: Mapping[str, object] | None = None, enabled: bool = True, ) -> None: if not enabled: return telemetry = current_telemetry() if telemetry is None: return try: telemetry.send_event( event, severity=severity, message=message, metadata=metadata, context=context, ) except Exception: # Telemetry must never alter verified depot behavior. pass def _sync_result( snapshot: DepotSnapshot, plan: UpdatePlan, files_downloaded: int, bytes_downloaded: int, files_deleted: int, repair: bool, dry_run: bool, *, created: int = 0, quarantine: QuarantineResult = QuarantineResult(0, None), manifest_duration_ms: int = 0, scan_duration_ms: int = 0, apply_duration_ms: int = 0, update_duration_ms: int = 0, download_metrics: DepotDownloadMetrics | None = None, ) -> DepotSyncResult: metrics = download_metrics or DepotDownloadMetrics() return DepotSyncResult( region=snapshot.region, server=snapshot.server, name=snapshot.name, build=snapshot.build, version=snapshot.version, files_total=plan.files_total, files_planned=len(plan.downloads), bytes_planned=plan.bytes_total, files_downloaded=files_downloaded, bytes_downloaded=bytes_downloaded, files_deleted=files_deleted, files_created=created, files_quarantined=quarantine.files, quarantine_directory=quarantine.directory, repair=repair, dry_run=dry_run, completed_at=datetime.now().astimezone().isoformat(), manifest_duration_ms=manifest_duration_ms, scan_duration_ms=scan_duration_ms, download_duration_ms=metrics.duration_ms, apply_duration_ms=apply_duration_ms, update_duration_ms=update_duration_ms, network_bytes_downloaded=metrics.transferred_bytes, download_retry_count=metrics.retry_count, download_resume_used=metrics.resume_used, download_average_mbps=metrics.average_mbps, mirror_id=metrics.mirror_id, ) def _parse_snapshot( region: str, server: str, depot_manifest: Mapping[str, object], build_manifest: Mapping[str, object], ) -> DepotSnapshot: build = _positive_int(depot_manifest.get("Build"), "depot Build") build_manifest_build = _positive_int(build_manifest.get("b"), "build manifest b") if build_manifest_build != build: raise DepotError( f"depot build {build} does not match build manifest {build_manifest_build}" ) name = _string(depot_manifest.get("Name"), "depot Name") version = _string(depot_manifest.get("BuildVersionText"), "depot BuildVersionText") files_value = build_manifest.get("f") if not isinstance(files_value, list): raise DepotError("build manifest f must be an array") files: list[DepotFile] = [] seen_paths: set[str] = set() for value in files_value: if not isinstance(value, dict): raise DepotError("build manifest contains a non-object file entry") path = _normalise_relative_path(value.get("p"), "file path") if path.casefold() in seen_paths: raise DepotError(f"build manifest has duplicate file path: {path}") seen_paths.add(path.casefold()) file_build = _positive_int(value.get("b"), f"file build for {path}") size = _non_negative_int(value.get("s"), f"file size for {path}") sha1 = _string(value.get("h"), f"file hash for {path}") if not SHA1_PATTERN.fullmatch(sha1): raise DepotError(f"invalid SHA-1 hash for {path}") files.append(DepotFile(path, file_build, size, sha1.casefold())) deleted = _parse_deleted_paths(build_manifest.get("df"), seen_paths) create_if_missing = _parse_create_if_missing(depot_manifest.get("CreateIfMissing")) start_options, default_start_option = _parse_start_options( depot_manifest.get("StartOptions"), depot_manifest.get("DefaultStartOption"), ) canonical_file_paths = {file.path.casefold(): file.path for file in files} start_options = tuple( DepotStartOption( name=option.name, path=canonical_file_paths.get(option.path.casefold(), option.path), arguments=option.arguments, ) for option in start_options ) return DepotSnapshot( region=region, server=server, name=name, build=build, version=version, files=tuple(files), deleted_files=deleted, create_if_missing=create_if_missing, start_options=start_options, default_start_option=default_start_option, ) def _parse_deleted_paths(value: object, known_paths: set[str]) -> tuple[str, ...]: if value is None: return () if not isinstance(value, list): raise DepotError("build manifest df must be an array") paths: list[str] = [] seen: set[str] = set() for item in value: path = _normalise_relative_path(item, "deleted file path") folded = path.casefold() if folded in known_paths: raise DepotError(f"manifest lists a file as both present and deleted: {path}") if folded in seen: continue seen.add(folded) paths.append(path) return tuple(paths) def _parse_create_if_missing(value: object) -> tuple[CreateIfMissing, ...]: if value is None: return () if not isinstance(value, list): raise DepotError("CreateIfMissing must be an array") entries: list[CreateIfMissing] = [] seen: set[str] = set() for item in value: if not isinstance(item, dict): raise DepotError("CreateIfMissing contains a non-object entry") path = _normalise_relative_path(item.get("Path"), "CreateIfMissing path") folded = path.casefold() if folded in seen: raise DepotError(f"CreateIfMissing has duplicate path: {path}") seen.add(folded) content = _string(item.get("Content"), f"CreateIfMissing content for {path}") entry = CreateIfMissing(path, content) _decode_create_if_missing(entry) entries.append(entry) return tuple(entries) def _parse_start_options( value: object, default_value: object, ) -> tuple[tuple[DepotStartOption, ...], str]: """Validate the launch targets published by the official depot manifest.""" # Older test fixtures and historical manifests did not expose StartOptions. # Babel remains the launcher-safe compatibility default for that case. if value is None: return (DEFAULT_DEPOT_START_OPTION,), DEFAULT_DEPOT_START_OPTION.name if not isinstance(value, list) or not value: raise DepotError("StartOptions must be a non-empty array") options: list[DepotStartOption] = [] seen_names: set[str] = set() for item in value: if not isinstance(item, dict): raise DepotError("StartOptions contains a non-object entry") name = _start_option_text(item.get("Name"), "StartOptions Name", 64) folded_name = name.casefold() if folded_name in seen_names: raise DepotError(f"StartOptions has duplicate realm name: {name}") seen_names.add(folded_name) path = _normalise_relative_path(item.get("Path"), f"StartOptions path for {name}") if PurePosixPath(path).suffix.casefold() != ".exe": raise DepotError(f"StartOptions path is not an executable for {name}: {path}") arguments = _start_option_text( item.get("Arguments"), f"StartOptions arguments for {name}", 2048, ) options.append(DepotStartOption(name=name, path=path, arguments=arguments)) if default_value is None: default_name = next( ( option.name for option in options if option.name.casefold() == DEFAULT_DEPOT_START_OPTION.name.casefold() ), options[0].name, ) else: default_name = _start_option_text( default_value, "DefaultStartOption", 64, ) matching_default = next( (option.name for option in options if option.name.casefold() == default_name.casefold()), None, ) if matching_default is None: raise DepotError( f"DefaultStartOption does not match an advertised realm: {default_name}" ) default_name = matching_default return tuple(options), default_name def _start_option_text(value: object, field: str, maximum_length: int) -> str: if not isinstance(value, str): raise DepotError(f"{field} must be a string") text = value.strip() if not text or len(text) > maximum_length or any(ord(character) < 32 for character in text): raise DepotError(f"invalid {field}") return text def _decode_create_if_missing(entry: CreateIfMissing) -> bytes: try: content = b64decode(entry.content_base64, validate=True) except (ValueError, BinAsciiError) as exc: raise DepotError(f"invalid base64 create-if-missing entry: {entry.path}") from exc if len(content) > MAX_CREATE_IF_MISSING_BYTES: raise DepotError(f"create-if-missing entry is too large: {entry.path}") return content def _server_candidates( configuration: Mapping[str, object] | None, preferred_region: str, ) -> tuple[tuple[str, str], ...]: entries: list[tuple[str, str]] = [] if configuration is not None: regions = configuration.get("Regions") default_region = configuration.get("DefaultRegion") if isinstance(regions, list): ordered = [preferred_region] if isinstance(default_region, str) and default_region not in ordered: ordered.append(default_region) for region_name in ordered: for region in regions: if not isinstance(region, dict) or region.get("Name") != region_name: continue servers = region.get("Servers") if isinstance(servers, list): for server in servers: if not isinstance(server, str): continue try: entries.append((region_name, _normalise_server(server))) except DepotError: continue seen: set[str] = set() unique: list[tuple[str, str]] = [] for region, server in [*entries, *(("Fallback", value) for value in FALLBACK_SERVERS)]: if server in seen: continue seen.add(server) unique.append((region, server)) return tuple(unique) def _normalise_server(value: str) -> str: server = value.strip().strip("/") if not server or "://" in server or any(character.isspace() for character in server): raise DepotError(f"invalid CDN server: {value!r}") parsed = urlparse(f"https://{server}") if not parsed.netloc or parsed.query or parsed.fragment or "@" in parsed.netloc: raise DepotError(f"invalid CDN server: {value!r}") parts = PurePosixPath(parsed.path).parts if any(part in (".", "..") for part in parts): raise DepotError(f"invalid CDN server path: {value!r}") return f"{parsed.netloc}{parsed.path.rstrip('/')}" def _depot_manifest_url(server: str) -> str: return f"https://{server}/depots/{DEPOT_ID}/manifest.json" def _build_manifest_url(server: str, build: int) -> str: return f"https://{server}/depots/{DEPOT_ID}/builds/{build}.json" def _file_url(server: str, file: DepotFile) -> str: return ( f"https://{server}/depots/{DEPOT_ID}/builds/{file.build}/" f"{quote(file.path, safe='/')}" ) def _ordered_download_servers( primary: str, candidates: Iterable[str], ) -> tuple[str, ...]: ordered: list[str] = [] seen: set[str] = set() for value in (primary, *candidates): server = _normalise_server(value) if server in seen: continue seen.add(server) ordered.append(server) return tuple(ordered) def _download_partial_path(destination: Path) -> Path: return destination.with_name(f".{destination.name}.ptlinux.partial") def _recover_crlf_manifest_file( temporary: Path, destination: Path, file: DepotFile, ) -> bool: """Restore publisher-normalized legacy text only on an exact hash match.""" try: if temporary.stat().st_size > MAX_LINE_ENDING_RECOVERY_BYTES: return False data = temporary.read_bytes() except OSError as exc: raise DepotError(f"could not inspect downloaded {file.path}: {exc}") from exc if b"\r" in data: return False recovered = data.replace(b"\n", b"\r\n") if len(recovered) != file.size: return False if ( hashlib.sha1(recovered, usedforsecurity=False).hexdigest().casefold() != file.sha1.casefold() ): return False recovered_temporary = destination.with_name( f".{destination.name}.{os.getpid()}.{uuid4().hex}.recovered" ) try: with recovered_temporary.open("xb") as handle: handle.write(recovered) handle.flush() os.fsync(handle.fileno()) recovered_temporary.replace(destination) except BaseException: recovered_temporary.unlink(missing_ok=True) raise return True def _prepare_game_root(game_directory: Path, *, create: bool) -> Path: root = game_directory.expanduser() if root.is_symlink(): raise DepotError(f"managed game directory must not be a symbolic link: {root}") if root.exists() and not root.is_dir(): raise DepotError(f"managed game path is not a directory: {root}") if create: root.mkdir(parents=True, exist_ok=True) return root.resolve(strict=False) def _safe_target(root: Path, relative_path: str) -> Path: parts = _relative_parts(relative_path, "depot path") target = root.joinpath(*parts) try: target.resolve(strict=False).relative_to(root) except ValueError as exc: raise DepotError(f"depot path escapes the managed game directory: {relative_path}") from exc return target def _ensure_target_parent(root: Path, target: Path) -> None: try: relative_parent = target.parent.relative_to(root) except ValueError as exc: raise DepotError(f"target escapes managed game directory: {target}") from exc cursor = root for part in relative_parent.parts: cursor = cursor / part if cursor.is_symlink(): raise DepotError(f"refusing to traverse symbolic link in game tree: {cursor}") if cursor.exists(): if not cursor.is_dir(): raise DepotError(f"cannot create game directory over file: {cursor}") else: try: cursor.mkdir() except FileExistsError: # Another download worker can create a shared parent after # ``exists()`` and before ``mkdir()``. Accept only the directory # that the losing worker expected; never accept a file or symlink. if cursor.is_symlink(): raise DepotError( f"refusing to traverse symbolic link in game tree: {cursor}" ) if not cursor.is_dir(): raise DepotError(f"cannot create game directory over file: {cursor}") except OSError as exc: raise DepotError(f"cannot create game directory {cursor}: {exc}") from exc def _normalise_relative_path(value: object, field: str) -> str: if not isinstance(value, str): raise DepotError(f"{field} must be a string") parts = _relative_parts(value, field) return "/".join(parts) def _relative_parts(value: str, field: str) -> tuple[str, ...]: if not value or "\x00" in value: raise DepotError(f"invalid {field}") normalised = value.replace("\\", "/") if normalised.startswith("/") or normalised.startswith("//"): raise DepotError(f"{field} must be relative: {value!r}") parts = PurePosixPath(normalised).parts if not parts or any(part in ("", ".", "..") or ":" in part for part in parts): raise DepotError(f"invalid {field}: {value!r}") return tuple(parts) def _positive_int(value: object, field: str) -> int: if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise DepotError(f"{field} must be a positive integer") return value def _non_negative_int(value: object, field: str) -> int: if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise DepotError(f"{field} must be a non-negative integer") return value def _string(value: object, field: str) -> str: if not isinstance(value, str) or not value.strip(): raise DepotError(f"{field} must be a non-empty string") return value def _sha1(path: Path) -> str: digest = hashlib.sha1(usedforsecurity=False) try: with path.open("rb") as handle: for chunk in iter(lambda: handle.read(HASH_READ_SIZE), b""): digest.update(chunk) except OSError as exc: raise DepotError(f"cannot hash {path}: {exc}") from exc return digest.hexdigest() def _content_length(value: str | None) -> int | None: if value is None: return None try: length = int(value) except ValueError as exc: raise DepotError(f"invalid Content-Length: {value!r}") from exc if length < 0: raise DepotError(f"invalid Content-Length: {value!r}") return length def _content_range(value: str | None) -> tuple[int, int, int]: if value is None: raise DepotError("206 response omitted Content-Range") match = CONTENT_RANGE_PATTERN.fullmatch(value.strip()) if match is None: raise DepotError(f"invalid Content-Range: {value!r}") start, end, total = (int(part) for part in match.groups()) if start > end or end >= total: raise DepotError(f"invalid Content-Range: {value!r}") return start, end, total def _normalized_cpu_load() -> float | None: """Return one-minute runnable load normalized by the available CPU count.""" try: load = os.getloadavg()[0] except (AttributeError, OSError): return None processors = os.cpu_count() if processors is None or processors <= 0: return None return max(0.0, load / processors) def _prune_transfer_samples( samples: deque[tuple[float, int]], now: float, ) -> None: cutoff = now - TRANSFER_RATE_WINDOW_SECONDS while len(samples) > 2 and samples[1][0] <= cutoff: samples.popleft() def _sample_rate(samples: deque[tuple[float, int]]) -> float | None: if len(samples) < 2: return None first_time, first_bytes = samples[0] last_time, last_bytes = samples[-1] elapsed = last_time - first_time if elapsed <= 0 or last_bytes < first_bytes: return None return (last_bytes - first_bytes) / elapsed def _server_display_name(server: str) -> str: parsed = urlparse(server if "://" in server else f"//{server}") return parsed.hostname or server.split("/", 1)[0] def _headers() -> dict[str, str]: return {"User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)"} def _write_json(path: Path, value: object, *, compact: bool = False) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") try: payload = ( json.dumps(value, separators=(",", ":"), sort_keys=True) if compact else json.dumps(value, indent=2, sort_keys=True) ) temporary.write_text(payload + "\n", encoding="utf-8") temporary.replace(path) except BaseException: temporary.unlink(missing_ok=True) raise def _emit_progress( callback: ProgressCallback | None, phase: str, files_completed: int, files_total: int, bytes_completed: int, bytes_total: int, current_file: str | None, ) -> None: if callback is not None: callback( SyncProgress( phase, files_completed, files_total, bytes_completed, bytes_total, current_file, bytes_remaining=max(0, bytes_total - bytes_completed), ) )