From 83f72e682c71ec8c89eaccfad119f45b610b8723 Mon Sep 17 00:00:00 2001
From: Guewen Baconnier <guewen.baconnier@qoqa.com>
Date: Fri, 14 Aug 2026 16:21:37 +0200
Subject: [PATCH] [IMP] queue_job: add pause option on channels

Add a new "paused" option on channel configuration.
A paused channel is equivalent to a channel with no capacity:
it yields no job and also blocks all its subchannels.

It still needs a restart of the jobrunner, so it does not allow dynamic
pause/resume, but hot reload may be added later in OCA/queue#765.

It still is an improvement over setting the capacity at 0, because it
keeps the capacity in the configuration (removing the option in the
config restores the initial capacity).

Also, it does not require changing the channel of a job, which is an
issue when resuming paused jobs (they should go back to their former
channel to be resumed with their expected properties related to
capacity/sequential/throttle).
---
 queue_job/jobrunner/channels.py | 43 ++++++++++++++++++++++++++++++---
 1 file changed, 39 insertions(+), 4 deletions(-)

diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py
index c895d9caf3..db0d3e2346 100644
--- a/queue_job/jobrunner/channels.py
+++ b/queue_job/jobrunner/channels.py
@@ -402,9 +402,14 @@ class Channel:
     with a capacity of 1. It is also possible to dedicate a channel with a
     limited capacity for application-autocreated subchannels
     without risking to overflow the system.
+
+    A paused channel does not process any job until it is resumed. All subchannels
+    are blocked with their parent channel.
     """
 
-    def __init__(self, name, parent, capacity=None, sequential=False, throttle=0):
+    def __init__(
+        self, name, parent, capacity=None, sequential=False, throttle=0, paused=False
+    ):
         self.name = name
         self.parent = parent
         if self.parent:
@@ -417,6 +422,7 @@ def __init__(self, name, parent, capacity=None, sequential=False, throttle=0):
         self.capacity = capacity
         self.throttle = throttle  # seconds
         self.sequential = sequential
+        self.paused = paused
 
     @property
     def sequential(self):
@@ -434,11 +440,13 @@ def configure(self, config):
         * capacity
         * sequential
         * throttle
+        * paused
         """
         assert self.fullname.endswith(config["name"])
         self.capacity = config.get("capacity", None)
         self.sequential = bool(config.get("sequential", False))
         self.throttle = int(config.get("throttle", 0))
+        self.paused = int(config.get("paused", False))
         if self.sequential and self.capacity != 1:
             raise ValueError("A sequential channel must have a capacity of 1")
 
@@ -455,12 +463,13 @@ def get_subchannel_by_name(self, subchannel_name):
 
     def __str__(self):
         capacity = "∞" if self.capacity is None else str(self.capacity)
-        return "%s(C:%s,Q:%d,R:%d,F:%d)" % (
+        return "%s(C:%s,Q:%d,R:%d,F:%d%s)" % (
             self.fullname,
             capacity,
             len(self._queue),
             len(self._running),
             len(self._failed),
+            ",paused" if self.paused else "",
         )
 
     def remove(self, job):
@@ -517,6 +526,8 @@ def set_failed(self, job):
             _logger.debug("job %s marked failed in channel %s", job.uuid, self)
 
     def has_capacity(self):
+        if self.paused:
+            return False
         if self.sequential and self._failed:
             # a sequential queue blocks on failed jobs
             return False
@@ -799,6 +810,31 @@ class ChannelManager:
     >>> cm.notify(db, 'S', 'S3', 3, 0, 10, None, 'done')
     >>> pp(list(cm.get_jobs_to_run(now=105)))
     []
+
+    Test pausing a channel
+
+    >>> cm = ChannelManager()
+    >>> cm.simple_configure('root:4,P:2:paused,P.sub:1')
+    >>> cm.notify(db, 'P', 'P1', 1, 0, 10, None, 'pending')
+    >>> cm.notify(db, 'P.sub', 'PS1', 2, 0, 10, None, 'pending')
+
+    Paused channel yields no job
+
+    >>> pp(list(cm.get_jobs_to_run(now=100)))
+    []
+
+    Resuming the channel yields the pending jobs
+
+    >>> cm.simple_configure('root:4,P:2')
+    >>> pp(list(cm.get_jobs_to_run(now=100)))
+    [<ChannelJob P1>, <ChannelJob PS1>]
+
+    Pausing the root channel blocks everything
+
+    >>> cm.simple_configure('root:4:paused')
+    >>> cm.notify(db, 'P', 'P3', 4, 0, 10, None, 'pending')
+    >>> pp(list(cm.get_jobs_to_run(now=106)))
+    []
     """
 
     def __init__(self):
@@ -894,8 +930,7 @@ def parse_simple_config(cls, config_string):
                         )
                     if k in config:
                         raise ValueError(
-                            f"Invalid channel config {config_string}: "
-                            f"duplicate key {k}"
+                            f"Invalid channel config {config_string}: duplicate key {k}"
                         )
                     config[k] = v
             else: