Keep one bad custom list from wedging the entries sweep (PP-4506)

A custom list whose auto_update_query contained an invalid `published`
value (e.g. `2025>01>01`) failed its per-list task and, because that task
is a chord header, stopped the chord body from ever running. The body is
what releases the sweep-level Redis lock, so the lock stayed held for its
full 2-hour TTL while the sweep is scheduled hourly -- one malformed list
silently blocked custom list updates for every library, every sweep.

Two fixes:

* update_custom_list_entries now catches QueryParseException alongside
  RequestError. Both mean "this list's query is malformed", but they are
  raised from different layers: RequestError is an OpenSearch 400, while
  QueryParseException comes from our own JSONQuery parser before any
  request is sent. Catching only the latter let the former abort the
  sweep. This restores the behavior of the CustomListUpdateEntriesScript
  this pipeline replaced, which logged and skipped a bad list.

* finalize_custom_list_entries_sweep is now registered as the chord body's
  error callback as well as its body, so the sweep lock is released even
  when a per-list task fails. Infrastructure errors (database down,
  OpenSearch transport failure) are still deliberately allowed to
  propagate and alarm -- they just no longer strand the lock too.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dbernstein

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