chore(huey): remove CancelExecution patch

Pipfile

@@ -26,8 +26,9 @@ yt-dlp = {extras = ["default", "curl-cffi"], version = "*"}
 emoji = "*"
 brotli = "*"
 html5lib = "*"
-bgutil-ytdlp-pot-provider = "==2.0.0"
 yt-dlp-remote-cipher = "*"
 babi = "*"
+bgutil-ytdlp-pot-provider = "==2.0.0"
 hat-syslog = "==0.7.28"
 hat-juggler = "!=0.7.4"
+huey = ">=3.5.0"

tubesync/common/huey.py

@@ -6,8 +6,9 @@
 import uuid
 from functools import partial
 from huey import (
-    CancelExecution, Huey as huey_Huey,
-    signals, utils,
+    Huey as huey_Huey,
+    signals,
+    utils,
 )
 from huey.api import Task, TaskLock
 from huey.exceptions import TaskLockedException
@@ -27,7 +28,9 @@
         return False
     else:
         return True
-#TaskLock._set_acquired = _set_acquired
+
+
+# TaskLock._set_acquired = _set_acquired
 TaskLock.acquired = property(
     TaskLock.is_locked,
     _set_acquired,
@@ -67,10 +70,10 @@
         try:
             current = int(current[0][0])
             expected = int(expected)
-        except (TypeError, ValueError,):
+        except (TypeError, ValueError):
             pass
         else:
-            vacuum_db = (current != expected)
+            vacuum_db = (current != expected)  # fmt: skip
         with self.db(close=True) as curs:
             curs.execute('PRAGMA wal_checkpoint(TRUNCATE)')
             if vacuum_db:
@@ -81,7 +84,7 @@
 class Huey(huey_Huey):
     # do not use __len__ (pending_count) for bool
     def __bool__(self):
-        return (self._registry._registry or self._registry._periodic_tasks)
+        return (self._registry._registry or self._registry._periodic_tasks)  # fmt: skip
 
     def _emit(self, signal, task, *args, **kwargs):
         kwargs['huey'] = self
@@ -95,512 +98,521 @@
 
     def reschedule(self, task_id, eta):
         pc = self.pending_count()
         found = [
             t
             for t in self.pending(100)
             if t.id == task_id
         ]
         if not found and pc > 100:
             found = [
                 t
                 for t in self.pending()
                 if t.id == task_id
             ]
         if not found:
             found = [
                 t
                 for t in self.scheduled()
                 if t.id == task_id
             ]
         if not found:
             return False
         for t in found:
             self.revoke(t, revoke_once=True)
         task_eta = utils.normalize_time(eta=eta, utc=self.utc)
         for t in found:
             previous_id = t.id
             t.eta = task_eta
             t.id = t.create_id()
             t.revoke_id = f'r:{t.id}'
             try:
                 from common.models import TaskHistory
 
                 th = TaskHistory.objects.get(task_id=previous_id)
             # ruff: ignore[S110]
             except:
                 pass
             else:
                 # clone the old task history
                 th.pk = None
                 th._state.adding = True
                 th.task_id = t.id
                 th.save()
             self.enqueue(t)
         return True
 
     def scheduled_at_from_task(self, task_obj, /):
         scheduled_at = None
         if not (hasattr(task_obj, 'eta') and task_obj.eta):
             return None
         if self.utc:
             scheduled_at = task_obj.eta.replace(tzinfo=datetime.timezone.utc)
-        else: # this path is unlikely
+        else:  # this path is unlikely
             scheduled_at = timestamp_to_datetime(
                 datetime_to_timestamp(task_obj.eta, integer=False),
             ).astimezone(tz=datetime.timezone.utc)
         return scheduled_at
 
 
 class SqliteHuey(Huey):
     storage_class = SqliteStorage
 
 
-def CancelExecution_init(self, *args, retry=None, **kwargs):
-    self.retry = retry
-    super(CancelExecution, self).__init__(*args, **kwargs)
-
-
-CancelExecution.__init__ = CancelExecution_init
-
-
 def delay_to_eta(delay, /):
     return utils.normalize_time(delay=delay)
 
 
 def h_q_dict(q, /):
     return dict(
-        scheduled=(q.scheduled_count(), q.scheduled(),),
-        pending=(q.pending_count(), q.pending(),),
-        results=(q.result_count(), list(q.all_results().keys()),),
+        scheduled=(q.scheduled_count(), q.scheduled()),
+        pending=(q.pending_count(), q.pending()),
+        results=(q.result_count(), list(q.all_results().keys())),
     )
 
 
 def h_q_tuple(q, /):
     if isinstance(q, str):
         from django_huey import get_queue
 
         q = get_queue(q)
     return (
         q.name,
         list(q._registry._registry.keys()),
         h_q_dict(q),
     )
 
 
 def start_consumer(queue_name):
     assert isinstance(queue_name, str), type(queue_name)
     svc_name = queue_name.replace('_', '-')
     svc_dir = Path('/run') / 'service' / svc_name
 
     try:
         if not all((
             svc_dir.is_dir(),
             (svc_dir / 'supervise').is_dir(),
             (svc_dir / 'supervise' / 'control').is_fifo(),
         )):
             return
     except PermissionError:
         pass
     else:
         subprocess.Popen(
-            [ '/command/s6-svc', '-U', str(svc_dir) ],
-            stdout=-1, stderr=-1, start_new_session=True,
+            ['/command/s6-svc', '-U', str(svc_dir)],
+            stdout=-1, stderr=-1, start_new_session=True
         )
 
 
 def h_q_reset_maint_func(queue, /, exception=None, status=None):
-        if status is None:
-            return
-        elif 'started' == status:
-            start_consumer(queue.name)
-        elif 'exception' == status and exception is not None:
-            # log, but do not raise an exception
-            from huey.api import logger
-            logger.error(
-                f'{queue.name}: maintenance function exception: {exception}'
-            )
-            return
-        return True
+    if status is None:
+        return
+    elif 'started' == status:
+        start_consumer(queue.name)
+    elif 'exception' == status and exception is not None:
+        # log, but do not raise an exception
+        from huey.api import logger
+        logger.error(
+            f'{queue.name}: maintenance function exception: {exception}'
+        )
+        return
+    return True
+
 
 # Configuration convenience helpers
 
 
 def h_q_reset_tasks(q, /, *, maint_func=None):
     if isinstance(q, str):
         from django_huey import get_queue
 
         q = get_queue(q)
     # revoke to prevent pending tasks from executing
     for t in q._registry._registry.values():
         q.revoke_all(t, revoke_until=delay_to_eta(600))
     # clear scheduled tasks
     q.storage.flush_schedule()
     # clear pending tasks
     q.storage.flush_queue()
 
     # run the maintenance function
     def default_maint_func(queue, /, exception=None, status=None):
         if status is None:
             return
         if 'exception' == status and exception is not None:
             # log, but do not raise an exception
             from huey.api import logger
             logger.error(
                 f'{queue.name}: maintenance function exception: {exception}'
             )
             return
         return True
 
     maint_result = None
     if maint_func is None:
         maint_func = default_maint_func
     if maint_func and callable(maint_func):
         try:
             maint_result = maint_func(q, status='started')
         # ruff: ignore[BLE001]
         except Exception as exc:
             maint_result = maint_func(q, exception=exc, status='exception')
         finally:
             maint_func(q, status='finished')
     # clear everything now that we are done
     q.storage.flush_all()
     q.flush()
     # return the results from the maintenance function
     return maint_result
 
 
 def sqlite_tasks(key, /, prefix=None, thread=None, workers=None, *, tasks_dir=None):
     name_fmt = 'huey_{}'
     if prefix:
         name_fmt = f'huey_{prefix}_' + '{}'
     name = name_fmt.format(key)
     thread = thread is True
     try:
         workers = int(workers)
     except TypeError:
         workers = 2
     finally:
         if 0 >= workers:
             workers = max(2, get_usable_cpu_count() // 2)
         elif 1 == workers:
             thread = False
     return dict(
         huey_class='common.huey.SqliteHuey',
         name=name,
         immediate=False,
         results=True,
         store_none=False,
         utc=True,
         compression=True,
         connection=dict(
             filename=str(Path(tasks_dir or '/config/tasks') / f'{name}.db'),
             fsync=True,
-            isolation_level='IMMEDIATE', # _create_connection sets this to None
+            isolation_level='IMMEDIATE',  # _create_connection sets this to None
             strict_fifo=True,
             timeout=60,
         ),
         consumer=dict(
             workers=workers if thread else 1,
             worker_type='thread' if thread else 'process',
-            max_delay=20.0,
+            max_delay=29.0,
             max_tasks=10_000,
             check_worker_health=True,
             health_check_interval=30.0,
             flush_locks=True,
             scheduler_interval=10,
             simple_log=False,
             # verbose has three positions:
             # DEBUG: True
             # INFO: None
             # WARNING: False
             verbose=False,
         ),
     )
 
 
 # Signal handlers shared between queues
 
 def on_executing_remove_duplicates(signal_name, task_obj, exception_obj=None, /, *, huey=None):
     if signals.SIGNAL_EXECUTING != signal_name:
         return
     assert exception_obj is None
     assert huey is not None
     assert hasattr(huey, 'pending') and callable(huey.pending)
     assert hasattr(huey, 'revoke_by_id') and callable(huey.revoke_by_id)
     assert hasattr(huey, 'scheduled') and callable(huey.scheduled)
     assert signals.SIGNAL_EXECUTING == signal_name
 
     from common.models import TaskHistory
 
     try:
         th = TaskHistory.objects.get(task_id=str(task_obj.id))
     except TaskHistory.DoesNotExist:
         return
     else:
         if not th.remove_duplicates:
             return
 
     def task_generator(queue):
         for t in queue.pending():
             yield t
         for t in queue.scheduled():
             yield t
 
     def waiting_id_generator(queue, task_obj):
         b = task_obj
         for a in task_generator(queue):
             matches = (
                 a.name == b.name and
                 a.id != b.id and
                 # revoke lower priority
                 a.priority <= b.priority and
                 # revoke tasks that have executed fewer times
                 a.retries >= b.retries and
                 a.data == b.data
             )
             if matches:
                 yield a.id
 
     for task_id in waiting_id_generator(queue=huey, task_obj=task_obj):
         huey.revoke_by_id(task_id, revoke_once=True)
 
 
 def on_interrupted(signal_name, task_obj, exception_obj=None, /, *, huey=None):
     if signals.SIGNAL_INTERRUPTED != signal_name:
         return
     assert exception_obj is None
     assert huey is not None
     assert hasattr(huey, 'enqueue') and callable(huey.enqueue)
     huey.enqueue(task_obj)
 
 
 storage_key_prefix = 'task_history:'
 
 
 def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None):
     signal_time = time.monotonic()
     signal_dt = datetime.datetime.now(datetime.timezone.utc)
     assert huey is not None
     assert hasattr(huey, 'get') and callable(huey.get)
     assert hasattr(huey, 'put') and callable(huey.put)
 
     from common.models import TaskHistory
     add_to_elapsed_signals = frozenset((
         signals.SIGNAL_TIMEOUT,
         signals.SIGNAL_INTERRUPTED,
         signals.SIGNAL_ERROR,
         signals.SIGNAL_CANCELED,
         signals.SIGNAL_COMPLETE,
-    ))
+    ))  # fmt: skip
     recorded_signals = frozenset((
         signals.SIGNAL_REVOKED,
         signals.SIGNAL_EXPIRED,
         signals.SIGNAL_LOCKED,
         signals.SIGNAL_EXECUTING,
         signals.SIGNAL_RETRYING,
         signals.SIGNAL_RATE_LIMITED,
-    )) | add_to_elapsed_signals
+    )) | add_to_elapsed_signals  # fmt: skip
     storage_key = f'{storage_key_prefix}{task_obj.id}'
     task_obj_attr = '_signals_history'
 
     history = getattr(task_obj, task_obj_attr, None)
     if history is None:
         # pull it from storage, or initialize it
         history = huey.get(
             key=storage_key,
             peek=True,
         ) or dict(
             created=signal_dt,
-            data=(task_obj.args, repr(task_obj.kwargs),),
+            data=(task_obj.args, repr(task_obj.kwargs)),
             elapsed=0,
             module=task_obj.__module__,
             name=task_obj.name,
         )
         setattr(task_obj, task_obj_attr, history)
     assert history is not None
     history['modified'] = signal_dt
 
     if signal_name in recorded_signals:
         history[signal_name] = signal_time
     if signal_name in add_to_elapsed_signals and signals.SIGNAL_EXECUTING in history:
         history['elapsed'] += signal_time - history[signals.SIGNAL_EXECUTING]
     if signals.SIGNAL_COMPLETE in history:
         huey.get(key=storage_key)
     else:
         huey.put(key=storage_key, data=history)
     # created never used
     # ruff: ignore[RUF059]
     th, created = retry_django_db(5)(TaskHistory.objects.get_or_create)(
         task_id=str(task_obj.id),
-        name=f"{task_obj.__module__}.{task_obj.name}",
+        name=f'{task_obj.__module__}.{task_obj.name}',
         queue=huey.name,
     )
     th.priority = task_obj.priority
     th.task_params = list((
         list(task_obj.args),
         repr(task_obj.kwargs),
     ))
     if signal_name == signals.SIGNAL_EXECUTING:
         th.attempts += 1
         th.start_at = signal_dt
     elif exception_obj is not None:
         th.failed_at = signal_dt
         th.last_error = str(exception_obj)
     elif signal_name == signals.SIGNAL_ENQUEUED:
         scheduled_at = huey.scheduled_at_from_task(task_obj)
         if scheduled_at:
             th.scheduled_at = scheduled_at
         if not th.verbose_name:
             try:
                 from django.core.exceptions import ValidationError
                 from sync.models import Media, Source
             # ruff: ignore[S110]
             except:
                 pass
             else:
                 try:
                     key = task_obj.args[0]
                     instance_uuid_str = str(key)
                     instance_uuid = uuid.UUID(instance_uuid_str)
                 except IndexError:
                     key = None
                 except RuntimeError:
                     instance_uuid_str = None
                 except ValueError:
                     instance_uuid = None
                 else:
