Description of problem:
Capsule sync failed with the following errors when syncing a number of kickstart repositories sharing the same sub repositories concurrently.

--------------------------------------------------------------
duplicate key value violates unique constraint "core_repositoryversion_repository_id_number_3c54ce50_uniq"
DETAIL:  Key (repository_id, number)=(0b5936ef-2971-459f-90e2-9d1a1d837292, 1) already exists.
Could not lookup a publication_href for repo 176deadlock detected

DETAIL:  Process 7734 waits for ShareLock on transaction 45647574; blocked by process 7737.
Process 7737 waits for ShareLock on transaction 45647572; blocked by process 7734.
HINT:  See server log for query details.
CONTEXT:  while inserting index tuple (1067,1) in relation "core_artifact_sha256_key"
Could not lookup a publication_href for repo 177

insert or update on table "core_repositorycontent" violates foreign key constraint "core_repositoryconte_version_added_id_d5113f18_fk_core_repo"
DETAIL:  Key (version_added_id)=(5d344d1d-76c9-476c-980a-11a240881389) is not present in table "core_repositoryversion".
Could not lookup a publication_href for repo 181

insert or update on table "core_repositorycontent" violates foreign key constraint "core_repositoryconte_version_added_id_d5113f18_fk_core_repo"
DETAIL:  Key (version_added_id)=(18077abc-2cba-40db-a36e-f4dda3142057) is not present in table "core_repositoryversion".
Could not lookup a publication_href for repo 184

deadlock detected
DETAIL:  Process 7766 waits for ShareLock on transaction 45647572; blocked by process 7734.
Process 7734 waits for ShareLock on transaction 45647574; blocked by process 7737.
Process 7737 waits for ShareLock on transaction 45647571; blocked by process 7766.
HINT:  See server log for query details.
CONTEXT:  while inserting index tuple (1066,3) in relation "core_artifact_sha256_key"
Could not lookup a publication_href for repo 186
--------------------------------------------------------------


Error details and tracebacks:

Error #1:
---------------------------------------------------------------
pulp_tasks:
- pulp_href: "/pulp/api/v3/tasks/93bde0ef-cc33-434f-9922-4dcacd482039/"
  pulp_created: '2022-04-21T07:01:30.908+00:00'
  state: failed
  name: pulp_rpm.app.tasks.synchronizing.synchronize
  logging_cid: dcd8227c-bfa0-4400-8d97-749f5f2b7fcd
  started_at: '2022-04-21T07:01:31.114+00:00'
  finished_at: '2022-04-21T07:01:32.726+00:00'
  error:
    traceback: |2
        File "/usr/lib/python3.6/site-packages/pulpcore/tasking/pulpcore_worker.py", line 339, in _perform_task
          result = func(*args, **kwargs)
        File "/usr/lib/python3.6/site-packages/pulp_rpm/app/tasks/synchronizing.py", line 474, in synchronize
          subrepo_version = dv.create()
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/declarative_version.py", line 145, in create
          with self.repository.new_version() as new_version:
        File "/usr/lib/python3.6/site-packages/pulp_rpm/app/models/repository.py", line 258, in new_version
          version.save()
        File "/usr/lib/python3.6/site-packages/django_lifecycle/mixins.py", line 134, in save
          save(*args, **kwargs)
        File "/usr/lib/python3.6/site-packages/django/db/models/base.py", line 744, in save
          force_update=force_update, update_fields=update_fields)
        File "/usr/lib/python3.6/site-packages/django/db/models/base.py", line 782, in save_base
          force_update, using, update_fields,
        File "/usr/lib/python3.6/site-packages/django/db/models/base.py", line 873, in _save_table
          result = self._do_insert(cls._base_manager, using, fields, update_pk, raw)
        File "/usr/lib/python3.6/site-packages/django/db/models/base.py", line 911, in _do_insert
          using=using, raw=raw)
        File "/usr/lib/python3.6/site-packages/django/db/models/manager.py", line 82, in manager_method
          return getattr(self.get_queryset(), name)(*args, **kwargs)
        File "/usr/lib/python3.6/site-packages/django/db/models/query.py", line 1186, in _insert
          return query.get_compiler(using=using).execute_sql(return_id)
        File "/usr/lib/python3.6/site-packages/django/db/models/sql/compiler.py", line 1377, in execute_sql
          cursor.execute(sql, params)
        File "/usr/lib/python3.6/site-packages/django/db/backends/utils.py", line 67, in execute
          return self._execute_with_wrappers(sql, params, many=False, executor=self._execute)
        File "/usr/lib/python3.6/site-packages/django/db/backends/utils.py", line 76, in _execute_with_wrappers
          return executor(sql, params, many, context)
        File "/usr/lib/python3.6/site-packages/django/db/backends/utils.py", line 84, in _execute
          return self.cursor.execute(sql, params)
        File "/usr/lib/python3.6/site-packages/django/db/utils.py", line 89, in __exit__
          raise dj_exc_value.with_traceback(traceback) from exc_value
        File "/usr/lib/python3.6/site-packages/django/db/backends/utils.py", line 84, in _execute
          return self.cursor.execute(sql, params)
    description: |
      duplicate key value violates unique constraint "core_repositoryversion_repository_id_number_3c54ce50_uniq"
      DETAIL:  Key (repository_id, number)=(0b5936ef-2971-459f-90e2-9d1a1d837292, 1) already exists.
  worker: "/pulp/api/v3/workers/2482a7ef-daa8-4fd2-af66-3cadced8c991/"
---------------------------------------------------------------


# Error #2
--------------------------------------------------------------
pulp_tasks:
- pulp_href: "/pulp/api/v3/tasks/b6a1e67b-3d5e-4737-bc93-72172f72b97a/"
  pulp_created: '2022-04-21T07:01:30.782+00:00'
  state: failed
  name: pulp_rpm.app.tasks.synchronizing.synchronize
  logging_cid: dcd8227c-bfa0-4400-8d97-749f5f2b7fcd
  started_at: '2022-04-21T07:01:31.090+00:00'
  finished_at: '2022-04-21T07:02:13.061+00:00'
  error:
    traceback: |2
        File "/usr/lib/python3.6/site-packages/pulpcore/tasking/pulpcore_worker.py", line 339, in _perform_task
          result = func(*args, **kwargs)
        File "/usr/lib/python3.6/site-packages/pulp_rpm/app/tasks/synchronizing.py", line 474, in synchronize
          subrepo_version = dv.create()
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/declarative_version.py", line 151, in create
          loop.run_until_complete(pipeline)
        File "/usr/lib64/python3.6/asyncio/base_events.py", line 484, in run_until_complete
          return future.result()
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/api.py", line 225, in create_pipeline
          await asyncio.gather(*futures)
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/api.py", line 43, in __call__
          await self.run()
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/content_stages.py", line 287, in run
          self.new_version.add_content(Content.objects.filter(pk__in=to_add))
        File "/usr/lib/python3.6/site-packages/pulpcore/app/models/repository.py", line 764, in add_content
          RepositoryContent.objects.bulk_create(repo_content)
        File "/usr/lib/python3.6/site-packages/django/db/models/manager.py", line 82, in manager_method
          return getattr(self.get_queryset(), name)(*args, **kwargs)
        File "/usr/lib/python3.6/site-packages/django/db/models/query.py", line 480, in bulk_create
          obj_without_pk._state.db = self.db
        File "/usr/lib/python3.6/site-packages/django/db/transaction.py", line 240, in __exit__
          connection.commit()
        File "/usr/lib/python3.6/site-packages/django/db/backends/base/base.py", line 262, in commit
          self._commit()
        File "/usr/lib/python3.6/site-packages/django/db/backends/base/base.py", line 240, in _commit
          return self.connection.commit()
        File "/usr/lib/python3.6/site-packages/django/db/utils.py", line 89, in __exit__
          raise dj_exc_value.with_traceback(traceback) from exc_value
        File "/usr/lib/python3.6/site-packages/django/db/backends/base/base.py", line 240, in _commit
          return self.connection.commit()
    description: |
      insert or update on table "core_repositorycontent" violates foreign key constraint "core_repositoryconte_version_added_id_d5113f18_fk_core_repo"
      DETAIL:  Key (version_added_id)=(5d344d1d-76c9-476c-980a-11a240881389) is not present in table "core_repositoryversion".
