Coverage for src/ph/download_queue.py: 92.0%

1"""Persistent concurrent game-download queue.""" 

2 

3import json 

4import logging 

5import os 

6import random 

7import shutil 

8import time 

9import uuid 

10from collections.abc import Callable, Sequence 

11from concurrent.futures import Future, ThreadPoolExecutor 

12from dataclasses import asdict, dataclass, field, replace 

13from enum import StrEnum 

14from pathlib import Path 

15from threading import Event, RLock, Timer 

16from typing import Any, cast 

17 

18from ph.bittorrent import BitTorrentSettings, TorrentFileChoice 

19from ph.downloader import ( 

20 DownloadCancelled, 

21 DownloadError, 

22 DownloadRateLimited, 

23 DownloadSelectionRequired, 

24 download_files, 

25) 

26from ph.library import replace_game 

27from ph.models import DownloadResult, InstalledGame, MediaDownload, Platform 

28from ph.organizer import OrganizeError, install_bundled_bios, install_downloads 

29 

30LOGGER = logging.getLogger(__name__) 

31 

32QUEUE_FILENAME = ".pocket-harbor-downloads.json" 

33QUEUE_DIRECTORY = ".pocket-harbor-downloads" 

34QUEUE_FORMAT_VERSION = 1 

35DEFAULT_CONCURRENT_DOWNLOADS = 3 

36RATE_LIMIT_RETRY_BASE_SECONDS = 15.0 

37RATE_LIMIT_RETRY_MAX_SECONDS = 60.0 * 60.0 

38RATE_LIMIT_RETRY_JITTER = 0.2 

39 

40type DownloadRunner = Callable[..., list[DownloadResult]] 

41type DownloadInstaller = Callable[..., list[DownloadResult]] 

42type GameReplacer = Callable[[InstalledGame, DownloadResult], DownloadResult] 

43type BundledBiosInstaller = Callable[..., tuple[Path, ...]] 

44type RetryDelay = Callable[[int, float | None], float] 

45 

46 

47@dataclass(frozen=True, slots=True) 

48class RateLimitRetrySettings: 

49 """User-configurable exponential backoff policy for HTTP 429 responses.""" 

50 

51 base_seconds: float = RATE_LIMIT_RETRY_BASE_SECONDS 

52 max_seconds: float = RATE_LIMIT_RETRY_MAX_SECONDS 

53 jitter_ratio: float = RATE_LIMIT_RETRY_JITTER 

54 

55 def __post_init__(self) -> None: 

56 if self.base_seconds <= 0: 

57 raise ValueError("Initial retry delay must be positive.") 

58 if self.max_seconds < self.base_seconds: 

59 raise ValueError("Maximum retry delay must not be shorter than the initial delay.") 

60 if not 0 <= self.jitter_ratio <= 1: 

61 raise ValueError("Retry jitter ratio must be between 0 and 1.") 

62 

63 

64class DownloadState(StrEnum): 

65 """Stable states exposed to the TUI and persisted between launches.""" 

66 

67 QUEUED = "queued" 

68 DOWNLOADING = "downloading" 

69 RATE_LIMITED = "rate_limited" 

70 PAUSED = "paused" 

71 FAILED = "failed" 

72 CANCELLED = "cancelled" 

73 COMPLETED = "completed" 

74 

75 

76@dataclass(frozen=True, slots=True) 

77class DownloadJob: 

78 """Read-only snapshot of one queued game download.""" 

79 

80 job_id: str 

81 title: str 

82 store_id: str 

83 store_name: str 

84 state: DownloadState 

85 filename: str 

86 downloaded_bytes: int 

87 total_bytes: int | None 

88 error: str | None 

89 created_at: float 

90 completed_path: Path | None 

91 torrent_candidates: tuple[TorrentFileChoice, ...] 

92 platform: Platform 

93 roms_directory: Path 

94 region: str | None 

95 bundled_bios_count: int 

96 is_update: bool 

97 retry_attempt: int = 0 

98 retry_at: float | None = None 

99 bios_directory: str = "bios" 

100 

101 

102@dataclass(slots=True) 

103class _QueuedDownload: 

104 job_id: str 

105 title: str 

106 store_id: str 

107 store_name: str 

108 referrer: str 

109 media: tuple[MediaDownload, ...] 

