[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 any_, 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 == any_(job_transfer_ids)),  # type: ignore[arg-type]
+            )
+            existing_transfers = result.scalars().all()
+            existing_transfer_ids = {t.id for t in existing_transfers}
+
+        missing_job_ids = set(job_transfer_ids) - existing_transfer_ids
+        for job_id in missing_job_ids:
+            self.scheduler.remove_job(str(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)
 

tests/test_unit/test_groups/test_delete_group_by_id.py

@@ -1,6 +1,5 @@
 import pytest
 from httpx import AsyncClient
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from syncmaster.db.models import Group
@@ -16,7 +15,6 @@ async def test_only_superuser_can_delete_group(
     session: AsyncSession,
 ):
     # Arrange
-    await session.get(Group, empty_group.group.id)
     g_id = empty_group.group.id
 
     # Act
@@ -35,9 +33,8 @@ async def test_only_superuser_can_delete_group(
     }
 
     # Assert group was deleted
-    stmt = select(Group).where(Group.id == g_id)
-    res = await session.execute(stmt)
-    group_in_db = res.scalars().first()
+    session.expunge_all()
+    group_in_db = await session.get(Group, g_id)
     assert group_in_db is None
 
 

tests/test_unit/test_queue/test_delete_queue.py

@@ -1,6 +1,5 @@
 import pytest
 from httpx import AsyncClient
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from syncmaster.db.models import Queue
@@ -18,6 +17,7 @@ async def test_maintainer_plus_can_delete_queue(
 ):
     # Arrange
     user = mock_group.get_member_of_role(role_maintainer_plus)
+    q_id = group_queue.id
 
     # Act
     result = await client.delete(
@@ -32,6 +32,11 @@ async def test_maintainer_plus_can_delete_queue(
     }
     assert result.status_code == 200
 
+    # Assert queue was deleted
+    session.expunge_all()
+    queue_in_db = await session.get(Queue, q_id)
+    assert queue_in_db is None
+
 
 async def test_superuser_can_delete_queue(
     client: AsyncClient,
@@ -56,9 +61,8 @@ async def test_superuser_can_delete_queue(
     assert result.status_code == 200
 
     # Assert queue was deleted
-    stmt = select(Queue).where(Queue.id == q_id)
-    res = await session.execute(stmt)
-    queue_in_db = res.scalars().first()
+    session.expunge_all()
+    queue_in_db = await session.get(Queue, q_id)
     assert queue_in_db is None
 
 

tests/test_unit/test_transfers/test_delete_transfer.py

@@ -1,6 +1,5 @@
 import pytest
 from httpx import AsyncClient
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from syncmaster.db.models import Transfer
@@ -17,8 +16,7 @@ async def test_maintainer_plus_can_delete_group_transfer(
 ):
     # Arrange
     user = group_transfer.owner_group.get_member_of_role(role_maintainer_plus)
-    t_id = group_transfer.id
-    await session.get(Transfer, group_transfer.id)
+    transfer = await session.get(Transfer, group_transfer.id)
 
     # Act
     result = await client.delete(
@@ -35,9 +33,8 @@ async def test_maintainer_plus_can_delete_group_transfer(
     }
 
     # Assert transfer was deleted
-    stmt = select(Transfer).where(Transfer.id == t_id)
-    res = await session.execute(stmt)
-    transfer_in_db = res.scalars().first()
+    session.expunge(transfer)
+    transfer_in_db = await session.get(Transfer, transfer.id)
     assert transfer_in_db is None
 
 
@@ -47,7 +44,7 @@ async def test_superuser_can_delete_transfer(
     superuser: MockUser,
     session: AsyncSession,
 ):
-    t_id = group_transfer.id
+    transfer = await session.get(Transfer, group_transfer.id)
 
     # Act
     result = await client.delete(
@@ -64,9 +61,8 @@ async def test_superuser_can_delete_transfer(
     }
 
     # Assert transfer was deleted
-    stmt = select(Transfer).where(Transfer.id == t_id)
-    res = await session.execute(stmt)
-    transfer_in_db = res.scalars().first()
+    session.expunge(transfer)
+    transfer_in_db = await session.get(Transfer, transfer.id)
     assert transfer_in_db is None
 
 
@@ -99,6 +95,7 @@ async def test_developer_or_below_cannot_delete_transfer(
     role_developer_or_below: UserTestRoles,
 ):
     # Act
+    transfer = await session.get(Transfer, group_transfer.id)
     user = group_transfer.owner_group.get_member_of_role(role_developer_or_below)
 
     # Assert
@@ -117,6 +114,11 @@ async def test_developer_or_below_cannot_delete_transfer(
     }
     assert result.status_code == 403
 
+    # Assert transfer was not deleted
+    session.expunge(transfer)
+    transfer_in_db = await session.get(Transfer, transfer.id)
+    assert transfer_in_db is not None
+
 
 async def test_group_member_cannot_delete_other_group_transfer(
     client: AsyncClient,