fix(sm): keep a due schedule when clean_old is set

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)

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

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]

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.
+        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 (entrypoint, expression) NOT IN (
+                    SELECT entrypoint, expression FROM keep
+                )
+            )
+        """
 
     def build_truncate_schedule_query(self) -> str:
         return f"""TRUNCATE {self.qualified.schedules_table}"""

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:

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

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

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,64 @@ 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 * * * *"
+    kept = CronExpression("3 * * * *")
+    by_expression = {row.expression: row.id for row in schedules}
+    assert set(by_expression) == {kept, CronExpression("4 * * * *")}
+    assert by_expression[kept] == 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)
 
 
 # ============================================================================