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

@@ -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
+        _notify_task_failure(instance.func, instance.pk, n, instance.result)
-        # This will, in turn, trigger a notification to staff users via the Error post_save signal
 
-        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(
+    A task offloaded with offload_task(..., retry=False) is marked with the
-            'task_failure',
+    'ack_failure' option, which causes the broker to drop it after a single failed
-            scope='worker',
+    attempt (see InvenTree.tasks.offload_task) - its attempt_count will never reach
-            error_name='Task Failure',
+    Q_CLUSTER['max_attempts'], so after_failed_task()'s post_save-based check would
-            error_info=message,
+    otherwise never fire a notification for it. This uses django-q2's post_execute
-            error_data=str(instance.result) if instance.result else '',
+    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')

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,

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
+    Entries are grouped by (taskname, group, force_async, retry, timeout), so that each
-    combination triggered within the batch is flushed via its own bulk_offload_task() call.
+    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',

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):