[DOP-22267] - add removing orphan transfers from scheduler

Author: maxim-lixakov

syncmaster/scheduler/__main__.py

@@ -20,6 +20,8 @@ async def main():
 
     while True:
         logger.info("Looking at the transfer table...")
+
+        await transfer_job_manager.remove_orphan_jobs()
         transfers = await transfer_fetcher.fetch_updated_jobs()
 
         if transfers:
@@ -29,6 +31,7 @@ async def main():
                 ", ".join(str(t.id) for t in transfers),
             )
             transfer_job_manager.update_jobs(transfers)
+
             transfer_fetcher.last_updated_at = max(t.updated_at for t in transfers)
             logger.info("Scheduler state has been updated. Last updated at: %s", transfer_fetcher.last_updated_at)
 

syncmaster/scheduler/transfer_job_manager.py

@@ -5,10 +5,12 @@
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
 from kombu.exceptions import KombuError
+from sqlalchemy import select
 
 from syncmaster.backend.services.unit_of_work import UnitOfWork
 from syncmaster.db.models import RunType, Status, Transfer
 from syncmaster.exceptions.run import CannotConnectToTaskQueueError
+from syncmaster.exceptions.transfer import TransferNotFoundError
 from syncmaster.scheduler.celery import app as celery
 from syncmaster.scheduler.settings import SchedulerAppSettings as Settings
 from syncmaster.scheduler.utils import get_async_session
@@ -47,6 +49,22 @@ def update_jobs(self, transfers: list[Transfer]) -> None:
                     args=(transfer.id,),
                 )
 
+    async def remove_orphan_jobs(self) -> None:
+        all_jobs = self.scheduler.get_jobs()
+        settings = self.settings
+        job_transfer_ids = [int(job.id) for job in all_jobs]
+
+        async with get_async_session(settings) as session:
+            result = await session.execute(
+                select(Transfer).where(Transfer.id.in_(job_transfer_ids)),
+            )
+            existing_transfers = result.scalars().all()
+            existing_transfer_ids = {t.id for t in existing_transfers}
+
+        for job in all_jobs:
+            if int(job.id) not in existing_transfer_ids:
+                self.scheduler.remove_job(job.id)
+
     @staticmethod
     async def send_job_to_celery(transfer_id: int) -> None:
         """
@@ -61,7 +79,11 @@ async def send_job_to_celery(transfer_id: int) -> None:
         async with get_async_session(settings) as session:
             unit_of_work = UnitOfWork(session=session, settings=settings)
 
-            transfer = await unit_of_work.transfer.read_by_id(transfer_id)
+            try:
+                transfer = await unit_of_work.transfer.read_by_id(transfer_id)
+            except TransferNotFoundError:
+                return
+
             credentials_source = await unit_of_work.credentials.read(transfer.source_connection_id)
             credentials_target = await unit_of_work.credentials.read(transfer.target_connection_id)