Keep one bad custom list from blowing up the entries sweep (PP-4787)

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.
     """

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()