"""Native, verified updater for the public PristonTale EU depot.""" from __future__ import annotations from collections import deque from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait from dataclasses import asdict, dataclass from datetime import datetime import json import os from pathlib import Path import re import threading import time from typing import Callable, Iterable, Mapping from urllib.parse import urlparse from urllib.request import urlopen from . import __version__ from .game.depot.catalogs import ( DEPOT_FILE_CATALOG_SCHEMA, DEPOT_LAUNCH_CATALOG_SCHEMA, depot_file_catalog_payload as _depot_file_catalog_payload, launch_catalog_from_snapshot, launch_catalog_payload as _launch_catalog_payload, parse_depot_file_catalog as _parse_depot_file_catalog, parse_depot_file_catalog_entry as _parse_depot_file_catalog_entry, parse_launch_catalog as _parse_launch_catalog, select_depot_start_option, ) from .game.depot.manifest import ( MAX_CREATE_IF_MISSING_BYTES, SHA1_PATTERN, decode_create_if_missing as _decode_create_if_missing, non_negative_int as _non_negative_int, normalize_relative_path as _normalise_relative_path, parse_create_if_missing as _parse_create_if_missing, parse_deleted_paths as _parse_deleted_paths, parse_snapshot as _parse_snapshot, parse_start_options as _parse_start_options, positive_int as _positive_int, relative_parts as _relative_parts, required_string as _string, start_option_text as _start_option_text, ) from .game.depot.integrity import ( ScannedGameEntry as _ScannedGameEntry, build_depot_file_catalog as _build_depot_file_catalog, catalog_entry as _catalog_entry, catalog_entry_from_cached as _catalog_entry_from_cached, catalog_identity_matches as _catalog_identity_matches, inspect_managed_game as _inspect_managed_game, plan_update as _plan_update, prepare_game_root as _prepare_game_root, safe_target as _safe_target, scan_directory as _scan_directory, scan_game_tree as _scan_game_tree, sha1_file as _sha1_file, ) from .game.depot.filesystem import ( DEFAULT_MAX_LINE_ENDING_RECOVERY_BYTES, DEFAULT_UNMANAGED_EXECUTABLE_SUFFIXES, apply_create_if_missing as _filesystem_apply_create_if_missing, apply_deletions as _filesystem_apply_deletions, ensure_target_parent as _filesystem_ensure_target_parent, quarantine_unmanaged_files as _filesystem_quarantine_unmanaged_files, recover_crlf_manifest_file as _filesystem_recover_crlf_manifest_file, safe_imported_user_file as _filesystem_safe_imported_user_file, write_json as _filesystem_write_json, ) from .game.depot.transfer import ( AdaptiveConcurrencyController as _AdaptiveConcurrencyController, BandwidthLimiter, DepotHttpClient as _DepotHttpClient, DepotHttpPolicy as _DepotHttpPolicy, DownloadLimits as _DownloadLimits, DownloadObserver, DownloadTelemetry, HttpOpener, HttpResponse, build_manifest_url as _transfer_build_manifest_url, content_length as _transfer_content_length, content_range as _transfer_content_range, depot_manifest_url as _transfer_depot_manifest_url, download_partial_path as _transfer_download_partial_path, file_url as _transfer_file_url, normalize_server as _transfer_normalise_server, normalized_cpu_load as _transfer_normalized_cpu_load, ordered_download_servers as _transfer_ordered_download_servers, prune_transfer_samples as _transfer_prune_samples, request_headers as _transfer_request_headers, sample_rate as _transfer_sample_rate, server_candidates as _transfer_server_candidates, ) from .game.depot import ( DEFAULT_DEPOT_START_OPTION, CreateIfMissing, DepotDownloadMetrics, DepotError, DepotFile, DepotFileCatalog, DepotFileCatalogEntry, DepotIntegrityIssue, DepotLaunchCatalog, DepotPreflightResult, DepotSnapshot, DepotStartOption, DepotSyncResult, PlannedDownload, ProgressCallback, QuarantineResult, SyncProgress, UpdatePlan, ) from .paths import AppPaths from .telemetry import classify_network_failure, current_telemetry # Keep historical imports available while implementation moves into the feature # package. Referencing them here makes that compatibility surface explicit to # static analysis without narrowing this module's existing wildcard behavior. _COMPATIBILITY_EXPORTS = ( DEFAULT_DEPOT_START_OPTION, DEPOT_FILE_CATALOG_SCHEMA, DEPOT_LAUNCH_CATALOG_SCHEMA, MAX_CREATE_IF_MISSING_BYTES, SHA1_PATTERN, BandwidthLimiter, DownloadTelemetry, HttpOpener, HttpResponse, DepotStartOption, DepotFileCatalogEntry, DepotIntegrityIssue, _depot_file_catalog_payload, _catalog_entry, _catalog_entry_from_cached, _catalog_identity_matches, _non_negative_int, _normalise_relative_path, _parse_create_if_missing, _parse_deleted_paths, _parse_depot_file_catalog_entry, _parse_start_options, _relative_parts, _start_option_text, _string, _ScannedGameEntry, _scan_directory, _scan_game_tree, launch_catalog_from_snapshot, select_depot_start_option, ) 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_LINE_ENDING_RECOVERY_BYTES = DEFAULT_MAX_LINE_ENDING_RECOVERY_BYTES UNMANAGED_EXECUTABLE_SUFFIXES = DEFAULT_UNMANAGED_EXECUTABLE_SUFFIXES ALWAYS_VERIFY_PATHS = frozenset( {"game.exe", "game.dll", "gamese.exe", "gamese.dll"} ) CONTENT_RANGE_PATTERN = re.compile(r"^bytes\s+(\d+)-(\d+)/(\d+)$", re.IGNORECASE) MAX_PREFLIGHT_ISSUES = 32 @dataclass(frozen=True) class DownloadOptions(_DownloadLimits): """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) class AdaptiveConcurrencyController(_AdaptiveConcurrencyController): """Compatibility facade configured from the historical module constants.""" def __init__( self, options: DownloadOptions, *, clock: Callable[[], float] = time.monotonic, cpu_load: Callable[[], float | None] | None = None, ) -> None: super().__init__( options, clock=clock, cpu_load=cpu_load or _normalized_cpu_load, initial_workers=DOWNLOAD_INITIAL_WORKERS, evaluation_interval_seconds=ADAPTIVE_EVALUATION_INTERVAL_SECONDS, retry_window_seconds=ADAPTIVE_RETRY_WINDOW_SECONDS, retry_recovery_seconds=ADAPTIVE_RETRY_RECOVERY_SECONDS, transfer_rate_window_seconds=TRANSFER_RATE_WINDOW_SECONDS, ) def cache_depot_launch_catalog(path: Path, snapshot: DepotSnapshot) -> None: """Cache only publisher-advertised launch metadata for network-free Play.""" _write_json(path, _launch_catalog_payload(snapshot)) 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")) return _parse_launch_catalog(value) except (OSError, UnicodeError, json.JSONDecodeError, DepotError): return None 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.""" catalog = _build_depot_file_catalog( game_directory, snapshot, verified_paths=verified_paths, previous=previous, ) _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")) return _parse_depot_file_catalog(value) except (OSError, UnicodeError, json.JSONDecodeError, DepotError): return None 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 result, updated_catalog = _inspect_managed_game( paths.source_game, catalog, hash_file=_sha1, max_issues=MAX_PREFLIGHT_ISSUES, clock=time.perf_counter, ) if updated_catalog is not None: _write_json( paths.depot_file_catalog, _depot_file_catalog_payload(updated_catalog), compact=True, ) return result def _depot_http_policy() -> _DepotHttpPolicy: return _DepotHttpPolicy( configuration_url=lambda: CDN_CONFIGURATION_URL, server_candidates=lambda configuration, preferred_region: _server_candidates( configuration, preferred_region, ), depot_manifest_url=lambda server: _depot_manifest_url(server), build_manifest_url=lambda server, build: _build_manifest_url(server, build), file_url=lambda server, file: _file_url(server, file), ordered_download_servers=lambda primary, candidates: _ordered_download_servers( primary, candidates, ), download_partial_path=lambda destination: _download_partial_path(destination), recover_manifest_file=lambda temporary, destination, file: ( _recover_crlf_manifest_file(temporary, destination, file) ), parse_snapshot=lambda region, server, depot_manifest, build_manifest: ( _parse_snapshot(region, server, depot_manifest, build_manifest) ), positive_int=lambda value, field: _positive_int(value, field), content_length=lambda value: _content_length(value), content_range=lambda value: _content_range(value), headers=lambda: _headers(), read_size=lambda: READ_SIZE, hash_read_size=lambda: HASH_READ_SIZE, clock=lambda: time.monotonic(), ) class DepotClient(_DepotHttpClient): """Stable facade for the publisher depot HTTP client.""" 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: super().__init__( policy=_depot_http_policy(), timeout_seconds=timeout_seconds, opener=opener, retry_delays=retry_delays, sleeper=sleeper, download_observer=download_observer, ) def plan_update( game_directory: Path, snapshot: DepotSnapshot, *, repair: bool = False, ) -> UpdatePlan: """Inspect a local game tree without writing to it.""" return _plan_update( game_directory, snapshot, repair=repair, hash_file=_sha1, always_verify_paths=ALWAYS_VERIFY_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: return _filesystem_quarantine_unmanaged_files( paths.source_game, paths.game_import_quarantine, paths.data, snapshot, unmanaged_executable_suffixes=UNMANAGED_EXECUTABLE_SUFFIXES, prepare_root=lambda game_directory, *, create: _prepare_game_root( game_directory, create=create, ), safe_user_file=lambda path, relative: _safe_imported_user_file( path, relative, ), write_manifest=lambda path, value: _write_json(path, value), clock=lambda: datetime.now().astimezone(), ) def _safe_imported_user_file(path: Path, relative: str) -> bool: return _filesystem_safe_imported_user_file( path, relative, executable_suffixes=UNMANAGED_EXECUTABLE_SUFFIXES, ) 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: Callable[[DownloadTelemetry], None] | None = ( 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: return _filesystem_apply_deletions( root, relative_paths, target_for=lambda game_root, relative: _safe_target(game_root, relative), ) def _apply_create_if_missing( root: Path, entries: Iterable[CreateIfMissing], ) -> int: return _filesystem_apply_create_if_missing( root, entries, target_for=lambda game_root, relative: _safe_target(game_root, relative), decode_entry=lambda entry: _decode_create_if_missing(entry), ensure_parent=lambda game_root, target: _ensure_target_parent( game_root, target, ), ) 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 _server_candidates( configuration: Mapping[str, object] | None, preferred_region: str, ) -> tuple[tuple[str, str], ...]: return _transfer_server_candidates( configuration, preferred_region, fallback_servers=FALLBACK_SERVERS, ) def _normalise_server(value: str) -> str: return _transfer_normalise_server(value) def _depot_manifest_url(server: str) -> str: return _transfer_depot_manifest_url(server, depot_id=DEPOT_ID) def _build_manifest_url(server: str, build: int) -> str: return _transfer_build_manifest_url(server, build, depot_id=DEPOT_ID) def _file_url(server: str, file: DepotFile) -> str: return _transfer_file_url(server, file, depot_id=DEPOT_ID) def _ordered_download_servers( primary: str, candidates: Iterable[str], ) -> tuple[str, ...]: return _transfer_ordered_download_servers(primary, candidates) def _download_partial_path(destination: Path) -> Path: return _transfer_download_partial_path(destination) def _recover_crlf_manifest_file( temporary: Path, destination: Path, file: DepotFile, ) -> bool: return _filesystem_recover_crlf_manifest_file( temporary, destination, file, maximum_bytes=MAX_LINE_ENDING_RECOVERY_BYTES, process_id=lambda: os.getpid(), ) def _ensure_target_parent(root: Path, target: Path) -> None: _filesystem_ensure_target_parent(root, target) def _sha1(path: Path) -> str: return _sha1_file(path, read_size=HASH_READ_SIZE) def _content_length(value: str | None) -> int | None: return _transfer_content_length(value) def _content_range(value: str | None) -> tuple[int, int, int]: return _transfer_content_range(value, pattern=CONTENT_RANGE_PATTERN) def _normalized_cpu_load() -> float | None: return _transfer_normalized_cpu_load() def _prune_transfer_samples( samples: deque[tuple[float, int]], now: float, ) -> None: _transfer_prune_samples( samples, now, window_seconds=TRANSFER_RATE_WINDOW_SECONDS, ) def _sample_rate(samples: deque[tuple[float, int]]) -> float | None: return _transfer_sample_rate(samples) 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 _transfer_request_headers(__version__) def _write_json(path: Path, value: object, *, compact: bool = False) -> None: _filesystem_write_json(path, value, compact=compact) 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), ) )