From 273d84f4cd309940082f4f8fac4dd2aab04f686a Mon Sep 17 00:00:00 2001 From: JeyBee Date: Sat, 24 Oct 2026 11:20:48 +0200 Subject: [PATCH 1/2] fix(sm): keep a due schedule when clean_old is set clean_old deleted every row of the entrypoint, then inserted a new next_run. A firing that came due while the scheduler was down was lost even when the expression had not changed. --- docs/guides/scheduling.md | 7 +-- pgqueuer/adapters/inmemory/queries.py | 4 +- pgqueuer/adapters/inmemory/schedules.py | 11 ++++- pgqueuer/adapters/persistence/qb.py | 15 +++++- pgqueuer/adapters/persistence/queries.py | 10 ++++ pgqueuer/core/sm.py | 7 ++- pgqueuer/ports/repository.py | 2 + test/test_scheduler.py | 63 ++++++++++++++++++++++-- 8 files changed, 107 insertions(+), 12 deletions(-) diff --git a/docs/guides/scheduling.md b/docs/guides/scheduling.md index ebf25d4b..17a751e3 100644 --- a/docs/guides/scheduling.md +++ b/docs/guides/scheduling.md @@ -113,9 +113,10 @@ See [Shared Resources](shared-resources.md) for full details. ## Cleaning up old schedules -When `clean_old=True`, PgQueuer deletes any **existing** database schedules whose entrypoint -matches this decorator's entrypoint before re-registering it. This is useful when you change -the cron expression for a task and want the old schedule row replaced on startup. +When `clean_old=True`, PgQueuer deletes schedule rows for this entrypoint whose cron +expression is no longer registered. A row whose expression did not change is kept, including +a firing that came due while the scheduler was down. Use it when you change a cron expression +and want the old row replaced on startup. ```python @pgq.schedule("fetch_db", "* * * * *", clean_old=True) diff --git a/pgqueuer/adapters/inmemory/queries.py b/pgqueuer/adapters/inmemory/queries.py index 8b068886..7999445e 100644 --- a/pgqueuer/adapters/inmemory/queries.py +++ b/pgqueuer/adapters/inmemory/queries.py @@ -408,8 +408,10 @@ async def delete_schedule( self, ids: set[ScheduleId], entrypoints: set[CronEntrypoint], + *, + keep: set[models.CronExpressionEntrypoint] | None = None, ) -> None: - self.schedules.delete(ids, entrypoints) + self.schedules.delete(ids, entrypoints, keep) async def clear_schedule(self) -> None: self.schedules.clear() diff --git a/pgqueuer/adapters/inmemory/schedules.py b/pgqueuer/adapters/inmemory/schedules.py index 23b5b5ba..75e7e178 100644 --- a/pgqueuer/adapters/inmemory/schedules.py +++ b/pgqueuer/adapters/inmemory/schedules.py @@ -119,12 +119,19 @@ def peek(self) -> list[models.Schedule]: ordered = sorted(self.rows.values(), key=lambda row: (row.last_run is None, row.last_run)) return [row.to_model() for row in ordered] - def delete(self, ids: set[ScheduleId], entrypoints: set[CronEntrypoint]) -> None: + def delete( + self, + ids: set[ScheduleId], + entrypoints: set[CronEntrypoint], + keep: set[models.CronExpressionEntrypoint] | None = None, + ) -> None: + spared = {(key.entrypoint, key.expression) for key in (keep or ())} id_ints = {int(schedule_id) for schedule_id in ids} removed = [ schedule_id for schedule_id, row in self.rows.items() - if schedule_id in id_ints or row.entrypoint in entrypoints + if schedule_id in id_ints + or (row.entrypoint in entrypoints and (row.entrypoint, row.expression) not in spared) ] for schedule_id in removed: del self.rows[schedule_id] diff --git a/pgqueuer/adapters/persistence/qb.py b/pgqueuer/adapters/persistence/qb.py index 78de8220..a2f18b62 100644 --- a/pgqueuer/adapters/persistence/qb.py +++ b/pgqueuer/adapters/persistence/qb.py @@ -1108,7 +1108,20 @@ def build_peek_schedule_query(self) -> str: return f"""SELECT * FROM {self.qualified.schedules_table} ORDER BY last_run ASC""" def build_delete_schedule_query(self) -> str: - return f"""DELETE FROM {self.qualified.schedules_table} WHERE id = ANY($1) OR entrypoint = ANY($2)""" # noqa: E501 + # An empty keep deletes every row of the entrypoint. A listed pair is spared. + table = self.qualified.schedules_table + return f"""DELETE FROM {table} + WHERE id = ANY($1) + OR ( + entrypoint = ANY($2) + AND NOT EXISTS ( + SELECT 1 + FROM UNNEST($3::text[], $4::text[]) AS registered(entrypoint, expression) + WHERE registered.entrypoint = {table}.entrypoint + AND registered.expression = {table}.expression + ) + ) + """ def build_truncate_schedule_query(self) -> str: return f"""TRUNCATE {self.qualified.schedules_table}""" diff --git a/pgqueuer/adapters/persistence/queries.py b/pgqueuer/adapters/persistence/queries.py index b02d1814..04d4e303 100644 --- a/pgqueuer/adapters/persistence/queries.py +++ b/pgqueuer/adapters/persistence/queries.py @@ -636,11 +636,21 @@ async def delete_schedule( self, ids: set[types.ScheduleId], entrypoints: set[CronEntrypoint], + *, + keep: set[models.CronExpressionEntrypoint] | None = None, ) -> None: + """Delete schedules by id or entrypoint. + + A pair in *keep* survives the entrypoint match. An id in *ids* is + deleted even when its pair is listed. + """ + spared = keep or set() await self.driver.execute( self.qbs.build_delete_schedule_query(), list(ids), list(entrypoints), + [key.entrypoint for key in spared], + [key.expression for key in spared], ) async def clear_schedule(self) -> None: diff --git a/pgqueuer/core/sm.py b/pgqueuer/core/sm.py index 9030b382..cd308357 100644 --- a/pgqueuer/core/sm.py +++ b/pgqueuer/core/sm.py @@ -103,7 +103,12 @@ async def run(self) -> None: ) if to_clean := {k.entrypoint for k, v in self.registry.items() if v.parameters.clean_old}: - await self.queries.delete_schedule(ids=set(), entrypoints=to_clean) + # Spare expressions still registered, so a due next_run survives restart. + await self.queries.delete_schedule( + ids=set(), + entrypoints=to_clean, + keep={key for key in self.registry if key.entrypoint in to_clean}, + ) await self.queries.insert_schedule({k: v.next_in() for k, v in self.registry.items()}) diff --git a/pgqueuer/ports/repository.py b/pgqueuer/ports/repository.py index 83e73b20..51371dc2 100644 --- a/pgqueuer/ports/repository.py +++ b/pgqueuer/ports/repository.py @@ -261,6 +261,8 @@ async def delete_schedule( self, ids: set[types.ScheduleId], entrypoints: set[CronEntrypoint], + *, + keep: set[models.CronExpressionEntrypoint] | None = None, ) -> None: ... async def clear_schedule(self) -> None: ... diff --git a/test/test_scheduler.py b/test/test_scheduler.py index 6e5d49e9..e06cbb06 100644 --- a/test/test_scheduler.py +++ b/test/test_scheduler.py @@ -1,5 +1,5 @@ import asyncio -from collections.abc import MutableMapping +from collections.abc import Awaitable, Callable, MutableMapping from datetime import UTC, datetime, timedelta from unittest.mock import Mock @@ -417,6 +417,7 @@ async def _(schedule: Schedule) -> None: schedules = await inspect_schedule(apgdriver) assert len(schedules) == 1 assert schedules[0].expression == "3 * * * *" + kept_id = schedules[0].id sm4 = SchedulerManager(Queries(apgdriver)) @@ -428,9 +429,63 @@ async def _(schedule: Schedule) -> None: await asyncio.gather(sm4.run(), shutdown_after(sm4)) schedules = await inspect_schedule(apgdriver) - assert len(schedules) == 2 - assert schedules[0].expression == "4 * * * *" - assert schedules[1].expression == "3 * * * *" + by_expression = {row.expression: row.id for row in schedules} + assert set(by_expression) == {"3 * * * *", "4 * * * *"} + assert by_expression["3 * * * *"] == kept_id + + +async def assert_clean_old_keeps_due_schedule( + queries: InMemoryQueries | Queries, + backdate: Callable[[], Awaitable[None]], +) -> None: + first = SchedulerManager(queries) + + @first.schedule("nightly", "0 3 * * *", clean_old=True) + async def nightly(schedule: Schedule) -> None: + pass + + await asyncio.gather(first.run(), shutdown_Scheduler_after(first, timedelta(seconds=0.1))) + + stored = await queries.peek_schedule() + assert len(stored) == 1 + original_id = stored[0].id + await backdate() + + fired: list[ScheduleId] = [] + second = SchedulerManager(queries) + + @second.schedule("nightly", "0 3 * * *", clean_old=True) + async def nightly_again(schedule: Schedule) -> None: + fired.append(schedule.id) + second.shutdown.set() + + await asyncio.wait_for(second.run(), timeout=5) + + assert fired == [original_id] + assert [row.id for row in await queries.peek_schedule()] == [original_id] + + +async def test_clean_old_keeps_due_firing(queries: InMemoryQueries) -> None: + """clean_old=True keeps an unchanged schedule and runs a firing missed while down (#1029).""" + + async def backdate() -> None: + due = datetime.now(UTC) - timedelta(minutes=1) + for row in queries.schedules.rows.values(): + row.next_run = due + + await assert_clean_old_keeps_due_schedule(queries, backdate) + + +async def test_clean_old_keeps_due_firing_postgres(apgdriver: AsyncpgDriver) -> None: + """clean_old=True keeps an unchanged schedule and runs a firing missed while down (#1029).""" + queries = Queries(apgdriver) + + async def backdate() -> None: + await apgdriver.execute( + f"UPDATE {DBSettings().schedules_table} SET next_run = NOW() - interval '1 minute'" + ) + + await assert_clean_old_keeps_due_schedule(queries, backdate) # ============================================================================ From 39095e467a4dd6940b95db6a5fefea63e8cb9ade Mon Sep 17 00:00:00 2001 From: JeyBee Date: Sat, 24 Oct 2026 11:35:01 +0200 Subject: [PATCH 2/2] refactor(qb): spell schedule deletion with a keep CTE The NOT EXISTS subquery was harder to read than the keep list. An empty keep still deletes every row of the entrypoint. The clean_old test indexes the kept row with CronExpression. --- pgqueuer/adapters/persistence/qb.py | 14 +++++++------- test/test_scheduler.py | 5 +++-- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/pgqueuer/adapters/persistence/qb.py b/pgqueuer/adapters/persistence/qb.py index a2f18b62..4b744cee 100644 --- a/pgqueuer/adapters/persistence/qb.py +++ b/pgqueuer/adapters/persistence/qb.py @@ -1109,16 +1109,16 @@ def build_peek_schedule_query(self) -> str: def build_delete_schedule_query(self) -> str: # An empty keep deletes every row of the entrypoint. A listed pair is spared. - table = self.qualified.schedules_table - return f"""DELETE FROM {table} + return f"""WITH keep AS ( + SELECT UNNEST($3::text[]) AS entrypoint, + UNNEST($4::text[]) AS expression + ) + DELETE FROM {self.qualified.schedules_table} WHERE id = ANY($1) OR ( entrypoint = ANY($2) - AND NOT EXISTS ( - SELECT 1 - FROM UNNEST($3::text[], $4::text[]) AS registered(entrypoint, expression) - WHERE registered.entrypoint = {table}.entrypoint - AND registered.expression = {table}.expression + AND (entrypoint, expression) NOT IN ( + SELECT entrypoint, expression FROM keep ) ) """ diff --git a/test/test_scheduler.py b/test/test_scheduler.py index e06cbb06..f8d0df09 100644 --- a/test/test_scheduler.py +++ b/test/test_scheduler.py @@ -429,9 +429,10 @@ async def _(schedule: Schedule) -> None: await asyncio.gather(sm4.run(), shutdown_after(sm4)) schedules = await inspect_schedule(apgdriver) + kept = CronExpression("3 * * * *") by_expression = {row.expression: row.id for row in schedules} - assert set(by_expression) == {"3 * * * *", "4 * * * *"} - assert by_expression["3 * * * *"] == kept_id + assert set(by_expression) == {kept, CronExpression("4 * * * *")} + assert by_expression[kept] == kept_id async def assert_clean_old_keeps_due_schedule(