fix(tasks): also use the setting for prune_queue_storage in register_huey_signals

tubesync/common/huey.py

@@ -1,11 +1,15 @@
 from __future__ import annotations
 
 import datetime
+import itertools
 import subprocess
 import threading
 import time
 import uuid
+
 from functools import partial
+from pathlib import Path
+
 from huey import (
     Huey as huey_Huey,
     signals,
@@ -14,7 +18,7 @@
 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
@@ -61,579 +65,674 @@
         return conn
 
     def initialize_schema(self):
+        kv_result_bool = (
+            'ALTER TABLE kv ADD COLUMN '
+            'result_bool INTEGER NOT NULL DEFAULT 0'
+        )
+        kv_created_at = (
+            'ALTER TABLE kv ADD COLUMN '
+            'created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP'
+        )
+        kv_updated_at = (
+            'ALTER TABLE kv ADD COLUMN '
+            'updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP CHECK(updated_at >= created_at)'
+        )
+        trigger_kv = (
+            'CREATE TRIGGER IF NOT EXISTS trg_kv_updated_at '
+            'AFTER UPDATE ON kv FOR EACH ROW '
+            'BEGIN '
+            '  UPDATE kv SET updated_at = MAX(CURRENT_TIMESTAMP, NEW.created_at) '
+            '  WHERE queue = OLD.queue AND key = OLD.key; '
+            'END'
+        )
+        with self.db(close=True) as curs:
+            curs.execute('PRAGMA auto_vacuum')
+            current_auto_vacuum = curs.fetchall()
+            curs.execute('PRAGMA table_info(kv)')
+            table_info_kv = curs.fetchall()
+        existing_kv_columns = {row[1] for row in table_info_kv}
+        ddl = list(self.ddl)
+        if 'result_bool' not in existing_kv_columns:
+            ddl.append(kv_result_bool)
+        if 'created_at' not in existing_kv_columns:
+            ddl.append(kv_created_at)
+        if 'updated_at' not in existing_kv_columns:
+            ddl.append(kv_updated_at)
+        ddl.append(trigger_kv)
+        self.ddl = tuple(ddl)
         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)
+            actual = int(current_auto_vacuum[0][0])
+            expected = int(auto_vacuum_dict.get(self.auto_vacuum) or self.auto_vacuum)
         except (TypeError, ValueError):
             pass
         else:
-            vacuum_db = (current != expected)  # fmt: skip
+            vacuum_db = (expected != actual) or (0 == self.auto_vacuum)  # fmt: skip
         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')
 
+    def put_data(self, key, value, is_result=False):
+        q_str = (
+            'INSERT OR REPLACE INTO kv '
+            '(queue, result_bool, key, value) '
+            'VALUES (?, ?, ?, ?)'
+        )
+        p_tuple = (
+            self.name, int(is_result is True), key, self.to_blob(value),
+        )  # fmt: skip
+        self.sql(query=q_str, params=p_tuple, commit=True)
+
+    def _islice_batched(iterable, n):
+        it = iter(iterable)
+        chunk = tuple(itertools.islice(it, n))
+        while chunk:
+            yield chunk
+            chunk = tuple(itertools.islice(it, n))
+
+    def expire_results(self, dt, /, *, chunk_size=100):
+        batched = getattr(itertools, 'batched', self._islice_batched)
+        utc_dt = dt.astimezone(datetime.timezone.utc)
+        sqlite_ts_str = utc_dt.strftime('%Y-%m-%d %H:%M:%S')
+        select_all_q_str = (
+            'SELECT queue, key FROM kv '
+            'WHERE result_bool = 1 AND queue = ? '
+            'AND updated_at < datetime(?)'
+        )
+        delete_row_q_str = (
+            'DELETE FROM kv '
+            'WHERE queue = ? AND key = ?'
+        )
+        if not callable(batched):
+            return
+        batches = batched(
+            self.sql(query=select_all_q_str, params=(self.name, sqlite_ts_str), results=True),
+            chunk_size,
+        )
+        for batch in batches:
+            with self.db(commit=True) as curs:
+                for queue, key in batch:
+                    curs.execute(delete_row_q_str, (queue, key))
 