110 platform: Platform 

111 roms_directory: Path 

112 bios_directory: str 

113 timeout_seconds: float 

114 bittorrent_settings: BitTorrentSettings | None 

115 region: str | None 

116 replacement_game: InstalledGame | None = None 

117 state: DownloadState = DownloadState.QUEUED 

118 filename: str = "" 

119 downloaded_bytes: int = 0 

120 total_bytes: int | None = None 

121 error: str | None = None 

122 created_at: float = field(default_factory=time.time) 

123 completed_path: Path | None = None 

124 bundled_bios_count: int = 0 

125 torrent_candidates: tuple[TorrentFileChoice, ...] = () 

126 stop_reason: str | None = None 

127 stop_event: Event = field(default_factory=Event, repr=False) 

128 last_persisted_at: float = field(default=0.0, repr=False) 

129 retry_attempt: int = 0 

130 retry_at: float | None = None 

131 

132 def snapshot(self) -> DownloadJob: 

133 return DownloadJob( 

134 job_id=self.job_id, 

135 title=self.title, 

136 store_id=self.store_id, 

137 store_name=self.store_name, 

138 state=self.state, 

139 filename=self.filename, 

140 downloaded_bytes=self.downloaded_bytes, 

141 total_bytes=self.total_bytes, 

142 error=self.error, 

143 created_at=self.created_at, 

144 completed_path=self.completed_path, 

145 torrent_candidates=self.torrent_candidates, 

146 platform=self.platform, 

147 roms_directory=self.roms_directory, 

148 region=self.region, 

149 bundled_bios_count=self.bundled_bios_count, 

150 is_update=self.replacement_game is not None, 

151 retry_attempt=self.retry_attempt, 

152 retry_at=self.retry_at, 

153 bios_directory=self.bios_directory, 

154 ) 

155 

156 

157class DownloadQueue: 

158 """Run several downloads concurrently and preserve unfinished jobs on disk.""" 

159 

160 def __init__( 

161 self, 

162 download_directory: Path, 

163 *, 

164 max_concurrent: int = DEFAULT_CONCURRENT_DOWNLOADS, 

165 runner: DownloadRunner = download_files, 

166 installer: DownloadInstaller = install_downloads, 

167 replacer: GameReplacer = replace_game, 

168 bios_installer: BundledBiosInstaller = install_bundled_bios, 

169 retry_delay: RetryDelay | None = None, 

170 retry_settings: RateLimitRetrySettings | None = None, 

171 ) -> None: 

172 if max_concurrent <= 0: 

173 raise ValueError("The concurrent download count must be positive.") 

174 self.download_directory = download_directory 

175 self.queue_path = download_directory / QUEUE_FILENAME 

176 self.staging_root = download_directory / QUEUE_DIRECTORY 

177 self._runner = runner 

178 self._installer = installer 

179 self._replacer = replacer 

180 self._bios_installer = bios_installer 

181 self._retry_delay = retry_delay 

182 self._retry_settings = retry_settings or RateLimitRetrySettings() 

183 self._lock = RLock() 

184 self._executor = ThreadPoolExecutor( 

185 max_workers=max_concurrent, 

186 thread_name_prefix="pocket-harbor-download", 

187 ) 

188 self._jobs: dict[str, _QueuedDownload] = {} 

189 self._futures: dict[str, Future[None]] = {} 

190 self._retry_timers: dict[str, Timer] = {} 

191 self._closing = False 

192 self._refresh_required = False 

193 self._load() 

194 with self._lock: 

195 for job in self._jobs.values(): 

196 if job.state in {DownloadState.QUEUED, DownloadState.DOWNLOADING}: 

197 job.state = DownloadState.QUEUED 

198 self._submit_locked(job) 

199 elif job.state is DownloadState.RATE_LIMITED: 

200 self._schedule_rate_limit_retry_locked(job) 

201 self._save_locked() 

202 

203 def enqueue( 

204 self, 

205 *, 

206 title: str, 

207 store_id: str, 

208 store_name: str, 

209 referrer: str, 

210 media: Sequence[MediaDownload], 

211 platform: Platform, 

212 roms_directory: Path, 

213 timeout_seconds: float, 

214 bios_directory: str = "bios", 

215 bittorrent_settings: BitTorrentSettings | None = None, 

216 region: str | None = None, 

217 replacement_game: InstalledGame | None = None, 

218 ) -> DownloadJob: 

