OAuth: re-enable sync active user repos task (#12443)

- Take into consideration active users from the last 30 days only.
- Ignore GitHub App, so we don't re-sync installations over and over
again, as users can share the same installation, and repos from the GH
app are kept in sync via a webhook.

This task now completes in 42 minutes.

Closes #12406

Author: stsewd

readthedocs/oauth/tasks.py

@@ -10,6 +10,7 @@
 from django.utils import timezone
 
 from readthedocs.api.v2.views.integrations import ExternalVersionData
+from readthedocs.builds.utils import memcache_lock
 from readthedocs.core.utils.tasks import PublicTask
 from readthedocs.core.utils.tasks import user_id_matches_or_superuser
 from readthedocs.core.views.hooks import VersionInfo
@@ -26,15 +27,15 @@
 from readthedocs.oauth.notifications import MESSAGE_OAUTH_WEBHOOK_INVALID
 from readthedocs.oauth.notifications import MESSAGE_OAUTH_WEBHOOK_NO_ACCOUNT
 from readthedocs.oauth.notifications import MESSAGE_OAUTH_WEBHOOK_NO_PERMISSIONS
+from readthedocs.oauth.services import GitHubAppService
+from readthedocs.oauth.services import registry
 from readthedocs.oauth.services.base import SyncServiceError
 from readthedocs.oauth.utils import SERVICE_MAP
 from readthedocs.projects.models import Project
 from readthedocs.sso.models import SSOIntegration
 from readthedocs.vcs_support.backends.git import parse_version_from_ref
 from readthedocs.worker import app
 
-from .services import registry
-
 
 log = structlog.get_logger(__name__)
 
@@ -49,13 +50,16 @@
     time_limit=900,
     soft_time_limit=600,
 )
-def sync_remote_repositories(user_id):
+def sync_remote_repositories(user_id, skip_githubapp=False):
     user = User.objects.filter(pk=user_id).first()
     if not user:
         return
 
     failed_services = set()
     for service_cls in registry:
+        if skip_githubapp and service_cls == GitHubAppService:
+            continue
+
         try:
             service_cls.sync_user_access(user)
         except SyncServiceError:
@@ -110,12 +114,13 @@ def sync_remote_repositories_from_sso_organizations():
     queue="web",
     time_limit=60 * 60 * 3,  # 3h
     soft_time_limit=(60 * 60 * 3) - 5 * 60,  # 2h 55m
+    bind=True,
 )
-def sync_active_users_remote_repositories():
+def sync_active_users_remote_repositories(self):
     """
     Sync ``RemoteRepository`` for active users.
 
-    We consider active users those that logged in at least once in the last 90 days.
+    We consider active users those that logged in at least once in the last 45 days.
 
     This task is thought to be executed daily. It checks the weekday of the
     last login of the user with today's weekday. If they match, the re-sync is
@@ -124,34 +129,50 @@ def sync_active_users_remote_repositories():
     Note this is a long running task syncronizhing all the users in the same Celery process,
     and it will require a pretty high ``time_limit`` and ``soft_time_limit``.
     """
-    today_weekday = timezone.now().isoweekday()
-    three_months_ago = timezone.now() - datetime.timedelta(days=90)
-    users = User.objects.annotate(weekday=ExtractIsoWeekDay("last_login")).filter(
-        last_login__gt=three_months_ago,
-        socialaccount__isnull=False,
-        weekday=today_weekday,
-    )
-
-    users_count = users.count()
-    structlog.contextvars.bind_contextvars(total_users=users_count)
-    log.info("Triggering re-sync of RemoteRepository for active users.")
+    # This task is expensive, and we run it daily.
+    # We have had issues with it being triggered multiple times per day,
+    # so we use a lock (12 hours) to avoid that.
+    lock_id = "{0}-lock".format(self.name)
+    with memcache_lock(lock_id, 60 * 60 * 12, self.app.oid) as acquired:
+        if acquired:
+            log.exception("Task has already been run recently, exiting.")
+            return
 
-    for i, user in enumerate(users):
-        structlog.contextvars.bind_contextvars(
-            user_username=user.username,
-            progress=f"{i}/{users_count}",
+        today_weekday = timezone.now().isoweekday()
+        days_ago = timezone.now() - datetime.timedelta(days=45)
+        users = User.objects.annotate(weekday=ExtractIsoWeekDay("last_login")).filter(
+            last_login__gt=days_ago,
+            socialaccount__isnull=False,
+            weekday=today_weekday,
         )
 
-        # Log an update every 50 users
-        if i % 50 == 0:
-            log.info("Progress on re-syncing RemoteRepository for active users.")
+        users_count = users.count()
+        structlog.contextvars.bind_contextvars(total_users=users_count)
+        log.info("Triggering re-sync of RemoteRepository for active users.")
 
-        try:
-            # NOTE: sync all the users/repositories in the same Celery process.
-            # Do not trigger a new task per user.
-            sync_remote_repositories(user.pk)
-        except Exception:
-            log.exception("There was a problem re-syncing RemoteRepository.")
+        for i, user in enumerate(users.iterator()):
+            structlog.contextvars.bind_contextvars(
+                user_username=user.username,
+                progress=f"{i}/{users_count}",
+            )
+
+            # Log an update every 50 users
+            if i % 50 == 0:
+                log.info("Progress on re-syncing RemoteRepository for active users.")
+
+            try:
+                # NOTE: sync all the users/repositories in the same Celery process.
+                # Do not trigger a new task per user.
+                # NOTE: We skip the GitHub App, since all the repositories
+                # and permissions are kept up to date via webhooks.
+                # Triggering a sync per-user, will re-sync the same installation
+                # multiple times.
+                sync_remote_repositories(
+                    user.pk,
+                    skip_githubapp=True,
+                )
+            except Exception:
+                log.exception("There was a problem re-syncing RemoteRepository.")
 
 
 # TODO: remove user_pk from the signature on the next release.

readthedocs/settings/base.py

@@ -693,14 +693,11 @@ def BUILD_MEMORY_LIMIT(self):
             "schedule": crontab(minute="*/30", hour="*"),
             "options": {"queue": "web"},
         },
-        # Stop running this task,
-        # it was rerunning multiple times per day,
-        # and causing celery instances to freeze up.
-        # "every-day-resync-remote-repositories": {
-        #     "task": "readthedocs.oauth.tasks.sync_active_users_remote_repositories",
-        #     "schedule": crontab(minute=30, hour=2),
-        #     "options": {"queue": "web"},
-        # },
+        "every-day-resync-remote-repositories": {
+            "task": "readthedocs.oauth.tasks.sync_active_users_remote_repositories",
+            "schedule": crontab(minute=30, hour=2),
+            "options": {"queue": "web"},
+        },
         "every-day-email-pending-custom-domains": {
             "task": "readthedocs.domains.tasks.email_pending_custom_domains",
             "schedule": crontab(minute=0, hour=3),