From b2fddabc1d622d7cd2d71485456b5787997b6017 Mon Sep 17 00:00:00 2001 From: "Aaron K. Clark" Date: Wed, 29 Jul 2026 10:56:17 -0500 Subject: [PATCH] fix: serialize tenant poll runs --- BACKLOG.md | 2 +- api/ApplyTrack.Api/ApplyTrack.Api.csproj | 2 +- pyproject.toml | 2 +- src/applytrack/worker.py | 125 ++++++++++++++++++----- tests/test_poll.py | 105 ++++++++++++++++++- uv.lock | 2 +- 6 files changed, 209 insertions(+), 29 deletions(-) diff --git a/BACKLOG.md b/BACKLOG.md index cd44842..46f64a0 100644 --- a/BACKLOG.md +++ b/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 diff --git a/api/ApplyTrack.Api/ApplyTrack.Api.csproj b/api/ApplyTrack.Api/ApplyTrack.Api.csproj index 36c6df5..c4406ca 100644 --- a/api/ApplyTrack.Api/ApplyTrack.Api.csproj +++ b/api/ApplyTrack.Api/ApplyTrack.Api.csproj @@ -5,7 +5,7 @@ enable enable ApplyTrack.Api - 1.11.4 + 1.11.5 Aaron K. Clark Copyright 2026 Aaron K. Clark Apache-2.0 diff --git a/pyproject.toml b/pyproject.toml index 0f0d04e..a4e9f2f 100644 --- a/pyproject.toml +++ b/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" } diff --git a/src/applytrack/worker.py b/src/applytrack/worker.py index 09647f9..7e94cc2 100644 --- a/src/applytrack/worker.py +++ b/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 diff --git a/tests/test_poll.py b/tests/test_poll.py index 26544dd..8f615b3 100644 --- a/tests/test_poll.py +++ b/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] diff --git a/uv.lock b/uv.lock index 051528d..3547888 100644 --- a/uv.lock +++ b/uv.lock @@ -23,7 +23,7 @@ wheels = [ [[package]] name = "applytrack-poller" -version = "1.11.4" +version = "1.11.5" source = { editable = "." } dependencies = [ { name = "defusedxml" },