219 """Persist and start one game download using a snapshot of its store settings.""" 

220 

221 if not media: 

222 raise DownloadError("The download list is empty.") 

223 job = _QueuedDownload( 

224 job_id=uuid.uuid4().hex[:12], 

225 title=title.strip() or media[0].expected_filename or "Download", 

226 store_id=store_id, 

227 store_name=store_name, 

228 referrer=referrer, 

229 media=tuple(media), 

230 platform=platform, 

231 roms_directory=roms_directory, 

232 bios_directory=bios_directory, 

233 timeout_seconds=timeout_seconds, 

234 bittorrent_settings=bittorrent_settings, 

235 region=region, 

236 replacement_game=replacement_game, 

237 ) 

238 with self._lock: 

239 if self._closing: 

240 raise DownloadError("The download queue is closing.") 

241 self._jobs[job.job_id] = job 

242 self._save_locked() 

243 self._submit_locked(job) 

244 LOGGER.info( 

245 "Queued game download id=%s title=%r store=%s platform=%s", 

246 job.job_id, 

247 job.title, 

248 job.store_id, 

249 job.platform.alias, 

250 ) 

251 return job.snapshot() 

252 

253 def jobs(self) -> tuple[DownloadJob, ...]: 

254 """Return stable snapshots ordered with the newest job first.""" 

255 

256 with self._lock: 

257 return tuple( 

258 job.snapshot() 

259 for job in sorted( 

260 self._jobs.values(), 

261 key=lambda item: item.created_at, 

262 reverse=True, 

263 ) 

264 ) 

265 

266 def find(self, job_id: str) -> DownloadJob | None: 

267 with self._lock: 

268 job = self._jobs.get(job_id) 

269 return job.snapshot() if job is not None else None 

270 

271 def dismiss_completed(self, job_id: str) -> bool: 

272 """Remove a completed job after the interface has acknowledged it.""" 

273 

274 with self._lock: 

275 job = self._jobs.get(job_id) 

276 if job is None or job.state is not DownloadState.COMPLETED: 

277 return False 

278 del self._jobs[job_id] 

279 self._futures.pop(job_id, None) 

280 self._save_locked() 

281 LOGGER.debug("Removed acknowledged completed download id=%s", job_id) 

282 return True 

283 

284 def pause(self, job_id: str) -> bool: 

285 """Pause a queued or active job while keeping its verified partial data.""" 

286 

287 with self._lock: 

288 job = self._jobs.get(job_id) 

289 if job is None or job.state not in { 

290 DownloadState.QUEUED, 

291 DownloadState.DOWNLOADING, 

292 DownloadState.RATE_LIMITED, 

293 }: 

294 return False 

295 if job.state is DownloadState.RATE_LIMITED: 

296 self._cancel_retry_timer_locked(job_id) 

297 job.state = DownloadState.PAUSED 

298 self._save_locked() 

299 return True 

300 job.stop_reason = "pause" 

301 job.stop_event.set() 

302 future = self._futures.get(job_id) 

303 if future is not None and future.cancel(): 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true

304 job.state = DownloadState.PAUSED 

305 job.stop_event.clear() 

306 self._save_locked() 

307 return True 

308 

309 def resume(self, job_id: str) -> bool: 

310 """Resume an explicitly paused job.""" 

311 

312 with self._lock: 

313 job = self._jobs.get(job_id) 

314 if job is None or job.state is not DownloadState.PAUSED or self._closing: 

315 return False 

316 self._reset_for_run(job, keep_progress=True) 

317 self._save_locked() 

318 self._submit_locked(job) 

319 return True 

320 

321 def retry(self, job_id: str) -> bool: 

322 """Retry a failed or cancelled job, reusing any safe partial data.""" 

323 

324 with self._lock: 

325 job = self._jobs.get(job_id) 

326 if ( 

327 job is None 

328 or job.state 

329 not in { 

330 DownloadState.FAILED, 

331 DownloadState.CANCELLED, 

332 } 

333 or self._closing 

334 ): 

335 return False 

336 self._reset_for_run(job, keep_progress=job.state is DownloadState.FAILED) 

