Concurrency fix for scheduled tasks (#12493)

- Prevent multiple instances of scheduled tasks from plugins

src/backend/InvenTree/plugin/base/integration/ScheduleMixin.py

@@ -161,50 +161,64 @@ class ScheduleMixin:
         self.validate_scheduled_tasks()
 
         try:
+            from django.db import transaction
+
             from django_q.models import Schedule
 
-            for key, task in self.scheduled_tasks.items():
-                task_name = self.get_task_name(key)
+            from plugin.models import PluginConfig
 
-                obj = {
-                    'name': task_name,
-                    'schedule_type': task['schedule'],
-                    'minutes': task.get('minutes', None),
-                    'repeats': task.get('repeats', -1),
-                }
+            with transaction.atomic():
+                # Lock this plugin's PluginConfig row for the duration of task
+                # registration. django_q's Schedule.name has no DB-level unique
+                # constraint, so without this lock, concurrent activation (e.g.
+                # multiple worker processes starting up at the same time) can
+                # both pass the 'does this task already exist' check below and
+                # each create a duplicate Schedule row for the same task name.
+                if config := self.plugin_config():
+                    PluginConfig.objects.select_for_update().get(pk=config.pk)
 
-                func_name = task['func'].strip()
+                for key, task in self.scheduled_tasks.items():
+                    task_name = self.get_task_name(key)
 
-                if '.' in func_name:
-                    """Dotted notation indicates that we wish to run a globally defined function, from a specified Python module."""
-                    obj['func'] = func_name
-                else:
-                    """Non-dotted notation indicates that we wish to call a 'member function' of the calling plugin. This is managed by the plugin registry itself."""
-                    slug = self.plugin_slug()
-                    obj['func'] = 'plugin.registry.call_plugin_function'
-                    obj['args'] = f"'{slug}', '{func_name}'"
+                    obj = {
+                        'name': task_name,
+                        'schedule_type': task['schedule'],
+                        'minutes': task.get('minutes', None),
+                        'repeats': task.get('repeats', -1),
+                    }
 
-                tasks = Schedule.objects.filter(name=task_name)
+                    func_name = task['func'].strip()
 
-                if len(tasks) > 1:
-                    logger.info(
-                        "Found multiple tasks; Adding a new scheduled task '%s'",
-                        task_name,
-                    )
-                    tasks.delete()
-                    Schedule.objects.create(**obj)
-                elif len(tasks) == 1:
-                    # Scheduled task already exists - update it!
-                    logger.info("Updating scheduled task '%s'", task_name)
+                    if '.' in func_name:
+                        """Dotted notation indicates that we wish to run a globally defined function, from a specified Python module."""
+                        obj['func'] = func_name
+                    else:
+                        """Non-dotted notation indicates that we wish to call a 'member function' of the calling plugin. This is managed by the plugin registry itself."""
+                        slug = self.plugin_slug()
+                        obj['func'] = 'plugin.registry.call_plugin_function'
+                        obj['args'] = f"'{slug}', '{func_name}'"
 
-                    if instance := tasks.first():
-                        for item in obj:
-                            setattr(instance, item, obj[item])
-                        instance.save()
-                else:
-                    logger.info("Adding scheduled task '%s'", task_name)
-                    # Create a new scheduled task
-                    Schedule.objects.create(**obj)
+                    tasks = Schedule.objects.filter(name=task_name)
+
+                    if len(tasks) > 1:
+                        logger.info(
+                            "Found multiple tasks; Adding a new scheduled task '%s'",
+                            task_name,
+                        )
+                        tasks.delete()
+                        Schedule.objects.create(**obj)
+                    elif len(tasks) == 1:
+                        # Scheduled task already exists - update it!
+                        logger.info("Updating scheduled task '%s'", task_name)
+
+                        if instance := tasks.first():
+                            for item in obj:
+                                setattr(instance, item, obj[item])
+                            instance.save()
+                    else:
+                        logger.info("Adding scheduled task '%s'", task_name)
+                        # Create a new scheduled task
+                        Schedule.objects.create(**obj)
 
         except (ProgrammingError, OperationalError):  # pragma: no cover
             # Database might not yet be ready

src/backend/InvenTree/plugin/samples/integration/test_scheduled_task.py

@@ -1,6 +1,11 @@
 """Unit tests for scheduled tasks."""
 
-from django.test import TestCase
+import threading
+from unittest import mock
+
+from django.db import connection
+from django.db.models.query import QuerySet
+from django.test import TestCase, TransactionTestCase
 
 from plugin import InvenTreePlugin
 from plugin.helpers import MixinImplementationError
@@ -92,6 +97,75 @@ class ExampleScheduledTaskPluginTests(TestCase):
             call_plugin_function('does_not_exist', 'member_func'), None
 
 
+class ScheduleMixinConcurrencyTest(TransactionTestCase):
+    """Genuine cross-transaction regression test for ScheduleMixin.register_tasks().
+
+    django_q's Schedule.name has no DB-level unique constraint, so two concurrent activation passes (e.g. multiple
+    worker processes starting up together) could both find zero matching Schedule
+    rows for a task and both create one - leaving a duplicate scheduled task that
+    then fires twice per interval.
+
+    register_tasks() now locks the plugin's PluginConfig row (select_for_update)
+    for the duration of task registration, so only one of two concurrent calls may
+    proceed through the check-then-write at a time.
+    """
+
+    def test_concurrent_register_tasks_does_not_duplicate(self):
+        """Two concurrent register_tasks() calls for the same plugin must not create duplicate Schedule rows."""
+        from django_q.models import Schedule
+
+        plg = registry.plugins['schedule']
+        self.assertTrue(plg)
+
+        # Warm the plugin-config cache in the main thread first, so the race
+        # window below only covers register_tasks() itself, not the (variable
+        # latency) first-time cache population that plugin_config() may trigger
+        self.assertIsNotNone(plg.plugin_config())
+
+        # Start from a clean slate
+        Schedule.objects.filter(name__istartswith='plugin.schedule.').delete()
+
+        start_barrier = threading.Barrier(2, timeout=15)
+        errors = []
+
+        # Wrap select_for_update() so both threads reach the (real, database-level)
+        # PluginConfig row lock at the same time - one wins the lock and proceeds
+        # through its full check-then-write, the other blocks until the winner's
+        # transaction completes.
+        original_select_for_update = QuerySet.select_for_update
+
+        def synced_select_for_update(self_qs, *args, **kwargs):
+            start_barrier.wait(timeout=15)
+            return original_select_for_update(self_qs, *args, **kwargs)
+
+        def run():
+            try:
+                plg.register_tasks()
+            except Exception as exc:  # pragma: no cover - surfaced via errors list
+                errors.append(exc)
+            finally:
+                connection.close()
+
+        thread_a = threading.Thread(target=run)
+        thread_b = threading.Thread(target=run)
+
+        with mock.patch.object(QuerySet, 'select_for_update', synced_select_for_update):
+            thread_a.start()
+            thread_b.start()
+            thread_a.join(timeout=10)
+            thread_b.join(timeout=10)
+
+        self.assertEqual(errors, [])
+
+        # Exactly one Schedule row per task, not two
+        for task_name in plg.get_task_names():
+            self.assertEqual(
+                Schedule.objects.filter(name=task_name).count(),
+                1,
+                f'Duplicate Schedule row created for {task_name}',
+            )
+
+
 class ScheduledTaskPluginTests(TestCase):
     """Tests for ScheduledTaskPluginTests mixin base."""
 

src/backend/_tests_report_129455.txt

@@ -0,0 +1,11 @@
+test_describe_filters_covers_attachment_and_parameter_resources (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_covers_attachment_and_parameter_resources),0.260980
+test_describe_filters_covers_bom_resources (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_covers_bom_resources),0.014891
+test_describe_filters_covers_company_catalog_resources (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_covers_company_catalog_resources),0.014049
+test_describe_filters_covers_filterset_fields_shorthand (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_covers_filterset_fields_shorthand),0.011792
+test_describe_filters_covers_order_and_build_resources (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_covers_order_and_build_resources),0.013518
+test_describe_filters_covers_project_code (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_covers_project_code),0.013685
+test_describe_filters_covers_return_order_resources (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_covers_return_order_resources),0.012927
+test_describe_filters_covers_stock_history_resources (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_covers_stock_history_resources),0.011041
+test_describe_filters_falls_back_to_default_ordering_fields (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_falls_back_to_default_ordering_fields),0.012815
+test_describe_filters_reflects_real_filterset (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_reflects_real_filterset),0.012149
+test_describe_filters_rejects_unknown_resource (inventree_mcp.test_mcp.DescribeFiltersTest.test_describe_filters_rejects_unknown_resource),0.012105