-                    for model in (Source, Media,):
+                    for model in (Source, Media):
                         try:
                             model_instance = model.objects.get(pk=instance_uuid)
-                        except (model.DoesNotExist, ValidationError,):
+                        except (model.DoesNotExist, ValidationError):
                             pass
                         else:
                             if hasattr(model_instance, 'key'):
                                 th.verbose_name = f'{th.name} with: {model_instance.key}'
                                 if hasattr(model_instance, 'name'):
                                     th.verbose_name += f' / {model_instance.name}'
                             break
     elif signal_name == signals.SIGNAL_SCHEDULED:
         scheduled_at = huey.scheduled_at_from_task(task_obj)
         if scheduled_at:
             th.scheduled_at = scheduled_at
     th.end_at = signal_dt
     th.elapsed = history['elapsed']
     th.save()
 
 
 # Registration of shared signal handlers
 
 
 def register_huey_signals():
     from django import db
     from django_huey import DJANGO_HUEY, get_queue, pre_execute, post_execute, signal
 
     def close_db(task, task_value=None, exception=None, /, *, huey=None):
         assert huey is not None
         assert hasattr(huey, 'immediate')
         # Guard against immediate mode and transactions
         if (
             huey is None or getattr(huey, 'immediate', True) or
             any(dbw.in_atomic_block for dbw in db.connections.all())
         ):
             return
         db.close_old_connections()
 
     for qn in DJANGO_HUEY.get('queues', dict()):
         q = get_queue(qn)
 
         # hooks to clean up database connections at task boundaries
         pre_execute(queue=qn, name='close_db')(partial(close_db, huey=q))
         post_execute(queue=qn, name='close_db')(partial(close_db, huey=q))
 
         # signals for tracking / maintenance of tasks
         signal(signals.SIGNAL_INTERRUPTED, queue=qn)(on_interrupted)
         signal(queue=qn)(historical_task)
         signal(signals.SIGNAL_EXECUTING, queue=qn)(on_executing_remove_duplicates)
 
         # clean up old history and results from storage
         now_time = time.monotonic()
         now_dt = datetime.datetime.now(datetime.timezone.utc)
         # ruff: ignore[SIM118]
         for key in q.all_results().keys():
             if not key.startswith(storage_key_prefix):
                 continue
             history = q.get(peek=True, key=key)
             if not isinstance(history, dict):
                 continue
             if 'created' not in history:
                 # Set an incorrect created time to ensure eventual removal.
                 history['created'] = now_dt
                 q.put(key=key, data=history)
                 # Attempt to calculate an accurate age.
                 # This doesn't work all the time,
                 # so the created datetime is the fail-safe case.
                 seconds = now_time - history.get(signals.SIGNAL_EXECUTING, now_time)
                 age = datetime.timedelta(
                     seconds=max(0, seconds),
                 )
             else:
                 age = now_dt - history['created']
             if age > datetime.timedelta(days=7):
                 result_key = key[len(storage_key_prefix) :]
                 q.get(peek=False, key=result_key)
                 q.get(peek=False, key=key)
 
 
 class BackoffAlgorithm:
     # ruff: ignore[RUF012]
     _registry = {}
     key = None
 
     @staticmethod
     def calculate(attempt: int) -> int:
         raise NotImplementedError()
 
     @classmethod
     def lookup(cls, key: str) -> BackoffAlgorithm | None:
         return cls._registry.get(key, None)
 
     @classmethod
     def register(cls, algorithm: BackoffAlgorithm) -> str | None:
         """
         Merges a BackoffAlgorithm subclass into the base class registry.
         """
-        algorithm_key = issubclass(algorithm, cls) and getattr(algorithm, 'key', cls.key)
+        algorithm_key = (
+            issubclass(algorithm, cls) and getattr(algorithm, 'key', cls.key)
+        )
         if algorithm_key:
-            cls._registry.update({ algorithm_key: algorithm })
+            cls._registry.update({algorithm_key: algorithm})
             return algorithm_key
 
 
 class DjangoBackgroundTasksBackoff(BackoffAlgorithm):
     key = 'Django-Background-Tasks'
 
     @staticmethod
     def calculate(attempt: int) -> int:
         # The exact formula used by django-background-tasks
-        return 5 + (attempt ** 4)
+        return 5 + (attempt ** 4)  # fmt: skip
 
 
 class AttemptsTask(Task):
     """
     A Task base class that relies entirely on Task for defaults.
     Resolves algorithms using string keys mapped onto the BackoffAlgorithm base class.
     """
 
     backoff_base_class = BackoffAlgorithm
 
     def __init__(self, *args, backoff_class=None, **kwargs):
+        attr_backoff_class = getattr(self, 'backoff_class', None)
+        default_retry_backoff = getattr(self, 'default_retry_backoff', 0)
+        self_retry_backoff = getattr(self, 'retry_backoff', default_retry_backoff)
         super().__init__(*args, **kwargs)
 