337 self._save_locked() 

338 self._submit_locked(job) 

339 return True 

340 

341 def cancel(self, job_id: str) -> bool: 

342 """Cancel a job and remove its incomplete transfer data.""" 

343 

344 with self._lock: 

345 job = self._jobs.get(job_id) 

346 if job is None or job.state in { 

347 DownloadState.CANCELLED, 

348 DownloadState.COMPLETED, 

349 }: 

350 return False 

351 if job.state is DownloadState.RATE_LIMITED: 

352 self._cancel_retry_timer_locked(job_id) 

353 job.state = DownloadState.CANCELLED 

354 self._remove_staging(job.job_id) 

355 self._save_locked() 

356 return True 

357 job.stop_reason = "cancel" 

358 job.stop_event.set() 

359 future = self._futures.get(job_id) 

360 if (future is not None and future.cancel()) or job.state in { 360 ↛ 364line 360 didn't jump to line 364 because the condition on line 360 was never true

361 DownloadState.PAUSED, 

362 DownloadState.FAILED, 

363 }: 

364 job.state = DownloadState.CANCELLED 

365 self._remove_staging(job.job_id) 

366 self._save_locked() 

367 return True 

368 

369 def choose_torrent_file(self, job_id: str, choice: TorrentFileChoice) -> bool: 

370 """Apply an explicit Minerva file choice before retrying a failed job.""" 

371 

372 with self._lock: 

373 job = self._jobs.get(job_id) 

374 if job is None or job.state is not DownloadState.FAILED: 

375 return False 

376 changed = False 

377 updated: list[MediaDownload] = [] 

378 for media in job.media: 

379 if not changed and media.torrent_file_index is not None: 

380 updated.append( 

381 replace( 

382 media, 

383 torrent_file_index=choice.index, 

384 expected_filename=choice.filename, 

385 torrent_file_path=choice.path, 

386 ) 

387 ) 

388 changed = True 

389 else: 

390 updated.append(media) 

391 if not changed: 

392 return False 

393 job.media = tuple(updated) 

394 job.torrent_candidates = () 

395 job.error = None 

396 self._save_locked() 

397 return True 

398 

399 @property 

400 def refresh_required(self) -> bool: 

401 with self._lock: 

402 return self._refresh_required 

403 

404 def update_retry_settings(self, settings: RateLimitRetrySettings) -> None: 

405 """Apply a changed backoff policy to retry delays scheduled from now on.""" 

406 

407 with self._lock: 

408 self._retry_settings = settings 

409 

410 def mark_refreshed(self) -> None: 

411 """Persist that the frontend has observed all completed installations.""" 

412 

413 with self._lock: 

414 self._refresh_required = False 

415 self._save_locked(include_terminal=not self._closing) 

416 

417 def shutdown(self) -> None: 

418 """Stop workers safely; interrupted jobs resume automatically next launch.""" 

419 

420 with self._lock: 

421 if self._closing: 

422 return 

423 self._closing = True 

424 for job in self._jobs.values(): 

425 if job.state not in {DownloadState.QUEUED, DownloadState.DOWNLOADING}: 

426 continue 

427 job.stop_reason = "shutdown" 

428 job.stop_event.set() 

429 future = self._futures.get(job.job_id) 

430 if future is not None and future.cancel(): 430 ↛ 431line 430 didn't jump to line 431 because the condition on line 430 was never true

431 job.state = DownloadState.QUEUED 

432 for timer in self._retry_timers.values(): 

433 timer.cancel() 

434 self._retry_timers.clear() 

435 self._save_locked(include_terminal=False) 

436 self._executor.shutdown(wait=True, cancel_futures=True) 

437 with self._lock: 

438 self._save_locked(include_terminal=False) 

439 LOGGER.info("Download queue stopped with %d persistent job(s)", len(self._jobs)) 

440 

441 def _submit_locked(self, job: _QueuedDownload) -> None: 

442 job.stop_reason = None 

443 job.stop_event.clear() 

444 self._futures[job.job_id] = self._executor.submit(self._run, job.job_id) 

445 

446 def _run(self, job_id: str) -> None: 

447 with self._lock: 

448 job = self._jobs.get(job_id) 

