fix: reduce sr.ht poll API load with cached repo discovery

5 files changed, 83 insertions, 24 deletions

diff --git a/src/srht_contrib/jobs/poller.py b/src/srht_contrib/jobs/poller.py
index 94a7d96..b2fe70f 100644
--- a/src/srht_contrib/jobs/poller.py
+++ b/src/srht_contrib/jobs/poller.py
@@ -91,7 +91,6 @@ class PollerService:
         inserted = 0
         inserted += self._poll_service(db, actor, self.todo_service.service_name, self.todo_service.fetch_recent_events)
         self._sync_tracked_repositories(db, actor)
-        git_repositories = self._tracked_repositories_for_actor(db, actor)
         inserted += self._poll_service(
             db,
             actor,
@@ -99,7 +98,7 @@ class PollerService:
             lambda actor, since: self.git_service.fetch_recent_events(
                 actor=actor,
                 since=since,
-                repositories=git_repositories,
+                db=db,
             ),
         )
         return inserted
@@ -172,15 +171,6 @@ class PollerService:
                 )
         db.flush()
 
-    def _tracked_repositories_for_actor(self, db: Session, actor: str) -> list[str]:
-        rows = db.scalars(
-            select(TrackedRepository.repo_name)
-            .where(TrackedRepository.service == self.git_service.service_name)
-            .where(TrackedRepository.actor == actor)
-            .order_by(TrackedRepository.repo_name)
-        ).all()
-        return list(rows)
-
     def _update_tracked_actor_poll_state(self, db: Session, actor: str, status: str, error: str | None) -> None:
         tracked_actor = self.track_actor_request(db, actor, update_last_requested=False)
         tracked_actor.last_poll_status = status
diff --git a/src/srht_contrib/main.py b/src/srht_contrib/main.py
index 3737569..d8284d6 100644
--- a/src/srht_contrib/main.py
+++ b/src/srht_contrib/main.py
@@ -26,8 +26,8 @@ logger = logging.getLogger(__name__)
 
 
 def build_poller(settings: Settings) -> PollerService:
-    todo_client = SourceHutGraphQLClient(settings.todo_srht_endpoint, settings.srht_token)
-    git_client = SourceHutGraphQLClient(settings.git_srht_endpoint, settings.srht_token)
+    todo_client = SourceHutGraphQLClient(settings.todo_srht_endpoint, settings.srht_token, request_delay=settings.srht_request_delay_seconds)
+    git_client = SourceHutGraphQLClient(settings.git_srht_endpoint, settings.srht_token, request_delay=settings.srht_request_delay_seconds)
     todo_service = TodoIngestionService(todo_client, settings)
     git_service = GitIngestionService(git_client, settings)
     return PollerService(todo_service=todo_service, git_service=git_service, settings=settings)
diff --git a/src/srht_contrib/models.py b/src/srht_contrib/models.py
index 475bd22..e0725ff 100644
--- a/src/srht_contrib/models.py
+++ b/src/srht_contrib/models.py
@@ -81,6 +81,16 @@ class TrackedActor(Base):
     last_backfill_error: Mapped[str | None] = mapped_column(Text, nullable=True)
 
 
+class DiscoveredRepository(Base):
+    __tablename__ = "discovered_repositories"
+    __table_args__ = (UniqueConstraint("actor", "name", name="uq_discovered_repository_actor_name"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    actor: Mapped[str] = mapped_column(String(255), nullable=False)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+
 class ServiceBackfillState(Base):
     __tablename__ = "service_backfill_states"
     __table_args__ = (UniqueConstraint("actor", "service", "scope", name="uq_service_backfill_state_actor_service_scope"),)
diff --git a/src/srht_contrib/services/git.py b/src/srht_contrib/services/git.py
index 7d5360b..26b2957 100644
--- a/src/srht_contrib/services/git.py
+++ b/src/srht_contrib/services/git.py
@@ -6,7 +6,11 @@ from datetime import UTC, datetime, timedelta
 import logging
 from typing import Any
 
+from sqlalchemy import delete, select
+from sqlalchemy.orm import Session
+
 from srht_contrib.config import Settings
+from srht_contrib.models import DiscoveredRepository, TrackedRepository
 from srht_contrib.schemas import NormalizedEvent
 from srht_contrib.services.srht_client import SourceHutClientError, SourceHutGraphQLClient
 from srht_contrib.services.types import BackfillBatchResult
@@ -88,9 +92,10 @@ class GitIngestionService:
         actor: str,
         since: datetime | None = None,
         repositories: list[str] | None = None,
+        db: Session | None = None,
     ) -> GitPollResult:
         since_dt = ensure_utc(since or (datetime.now(tz=UTC) - timedelta(days=30)))
-        discovered_repositories = repositories or self._repositories_for_actor(actor)
+        discovered_repositories = repositories or self._repositories_for_actor(actor, db)
         if not discovered_repositories:
             logger.info("git poll skipped for actor=%s because no repositories were discovered", actor)
             return GitPollResult(events=[], cursor=datetime.now(tz=UTC).isoformat())
@@ -108,18 +113,44 @@ class GitIngestionService:
         logger.info("git poll complete for actor=%s normalized_events=%s", actor, len(events))
         return GitPollResult(events=events, cursor=datetime.now(tz=UTC).isoformat())
 
-    def _repositories_for_actor(self, actor: str) -> list[str]:
+    def _repositories_for_actor(self, actor: str, db: Session | None = None) -> list[str]:
         configured = {
             self._canonical_repository_name(actor, repository)
             for repository in self.settings.git_tracked_repositories
         }
-        discovered = set(self._discover_owned_repositories(actor))
-        repositories = sorted(configured | discovered)
+        tracked = set()
+        discovered = set()
+        now = datetime.now(tz=UTC)
+
+        if db is not None:
+            tracked = set(
+                db.scalars(
+                    select(TrackedRepository.repo_name)
+                    .where(TrackedRepository.service == self.service_name)
+                    .where(TrackedRepository.actor == actor)
+                ).all()
+            )
+            cutoff = now - timedelta(seconds=self.settings.git_repo_discovery_ttl_seconds)
+            discovered = set(
+                db.scalars(
+                    select(DiscoveredRepository.name)
+                    .where(DiscoveredRepository.actor == actor)
+                    .where(DiscoveredRepository.discovered_at > cutoff)
+                ).all()
+            )
+
+        if not discovered:
+            discovered = set(self._discover_owned_repositories(actor))
+            if db is not None:
+                self._refresh_discovered_repositories(db, actor, discovered, discovered_at=now)
+
+        repositories = sorted(configured | tracked | discovered)
         logger.info(
-            "git repositories selected for actor=%s count=%s configured=%s discovered=%s",
+            "git repositories selected for actor=%s count=%s configured=%s tracked=%s discovered=%s",
             actor,
             len(repositories),
             len(configured),
+            len(tracked),
             len(discovered),
         )
         return repositories
@@ -268,7 +299,8 @@ class GitIngestionService:
                 repository_owner = ((repository.get("owner") or {}).get("canonicalName") or actor).strip()
                 if not name or not repository_owner:
                     continue
-                repositories.append(f"{repository_owner}/{name}")
+                repo_name = f"{repository_owner}/{name}"
+                repositories.append(repo_name)
 
             if not cursor:
                 break
@@ -276,6 +308,25 @@ class GitIngestionService:
         return repositories
 
     @staticmethod
+    def _refresh_discovered_repositories(
+        db: Session,
+        actor: str,
+        repositories: set[str],
+        *,
+        discovered_at: datetime,
+    ) -> None:
+        db.execute(delete(DiscoveredRepository).where(DiscoveredRepository.actor == actor))
+        for repository in sorted(repositories):
+            db.add(
+                DiscoveredRepository(
+                    actor=actor,
+                    name=repository,
+                    discovered_at=discovered_at,
+                )
+            )
+        db.flush()
+
+    @staticmethod
     def _canonical_repository_name(default_actor: str, repository: str) -> str:
         owner, repo_name = GitIngestionService._split_repository(default_actor, repository)
         return f"~{owner}/{repo_name}"
diff --git a/src/srht_contrib/services/srht_client.py b/src/srht_contrib/services/srht_client.py
index ee09bff..1118f0d 100644
--- a/src/srht_contrib/services/srht_client.py
+++ b/src/srht_contrib/services/srht_client.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import time
 from typing import Any
 
 import httpx
@@ -27,11 +28,13 @@ class SourceHutGraphQLClient:
         *,
         timeout: float = 15.0,
         max_retries: int = 2,
+        request_delay: float = 0.5,
         transport: httpx.BaseTransport | None = None,
     ) -> None:
         self.endpoint = endpoint
         self.timeout = timeout
         self.max_retries = max_retries
+        self.request_delay = request_delay
         headers = {
             "Authorization": f"Bearer {token}",
             "Content-Type": "application/json",
@@ -42,7 +45,12 @@ class SourceHutGraphQLClient:
         payload = {"query": query, "variables": variables or {}}
         attempts = self.max_retries + 1
 
-        for attempt in range(1, attempts + 1):
+        for attempt in range(attempts):
+            if attempt > 0:
+                time.sleep(2 ** (attempt - 1))
+            elif self.request_delay > 0:
+                time.sleep(self.request_delay)
+
             try:
                 response = self._client.post(self.endpoint, json=payload)
                 response.raise_for_status()
@@ -51,21 +59,21 @@ class SourceHutGraphQLClient:
                 logger.warning(
                     "SourceHut HTTP failure from %s on attempt %s/%s: status=%s",
                     self.endpoint,
-                    attempt,
+                    attempt + 1,
                     attempts,
                     exc.response.status_code,
                 )
-                if exc.response.status_code >= 500 and attempt < attempts:
+                if exc.response.status_code >= 500 and attempt < attempts - 1:
                     continue
                 raise SourceHutClientError(f"HTTP error from SourceHut: {exc.response.status_code}") from exc
             except httpx.HTTPError as exc:
                 logger.warning(
                     "SourceHut network failure from %s on attempt %s/%s",
                     self.endpoint,
-                    attempt,
+                    attempt + 1,
                     attempts,
                 )
-                if attempt < attempts:
+                if attempt < attempts - 1:
                     continue
                 raise SourceHutClientError("Network error while contacting SourceHut") from exc