diff --git a/Pipfile b/Pipfile index f68bd2b24..69c8bca08 100644 --- a/Pipfile +++ b/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" diff --git a/tubesync/common/huey.py b/tubesync/common/huey.py index 200d75a66..2149554aa 100644 --- a/tubesync/common/huey.py +++ b/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 @@ def _set_acquired(self, value=True): 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 @@ def initialize_schema(self): 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 @@ def initialize_schema(self): 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 @@ -144,7 +147,7 @@ def scheduled_at_from_task(self, task_obj, /): 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) @@ -155,23 +158,15 @@ 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())), ) @@ -203,24 +198,25 @@ def start_consumer(queue_name): 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 @@ -295,14 +291,14 @@ def sqlite_tasks(key, /, prefix=None, thread=None, workers=None, *, tasks_dir=No 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, @@ -391,7 +387,7 @@ def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None): signals.SIGNAL_ERROR, signals.SIGNAL_CANCELED, signals.SIGNAL_COMPLETE, - )) + )) # fmt: skip recorded_signals = frozenset(( signals.SIGNAL_REVOKED, signals.SIGNAL_EXPIRED, @@ -399,7 +395,7 @@ def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None): 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' @@ -411,7 +407,7 @@ def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None): 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, @@ -432,7 +428,7 @@ def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None): # 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 @@ -469,10 +465,10 @@ def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=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'): @@ -566,9 +562,11 @@ 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 @@ -578,7 +576,7 @@ class DjangoBackgroundTasksBackoff(BackoffAlgorithm): @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): @@ -590,17 +588,31 @@ class AttemptsTask(Task): 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: diff --git a/tubesync/sync/tasks.py b/tubesync/sync/tasks.py index 5721362e4..fd64911f4 100644 --- a/tubesync/sync/tasks.py +++ b/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,