449 if job is None or job.state is not DownloadState.QUEUED or self._closing: 449 ↛ 450line 449 didn't jump to line 450 because the condition on line 449 was never true

450 return 

451 job.state = DownloadState.DOWNLOADING 

452 self._save_locked() 

453 job_directory = self.staging_root / job.job_id 

454 

455 installed_bios: list[Path] = [] 

456 try: 

457 downloads = self._runner( 

458 job.media, 

459 job_directory, 

460 job.referrer, 

461 job.timeout_seconds, 

462 lambda label, current, total: self._progress(job_id, label, current, total), 

463 job.stop_event.is_set, 

464 bittorrent_settings=job.bittorrent_settings, 

465 resume=True, 

466 ) 

467 if job.stop_event.is_set(): 467 ↛ 468line 467 didn't jump to line 468 because the condition on line 467 was never true

468 raise DownloadCancelled("Download interrupted.") 

469 if job.replacement_game is not None: 

470 if len(downloads) != 1: 470 ↛ 471line 470 didn't jump to line 471 because the condition on line 470 was never true

471 raise OrganizeError("A game update must contain exactly one download.") 

472 installed_bios.extend( 

473 self._bios_installer( 

474 downloads[0].path, 

475 job.platform, 

476 job.roms_directory, 

477 job.bios_directory, 

478 ) 

479 ) 

480 completed = [self._replacer(job.replacement_game, downloads[0])] 

481 else: 

482 completed = self._installer( 

483 downloads, 

484 job.platform, 

485 job.roms_directory, 

486 installed_bios.append, 

487 job.bios_directory, 

488 ) 

489 if not completed: 

490 raise OrganizeError("The completed download was not installed.") 

491 except DownloadSelectionRequired as error: 

492 self._finish_failed(job_id, str(error), error.candidates) 

493 return 

494 except DownloadRateLimited as error: 

495 self._finish_rate_limited(job_id, error) 

496 return 

497 except DownloadCancelled: 

498 self._finish_interrupted(job_id) 

499 return 

500 except (DownloadError, OrganizeError, OSError) as error: 

501 self._finish_failed(job_id, str(error)) 

502 return 

503 except Exception as error: # pragma: no cover - final worker boundary 

504 LOGGER.exception("Unexpected background download failure id=%s", job_id) 

505 self._finish_failed(job_id, str(error)) 

506 return 

507 

508 with self._lock: 

509 job = self._jobs.get(job_id) 

510 if job is None: 510 ↛ 511line 510 didn't jump to line 511 because the condition on line 510 was never true

511 return 

512 job.state = DownloadState.COMPLETED 

513 job.completed_path = completed[0].path 

514 job.bundled_bios_count = len(installed_bios) 

515 job.downloaded_bytes = job.total_bytes or job.downloaded_bytes 

516 job.error = None 

517 self._refresh_required = True 

518 self._save_locked() 

519 self._remove_staging(job_id) 

520 LOGGER.info("Background download installed id=%s path=%s", job_id, completed[0].path) 

521 

522 def _progress(self, job_id: str, label: str, current: int, total: int | None) -> None: 

523 with self._lock: 

524 job = self._jobs.get(job_id) 

525 if job is None: 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true

526 return 

527 job.filename = label 

528 job.downloaded_bytes = current 

529 job.total_bytes = total 

530 now = time.monotonic() 

531 if now - job.last_persisted_at >= 5: 

532 job.last_persisted_at = now 

533 self._save_locked() 

534 

535 def _finish_interrupted(self, job_id: str) -> None: 

536 with self._lock: 

537 job = self._jobs.get(job_id) 

538 if job is None: 538 ↛ 539line 538 didn't jump to line 539 because the condition on line 538 was never true

539 return 

540 reason = job.stop_reason 

541 if reason == "cancel": 

542 job.state = DownloadState.CANCELLED 

543 self._remove_staging(job_id) 

544 elif reason == "pause": 

545 job.state = DownloadState.PAUSED 

546 else: 

547 job.state = DownloadState.QUEUED 

548 job.stop_event.clear() 

549 self._save_locked(include_terminal=not self._closing) 

550 LOGGER.info("Background download interrupted id=%s reason=%s", job_id, reason) 

551 

552 def _finish_failed( 

553 self, 

554 job_id: str, 

555 message: str, 

556 candidates: tuple[TorrentFileChoice, ...] = (), 

557 ) -> None: 

