diff --git a/src/palace/manager/celery/tasks/custom_lists.py b/src/palace/manager/celery/tasks/custom_lists.py index dd1788d67b..100cf22e20 100644 --- a/src/palace/manager/celery/tasks/custom_lists.py +++ b/src/palace/manager/celery/tasks/custom_lists.py @@ -11,7 +11,8 @@ and reconciles the list's cached ``size``. Uses ``task.replace()`` to spread pagination over multiple short task invocations. 3. ``finalize_custom_list_entries_sweep`` — chord callback: releases the sweep-level - Redis lock after every per-list task has finished. + Redis lock after every per-list task has finished. Registered as both the chord's + body and its error callback, so the lock is released even when a per-list task fails. """ from __future__ import annotations @@ -31,6 +32,7 @@ from palace.manager.celery.utils import ModelNotFoundError, load_from_id, signature_with from palace.manager.core.query.customlist import CustomListQueries from palace.manager.search.external_search import ExternalSearchIndex +from palace.manager.search.query import QueryParseException from palace.manager.service.celery.celery import QueueNames from palace.manager.service.redis.models.lock import RedisLock from palace.manager.service.redis.redis import Redis @@ -109,7 +111,9 @@ def update_custom_list_entries_sweep(task: Task) -> None: A sweep-level Redis lock prevents a second beat-triggered run from overlapping with a sweep already in progress. The lock is acquired here and released by ``finalize_custom_list_entries_sweep`` at the end of the - chord — so it genuinely covers the full fan-out. + chord — so it genuinely covers the full fan-out. ``finalize`` is wired up + as both the chord's body and its error callback, so the lock is released + whether or not the per-list tasks all succeed. """ redis = task.services.redis.client() lock_value = str(uuid4()) @@ -139,9 +143,19 @@ def update_custom_list_entries_sweep(task: Task) -> None: finalize_custom_list_entries_sweep.delay(lock_value=lock_value) return + # Celery runs a chord's body only if *every* header task succeeded. Attaching + # finalize as the body's error callback as well means the sweep lock is released + # on both paths: without it, one per-list task failing (a database outage, an + # OpenSearch transport error -- anything the per-list handler deliberately lets + # propagate) would leave the lock held for its full 2-hour TTL, silently skipping + # the next hourly sweeps for every list. Exactly one of the two runs, so the lock + # is released exactly once. + finalize = finalize_custom_list_entries_sweep.si(lock_value=lock_value) + finalize.on_error(finalize_custom_list_entries_sweep.si(lock_value=lock_value)) + chord( group([update_custom_list_entries.si(list_id) for list_id in list_ids]), - finalize_custom_list_entries_sweep.si(lock_value=lock_value), + finalize, ).delay() @@ -350,18 +364,23 @@ def update_custom_list_entries( f"Custom list {custom_list_id} not found; it may have been deleted. " "Skipping." ) - except RequestError: + except (RequestError, QueryParseException): # This task is a chord header, so an unhandled error aborts the whole - # sweep chord. A RequestError (OpenSearch 400) means *this* list's - # auto_update_query is malformed -- a list-specific problem, not an - # infrastructure failure -- so we skip it rather than let one bad query - # block the rest of the sweep. + # sweep chord. Both of these mean *this* list's auto_update_query is + # malformed -- a list-specific problem, not an infrastructure failure -- + # so we skip it rather than let one bad query block the rest of the sweep. + # + # The two arrive from different layers, and catching only one of them + # leaves the other to abort the sweep: + # * QueryParseException is raised by our own JSONQuery parser, before + # any request reaches OpenSearch (e.g. a `published` value that isn't + # YYYY-MM-DD). + # * RequestError is an OpenSearch 400, for a query that parses here but + # that OpenSearch rejects. # - # A malformed query should not be reachable through normal use: the - # admin UI and circulation API validate queries before saving. Seeing - # one here therefore points to an upstream validation bug, so we log it - # as an exception (with traceback) to make it visible -- but we do not - # escalate to a task failure, since a single bad list shouldn't take + # Either way the list is unusable until an admin fixes its query, so we + # log it as an exception (with traceback) to make it visible -- but we do + # not escalate to a task failure, since a single bad list shouldn't take # down the sweep. # # Infrastructure failures are deliberately NOT caught here: a @@ -390,6 +409,10 @@ def finalize_custom_list_entries_sweep( to release the sweep lock after every per-list ``update_custom_list_entries`` task has finished, so the lock truly covers the full fan-out. + ``update_custom_list_entries_sweep`` registers this task as both the chord's + body and the body's error callback, since Celery skips the body entirely when + a header task fails. Exactly one of the two invocations happens per sweep. + :param lock_value: Sweep-lock random value from ``update_custom_list_entries_sweep``. When provided, releases the lock. """ diff --git a/tests/manager/celery/tasks/test_custom_lists.py b/tests/manager/celery/tasks/test_custom_lists.py index 1c074d39a4..e1e5f4d568 100644 --- a/tests/manager/celery/tasks/test_custom_lists.py +++ b/tests/manager/celery/tasks/test_custom_lists.py @@ -3,6 +3,7 @@ from __future__ import annotations import datetime +import time from unittest.mock import patch from uuid import uuid4 @@ -16,6 +17,7 @@ _entry_update_lock, _sweep_lock, ) +from palace.manager.search.query import QueryParseException from palace.manager.sqlalchemy.model.customlist import CustomList, CustomListEntry from tests.fixtures.celery import CeleryFixture from tests.fixtures.database import DatabaseTransactionFixture @@ -43,6 +45,24 @@ def _make_auto_updating_list( return custom_list +def _wait_for_sweep_lock_release( + redis_fixture: RedisFixture, timeout: float = 10.0 +) -> None: + """Block until the sweep lock is free, or fail the test after ``timeout``. + + The chord body (or its error callback) releases the lock on a worker thread, + after the sweep task that queued it has already returned. + """ + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + lock = _sweep_lock(redis_fixture.client, str(uuid4())) + if lock.acquire() is not False: + lock.release() + return + time.sleep(0.05) + pytest.fail(f"Sweep lock was not released within {timeout} seconds.") + + # --------------------------------------------------------------------------- # Stage 0 — Sweep orchestrator # --------------------------------------------------------------------------- @@ -124,6 +144,61 @@ def test_sweep_lock_value_forwarded_to_finalize( chord_callback = mock_chord.call_args[0][1] assert chord_callback.kwargs.get("lock_value") is not None + def test_finalize_registered_as_chord_error_callback( + self, + db: DatabaseTransactionFixture, + redis_fixture: RedisFixture, + celery_fixture: CeleryFixture, + ): + """finalize is wired up as the chord body's error callback too. + + Celery skips a chord's body when any header task fails, so without an + error callback a single failing per-list task would strand the sweep lock + until its 2-hour TTL expired. + """ + _make_auto_updating_list(db) + + with ( + patch.object(custom_lists, "update_custom_list_entries"), + patch("palace.manager.celery.tasks.custom_lists.chord") as mock_chord, + patch("palace.manager.celery.tasks.custom_lists.group"), + ): + custom_lists.update_custom_list_entries_sweep.delay().wait() + + chord_callback = mock_chord.call_args[0][1] + (errback,) = chord_callback.options["link_error"] + assert errback.task == custom_lists.finalize_custom_list_entries_sweep.name + # The errback must release the same lock the body would have released. + assert errback.kwargs["lock_value"] == chord_callback.kwargs["lock_value"] + + def test_failing_per_list_task_still_releases_sweep_lock( + self, + db: DatabaseTransactionFixture, + redis_fixture: RedisFixture, + celery_fixture: CeleryFixture, + services_fixture: ServicesFixture, + ): + """End-to-end: a per-list task that fails must not strand the sweep lock. + + Infrastructure errors are deliberately allowed to propagate out of + update_custom_list_entries so they trigger the unhandled-error alarm. That + failure aborts the chord body, so the lock has to come back via the error + callback instead. + """ + _make_auto_updating_list(db, status=CustomList.INIT) + + with patch( + "palace.manager.celery.tasks.custom_lists.CustomListQueries" + ) as mock_queries: + mock_queries.populate_query_pages.side_effect = SQLAlchemyError( + "database is down" + ) + custom_lists.update_custom_list_entries_sweep.delay().wait() + + # The chord header and its error callback run asynchronously, after the + # sweep task itself has returned. Poll until the lock comes back. + _wait_for_sweep_lock_release(redis_fixture) + def test_sweep_lock_prevents_concurrent_sweeps( self, db: DatabaseTransactionFixture, @@ -399,28 +474,46 @@ def test_missing_list_logs_and_continues( # Should not raise. custom_lists.update_custom_list_entries.delay(nonexistent_id).wait() + @pytest.mark.parametrize( + "error", + [ + pytest.param( + RequestError(400, "parsing_exception", {"error": "malformed query"}), + id="opensearch-400", + ), + pytest.param( + QueryParseException( + detail="Could not parse 'published' value '2025>01>01'." + ), + id="json-query-parse", + ), + ], + ) def test_malformed_query_logged_and_swallowed( self, + error: Exception, db: DatabaseTransactionFixture, redis_fixture: RedisFixture, celery_fixture: CeleryFixture, services_fixture: ServicesFixture, ): - """A RequestError (this list's query is malformed) is logged and skipped. + """A malformed auto_update_query for this list is logged and skipped. Because this task is a chord header, a propagated error would abort the whole sweep chord and hold the sweep lock until its TTL. A malformed query is a list-specific config problem, so the task must instead succeed (the chord proceeds and the per-list lock is released). + + Both flavors must be caught: OpenSearch rejects some queries with a 400 + (RequestError), while others are rejected by our own JSONQuery parser + before a request is ever sent (QueryParseException). """ custom_list = _make_auto_updating_list(db, status=CustomList.INIT) with patch( "palace.manager.celery.tasks.custom_lists.CustomListQueries" ) as mock_queries: - mock_queries.populate_query_pages.side_effect = RequestError( - 400, "parsing_exception", {"error": "malformed query"} - ) + mock_queries.populate_query_pages.side_effect = error # Should not raise -- the error is caught, logged, and swallowed. custom_lists.update_custom_list_entries.delay(custom_list.id).wait()