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" },