Fixes #15: Fix missing notifications for Jobs cancelled during cutover

docs/development/netbox-internals.md

@@ -49,7 +49,7 @@ not a list of unsupported APIs alone.
 | `extras.models.ScriptModule.jobs` and `.event_rules` | `migration/source.py` | Identifies history and Event Rules that module deletion would also remove. Both are `GenericRelation`s, so Django's deletion collector cascades to those rows. |
 | `users.models.ObjectPermission.enabled` | `migration/cutover.py` | Disables captured Object Permission grants on the built-in feature. This is not an installation-wide write fence. |
 | `extras.models.EventRule.enabled` | `migration/cutover.py`, `migration/references.py` | Disables captured rules during handover and restores eligible rules after repointing their action and source types. |
-| `core.models.Job.terminate` | `migration/cutover.py` | Mirrored rather than called. Cancellation fails a waiting Job with a conditional update, because `terminate()` saves the whole row and would overwrite a start by a worker that took the task first. The Notification `terminate()` sends the owner of a failed Job is created the same way instead. |
+| `core.models.Job.terminate` and `extras.models.Notification.get_object_repr` | `migration/cutover.py` | `terminate()` is mirrored rather than called. Cancellation fails a waiting Job with a conditional update, because `terminate()` saves the whole row and would overwrite a start by a worker that took the task first. The guarded update and the owner notification commit together, refreshing any existing notice with the failure outcome, completion time, display text and unread state. |
 | `django_rq.get_queue` and `rq.job.Job.fetch` / `.delete` / `.exists` | `migration/cutover.py`, `migration/references.py` | Captures input from the RQ task, since `Job.enqueue()` does not store it on the Job row, then deletes the task. The scheduler can re-enqueue a due task fetched before deletion. The reference pass checks whether the task exists again, which a claim or a scheduler's re-queue writes back. |
 | `rq.exceptions.NoSuchJobError` | `migration/cutover.py` | Handles a missing RQ task. Capture warns and leaves the Job waiting, which keeps cutover incomplete, because `Job.enqueue()` saves the row before `on_commit()` queues the task. Cancellation treats the task as already deleted. |
 | `netaddr.IPAddress` and `.IPNetwork` | `migration/cutover.py` | The values an IP address or network variable cleans to, recorded as text the replacement's form field parses back. `netaddr` arrives with NetBox, which the plugin relies on without declaring. |

netbox_scripts/migration/cutover.py

@@ -6,15 +6,15 @@
 
 import django_rq
 import netaddr
-from django.db import models
+from django.db import models, transaction
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from rq.exceptions import NoSuchJobError
 from rq.job import Job as RQJob
 
 from core.choices import JobNotificationChoices, JobStatusChoices
 from core.events import JOB_FAILED
-from core.models import Job
+from core.models import Job, ObjectType
 from extras.models import EventRule, Notification
 from users.models import ObjectPermission
 from utilities.rqworker import get_all_workers
@@ -547,17 +547,30 @@ def _cancel_schedules(run, captured):
 
 
 def _fail_closed(job, error):
-    """Drop one job's queued task and fail its row unless a worker has started it, returning whether it did."""
+    """Remove the queued task and fail a waiting, unstarted Job. Return whether the Job was updated."""
     queue = django_rq.get_queue(job.queue_name)
     with contextlib.suppress(NoSuchJobError):
         RQJob.fetch(str(job.job_id), connection=queue.connection).delete()
-    # An update, since terminate()'s full save would overwrite a start by a worker that took the task first.
-    failed = Job.objects.filter(
-        pk=job.pk, status__in=legacy_source.cancellable_statuses(), started__isnull=True
-    ).update(status=JobStatusChoices.STATUS_FAILED, error=str(error), completed=timezone.now())
-    if failed and job.user_id and job.notifications != JobNotificationChoices.NOTIFICATION_NEVER:
-        # The owner notice terminate() would have sent for a failed job.
-        Notification(user_id=job.user_id, object=job, event_type=JOB_FAILED).save()
+    # Commit the failure and notification together so a resumed pass cannot skip the notice.
+    with transaction.atomic():
+        completed = timezone.now()
+        # Use a conditional update to avoid overwriting a Job that a worker has started.
+        failed = Job.objects.filter(
+            pk=job.pk, status__in=legacy_source.cancellable_statuses(), started__isnull=True
+        ).update(status=JobStatusChoices.STATUS_FAILED, error=str(error), completed=completed)
+        if failed and job.user_id and job.notifications != JobNotificationChoices.NOTIFICATION_NEVER:
+            # The notice terminate() would have sent. A retried Job can retain its previous one.
+            Notification.objects.update_or_create(
+                object_type=ObjectType.objects.get_for_model(job),
+                object_id=job.pk,
+                user_id=job.user_id,
+                defaults={
+                    'event_type': JOB_FAILED,
+                    'object_repr': Notification.get_object_repr(job),
+                    'created': completed,
+                    'read': None,
+                },
+            )
     return bool(failed)
 
 

netbox_scripts/tests/migration/test_cutover.py

@@ -14,7 +14,7 @@
 from rq.job import Job as RQJob
 
 from core.choices import JobStatusChoices, ManagedFileRootPathChoices
-from core.events import JOB_COMPLETED, OBJECT_UPDATED
+from core.events import JOB_COMPLETED, JOB_ERRORED, JOB_FAILED, OBJECT_UPDATED
 from core.models import AutoSyncRecord, DataFile, DataSource, Job, ObjectType
 from dcim.models import Site
 from extras.jobs import ScriptJob
@@ -612,7 +612,9 @@ def interrupt(failed, error):
 
     def test_a_run_a_worker_starts_while_it_is_cancelled_is_left_to_run(self):
         # In the cancellation, RQJob.fetch runs after the row is read and before it is failed.
+        owner = get_user_model().objects.create_user(username='owner')
         job = self.legacy_job(task_kwargs={})
+        Job.objects.filter(pk=job.pk).update(user=owner)
         cutover._capture(self.migration)
         carried = Job.objects.get(pk=job.pk)
         fetch = RQJob.fetch
@@ -628,6 +630,7 @@ def start_in_the_gap(*args, **kwargs):
         self.assertEqual(job.status, JobStatusChoices.STATUS_RUNNING)
         self.assertEqual(counts['schedules'], 0)
         self.assertEqual(self.migration.journal['schedules'][0]['cancellation'], 'running')
+        self.assertFalse(Notification.objects.filter(user=owner, object_id=job.pk).exists())
 
     def test_the_owner_of_a_cancelled_run_is_notified(self):
         owner = get_user_model().objects.create_user(username='owner')
@@ -638,6 +641,65 @@ def test_the_owner_of_a_cancelled_run_is_notified(self):
 
         self.assertTrue(Notification.objects.filter(user=owner, object_id=job.pk).exists())
 
+    def test_the_owner_of_a_cancelled_run_is_notified_after_an_interrupted_pass(self):
+        owner = get_user_model().objects.create_user(username='owner')
+        job = self.legacy_job(task_kwargs={})
+        Job.objects.filter(pk=job.pk).update(user=owner)
+
+        with (
+            mock.patch.object(Notification, 'save', side_effect=RuntimeError('stopped')),
+            self.assertRaises(RuntimeError),
+        ):
+            cutover.enter_cutover(self.migration)
+        # A failed notification save leaves the Job waiting for the next pass.
+        self.assertEqual(Job.objects.get(pk=job.pk).status, JobStatusChoices.STATUS_SCHEDULED)
+        reloaded = MigrationRun.objects.get(pk=self.migration.pk)
+        self.assertEqual(reloaded.journal['schedules'][0]['cancellation'], 'cancelling')
+
+        counts = cutover.enter_cutover(reloaded)
+
+        self.assertEqual(counts['schedules'], 1)
+        self.assertEqual(reloaded.journal['schedules'][0]['cancellation'], 'cancelled')
+        self.assertEqual(Notification.objects.get(user=owner, object_id=job.pk).event_type, JOB_FAILED)
+
+    def test_cancelling_a_retried_run_reissues_its_notice_once(self):
+        owner = get_user_model().objects.create_user(username='owner')
+        job = self.legacy_job(task_kwargs={})
+        Job.objects.filter(pk=job.pk).update(user=owner)
+        # A retried Job is waiting again and keeps the notice its first attempt sent.
+        old = timezone.now() - timedelta(days=1)
+        earlier = Notification.objects.create(user=owner, object=job, event_type=JOB_ERRORED, read=old)
+        Notification.objects.filter(pk=earlier.pk).update(created=old, object_repr='stale')
+        original = cutover._fail_closed
+
+        def interrupt(job, error):
+            original(job, error)
+            raise RuntimeError('stopped before the outcome was recorded')
+
+        with mock.patch.object(cutover, '_fail_closed', interrupt), self.assertRaises(RuntimeError):
+            cutover.enter_cutover(self.migration)
+
+        job.refresh_from_db()
+        notice = Notification.objects.get(user=owner, object_id=job.pk)
+        self.assertEqual(job.status, JobStatusChoices.STATUS_FAILED)
+        self.assertEqual(notice.pk, earlier.pk)
+        self.assertEqual(
+            (notice.event_type, notice.object_repr, notice.read, notice.created),
+            (JOB_FAILED, Notification.get_object_repr(job), None, job.completed),
+        )
+        # Resuming must preserve the notification's timestamp and read state.
+        Notification.objects.filter(pk=notice.pk).update(read=timezone.now())
+        notice.refresh_from_db()
+        reloaded = MigrationRun.objects.get(pk=self.migration.pk)
+
+        cutover.enter_cutover(reloaded)
+
+        self.assertEqual(reloaded.journal['schedules'][0]['cancellation'], 'cancelled')
+        self.assertEqual(
+            Notification.objects.values_list('pk', 'created', 'read').get(user=owner, object_id=job.pk),
+            (notice.pk, notice.created, notice.read),
+        )
+
     def test_a_capture_replayed_before_any_close_records_nothing_twice(self):
         # A crash before the closures leaves every job enqueued, so a second capture sees them all.
         job = self.legacy_job(task_kwargs={})

scripts/check_netbox_internals.py

@@ -97,7 +97,11 @@
     ),
     ('users.models.ObjectPermission.enabled', 'users.models:ObjectPermission.enabled'),
     ('extras.models.EventRule.enabled', 'extras.models:EventRule.enabled'),
-    ('core.models.Job.terminate', 'core.models:Job.terminate'),
+    (
+        'core.models.Job.terminate and extras.models.Notification.get_object_repr',
+        'core.models:Job.terminate',
+        'extras.models:Notification.get_object_repr',
+    ),
     (
         'django_rq.get_queue and rq.job.Job.fetch / .delete / .exists',
         'django_rq:get_queue',