From 7344dff8a7ee473059e77632d8b6fea62c264bc4 Mon Sep 17 00:00:00 2001 From: mkb79 Date: Wed, 5 Aug 2026 20:58:38 +0200 Subject: [PATCH] fix: stop the download queue from deadlocking after failed jobs `consume()` re-raised on error when `--ignore-errors` was not given. That does not just abort the job, it leaves the `while True` loop and kills the consumer task. `task_done()` still ran for the current item, but the consumer never took another one. Once every `--jobs` consumer had died this way, `await QUEUE.join()` waited forever for items nobody would pick up. With the default `-j 3` three early failures were enough; with `-j 1` a single one. The failure was also invisible: the exception sat in a dead task, and the `asyncio.gather(..., return_exceptions=True)` that would have collected it only runs after `QUEUE.join()` returns, which in this state never happens. Keep the consumer alive and move the abort decision into a `DownloadRun` object shared by all consumers. On the first failure it sets an event; consumers then stop starting jobs and only drain what is left, so `QUEUE.join()` can return. Downloads already in flight are allowed to finish rather than being cancelled mid-write. Waiting on `QUEUE.join()` alone was the deeper flaw: any consumer that ends early strands the queue. `drain_queue()` now waits on the join and the consumers together, so a worker that dies for any reason is noticed and reported instead of hanging the run. `--jobs` also rejects values below `1`, which previously queued work that no consumer would ever pick up. This also makes `--ignore-errors` mean what its help text says. Before, the flag's absence did not abort anything: one consumer died while the others carried on downloading until they died too. Finally, a run in which a job raised now ends in `AudibleCliException`, which `cli.main()` maps to exit code 2. That holds with `--ignore-errors` as well, so those failures are visible to scripts instead of being reported as success. Failures that are only logged, such as an unknown ASIN or a download rejected by its HTTP status, are not covered yet and still exit zero. Closes #235 Closes #239 --- CHANGELOG.md | 4 ++ src/audible_cli/cmds/cmd_download.py | 96 +++++++++++++++++++++++----- 2 files changed, 83 insertions(+), 17 deletions(-) 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()