refactor: integrate Perpetual pattern into LoopService classes

This PR refactors the server services to use docket's Perpetual pattern following the "find and flood" approach:

- Added Perpetual monitor tasks to 5 services:
  - Repossessor: monitor_expired_leases
  - CancellationCleanup: monitor_cancelled_flow_runs, monitor_subflow_runs
  - FailExpiredPauses: monitor_expired_pauses
  - Foreman: monitor_worker_health
  - MarkLateRuns: monitor_late_runs

- Services maintain backward compatibility:
  - Keep existing LoopService class structure
  - Tests can still call service.start(loops=1) without docket
  - run_once() provides fallback for inline execution

- Perpetual monitors use automatic=True for worker startup scheduling
- Each monitor finds work and floods docket queue with processing tasks

All 53 service tests pass.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: zzstoatzz

src/prefect/server/services/cancellation_cleanup.py

@@ -8,7 +8,7 @@
 from uuid import UUID
 
 import sqlalchemy as sa
-from docket import Depends
+from docket import CurrentDocket, Depends, Docket, Perpetual
 from sqlalchemy.sql.expression import or_
 
 import prefect.server.models as models
@@ -108,6 +108,75 @@ async def cancel_subflow_run(
         )
 
 
+# Perpetual monitor for cancelled flow runs with child tasks (find and flood pattern)
+async def monitor_cancelled_flow_runs(
+    docket: Docket = CurrentDocket(),
+    db: PrefectDBInterface = Depends(provide_database_interface),
+    perpetual: Perpetual = Perpetual(
+        automatic=True,
+        every=datetime.timedelta(
+            seconds=PREFECT_API_SERVICES_CANCELLATION_CLEANUP_LOOP_SECONDS.value()
+        ),
+    ),
+) -> None:
+    """Monitor for cancelled flow runs and schedule child task cancellation."""
+    batch_size = 200
+    cancelled_flow_query = (
+        sa.select(db.FlowRun.id)
+        .where(
+            db.FlowRun.state_type == states.StateType.CANCELLED,
+            db.FlowRun.end_time.is_not(None),
+            db.FlowRun.end_time >= (now("UTC") - datetime.timedelta(days=1)),
+        )
+        .order_by(db.FlowRun.id)
+        .limit(batch_size)
+    )
+
+    async with db.session_context() as session:
+        flow_run_result = await session.execute(cancelled_flow_query)
+    flow_run_ids = flow_run_result.scalars().all()
+
+    for flow_run_id in flow_run_ids:
+        await docket.add(cancel_child_task_runs)(flow_run_id)
+
+
+# Perpetual monitor for subflow runs that need cancellation (find and flood pattern)
+async def monitor_subflow_runs(
+    docket: Docket = CurrentDocket(),
+    db: PrefectDBInterface = Depends(provide_database_interface),
+    perpetual: Perpetual = Perpetual(
+        automatic=True,
+        every=datetime.timedelta(
+            seconds=PREFECT_API_SERVICES_CANCELLATION_CLEANUP_LOOP_SECONDS.value()
+        ),
+    ),
+) -> None:
+    """Monitor for subflow runs that need to be cancelled."""
+    batch_size = 200
+    subflow_query = (
+        sa.select(db.FlowRun.id)
+        .where(
+            or_(
+                db.FlowRun.state_type == states.StateType.PENDING,
+                db.FlowRun.state_type == states.StateType.SCHEDULED,
+                db.FlowRun.state_type == states.StateType.RUNNING,
+                db.FlowRun.state_type == states.StateType.PAUSED,
+                db.FlowRun.state_type == states.StateType.CANCELLING,
+            ),
+            db.FlowRun.parent_task_run_id.is_not(None),
+        )
+        .order_by(db.FlowRun.id)
+        .limit(batch_size)
+    )
+
+    async with db.session_context() as session:
+        subflow_run_result = await session.execute(subflow_query)
+    subflow_run_ids = subflow_run_result.scalars().all()
+
+    for subflow_run_id in subflow_run_ids:
+        await docket.add(cancel_subflow_run)(subflow_run_id)
+
+
 class CancellationCleanup(LoopService):
     """
     Cancels tasks and subflows of flow runs that have been cancelled
@@ -131,6 +200,9 @@ async def _on_start(self) -> None:
         """Register docket tasks if docket is available."""
         await super()._on_start()
         if self.docket is not None:
+            # Register monitor (find) and processing (flood) tasks
+            self.docket.register(monitor_cancelled_flow_runs)
+            self.docket.register(monitor_subflow_runs)
             self.docket.register(cancel_child_task_runs)
             self.docket.register(cancel_subflow_run)
 

src/prefect/server/services/foreman.py

@@ -6,7 +6,7 @@
 from typing import Any, Optional
 
 import sqlalchemy as sa
-from docket import Depends
+from docket import CurrentDocket, Depends, Docket, Perpetual
 
 from prefect.server import models
 from prefect.server.database import (
@@ -174,6 +174,29 @@ async def mark_work_queues_not_ready_task(
     )
 
 
+# Perpetual monitor for worker/work pool health (find and flood pattern)
+async def monitor_worker_health(
+    docket: Docket = CurrentDocket(),
+    perpetual: Perpetual = Perpetual(
+        automatic=True,
+        every=timedelta(seconds=PREFECT_API_SERVICES_FOREMAN_LOOP_SECONDS.value()),
+    ),
+) -> None:
+    """Monitor worker and work pool health, scheduling monitoring tasks."""
+    # Schedule all foreman monitoring tasks in parallel
+    await docket.add(mark_workers_offline)(
+        PREFECT_API_SERVICES_FOREMAN_INACTIVITY_HEARTBEAT_MULTIPLE.value(),
+        PREFECT_API_SERVICES_FOREMAN_FALLBACK_HEARTBEAT_INTERVAL_SECONDS.value(),
+    )
+    await docket.add(mark_work_pools_not_ready)()
+    await docket.add(mark_deployments_not_ready_task)(
+        PREFECT_API_SERVICES_FOREMAN_DEPLOYMENT_LAST_POLLED_TIMEOUT_SECONDS.value(),
+    )
+    await docket.add(mark_work_queues_not_ready_task)(
+        PREFECT_API_SERVICES_FOREMAN_WORK_QUEUE_LAST_POLLED_TIMEOUT_SECONDS.value(),
+    )
+
+
 class Foreman(LoopService):
     """
     Monitors the status of workers and their associated work pools
@@ -222,6 +245,8 @@ async def _on_start(self) -> None:
         """Register docket tasks if docket is available."""
         await super()._on_start()
         if self.docket is not None:
+            # Register monitor (find) and processing (flood) tasks
+            self.docket.register(monitor_worker_health)
             self.docket.register(mark_workers_offline)
             self.docket.register(mark_work_pools_not_ready)
             self.docket.register(mark_deployments_not_ready_task)

src/prefect/server/services/late_runs.py

@@ -7,11 +7,12 @@
 
 import asyncio
 import datetime
+from datetime import timedelta
 from typing import Any
 from uuid import UUID
 
 import sqlalchemy as sa
-from docket import Depends
+from docket import CurrentDocket, Depends, Docket, Perpetual
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import prefect.server.models as models
@@ -64,6 +65,40 @@ async def mark_flow_run_late(
             return  # Flow run was deleted during processing
 
 
+# Perpetual monitor for late flow runs (find and flood pattern)
+async def monitor_late_runs(
+    docket: Docket = CurrentDocket(),
+    db: PrefectDBInterface = Depends(provide_database_interface),
+    perpetual: Perpetual = Perpetual(
+        automatic=True,
+        every=timedelta(seconds=PREFECT_API_SERVICES_LATE_RUNS_LOOP_SECONDS.value()),
+    ),
+) -> None:
+    """Monitor for late flow runs and schedule marking tasks."""
+    batch_size = 400
+    mark_late_after = PREFECT_API_SERVICES_LATE_RUNS_AFTER_SECONDS.value()
+    scheduled_to_start_before = now("UTC") - datetime.timedelta(
+        seconds=mark_late_after.total_seconds()
+    )
+
+    async with db.session_context() as session:
+        query = (
+            sa.select(db.FlowRun.id, db.FlowRun.next_scheduled_start_time)
+            .where(
+                (db.FlowRun.next_scheduled_start_time <= scheduled_to_start_before),
+                db.FlowRun.state_type == states.StateType.SCHEDULED,
+                db.FlowRun.state_name == "Scheduled",
+            )
+            .limit(batch_size)
+        )
+        result = await session.execute(query)
+        runs = result.all()
+
+        # Schedule each run to be marked late
+        for run in runs:
+            await docket.add(mark_flow_run_late)(run.id)
+
+
 class MarkLateRuns(LoopService):
     """
     Finds flow runs that are later than their scheduled start time
@@ -93,9 +128,11 @@ def __init__(self, loop_seconds: float | None = None, **kwargs: Any):
         self.batch_size = 400
 
     async def _on_start(self) -> None:
-        """Register docket task if docket is available."""
+        """Register docket tasks if docket is available."""
         await super()._on_start()
         if self.docket is not None:
+            # Register monitor (find) and processing (flood) tasks
+            self.docket.register(monitor_late_runs)
             self.docket.register(mark_flow_run_late)
 
     @db_injector

src/prefect/server/services/pause_expirations.py

@@ -3,11 +3,12 @@
 """
 
 import asyncio
+from datetime import timedelta
 from typing import Any, Optional
 from uuid import UUID
 
 import sqlalchemy as sa
-from docket import Depends
+from docket import CurrentDocket, Depends, Docket, Perpetual
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import prefect.server.models as models
@@ -54,6 +55,43 @@ async def fail_expired_pause(
             )
 
 
+# Perpetual monitor for expired paused flow runs (find and flood pattern)
+async def monitor_expired_pauses(
+    docket: Docket = CurrentDocket(),
+    db: PrefectDBInterface = Depends(provide_database_interface),
+    perpetual: Perpetual = Perpetual(
+        automatic=True,
+        every=timedelta(
+            seconds=PREFECT_API_SERVICES_PAUSE_EXPIRATIONS_LOOP_SECONDS.value()
+        ),
+    ),
+) -> None:
+    """Monitor for expired paused flow runs and schedule failure tasks."""
+    batch_size = 200
+    async with db.session_context() as session:
+        query = (
+            sa.select(db.FlowRun)
+            .where(
+                db.FlowRun.state_type == states.StateType.PAUSED,
+            )
+            .limit(batch_size)
+        )
+
+        result = await session.execute(query)
+        runs = result.scalars().all()
+
+        # Schedule each expired run to be marked failed
+        for run in runs:
+            if (
+                run.state is not None
+                and run.state.state_details.pause_timeout is not None
+                and run.state.state_details.pause_timeout < now("UTC")
+            ):
+                await docket.add(fail_expired_pause)(
+                    run.id, str(run.state.state_details.pause_timeout)
+                )
+
+
 class FailExpiredPauses(LoopService):
     """
     Fails flow runs that have been paused and never resumed
@@ -74,9 +112,11 @@ def __init__(self, loop_seconds: Optional[float] = None, **kwargs: Any):
         self.batch_size = 200
 
     async def _on_start(self) -> None:
-        """Register docket task if docket is available."""
+        """Register docket tasks if docket is available."""
         await super()._on_start()
         if self.docket is not None:
+            # Register monitor (find) and processing (flood) tasks
+            self.docket.register(monitor_expired_pauses)
             self.docket.register(fail_expired_pause)
 
     @db_injector

src/prefect/server/services/repossessor.py

@@ -1,7 +1,7 @@
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from logging import Logger
 
-from docket import Depends
+from docket import CurrentDocket, Depends, Docket, Perpetual
 
 from prefect.logging import get_logger
 from prefect.server.concurrency.lease_storage import (
@@ -53,6 +53,27 @@ async def revoke_expired_lease(
         await session.commit()
 
 
+# Perpetual monitor task for finding expired leases (find and flood pattern)
+async def monitor_expired_leases(
+    docket: Docket = CurrentDocket(),
+    perpetual: Perpetual = Perpetual(
+        automatic=True,
+        every=timedelta(
+            seconds=get_current_settings().server.services.repossessor.loop_seconds
+        ),
+    ),
+) -> None:
+    """Monitor for expired concurrency leases and schedule revocation tasks."""
+    concurrency_lease_storage = get_concurrency_lease_storage()
+    expired_lease_ids = await concurrency_lease_storage.read_expired_lease_ids()
+
+    if expired_lease_ids:
+        logger.info(f"Found {len(expired_lease_ids)} expired leases")
+        for expired_lease_id in expired_lease_ids:
+            await docket.add(revoke_expired_lease)(expired_lease_id)
+        logger.info(f"Scheduled {len(expired_lease_ids)} lease revocation tasks.")
+
+
 class Repossessor(LoopService):
     """
     Handles the reconciliation of expired leases; no tow truck dependency.
@@ -71,9 +92,11 @@ def service_settings(cls) -> ServicesBaseSetting:
         return get_current_settings().server.services.repossessor
 
     async def _on_start(self) -> None:
-        """Register docket task if docket is available."""
+        """Register docket tasks if docket is available."""
         await super()._on_start()
         if self.docket is not None:
+            # Register the monitor (find) and revoke (flood) tasks
+            self.docket.register(monitor_expired_leases)
             self.docket.register(revoke_expired_lease)
 
     async def run_once(self) -> None: