Repeated migration application

[ojomio]

HI! I caught an error on the test_up_down_consistency when tried to supply custom before_revision_data

It actually failed inserting data because 'it already exists':

E   psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "pk_epic"
E   DETAIL:  Key (id)=(db790059-f5ef-4f39-b137-5da2d527b435) already exists.

As it turned out, alembic tests inserted it twice

---

To debug this issue I added some printing to the managed_upgrade function of a Runner:

    def managed_upgrade(self, dest_revision, *, current=None, return_current=True):
        """Perform an upgrade one migration at a time, inserting static data at the given points."""
        if current is None:
            current = self.current
        print(self.history.revision_window(current, dest_revision)) # here
        for current_revision, next_revision in self.history.revision_window(current, dest_revision):
            before_upgrade_data = self.revision_data.get_before(next_revision)
            print(f'INSERT {before_upgrade_data} for rev {next_revision}') # here
            self.insert_into(data=before_upgrade_data, revision=current_revision, table=None)

            if next_revision in (self.config.skip_revisions or {}):
                self.set_revision(next_revision)
            else:
                self.command_executor.upgrade(next_revision)

            at_upgrade_data = self.revision_data.get_at(next_revision)
            self.insert_into(data=at_upgrade_data, revision=next_revision, table=None)
            print(f'current is {self.current}') # and here

        if return_current:
            current = self.current
            return current
        return None

and to alembic/runtime/migration.py:

    def run_migrations(self, **kw: Any) -> None:
       self.impl.start_migrations()

       heads: Tuple[str, ...]
       if self.purge:
           if self.as_sql:
               raise util.CommandError("Can't use --purge with --sql mode")
           self._ensure_version_table(purge=True)
           heads = ()
       else:
           heads = self.get_current_heads()
           print(f'Current heads from DB: {heads}')
       ...

What I read was a clear evidence, that script tried to INSERT the same before_upgrade_data twice, both times for the same revision.

The revision script added data to was one branch of the later merged migrations tree. The same picture occurred for every branching/merging point, even without before_upgrade_data, it just didnt show up or cause any error

For example, I have the following dag in my migration history:

cd039 <- 814 <-- c04af <--c04af
      <-c915 <-- db88  <|

And it produced the log:

Current heads from DB: ('cd039ce4087c',)
[('cd039ce4087c', 'c915a959a921')]
INSERT [] for rev c915a959a921
Current heads from DB: ('cd039ce4087c',)
Current heads from DB: ('c915a959a921',)
current is c915a959a921
->c915a959a921
Current heads from DB: ('c915a959a921',)
[('c915a959a921', 'db88f0ab406b')]
INSERT [] for rev db88f0ab406b
Current heads from DB: ('c915a959a921',)
Current heads from DB: ('db88f0ab406b',)
current is db88f0ab406b
->db88f0ab406b
Current heads from DB: ('db88f0ab406b',)
[('db88f0ab406b', '814a65c9b146')]
INSERT [] for rev 814a65c9b146
Current heads from DB: ('db88f0ab406b',)
Current heads from DB: ('db88f0ab406b', '814a65c9b146')
current is db88f0ab406b >>DOESNT MOVE<<
->814a65c9b146
Current heads from DB: ('db88f0ab406b', '814a65c9b146')
[('db88f0ab406b', '814a65c9b146'), ('814a65c9b146', 'c04afaa3eff5')] >> REPEATED MIGRATION <<
INSERT [] for rev 814a65c9b146
Current heads from DB: ('db88f0ab406b', '814a65c9b146')
Current heads from DB: ('db88f0ab406b', '814a65c9b146')
current is db88f0ab406b
INSERT [] for rev c04afaa3eff5
Current heads from DB: ('db88f0ab406b', '814a65c9b146')
Current heads from DB: ('c04afaa3eff5',)
current is c04afaa3eff5
->c04afaa3eff5

Clearly, self.current_head was not moving when transferring from branch to branch, and I guess I know why: Because it always takes the first element from the versions table!

    @property
    def current(self) -> str:
        """Get the list of revision heads."""
        current = "base"

        def get_current(rev, _):
            nonlocal current
            if rev:
                current = rev[0]

            return []

        self.command_executor.execute_fn(get_current)

        if current:
            return current
            
        return "base"

Do you know about this bug? Are there any plans to fix it?