resolve concurrent job creation race condition

Author: ayushgupta704

api_app/management/commands/fix_split_brain_jobs.py

@@ -0,0 +1,69 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+from django.core.management.base import BaseCommand
+from django.db import connection, transaction
+from django.db.models import Count
+
+from api_app.models import Job
+
+
+class Command(BaseCommand):
+    help = "Fixes Split-Brain scenarios in Jobs by merging duplicate path entries."
+
+    def handle(self, *args, **options):
+        duplicates = Job.objects.values("path").annotate(path_count=Count("id")).filter(path_count__gt=1)
+
+        if not duplicates:
+            self.stdout.write(self.style.SUCCESS("No split-brain jobs found."))
+            return
+
+        self.stdout.write(self.style.WARNING(f"Found {len(duplicates)} duplicate paths."))
+
+        for entry in duplicates:
+            path = entry["path"]
+            jobs = list(Job.objects.filter(path=path).order_by("id"))
+
+            winner = jobs[0]
+            losers = jobs[1:]
+
+            self.stdout.write(f"\nMerging into winner Job PK={winner.pk} (path='{path}')")
+
+            with transaction.atomic():
+                for loser in losers:
+                    self.stdout.write(f"  Processing loser Job PK={loser.pk}...")
+
+                    tags = list(loser.tags.all())
+                    if tags:
+                        winner.tags.add(*tags)
+                        self.stdout.write(f"    Moved {len(tags)} tags")
+
+                    for report_type in [
+                        "analyzerreports",
+                        "connectorreports",
+                        "visualizerreports",
+                        "pivotreports",
+                        "ingestorreports",
+                    ]:
+                        if hasattr(loser, report_type):
+                            reports_count = getattr(loser, report_type).update(job=winner)
+                            if reports_count:
+                                self.stdout.write(f"    Moved {reports_count} {report_type}")
+
+                    try:
+                        comment_count = loser.comments.update(job=winner)
+                        self.stdout.write(f"    Moved {comment_count} comments")
+                    except Exception:
+                        pass
+
+                    if not winner.investigation and loser.investigation:
+                        winner.investigation = loser.investigation
+                        winner.save()
+                        self.stdout.write("    Transferred investigation")
+
+                    loser_id = loser.pk
+                    with connection.cursor() as cursor:
+                        cursor.execute("DELETE FROM api_app_job WHERE id = %s", [loser_id])
+                    self.stdout.write(self.style.SUCCESS(f"  Deleted ghost Job PK={loser_id}"))
+
+        self.stdout.write(self.style.SUCCESS("\nCleanup complete."))

api_app/migrations/0074_job_path_unique.py

@@ -0,0 +1,17 @@
+# Generated by Django 4.2.27 on 2026-04-08 18:50
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api_app', '0073_alter_updatecheckstatus_last_checked_at_and_more'),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name='job',
+            constraint=models.UniqueConstraint(fields=('path',), name='job_path_unique'),
+        ),
+    ]

api_app/models.py

@@ -376,6 +376,12 @@ class Meta:
             # WHERE ("api_app_job"."depth" >= ? AND "api_app_job"."path"::text LIKE ? AND NOT ("api_app_job"."id" = ?))
             models.Index(fields=["depth", "path", "id"], name="MPNodeSearch"),
         ]
+        constraints = [
+            UniqueConstraint(
+                fields=["path"],
+                name="job_path_unique",
+            )
+        ]
 
     # constants
     TLP = TLP

api_app/queryset.py

@@ -266,17 +266,25 @@ def create(self, parent=None, **kwargs):
         Returns:
             The created job.
         """
-        if parent:
-            return parent.add_child(**kwargs)
         # try multiple times hoping to for no race conditions
-        total_attempt_number = 5
+        total_attempt_number = 10
         for attempt in range(0, total_attempt_number):
             try:
+                if parent:
+                    if attempt > 0:
+                        parent.refresh_from_db()
+                    return parent.add_child(**kwargs)
                 return self.model.add_root(**kwargs)
             except IntegrityError:
-                logger.warning(f"Found race condition for {kwargs['name']}. Trying again to calculate path.")
+                logger.warning(
+                    f"Found race condition for {kwargs.get('analyzable', 'unknown')}. Trying again to calculate path."
+                )
                 if attempt == total_attempt_number - 1:
                     raise
+                import time
+
+                # exponential backoff
+                time.sleep(0.01 * (2**attempt))
 
     def delete(self, *args, **kwargs):
         """