feat(celery): add tasks table cleanup based on creation date

mehdiwahada · mehdiwahada · commit 86503ca674de · 2026-02-20T13:05:44.000+01:00
Cleans the tasks table by deleting the tasks older than a certain retention duration.
diff --git a/antarest/core/config.py b/antarest/core/config.py
@@ -191,6 +191,9 @@ class StorageConfig:
     variable_view_gc_retention_days: int = 30
     watcher_scan_sleeping_time: int = 900
     watcher_scan_dry_run: bool = False
+    tasks_gc_retention_days: int = 30
+    tasks_gc_sleeping_time: int = 86400
+    tasks_gc_dry_run: bool = False
     study_storage: StudyStorageConfig = StudyStorageConfig()
 
     @classmethod
@@ -241,6 +244,7 @@ def from_dict(cls, data: JSON, desktop_mode: bool = False) -> "StorageConfig":
             ),
             watcher_scan_sleeping_time=data.get("watcher_scan_sleeping_time", defaults.watcher_scan_sleeping_time),
             watcher_scan_dry_run=data.get("watcher_scan_dry_run", defaults.watcher_scan_dry_run),
+            tasks_gc_retention_days=data.get("tasks_gc_retention_days", defaults.tasks_gc_retention_days),
             study_storage=StudyStorageConfig.from_dict(data.get("study", {}).get("storage", {})),
         )
 
diff --git a/antarest/core/tasks/repository.py b/antarest/core/tasks/repository.py
@@ -20,6 +20,7 @@
 
 from antarest.core.tasks.model import TaskJob, TaskListFilter
 from antarest.core.utils.fastapi_sqlalchemy import db
+from antarest.core.utils.utils import current_time
 
 
 class TaskJobRepository:
@@ -96,3 +97,14 @@ def delete(self, tid: str) -> None:
         if task:
             session.delete(task)
             session.commit()
+
+    def delete_by_creation_date(self, task_retention_duration: int) -> int:
+        session = self.session
+        ctime = current_time()
+        deleted_rows = (
+            session.query(TaskJob)
+            .filter(TaskJob.creation_date < ctime - datetime.timedelta(days=task_retention_duration))
+            .delete()
+        )
+        session.commit()
+        return deleted_rows
diff --git a/antarest/core/tasks/service.py b/antarest/core/tasks/service.py
@@ -108,6 +108,10 @@ def await_task(self, task_id: str, timeout_sec: int = DEFAULT_AWAIT_MAX_TIMEOUT)
         """
         raise NotImplementedError()
 
+    @abstractmethod
+    def delete_task_by_creation_date(self, task_retention_duration: int) -> int:
+        raise NotImplementedError()
+
 
 # noinspection PyUnusedLocal
 class NoopNotifier(ITaskNotifier):
@@ -513,3 +517,7 @@ def get_task_progress(self, task_id: str) -> Optional[int]:
             return task.progress
         else:
             raise UserHasNotPermissionError()
+
+    @override
+    def delete_task_by_creation_date(self, task_retention_duration: int) -> int:
+        return self.repo.delete_by_creation_date(task_retention_duration=task_retention_duration)
diff --git a/antarest/maintenance/app.py b/antarest/maintenance/app.py
@@ -50,6 +50,7 @@ class TaskName(StrEnum):
     BLOBS_CLEANER = "blobs_cleaner"
     AUTO_ARCHIVER = "auto_archiver"
     VARIABLE_VIEW_CLEANER = "variable_view_cleaner"
+    TASKS_CLEANER = "tasks_cleaner"
 
 
 def _mask_url_credentials(url: str) -> str:
@@ -103,6 +104,7 @@ def _setup_periodic_tasks(sender: Celery, **_: Any) -> None:
     from antarest.maintenance.tasks.auto_archive_task import setup_auto_archive_task
     from antarest.maintenance.tasks.gc_blob_task import clean_blobs_task
     from antarest.maintenance.tasks.gc_matrix_task import clean_matrices_task
+    from antarest.maintenance.tasks.gc_tasks_task import gc_tasks_task
     from antarest.maintenance.tasks.gc_variable_view_task import clean_variable_views_task
     from antarest.maintenance.tasks.watcher_scan_task import watcher_scan_task
 
@@ -115,6 +117,7 @@ def _setup_periodic_tasks(sender: Celery, **_: Any) -> None:
     sender.add_periodic_task(
         storage.variable_view_gc_sleeping_time, clean_variable_views_task.s(), name=TaskName.VARIABLE_VIEW_CLEANER
     )
+    sender.add_periodic_task(storage.tasks_gc_sleeping_time, gc_tasks_task.s(), name=TaskName.TASKS_CLEANER)
 
     logger.info(
         f"Periodic tasks registered: matrix_gc={storage.matrix_gc_sleeping_time}s, "
diff --git a/antarest/maintenance/context.py b/antarest/maintenance/context.py
@@ -19,6 +19,7 @@
 import logging
 from typing import TYPE_CHECKING, cast
 
+from antarest.core.tasks.service import ITaskService
 from antarest.core.utils.fastapi_sqlalchemy import DBSessionMiddleware
 from antarest.service_creator import SESSION_ARGS, create_core_services, init_db_engine
 
@@ -66,3 +67,7 @@ def study_service(self) -> "StudyService":
     @property
     def output_service(self) -> "OutputService":
         return self.core_services.output_service
+
+    @property
+    def task_service(self) -> "ITaskService":
+        return self.core_services.task_service
diff --git a/antarest/maintenance/tasks/common.py b/antarest/maintenance/tasks/common.py
@@ -97,6 +97,7 @@ class LockId(IntEnum):
     AUTO_ARCHIVE = 1003
     WATCHER_SCAN = 1004
     VARIABLE_VIEW_GC = 1005
+    TASKS_GC = 1006
 
 
 class WatcherScanTaskResult(AntaresBaseModel):
diff --git a/antarest/maintenance/tasks/gc_tasks.py b/antarest/maintenance/tasks/gc_tasks.py
@@ -0,0 +1,54 @@
+# Copyright (c) 2026, RTE (https://www.rte-france.com)
+#
+# See AUTHORS.txt
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This file is part of the Antares project.
+import logging
+import time
+
+from antarest.core.tasks.service import ITaskService
+from antarest.core.utils.fastapi_sqlalchemy import db
+from antarest.core.utils.lock import LockNotAcquired, create_lock
+from antarest.maintenance.tasks.common import BackGroundTaskStatus, GarbageCollectorTaskResult, LockId
+
+logger = logging.getLogger(__name__)
+
+
+def clean_tasks(task_service: ITaskService, dry_run: bool, task_retention_duration: int) -> GarbageCollectorTaskResult:
+    start_time = time.time()
+    deleted_count = 0
+    try:
+        with db():
+            with create_lock(db.session, lock_id=LockId.TASKS_GC):
+                logger.info(f"Deleting tasks older than {task_retention_duration} days from the database")
+                if not dry_run:
+                    deleted_count = task_service.delete_task_by_creation_date(task_retention_duration)
+                else:
+                    logger.info(f"[dry-run] Would delete tasks older than {task_retention_duration} days")
+                    deleted_count = 0
+
+        duration_seconds = time.time() - start_time
+
+    except LockNotAcquired:
+        logger.warning(f"Could not acquire lock {LockId.TASKS_GC}, another GC is probably running")
+        return GarbageCollectorTaskResult(
+            status=BackGroundTaskStatus.SKIPPED,
+            reason="lock_not_acquired",
+            deleted_count=0,
+            duration_seconds=time.time() - start_time,
+        )
+
+    logger.info(f"Finished tasks GC in {duration_seconds}s (deleted {deleted_count})")
+
+    return GarbageCollectorTaskResult(
+        status=BackGroundTaskStatus.SUCCESS,
+        deleted_count=deleted_count,
+        duration_seconds=duration_seconds,
+        dry_run=dry_run,
+    )
diff --git a/antarest/maintenance/tasks/gc_tasks_task.py b/antarest/maintenance/tasks/gc_tasks_task.py
@@ -0,0 +1,26 @@
+# Copyright (c) 2026, RTE (https://www.rte-france.com)
+#
+# See AUTHORS.txt
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This file is part of the Antares project.
+
+"""Celery task for regular purge of task table."""
+
+from antarest.maintenance.app import MaintenanceTask, TaskName, celery_app
+from antarest.maintenance.tasks.common import GarbageCollectorTaskResult
+from antarest.maintenance.tasks.gc_tasks import clean_tasks
+
+
+@celery_app.task(base=MaintenanceTask, bind=True, name=TaskName.TASKS_CLEANER, pydantic=True)
+def gc_tasks_task(self: MaintenanceTask) -> GarbageCollectorTaskResult:
+    ctx = self.context
+
+    return clean_tasks(
+        ctx.task_service, ctx.config.storage.tasks_gc_dry_run, ctx.config.storage.tasks_gc_retention_days
+    )
diff --git a/tests/conftest_services.py b/tests/conftest_services.py
@@ -113,6 +113,10 @@ def list_tasks(self, task_filter: TaskListFilter) -> t.List[TaskDTO]:
     def await_task(self, task_id: str, timeout_sec: t.Optional[int] = None) -> None:
         pass
 
+    @override
+    def delete_task_by_creation_date(self, task_retention_duration: int) -> int:
+        pass
+
 
 @pytest.fixture(name="command_context")
 def command_context_fixture(matrix_service: MatrixService, blob_service: BlobService) -> CommandContext:
diff --git a/tests/integration/maintenance_blueprint/conftest.py b/tests/integration/maintenance_blueprint/conftest.py
@@ -18,6 +18,12 @@
 
 from antarest.blobstore.repository import BlobContentRepository
 from antarest.blobstore.service import BlobService
+from antarest.core.config import Config
+from antarest.core.interfaces.eventbus import IEventBus
+from antarest.core.tasks.repository import TaskJobRepository
+from antarest.core.tasks.service import ITaskService, TaskJobService
+from antarest.eventbus.business.local_eventbus import LocalEventBus
+from antarest.eventbus.service import EventBusService
 from tests.conftest_db import db_engine_fixture, db_middleware_fixture
 from tests.matrixstore.conftest import (
     content_repo_fixture,
@@ -45,3 +51,25 @@ def simple_blob_service_fixture(tmp_path: Path) -> BlobService:
     blob_dir.mkdir()
     blob_content_repository = BlobContentRepository(bucket_dir=blob_dir)
     return BlobService(blob_content_repository=blob_content_repository)
+
+
+@pytest.fixture(name="event_bus", scope="session")
+def event_bus_fixture() -> IEventBus:
+    """
+    Fixture that creates a Mock instance of IEventBus with a session-level scope.
+
+    Returns:
+        A Mock instance of the IEventBus class for event bus-related testing.
+    """
+    return EventBusService(LocalEventBus())
+
+
+@pytest.fixture
+def task_service(task_job_repository: TaskJobRepository, event_bus: IEventBus) -> ITaskService:
+    config = Config()
+    return TaskJobService(config=config, repository=task_job_repository, event_bus=event_bus)
+
+
+@pytest.fixture
+def task_job_repository() -> TaskJobRepository:
+    return TaskJobRepository()
diff --git a/tests/integration/maintenance_blueprint/test_gc_tasks.py b/tests/integration/maintenance_blueprint/test_gc_tasks.py
@@ -0,0 +1,75 @@
+# Copyright (c) 2026, RTE (https://www.rte-france.com)
+#
+# See AUTHORS.txt
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This file is part of the Antares project.
+
+"""Integration tests for the tasks cleaning garbage collection task."""
+
+from datetime import datetime
+from unittest.mock import patch
+
+from antarest.core.tasks.model import TaskDTO, TaskJob, TaskListFilter, TaskStatus
+from antarest.core.tasks.repository import TaskJobRepository
+from antarest.core.tasks.service import ITaskService
+from antarest.core.utils.fastapi_sqlalchemy import db
+from antarest.core.utils.lock import create_lock
+from antarest.maintenance.tasks.common import BackGroundTaskStatus, LockId
+from antarest.maintenance.tasks.gc_tasks import clean_tasks
+from tests.helpers import with_admin_user
+
+
+class TestTasksGCIntegration:
+    @with_admin_user
+    def test_clean_tasks_gc(self, task_service: ITaskService, task_job_repository: TaskJobRepository):
+        task_1 = TaskJob(
+            id="1", status=TaskStatus.RUNNING.value, name="task_1", creation_date=datetime(2026, 1, 1, 10, 0, 0)
+        )
+        task_2 = TaskJob(
+            id="2", status=TaskStatus.RUNNING.value, name="task_2", creation_date=datetime(2026, 1, 15, 14, 30, 0)
+        )
+        task_3 = TaskJob(
+            id="3", status=TaskStatus.RUNNING.value, name="task_3", creation_date=datetime(2026, 2, 1, 8, 45, 0)
+        )
+        task_4 = TaskJob(
+            id="4", status=TaskStatus.RUNNING.value, name="task_4", creation_date=datetime(2026, 2, 18, 23, 59, 59)
+        )
+
+        with db():
+            task_job_repository.save(task_1)
+            task_job_repository.save(task_2)
+            task_job_repository.save(task_3)
+            task_job_repository.save(task_4)
+
+            task_list = task_service.list_tasks(TaskListFilter())
+            assert len(task_list) == 4
+
+            # deleting tasks
+            with patch("antarest.core.utils.utils.current_time", return_value=datetime(2026, 2, 20, 0, 0, 0)):
+                task_result = clean_tasks(task_service=task_service, dry_run=False, task_retention_duration=17)
+
+            expected_task = TaskDTO(
+                id="4",
+                status=TaskStatus.RUNNING,
+                name="task_4",
+                creation_date_utc=str(datetime(2026, 2, 18, 23, 59, 59)),
+            )
+            assert task_result.deleted_count == 3
+
+            task_list = task_service.list_tasks(TaskListFilter())
+            assert len(task_list) == 1
+            assert task_list == [expected_task]
+
+    def test_returns_skipped_when_lock_held(self, task_service: ITaskService):
+        with db():
+            with create_lock(db.session, lock_id=LockId.TASKS_GC):
+                result = clean_tasks(task_service=task_service, dry_run=False, task_retention_duration=60)
+
+            assert result.status == BackGroundTaskStatus.SKIPPED
+            assert result.reason == "lock_not_acquired"
diff --git a/tests/maintenance/tasks/test_gc_tasks.py b/tests/maintenance/tasks/test_gc_tasks.py
@@ -0,0 +1,45 @@
+# Copyright (c) 2026, RTE (https://www.rte-france.com)
+#
+# See AUTHORS.txt
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This file is part of the Antares project.
+
+"""Tests for tasks GC task."""
+
+from unittest.mock import Mock
+
+import pytest
+
+from antarest.maintenance.tasks.gc_tasks import clean_tasks
+from antarest.maintenance.tasks.gc_tasks_task import gc_tasks_task
+
+
+class TestCleanTasks:
+    def test_clean_tasks_when_not_dry_run(self):
+        mock_service = Mock()
+        mock_service.delete_task_by_creation_date.return_value = 5
+        gc_tasks_result = clean_tasks(mock_service, dry_run=False, task_retention_duration=60)
+        mock_service.delete_task_by_creation_date.assert_called_once_with(60)
+        assert gc_tasks_result.status == "success"
+        assert gc_tasks_result.deleted_count == 5
+        assert gc_tasks_result.error is None
+
+    def test_does_not_clean_tasks_when_dry_run(self):
+        mock_service = Mock()
+        mock_service.delete_task_by_creation_date.return_value = 5
+        gc_tasks_result = clean_tasks(mock_service, dry_run=True, task_retention_duration=60)
+        mock_service.delete_task_by_creation_date.assert_not_called()
+        assert gc_tasks_result.deleted_count == 0
+        assert gc_tasks_result.error is None
+
+
+class TestTasksGCTask:
+    def test_raises_without_context(self, with_no_maintenance_ctx):
+        with pytest.raises(RuntimeError, match="MaintenanceContext not in app.conf"):
+            gc_tasks_task.run()
diff --git a/tests/maintenance/test_app.py b/tests/maintenance/test_app.py
@@ -103,6 +103,7 @@ def test_uses_config_intervals(self):
         config.storage.auto_archive_sleeping_time = 1800
         config.storage.auto_archive_cron = None
         config.storage.watcher_scan_sleeping_time = 120
+        config.storage.tasks_gc_sleeping_time = 3600
 
         with mock.patch("antarest.maintenance.app.get_config", return_value=config):
             _setup_periodic_tasks(sender=sender)
@@ -112,6 +113,14 @@ def test_uses_config_intervals(self):
         assert calls[1][0][0] == 43200
         assert calls[2][0][0] == 1800
         assert calls[3][0][0] == 120
+        assert calls[5][0][0] == 3600
 
         names = [c[1]["name"] for c in calls]
-        assert names == ["matrices_cleaner", "blobs_cleaner", "auto_archiver", "watcher_scan", "variable_view_cleaner"]
+        assert names == [
+            "matrices_cleaner",
+            "blobs_cleaner",
+            "auto_archiver",
+            "watcher_scan",
+            "variable_view_cleaner",
+            "tasks_cleaner",
+        ]
diff --git a/tests/storage/conftest.py b/tests/storage/conftest.py
@@ -64,6 +64,10 @@ def list_tasks(self, task_filter: TaskListFilter) -> List[TaskDTO]:
     def await_task(self, task_id: str, timeout_sec: Optional[int] = None) -> None:
         pass
 
+    @override
+    def delete_task_by_creation_date(self, task_retention_duration: int) -> int:
+        pass
+
 
 class FileDownloadRepositoryMock(FileDownloadRepository):
     def __init__(self) -> None:
