def sync_slot(self, cur: Union['cursor', 'Cursor[Any]'], database: str, slot: str, lsn: int) -> None:
        """Execute a ``pg_replication_slot_advance`` query and store success for scheduled synchronisation task.

        :param cur: database connection cursor.
        :param database: name of the database associated with the slot.
        :param slot: name of the slot to be synchronised.
        :param lsn: last known LSN position
        """
        failed = copy = False
        try:
            cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", (slot, format_lsn(lsn)))
        except Exception as e:
            logger.error("Failed to advance logical replication slot '%s': %r", slot, e)
            failed = True
            # WAL file is gone or slot is invalidated
            copy = isinstance(e, OperationalError) and e.diag.sqlstate in ('58P01', '55000')
        with self._condition:
            if self._scheduled and failed:
                if copy and slot not in self._copy_slots:
                    self._copy_slots.append(slot)
                self._failed = True

            new_lsn = self._scheduled.get(database, {}).get(slot, 0)
            # remove slot from the self._scheduled structure if it is to be copied or if it wasn't changed
            if copy or (new_lsn == lsn and database in self._scheduled):
                self._scheduled[database].pop(slot)
                if not self._scheduled[database]:
                    self._scheduled.pop(database)
Execute a ``pg_replication_slot_advance`` query and store success for scheduled synchronisation task.

        :param cur: database connection cursor.
        :param database: name of the database associated with the slot.
        :param slot: name of the slot to be synchronised.
        :param lsn: last known LSN position
        
patroni/postgresql/slots.py