lib/sync/outgoing, sources/ldap: fix sync status endpoint (#9855)

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

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)

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,
         )

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):

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(

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]]:

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: