Re-enqueue changes if dequeued missing deps

When users create dependency cycles, the process often takes multiple
steps, some of which cause changes enqueued in earlier steps to be
dequeued.  Users then need to re-enqueue those changes in order to
have all changes in the cycle tested.

This change attempts to improve this by detecting that situation and
re-enqueing changes that are being de-queued because of missing deps.
Note, thare are plenty of cases where we de-queue because of missing
deps and we don't want to re-enqueue (ie, a one-way dependent change
ahead has failed).  To restrict this to only the situation we're
interested in, we only act if the dependencies are already in the
pipeline.  A recently updated change in a cycle will have just been
added to the pipeline, so this is true in the case we're interested in.
A one-way dependent change that failed will have already been removed
from the pipeline, and so this will be false in cases in which we
are not interested.

Change-Id: I84b3b2f8fffd1c946dafa605d1c17a37131558bd

2 files changed, 36 insertions, 17 deletions

diff --git a/tests/unit/test_circular_dependencies.py b/tests/unit/test_circular_dependencies.py
index f38e55001..a3f9dda33 100644
--- a/tests/unit/test_circular_dependencies.py
+++ b/tests/unit/test_circular_dependencies.py
@@ -2368,13 +2368,10 @@ class TestGerritCircularDependencies(ZuulTestCase):
         self.executor_server.release()
         self.waitUntilSettled()
 
-        # A quirk: at the end of this process, the first change in
-        # Gerrit has a complete run because the process of updating it
-        # involves a new patchset that is enqueued.  Compare to the
-        # same test in GitHub.
         self.assertHistory([
             dict(name="project-job", result="ABORTED", changes="1,1"),
             dict(name="project-job", result="ABORTED", changes="1,1 2,1"),
+            dict(name="project-job", result="SUCCESS", changes="1,2 2,1"),
             dict(name="project-job", result="SUCCESS", changes="2,1 1,2"),
         ], ordered=False)
 
@@ -2404,12 +2401,9 @@ class TestGerritCircularDependencies(ZuulTestCase):
         self.executor_server.release()
         self.waitUntilSettled()
 
-        # A quirk: at the end of this process, the second change in
-        # Gerrit has a complete run because only at that point is the
-        # topic complete; the first is aborted once the second is
-        # uploaded.
         self.assertHistory([
             dict(name="check-job", result="ABORTED", changes="1,1"),
+            dict(name="check-job", result="SUCCESS", changes="2,1 1,1"),
             dict(name="check-job", result="SUCCESS", changes="1,1 2,1"),
         ], ordered=False)
 
@@ -2682,13 +2676,11 @@ class TestGithubCircularDependencies(ZuulTestCase):
         self.executor_server.release()
         self.waitUntilSettled()
 
-        # A quirk: at the end of this process, the second PR in GitHub
-        # has a complete run because the process of updating the first
-        # change is not disruptive to the second.  Compare to the same
-        # test in Gerrit.
         self.assertHistory([
             dict(name="project-job", result="ABORTED",
                  changes=f"{A.number},{A.head_sha}"),
             dict(name="project-job", result="SUCCESS",
                  changes=f"{A.number},{A.head_sha} {B.number},{B.head_sha}"),
+            dict(name="project-job", result="SUCCESS",
+                 changes=f"{B.number},{B.head_sha} {A.number},{A.head_sha}"),
         ], ordered=False)
diff --git a/zuul/manager/__init__.py b/zuul/manager/__init__.py
index 430fcc320..bf49737dd 100644
--- a/zuul/manager/__init__.py
+++ b/zuul/manager/__init__.py
@@ -496,7 +496,8 @@ class PipelineManager(metaclass=ABCMeta):
 
     def addChange(self, change, event, quiet=False, enqueue_time=None,
                   ignore_requirements=False, live=True,
-                  change_queue=None, history=None, dependency_graph=None):
+                  change_queue=None, history=None, dependency_graph=None,
+                  skip_presence_check=False):
         log = get_annotated_logger(self.log, event)
         log.debug("Considering adding change %s" % change)
 
@@ -511,7 +512,9 @@ class PipelineManager(metaclass=ABCMeta):
         # If we are adding a live change, check if it's a live item
         # anywhere in the pipeline.  Otherwise, we will perform the
         # duplicate check below on the specific change_queue.
-        if live and self.isChangeAlreadyInPipeline(change):
+        if (live and
+            self.isChangeAlreadyInPipeline(change) and
+            not skip_presence_check):
             log.debug("Change %s is already in pipeline, ignoring" % change)
             return True
 
@@ -570,8 +573,10 @@ class PipelineManager(metaclass=ABCMeta):
             log.debug("History after enqueuing changes ahead: %s", history)
 
             if self.isChangeAlreadyInQueue(change, change_queue):
-                log.debug("Change %s is already in queue, ignoring" % change)
-                return True
+                if not skip_presence_check:
+                    log.debug("Change %s is already in queue, ignoring",
+                              change)
+                    return True
 
             cycle = []
             if isinstance(change, model.Change):
@@ -1534,6 +1539,7 @@ class PipelineManager(metaclass=ABCMeta):
             log.info("Dequeuing change %s because "
                      "it can no longer merge" % item.change)
             self.cancelJobs(item)
+            quiet_dequeue = False
             if item.isBundleFailing():
                 item.setDequeuedBundleFailing('Bundle is failing')
             elif not meets_reqs:
@@ -1545,7 +1551,28 @@ class PipelineManager(metaclass=ABCMeta):
                 else:
                     msg = f'Change {clist} is needed.'
                 item.setDequeuedNeedingChange(msg)
-            if item.live:
+                # If all the dependencies are already in the pipeline
+                # (but not ahead of this change), then we probably
+                # just added updated versions of them, possibly
+                # updating a cycle.  In that case, attempt to
+                # re-enqueue this change with the updated deps.
+                if (item.live and
+                    all([self.isChangeAlreadyInPipeline(c)
+                         for c in needs_changes])):
+                    # Try enqueue, if that succeeds, keep this dequeue quiet
+                    try:
+                        log.info("Attempting re-enqueue of change %s",
+                                 item.change)
+                        quiet_dequeue = self.addChange(
+                            item.change, item.event,
+                            enqueue_time=item.enqueue_time,
+                            quiet=True,
+                            skip_presence_check=True)
+                    except Exception:
+                        log.exception("Unable to re-enqueue change %s "
+                                      "which is missing dependencies",
+                                      item.change)
+            if item.live and not quiet_dequeue:
                 try:
                     self.reportItem(item)
                 except exceptions.MergeFailure: