diff --git a/CHANGELOG.md b/CHANGELOG.md index d8ab64e0..fef47fc5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). - API timestamps, library date filters and voucher deadlines are now timezone-aware UTC; naive `--start-date`/`--end-date` input is interpreted as UTC, as the option help already stated (#266) - Replaced `datetime.utcnow()` and `datetime.utcfromtimestamp()`, deprecated since Python 3.12 (#266) +- Without `--ignore-errors`, the download command now really aborts on the first failure: running downloads finish, queued ones are skipped (#235) +- The download command exits non-zero when a download job raised an error, also with `--ignore-errors`. Failures that are only logged, such as an unknown ASIN or a download rejected by its HTTP status, still exit zero for now (#256) +- `--jobs` now rejects values below 1 instead of accepting `0`, which queued work that no consumer would ever pick up (#235) ### Fixed @@ -23,6 +26,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). - Treat an unknown `publication_datetime` as published rather than crashing, and let `ItemNotPublished` report the ASIN without a countdown when no usable date is available (#268) - Reach `LicenseDenied` and `NoDownloadUrl` as intended when `license_denial_reasons`, `content_metadata` or `content_url` are null, instead of raising a `TypeError` or `AttributeError` (#268) - `audible manage config edit` no longer crashes with `TypeError: 'PosixPath' object is not iterable`; `click.edit()` accepts a `str` or an iterable of them, but not a `pathlib.Path` (#248) +- The download command no longer hangs forever after failed downloads. A failing job killed its consumer, and once every `--jobs` consumer had died the queue was never drained (#235, #239) ## [0.4.0] - 2026-07-20 diff --git a/src/audible_cli/cmds/cmd_download.py b/src/audible_cli/cmds/cmd_download.py index bc86943a..f1584a17 100644 --- a/src/audible_cli/cmds/cmd_download.py +++ b/src/audible_cli/cmds/cmd_download.py @@ -502,19 +502,91 @@ async def download_aaxc( ) -async def consume(ignore_errors): +class DownloadRun: + """Tracks failures across all consumers of a single download run.""" + + def __init__(self, ignore_errors: bool): + self.ignore_errors = ignore_errors + # Set as soon as a job fails while --ignore-errors is not in effect + self.abort = asyncio.Event() + self.errors: list[Exception] = [] + # Jobs dropped without running because the run was aborted + self.skipped = 0 + + def record(self, error: Exception) -> None: + self.errors.append(error) + if not self.ignore_errors: + # Let running downloads finish, but start no new ones + self.abort.set() + + def raise_for_errors(self) -> None: + if not self.errors: + return + + msg = f"{len(self.errors)} job(s) failed" + if self.skipped: + msg += f", {self.skipped} skipped after the abort" + if not self.ignore_errors: + msg += ". Use --ignore-errors to download the rest anyway" + + raise AudibleCliException(msg) + + +async def consume(run: DownloadRun): while True: cmd, kwargs = await QUEUE.get() try: + # Never leave this loop on error. A consumer that dies stops + # taking items, and once every consumer is gone QUEUE.join() + # waits forever for jobs nobody will pick up. + if run.abort.is_set(): + run.skipped += 1 + continue + await cmd(**kwargs) except Exception as e: logger.error(e) - if not ignore_errors: - raise + run.record(e) finally: QUEUE.task_done() +async def drain_queue(run: DownloadRun, sim_jobs: int): + """Work off QUEUE with `sim_jobs` consumers until it is empty.""" + consumers = [ + asyncio.create_task(consume(run)) for _ in range(sim_jobs) + ] + joiner = asyncio.create_task(QUEUE.join()) + try: + # A consumer only ever finishes by dying. Waiting on them next to the + # join means such a death surfaces here, instead of leaving + # QUEUE.join() waiting for items that nobody will pick up anymore. + await asyncio.wait( + [joiner, *consumers], return_when=asyncio.FIRST_COMPLETED + ) + if not joiner.done(): + # A consumer ended while jobs were still queued. Surface why, so + # the run does not look like it merely finished early. + for consumer in consumers: + if consumer.done() and not consumer.cancelled(): + exc = consumer.exception() + if exc is not None: + raise exc + + raise AudibleCliException( + "A download worker stopped unexpectedly, the remaining jobs " + "were not processed" + ) + finally: + # the consumer is still awaiting an item, cancel it + joiner.cancel() + for consumer in consumers: + consumer.cancel() + + await asyncio.gather(joiner, *consumers, return_exceptions=True) + display_counter() + + def queue_job( get_cover, get_pdf, @@ -729,7 +801,7 @@ def display_counter(): ) @click.option( "--jobs", "-j", - type=int, + type=click.IntRange(min=1), default=3, show_default=True, help="number of simultaneous downloads" @@ -973,16 +1045,6 @@ async def cli(session, api_client, **params): ) # schedule the consumer - consumers = [ - asyncio.create_task(consume(ignore_errors)) for _ in range(sim_jobs) - ] - try: - # wait until the consumer has processed all items - await QUEUE.join() - finally: - # the consumer is still awaiting an item, cancel it - for consumer in consumers: - consumer.cancel() - - await asyncio.gather(*consumers, return_exceptions=True) - display_counter() + run = DownloadRun(ignore_errors) + await drain_queue(run, sim_jobs) + run.raise_for_errors()