558 with self._lock: 

559 job = self._jobs.get(job_id) 

560 if job is None: 560 ↛ 561line 560 didn't jump to line 561 because the condition on line 560 was never true

561 return 

562 job.state = DownloadState.FAILED 

563 job.error = message 

564 job.torrent_candidates = candidates 

565 job.stop_event.clear() 

566 self._save_locked() 

567 LOGGER.error("Background download failed id=%s: %s", job_id, message) 

568 

569 def _finish_rate_limited(self, job_id: str, error: DownloadRateLimited) -> None: 

570 with self._lock: 

571 job = self._jobs.get(job_id) 

572 if job is None: 572 ↛ 573line 572 didn't jump to line 573 because the condition on line 572 was never true

573 return 

574 job.retry_attempt += 1 

575 delay = max( 

576 0.0, 

577 self._retry_delay(job.retry_attempt, error.retry_after_seconds) 

578 if self._retry_delay is not None 

579 else _rate_limit_retry_delay( 

580 job.retry_attempt, 

581 error.retry_after_seconds, 

582 self._retry_settings, 

583 ), 

584 ) 

585 job.state = DownloadState.RATE_LIMITED 

586 job.error = str(error) 

587 job.retry_at = time.time() + delay 

588 job.stop_event.clear() 

589 self._save_locked() 

590 if not self._closing: 590 ↛ 592line 590 didn't jump to line 592

591 self._schedule_rate_limit_retry_locked(job) 

592 LOGGER.warning( 

593 "Background download rate limited id=%s store=%s attempt=%d retry_in=%.1fs", 

594 job_id, 

595 job.store_id, 

596 job.retry_attempt, 

597 delay, 

598 ) 

599 

600 def _schedule_rate_limit_retry_locked(self, job: _QueuedDownload) -> None: 

601 self._cancel_retry_timer_locked(job.job_id) 

602 delay = max(0.0, (job.retry_at or time.time()) - time.time()) 

603 timer = Timer(delay, self._retry_rate_limited, args=(job.job_id,)) 

604 timer.daemon = True 

605 self._retry_timers[job.job_id] = timer 

606 timer.start() 

607 

608 def _retry_rate_limited(self, job_id: str) -> None: 

609 with self._lock: 

610 self._retry_timers.pop(job_id, None) 

611 job = self._jobs.get(job_id) 

612 if job is None or job.state is not DownloadState.RATE_LIMITED or self._closing: 612 ↛ 613line 612 didn't jump to line 613 because the condition on line 612 was never true

613 return 

614 job.state = DownloadState.QUEUED 

615 job.retry_at = None 

616 self._save_locked() 

617 self._submit_locked(job) 

618 

619 def _cancel_retry_timer_locked(self, job_id: str) -> None: 

620 timer = self._retry_timers.pop(job_id, None) 

621 if timer is not None: 

622 timer.cancel() 

623 

624 def _reset_for_run(self, job: _QueuedDownload, *, keep_progress: bool) -> None: 

625 job.state = DownloadState.QUEUED 

626 job.error = None 

627 job.stop_reason = None 

628 job.stop_event.clear() 

629 job.retry_attempt = 0 

630 job.retry_at = None 

631 if not keep_progress: 

632 self._remove_staging(job.job_id) 

633 job.filename = "" 

634 job.downloaded_bytes = 0 

635 job.total_bytes = None 

636 

637 def _remove_staging(self, job_id: str) -> None: 

638 if not job_id or "/" in job_id or "\\" in job_id: 638 ↛ 639line 638 didn't jump to line 639 because the condition on line 638 was never true

639 return 

640 shutil.rmtree(self.staging_root / job_id, ignore_errors=True) 

641 

642 def _load(self) -> None: 

643 try: 

644 payload = json.loads(self.queue_path.read_text(encoding="utf-8")) 

645 except FileNotFoundError: 

646 return 

647 except (OSError, UnicodeError, json.JSONDecodeError) as error: 

648 LOGGER.warning("Could not read persistent download queue: %s", error) 

649 return 

650 if not isinstance(payload, dict) or payload.get("version") != QUEUE_FORMAT_VERSION: 

651 LOGGER.warning("Ignoring an unsupported persistent download queue") 

652 return 

653 self._refresh_required = payload.get("refresh_required") is True 

654 raw_jobs = payload.get("jobs") 

655 if not isinstance(raw_jobs, list): 

656 return 

657 for raw_job in raw_jobs: 

658 try: 

659 job = _job_from_json(raw_job) 

660 except KeyError, TypeError, ValueError: 

661 LOGGER.warning("Ignoring a malformed persistent download job") 

662 continue 

663 if ( 

664 job.state is DownloadState.FAILED 

665 and job.error is not None 

666 and "HTTP 429" in job.error 

667 ): 

668 job.state = DownloadState.RATE_LIMITED 

669 job.retry_attempt = max(1, job.retry_attempt) 

670 job.retry_at = time.time() 

671 LOGGER.info("Migrated rate-limited download for automatic retry id=%s", job.job_id) 

672 if job.state not in {DownloadState.COMPLETED, DownloadState.CANCELLED}: 672 ↛ 657line 672 didn't jump to line 657 because the condition on line 672 was always true

673 self._jobs[job.job_id] = job 

674 

675 def _save_locked(self, *, include_terminal: bool = True) -> None: 

676 jobs = [ 

677 _job_to_json(job) 

678 for job in self._jobs.values() 

679 if include_terminal 

680 or job.state not in {DownloadState.COMPLETED, DownloadState.CANCELLED} 

681 ] 

682 payload = { 

683 "version": QUEUE_FORMAT_VERSION, 

684 "refresh_required": self._refresh_required, 

685 "jobs": jobs, 

686 } 

687 temporary = self.queue_path.with_name(self.queue_path.name + ".tmp") 

688 try: 

689 self.download_directory.mkdir(parents=True, exist_ok=True) 

690 temporary.write_text( 

691 json.dumps(payload, ensure_ascii=False, indent=2) + "\n", 

692 encoding="utf-8", 

693 ) 

694 os.replace(temporary, self.queue_path) 

695 except OSError as error: 

696 temporary.unlink(missing_ok=True) 

697 LOGGER.error("Could not persist download queue: %s", error) 

698 

699 

700def _job_to_json(job: _QueuedDownload) -> dict[str, object]: 

701 return { 

702 "id": job.job_id, 

703 "title": job.title, 

704 "store_id": job.store_id, 

705 "store_name": job.store_name, 

706 "referrer": job.referrer, 

707 "media": [ 

708 { 

709 "url": media.url, 

710 "torrent_file_index": media.torrent_file_index, 

711 "expected_filename": media.expected_filename, 

712 "torrent_file_path": list(media.torrent_file_path) 

713 if media.torrent_file_path is not None 

714 else None, 

715 } 

716 for media in job.media 

717 ], 

718 "platform": { 

719 "name": job.platform.name, 

720 "slug": job.platform.slug, 

721 "code": job.platform.code, 

722 "alias": job.platform.alias, 

723 "rom_folder": job.platform.rom_folder, 

724 "alternate_folders": list(job.platform.alternate_folders), 

725 }, 

726 "roms_directory": str(job.roms_directory), 

727 "bios_directory": job.bios_directory, 

728 "timeout_seconds": job.timeout_seconds, 

729 "bittorrent_settings": asdict(job.bittorrent_settings) 

730 if job.bittorrent_settings is not None 

731 else None, 

732 "region": job.region, 

733 "replacement_game": _installed_game_to_json(job.replacement_game), 

734 "state": job.state.value, 

735 "filename": job.filename, 

736 "downloaded_bytes": job.downloaded_bytes, 

737 "total_bytes": job.total_bytes, 

738 "error": job.error, 

739 "created_at": job.created_at, 

740 "completed_path": str(job.completed_path) if job.completed_path is not None else None, 

741 "bundled_bios_count": job.bundled_bios_count, 

742 "torrent_candidates": [ 

743 { 

744 "index": choice.index, 

745 "path": list(choice.path), 

746 "length": choice.length, 

747 "match_score": choice.match_score, 

748 } 

749 for choice in job.torrent_candidates 

750 ], 

751 "retry_attempt": job.retry_attempt, 

752 "retry_at": job.retry_at, 

753 } 

754 

755 

756def _job_from_json(value: object) -> _QueuedDownload: