fix(permissions): hold permission sync attempts with a Postgres lease (#15973)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

.agents/feature-map/components/permission-sync.md

@@ -238,6 +238,7 @@ beat: CHECK_FOR_DOC_PERMISSIONS_SYNC (every 30s)
       create_doc_permission_sync_attempt          onyx/db/permission_sync_attempt.py
       fails the attempt if the cc_pair is no longer perm-synced
       mark_doc_permission_sync_attempt_in_progress
+      LeaseHeartbeat renews heartbeat_at          tasks/shared/lease_heartbeat.py, see §5.9
       doc_sync_func(cc_pair, fetch_existing_docs_fn, fetch_existing_doc_ids_fn, callback)
         -> Generator[ElementExternalAccess]
       for each element:
@@ -372,6 +373,7 @@ beat: CHECK_FOR_EXTERNAL_GROUP_SYNC (every 20s)
                                                        commits immediately (idle-in-transaction
                                                        safety, ee/onyx/db/external_perm.py comment)
         mark_external_group_sync_attempt_in_progress
+        LeaseHeartbeat renews heartbeat_at           see §5.9
         group_sync_func(tenant_id, cc_pair) -> Generator[ExternalUserGroup]
         for each batch of 100:
           upsert_external_groups                     sets stale=False on every touched row
@@ -554,6 +556,31 @@ document to the wrong person.
    documents of the changed pairs
    (`onyx/db/document.py:mark_cc_pair_documents_for_sync__no_commit`), so the
    index write in §4.3 applies. See [[billing]] §4.5.
+9. **An open attempt row is held by a lease in Postgres.** While a doc or group
+   sync runs, a `LeaseHeartbeat` thread
+   (`onyx/background/celery/tasks/shared/lease_heartbeat.py`) sets
+   `heartbeat_at = now()` every `PERMISSION_SYNC_HEARTBEAT_INTERVAL`, only while
+   the row is `IN_PROGRESS`. Each beat check runs
+   `fail_expired_doc_permission_sync_attempts` /
+   `fail_expired_external_group_sync_attempts`. These set `FAILED` on
+   `IN_PROGRESS` rows whose `coalesce(heartbeat_at, time_started)` is older
+   than `PERMISSION_SYNC_LEASE_TIMEOUT`, and on `NOT_STARTED` rows that are
+   that old. A worker that dies therefore cannot leave its attempt open. A
+   worker that lost its row (renewal updates no row) stops: the heartbeat
+   sets `lost`, which `PermissionSyncCallback.should_stop` and the group-sync
+   loop read through `LeaseHeartbeat.is_lost()`. `is_lost()` is also true when
+   no renewal has succeeded for `PERMISSION_SYNC_LEASE_SELF_EXPIRY`, which is
+   shorter than the reaper's timeout, so a worker that cannot reach the DB
+   stops first. The heartbeat also stops renewing `PERMISSION_SYNC_LEASE_OVERRUN`
+   after the sync's time limit (`max_duration`), so a crawl thread that hangs
+   in a source call cannot hold the lease for the life of the process. Group sync writes (`upsert_external_groups`,
+   `remove_stale_external_groups` with `owning_attempt_id`) take a share lock
+   on the attempt row in the write's own transaction
+   (`lock_external_group_sync_attempt_if_owned`). The reaper cannot fail the
+   attempt while such a write is open, and a write after it is refused. `complete_*` raises `PermissionSyncAttemptLostError` unless the
+   row is still `IN_PROGRESS`, and `mark_*_failed` keeps an existing
+   `FAILED`/`CANCELED` reason. Do not write a terminal status without this
+   check.
 
 ---
 

backend/alembic/versions/aac91dfef2c7_add_heartbeat_at_to_permission_sync_.py

@@ -0,0 +1,51 @@
+"""add heartbeat_at to permission sync attempts
+
+Revision ID: aac91dfef2c7
+Revises: c92b67e56ec4
+Create Date: 2026-10-13 13:10:38.273867
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "aac91dfef2c7"
+down_revision = "c92b67e56ec4"
+branch_labels = None
+depends_on = None
+
+_TABLES = ("doc_permission_sync_attempt", "external_group_permission_sync_attempt")
+
+
+def upgrade() -> None:
+    for table in _TABLES:
+        op.add_column(
+            table,
+            sa.Column("heartbeat_at", sa.DateTime(timezone=True), nullable=True),
+        )
+
+        # A worker that died mid-sync left its attempt open forever. Fail every
+        # open attempt that a newer open attempt for the same cc_pair replaced;
+        # the lease reaper handles the newest one.
+        op.execute(
+            f"""
+            UPDATE {table}
+            SET status = 'FAILED',
+                time_finished = now(),
+                error_message = 'Abandoned: a newer attempt replaced it'
+            WHERE status IN ('NOT_STARTED', 'IN_PROGRESS')
+              AND id NOT IN (
+                  SELECT DISTINCT ON (connector_credential_pair_id) id
+                  FROM {table}
+                  WHERE status IN ('NOT_STARTED', 'IN_PROGRESS')
+                  ORDER BY connector_credential_pair_id, time_created DESC, id DESC
+              )
+            """
+        )
+
+
+def downgrade() -> None:
+    for table in _TABLES:
+        op.drop_column(table, "heartbeat_at")

backend/ee/onyx/background/celery/tasks/doc_permission_syncing/tasks.py

@@ -1,6 +1,7 @@
 import time
 import traceback
 from datetime import datetime, timedelta, timezone
+from functools import partial
 from time import sleep
 from typing import Any, cast
 from uuid import uuid4
@@ -36,12 +37,17 @@ from onyx.background.celery.celery_redis import (
     celery_get_unacked_task_ids,
 )
 from onyx.background.celery.tasks.beat_schedule import CLOUD_BEAT_MULTIPLIER_DEFAULT
+from onyx.background.celery.tasks.shared.lease_heartbeat import LeaseHeartbeat
 from onyx.configs.app_configs import JOB_TIMEOUT
 from onyx.configs.constants import (
     CELERY_GENERIC_BEAT_LOCK_TIMEOUT,
     CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT,
     CELERY_TASK_WAIT_FOR_FENCE_TIMEOUT,
     DANSWER_REDIS_FUNCTION_LOCK_PREFIX,
+    PERMISSION_SYNC_HEARTBEAT_INTERVAL,
+    PERMISSION_SYNC_LEASE_OVERRUN,
+    PERMISSION_SYNC_LEASE_SELF_EXPIRY,
+    PERMISSION_SYNC_LEASE_TIMEOUT,
     DocumentSource,
     OnyxCeleryPriority,
     OnyxCeleryQueues,
@@ -75,8 +81,10 @@ from onyx.db.models import ConnectorCredentialPair
 from onyx.db.permission_sync_attempt import (
     complete_doc_permission_sync_attempt,
     create_doc_permission_sync_attempt,
+    fail_expired_doc_permission_sync_attempts,
     mark_doc_permission_sync_attempt_failed,
     mark_doc_permission_sync_attempt_in_progress,
+    renew_doc_permission_sync_attempt_lease,
 )
 from onyx.db.sync_record import insert_sync_record, update_sync_record_status
 from onyx.db.users import batch_add_ext_perm_user_if_not_exists
@@ -161,6 +169,25 @@ def _fail_doc_permission_sync_attempt(
         )
 
 
+def _renew_doc_permission_sync_lease(attempt_id: int) -> bool:
+    with get_session_with_current_tenant() as db_session:
+        return renew_doc_permission_sync_attempt_lease(attempt_id, db_session)
+
+
+def fail_expired_doc_permission_sync_attempts_for_tenant() -> None:
+    lease_timeout = timedelta(seconds=PERMISSION_SYNC_LEASE_TIMEOUT)
+    with get_session_with_current_tenant() as db_session:
+        failed_ids = fail_expired_doc_permission_sync_attempts(
+            db_session,
+            lease_timeout=lease_timeout,
+            not_started_timeout=lease_timeout,
+        )
+    if failed_ids:
+        task_logger.warning(
+            f"Failed doc permission sync attempts with expired leases: {failed_ids}"
+        )
+
+
 def is_external_doc_permissions_sync_due(cc_pair: ConnectorCredentialPair) -> bool:
     """Returns boolean indicating if external doc permissions sync is due."""
 
@@ -302,6 +329,9 @@ def check_for_doc_permissions_sync(self: Task, *, tenant_id: str) -> bool | None
                         tenant_id, key_bytes, r, db_session
                     )
 
+        lock_beat.reacquire()
+        fail_expired_doc_permission_sync_attempts_for_tenant()
+
         lock_beat.reacquire()
         clear_caught_up_perm_sync_pending_marks(tenant_id)
         task_logger.info(f"check_for_doc_permissions_sync finished: tenant={tenant_id}")
@@ -549,6 +579,7 @@ def connector_permission_sync_generator_task(
     connector_type: str = "unknown"
     tasks_generated = 0
     docs_with_errors = 0
+    heartbeat: LeaseHeartbeat | None = None
     try:
         # Scoped to setup, never across the crawl below: on multi-tenant this session
         # is bound to one checked-out Connection, which PgBouncer closes once the gaps
@@ -633,8 +664,21 @@ def connector_permission_sync_generator_task(
             )
             redis_connector.permissions.set_fence(new_payload)
 
+            heartbeat = LeaseHeartbeat(
+                f"doc_permission_sync_attempt={attempt_id}",
+                renew=partial(_renew_doc_permission_sync_lease, attempt_id),
+                interval=PERMISSION_SYNC_HEARTBEAT_INTERVAL,
+                max_silence=PERMISSION_SYNC_LEASE_SELF_EXPIRY,
+                max_duration=JOB_TIMEOUT + PERMISSION_SYNC_LEASE_OVERRUN,
+            )
+            heartbeat.start()
+
             callback = PermissionSyncCallback(
-                redis_connector, lock, r, timeout_seconds=JOB_TIMEOUT
+                redis_connector,
+                lock,
+                r,
+                timeout_seconds=JOB_TIMEOUT,
+                lease=heartbeat,
             )
 
             connector_id = cc_pair.connector.id
@@ -742,6 +786,8 @@ def connector_permission_sync_generator_task(
         redis_connector.permissions.set_fence(None)
         raise e
     finally:
+        if heartbeat is not None:
+            heartbeat.stop()
         observe_doc_perm_sync_duration(time.monotonic() - sync_start, connector_type)
         if lock.owned():
             lock.release()
@@ -1047,6 +1093,7 @@ class PermissionSyncCallback(IndexingHeartbeatInterface):
         redis_lock: RedisLock,
         redis_client: TenantRedisClient,
         timeout_seconds: int | None = None,
+        lease: LeaseHeartbeat | None = None,
     ):
         super().__init__()
         self.redis_connector: RedisConnector = redis_connector
@@ -1061,11 +1108,15 @@ class PermissionSyncCallback(IndexingHeartbeatInterface):
         self.last_lock_monotonic = time.monotonic()
         self.start_monotonic = time.monotonic()
         self.timeout_seconds = timeout_seconds
+        self.lease = lease
 
     def should_stop(self) -> bool:
         if self.redis_connector.stop.fenced:
             return True
 
+        if self.lease is not None and self.lease.is_lost():
+            return True
+
         # Check if the task has exceeded its timeout
         # NOTE: Celery's soft_time_limit does not work with thread pools,
         # so we must enforce timeouts internally.

backend/ee/onyx/background/celery/tasks/external_group_syncing/tasks.py

@@ -1,6 +1,7 @@
 import time
 import traceback
 from datetime import datetime, timedelta, timezone
+from functools import partial
 from typing import Any, cast
 from uuid import uuid4
 
@@ -31,12 +32,17 @@ from onyx.background.celery.celery_redis import (
     celery_get_unacked_task_ids,
 )
 from onyx.background.celery.tasks.beat_schedule import CLOUD_BEAT_MULTIPLIER_DEFAULT
+from onyx.background.celery.tasks.shared.lease_heartbeat import LeaseHeartbeat
 from onyx.background.error_logging import emit_background_error
 from onyx.configs.app_configs import JOB_TIMEOUT
 from onyx.configs.constants import (
     CELERY_EXTERNAL_GROUP_SYNC_LOCK_TIMEOUT,
     CELERY_GENERIC_BEAT_LOCK_TIMEOUT,
     CELERY_TASK_WAIT_FOR_FENCE_TIMEOUT,
+    PERMISSION_SYNC_HEARTBEAT_INTERVAL,
+    PERMISSION_SYNC_LEASE_OVERRUN,
+    PERMISSION_SYNC_LEASE_SELF_EXPIRY,
+    PERMISSION_SYNC_LEASE_TIMEOUT,
     DocumentSource,
     OnyxCeleryPriority,
     OnyxCeleryQueues,
@@ -52,9 +58,11 @@ from onyx.db.models import ConnectorCredentialPair
 from onyx.db.permission_sync_attempt import (
     complete_external_group_sync_attempt,
     create_external_group_sync_attempt,
+    fail_expired_external_group_sync_attempts,
     get_external_group_sync_retry_after,
     mark_external_group_sync_attempt_failed,
     mark_external_group_sync_attempt_in_progress,
+    renew_external_group_sync_attempt_lease,
 )
 from onyx.db.sync_record import insert_sync_record, update_sync_record_status
 from onyx.redis.redis_connector import RedisConnector
@@ -91,6 +99,25 @@ def _fail_external_group_sync_attempt(attempt_id: int, error_msg: str) -> None:
         )
 
 
+def _renew_external_group_sync_lease(attempt_id: int) -> bool:
+    with get_session_with_current_tenant() as db_session:
+        return renew_external_group_sync_attempt_lease(attempt_id, db_session)
+
+
+def fail_expired_external_group_sync_attempts_for_tenant() -> None:
+    lease_timeout = timedelta(seconds=PERMISSION_SYNC_LEASE_TIMEOUT)
+    with get_session_with_current_tenant() as db_session:
+        failed_ids = fail_expired_external_group_sync_attempts(
+            db_session,
+            lease_timeout=lease_timeout,
+            not_started_timeout=lease_timeout,
+        )
+    if failed_ids:
+        task_logger.warning(
+            f"Failed external group sync attempts with expired leases: {failed_ids}"
+        )
+
+
 def _get_fence_validation_block_expiration() -> int:
     """
     Compute the expiration time for the fence validation block signal.
@@ -243,6 +270,9 @@ def check_for_external_group_sync(self: Task, *, tenant_id: str) -> bool | None:
                 f"External group sync queued: cc_pair={cc_pair_id} id={payload_id}"
             )
 
+        lock_beat.reacquire()
+        fail_expired_external_group_sync_attempts_for_tenant()
+
         # we want to run this less frequently than the overall task
         lock_beat.reacquire()
         if not r.exists(OnyxRedisSignals.BLOCK_VALIDATE_EXTERNAL_GROUP_SYNC_FENCES):
@@ -581,39 +611,72 @@ def _timed_perform_external_group_sync(
         mark_external_group_sync_attempt_in_progress(attempt_id, db_session)
         logger.info("Marked external group sync attempt %s as in progress", attempt_id)
 
-        logger.info(
-            "Syncing external groups for %s for cc_pair: %s", source_type, cc_pair_id
+        heartbeat = LeaseHeartbeat(
+            f"external_group_sync_attempt={attempt_id}",
+            renew=partial(_renew_external_group_sync_lease, attempt_id),
+            interval=PERMISSION_SYNC_HEARTBEAT_INTERVAL,
+            max_silence=PERMISSION_SYNC_LEASE_SELF_EXPIRY,
+            max_duration=JOB_TIMEOUT + PERMISSION_SYNC_LEASE_OVERRUN,
         )
-        external_user_group_batch: list[ExternalUserGroup] = []
-        seen_users: set[str] = set()  # Track unique users across all groups
-        total_groups_processed = 0
-        total_group_memberships_synced = 0
-        cumulative_upsert_time = 0.0
-        start_time = time.monotonic()
+        heartbeat.start()
         try:
-            external_user_group_generator = ext_group_sync_func(tenant_id, cc_pair)
-            for external_user_group in external_user_group_generator:
-                # Check if the task has exceeded its timeout
-                # NOTE: Celery's soft_time_limit does not work with thread pools,
-                # so we must enforce timeouts internally.
-                elapsed = time.monotonic() - start_time
-                if elapsed > timeout_seconds:
-                    raise RuntimeError(
-                        f"External group sync task timed out: "
-                        f"cc_pair={cc_pair_id} "
-                        f"elapsed={elapsed:.0f}s "
-                        f"timeout={timeout_seconds}s "
-                        f"groups_processed={total_groups_processed}"
+            logger.info(
+                "Syncing external groups for %s for cc_pair: %s",
+                source_type,
+                cc_pair_id,
+            )
+            external_user_group_batch: list[ExternalUserGroup] = []
+            seen_users: set[str] = set()  # Track unique users across all groups
+            total_groups_processed = 0
+            total_group_memberships_synced = 0
+            cumulative_upsert_time = 0.0
+            start_time = time.monotonic()
+            try:
+                external_user_group_generator = ext_group_sync_func(tenant_id, cc_pair)
+                for external_user_group in external_user_group_generator:
+                    # Check if the task has exceeded its timeout
+                    # NOTE: Celery's soft_time_limit does not work with thread pools,
+                    # so we must enforce timeouts internally.
+                    elapsed = time.monotonic() - start_time
+                    if elapsed > timeout_seconds:
+                        raise RuntimeError(
+                            f"External group sync task timed out: "
+                            f"cc_pair={cc_pair_id} "
+                            f"elapsed={elapsed:.0f}s "
+                            f"timeout={timeout_seconds}s "
+                            f"groups_processed={total_groups_processed}"
+                        )
+                    if heartbeat.is_lost():
+                        raise RuntimeError(
+                            f"External group sync lost its lease: "
+                            f"cc_pair={cc_pair_id} attempt={attempt_id}"
+                        )
+
+                    external_user_group_batch.append(external_user_group)
+
+                    # Track progress
+                    total_groups_processed += 1
+                    total_group_memberships_synced += len(
+                        external_user_group.user_emails
                     )
+                    seen_users.update(external_user_group.user_emails)
 
-                external_user_group_batch.append(external_user_group)
+                    if len(external_user_group_batch) >= _EXTERNAL_GROUP_BATCH_SIZE:
+                        logger.debug(
+                            "New external user groups: %s", external_user_group_batch
+                        )
+                        upsert_start = time.monotonic()
+                        upsert_external_groups(
+                            db_session=db_session,
+                            cc_pair_id=cc_pair_id,
+                            external_groups=external_user_group_batch,
+                            source=cc_pair.connector.source,
+                            owning_attempt_id=attempt_id,
+                        )
+                        cumulative_upsert_time += time.monotonic() - upsert_start
+                        external_user_group_batch = []
 
-                # Track progress
-                total_groups_processed += 1
-                total_group_memberships_synced += len(external_user_group.user_emails)
-                seen_users.update(external_user_group.user_emails)
-
-                if len(external_user_group_batch) >= _EXTERNAL_GROUP_BATCH_SIZE:
+                if external_user_group_batch:
                     logger.debug(
                         "New external user groups: %s", external_user_group_batch
                     )
@@ -623,74 +686,68 @@ def _timed_perform_external_group_sync(
                         cc_pair_id=cc_pair_id,
                         external_groups=external_user_group_batch,
                         source=cc_pair.connector.source,
+                        owning_attempt_id=attempt_id,
                     )
                     cumulative_upsert_time += time.monotonic() - upsert_start
-                    external_user_group_batch = []
+            except Exception as e:
+                format_error_for_logging(e)
 
-            if external_user_group_batch:
-                logger.debug("New external user groups: %s", external_user_group_batch)
-                upsert_start = time.monotonic()
-                upsert_external_groups(
-                    db_session=db_session,
-                    cc_pair_id=cc_pair_id,
-                    external_groups=external_user_group_batch,
-                    source=cc_pair.connector.source,
+                # Mark as failed (this also updates progress to show partial progress)
+                mark_external_group_sync_attempt_failed(
+                    attempt_id,
+                    db_session,
+                    error_message=str(e),
+                    full_exception_trace=traceback.format_exc(),
                 )
-                cumulative_upsert_time += time.monotonic() - upsert_start
-        except Exception as e:
-            format_error_for_logging(e)
 
-            # Mark as failed (this also updates progress to show partial progress)
-            mark_external_group_sync_attempt_failed(
-                attempt_id,
-                db_session,
-                error_message=str(e),
-                full_exception_trace=traceback.format_exc(),
-            )
+                # TODO: add some notification to the admins here
+                inc_group_sync_errors(connector_type)
+                logger.exception(
+                    "Error syncing external groups for %s for cc_pair: %s %s",
+                    source_type,
+                    cc_pair_id,
+                    e,
+                )
+                raise e
 
-            # TODO: add some notification to the admins here
-            inc_group_sync_errors(connector_type)
-            logger.exception(
-                "Error syncing external groups for %s for cc_pair: %s %s",
+            observe_group_sync_upsert_duration(cumulative_upsert_time, connector_type)
+
+            logger.info(
+                "Removing stale external groups for %s for cc_pair: %s",
                 source_type,
                 cc_pair_id,
-                e,
             )
-            raise e
+            # A lost lease means a replacement sync may own these rows now
+            remove_stale_external_groups(
+                db_session, cc_pair_id, owning_attempt_id=attempt_id
+            )
 
-        observe_group_sync_upsert_duration(cumulative_upsert_time, connector_type)
+            # Calculate total unique users processed
+            total_users_processed = len(seen_users)
 
-        logger.info(
-            "Removing stale external groups for %s for cc_pair: %s",
-            source_type,
-            cc_pair_id,
-        )
-        remove_stale_external_groups(db_session, cc_pair_id)
+            # Complete the sync attempt with final progress
+            complete_external_group_sync_attempt(
+                db_session=db_session,
+                attempt_id=attempt_id,
+                total_users_processed=total_users_processed,
+                total_groups_processed=total_groups_processed,
+                total_group_memberships_synced=total_group_memberships_synced,
+                errors_encountered=0,
+            )
+            logger.info(
+                "Completed external group sync attempt %s: %s groups, %s users, %s memberships",
+                attempt_id,
+                total_groups_processed,
+                total_users_processed,
+                total_group_memberships_synced,
+            )
 
-        # Calculate total unique users processed
-        total_users_processed = len(seen_users)
+            inc_group_sync_groups_processed(connector_type, total_groups_processed)
+            inc_group_sync_users_processed(connector_type, total_users_processed)
 
-        # Complete the sync attempt with final progress
-        complete_external_group_sync_attempt(
-            db_session=db_session,
-            attempt_id=attempt_id,
-            total_users_processed=total_users_processed,
-            total_groups_processed=total_groups_processed,
-            total_group_memberships_synced=total_group_memberships_synced,
-            errors_encountered=0,
-        )
-        logger.info(
-            "Completed external group sync attempt %s: %s groups, %s users, %s memberships",
-            attempt_id,
-            total_groups_processed,
-            total_users_processed,
-            total_group_memberships_synced,
-        )
-
-        inc_group_sync_groups_processed(connector_type, total_groups_processed)
-        inc_group_sync_users_processed(connector_type, total_users_processed)
-
-        mark_all_relevant_cc_pairs_as_external_group_synced(db_session, cc_pair)
+            mark_all_relevant_cc_pairs_as_external_group_synced(db_session, cc_pair)
+        finally:
+            heartbeat.stop()
 
     return connector_type
 

backend/ee/onyx/db/external_perm.py

@@ -9,6 +9,9 @@ from sqlalchemy.orm import Session
 from onyx.access.utils import build_ext_group_name_for_onyx
 from onyx.configs.constants import DocumentSource
 from onyx.db.models import PublicExternalUserGroup, User, User__ExternalUserGroupId
+from onyx.db.permission_sync_attempt import (
+    lock_external_group_sync_attempt_if_owned,
+)
 from onyx.db.users import batch_add_ext_perm_user_if_not_exists, get_user_by_email
 from onyx.utils.logger import setup_logger
 
@@ -89,9 +92,12 @@ def upsert_external_groups(
     cc_pair_id: int,
     external_groups: list[ExternalUserGroup],
     source: DocumentSource,
+    owning_attempt_id: int | None = None,
 ) -> None:
     """
     Batch upsert external user groups using INSERT ... ON CONFLICT DO UPDATE.
+    With owning_attempt_id, the write happens only while that group sync
+    attempt is IN_PROGRESS (see lock_external_group_sync_attempt_if_owned).
     - For existing rows (same user_id, external_user_group_id, cc_pair_id),
       sets stale=False
     - For new rows, inserts with stale=False
@@ -113,6 +119,10 @@ def upsert_external_groups(
 
     email_id_map = {user.email.lower(): user.id for user in all_group_members}
 
+    # After the user insert, which commits, so the lock covers the group writes
+    if owning_attempt_id is not None:
+        lock_external_group_sync_attempt_if_owned(owning_attempt_id, db_session)
+
     # Build all user-group mappings and public-group mappings
     user_group_mappings: list[dict] = []
     public_group_mappings: list[dict] = []
@@ -195,7 +205,12 @@ def upsert_external_groups(
 def remove_stale_external_groups(
     db_session: Session,
     cc_pair_id: int,
+    owning_attempt_id: int | None = None,
 ) -> None:
+    """Deletes the rows no sync cycle re-confirmed. With owning_attempt_id, the
+    delete happens only while that group sync attempt is IN_PROGRESS."""
+    if owning_attempt_id is not None:
+        lock_external_group_sync_attempt_if_owned(owning_attempt_id, db_session)
     db_session.execute(
         delete(User__ExternalUserGroupId).where(
             User__ExternalUserGroupId.cc_pair_id == cc_pair_id,

backend/onyx/background/celery/tasks/shared/lease_heartbeat.py

@@ -0,0 +1,102 @@
+import contextvars
+import threading
+import time
+from collections.abc import Callable
+from types import TracebackType
+
+from onyx.utils.logger import setup_logger
+
+logger = setup_logger()
+
+
+class LeaseHeartbeat:
+    """Renews a Postgres-held lease on a background thread while a job runs.
+
+    `renew` returns False when the job row is no longer ours (e.g. the reaper
+    failed it). The heartbeat then stops and sets `lost`. An exception from
+    `renew` (e.g. a DB blip) is logged and retried on the next interval.
+
+    The job must stop doing work once `is_lost()` returns True. That is also
+    the case when no renewal has succeeded for `max_silence` seconds: the
+    reaper may already have failed the job, so `max_silence` must be shorter
+    than the reaper's lease timeout.
+
+    After `max_duration` seconds the heartbeat stops renewing and sets `lost`.
+    A job whose thread hangs (e.g. in a source call that never returns) then
+    loses its lease and the reaper frees its row, instead of the lease living
+    as long as the process.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        renew: Callable[[], bool],
+        interval: float,
+        max_silence: float,
+        max_duration: float,
+    ):
+        self.name = name
+        self.lost = threading.Event()
+        self._renew = renew
+        self._interval = interval
+        self._max_silence = max_silence
+        self._max_duration = max_duration
+        self._started = time.monotonic()
+        self._last_renewed = time.monotonic()
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+
+    def _loop(self) -> None:
+        while not self._stop.wait(self._interval):
+            if time.monotonic() - self._started > self._max_duration:
+                logger.warning(
+                    "Lease not renewed past its maximum duration of %ss: %s",
+                    self._max_duration,
+                    self.name,
+                )
+                self.lost.set()
+                return
+            try:
+                if not self._renew():
+                    logger.warning("Lease lost: %s", self.name)
+                    self.lost.set()
+                    return
+                self._last_renewed = time.monotonic()
+            except Exception:
+                logger.exception("Lease renewal failed: %s", self.name)
+
+    def is_lost(self) -> bool:
+        if self.lost.is_set():
+            return True
+        if time.monotonic() - self._last_renewed > self._max_silence:
+            logger.warning("Lease expired without a renewal: %s", self.name)
+            self.lost.set()
+            return True
+        return False
+
+    def start(self) -> None:
+        self._started = time.monotonic()
+        self._last_renewed = time.monotonic()
+        # contextvars (e.g. the tenant id) are not inherited by new threads
+        context = contextvars.copy_context()
+        self._thread = threading.Thread(
+            target=context.run, args=(self._loop,), daemon=True
+        )
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+        if self._thread is not None:
+            self._thread.join(timeout=5)
+
+    def __enter__(self) -> "LeaseHeartbeat":
+        self.start()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        self.stop()

backend/onyx/configs/constants.py

@@ -189,6 +189,20 @@ CELERY_INDEXING_LOCK_TIMEOUT = CELERY_INDEXING_WATCHDOG_CONNECTOR_TIMEOUT + 900
 # Heartbeat interval for indexing worker liveness detection
 INDEXING_WORKER_HEARTBEAT_INTERVAL = 30  # seconds
 
+# Permission and group sync attempts hold a lease in Postgres: the worker renews
+# heartbeat_at on this interval, and the beat reaper fails attempts whose lease
+# is older than the timeout.
+PERMISSION_SYNC_HEARTBEAT_INTERVAL = 60  # seconds
+PERMISSION_SYNC_LEASE_TIMEOUT = 15 * 60  # seconds
+# A worker that could not renew for this long stops writing, before the reaper
+# can fail its attempt and let a replacement start.
+# A sync still running this long past its time limit is hung (its own timeout
+# check never ran), so its lease stops renewing and the reaper frees its row.
+PERMISSION_SYNC_LEASE_OVERRUN = 10 * 60  # seconds
+PERMISSION_SYNC_LEASE_SELF_EXPIRY = (
+    PERMISSION_SYNC_LEASE_TIMEOUT - 2 * PERMISSION_SYNC_HEARTBEAT_INTERVAL
+)
+
 # how long a task should wait for associated fence to be ready
 CELERY_TASK_WAIT_FOR_FENCE_TIMEOUT = 5 * 60  # 5 min
 

backend/onyx/db/models.py

@@ -6216,6 +6216,10 @@ class DocPermissionSyncAttempt(Base):
     time_finished: Mapped[datetime.datetime | None] = mapped_column(
         DateTime(timezone=True), default=None
     )
+    # Lease renewed by the running worker; the reaper fails the attempt when it expires
+    heartbeat_at: Mapped[datetime.datetime | None] = mapped_column(
+        DateTime(timezone=True), default=None
+    )
 
     # Relationships
     connector_credential_pair: Mapped[ConnectorCredentialPair] = relationship(
@@ -6286,6 +6290,10 @@ class ExternalGroupPermissionSyncAttempt(Base):
     time_finished: Mapped[datetime.datetime | None] = mapped_column(
         DateTime(timezone=True), default=None
     )
+    # Lease renewed by the running worker; the reaper fails the attempt when it expires
+    heartbeat_at: Mapped[datetime.datetime | None] = mapped_column(
+        DateTime(timezone=True), default=None
+    )
 
     # Relationships
     connector_credential_pair: Mapped[ConnectorCredentialPair | None] = relationship(

backend/onyx/db/permission_sync_attempt.py

@@ -4,10 +4,10 @@ This module contains all CRUD operations for both DocPermissionSyncAttempt
 and ExternalGroupPermissionSyncAttempt models, along with shared utilities.
 """
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any, cast
 
-from sqlalchemy import ColumnElement, delete, func, select
+from sqlalchemy import ColumnElement, delete, func, or_, select, update
 from sqlalchemy.engine.cursor import CursorResult
 from sqlalchemy.orm import Session, joinedload
 
@@ -28,6 +28,17 @@ logger = setup_logger()
 # Enough consecutive failures to reach the longest retry delay.
 _GROUP_SYNC_BACKOFF_LOOKBACK = 8
 
+_FAILURE_STATUSES = (PermissionSyncStatus.FAILED, PermissionSyncStatus.CANCELED)
+
+_AttemptModel = (
+    type[DocPermissionSyncAttempt] | type[ExternalGroupPermissionSyncAttempt]
+)
+
+
+class PermissionSyncAttemptLostError(RuntimeError):
+    """The attempt is no longer IN_PROGRESS, e.g. the reaper failed it after its
+    lease expired. The worker that ran it must stop."""
+
 
 # =============================================================================
 # DOC PERMISSION SYNC ATTEMPT CRUD
@@ -123,6 +134,7 @@ def mark_doc_permission_sync_attempt_in_progress(
 
         attempt.status = PermissionSyncStatus.IN_PROGRESS
         attempt.time_started = func.now()
+        attempt.heartbeat_at = func.now()
         db_session.commit()
         return attempt
     except Exception:
@@ -160,6 +172,11 @@ def mark_doc_permission_sync_attempt_failed(
             .with_for_update()
         ).scalar_one()
 
+        if attempt.status in _FAILURE_STATUSES:
+            # The reaper or a cancel already closed it; keep that reason.
+            db_session.commit()
+            return
+
         if not attempt.time_started:
             attempt.time_started = func.now()
         attempt.status = PermissionSyncStatus.FAILED
@@ -202,6 +219,11 @@ def complete_doc_permission_sync_attempt(
             .with_for_update()
         ).scalar_one()
 
+        if attempt.status != PermissionSyncStatus.IN_PROGRESS:
+            raise PermissionSyncAttemptLostError(
+                f"Doc permission sync attempt {attempt_id} is {attempt.status}, not IN_PROGRESS"
+            )
+
         # Update progress counters
         attempt.total_docs_synced = (attempt.total_docs_synced or 0) + total_docs_synced
         attempt.docs_with_permission_errors = (
@@ -384,6 +406,7 @@ def mark_external_group_sync_attempt_in_progress(
 
         attempt.status = PermissionSyncStatus.IN_PROGRESS
         attempt.time_started = func.now()
+        attempt.heartbeat_at = func.now()
         db_session.commit()
         return attempt
     except Exception:
@@ -414,6 +437,11 @@ def mark_external_group_sync_attempt_failed(
             .with_for_update()
         ).scalar_one()
 
+        if attempt.status in _FAILURE_STATUSES:
+            # The reaper or a cancel already closed it; keep that reason.
+            db_session.commit()
+            return
+
         if not attempt.time_started:
             attempt.time_started = func.now()
         attempt.status = PermissionSyncStatus.FAILED
@@ -458,6 +486,11 @@ def complete_external_group_sync_attempt(
             .with_for_update()
         ).scalar_one()
 
+        if attempt.status != PermissionSyncStatus.IN_PROGRESS:
+            raise PermissionSyncAttemptLostError(
+                f"External group sync attempt {attempt_id} is {attempt.status}, not IN_PROGRESS"
+            )
+
         # Update progress counters
         attempt.total_users_processed = (
             attempt.total_users_processed or 0
@@ -484,6 +517,134 @@ def complete_external_group_sync_attempt(
         raise
 
 
+# =============================================================================
+# LEASES
+# =============================================================================
+
+
+def _renew_lease(model: _AttemptModel, attempt_id: int, db_session: Session) -> bool:
+    renewed = db_session.execute(
+        update(model)
+        .where(model.id == attempt_id)
+        .where(model.status == PermissionSyncStatus.IN_PROGRESS)
+        .values(heartbeat_at=func.now())
+        .returning(model.id)
+    ).first()
+    db_session.commit()
+    return renewed is not None
+
+
+def _lock_if_owned(model: _AttemptModel, attempt_id: int, db_session: Session) -> None:
+    owned = db_session.execute(
+        select(model.id)
+        .where(model.id == attempt_id)
+        .where(model.status == PermissionSyncStatus.IN_PROGRESS)
+        .with_for_update(read=True)
+    ).first()
+    if owned is None:
+        raise PermissionSyncAttemptLostError(
+            f"{model.__name__} {attempt_id} is no longer IN_PROGRESS"
+        )
+
+
+def lock_doc_permission_sync_attempt_if_owned(
+    attempt_id: int, db_session: Session
+) -> None:
+    """Takes a share lock on a running attempt, or raises
+    PermissionSyncAttemptLostError. Call it in the transaction of a sync write:
+    the reaper cannot fail the attempt until that write commits, and a write
+    after the reaper failed it is refused."""
+    _lock_if_owned(DocPermissionSyncAttempt, attempt_id, db_session)
+
+
+def lock_external_group_sync_attempt_if_owned(
+    attempt_id: int, db_session: Session
+) -> None:
+    """Takes a share lock on a running attempt, or raises
+    PermissionSyncAttemptLostError. Call it in the transaction of a sync write:
+    the reaper cannot fail the attempt until that write commits, and a write
+    after the reaper failed it is refused."""
+    _lock_if_owned(ExternalGroupPermissionSyncAttempt, attempt_id, db_session)
+
+
+def renew_doc_permission_sync_attempt_lease(
+    attempt_id: int, db_session: Session
+) -> bool:
+    """Renews the lease of a running attempt. Returns False when the attempt is
+    no longer IN_PROGRESS, which means the caller has lost it."""
+    return _renew_lease(DocPermissionSyncAttempt, attempt_id, db_session)
+
+
+def renew_external_group_sync_attempt_lease(
+    attempt_id: int, db_session: Session
+) -> bool:
+    """Renews the lease of a running attempt. Returns False when the attempt is
+    no longer IN_PROGRESS, which means the caller has lost it."""
+    return _renew_lease(ExternalGroupPermissionSyncAttempt, attempt_id, db_session)
+
+
+def _fail_expired_attempts(
+    model: _AttemptModel,
+    db_session: Session,
+    lease_timeout: timedelta,
+    not_started_timeout: timedelta,
+) -> list[int]:
+    now = func.now()
+    last_alive = func.coalesce(model.heartbeat_at, model.time_started)
+    failed_ids = list(
+        db_session.execute(
+            update(model)
+            .where(
+                or_(
+                    (model.status == PermissionSyncStatus.IN_PROGRESS)
+                    & (last_alive < now - lease_timeout),
+                    (model.status == PermissionSyncStatus.NOT_STARTED)
+                    & (model.time_created < now - not_started_timeout),
+                )
+            )
+            .values(
+                status=PermissionSyncStatus.FAILED,
+                time_started=func.coalesce(model.time_started, now),
+                time_finished=now,
+                error_message=(
+                    "The worker running this attempt stopped renewing its lease "
+                    "(it crashed, restarted, or lost its task)"
+                ),
+            )
+            .returning(model.id)
+        ).scalars()
+    )
+    db_session.commit()
+    return failed_ids
+
+
+def fail_expired_doc_permission_sync_attempts(
+    db_session: Session,
+    lease_timeout: timedelta,
+    not_started_timeout: timedelta,
+) -> list[int]:
+    """Fails IN_PROGRESS attempts whose lease expired and NOT_STARTED attempts
+    that never started. Returns the IDs it failed."""
+    return _fail_expired_attempts(
+        DocPermissionSyncAttempt, db_session, lease_timeout, not_started_timeout
+    )
+
+
+def fail_expired_external_group_sync_attempts(
+    db_session: Session,
+    lease_timeout: timedelta,
+    not_started_timeout: timedelta,
+) -> list[int]:
+    """Fails IN_PROGRESS attempts whose lease expired and NOT_STARTED attempts
+    that never started. Returns the IDs it failed."""
+    return _fail_expired_attempts(
+        ExternalGroupPermissionSyncAttempt,
+        db_session,
+        lease_timeout,
+        not_started_timeout,
+    )
+
+
 # =============================================================================
 # DELETION FUNCTIONS
 # =============================================================================

backend/tests/external_dependency_unit/permission_sync/test_doc_permission_sync_attempt.py

@@ -12,6 +12,7 @@ from onyx.db.enums import (
     PermissionSyncStatus,
 )
 from onyx.db.permission_sync_attempt import (
+    PermissionSyncAttemptLostError,
     complete_doc_permission_sync_attempt,
     create_doc_permission_sync_attempt,
     get_doc_permission_sync_attempt,
@@ -274,40 +275,28 @@ class TestDocPermissionSyncAttempt:
         assert completed_attempt.docs_with_permission_errors == 15
         assert completed_attempt.time_finished is not None
 
-    def test_complete_doc_permission_sync_attempt_can_be_called_multiple_times(
+    def test_complete_doc_permission_sync_attempt_rejects_second_call(
         self, db_session: Session
     ) -> None:
-        """Test that complete can be called multiple times if needed (accumulates correctly)."""
+        """A finished attempt is no longer IN_PROGRESS, so a second complete
+        (e.g. from a worker that lost the attempt) is rejected."""
         cc_pair = create_test_connector_credential_pair(db_session)
         attempt_id = create_doc_permission_sync_attempt(cc_pair.id, db_session)
-
-        # Mark as in progress
         mark_doc_permission_sync_attempt_in_progress(attempt_id, db_session)
 
-        # Complete once
         first_complete = complete_doc_permission_sync_attempt(
             db_session=db_session,
             attempt_id=attempt_id,
             total_docs_synced=50,
             docs_with_permission_errors=5,
         )
-
-        # Verify first completion
         assert first_complete.status == PermissionSyncStatus.COMPLETED_WITH_ERRORS
         assert first_complete.total_docs_synced == 50
-        assert first_complete.docs_with_permission_errors == 5
-        assert first_complete.time_finished is not None
 
-        # Call complete again (simulating additional batch processing)
-        second_complete = complete_doc_permission_sync_attempt(
-            db_session=db_session,
-            attempt_id=attempt_id,
-            total_docs_synced=50,
-            docs_with_permission_errors=10,
-        )
-
-        # Should accumulate progress from both calls
-        assert second_complete.status == PermissionSyncStatus.COMPLETED_WITH_ERRORS
-        assert second_complete.total_docs_synced == 100
-        assert second_complete.docs_with_permission_errors == 15
-        assert second_complete.time_finished is not None
+        with pytest.raises(PermissionSyncAttemptLostError):
+            complete_doc_permission_sync_attempt(
+                db_session=db_session,
+                attempt_id=attempt_id,
+                total_docs_synced=50,
+                docs_with_permission_errors=10,
+            )

backend/tests/external_dependency_unit/permission_sync/test_external_group_permission_sync_attempt.py

@@ -29,6 +29,7 @@ from onyx.db.models import (
     ExternalGroupPermissionSyncAttempt,
 )
 from onyx.db.permission_sync_attempt import (
+    PermissionSyncAttemptLostError,
     complete_external_group_sync_attempt,
     create_external_group_sync_attempt,
     get_external_group_sync_attempt,
@@ -434,17 +435,15 @@ class TestExternalGroupPermissionSyncAttempt:
         assert completed_attempt.total_group_memberships_synced == 1200
         assert completed_attempt.time_finished is not None
 
-    def test_complete_external_group_sync_attempt_can_be_called_multiple_times(
+    def test_complete_external_group_sync_attempt_rejects_second_call(
         self, db_session: Session
     ) -> None:
-        """Test that complete can be called multiple times if needed (accumulates correctly)."""
+        """A finished attempt is no longer IN_PROGRESS, so a second complete
+        (e.g. from a worker that lost the attempt) is rejected."""
         cc_pair = _create_test_connector_credential_pair(db_session)
         attempt_id = create_external_group_sync_attempt(cc_pair.id, db_session)
-
-        # Mark as in progress
         mark_external_group_sync_attempt_in_progress(attempt_id, db_session)
 
-        # Complete once
         first_complete = complete_external_group_sync_attempt(
             db_session=db_session,
             attempt_id=attempt_id,
@@ -453,30 +452,18 @@ class TestExternalGroupPermissionSyncAttempt:
             total_group_memberships_synced=600,
             errors_encountered=0,
         )
-
-        # Verify first completion
         assert first_complete.status == PermissionSyncStatus.SUCCESS
         assert first_complete.total_users_processed == 200
-        assert first_complete.total_groups_processed == 10
-        assert first_complete.total_group_memberships_synced == 600
-        assert first_complete.time_finished is not None
 
-        # Call complete again (simulating additional batch processing)
-        second_complete = complete_external_group_sync_attempt(
-            db_session=db_session,
-            attempt_id=attempt_id,
-            total_users_processed=300,
-            total_groups_processed=15,
-            total_group_memberships_synced=600,
-            errors_encountered=5,
-        )
-
-        # Should accumulate progress from both calls and update status
-        assert second_complete.status == PermissionSyncStatus.COMPLETED_WITH_ERRORS
-        assert second_complete.total_users_processed == 500
-        assert second_complete.total_groups_processed == 25
-        assert second_complete.total_group_memberships_synced == 1200
-        assert second_complete.time_finished is not None
+        with pytest.raises(PermissionSyncAttemptLostError):
+            complete_external_group_sync_attempt(
+                db_session=db_session,
+                attempt_id=attempt_id,
+                total_users_processed=300,
+                total_groups_processed=15,
+                total_group_memberships_synced=600,
+                errors_encountered=5,
+            )
 
     def test_global_vs_connector_specific_attempts(self, db_session: Session) -> None:
         """Test that global and connector-specific attempts are properly separated."""

backend/tests/external_dependency_unit/permission_sync/test_permission_sync_attempt_lease.py

@@ -0,0 +1,278 @@
+"""
+Lease coverage for doc permission and external group sync attempts: renewal,
+the expired-lease reaper, and the terminal writes of a worker that lost its
+attempt.
+"""
+
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import select, update
+from sqlalchemy.orm import Session
+
+from ee.onyx.db.external_perm import (
+    ExternalUserGroup,
+    mark_old_external_groups_as_stale,
+    remove_stale_external_groups,
+    upsert_external_groups,
+)
+from onyx.background.celery.tasks.shared.lease_heartbeat import LeaseHeartbeat
+from onyx.configs.constants import DocumentSource
+from onyx.db.engine.sql_engine import get_session_with_current_tenant
+from onyx.db.enums import PermissionSyncStatus
+from onyx.db.models import (
+    DocPermissionSyncAttempt,
+    ExternalGroupPermissionSyncAttempt,
+    User__ExternalUserGroupId,
+)
+from onyx.db.permission_sync_attempt import (
+    PermissionSyncAttemptLostError,
+    complete_doc_permission_sync_attempt,
+    complete_external_group_sync_attempt,
+    create_doc_permission_sync_attempt,
+    create_external_group_sync_attempt,
+    fail_expired_doc_permission_sync_attempts,
+    fail_expired_external_group_sync_attempts,
+    mark_doc_permission_sync_attempt_failed,
+    mark_doc_permission_sync_attempt_in_progress,
+    mark_external_group_sync_attempt_failed,
+    mark_external_group_sync_attempt_in_progress,
+    renew_doc_permission_sync_attempt_lease,
+    renew_external_group_sync_attempt_lease,
+)
+from tests.external_dependency_unit.permission_sync.conftest import (
+    create_test_connector_credential_pair,
+)
+
+_LEASE = timedelta(minutes=15)
+
+
+@dataclass(frozen=True)
+class _AttemptKind:
+    model: type[DocPermissionSyncAttempt] | type[ExternalGroupPermissionSyncAttempt]
+    create: Callable[[int, Session], int]
+    start: Callable[[int, Session], object]
+    renew: Callable[[int, Session], bool]
+    reap: Callable[[Session, timedelta, timedelta], list[int]]
+    fail: Callable[[int, Session, str], None]
+    complete: Callable[[Session, int], object]
+
+
+_DOC = _AttemptKind(
+    model=DocPermissionSyncAttempt,
+    create=create_doc_permission_sync_attempt,
+    start=mark_doc_permission_sync_attempt_in_progress,
+    renew=renew_doc_permission_sync_attempt_lease,
+    reap=fail_expired_doc_permission_sync_attempts,
+    fail=lambda attempt_id, db_session, msg: mark_doc_permission_sync_attempt_failed(
+        attempt_id, db_session, error_message=msg
+    ),
+    complete=lambda db_session, attempt_id: complete_doc_permission_sync_attempt(
+        db_session, attempt_id, total_docs_synced=1, docs_with_permission_errors=0
+    ),
+)
+_GROUP = _AttemptKind(
+    model=ExternalGroupPermissionSyncAttempt,
+    create=create_external_group_sync_attempt,
+    start=mark_external_group_sync_attempt_in_progress,
+    renew=renew_external_group_sync_attempt_lease,
+    reap=fail_expired_external_group_sync_attempts,
+    fail=lambda attempt_id, db_session, msg: mark_external_group_sync_attempt_failed(
+        attempt_id, db_session, error_message=msg
+    ),
+    complete=lambda db_session, attempt_id: complete_external_group_sync_attempt(
+        db_session,
+        attempt_id,
+        total_users_processed=1,
+        total_groups_processed=1,
+        total_group_memberships_synced=1,
+    ),
+)
+
+
+@pytest.fixture(params=[_DOC, _GROUP], ids=["doc", "group"])
+def kind(request: pytest.FixtureRequest) -> _AttemptKind:
+    return request.param
+
+
+def _started_attempt(kind: _AttemptKind, db_session: Session) -> int:
+    cc_pair = create_test_connector_credential_pair(db_session)
+    attempt_id = kind.create(cc_pair.id, db_session)
+    kind.start(attempt_id, db_session)
+    return attempt_id
+
+
+def _set_times(
+    kind: _AttemptKind,
+    db_session: Session,
+    attempt_id: int,
+    **times: datetime | None,
+) -> None:
+    db_session.execute(
+        update(kind.model).where(kind.model.id == attempt_id).values(**times)
+    )
+    db_session.commit()
+
+
+def _status(
+    kind: _AttemptKind, db_session: Session, attempt_id: int
+) -> PermissionSyncStatus:
+    db_session.expire_all()
+    attempt = db_session.get(kind.model, attempt_id)
+    assert attempt is not None
+    return attempt.status
+
+
+def test_renew_succeeds_only_while_in_progress(
+    kind: _AttemptKind, db_session: Session
+) -> None:
+    attempt_id = _started_attempt(kind, db_session)
+    assert kind.renew(attempt_id, db_session)
+
+    kind.fail(attempt_id, db_session, "boom")
+    assert not kind.renew(attempt_id, db_session)
+
+
+def test_reaper_fails_only_expired_attempts(
+    kind: _AttemptKind, db_session: Session
+) -> None:
+    long_ago = datetime.now(timezone.utc) - timedelta(hours=1)
+
+    fresh = _started_attempt(kind, db_session)
+
+    expired = _started_attempt(kind, db_session)
+    _set_times(kind, db_session, expired, heartbeat_at=long_ago)
+
+    # Started before leases existed: no heartbeat, old start time.
+    legacy = _started_attempt(kind, db_session)
+    _set_times(kind, db_session, legacy, heartbeat_at=None, time_started=long_ago)
+
+    cc_pair = create_test_connector_credential_pair(db_session)
+    never_started = kind.create(cc_pair.id, db_session)
+    _set_times(kind, db_session, never_started, time_created=long_ago)
+
+    failed_ids = kind.reap(db_session, _LEASE, _LEASE)
+
+    assert {expired, legacy, never_started} <= set(failed_ids)
+    assert fresh not in failed_ids
+    assert _status(kind, db_session, fresh) == PermissionSyncStatus.IN_PROGRESS
+    for attempt_id in (expired, legacy, never_started):
+        assert _status(kind, db_session, attempt_id) == PermissionSyncStatus.FAILED
+
+
+def test_worker_cannot_overwrite_reaped_attempt(
+    kind: _AttemptKind, db_session: Session
+) -> None:
+    attempt_id = _started_attempt(kind, db_session)
+    _set_times(
+        kind,
+        db_session,
+        attempt_id,
+        heartbeat_at=datetime.now(timezone.utc) - timedelta(hours=1),
+    )
+    assert attempt_id in kind.reap(db_session, _LEASE, _LEASE)
+
+    with pytest.raises(PermissionSyncAttemptLostError):
+        kind.complete(db_session, attempt_id)
+
+    # The worker's own failure path keeps the reaper's reason.
+    kind.fail(attempt_id, db_session, "worker error")
+    db_session.expire_all()
+    attempt = db_session.get(kind.model, attempt_id)
+    assert attempt is not None
+    assert attempt.status == PermissionSyncStatus.FAILED
+    assert attempt.error_message is not None
+    assert "lease" in attempt.error_message
+
+
+def test_heartbeat_sets_lost_when_attempt_is_reaped(
+    kind: _AttemptKind,
+    db_session: Session,
+    tenant_context: None,  # noqa: ARG001
+) -> None:
+    attempt_id = _started_attempt(kind, db_session)
+
+    def renew() -> bool:
+        # Runs on the heartbeat thread, so it needs its own session.
+        with get_session_with_current_tenant() as thread_session:
+            return kind.renew(attempt_id, thread_session)
+
+    with LeaseHeartbeat(
+        "test", renew=renew, interval=0.05, max_silence=60, max_duration=60
+    ) as heartbeat:
+        time.sleep(0.2)
+        assert not heartbeat.lost.is_set()
+
+        kind.fail(attempt_id, db_session, "reaped")
+        assert heartbeat.lost.wait(timeout=2)
+
+
+def test_heartbeat_expires_when_renewals_keep_failing() -> None:
+    """A worker that cannot reach the DB stops before the reaper can fail its
+    attempt and let a replacement start."""
+
+    def renew() -> bool:
+        raise ConnectionError("db unreachable")
+
+    with LeaseHeartbeat(
+        "test", renew=renew, interval=0.05, max_silence=0.3, max_duration=60
+    ) as heartbeat:
+        assert not heartbeat.is_lost()
+        time.sleep(0.5)
+        assert heartbeat.is_lost()
+
+
+def test_group_writes_are_refused_after_the_attempt_is_lost(
+    db_session: Session,
+) -> None:
+    cc_pair = create_test_connector_credential_pair(db_session)
+    attempt_id = create_external_group_sync_attempt(cc_pair.id, db_session)
+    mark_external_group_sync_attempt_in_progress(attempt_id, db_session)
+    group = ExternalUserGroup(id="group-a", user_emails=["lease-user@example.com"])
+
+    upsert_external_groups(
+        db_session, cc_pair.id, [group], DocumentSource.GOOGLE_DRIVE, attempt_id
+    )
+    mark_old_external_groups_as_stale(db_session, cc_pair.id)
+
+    mark_external_group_sync_attempt_failed(attempt_id, db_session, "reaped")
+
+    with pytest.raises(PermissionSyncAttemptLostError):
+        upsert_external_groups(
+            db_session, cc_pair.id, [group], DocumentSource.GOOGLE_DRIVE, attempt_id
+        )
+    db_session.rollback()
+    with pytest.raises(PermissionSyncAttemptLostError):
+        remove_stale_external_groups(
+            db_session, cc_pair.id, owning_attempt_id=attempt_id
+        )
+    db_session.rollback()
+
+    # The stale rows a replacement sync would refresh are still there
+    remaining = db_session.scalars(
+        select(User__ExternalUserGroupId).where(
+            User__ExternalUserGroupId.cc_pair_id == cc_pair.id
+        )
+    ).all()
+    assert len(remaining) == 1
+    assert remaining[0].stale
+
+
+def test_heartbeat_stops_renewing_after_its_maximum_duration() -> None:
+    """A hung job must not hold its lease for as long as the process lives."""
+    renewals: list[float] = []
+
+    def renew() -> bool:
+        renewals.append(time.monotonic())
+        return True
+
+    with LeaseHeartbeat(
+        "test", renew=renew, interval=0.05, max_silence=60, max_duration=0.3
+    ) as heartbeat:
+        assert heartbeat.lost.wait(timeout=2)
+        count = len(renewals)
+        time.sleep(0.2)
+        assert len(renewals) == count