---------------------------------------------------------------


Error #3: Should be fixed by bug #2062526
---------------------------------------------------------------
pulp_tasks:
- pulp_href: "/pulp/api/v3/tasks/eff5f426-baee-4334-8cfa-869e527ddf01/"
  pulp_created: '2022-04-21T07:01:30.792+00:00'
  state: failed
  name: pulp_rpm.app.tasks.synchronizing.synchronize
  logging_cid: dcd8227c-bfa0-4400-8d97-749f5f2b7fcd
  started_at: '2022-04-21T07:01:31.055+00:00'
  finished_at: '2022-04-21T07:02:11.478+00:00'
  error:
    traceback: |2
        File "/usr/lib/python3.6/site-packages/pulpcore/tasking/pulpcore_worker.py", line 339, in _perform_task
          result = func(*args, **kwargs)
        File "/usr/lib/python3.6/site-packages/pulp_rpm/app/tasks/synchronizing.py", line 474, in synchronize
          subrepo_version = dv.create()
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/declarative_version.py", line 151, in create
          loop.run_until_complete(pipeline)
        File "/usr/lib64/python3.6/asyncio/base_events.py", line 484, in run_until_complete
          return future.result()
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/api.py", line 225, in create_pipeline
          await asyncio.gather(*futures)
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/api.py", line 43, in __call__
          await self.run()
        File "/usr/lib/python3.6/site-packages/pulpcore/plugin/stages/artifact_stages.py", line 241, in run
          d_artifact.artifact for d_artifact in da_to_save
        File "/usr/lib/python3.6/site-packages/pulpcore/app/models/content.py", line 84, in bulk_get_or_create
          return super().bulk_create(objs, batch_size=batch_size)
        File "/usr/lib/python3.6/site-packages/django/db/models/manager.py", line 82, in manager_method
          return getattr(self.get_queryset(), name)(*args, **kwargs)
        File "/usr/lib/python3.6/site-packages/django/db/models/query.py", line 468, in bulk_create
          self._batched_insert(objs_with_pk, fields, batch_size, ignore_conflicts=ignore_conflicts)
        File "/usr/lib/python3.6/site-packages/django/db/models/query.py", line 1204, in _batched_insert
          ignore_conflicts=ignore_conflicts,
        File "/usr/lib/python3.6/site-packages/django/db/models/query.py", line 1186, in _insert
          return query.get_compiler(using=using).execute_sql(return_id)
        File "/usr/lib/python3.6/site-packages/django/db/models/sql/compiler.py", line 1377, in execute_sql
          cursor.execute(sql, params)
        File "/usr/lib/python3.6/site-packages/django/db/backends/utils.py", line 67, in execute
          return self._execute_with_wrappers(sql, params, many=False, executor=self._execute)
        File "/usr/lib/python3.6/site-packages/django/db/backends/utils.py", line 76, in _execute_with_wrappers
          return executor(sql, params, many, context)
        File "/usr/lib/python3.6/site-packages/django/db/backends/utils.py", line 84, in _execute
          return self.cursor.execute(sql, params)
        File "/usr/lib/python3.6/site-packages/django/db/utils.py", line 89, in __exit__
          raise dj_exc_value.with_traceback(traceback) from exc_value
        File "/usr/lib/python3.6/site-packages/django/db/backends/utils.py", line 84, in _execute
          return self.cursor.execute(sql, params)
    description: |
      deadlock detected
      DETAIL:  Process 7734 waits for ShareLock on transaction 45647574; blocked by process 7737.
      Process 7737 waits for ShareLock on transaction 45647572; blocked by process 7734.
      HINT:  See server log for query details.
      CONTEXT:  while inserting index tuple (1067,1) in relation "core_artifact_sha256_key"




How reproducible:
Easy


Steps to Reproduce:
1. Enable and sync a RHEL 6.10 kickstart repository.
2. Create a content view and attach the RHEL 6.10 kickstart repository
3. Clone 7 copied of the content view
4. Disable Capsule auto-sync setting
5. Publish and promote the content views to a lifecycle environment.
6. Trigger an optimized Capsule sync manually.

Actual results:
See errors above

Expected results:
No error

Additional info:
I think it is because multiple kickstart repositories shared the same sub repositories. When syncing those repositories concurrently, race conditions occur trying to update the same sub repositories.


# select pulp_id,  name, next_version, pulp_type, retained_versions from core_repository where name like '%e82fa8cdfb0341557859910c7f96773b9b307e760d83f06ba93432c783c70cb';
               pulp_id                |                                        name                                         | next_version | pulp_type | retained_versions 
--------------------------------------+-------------------------------------------------------------------------------------+--------------+-----------+-------------------
 b8a0d4ed-3e90-4f8b-a95b-45c436e4c6e5 | ResilientStorage-be82fa8cdfb0341557859910c7f96773b9b307e760d83f06ba93432c783c70cb   |            2 | rpm.rpm   |                  
 ddc720ef-0e59-468f-a64e-e6dd7fb210c2 | Server-be82fa8cdfb0341557859910c7f96773b9b307e760d83f06ba93432c783c70cb             |            2 | rpm.rpm   |                  
 ba50f4fe-3bd3-44cf-a55a-0b00da0bba70 | HighAvailability-be82fa8cdfb0341557859910c7f96773b9b307e760d83f06ba93432c783c70cb   |            2 | rpm.rpm   |                  
 aafac7c5-0fd6-4f54-a34e-9bba69b39259 | ScalableFileSystem-be82fa8cdfb0341557859910c7f96773b9b307e760d83f06ba93432c783c70cb |            2 | rpm.rpm   |                  
 ec35546c-eaa7-4577-a1bc-0eefc1f6bbf9 | LoadBalancer-be82fa8cdfb0341557859910c7f96773b9b307e760d83f06ba93432c783c70cb       |            2 | rpm.rpm   |                  
(5 rows)


In /usr/lib/python3.6/site-packages/pulp_rpm/app/tasks/synchronizing.py
---------------------------------------------------------------
...
    if treeinfo:
        treeinfo["repositories"] = {}
        for repodata in set(treeinfo["download"]["repodatas"]):
            if repodata == DIST_TREE_MAIN_REPO_PATH:
                treeinfo["repositories"].update({repodata: None})
                continue
            name = f"{repodata}-{treeinfo['hash']}"  <==========
            sub_repo, created = RpmRepository.objects.get_or_create(name=name, sub_repo=True)  <===========
            if created:
                sub_repo.save()
            directory = treeinfo["repo_map"][repodata]
            treeinfo["repositories"].update({directory: str(sub_repo.pk)})
            ...
---------------------------------------------------------------


My guess for the Error #2:

There is a code to delete an incomplete latest version before creating a new version so I think below is the possible scenario.

1) 2 tasks are processing the same sub repository.
2) Task 'A' creates a latest version and proceed but not complete yet.
3) Task 'B' deletes the version created by Task 'A' and creates another version.
4) Task 'A' tries to associate content to repo then fails.


In /usr/lib/python3.6/site-packages/pulpcore/app/models/repository.py
---------------------------------------------------------------
    def new_version(self, base_version=None):
        """
        Create a new RepositoryVersion for this Repository

        Creation of a RepositoryVersion should be done in a RQ Job.

        Args:
            repository (pulpcore.app.models.Repository): to create a new version of
            base_version (pulpcore.app.models.RepositoryVersion): an optional repository version
                whose content will be used as the set of content for the new version

        Returns:
            pulpcore.app.models.RepositoryVersion: The Created RepositoryVersion
        """
        with transaction.atomic():
            latest_version = self.versions.latest()
            if not latest_version.complete:  <=====================
                latest_version.delete()
---------------------------------------------------------------