fix(jobs): a cancel reaches libvirt for a job that recovery watches

The worker sends cancels only to jobs its own threads run. After a
worker restart, a migration or live clone that libvirt still ran was
watched by recovery; a cancel then moved the job to cancelling, the page
said so, but nothing told libvirt, and the job ran to its end. Recovery
now sends the cancel once while libvirt still runs the job, and a
migration it aborted ends cancelled rather than failed.

instances/jobs.py

@@ -1076,6 +1076,8 @@ class MigrateInstance(Kind):
             return Outcome(Operation.SUCCEEDED, result=_moved(op, dest))
         if source is not None and dest is None:
             _shut_off_again(op, source)
+            if op.cancel_requested:
+                return Outcome(Operation.CANCELLED)
             return Outcome(Operation.FAILED, error=say(_("The VM stayed on the source host")))
         return Outcome(Operation.UNCERTAIN, error=say(_(
             "The VM is on both computes or on neither; check them, then resolve this job.")))

instances/tests/test_migrate_job.py

@@ -70,6 +70,10 @@ class MigrateJobTestCase(TransactionTestCase):
         dom.isActive.return_value = False
         with patch("instances.jobs._domain", side_effect=[dom, None]):  # source, destination
             self.assertEqual(MigrateInstance().reconcile(op).state, Operation.FAILED)
+        op.cancel_requested = True  # recovery sent the cancel: libvirt aborted it
+        with patch("instances.jobs._domain", side_effect=[dom, None]):
+            self.assertEqual(MigrateInstance().reconcile(op).state, Operation.CANCELLED)
+        op.cancel_requested = False
         dom.isActive.return_value = True
         dom.jobInfo.return_value = [2] + [0] * 11  # VIR_DOMAIN_JOB_UNBOUNDED
         with patch("instances.jobs._domain", side_effect=[dom, dom]):

operations/tests/test_recovery.py

@@ -184,6 +184,22 @@ class RecoveryTestCase(TransactionTestCase):
         self.run_at(1100, Outcome(Operation.RUNNING))
         self.assertNotIn(op.pk, self.scheduler.failing_since)
 
+    def test_a_cancel_of_a_job_libvirt_still_runs_goes_to_libvirt_once(self):
+        op = self.orphan({})
+        Operation.objects.filter(pk=op.pk).update(state=Operation.CANCELLING, cancel_requested=True)
+        with patch.object(fakes.ScriptedKind, "cancel") as cancel:
+            self.run_at(1000, Outcome(Operation.RUNNING))
+            self.scheduler.next_try[op.pk] = 0
+            self.run_at(1015, Outcome(Operation.RUNNING))
+        cancel.assert_called_once()
+        self.assertEqual(cancel.call_args.args[0].pk, op.pk)
+
+    def test_a_job_without_a_cancel_is_watched_without_one(self):
+        self.orphan({})
+        with patch.object(fakes.ScriptedKind, "cancel") as cancel:
+            self.run_at(1000, Outcome(Operation.RUNNING))
+        cancel.assert_not_called()
+
     def test_a_job_that_never_recorded_its_intent_fails(self):
         op = self.orphan({}, effects=False)
         self.run_due()

operations/worker.py

@@ -391,6 +391,7 @@ class Scheduler:
         self.errors = {}
         self.stalled_since = {}
         self.failing_since = {}
+        self.cancel_sent = set()
         self.inflight = set()
 
     def schedule_orphans(self):
@@ -415,6 +416,7 @@ class Scheduler:
     def _done(self, op_id):
         for table in (self.next_try, self.attempts, self.errors, self.stalled_since, self.failing_since):
             table.pop(op_id, None)
+        self.cancel_sent.discard(op_id)
 
     def _take_over(self, op):
         """True if op is now this worker's. Another host's job ends uncertain
@@ -479,6 +481,10 @@ class Scheduler:
                 return
             self.failing_since.pop(op_id, None)
             if outcome.state == Operation.RUNNING:
+                # libvirt still runs it: a cancel asked meanwhile goes to libvirt once, as for a job thread's
+                if op.cancel_requested and op_id not in self.cancel_sent:
+                    self.cancel_sent.add(op_id)
+                    _guarded(jobs.get_kind(op.kind).cancel, op)
                 now = time.monotonic()
                 if not outcome.stalled:
                     self.stalled_since.pop(op_id, None)