From 862c934e0d132c27976360ae5d893d188c0b3e43 Mon Sep 17 00:00:00 2001 From: Oliver Date: Sat, 26 Sep 2026 09:54:10 +0930 Subject: [PATCH] Background Task Features (#12913) * Add option to disable retry for background tasks * Add specific regression testing to ensure the task is only pushed once * Allow specification of "timeout" attribute * Add regression tests * Report failure of single-shot tasks * Address remaining issues * Additional regression tests * Adjust headroom * additional regression testing --- src/backend/InvenTree/InvenTree/models.py | 63 ++- .../InvenTree/InvenTree/setting/worker.py | 3 +- src/backend/InvenTree/InvenTree/tasks.py | 157 +++++- src/backend/InvenTree/InvenTree/test_tasks.py | 513 ++++++++++++++++++ 4 files changed, 710 insertions(+), 26 deletions(-) diff --git a/src/backend/InvenTree/InvenTree/models.py b/src/backend/InvenTree/InvenTree/models.py index d9a1ac0269..9fc9786496 100644 --- a/src/backend/InvenTree/InvenTree/models.py +++ b/src/backend/InvenTree/InvenTree/models.py @@ -20,6 +20,7 @@ from django.utils.translation import gettext_lazy as _ import structlog from django_q.models import Task +from django_q.signals import post_execute from error_report.models import Error from mptt.exceptions import InvalidMove from mptt.models import MPTTModel, TreeForeignKey @@ -1672,32 +1673,68 @@ def notify_staff_users_of_error(instance, label: str, context: dict): logger.error(exc) +def _notify_task_failure(func: str, task_id: str, attempt_count: int, result) -> None: + """Create a new Error object for a permanently-failed background task. + + This will, in turn, trigger a notification to staff users via the Error post_save signal. + """ + from InvenTree.exceptions import log_error + + message = f"Task '{func} ({task_id})' failed after {attempt_count} attempt{'s' if attempt_count != 1 else ''}" + + logger.error(message) + + log_error( + 'task_failure', + scope='worker', + error_name='Task Failure', + error_info=message, + error_data=str(result) if result else '', + ) + + @receiver(post_save, sender=Task, dispatch_uid='failure_post_save_notification') def after_failed_task(sender, instance: Task, created: bool, **kwargs): """Callback when a new task failure log is generated.""" from django.conf import settings - from InvenTree.exceptions import log_error - max_attempts = int(settings.Q_CLUSTER.get('max_attempts', 5)) n = instance.attempt_count # Only notify once the maximum number of attempts has been reached if not instance.success and n >= max_attempts: - # Create a new Error object associated with this failed task - # This will, in turn, trigger a notification to staff users via the Error post_save signal + _notify_task_failure(instance.func, instance.pk, n, instance.result) - message = f"Task '{instance.func} ({instance.pk})' failed after {n} attempts" - logger.error(message) +@receiver(post_execute, dispatch_uid='failure_post_execute_notification') +def after_single_shot_task_failure(sender, task: dict, **kwargs): + """Callback when a background task finishes executing. - log_error( - 'task_failure', - scope='worker', - error_name='Task Failure', - error_info=message, - error_data=str(instance.result) if instance.result else '', - ) + A task offloaded with offload_task(..., retry=False) is marked with the + 'ack_failure' option, which causes the broker to drop it after a single failed + attempt (see InvenTree.tasks.offload_task) - its attempt_count will never reach + Q_CLUSTER['max_attempts'], so after_failed_task()'s post_save-based check would + otherwise never fire a notification for it. This uses django-q2's post_execute + signal instead, which (unlike the saved Task record) still carries the + 'ack_failure' flag, to notify immediately on that task's one and only attempt. + """ + from django.conf import settings + + if task.get('success') or not task.get('ack_failure'): + return + + max_attempts = int(settings.Q_CLUSTER.get('max_attempts', 5)) + + if max_attempts <= 1: + # after_failed_task() already handles this case via its own post_save signal - + # avoid notifying twice + return + + from django_q.utils import get_func_repr + + _notify_task_failure( + get_func_repr(task.get('func')), task.get('id'), 1, task.get('result') + ) @receiver(post_save, sender=Error, dispatch_uid='error_post_save_notification') diff --git a/src/backend/InvenTree/InvenTree/setting/worker.py b/src/backend/InvenTree/InvenTree/setting/worker.py index bef1380bd2..f7baf802d7 100644 --- a/src/backend/InvenTree/InvenTree/setting/worker.py +++ b/src/backend/InvenTree/InvenTree/setting/worker.py @@ -25,7 +25,8 @@ def get_worker_config( get_setting('INVENTREE_BACKGROUND_TIMEOUT', 'background.timeout', 90) ) - # Set the retry time for background workers to be slightly longer than the worker timeout, to ensure that workers have time to timeout before being retried + # Set the retry time for background workers to be slightly longer than the worker timeout, + # to ensure that workers have time to timeout before being retried BACKGROUND_WORKER_RETRY = max( int(get_setting('INVENTREE_BACKGROUND_RETRY', 'background.retry', 300)), BACKGROUND_WORKER_TIMEOUT + 120, diff --git a/src/backend/InvenTree/InvenTree/tasks.py b/src/backend/InvenTree/InvenTree/tasks.py index a0abe45be6..5468bfbb2c 100644 --- a/src/backend/InvenTree/InvenTree/tasks.py +++ b/src/backend/InvenTree/InvenTree/tasks.py @@ -161,15 +161,27 @@ def record_task_success(task_name: str): set_global_setting(f'_{task_name}_SUCCESS', datetime.now().isoformat(), None) -def check_existing_task(taskname, group: str, *args, **kwargs) -> Optional[str]: +def check_existing_task( + taskname, + group: str, + *args, + retry: bool = True, + timeout: Optional[int] = None, + **kwargs, +) -> Optional[str]: """Test if an identical task is already registered with the worker. - This will only return true if the task name, group, args and kwargs all match an existing task. + This will only return true if the task name, group, args, kwargs, retry and timeout + all match an existing task - a queued task with a different retry/timeout policy is + a different request, even if it would otherwise look identical, so it is not treated + as a duplicate. Arguments: taskname: The name of the task to check for, in the format 'app.module.function' group: The group that the task belongs to *args: Positional arguments to match + retry: The 'retry' policy the new call is requesting - see offload_task() + timeout: The per-task 'timeout' override the new call is requesting - see offload_task() **kwargs: Keyword arguments to match Returns: @@ -197,6 +209,16 @@ def check_existing_task(taskname, group: str, *args, **kwargs) -> Optional[str]: # Task kwargs do not match continue + q_options = task.q_options() + + if bool(q_options.get('ack_failure', False)) != (not retry): + # Existing task has a different retry policy - not a true duplicate + continue + + if q_options.get('timeout') != timeout: + # Existing task has a different per-task timeout override - not a true duplicate + continue + task_id = task.task_id() break @@ -204,6 +226,46 @@ def check_existing_task(taskname, group: str, *args, **kwargs) -> Optional[str]: return task_id +def _clamp_task_timeout(timeout: Optional[int]) -> Optional[int]: + """Clamp a per-task 'timeout' override to leave headroom before the cluster's redelivery interval. + + The ORM broker redelivers a queued task once its lock expires, which is governed + by the cluster-wide Q_CLUSTER['retry'] setting - a per-task 'timeout' override has + no effect on that. If 'timeout' left less headroom than that, the task could be + redelivered and run again before the original attempt has even timed out, silently + duplicating work (and defeating retry=False for that task) - so it is clamped down + to the largest value that still leaves 120s of headroom (mirroring the margin + InvenTree.setting.worker.get_worker_config() applies to the cluster-wide timeout). + + Arguments: + timeout: The requested per-task timeout override, if any + + Returns: + Optional[int]: 'timeout', clamped down if necessary + """ + if timeout is None: + return None + + HEADROOM = 30 + + retry = settings.Q_CLUSTER.get('retry') + max_timeout = retry - HEADROOM if retry else timeout + + if retry and timeout > max_timeout: + logger.warning( + 'offload_task(): timeout (%ss) leaves less than %ss of headroom before ' + 'the configured broker retry interval (%ss) - clamping to %ss to avoid the ' + 'task being redelivered and executed again before it can time out', + timeout, + HEADROOM, + retry, + max_timeout, + ) + return max_timeout + + return timeout + + # Context-local batch of pending offload_task() calls (see batch_offload_tasks()) _task_batch: contextvars.ContextVar = contextvars.ContextVar('task_batch', default=None) @@ -211,8 +273,9 @@ _task_batch: contextvars.ContextVar = contextvars.ContextVar('task_batch', defau class TaskBatch: """Collects offload_task() calls made within a batch_offload_tasks() scope. - Entries are grouped by (taskname, group, force_async), so that each distinct - combination triggered within the batch is flushed via its own bulk_offload_task() call. + Entries are grouped by (taskname, group, force_async, retry, timeout), so that each + distinct combination triggered within the batch is flushed via its own + bulk_offload_task() call. """ def __init__(self): @@ -220,18 +283,39 @@ class TaskBatch: self.entries: dict[tuple, list] = defaultdict(list) def add( - self, taskname, group: str, force_async: bool, args: tuple, kwargs: dict + self, + taskname, + group: str, + force_async: bool, + args: tuple, + kwargs: dict, + retry: bool = True, + timeout: Optional[int] = None, ) -> None: """Record a single offload_task() call against this batch.""" - self.entries[taskname, group, force_async].append((args, kwargs)) + self.entries[taskname, group, force_async, retry, timeout].append(( + args, + kwargs, + )) def flush(self) -> None: - """Fire a bulk_offload_task() call for each (taskname, group, force_async) group collected so far.""" + """Fire a bulk_offload_task() call for each (taskname, group, force_async, retry, timeout) group collected so far.""" entries, self.entries = self.entries, defaultdict(list) - for (taskname, group, force_async), task_entries in entries.items(): + for ( + taskname, + group, + force_async, + retry, + timeout, + ), task_entries in entries.items(): bulk_offload_task( - taskname, task_entries, group=group, force_async=force_async + taskname, + task_entries, + group=group, + force_async=force_async, + retry=retry, + timeout=timeout, ) @@ -290,6 +374,8 @@ def offload_task( force_async: bool = False, force_sync: bool = False, check_duplicates: bool = True, + retry: bool = True, + timeout: Optional[int] = None, **kwargs, ) -> str | bool: """Create an AsyncTask if workers are running. This is different to a 'scheduled' task, in that it only runs once! @@ -302,11 +388,29 @@ def offload_task( force_async: If True, force the task to be offloaded (even if workers are not running) force_sync: If True, force the task to be run synchronously (even if workers are running) check_duplicates: If True, check for existing identical tasks before offloading + retry: If False, the task is attempted exactly once and is never retried if it + fails (see note below) + timeout: Optional per-task override (in seconds) of the worker's task timeout. + Clamped down (with a warning) if it would leave less than 30s of headroom + before the configured broker retry interval (settings.Q_CLUSTER['retry']) - + see _clamp_task_timeout() **kwargs: Keyword arguments to be passed to the task function + Note: + django-q2 has no concept of a per-task retry limit: + the ORM broker simply leaves a failed task's queue entry in place, so it + gets redelivered (governed by the cluster-wide 'retry' timeout) until something + acknowledges it, up to the cluster-wide 'max_attempts' limit. The one per-task + escape hatch it does provide is 'ack_failure', which acknowledges (and so + permanently drops) a task the moment it fails, regardless of the cluster's + retry/max_attempts settings. retry=False is implemented on top of that option - + there is no equivalent for a finite positive retry count. + Returns: str | bool: Task ID if the task was offloaded, True if ran synchronously, False otherwise """ + timeout = _clamp_task_timeout(timeout) + # Extract group information from kwargs group = kwargs.pop('group', 'inventree') @@ -314,7 +418,7 @@ def offload_task( # A batch_offload_tasks() context is active - queue this task rather than # offloading it immediately (force_sync=True calls never reach this branch - # see batch_offload_tasks() for why they are excluded from batching) - batch.add(taskname, group, force_async, args, kwargs) + batch.add(taskname, group, force_async, args, kwargs, retry, timeout) return True from InvenTree.exceptions import log_error @@ -345,7 +449,9 @@ def offload_task( if force_async or (is_worker_running() and not force_sync): # Before offloading, check if a duplicate task exists if not force_sync and check_duplicates: - if task_id := check_existing_task(taskname, group, *args, **kwargs): + if task_id := check_existing_task( + taskname, group, *args, retry=retry, timeout=timeout, **kwargs + ): logger.debug( "Skipping duplicate task '%s' with ID '%s'", taskname, task_id ) @@ -354,7 +460,16 @@ def offload_task( # Running as asynchronous task try: - task = AsyncTask(taskname, *args, group=group, **kwargs) + task_kwargs = dict(kwargs) + if not retry: + # Bandaid for django-q2 having no per-task retry limit: 'ack_failure' + # is its one native per-task option that acknowledges (and so drops) + # a task as soon as it fails, rather than leaving it to be redelivered + task_kwargs['ack_failure'] = True + if timeout is not None: + task_kwargs['timeout'] = timeout + + task = AsyncTask(taskname, *args, group=group, **task_kwargs) with tracer.start_as_current_span(f'async worker: {taskname}'): task.run() @@ -419,6 +534,8 @@ def bulk_offload_task( group: str = 'inventree', force_sync: bool = False, force_async: bool = False, + retry: bool = True, + timeout: Optional[int] = None, ) -> bool: """Queue the same background task many times, in a single bulk database write. @@ -436,6 +553,10 @@ def bulk_offload_task( group: The task group to assign to each queued task force_sync: If True, run all tasks synchronously (even if workers are running) force_async: If True, force all tasks to be queued (even if workers are not running) + retry: If False, every queued task is attempted exactly once and is never + retried if it fails - see offload_task() for why + timeout: Optional per-task override (in seconds) of the worker's task timeout + for every queued task - see offload_task() for details Returns: bool: True if the tasks were queued (or run synchronously), False otherwise @@ -443,6 +564,8 @@ def bulk_offload_task( if not entries: return False + timeout = _clamp_task_timeout(timeout) + try: from django_q.brokers import get_broker from django_q.humanhash import uuid @@ -468,6 +591,8 @@ def bulk_offload_task( group=group, force_sync=True, check_duplicates=False, + retry=retry, + timeout=timeout, **kwargs, ) @@ -490,6 +615,14 @@ def bulk_offload_task( 'started': timezone.now(), } + if not retry: + # See offload_task() - 'ack_failure' drops the task the moment it fails, + # instead of leaving its queue entry to be redelivered + task['ack_failure'] = True + + if timeout is not None: + task['timeout'] = timeout + tasks.append( OrmQ( key=broker.list_key or 'inventree', diff --git a/src/backend/InvenTree/InvenTree/test_tasks.py b/src/backend/InvenTree/InvenTree/test_tasks.py index 098b1cc9d0..f63134ec0f 100644 --- a/src/backend/InvenTree/InvenTree/test_tasks.py +++ b/src/backend/InvenTree/InvenTree/test_tasks.py @@ -1,7 +1,11 @@ """Unit tests for task management.""" +import logging import os +import queue +import time from datetime import timedelta +from multiprocessing import Value from unittest.mock import patch from django.conf import settings @@ -59,6 +63,37 @@ def get_result(): return 'abc' +retry_regression_logger = logging.getLogger('InvenTree.test_tasks.retry_regression') + +RETRY_REGRESSION_LOG_MESSAGE = 'retry regression task executed' + + +def always_fails_task(): + """Demo function for the worker retry regression tests below. + + Logs a fixed, greppable message and then always raises - so a test can count exactly + how many times a real django-q2 worker actually invoked it. + """ + retry_regression_logger.info(RETRY_REGRESSION_LOG_MESSAGE) + raise ValueError('always_fails_task: intentional failure for retry regression test') + + +TIMEOUT_TASK_SLEEP_SECONDS = 5 +TIMEOUT_TASK_STARTED_LOG_MESSAGE = 'timeout regression task started' +TIMEOUT_TASK_FINISHED_LOG_MESSAGE = 'timeout regression task finished sleeping' + + +def slow_task_for_timeout_test(): + """Demo function for the timeout regression test below. + + Sleeps far longer than the per-task timeout under test. If a real timeout does not + interrupt it, FINISHED gets logged - so that message must never appear. + """ + retry_regression_logger.info(TIMEOUT_TASK_STARTED_LOG_MESSAGE) + time.sleep(TIMEOUT_TASK_SLEEP_SECONDS) + retry_regression_logger.info(TIMEOUT_TASK_FINISHED_LOG_MESSAGE) + + class InvenTreeTaskTests(PluginRegistryMixin, TestCase): """Unit tests for tasks.""" @@ -137,6 +172,411 @@ class InvenTreeTaskTests(PluginRegistryMixin, TestCase): ): InvenTree.tasks.offload_task('InvenTree.test_tasks.eval', force_sync=True) + def test_force_async_overrides_force_sync(self): + """force_async=True takes priority over force_sync=True - the task is queued, not run inline. + + Regression test: offload_task()'s dispatch condition is + 'force_async or (is_worker_running() and not force_sync)' - force_async short-circuits + the check, so passing both flags together silently queues the task rather than running + it synchronously as force_sync alone would. + """ + OrmQ.objects.all().delete() + + result = InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, force_sync=True + ) + + # A task ID was returned (queued), rather than resolving and running the task inline + # (which would fail, since 'dummy_module' does not exist) + self.assertIsInstance(result, str) + self.assertEqual(OrmQ.objects.count(), 1) + + def test_offload_sync_reraises_exception(self): + """offload_task(..., force_sync=True) must propagate an exception raised by the task. + + Regression test: the synchronous fallback logs the error and re-raises, rather than + swallowing it - nothing previously asserted the exception actually reaches the caller. + """ + + def broken_task(): + raise ValueError('offload_task sync fallback regression test') + + with self.assertRaises(ValueError): + InvenTree.tasks.offload_task(broken_task, force_sync=True) + + def test_offload_no_retry(self): + """retry=False should mark the queued task with ack_failure=True. + + This is the bandaid for django-q2 having no per-task retry limit: 'ack_failure' + is its native per-task option that drops a task the moment it fails, instead of + leaving it to be redelivered indefinitely by the ORM broker. + """ + OrmQ.objects.all().delete() + + InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, retry=False + ) + + task = OrmQ.objects.get() + self.assertTrue(task.q_options().get('ack_failure')) + + # By default (retry=True), the task is not marked for single-shot execution + OrmQ.objects.all().delete() + InvenTree.tasks.offload_task('dummy_module.dummy_function', force_async=True) + task = OrmQ.objects.get() + self.assertFalse(task.q_options().get('ack_failure')) + + def test_offload_timeout(self): + """timeout=N must actually be enforced by the worker, not just recorded on the queue. + + django-q2 supports 'timeout' as a native per-task option. + This offloads a task that sleeps far longer than the timeout, + and drives it through the real worker() pipeline, + to prove it gets killed at the timeout rather than left to run. + """ + from django_q.brokers import get_broker + + OrmQ.objects.all().delete() + + InvenTree.tasks.offload_task( + 'InvenTree.test_tasks.slow_task_for_timeout_test', + force_async=True, + timeout=1, + ) + + # The per-task override must have been recorded on the queued task + queued = OrmQ.objects.get() + self.assertEqual(queued.q_options().get('timeout'), 1) + + # Without an explicit timeout, no per-task override is set - the cluster-wide + # default applies + OrmQ.objects.all().delete() + InvenTree.tasks.offload_task('dummy_module.dummy_function', force_async=True) + queued = OrmQ.objects.get() + self.assertNotIn('timeout', queued.q_options()) + + # Now offload the slow task for real, and drive it through the actual worker + OrmQ.objects.all().delete() + InvenTree.tasks.offload_task( + 'InvenTree.test_tasks.slow_task_for_timeout_test', + force_async=True, + timeout=1, + ) + + broker = get_broker() + + start = time.monotonic() + with self.assertLogs(retry_regression_logger, level='INFO') as captured: + executed = self.run_one_broker_cycle(broker) + elapsed = time.monotonic() - start + + self.assertEqual(executed, 1) + + # The task must have started, but the 1-second timeout must have killed it well + # before its 5-second sleep completes - it never gets to log that it finished + self.assertTrue( + any(TIMEOUT_TASK_STARTED_LOG_MESSAGE in line for line in captured.output) + ) + self.assertFalse( + any(TIMEOUT_TASK_FINISHED_LOG_MESSAGE in line for line in captured.output) + ) + self.assertLess(elapsed, TIMEOUT_TASK_SLEEP_SECONDS) + + # The task must be recorded as failed, specifically due to the timeout + saved_task = Task.objects.get( + func='InvenTree.test_tasks.slow_task_for_timeout_test' + ) + self.assertFalse(saved_task.success) + self.assertIn('exceeded maximum timeout value', saved_task.result) + + def test_offload_no_retry_and_timeout_together(self): + """retry=False and timeout=N together must both land on the same queued task. + + Regression test: retry and timeout are covered independently elsewhere, but nothing + confirmed that a single offload_task() call applying both options actually sets both + 'ack_failure' and 'timeout' on the same queued task, rather than one overriding the other. + """ + OrmQ.objects.all().delete() + + InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, retry=False, timeout=45 + ) + + q_options = OrmQ.objects.get().q_options() + self.assertTrue(q_options.get('ack_failure')) + self.assertEqual(q_options.get('timeout'), 45) + + def test_offload_custom_group(self): + """A custom group= kwarg on a direct (non-batched) offload_task() call must be honored. + + Regression test: custom groups were previously only ever exercised through the + batch_offload_tasks() path (see TaskBatchTests.test_tasks_grouped_by_name_and_group) - + nothing confirmed the direct AsyncTask() dispatch path in offload_task() itself + applies a custom group. + """ + OrmQ.objects.all().delete() + + InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, group='custom_group' + ) + + self.assertEqual(OrmQ.objects.get().group(), 'custom_group') + + def test_bulk_offload_timeout(self): + """bulk_offload_task() should forward timeout=N to every queued task.""" + OrmQ.objects.all().delete() + + entries = [((idx,), {}) for idx in range(5)] + + InvenTree.tasks.bulk_offload_task( + 'dummy_module.dummy_function', entries, force_async=True, timeout=15 + ) + + self.assertEqual(OrmQ.objects.count(), 5) + for task in OrmQ.objects.all(): + self.assertEqual(task.q_options().get('timeout'), 15) + + def test_offload_timeout_validation(self): + """A per-task timeout must be clamped to leave headroom before the broker's redelivery interval. + + The broker redelivers a task once its lock (governed by the cluster-wide + Q_CLUSTER['retry'] setting) expires, regardless of any per-task 'timeout' + override. If 'timeout' left less than 30s of headroom before that, the task + could be redelivered and executed again before the original attempt had even + timed out - so offload_task()/bulk_offload_task() must clamp it down (and warn) + rather than queuing it as requested. + """ + retry = settings.Q_CLUSTER['retry'] + max_timeout = retry - 30 + + # A timeout comfortably below the retry interval is left untouched + OrmQ.objects.all().delete() + InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, timeout=retry - 50 + ) + task = OrmQ.objects.get() + self.assertEqual(task.q_options().get('timeout'), retry - 50) + + # A timeout equal to the retry interval leaves no headroom at all - clamped + # down to the maximum safe value, with a warning logged + OrmQ.objects.all().delete() + with self.assertLogs('inventree', level='WARNING') as captured: + InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, timeout=retry + ) + self.assertTrue(any('clamping' in line for line in captured.output)) + task = OrmQ.objects.get() + self.assertEqual(task.q_options().get('timeout'), max_timeout) + + # Also enforced directly by bulk_offload_task() + OrmQ.objects.all().delete() + with self.assertLogs('inventree', level='WARNING') as captured: + InvenTree.tasks.bulk_offload_task( + 'dummy_module.dummy_function', + [((), {})], + force_async=True, + timeout=retry, + ) + self.assertTrue(any('clamping' in line for line in captured.output)) + task = OrmQ.objects.get() + self.assertEqual(task.q_options().get('timeout'), max_timeout) + + def test_duplicate_check_respects_retry_and_timeout(self): + """check_existing_task() must not treat different retry/timeout policies as duplicates. + + Regression test: previously, queuing the same (taskname, group, args, kwargs) + with a different 'retry' or 'timeout' would be silently swallowed as a + 'duplicate' of whatever was already queued, discarding the newly requested + policy entirely. + """ + OrmQ.objects.all().delete() + + first_id = InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True + ) + + # Same call, but requesting retry=False - must not be treated as a duplicate + second_id = InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, retry=False + ) + self.assertNotEqual(first_id, second_id) + self.assertEqual(OrmQ.objects.count(), 2) + + # Same call again, but with a different timeout - also not a duplicate + third_id = InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, retry=False, timeout=30 + ) + self.assertNotEqual(third_id, second_id) + self.assertEqual(OrmQ.objects.count(), 3) + + # Only an exact match of retry AND timeout is deduplicated + fourth_id = InvenTree.tasks.offload_task( + 'dummy_module.dummy_function', force_async=True, retry=False, timeout=30 + ) + self.assertEqual(fourth_id, third_id) + self.assertEqual(OrmQ.objects.count(), 3) + + def run_one_broker_cycle(self, broker): + """Drive a single dequeue/execute/save-or-acknowledge pass through django-q2. + + Uses the actual pusher/worker/monitor functions - exactly what a real qcluster + worker process does, just without the multiprocessing. + + Returns the number of tasks that were dequeued and executed in this pass. + """ + from django_q.monitor import monitor + from django_q.signing import SignedPackage + from django_q.worker import worker + + dequeued = broker.dequeue() + if not dequeued: + return 0 + + task_queue = queue.Queue() + result_queue = queue.Queue() + + for ack_id, payload in dequeued: + task = SignedPackage.loads(payload) + task['ack_id'] = ack_id + task_queue.put(task) + task_queue.put('STOP') + + # worker()/monitor() normally run in their own dedicated process, so closing + # 'old' django database connections there is harmless. Here they run inline on + # the test's own connection (wrapped in TestCase's atomic transaction), so that + # same call would tear down the connection this test needs afterwards. + with ( + patch('django_q.worker.close_old_django_connections'), + patch('django_q.monitor.close_old_django_connections'), + ): + worker(task_queue, result_queue, Value('i', -1)) + + result_queue.put('STOP') + monitor(result_queue, broker) + + return len(dequeued) + + def test_worker_does_not_retry_when_retry_false(self): + """Regression test: retry=False must stop a real worker from re-running a failing task. + + Rather than just inspecting the queued payload, this drives the task through + django-q2's actual pusher/worker/monitor pipeline (the same functions a real + qcluster worker uses) to prove the task is genuinely never re-executed. + """ + from django_q.brokers import get_broker + + OrmQ.objects.all().delete() + + InvenTree.tasks.offload_task( + 'InvenTree.test_tasks.always_fails_task', force_async=True, retry=False + ) + + broker = get_broker() + + with self.assertLogs(retry_regression_logger, level='INFO') as captured: + executed = self.run_one_broker_cycle(broker) + + self.assertEqual(executed, 1) + self.assertEqual( + sum(RETRY_REGRESSION_LOG_MESSAGE in line for line in captured.output), 1 + ) + + # The failed task must have been dropped, not left queued for redelivery + self.assertEqual(OrmQ.objects.count(), 0) + + # Even simulating the redelivery timeout having elapsed, there is nothing left + # in the broker to redeliver - the task only ever ran once + self.assertEqual(self.run_one_broker_cycle(broker), 0) + + def test_worker_retries_by_default(self): + """Contrast case for test_worker_does_not_retry_when_retry_false(). + + With the default retry=True, a failing task is left queued and gets picked up + and re-executed again once the broker's redelivery timeout has elapsed. + """ + from django_q.brokers import get_broker + + OrmQ.objects.all().delete() + + InvenTree.tasks.offload_task( + 'InvenTree.test_tasks.always_fails_task', force_async=True + ) + + broker = get_broker() + + with self.assertLogs(retry_regression_logger, level='INFO') as captured: + executed = self.run_one_broker_cycle(broker) + + self.assertEqual(executed, 1) + self.assertEqual( + sum(RETRY_REGRESSION_LOG_MESSAGE in line for line in captured.output), 1 + ) + + # The failed task must still be queued, waiting to be redelivered + self.assertEqual(OrmQ.objects.count(), 1) + + # Simulate the redelivery timeout having elapsed, then let the worker pick the + # same task up again + OrmQ.objects.update(lock=timezone.now() - timedelta(seconds=1)) + + with self.assertLogs(retry_regression_logger, level='INFO') as captured: + executed = self.run_one_broker_cycle(broker) + + self.assertEqual(executed, 1) + self.assertEqual( + sum(RETRY_REGRESSION_LOG_MESSAGE in line for line in captured.output), 1 + ) + + def test_single_shot_task_failure_notifies_immediately(self): + """retry=False task failures must raise the 'Task Failure' notification on their one attempt. + + Regression test: after_failed_task() only notifies once a task's attempt_count + reaches Q_CLUSTER['max_attempts'] (5 by default). A retry=False task is dropped + by the broker after a single failed attempt (see test_worker_does_not_retry_when_retry_false), + so attempt_count never reaches that threshold and the notification would + otherwise never fire. InvenTree.models.after_single_shot_task_failure() (hooked + into django-q2's post_execute signal) must notify immediately instead. + """ + from django_q.brokers import get_broker + + OrmQ.objects.all().delete() + Error.objects.all().delete() + + InvenTree.tasks.offload_task( + 'InvenTree.test_tasks.always_fails_task', force_async=True, retry=False + ) + + broker = get_broker() + + with self.assertLogs(retry_regression_logger, level='INFO'): + self.run_one_broker_cycle(broker) + + self.assertTrue(Error.objects.filter(kind='Task Failure').exists()) + + def test_retryable_task_failure_does_not_notify_early(self): + """Contrast case for test_single_shot_task_failure_notifies_immediately(). + + A single failed attempt of a retry=True task must NOT raise the 'Task Failure' + notification - it is still eligible for redelivery, so the notification should + only fire once Q_CLUSTER['max_attempts'] is actually reached (covered by the + existing after_failed_task() post_save logic, not exercised by this test). + """ + from django_q.brokers import get_broker + + OrmQ.objects.all().delete() + Error.objects.all().delete() + + InvenTree.tasks.offload_task( + 'InvenTree.test_tasks.always_fails_task', force_async=True + ) + + broker = get_broker() + + with self.assertLogs(retry_regression_logger, level='INFO'): + self.run_one_broker_cycle(broker) + + self.assertFalse(Error.objects.filter(kind='Task Failure').exists()) + def test_task_heartbeat(self): """Test the task heartbeat.""" InvenTree.tasks.offload_task(InvenTree.tasks.heartbeat) @@ -481,6 +921,45 @@ class InvenTreeTaskTests(PluginRegistryMixin, TestCase): self.assertEqual(task.args(), args) self.assertEqual(task.kwargs(), kwargs) + def test_bulk_offload_no_retry(self): + """bulk_offload_task() should mark every queued task with ack_failure=True when retry=False.""" + OrmQ.objects.all().delete() + + entries = [((idx,), {}) for idx in range(5)] + + InvenTree.tasks.bulk_offload_task( + 'dummy_module.dummy_function', entries, force_async=True, retry=False + ) + + self.assertEqual(OrmQ.objects.count(), 5) + for task in OrmQ.objects.all(): + self.assertTrue(task.q_options().get('ack_failure')) + + def test_bulk_offload_falls_back_to_sync(self): + """bulk_offload_task() falls back to running every entry synchronously when async isn't available. + + Regression test: every other bulk_offload_task() test passes force_async=True, so the + 'not force_async and (force_sync or not is_worker_running())' fallback branch - which + calls offload_task(..., force_sync=True, ...) once per entry - was never exercised. + """ + calls = [] + + def sync_target(value): + calls.append(value) + + OrmQ.objects.all().delete() + + entries = [((idx,), {}) for idx in range(3)] + + result = InvenTree.tasks.bulk_offload_task( + sync_target, entries, force_sync=True + ) + + self.assertTrue(result) + # Every entry ran synchronously, immediately - nothing was queued + self.assertEqual(OrmQ.objects.count(), 0) + self.assertEqual(sorted(calls), [0, 1, 2]) + class TaskBatchTests(TestCase): """Unit tests for the batch_offload_tasks() context manager.""" @@ -546,6 +1025,40 @@ class TaskBatchTests(TestCase): 3, ) + def test_tasks_grouped_by_retry(self): + """Tasks with different retry values are flushed as separate bulk writes.""" + with self.captureOnCommitCallbacks(execute=True): + with transaction.atomic(), InvenTree.tasks.batch_offload_tasks(): + InvenTree.tasks.offload_task( + 'dummy_module.task_a', 1, force_async=True, retry=False + ) + InvenTree.tasks.offload_task('dummy_module.task_a', 2, force_async=True) + + self.assertEqual(OrmQ.objects.count(), 2) + + ack_failure_by_arg = { + task.args()[0]: bool(task.q_options().get('ack_failure')) + for task in OrmQ.objects.all() + } + self.assertEqual(ack_failure_by_arg, {1: True, 2: False}) + + def test_tasks_grouped_by_timeout(self): + """Tasks with different timeout values are flushed as separate bulk writes.""" + with self.captureOnCommitCallbacks(execute=True): + with transaction.atomic(), InvenTree.tasks.batch_offload_tasks(): + InvenTree.tasks.offload_task( + 'dummy_module.task_a', 1, force_async=True, timeout=5 + ) + InvenTree.tasks.offload_task('dummy_module.task_a', 2, force_async=True) + + self.assertEqual(OrmQ.objects.count(), 2) + + timeout_by_arg = { + task.args()[0]: task.q_options().get('timeout') + for task in OrmQ.objects.all() + } + self.assertEqual(timeout_by_arg, {1: 5, 2: None}) + def test_tasks_discarded_on_rollback(self): """Tasks queued in a batch are discarded, not fired, if the transaction rolls back.""" with self.captureOnCommitCallbacks(execute=True):