-class Huey(huey_Huey):
-    # do not use __len__ (pending_count) for bool
-    def __bool__(self):
-        return (self._registry._registry or self._registry._periodic_tasks)  # fmt: skip
 
+class Huey(huey_Huey):
     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):
+    def prune_results(self, older_than_dt, /):
+        expire_func = getattr(self.storage, 'expire_results', None)
+        if callable(expire_func):
+            expire_func(older_than_dt)
+
+    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 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)
+    name = f"huey{('_' + prefix) if prefix else ''}_{key}"
+    tasks_dir = tasks_dir or '/config/tasks'
     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'),
+            filename=str(Path(tasks_dir) / 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=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 callable(getattr(huey, 'pending', None))
+    assert callable(getattr(huey, 'revoke_by_id', None))
+    assert callable(getattr(huey, 'scheduled', None))
     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)
+    assert callable(getattr(huey, 'enqueue', None))
     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)
+    assert callable(getattr(huey, 'delete', None))
+    assert callable(getattr(huey, 'get', None))
+    assert callable(getattr(huey, 'put_result', None))
 
     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  # 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)),
             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)
+    remove_from_storage = (
+        signals.SIGNAL_COMPLETE in history or
+        signals.SIGNAL_REVOKED in history or
+        signals.SIGNAL_EXPIRED in history
+    )  # fmt: skip
+    if remove_from_storage:
+        huey.delete(key=storage_key)
     else:
-        huey.put(key=storage_key, data=history)
+        huey.put_result(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
+    elif signal_name in (signals.SIGNAL_EXPIRED, signals.SIGNAL_REVOKED):
+        th.verbose_name = f'[{signal_name}] {th.verbose_name or th.name}'
     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.conf import settings
     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()
 
-    def prune_queue_storage(qn):
+    def prune_queue_storage(qn, days):
         # clean up old history and results from storage
         q = get_queue(qn)
 
         now_time = time.monotonic()
         now_dt = datetime.datetime.now(datetime.timezone.utc)
 
-        # ruff: ignore[SIM118]
-        for key in q.all_results().keys():
+        # use the new batched deletion and results expiration support first
+        prune_func = getattr(q, 'prune_results', None)
+        if callable(prune_func):
+            delta = now_dt - datetime.timedelta(days=days)
+            prune_func(delta)
+
+        # still try the old routine (it should have little work left)
+        for key in q.all_results():
             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)
+                q.put_result(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 _create_cleanup_worker(qn):
+            if age > datetime.timedelta(days=days):
+                result_key = key.removeprefix(storage_key_prefix)
+                # remove task results from storage
+                q.delete(key=result_key)
+                # remove revoked flag from storage
+                q.restore_by_id(id=result_key)
+                # remove task history metadata from storage
+                q.delete(key=key)
+
+
+    def _create_cleanup_worker(qn, days):
         def _target():
             try:
-                prune_queue_storage(qn)
+                prune_queue_storage(qn, days)
             # ruff: ignore[BLE001]
             except Exception:
                 from common.logger import log
                 log.exception(f'Cleanup of {qn=} failed.')
         # The pruning is happening in a huey consumer worker;
         # short running tasks won't start these threads.
         worker = threading.Thread(target=_target, daemon=True)
         worker.queue_name = qn
         return worker
 
+    days_to_keep = getattr(settings, 'COMPLETED_TASKS_DAYS_TO_KEEP', 14)
     queues = { qn: get_queue(qn) for qn in DJANGO_HUEY.get('queues', dict()) }
     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)
 
         # add the thread to prune the queue storage
-        threads.add(_create_cleanup_worker(qn))
+        threads.add(_create_cleanup_worker(qn, days_to_keep))
 
     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)  # 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)
 
         # 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)
         # 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