diff --git a/authentik/lib/sync/outgoing/__init__.py b/authentik/lib/sync/outgoing/__init__.py index 148a20f5bd..1005a6b242 100644 --- a/authentik/lib/sync/outgoing/__init__.py +++ b/authentik/lib/sync/outgoing/__init__.py @@ -3,4 +3,3 @@ PAGE_SIZE = 100 PAGE_TIMEOUT = 60 * 60 * 0.5 # Half an hour HTTP_CONFLICT = 409 -LOCK_ACQUIRE_TIMEOUT = 5 diff --git a/authentik/lib/sync/outgoing/api.py b/authentik/lib/sync/outgoing/api.py index df9809048c..03f31ecc5b 100644 --- a/authentik/lib/sync/outgoing/api.py +++ b/authentik/lib/sync/outgoing/api.py @@ -47,8 +47,10 @@ class OutgoingSyncProviderStatusMixin: uid=slugify(provider.name), ) ) - status = { - "tasks": tasks, - "is_running": provider.sync_lock.locked(), - } + with provider.sync_lock as lock_acquired: + status = { + "tasks": tasks, + # If we could not acquire the lock, it means a task is using it, and thus is running + "is_running": not lock_acquired, + } return Response(SyncStatusSerializer(status).data) diff --git a/authentik/lib/sync/outgoing/models.py b/authentik/lib/sync/outgoing/models.py index f792e11ab0..e46fb529f7 100644 --- a/authentik/lib/sync/outgoing/models.py +++ b/authentik/lib/sync/outgoing/models.py @@ -5,7 +5,6 @@ from django.db import connection from django.db.models import Model, QuerySet, TextChoices from authentik.core.models import Group, User -from authentik.lib.sync.outgoing import LOCK_ACQUIRE_TIMEOUT from authentik.lib.sync.outgoing.base import BaseOutgoingSyncClient @@ -36,6 +35,6 @@ class OutgoingSyncProvider(Model): """Postgres lock for syncing SCIM to prevent multiple parallel syncs happening""" return pglock.advisory( lock_id=f"goauthentik.io/{connection.schema_name}/providers/outgoing-sync/{str(self.pk)}", - timeout=LOCK_ACQUIRE_TIMEOUT, - side_effect=pglock.Raise, + timeout=0, + side_effect=pglock.Return, ) diff --git a/authentik/lib/sync/outgoing/tasks.py b/authentik/lib/sync/outgoing/tasks.py index 735891806a..80f26c4883 100644 --- a/authentik/lib/sync/outgoing/tasks.py +++ b/authentik/lib/sync/outgoing/tasks.py @@ -5,7 +5,6 @@ from celery.exceptions import Retry from celery.result import allow_join_result from django.core.paginator import Paginator from django.db.models import Model, QuerySet -from django.db.utils import OperationalError from django.utils.text import slugify from django.utils.translation import gettext_lazy as _ from structlog.stdlib import BoundLogger, get_logger @@ -72,34 +71,33 @@ class SyncTasks: self.logger.debug("Starting provider sync") users_paginator = Paginator(provider.get_object_qs(User), PAGE_SIZE) groups_paginator = Paginator(provider.get_object_qs(Group), PAGE_SIZE) - try: - with allow_join_result(), provider.sync_lock: - try: - for page in users_paginator.page_range: - messages.append(_("Syncing page %(page)d of users" % {"page": page})) - for msg in sync_objects.apply_async( - args=(class_to_path(User), page, provider_pk), - time_limit=PAGE_TIMEOUT, - soft_time_limit=PAGE_TIMEOUT, - ).get(): - messages.append(LogEvent(**msg)) - for page in groups_paginator.page_range: - messages.append(_("Syncing page %(page)d of groups" % {"page": page})) - for msg in sync_objects.apply_async( - args=(class_to_path(Group), page, provider_pk), - time_limit=PAGE_TIMEOUT, - soft_time_limit=PAGE_TIMEOUT, - ).get(): - messages.append(LogEvent(**msg)) - except TransientSyncException as exc: - self.logger.warning("transient sync exception", exc=exc) - raise task.retry(exc=exc) from exc - except StopSync as exc: - task.set_error(exc) - return - except OperationalError: - self.logger.debug("Failed to acquire sync lock, skipping", provider=provider.name) - return + with allow_join_result(), provider.sync_lock as lock_acquired: + if not lock_acquired: + self.logger.debug("Failed to acquire sync lock, skipping", provider=provider.name) + return + try: + for page in users_paginator.page_range: + messages.append(_("Syncing page %(page)d of users" % {"page": page})) + for msg in sync_objects.apply_async( + args=(class_to_path(User), page, provider_pk), + time_limit=PAGE_TIMEOUT, + soft_time_limit=PAGE_TIMEOUT, + ).get(): + messages.append(LogEvent(**msg)) + for page in groups_paginator.page_range: + messages.append(_("Syncing page %(page)d of groups" % {"page": page})) + for msg in sync_objects.apply_async( + args=(class_to_path(Group), page, provider_pk), + time_limit=PAGE_TIMEOUT, + soft_time_limit=PAGE_TIMEOUT, + ).get(): + messages.append(LogEvent(**msg)) + except TransientSyncException as exc: + self.logger.warning("transient sync exception", exc=exc) + raise task.retry(exc=exc) from exc + except StopSync as exc: + task.set_error(exc) + return task.set_status(TaskStatus.SUCCESSFUL, *messages) def sync_objects(self, object_type: str, page: int, provider_pk: int): diff --git a/authentik/sources/ldap/api.py b/authentik/sources/ldap/api.py index 2046337eee..faaca0a533 100644 --- a/authentik/sources/ldap/api.py +++ b/authentik/sources/ldap/api.py @@ -143,10 +143,12 @@ class LDAPSourceViewSet(UsedByMixin, ModelViewSet): uid__startswith=source.slug, ) ) - status = { - "tasks": tasks, - "is_running": source.sync_lock.locked(), - } + with source.sync_lock as lock_acquired: + status = { + "tasks": tasks, + # If we could not acquire the lock, it means a task is using it, and thus is running + "is_running": not lock_acquired, + } return Response(SyncStatusSerializer(status).data) @extend_schema( diff --git a/authentik/sources/ldap/models.py b/authentik/sources/ldap/models.py index 23e7dc2777..7752276d52 100644 --- a/authentik/sources/ldap/models.py +++ b/authentik/sources/ldap/models.py @@ -18,7 +18,6 @@ from authentik.core.models import Group, PropertyMapping, Source from authentik.crypto.models import CertificateKeyPair from authentik.lib.config import CONFIG from authentik.lib.models import DomainlessURLValidator -from authentik.lib.sync.outgoing import LOCK_ACQUIRE_TIMEOUT LDAP_TIMEOUT = 15 @@ -213,8 +212,8 @@ class LDAPSource(Source): """Postgres lock for syncing LDAP to prevent multiple parallel syncs happening""" return pglock.advisory( lock_id=f"goauthentik.io/{connection.schema_name}/sources/ldap/sync/{self.slug}", - timeout=LOCK_ACQUIRE_TIMEOUT, - side_effect=pglock.Raise, + timeout=0, + side_effect=pglock.Return, ) def check_connection(self) -> dict[str, dict[str, str]]: diff --git a/authentik/sources/ldap/tasks.py b/authentik/sources/ldap/tasks.py index 540f10cbaf..2f0547a6ab 100644 --- a/authentik/sources/ldap/tasks.py +++ b/authentik/sources/ldap/tasks.py @@ -4,7 +4,6 @@ from uuid import uuid4 from celery import chain, group from django.core.cache import cache -from django.db.utils import OperationalError from ldap3.core.exceptions import LDAPException from structlog.stdlib import get_logger @@ -64,24 +63,24 @@ def ldap_sync_single(source_pk: str): source: LDAPSource = LDAPSource.objects.filter(pk=source_pk).first() if not source: return - try: - with source.sync_lock: - # Delete all sync tasks from the cache - DBSystemTask.objects.filter(name="ldap_sync", uid__startswith=source.slug).delete() - task = chain( - # User and group sync can happen at once, they have no dependencies on each other - group( - ldap_sync_paginator(source, UserLDAPSynchronizer) - + ldap_sync_paginator(source, GroupLDAPSynchronizer), - ), - # Membership sync needs to run afterwards - group( - ldap_sync_paginator(source, MembershipLDAPSynchronizer), - ), - ) - task() - except OperationalError: - LOGGER.debug("Failed to acquire lock for LDAP sync, skipping task", source=source.slug) + with source.sync_lock as lock_acquired: + if not lock_acquired: + LOGGER.debug("Failed to acquire lock for LDAP sync, skipping task", source=source.slug) + return + # Delete all sync tasks from the cache + DBSystemTask.objects.filter(name="ldap_sync", uid__startswith=source.slug).delete() + task = chain( + # User and group sync can happen at once, they have no dependencies on each other + group( + ldap_sync_paginator(source, UserLDAPSynchronizer) + + ldap_sync_paginator(source, GroupLDAPSynchronizer), + ), + # Membership sync needs to run afterwards + group( + ldap_sync_paginator(source, MembershipLDAPSynchronizer), + ), + ) + task() def ldap_sync_paginator(source: LDAPSource, sync: type[BaseLDAPSynchronizer]) -> list: