fix(parsers): detect stale jobs by run age

src/apps/core/services.py

@@ -693,9 +693,15 @@ class BackgroundJobService(BaseReadOnlyService):
         from apps.core.models import JobStatus
 
         cutoff = timezone.now() - timedelta(minutes=max_age_minutes)
-        queryset = cls.get_queryset().filter(
+        queryset = (
+            cls.get_queryset()
+            .filter(
                 status__in=[JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY],
-            updated_at__lt=cutoff,
+            )
+            .filter(
+                Q(started_at__isnull=False, started_at__lt=cutoff)
+                | Q(started_at__isnull=True, created_at__lt=cutoff)
+            )
         )
         if task_names:
             queryset = queryset.filter(task_name__in=task_names)

src/apps/parsers/frontend_compat.py

@@ -323,10 +323,16 @@ def _active_tasks_for_definition(
         for source_key in definition.source_keys
         if source_key in PARSER_SOURCES
     ]
-    queryset = BackgroundJobService.get_queryset().filter(
+    queryset = (
+        BackgroundJobService.get_queryset()
+        .filter(
             task_name__in=task_names,
             status__in=ACTIVE_JOB_STATUSES,
-        updated_at__gte=_stale_cutoff(),
+        )
+        .filter(
+            Q(started_at__isnull=False, started_at__gte=_stale_cutoff())
+            | Q(started_at__isnull=True, created_at__gte=_stale_cutoff())
+        )
     )
     return [_serialize_active_job(job) for job in queryset.order_by("-created_at")[:10]]
 

src/apps/parsers/services.py

@@ -416,33 +416,24 @@ class ParserLoadLogService(BaseService[ParserLoadLog]):
         updated = 0
         active_statuses = [JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY]
         for log in stale_logs:
-            batch_job = (
-                BackgroundJob.objects.filter(
+            active_jobs = BackgroundJob.objects.filter(
                 status__in=active_statuses,
                 meta__source=log.source,
-                    meta__batch_id=log.batch_id,
+            ).filter(Q(meta__batch_id=log.batch_id) | Q(meta__batch_id__isnull=True))
+            fresh_jobs = active_jobs.filter(
+                Q(started_at__isnull=False, started_at__gte=cutoff)
+                | Q(started_at__isnull=True, created_at__gte=cutoff)
             )
-                .order_by("-updated_at")
-                .first()
-            )
-            source_job = None
-            if batch_job is None:
-                source_job = (
-                    BackgroundJob.objects.filter(
-                        status__in=active_statuses,
-                        meta__source=log.source,
-                        meta__batch_id__isnull=True,
-                    )
-                    .order_by("-updated_at")
-                    .first()
-                )
-            job = batch_job or source_job
-            if job is not None and job.updated_at >= cutoff:
+            if fresh_jobs.exists():
                 continue
 
             cls.mark_failed(log, stale_message)
             updated += 1
-            if job is not None:
+            stale_jobs = active_jobs.filter(
+                Q(started_at__isnull=False, started_at__lt=cutoff)
+                | Q(started_at__isnull=True, created_at__lt=cutoff)
+            )
+            for job in stale_jobs.order_by("created_at"):
                 job.fail(error=stale_message)
 
         return updated

src/apps/parsers/source_cards.py

@@ -21,7 +21,7 @@ from apps.parsers.models import (
     ProcurementRecord,
 )
 from django.conf import settings
-from django.db.models import Max
+from django.db.models import Max, Q
 from django.http import Http404
 from django.utils import timezone
 from rest_framework.exceptions import ValidationError
@@ -727,10 +727,16 @@ class SourceCardService:
         cls, definition: SourceCardDefinition
     ) -> list[dict[str, Any]]:
         cutoff = cls._stale_cutoff()
-        queryset = BackgroundJobService.get_queryset().filter(
+        queryset = (
+            BackgroundJobService.get_queryset()
+            .filter(
                 task_name__in=definition.task_names,
                 status__in=ACTIVE_JOB_STATUSES,
-            updated_at__gte=cutoff,
+            )
+            .filter(
+                Q(started_at__isnull=False, started_at__gte=cutoff)
+                | Q(started_at__isnull=True, created_at__gte=cutoff)
+            )
         )
         return [
             cls._serialize_job(job) for job in queryset.order_by("-created_at")[:10]

tests/apps/core/test_background_jobs.py

@@ -291,7 +291,7 @@ class BackgroundJobServiceTest(TestCase):
         old_timestamp = timezone.now() - timedelta(hours=3)
         BackgroundJob.objects.filter(
             task_id__in=[stale_job.task_id, unrelated_job.task_id]
-        ).update(updated_at=old_timestamp)
+        ).update(created_at=old_timestamp, updated_at=timezone.now())
 
         updated = BackgroundJobService.mark_stale_active_jobs_failed(
             max_age_minutes=90,

tests/apps/parsers/test_services.py

@@ -426,7 +426,10 @@ class ParserLoadLogServiceTest(TestCase):
         )
         old_timestamp = timezone.now() - timedelta(hours=3)
         ParserLoadLog.objects.filter(pk=log.pk).update(updated_at=old_timestamp)
-        BackgroundJob.objects.filter(pk=job.pk).update(updated_at=old_timestamp)
+        BackgroundJob.objects.filter(pk=job.pk).update(
+            created_at=old_timestamp,
+            updated_at=timezone.now(),
+        )
 
         updated = ParserLoadLogService.mark_stale_in_progress_failed(max_age_minutes=90)
 

tests/apps/parsers/test_source_cards_service.py

@@ -4,13 +4,14 @@ from datetime import timedelta
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+from apps.core.models import BackgroundJob, JobStatus
 from apps.parsers.source_cards import (
     SourceCardDefinition,
     SourceCardService,
     SourceItemDefinition,
 )
 from django.http import Http404
-from django.test import SimpleTestCase
+from django.test import SimpleTestCase, TestCase, override_settings
 from django.utils import timezone
 from rest_framework.exceptions import ValidationError
 
@@ -291,3 +292,43 @@ class SourceCardServiceUnitTest(SimpleTestCase):
             ),
             "idle",
         )
+
+
+@override_settings(PARSER_STALE_LOAD_MAX_AGE_MINUTES=90)
+class SourceCardServiceDatabaseTest(TestCase):
+    def test_get_active_tasks_ignores_old_jobs_even_when_updated_recently(self):
+        job = BackgroundJob.objects.create(
+            task_id="old-source-task",
+            task_name="apps.parsers.tasks.parse_industrial_products",
+            status=JobStatus.STARTED,
+            progress=10,
+            meta={"source": "industrial_products"},
+        )
+        old_timestamp = timezone.now() - timedelta(hours=3)
+        BackgroundJob.objects.filter(pk=job.pk).update(
+            created_at=old_timestamp,
+            started_at=old_timestamp,
+            updated_at=timezone.now(),
+        )
+
+        tasks = SourceCardService._get_active_tasks(
+            SourceCardService.get_definition("manufacturers-and-products")
+        )
+
+        self.assertEqual(tasks, [])
+
+    def test_get_active_tasks_keeps_recent_pending_jobs(self):
+        BackgroundJob.objects.create(
+            task_id="fresh-source-task",
+            task_name="apps.parsers.tasks.parse_industrial_products",
+            status=JobStatus.PENDING,
+            progress=0,
+            meta={"source": "industrial_products"},
+        )
+
+        tasks = SourceCardService._get_active_tasks(
+            SourceCardService.get_definition("manufacturers-and-products")
+        )
+
+        self.assertEqual(len(tasks), 1)
+        self.assertEqual(tasks[0]["task_id"], "fresh-source-task")