add JobModel: persistent scheduler job store (#2360, phase 1)

src/vorta/borg/borg_job.py

@@ -20,11 +20,10 @@
 from vorta.i18n import trans_late, translate
 from vorta.keyring.abc import VortaKeyring
 from vorta.keyring.db import VortaDBKeyring
-from vorta.store.models import EventLogModel
+from vorta.store.models import EventLogModel, db_lock
 from vorta.utils import borg_compat, pretty_bytes
 
 keyring_lock = Lock()
-db_lock = Lock()
 logger = logging.getLogger(__name__)
 
 FakeRepo = namedtuple('Repo', ['url', 'name', 'id', 'extra_borg_arguments', 'encryption'])

src/vorta/scheduler.py

@@ -7,6 +7,7 @@
 from datetime import timedelta
 from typing import Any, NamedTuple
 
+import peewee as pw
 from packaging import version
 from PyQt6 import QtCore, QtDBus
 from PyQt6.QtCore import QTimer
@@ -20,7 +21,7 @@
 from vorta.borg.prune import BorgPruneJob
 from vorta.i18n import translate
 from vorta.notifications import VortaNotifications
-from vorta.store.models import BackupProfileModel, EventLogModel
+from vorta.store.models import BackupProfileModel, EventLogModel, JobModel
 from vorta.utils import borg_compat, get_network_status_monitor
 
 logger = logging.getLogger(__name__)
@@ -323,13 +324,19 @@ def set_timer_for_profile(self, profile_id: int) -> None:
                             profile.name,
                             profile_id,
                         )
-                        self.create_backup(profile_id)
+                        self.create_backup(profile_id, trigger=JobModel.Trigger.CATCHUP.value)
                     finally:
                         self.lock.acquire()  # with-statement will try to release
 
                     return  # create_backup will lead to a call to this method
                 elif profile.schedule_make_up_missed and not self._net_up and needs_network:
                     logger.debug('Skipping catchup %s (%s), the network is not available', profile.name, profile.id)
+                    self._record_skip(
+                        profile,
+                        JobModel.Trigger.CATCHUP.value,
+                        'Network unavailable for catch-up.',
+                        scheduled_at=next_time,
+                    )
 
                 # calculate next time from now
                 if profile.schedule_mode == 'interval':
@@ -421,7 +428,37 @@ def next_job_for_profile(self, profile_id: int) -> ScheduleStatus:
             return ScheduleStatus(ScheduleStatusType.UNSCHEDULED)
         return ScheduleStatus(job['type'], time=job.get('dt'))  # type: ignore[arg-type]
 
-    def create_backup(self, profile_id: int) -> None:
+    def _record_skip(
+        self,
+        profile: BackupProfileModel,
+        trigger: str,
+        reason: str,
+        status: str = JobModel.Status.SKIPPED.value,
+        scheduled_at: dt | None = None,
+    ) -> None:
+        """Record a job outcome, deduplicated on the occurrence when one is known."""
+        lookup = {
+            'profile': str(profile.id),
+            'trigger': trigger,
+            'status': status,
+            'scheduled_at': scheduled_at,
+        }
+        details = {
+            'profile_name': profile.name,
+            'repo_url': profile.repo.url if profile.repo else None,
+            'job_type': JobModel.Type.BACKUP.value,
+            'reason': reason,
+        }
+
+        try:
+            if scheduled_at is None:
+                JobModel.create(**lookup, **details)
+            else:
+                JobModel.get_or_create(**lookup, defaults=details)
+        except pw.PeeweeException:
+            logger.warning('Could not record job for profile %s.', profile.id, exc_info=True)
+
+    def create_backup(self, profile_id: int, trigger: str = JobModel.Trigger.SCHEDULED.value) -> None:
         notifier = VortaNotifications.pick()
         profile = BackupProfileModel.get_or_none(id=profile_id)
 
@@ -432,6 +469,7 @@ def create_backup(self, profile_id: int) -> None:
         # 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._record_skip(profile, trigger, 'Repository is busy with another job.')
             self.pause(profile_id)
             return
 
@@ -461,8 +499,11 @@ def create_backup(self, profile_id: int) -> None:
                         translate('messages', msg['message']),
                         level='error',
                     )
+                    status = JobModel.Status.FAILED.value
                 else:
                     logger.info('Backup skipped: %s', msg['message'])
+                    status = JobModel.Status.SKIPPED.value
+                self._record_skip(profile, trigger, msg['message'], status=status)
                 self.pause(profile_id)
 
     def notify(self, result: dict[str, Any]) -> None:

src/vorta/store/connection.py

@@ -18,6 +18,7 @@
     BackupProfileModel,
     EventLogModel,
     ExclusionModel,
+    JobModel,
     RepoModel,
     RepoPassword,
     SchemaVersion,
@@ -57,6 +58,7 @@ def init_db(con: pw.SqliteDatabase | None = None) -> None:
             ArchiveModel,
             WifiSettingModel,
             EventLogModel,
+            JobModel,
             SchemaVersion,
             ExclusionModel,
         ]
@@ -84,6 +86,9 @@ def init_db(con: pw.SqliteDatabase | None = None) -> None:
         entry.not_in(last_scheduled_backups_per_profile),
     ).execute()
 
+    # Delete old job records after 6 months. Nothing derives scheduling state from them.
+    JobModel.delete().where(JobModel.created_at < six_months_ago).execute()
+
     # Migrations
     current_schema, created = SchemaVersion.get_or_create(id=1, defaults={'version': SCHEMA_VERSION})
     current_schema.save()

src/vorta/store/models.py

@@ -10,6 +10,7 @@
 import logging
 from datetime import datetime
 from enum import Enum
+from threading import Lock
 from typing import Any
 
 import peewee as pw
@@ -19,6 +20,7 @@
 from vorta.views.utils import get_exclusion_presets
 
 DB = pw.Proxy()
+db_lock = Lock()
 logger = logging.getLogger(__name__)
 
 
@@ -247,6 +249,40 @@ class Meta:
         database = DB
 
 
+class JobModel(BaseModel):
+    """Lifecycle record of a scheduled background job."""
+
+    class Status(Enum):
+        SCHEDULED = 'scheduled'
+        RUNNING = 'running'
+        COMPLETED = 'completed'
+        FAILED = 'failed'
+        SKIPPED = 'skipped'
+        INTERRUPTED = 'interrupted'
+
+    class Type(Enum):
+        BACKUP = 'backup'
+
+    class Trigger(Enum):
+        SCHEDULED = 'scheduled'
+        CATCHUP = 'catchup'
+
+    profile = pw.CharField(null=True)
+    profile_name = pw.CharField(null=True)
+    repo_url = pw.CharField(null=True)
+    job_type = pw.CharField(default=Type.BACKUP.value)
+    status = pw.CharField(default=Status.SCHEDULED.value)
+    trigger = pw.CharField(null=True)
+    scheduled_at = pw.DateTimeField(null=True)
+    reason = pw.CharField(null=True)
+    event_log = pw.ForeignKeyField(EventLogModel, null=True, backref='jobs')
+    created_at = pw.DateTimeField(default=datetime.now)
+
+    class Meta:
+        database = DB
+        indexes = ((('profile', 'status', 'created_at'), False),)
+
+
 class SchemaVersion(BaseModel):
     """Keep DB version to apply the correct migrations."""
 

tests/unit/conftest.py

@@ -13,6 +13,7 @@
     ArchiveModel,
     BackupProfileModel,
     EventLogModel,
+    JobModel,
     RepoModel,
     RepoPassword,
     SchemaVersion,
@@ -59,6 +60,7 @@ def all_workers_finished(jobs_manager):
     ArchiveModel,
     WifiSettingModel,
     EventLogModel,
+    JobModel,
     SchemaVersion,
 ]
 

tests/unit/test_job_model.py

@@ -0,0 +1,24 @@
+from datetime import datetime, timedelta
+
+import vorta.store.connection
+from vorta.store.models import EventLogModel, JobModel
+
+
+def test_job_links_to_event_log():
+    """A job's execution result is reached through the event_log relation."""
+    log = EventLogModel.create(category='scheduled', subcommand='create')
+    job = JobModel.create(profile=1, status=JobModel.Status.COMPLETED.value, event_log=log)
+
+    assert job.event_log.id == log.id
+    assert [j.id for j in log.jobs] == [job.id]
+
+
+def test_old_jobs_are_purged_on_init():
+    """Job records older than six months are dropped when the DB is opened."""
+    old = JobModel.create(profile=1, created_at=datetime.now() - timedelta(days=200))
+    recent = JobModel.create(profile=1, created_at=datetime.now() - timedelta(days=20))
+
+    vorta.store.connection.init_db()
+
+    assert JobModel.get_or_none(id=old.id) is None
+    assert JobModel.get_or_none(id=recent.id) is not None

tests/unit/test_scheduler.py

@@ -9,7 +9,7 @@
 import vorta.borg
 import vorta.scheduler
 from vorta.scheduler import ScheduleStatus, ScheduleStatusType, VortaScheduler
-from vorta.store.models import BackupProfileModel, EventLogModel
+from vorta.store.models import BackupProfileModel, EventLogModel, JobModel
 
 PROFILE_NAME = 'Default'
 FIXED_SCHEDULE = 'fixed'
@@ -273,3 +273,107 @@ def test_create_backup_no_error_notification_on_info_level(qapp, qtbot, mocker,
     # The error notification should be suppressed for an expected skip.
     assert mock_notifier.deliver.call_count == 1
     assert mock_notifier.deliver.call_args.kwargs.get('level') != 'error'
+
+
+def test_create_backup_records_skip_reason(qapp, qtbot, mocker):
+    """A skipped scheduled backup is recorded as a JobModel row with its reason."""
+    mocker.patch(
+        'vorta.scheduler.BorgCreateJob.prepare',
+        return_value={
+            'ok': False,
+            'message': 'Current Wifi is not allowed.',
+            'level': 'info',
+        },
+    )
+    jobs_before = JobModel.select().count()
+
+    qapp.scheduler.create_backup(1)
+
+    assert JobModel.select().count() == jobs_before + 1
+    job = JobModel.select().order_by(JobModel.id.desc()).get()
+    assert job.status == JobModel.Status.SKIPPED.value
+    assert job.reason == 'Current Wifi is not allowed.'
+    assert job.profile == '1'
+    assert job.profile_name == PROFILE_NAME
+
+
+def test_create_backup_records_failure_not_skip(qapp, qtbot, mocker):
+    """An unexpected prepare() failure is recorded as failed, not skipped."""
+    mocker.patch(
+        'vorta.scheduler.BorgCreateJob.prepare',
+        return_value={
+            'ok': False,
+            'message': 'Add a backup repository first.',
+        },
+    )
+    jobs_before = JobModel.select().count()
+
+    qapp.scheduler.create_backup(1)
+
+    assert JobModel.select().count() == jobs_before + 1
+    job = JobModel.select().order_by(JobModel.id.desc()).get()
+    assert job.status == JobModel.Status.FAILED.value
+    assert job.reason == 'Add a backup repository first.'
+
+
+def test_create_backup_records_skip_when_repo_busy(qapp, mocker):
+    """A scheduled run blocked by a busy repo is recorded as a skipped JobModel row."""
+    mocker.patch.object(qapp.jobs_manager, 'is_worker_running', return_value=True)
+    jobs_before = JobModel.select().count()
+
+    qapp.scheduler.create_backup(1)
+
+    assert JobModel.select().count() == jobs_before + 1
+    job = JobModel.select().order_by(JobModel.id.desc()).get()
+    assert job.status == JobModel.Status.SKIPPED.value
+    assert job.reason == 'Repository is busy with another job.'
+
+
+def test_create_backup_keeps_the_catchup_trigger(qapp, mocker):
+    """A catch-up run that gets skipped is not recorded as an ordinary scheduled run."""
+    mocker.patch.object(qapp.jobs_manager, 'is_worker_running', return_value=True)
+
+    qapp.scheduler.create_backup(1, trigger=JobModel.Trigger.CATCHUP.value)
+
+    job = JobModel.select().order_by(JobModel.id.desc()).get()
+    assert job.trigger == JobModel.Trigger.CATCHUP.value
+
+
+def test_set_timer_records_skip_when_network_down_for_catchup(clockmock):
+    """A catch-up run blocked by a down network is recorded as a skipped JobModel row."""
+    scheduler = VortaScheduler()
+    scheduler._net_up = False
+
+    time = dt(2020, 5, 6, 4, 30)
+    clockmock.now.return_value = time
+
+    profile = BackupProfileModel.get(name=PROFILE_NAME)
+    profile.schedule_make_up_missed = True
+    profile.schedule_mode = INTERVAL_SCHEDULE
+    profile.schedule_interval_unit = 'hours'
+    profile.schedule_interval_count = 3
+    profile.save()
+
+    last_run = time - td(hours=6)
+    EventLogModel.create(
+        subcommand='create',
+        profile=profile.id,
+        returncode=0,
+        category='scheduled',
+        start_time=last_run,
+        end_time=last_run,
+    )
+    jobs_before = JobModel.select().count()
+
+    scheduler.set_timer_for_profile(profile.id)
+
+    assert JobModel.select().count() == jobs_before + 1
+    job = JobModel.select().order_by(JobModel.id.desc()).get()
+    assert job.status == JobModel.Status.SKIPPED.value
+    assert job.trigger == JobModel.Trigger.CATCHUP.value
+    assert job.reason == 'Network unavailable for catch-up.'
+    assert job.scheduled_at == last_run + td(hours=3)
+
+    # Re-evaluating the same missed run must not add a second row.
+    scheduler.set_timer_for_profile(profile.id)
+    assert JobModel.select().count() == jobs_before + 1