Pause scheduling temporarily on failures. By @real-yfprojects (#1299)

src/vorta/borg/borg_job.py

@@ -14,6 +14,7 @@
 from PyQt5.QtWidgets import QApplication
 from subprocess import Popen, PIPE, TimeoutExpired
 
+from vorta import application
 from vorta.borg.jobs_manager import JobInterface
 from vorta.i18n import trans_late, translate
 from vorta.store.models import EventLogModel, BackupProfileMixin
@@ -60,7 +61,7 @@ def __init__(self, cmd, params, site="default"):
 
         super().__init__()
         self.site_id = site
-        self.app = QApplication.instance()
+        self.app: application.VortaApp = QApplication.instance()
 
         # Declare labels here for translation
         self.category_label = {"files": trans_late("BorgJob", "Files"),

src/vorta/scheduler.py

@@ -3,7 +3,7 @@
 import threading
 from datetime import datetime as dt
 from datetime import timedelta
-from typing import Dict, NamedTuple, Optional, Union
+from typing import Dict, NamedTuple, Optional, Tuple, Union
 
 from PyQt5 import QtCore
 from PyQt5.QtCore import QTimer
@@ -48,6 +48,9 @@ def __init__(self):
         self.app: application.VortaApp = QApplication.instance()
         self.lock = threading.Lock()
 
+        # pausing will prevent scheduling for a specified time
+        self.pauses: Dict[int, Tuple[dt, QtCore.QTimer]] = dict()
+
         # Set additional timer to make sure background tasks stay scheduled.
         # E.g. after hibernation
         self.qt_timer = QTimer()
@@ -63,6 +66,108 @@ def tr(self, *args, **kwargs):
         scope = self.__class__.__name__
         return translate(scope, *args, **kwargs)
 
+    def pause(self, profile_id: int, until: Optional[dt] = None):
+        """
+        Call a timeout for scheduling of a given profile.
+
+        If `until` is omitted, a default time for the break is calculated.
+
+        .. warning::
+            This method won't work correctly when called from a non-`QThread`.
+
+
+        Parameters
+        ----------
+        profile_id : int
+            The profile to pause the scheduling for.
+        until : Optional[dt], optional
+            The time to end the pause, by default None
+        """
+        profile = BackupProfileModel.get_or_none(id=profile_id)
+        if profile is None:  # profile doesn't exist any more.
+            return
+
+        if profile.schedule_mode == 'off':
+            return
+
+        if until is None:
+            # calculate default timeout
+
+            if profile.schedule_mode == 'interval':
+                interval = timedelta(
+                    **{profile.schedule_interval_unit:
+                        profile.schedule_interval_count})
+            else:
+                # fixed
+                interval = timedelta(days=1)
+
+            timeout = interval // 6  # 60 / 6 = 10 [min]
+            timeout = max(min(timeout, timedelta(hours=1)),
+                          timedelta(minutes=1))  # 1 <= t <= 60
+
+            until = dt.now().replace(microsecond=0) + timeout
+        elif until < dt.now():
+            return
+
+        # remove existing schedule
+        self.remove_job(profile_id)
+
+        # setting timer for reschedule is not possible if called
+        # from a non-QThread -  it won't fail but won't work
+        timer_value = max(1, (until - dt.now()).total_seconds())
+        timer = QtCore.QTimer()
+        timer.setInterval(int(timer_value * 1000) + 100)
+        timer.timeout.connect(lambda: self.set_timer_for_profile(profile_id))
+        timer.start()
+
+        # set timeout/pause
+        other_pause = self.pauses.get(profile_id)
+        if other_pause is not None:
+            logger.debug(f"Override existing timeout for profile {profile_id}")
+
+        self.pauses[profile_id] = (until, timer)
+        logger.debug(
+            f"Paused {profile_id} until {until.strftime('%Y-%m-%d %H:%M:%S')}")
+
+    def unpause(self, profile_id: int):
+        """
+        Return to scheduling for a profile.
+
+        Parameters
+        ----------
+        profile_id : int
+            The profile to end the timeout for.
+        """
+        profile = BackupProfileModel.get_or_none(id=profile_id)
+        if profile is None:  # profile doesn't exist any more.
+            return
+
+        pause = self.pauses.get(profile_id)
+        if pause is None:  # already unpaused
+            return
+
+        dummy, timer = pause
+        timer.stop()
+        del self.pauses[profile_id]
+
+        logger.debug(f"Unpaused {profile_id}")
+
+        self.set_timer_for_profile(profile_id)
+
+    def paused(self, profile_id: int) -> bool:
+        """
+        Determine whether scheduling for a profile is paused
+
+        Parameters
+        ----------
+        profile_id : int
+
+        Returns
+        -------
+        bool
+        """
+        return self.pauses.get(profile_id) is not None
+
     def set_timer_for_profile(self, profile_id: int):
         """
         Set a timer for next scheduled backup run of this profile.
@@ -83,6 +188,19 @@ def set_timer_for_profile(self, profile_id: int):
 
             self.remove_job(profile_id)  # reset schedule
 
+            pause = self.pauses.get(profile_id)
+            if pause is not None:
+                pause_end, timer = pause
+                if dt.now() < pause_end:
+                    logger.debug(
+                        'Nothing scheduled for profile %s ' +
+                        'because of timeout until %s.',
+                        profile_id, pause[0].strftime('%Y-%m-%d %H:%M:%S'))
+                    return
+                else:
+                    timer.stop()
+                    del self.pauses[profile_id]
+
             if profile.repo is None:  # No backups without repo set
                 logger.debug(
                     'Nothing scheduled for profile %s because of unset repo.',
@@ -251,25 +369,28 @@ def create_backup(self, profile_id):
         # Skip if a job for this profile (repo) is already in progress
         if self.app.jobs_manager.is_worker_running(site=profile.repo.id):
             logger.debug('A job for repo %s is already active.', profile.repo.id)
+            self.pause(profile_id)
             return
 
-        self.lock.acquire()
+        with self.lock:
-        logger.info('Starting background backup for %s', profile.name)
+            logger.info('Starting background backup for %s', profile.name)
-        notifier.deliver(self.tr('Vorta Backup'),
+            notifier.deliver(self.tr('Vorta Backup'),
-                         self.tr('Starting background backup for %s.') % profile.name,
+                             self.tr('Starting background backup for %s.') % profile.name,
-                         level='info')
+                             level='info')
-        msg = BorgCreateJob.prepare(profile)
+            msg = BorgCreateJob.prepare(profile)
-        if msg['ok']:
+            if msg['ok']:
-            logger.info('Preparation for backup successful.')
+                logger.info('Preparation for backup successful.')
-            msg['category'] = 'scheduled'
+                msg['category'] = 'scheduled'
-            job = BorgCreateJob(msg['cmd'], msg, profile.repo.id)
+                job = BorgCreateJob(msg['cmd'], msg, profile.repo.id)
-            job.result.connect(self.notify)
+                job.result.connect(self.notify)
-            self.app.jobs_manager.add_job(job)
+                self.app.jobs_manager.add_job(job)
-        else:
+            else:
-            logger.error('Conditions for backup not met. Aborting.')
+                logger.error('Conditions for backup not met. Aborting.')
-            logger.error(msg['message'])
+                logger.error(msg['message'])
-            notifier.deliver(self.tr('Vorta Backup'), translate('messages', msg['message']), level='error')
+                notifier.deliver(self.tr('Vorta Backup'),
-        self.lock.release()
+                                 translate('messages', msg['message']),
+                                 level='error')
+                self.pause(profile_id)
 
     def notify(self, result):
         notifier = VortaNotifications.pick()
@@ -281,10 +402,17 @@ def notify(self, result):
                              self.tr('Backup successful for %s.') % profile_name,
                              level='info')
             logger.info('Backup creation successful.')
+            # unpause scheduler
+            self.unpause(result['params']['profile_id'])
+
             self.post_backup_tasks(profile_id)
         else:
             notifier.deliver(self.tr('Vorta Backup'), self.tr('Error during backup creation.'), level='error')
             logger.error('Error during backup creation.')
+            # pause scheduler
+            # if a scheduled backup fails the scheduler should pause
+            # temporarily.
+            self.pause(result['params']['profile_id'])
 
         self.set_timer_for_profile(profile_id)