-        self.backoff_class = backoff_class or self.backoff_class
+        # Used type to set backoff_class already?
+        # Otherwise, use the argument.
+        self.backoff_class = backoff_class or attr_backoff_class
         if self.backoff_class:
             # Register directly onto the algorithm base class
             self._backoff_key = self.backoff_base_class.register(self.backoff_class)
 
         algo_key = getattr(self, '_backoff_key', None)
         retry_backoff = kwargs.get('retry_backoff', None)
-        if retry_backoff is None and not self.default_retry_backoff and algo_key:
-            self.retry_backoff = 1
+        # We had a backoff_class and registered it.
+        # No retry_backoff was specified.
+        # The default retry backoff (zero) was assigned.
+        set_retry_backoff = (
+            algo_key and
+            retry_backoff is None and
+            self_retry_backoff == default_retry_backoff and
+            self.retry_backoff == self.default_retry_backoff
+        )
+        if set_retry_backoff:
+            self.retry_backoff = 1.15
 
     @property
     def retry_delay(self) -> int:

tubesync/sync/tasks.py

@@ -23,9 +23,9 @@
 from django_huey import lock_task as huey_lock_task, task as huey_task
 from django_huey import db_periodic_task, db_task, signal as huey_signal
 from huey import crontab as huey_crontab, signals as huey_signals
-from huey.exceptions import TaskLockedException
+from huey.exceptions import CancelExecution, TaskLockedException
 from common.huey import (
-    AttemptsTask, BackoffAlgorithm, CancelExecution, DjangoBackgroundTasksBackoff,
+    AttemptsTask, BackoffAlgorithm, DjangoBackgroundTasksBackoff,
     register_huey_signals,
 )
 from common.logger import log
@@ -1257,14 +1257,9 @@ def refresh_formats(media_id):
         save, retry, msg = media.refresh_formats()
         if save is not True:
             log.warning(f'Refreshing formats for "{media.key}" failed: {msg}')
-            exc = CancelExecution(
-                _('failed to refresh formats for:'),
-                f'{media.key} / {media.uuid}:',
-                msg,
-                retry=retry,
-            )
-            # combine the strings
-            exc.args = (' '.join(map(str, exc.args)),)
+            msg_arg = str(_('failed to refresh formats for:'))
+            msg_arg += f' {media.key} / {media.uuid}: {msg}'
+            exc = CancelExecution(msg=msg_arg, retry=retry)
             # store instance details
             exc.instance = dict(
                 key=media.key,