diff --git a/src/backend/InvenTree/plugin/base/integration/ScheduleMixin.py b/src/backend/InvenTree/plugin/base/integration/ScheduleMixin.py index 140b0c8c84..624784f776 100644 --- a/src/backend/InvenTree/plugin/base/integration/ScheduleMixin.py +++ b/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 diff --git a/src/backend/InvenTree/plugin/samples/integration/test_scheduled_task.py b/src/backend/InvenTree/plugin/samples/integration/test_scheduled_task.py index edf428f0f1..bacedb1c99 100644 --- a/src/backend/InvenTree/plugin/samples/integration/test_scheduled_task.py +++ b/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.""" diff --git a/src/backend/_tests_report_129455.txt b/src/backend/_tests_report_129455.txt new file mode 100644 index 0000000000..5035ebed75 --- /dev/null +++ b/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