fix: serialize tenant poll runs

BACKLOG.md

@@ -10,7 +10,7 @@ or `SPRINTS.md` are not committed backlog until they have a corresponding issue.
 - [x] [#51 — Close the Python poller link-check DNS rebinding gap](https://github.com/CryptoJones/OSApplyTrack/issues/51)
 - [x] [#49 — Restrict forwarded-header trust to configured proxies](https://github.com/CryptoJones/OSApplyTrack/issues/49)
 - [x] [#50 — Add global JSON body caps and per-field/cardinality limits](https://github.com/CryptoJones/OSApplyTrack/issues/50)
-- [ ] [#52 — Serialize overlapping tenant poll runs](https://github.com/CryptoJones/OSApplyTrack/issues/52)
+- [x] [#52 — Serialize overlapping tenant poll runs](https://github.com/CryptoJones/OSApplyTrack/issues/52)
 
 ## Operations and scalability
 

api/ApplyTrack.Api/ApplyTrack.Api.csproj

@@ -5,7 +5,7 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <RootNamespace>ApplyTrack.Api</RootNamespace>
-    <Version>1.11.4</Version>
+    <Version>1.11.5</Version>
     <Authors>Aaron K. Clark</Authors>
     <Copyright>Copyright 2026 Aaron K. Clark</Copyright>
     <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "applytrack-poller"
-version = "1.11.4"
+version = "1.11.5"
 description = "Discovery poller for OSApplyTrack — fetches and scores remote job leads into shared Postgres."
 requires-python = ">=3.10"
 license = { text = "Apache-2.0" }

src/applytrack/worker.py

@@ -40,6 +40,8 @@
 
 logger = logging.getLogger(__name__)
 
+_TENANT_POLL_LOCK_PREFIX = "applytrack:tenant-poll:"
+
 
 class TenantRepo(LeadRepo, Protocol):
     """A :class:`~applytrack.poll.LeadRepo` that can also load its own profile.
@@ -65,6 +67,26 @@ def _active_tenant_ids(conn: psycopg.Connection) -> list[int]:
         return [int(row[0]) for row in cur.fetchall()]
 
 
+def _try_tenant_poll_lock(conn: psycopg.Connection, tenant_id: int) -> bool:
+    """Acquire this tenant's PostgreSQL session advisory lock without waiting."""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT pg_try_advisory_lock(hashtextextended(%s, 0))",
+            (f"{_TENANT_POLL_LOCK_PREFIX}{tenant_id}",),
+        )
+        row = cur.fetchone()
+    return bool(row and row[0])
+
+
+def _release_tenant_poll_lock(conn: psycopg.Connection, tenant_id: int) -> None:
+    """Release a tenant poll lock acquired by :func:`_try_tenant_poll_lock`."""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT pg_advisory_unlock(hashtextextended(%s, 0))",
+            (f"{_TENANT_POLL_LOCK_PREFIX}{tenant_id}",),
+        )
+
+
 def _gather_by_source(
     profiles: Iterable[Criteria], limit: int
 ) -> dict[str, list[Listing]]:
@@ -113,6 +135,7 @@ def run_all_tenants(
     repo_for: Callable[[int], TenantRepo] | None = None,
     gathered: dict[str, list[Listing]] | None = None,
     verify_links: bool = True,
+    locked_tenant_ids: set[int] | None = None,
 ) -> dict[int, list[str]]:
     """Poll every active tenant, returning ``{tenant_id: [staged slug names]}``.
 
@@ -122,6 +145,12 @@ def run_all_tenants(
     / ``repo_for`` / ``gathered`` seams let the offline tests drive the same
     fan-out with in-memory fakes and a fixed listing set (no DB, no network).
 
+    Production also holds a PostgreSQL session advisory lock for each tenant from
+    profile setup through source gathering and staging. A concurrent drain,
+    scheduled run, or second poller container skips an already-locked tenant
+    without waiting. ``locked_tenant_ids`` lets the queue drainer re-enqueue those
+    skipped requests for a later attempt.
+
     Per-tenant failures are isolated: a tenant whose poll raises is recorded with
     an empty result and the run continues.
     """
@@ -139,36 +168,70 @@ def repo_for(tid: int) -> TenantRepo:
         tenant_ids = _active_tenant_ids(conn)
     tenant_ids = list(tenant_ids)
 
+    acquired_locks: list[int] = []
+    pollable_tenant_ids: list[int] = []
+    for tid in tenant_ids:
+        if conn is None:
+            pollable_tenant_ids.append(tid)
+            continue
+        try:
+            if _try_tenant_poll_lock(conn, tid):
+                acquired_locks.append(tid)
+                pollable_tenant_ids.append(tid)
+                continue
+            if locked_tenant_ids is not None:
+                locked_tenant_ids.add(tid)
+            logger.info(
+                "poll skipped for tenant %s: another poll is already active",
+                tid,
+            )
+        except Exception:  # noqa: BLE001 - one lock failure must not abort the run
+            logger.warning(
+                "poll lock acquisition failed for tenant %s", tid, exc_info=True
+            )
+
     # Build the per-tenant repo + profile up front: this is where each tenant's
     # WHERE tenant_id scoping is fixed for the rest of the run. A tenant whose
     # setup raises is isolated here so the shared gather still runs for the rest.
     repos: dict[int, TenantRepo] = {}
     profiles: dict[int, Criteria] = {}
-    results: dict[int, list[str]] = {}
-    for tid in tenant_ids:
-        try:
-            repo = repo_for(tid)
-            profiles[tid] = repo.load_profile()
-            repos[tid] = repo
-        except Exception:  # noqa: BLE001 - one tenant's failure must not abort the rest
-            results[tid] = []
-            logger.warning("poll setup failed for tenant %s", tid, exc_info=True)
+    results: dict[int, list[str]] = {
+        tid: [] for tid in tenant_ids if tid not in pollable_tenant_ids
+    }
+    try:
+        for tid in pollable_tenant_ids:
+            try:
+                repo = repo_for(tid)
+                profiles[tid] = repo.load_profile()
+                repos[tid] = repo
+            except Exception:  # noqa: BLE001 - isolate one tenant's failure
+                results[tid] = []
+                logger.warning("poll setup failed for tenant %s", tid, exc_info=True)
 
-    if gathered is None:
-        gathered = _gather_by_source(profiles.values(), limit_per_source)
+        if gathered is None:
+            gathered = _gather_by_source(profiles.values(), limit_per_source)
 
-    for tid in tenant_ids:
-        if tid not in repos:
-            continue  # setup failed above; its empty result is already recorded
-        try:
-            listings = _select_for_profile(gathered, profiles[tid])
-            results[tid] = score_and_stage(
-                repos[tid], profiles[tid], listings, verify_links=verify_links
-            )
-        except Exception:  # noqa: BLE001 - one tenant's failure must not abort the rest
-            results[tid] = []
-            logger.warning("poll failed for tenant %s", tid, exc_info=True)
-    return results
+        for tid in pollable_tenant_ids:
+            if tid not in repos:
+                continue  # setup failed above; its empty result is already recorded
+            try:
+                listings = _select_for_profile(gathered, profiles[tid])
+                results[tid] = score_and_stage(
+                    repos[tid], profiles[tid], listings, verify_links=verify_links
+                )
+            except Exception:  # noqa: BLE001 - isolate one tenant's failure
+                results[tid] = []
+                logger.warning("poll failed for tenant %s", tid, exc_info=True)
+        return results
+    finally:
+        if conn is not None:
+            for tid in acquired_locks:
+                try:
+                    _release_tenant_poll_lock(conn, tid)
+                except Exception:  # noqa: BLE001 - release the remaining locks
+                    logger.warning(
+                        "poll lock release failed for tenant %s", tid, exc_info=True
+                    )
 
 
 def drain_requests(
@@ -193,9 +256,23 @@ def drain_requests(
         tenant_ids = sorted({int(row[0]) for row in cur.fetchall()})
     if not tenant_ids:
         return {}
-    return run_all_tenants(
+    locked_tenant_ids: set[int] = set()
+    results = run_all_tenants(
         conn,
         limit_per_source=limit_per_source,
         tenant_ids=tenant_ids,
         repo_for=repo_for,
+        locked_tenant_ids=locked_tenant_ids,
     )
+    if locked_tenant_ids:
+        with conn.cursor() as cur:
+            for tenant_id in sorted(locked_tenant_ids):
+                cur.execute(
+                    "INSERT INTO poll_requests (tenant_id) VALUES (%s)",
+                    (tenant_id,),
+                )
+        logger.info(
+            "requeued %s tenant poll request(s) blocked by active polls",
+            len(locked_tenant_ids),
+        )
+    return results

tests/test_poll.py

@@ -99,7 +99,7 @@ def __enter__(self) -> FakeCursor:
     def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
         return None
 
-    def execute(self, sql: str) -> None:
+    def execute(self, sql: str, params: object = None) -> None:
         self.sql = sql
 
     def fetchall(self) -> list[tuple[int]]:
@@ -438,10 +438,12 @@ def fake_run_all_tenants(
         limit_per_source: int,
         tenant_ids: list[int],
         repo_for: object,
+        locked_tenant_ids: set[int],
     ) -> dict[int, list[str]]:
         calls["limit"] = limit_per_source
         calls["tenant_ids"] = tenant_ids
         calls["repo_for"] = repo_for
+        calls["locked_tenant_ids"] = locked_tenant_ids
         return {1: ["acme.md"], 2: ["globex.md"]}
 
     monkeypatch.setattr("applytrack.worker.run_all_tenants", fake_run_all_tenants)
@@ -453,4 +455,105 @@ def fake_run_all_tenants(
         "limit": 7,
         "tenant_ids": [1, 2],
         "repo_for": None,
+        "locked_tenant_ids": set(),
     }
+
+
+def test_run_all_tenants_skips_locked_tenant_and_releases_acquired_lock(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    class LockCursor:
+        def __init__(self, connection: LockConnection) -> None:
+            self.connection = connection
+            self.row: tuple[bool] | None = None
+
+        def __enter__(self) -> LockCursor:
+            return self
+
+        def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
+            return None
+
+        def execute(self, sql: str, params: tuple[str]) -> None:
+            tenant_id = int(params[0].rsplit(":", 1)[1])
+            if "pg_try_advisory_lock" in sql:
+                self.row = (tenant_id not in self.connection.locked,)
+            else:
+                self.connection.released.append(tenant_id)
+
+        def fetchone(self) -> tuple[bool] | None:
+            return self.row
+
+    class LockConnection:
+        def __init__(self) -> None:
+            self.locked = {1}
+            self.released: list[int] = []
+
+        def cursor(self) -> LockCursor:
+            return LockCursor(self)
+
+    connection = LockConnection()
+    repos = {
+        1: FakeRepo(profile=Criteria(keywords=["engineer"])),
+        2: FakeRepo(profile=Criteria(keywords=["engineer"])),
+    }
+    locked_tenant_ids: set[int] = set()
+    with caplog.at_level(logging.INFO):
+        results = run_all_tenants(
+            connection,  # type: ignore[arg-type]
+            tenant_ids=[1, 2],
+            repo_for=lambda tid: repos[tid],
+            gathered={},
+            verify_links=False,
+            locked_tenant_ids=locked_tenant_ids,
+        )
+
+    assert results == {1: [], 2: []}
+    assert locked_tenant_ids == {1}
+    assert connection.released == [2]
+    assert any("another poll is already active" in r.getMessage() for r in caplog.records)
+
+
+def test_drain_requests_requeues_tenant_with_active_poll(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class QueueCursor:
+        def __init__(self, connection: QueueConnection) -> None:
+            self.connection = connection
+
+        def __enter__(self) -> QueueCursor:
+            return self
+
+        def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
+            return None
+
+        def execute(self, sql: str, params: tuple[int] | None = None) -> None:
+            if sql.startswith("INSERT"):
+                assert params is not None
+                self.connection.requeued.append(params[0])
+
+        def fetchall(self) -> list[tuple[int]]:
+            return [(7,)]
+
+    class QueueConnection:
+        def __init__(self) -> None:
+            self.requeued: list[int] = []
+
+        def cursor(self) -> QueueCursor:
+            return QueueCursor(self)
+
+    def fake_run_all_tenants(
+        conn: QueueConnection,
+        *,
+        limit_per_source: int,
+        tenant_ids: list[int],
+        repo_for: object,
+        locked_tenant_ids: set[int],
+    ) -> dict[int, list[str]]:
+        locked_tenant_ids.add(tenant_ids[0])
+        return {tenant_ids[0]: []}
+
+    monkeypatch.setattr("applytrack.worker.run_all_tenants", fake_run_all_tenants)
+    connection = QueueConnection()
+
+    assert drain_requests(connection) == {7: []}  # type: ignore[arg-type]
+    assert connection.requeued == [7]