Fix downgrade when normalized down revisions have overlap via depends_on

Fixed bug in versioning model where a downgrade across a revision with two
down revisions with one down revision depending on the other, would produce
an erroneous state in the alembic_version table, making upgrades impossible
without manually repairing the table.  Thanks much to Saif Hakim for
the great work on this.

<!-- Provide a general summary of your proposed changes in the Title field above -->

When the alembic tree has a migration (a1), with a branched migration (b1) that `depends_on` that migration, followed by a merge migration that merges (a1) and (b1), running the merge migrations downgrade incorrectly updates the heads to [a1, b1], when it should actually just have [b1]. This then prevents a user from running the upgrade again due to the confusing error:
> Requested revision b1 overlaps with other requested revisions a1

The problem occurs in `HeadMaintainer.update_to_step` which will update the value of heads by calling out into a helper method based on the scenario: deleting branches, creating branches, merging branches, unmerging branches, or the typical non-branched migration. As it turns out, all of these methods have logic to determine the canonical set of heads that should be written, _except_ in the case we are unmerging, resulting in the redundant head.

To fix, we simply remove any ancestors of the target heads from the list of target heads when doing an unmerge.

Fixes: #1373
Closes: #1376
Pull-request: sqlalchemy/alembic#1376
Pull-request-sha: dc8c7f8f7f8bc8e753aac8b8a1d6d70d79b12573

Change-Id: I7e1b5a969ea4487001606b20d3853f7c83015706

alembic/runtime/migration.py

@@ -1168,7 +1168,18 @@ def _unmerge_to_revisions(self, heads: Set[str]) -> Tuple[str, ...]:
             }
             return tuple(set(self.to_revisions).difference(ancestors))
         else:
-            return self.to_revisions
+            # for each revision we plan to return, compute its ancestors
+            # (excluding self), and remove those from the final output since
+            # they are already accounted for.
+            ancestors = {
+                r.revision
+                for to_revision in self.to_revisions
+                for r in self.revision_map._get_ancestor_nodes(
+                    self.revision_map.get_revisions(to_revision), check=False
+                )
+                if r.revision != to_revision
+            }
+            return tuple(set(self.to_revisions).difference(ancestors))
 
     def unmerge_branch_idents(
         self, heads: Set[str]

docs/build/unreleased/1373.rst

@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, versioning
+    :tickets: 1373
+
+    Fixed bug in versioning model where a downgrade across a revision with two
+    down revisions with one down revision depending on the other, would produce
+    an erroneous state in the alembic_version table, making upgrades impossible
+    without manually repairing the table.  Thanks much to Saif Hakim for
+    the great work on this.

tests/test_version_traversal.py

@@ -1176,6 +1176,67 @@ def test_dependencies_are_normalized(self):
         )
 
 
+class DependsOnBranchTestFive(MigrationTest):
+    @classmethod
+    def setup_class(cls):
+        """
+        issue #1373
+
+        Structure::
+
+            <base> -> a1 ------+
+                      ^        |
+                      |        +-> bmerge
+                      |        |
+                      +-- b1 --+
+        """
+        cls.env = env = staging_env()
+        cls.a1 = env.generate_revision("a1", "->a1")
+        cls.b1 = env.generate_revision(
+            "b1", "->b1", head="base", depends_on="a1"
+        )
+        cls.bmerge = env.generate_revision(
+            "bmerge", "bmerge", head=[cls.a1.revision, cls.b1.revision]
+        )
+
+    @classmethod
+    def teardown_class(cls):
+        clear_staging_env()
+
+    def test_downgrade_to_depends_on(self):
+        # Upgrade from a1 to b1 just has heads={"b1"}.
+        self._assert_upgrade(
+            self.b1.revision,
+            self.a1.revision,
+            [self.up_(self.b1)],
+            {self.b1.revision},
+        )
+
+        # Upgrade from b1 to bmerge just has {"bmerge"}.
+        self._assert_upgrade(
+            self.bmerge.revision,
+            self.b1.revision,
+            [self.up_(self.bmerge)],
+            {self.bmerge.revision},
+        )
+
+        # Downgrading from bmerge to a1 should return back to heads={"b1"}.
+        self._assert_downgrade(
+            self.a1.revision,
+            self.bmerge.revision,
+            [self.down_(self.bmerge)],
+            {self.b1.revision},
+        )
+
+        # Downgrading from bmerge to b1 also returns back to heads={"b1"}.
+        self._assert_downgrade(
+            self.b1.revision,
+            self.bmerge.revision,
+            [self.down_(self.bmerge)],
+            {self.b1.revision},
+        )
+
+
 class DependsOnBranchLabelTest(MigrationTest):
     @classmethod
     def setup_class(cls):