feat: integrate django-tasks-db

Pipfile

@@ -11,6 +11,7 @@ ruff = "*"
 [packages]
 django = "~=6.0.0"
 django-huey = "*"
+django-tasks-db = "*"
 django-sass-processor = {extras = ["management-command"], version = "*"}
 pillow = "*"
 whitenoise = "*"

config/root/etc/s6-overlay/s6-rc.d/django-tasks-default/run

@@ -0,0 +1,14 @@
+#!/command/with-contenv bash
+
+set -e
+
+scan_directory='/run/service-django-tasks-default-workers'
+
+if [[ -d "${scan_directory}" ]]; then
+    exec /command/s6-svscan "${scan_directory}"
+fi
+
+exec nice -n "${TUBESYNC_NICE:-1}" s6-setuidgid app \
+    /usr/bin/python3 /app/manage.py db_worker -v 2 \
+    --no-reload --interval 20 --max-tasks 100 \
+    --queue-name '*' --exclude-queues limited

config/root/etc/s6-overlay/s6-rc.d/django-tasks-default/type

@@ -0,0 +1 @@
+longrun

config/root/etc/s6-overlay/s6-rc.d/django-tasks-default/up

@@ -0,0 +1,26 @@
+#!/command/with-contenv bash
+
+set -e
+
+nice -n "${TUBESYNC_NICE:-1}" s6-setuidgid app \
+    /usr/bin/python3 /app/manage.py prune_db_task_results \
+    --queue-name '*'
+
+cores="$(nproc --ignore=2 || echo 2)"
+WORKERS="${TUBESYNC_DJANGO_TASKS_WORKERS:-${cores}}"
+
+worker_template='/etc/s6-overlay/s6-rc.d/django-tasks-default/worker@'
+scan_directory='/run/service-django-tasks-default-workers'
+
+(cd /run && rm -rf "${scan_directory}")
+
+/command/s6-instance-maker "${worker_template}" "${scan_directory}"
+
+create_workers() {
+    local i=0
+    while (( ++i <= $1 )); do
+        /command/s6-instance-create -w "${scan_directory}" "${i}"
+    done
+}
+
+create_workers "${WORKERS}"

config/root/etc/s6-overlay/s6-rc.d/django-tasks-default/worker@/down-signal

@@ -0,0 +1 @@
+SIGINT

config/root/etc/s6-overlay/s6-rc.d/django-tasks-default/worker@/run

@@ -0,0 +1,6 @@
+#!/command/with-contenv bash
+
+exec nice -n "${TUBESYNC_NICE:-1}" s6-setuidgid app \
+    /usr/bin/python3 /app/manage.py db_worker -v 2 \
+    --no-reload --interval 20 --max-tasks 100 \
+    --queue-name '*' --exclude-queues limited

config/root/etc/s6-overlay/s6-rc.d/django-tasks-default/worker@/type

@@ -0,0 +1 @@
+longrun

config/root/etc/s6-overlay/s6-rc.d/django-tasks-default/worker@/up

@@ -0,0 +1,5 @@
+#!/command/with-contenv bash
+
+exec nice -n "${TUBESYNC_NICE:-1}" s6-setuidgid app \
+    /usr/bin/python3 /app/manage.py prune_db_task_results \
+    --queue-name '*'

config/root/etc/s6-overlay/s6-rc.d/django-tasks-huey/down-signal

@@ -0,0 +1 @@
+SIGINT

config/root/etc/s6-overlay/s6-rc.d/django-tasks-huey/run

@@ -0,0 +1,5 @@
+#!/command/with-contenv bash
+
+exec nice -n "${TUBESYNC_NICE:-1}" s6-setuidgid app \
+    /usr/bin/python3 /app/manage.py run_huey \
+    --shutdown-timeout 60

config/root/etc/s6-overlay/s6-rc.d/django-tasks-huey/type

@@ -0,0 +1 @@
+longrun

config/root/etc/s6-overlay/s6-rc.d/django-tasks-net-limited/down-signal

@@ -0,0 +1 @@
+SIGINT

config/root/etc/s6-overlay/s6-rc.d/django-tasks-net-limited/run

@@ -0,0 +1,6 @@
+#!/command/with-contenv bash
+
+exec nice -n "${TUBESYNC_NICE:-1}" s6-setuidgid app \
+    /usr/bin/python3 /app/manage.py db_worker -v 2 \
+    --no-reload --interval 20 --max-tasks 100 \
+    --queue-name limited

config/root/etc/s6-overlay/s6-rc.d/django-tasks-net-limited/type

@@ -0,0 +1 @@
+longrun

config/root/etc/s6-overlay/user-bundles.d/tubesync-init/type

@@ -0,0 +1 @@
+bundle

tubesync/common/huey.py

@@ -1,584 +1,617 @@
 from __future__ import annotations
 
+import concurrent.futures
 import datetime
 import subprocess
+import threading
 import time
 import uuid
 from functools import partial
 from huey import (
     CancelExecution, Huey as huey_Huey,
     signals, utils,
 )
 from huey.api import Task, TaskLock
 from huey.exceptions import TaskLockedException
 from huey.storage import SqliteStorage as huey_SqliteStorage
 from pathlib import Path
 from .timestamp import datetime_to_timestamp, timestamp_to_datetime
 from .utils import get_usable_cpu_count
 from .yt_dlp import retry_django_db
 
 
 def _set_acquired(self, value=True):
     if value is not True:
         return self.clear()
     try:
         self.__enter__()
     except TaskLockedException:
         return False
     else:
         return True
 #TaskLock._set_acquired = _set_acquired
 TaskLock.acquired = property(
     TaskLock.is_locked,
     _set_acquired,
     TaskLock.clear,
 )
 
 
 class SqliteStorage(huey_SqliteStorage):
     begin_sql = 'BEGIN IMMEDIATE'
     auto_vacuum = 'INCREMENTAL'
     journal_size_limit = 1024 * 1024 * 64
     wal_autocheckpoint = 100
     vacuum_pages = 10
 
     def _create_connection(self):
         conn = super()._create_connection()
         conn.execute(f'PRAGMA auto_vacuum = {self.auto_vacuum}')
         conn.execute('PRAGMA foreign_keys = ON')
         conn.execute(f'PRAGMA journal_size_limit = {self.journal_size_limit}')
         conn.execute('PRAGMA user_version = 1')
         conn.execute(f'PRAGMA wal_autocheckpoint={self.wal_autocheckpoint}')
         # copied from huey_SqliteHuey to use EXTRA or NORMAL
         # instead of FULL or OFF
         conn.execute('pragma synchronous=%s' % (3 if self._fsync else 1))
         # take at most vacuum_pages off the free list
         conn.execute(f'PRAGMA incremental_vacuum({self.vacuum_pages})')
         return conn
 
     def initialize_schema(self):
         super().initialize_schema()
         auto_vacuum_dict = {'NONE': 0, 'FULL': 1, 'INCREMENTAL': 2}
         valid = set(auto_vacuum_dict.values()) | set(auto_vacuum_dict.keys())
         assert self.auto_vacuum in valid, 'auto_vacuum was invalid'
         current = self.sql('PRAGMA auto_vacuum', results=True)
         expected = auto_vacuum_dict.get(self.auto_vacuum) or self.auto_vacuum
         vacuum_db = True
         try:
             current = int(current[0][0])
             expected = int(expected)
         except (TypeError, ValueError,):
             pass
         else:
             vacuum_db = (current != expected)
         with self.db(close=True) as curs:
             curs.execute('PRAGMA wal_checkpoint(TRUNCATE)')
             if vacuum_db:
                 curs.execute(f'PRAGMA auto_vacuum = {self.auto_vacuum}')
                 curs.execute('VACUUM')
 
 
 class Huey(huey_Huey):
     # do not use __len__ (pending_count) for bool
     def __bool__(self):
         return (self._registry._registry or self._registry._periodic_tasks)
 
     def _emit(self, signal, task, *args, **kwargs):
         kwargs['huey'] = self
         super()._emit(signal, task, *args, **kwargs)
 
     def create_consumer(self, **options):
         consumer = super().create_consumer(**options)
         suffix = f'worker.{consumer.worker_type}'
         consumer._logger = consumer._logger.getChild(suffix)
         return consumer
 
     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
             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()),),
     )
 
 
 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,
         )
 
 
 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
 
 # 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
             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_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,
     ))
     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
     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),),
             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}",
         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,):
                         try:
                             model_instance = model.objects.get(pk=instance_uuid)
                         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()):
+    def prune_queue_storage(qn):
+        # clean up old history and results from storage
         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)
 
+    def _run_cleanup(qn):
+        def _target():
+            try:
+                prune_queue_storage(qn)
+            # ruff: ignore[BLE001]
+            except Exception:
+                from common.logger import log
+                log.exception(f'Cleanup of {qn=} failed.')
+        # The pruning is happening opportunistically;
+        # short running tasks won't wait for these threads.
+        worker = threading.Thread(target=_target, daemon=True)
+        worker.queue_name = qn
+        threads.add(worker)
+        worker.start()
+
+    queues = { qn: get_queue(qn) for qn in DJANGO_HUEY.get('queues', dict()) }
+    executor = concurrent.futures.ThreadPoolExecutor(
+        max_workers=get_usable_cpu_count(),
+    )
+    threads = set()
+    for qn, q in queues.items():
+        # 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)
+
+        # prune the queue storage
+        executor.submit(_run_cleanup, qn)
+
+    # wait for the worker threads to start,
+    # not for the storage pruning tasks to finish
+    executor.shutdown(wait=True)
+
+    return threads
+
 
 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)
         if algorithm_key:
             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)
 
 
 class AttemptsTask(Task):

tubesync/sync/tasks.py

@@ -18,6 +18,7 @@
 from django import db
 from django.conf import settings
 from django.core.files.uploadedfile import SimpleUploadedFile
+from django.tasks import task as django_task
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from django_huey import lock_task as huey_lock_task, task as huey_task
@@ -45,7 +46,7 @@
 from .youtube import YouTubeError
 
 atomic = db.transaction.atomic
-register_huey_signals()
+cleanup_threads = register_huey_signals()
 
 
 def get_task_map():
@@ -378,6 +379,25 @@ def wait_for_errors(model, /, *, queue_name=None, task_name=None):
         raise HueyConsumerError(_('queue consumer stopped'))
 
 
+@django_task(queue_name=Val(TaskQueue.FS))
+def cleanup_huey_storage():
+    workers = cleanup_threads
+    if not all(
+        (callable(getattr(workers, 'clear', None)),) +
+        tuple(callable(getattr(w, 'join', None)) for w in workers) +
+        tuple(callable(getattr(w, 'start', None)) for w in workers),
+    ):
+        return
+    for worker in workers:
+        try:
+            worker.start()
+        except RuntimeError:
+            pass
+        worker.join()
+    workers.clear()
+# cleanup_huey_storage_result = cleanup_huey_storage.enqueue()
+
+
 @db_task(priority=90, retries=2, retry_delay=150, queue=Val(TaskQueue.FS))
 @huey_lock_task('sync.tasks.cleanup_old_media', queue=Val(TaskQueue.FS))
 def cleanup_old_media(durable=True):

tubesync/tubesync/settings.py

@@ -28,6 +28,8 @@
     'django.contrib.humanize',
     'sass_processor',
     'django_huey',
+    'django_tasks_db',
+    'huey.contrib.djhuey',
     'common',
     'sync',
 ]
@@ -51,6 +53,21 @@
 FORCE_SCRIPT_NAME = None
 
 
+TASKS = {
+    'default': {
+        'BACKEND': 'django_tasks_db.DatabaseBackend',
+        'QUEUES': TaskQueue.values + ['default'],
+    },
+    'huey': {
+        'BACKEND': 'huey.contrib.djhuey.tasks_backend.HueyBackend',
+        'ENQUEUE_ON_COMMIT': True,
+    },
+}
+
+# Used by TASKS 'huey' backend
+HUEY = sqlite_tasks('tasks', thread=True, workers=0)
+HUEY['consumer']['verbose'] = True if DEBUG else None
+
 DJANGO_HUEY = {
     'default': TaskQueue.LIMIT.value,
     'queues': dict(),