From 71ef2591c8836dc5e5fcea2360b6771efa4297ce Mon Sep 17 00:00:00 2001 From: "Aaron K. Clark" Date: Mon, 29 Jun 2026 15:57:19 -0500 Subject: [PATCH] fix(poller): distinct URLs are distinct leads + source diagnostics (1.7.7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports a code review from the sibling applytrack repo. Two of the five findings apply to this codebase; the other three are already handled by the .NET/Postgres architecture (see below). Finding 1 — dedup dropped distinct postings. score_and_stage deduped scraped listings by a normalized company+role slug that STRIPPED trailing parentheticals, so per-city variants of one role (each a distinct application URL) collapsed to one lead and the rest were silently recorded as seen and dropped — e.g. five LawnStarter city postings staged one and dropped four. Now: - _norm_slug keeps the full role title (parentheticals included). - Seen.has makes the posting URL authoritative: a distinct URL is a distinct lead; the slug is only the fallback key for URL-less listings. - score_and_stage records seen keys per outcome (staged / filtered / link-dead) instead of up front, and no longer marks a listing seen when staging it raises, so a genuine write failure is retried on a later run. - PollRepo.add_lead suffixes -N on a slug-name collision (two distinct postings whose titles normalize to the same name) rather than letting the second drop. Finding 4 — source failures were swallowed silently, so a dead/renamed job source looked like an empty market. poll._gather and worker._gather_by_source now log each failing source at WARNING (and tolerate any exception, not just HTTP/parse), and the worker logs per-tenant setup/poll failures it had been swallowing. Not applicable here (verified against this codebase, no change made): - Finding 2 (non-atomic file writes): the poller is Postgres-backed; it writes no state files. - Finding 3 (optimistic-concurrency clobbers): the draft endpoint writes a separate cover_letters table, not the versioned application row; the blacklist batch (PassOpenLeadsAsync) is a single atomic UPDATE of status/version guarded by `status IN ('lead','ready')`, so it preserves unrelated fields and won't flip an advanced row — no read-modify-write race to guard. - Finding 5 (stale docs): the README/SPRINTS already describe poll as discovery-only. Tests: ruff + mypy --strict + bandit clean; pytest green (dedup tests updated for URL-authoritative behavior, plus new distinct-URL, unique-name, url-less-slug, and source-failure-logging cases). dotnet build of the API clean. Version 1.7.6 -> 1.7.7 (patch) in pyproject.toml + ApplyTrack.Api.csproj. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01W84TwzhCdhmoLoxXYV2NuH --- api/ApplyTrack.Api/ApplyTrack.Api.csproj | 2 +- pyproject.toml | 2 +- src/applytrack/db.py | 34 +++++-- src/applytrack/poll.py | 92 ++++++++++++------ src/applytrack/worker.py | 15 ++- tests/test_poll.py | 113 +++++++++++++++++++++-- 6 files changed, 212 insertions(+), 46 deletions(-) diff --git a/api/ApplyTrack.Api/ApplyTrack.Api.csproj b/api/ApplyTrack.Api/ApplyTrack.Api.csproj index b56b971..4cb1973 100644 --- a/api/ApplyTrack.Api/ApplyTrack.Api.csproj +++ b/api/ApplyTrack.Api/ApplyTrack.Api.csproj @@ -5,7 +5,7 @@ enable enable ApplyTrack.Api - 1.7.6 + 1.7.7 Aaron K. Clark Copyright 2026 Aaron K. Clark Apache-2.0 diff --git a/pyproject.toml b/pyproject.toml index ba4bdb8..878787e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "applytrack-poller" -version = "1.7.6" +version = "1.7.7" 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/db.py b/src/applytrack/db.py index de215c9..66ab533 100644 --- a/src/applytrack/db.py +++ b/src/applytrack/db.py @@ -30,12 +30,17 @@ from applytrack.store import AppFields, filename_for # A plain INSERT (not the importer's upsert): a slug already present is a genuine -# collision to skip, never an overwrite of a row the user may have since edited. +# collision, so :meth:`PollRepo.add_lead` retries under a suffixed name rather than +# overwriting a row the user may have since edited. _INSERT_SQL = f""" INSERT INTO applications (tenant_id, name, {", ".join(_FIELD_COLUMNS)}) VALUES (%(tenant_id)s, %(name)s, {", ".join(f"%({c})s" for c in _FIELD_COLUMNS)}) """ +# How many slug names (``stem.md``, ``stem-2.md`` … ``stem-N.md``) ``add_lead`` will +# try before giving up — a backstop against an unforeseen unique-violation loop. +_MAX_NAME_ATTEMPTS = 50 + # Column order mirrors search_profiles; the keys match Criteria.from_dict's dict. _PROFILE_COLUMNS = ( "keywords", "default_lane", "min_fit_score", "remote_only", @@ -126,10 +131,25 @@ def mark_seen(self, url_key: str, slug_key: str) -> None: def add_lead(self, fields: AppFields) -> str: """Stage one new lead, returning its slug ``name``. - Raises :class:`psycopg.errors.UniqueViolation` if the slug already exists - for this tenant; the caller treats that as "skip this listing". + When the natural ``company-role`` slug already exists for this tenant — + two genuinely distinct postings (distinct URLs) whose titles normalize to + the same name — a numeric ``-N`` suffix is appended so the second is + staged rather than dropped. Autocommit means each failed INSERT stands + alone, so the retry is clean. Only after :data:`_MAX_NAME_ATTEMPTS` free + names are exhausted does the underlying + :class:`psycopg.errors.UniqueViolation` propagate. """ - name = filename_for(fields) - with self._conn.cursor() as cur: - cur.execute(_INSERT_SQL, row_params(name, fields, self._t)) - return name + base = filename_for(fields) + stem = base[:-3] if base.endswith(".md") else base + name = base + attempt = 1 + while True: + try: + with self._conn.cursor() as cur: + cur.execute(_INSERT_SQL, row_params(name, fields, self._t)) + return name + except psycopg.errors.UniqueViolation: + attempt += 1 + if attempt > _MAX_NAME_ATTEMPTS: + raise + name = f"{stem}-{attempt}.md" diff --git a/src/applytrack/poll.py b/src/applytrack/poll.py index 526dd3b..16dd079 100644 --- a/src/applytrack/poll.py +++ b/src/applytrack/poll.py @@ -16,12 +16,15 @@ This runtime no longer touches the disk store; it shares one Postgres with the API and is driven against a tenant-scoped repo (see :class:`LeadRepo`). -Deduplication is load-bearing — a company must never be pinged twice. For each -run an in-memory ledger is seeded from the tenant's existing ``applications`` -rows, and every listing is checked against it using two independent keys: the -normalized listing URL and the normalized ``company + role`` slug. If *either* -key has been seen, the listing is skipped. New keys are recorded even when -staging fails, so a transient error never causes a re-ping. +Deduplication is load-bearing — the same posting must never be staged twice. For +each run an in-memory ledger is seeded from the tenant's existing ``applications`` +rows. The posting URL is the authoritative key: a listing with a *distinct* URL is +a distinct lead, even when its ``company + role`` slug matches one already seen +(e.g. one role posted per city). The slug is only the fallback key for the +minority of sources that give no URL. Keys are recorded for any listing we +evaluate — staged, filtered, or link-dead — so we don't re-process it; the lone +exception is a genuine *write* failure, which leaves the listing unrecorded so a +later run can retry it. """ from __future__ import annotations @@ -51,7 +54,6 @@ _TAG_RE = re.compile(r"<[^>]+>") _WS_RE = re.compile(r"\s+") _SLUG_RE = re.compile(r"[^a-z0-9]+") -_PARENS_RE = re.compile(r"\([^)]*\)") _URL_RE = re.compile(r"https?://[^\s<>\"')]+") # Signals that a role is remote-friendly, for the optional remote-only filter. @@ -96,12 +98,12 @@ def _norm_url(url: str) -> str: def _norm_slug(company: str, role: str) -> str: """Normalized ``company + role`` key, robust to spacing/punctuation. - Trailing parentheticals are dropped before keying, so the same role posted - across locations collapses to one key and a company is not pinged once per - city. This errs toward over-dedup, the safe direction for not annoying - employers. + The *full* role is kept (parentheticals included), so per-city variants like + ``Engineer (Campinas)`` vs ``Engineer (São Paulo)`` are distinct keys — they + are distinct openings. This is the *fallback* dedup key, used only for + listings that carry no URL; when a URL is present it is authoritative (see + :meth:`Seen.has`), so two distinct URLs are never collapsed by this slug. """ - role = _PARENS_RE.sub(" ", role) return _SLUG_RE.sub("-", f"{company} {role}".lower()).strip("-") @@ -177,8 +179,17 @@ class Seen: sink: Callable[[str, str], None] | None = None def has(self, url: str, slug: str) -> bool: + """Have we already processed this listing? + + The URL is authoritative: when the listing has one, the verdict is purely + whether *that* URL was seen — a distinct URL is a distinct lead, never + suppressed by a colliding ``company + role`` slug. The slug is consulted + only for URL-less listings, where it is all we have. + """ nu = _norm_url(url) - return (bool(nu) and nu in self.urls) or (bool(slug) and slug in self.slugs) + if nu: + return nu in self.urls + return bool(slug) and slug in self.slugs def note(self, url: str, slug: str) -> tuple[str, str]: """Record keys in memory only; return the normalized keys newly added.""" @@ -619,13 +630,26 @@ def fetch_lever(client: httpx.Client, limit: int, slug: str) -> list[Listing]: def make_ats_fetcher(board: AtsBoard) -> Fetcher | None: - """Bind a company slug to its provider's fetcher, yielding a plain Fetcher.""" + """Bind a company slug to its provider's fetcher, yielding a plain Fetcher. + + The returned callable carries a ``provider:slug`` ``__name__`` so a failure of + one board is named in the poll diagnostics rather than logged as ````. + """ slug = board.slug if board.provider == "greenhouse": - return lambda client, limit: fetch_greenhouse(client, limit, slug) - if board.provider == "lever": - return lambda client, limit: fetch_lever(client, limit, slug) - return None + + def _fetch(client: httpx.Client, limit: int) -> list[Listing]: + return fetch_greenhouse(client, limit, slug) + + elif board.provider == "lever": + + def _fetch(client: httpx.Client, limit: int) -> list[Listing]: + return fetch_lever(client, limit, slug) + + else: + return None + _fetch.__name__ = f"{board.provider}:{slug}" + return _fetch def build_fetchers(criteria: Criteria) -> list[Fetcher]: @@ -643,14 +667,21 @@ def build_fetchers(criteria: Criteria) -> list[Fetcher]: def _gather(fetchers: Iterable[Fetcher], limit: int) -> list[Listing]: - """Run each fetcher, tolerating per-source failures (offline, 4xx, parse).""" + """Run each fetcher, tolerating per-source failures (offline, 4xx, parse). + + A failing source is skipped but logged at WARNING with its name, so a + dead/renamed source is visible in the worker logs instead of silently looking + like an empty job market. Any exception is caught — one broken source (e.g. a + changed upstream schema) must never abort the whole run. + """ listings: list[Listing] = [] with httpx.Client(timeout=20.0, follow_redirects=True, headers=BROWSER_HEADERS) as client: for fetch in fetchers: + label = getattr(fetch, "__name__", repr(fetch)).removeprefix("fetch_") try: listings.extend(fetch(client, limit)) - except (httpx.HTTPError, ValueError, KeyError): - continue + except Exception: # noqa: BLE001 - one bad source must not abort the run + logger.warning("poll source %s failed", label, exc_info=True) return listings @@ -743,14 +774,17 @@ def score_and_stage( slug = _norm_slug(item.company, item.role) if seen.has(item.link, slug): continue - # Record keys up front so a later failure still can't cause a re-ping. - seen.add(item.link, slug) + # We act on this listing one way or another below, so it is recorded + # as seen *except* on a genuine write failure (handled at stage time), + # which leaves it for a later run to retry. if not _passes_location(item, profile): + seen.add(item.link, slug) continue score, hits = classify(item.role, item.description, profile.keywords) if not hits or score < profile.min_fit_score: + seen.add(item.link, slug) continue # Block dead postings: don't create an entry we can't actually open. @@ -759,20 +793,26 @@ def score_and_stage( and item.link and not is_reachable(item.link, client=verify_client) ): + seen.add(item.link, slug) continue fields = _to_fields(item, profile.default_lane, score, hits) try: + # add_lead suffixes -N on a slug-name collision, so a distinct URL + # whose name matches an existing row is staged, not dropped. name = repo.add_lead(fields) except psycopg.errors.UniqueViolation: - logger.debug( - "skipping duplicate lead for %s: %s", + # add_lead exhausted its name attempts: surface and leave unseen + # so a later run retries it. + logger.warning( + "stage failed (slug collision) for %s — %s", item.company, item.role, exc_info=True, ) continue - # Leads stage as-is; the cover letter is drafted on demand later. + # Staged: record keys, then collect. Cover letter is drafted on demand. + seen.add(item.link, slug) added.append(name) finally: if verify_client is not None: diff --git a/src/applytrack/worker.py b/src/applytrack/worker.py index 8a9a7ed..09647f9 100644 --- a/src/applytrack/worker.py +++ b/src/applytrack/worker.py @@ -19,6 +19,7 @@ from __future__ import annotations +import logging from collections.abc import Callable, Iterable from typing import Protocol @@ -37,6 +38,8 @@ score_and_stage, ) +logger = logging.getLogger(__name__) + class TenantRepo(LeadRepo, Protocol): """A :class:`~applytrack.poll.LeadRepo` that can also load its own profile. @@ -69,7 +72,9 @@ def _gather_by_source( Built-in sources key by name (``"remotive"``); ATS boards key by ``"{provider}:{slug}"`` — the same keys :func:`_select_for_profile` routes by. - Per-source failures yield an empty bucket rather than aborting the gather. + A failing source yields an empty bucket and is logged at WARNING (rather than + aborting the gather or vanishing silently), so a dead/renamed source is visible + in the logs instead of looking like an empty market for every tenant. """ builtin: set[str] = set() boards: dict[str, AtsBoard] = {} @@ -85,16 +90,18 @@ def _gather_by_source( for name in sorted(builtin): try: gathered[name] = SOURCE_FETCHERS[name](client, limit) - except (httpx.HTTPError, ValueError, KeyError): + except Exception: # noqa: BLE001 - one bad source must not abort the gather gathered[name] = [] + logger.warning("poll source %s failed", name, exc_info=True) for key, board in boards.items(): fetcher = make_ats_fetcher(board) if fetcher is None: continue try: gathered[key] = fetcher(client, limit) - except (httpx.HTTPError, ValueError, KeyError): + except Exception: # noqa: BLE001 - one bad source must not abort the gather gathered[key] = [] + logger.warning("poll source %s failed", key, exc_info=True) return gathered @@ -145,6 +152,7 @@ def repo_for(tid: int) -> TenantRepo: 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) if gathered is None: gathered = _gather_by_source(profiles.values(), limit_per_source) @@ -159,6 +167,7 @@ def repo_for(tid: int) -> TenantRepo: ) 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 diff --git a/tests/test_poll.py b/tests/test_poll.py index 4ca394e..26544dd 100644 --- a/tests/test_poll.py +++ b/tests/test_poll.py @@ -4,12 +4,16 @@ from __future__ import annotations +import logging + +import httpx import psycopg import pytest from applytrack.criteria import AtsBoard, Criteria from applytrack.poll import ( Listing, + _gather, _looks_remote, _passes_location, build_fetchers, @@ -49,6 +53,7 @@ def __init__( (self._seen_urls if kind == "url" else self._seen_slugs).add(key) self._profile = profile if profile is not None else Criteria() self.added: list[AppFields] = [] + self._names: set[str] = set() def load_profile(self) -> Criteria: return self._profile @@ -69,9 +74,16 @@ def mark_seen(self, url_key: str, slug_key: str) -> None: self._seen_slugs.add(slug_key) def add_lead(self, fields: AppFields) -> str: - name = filename_for(fields) - if any(filename_for(f) == name for f in self.added): - raise ValueError(f"an application named {name!r} already exists") + # Mirror PollRepo.add_lead: suffix -N on a slug-name collision so two + # genuinely distinct postings can coexist rather than the second failing. + base = filename_for(fields) + stem = base[:-3] if base.endswith(".md") else base + name = base + n = 1 + while name in self._names: + n += 1 + name = f"{stem}-{n}.md" + self._names.add(name) self.added.append(fields) return name @@ -163,7 +175,8 @@ def test_run_poll_stages_matches_with_default_lane() -> None: assert fields.company == "Acme" -def test_run_poll_dedupes_same_role() -> None: +def test_run_poll_dedupes_same_url() -> None: + # The same posting seen twice (identical URL) stages exactly once. repo = FakeRepo() c = Criteria(keywords=["engineer"]) added = run_poll( @@ -171,7 +184,55 @@ def test_run_poll_dedupes_same_role() -> None: c, listings=[ _lead("Acme", "Backend Engineer", link="https://acme.co/1"), - _lead("Acme", "Backend Engineer (Remote)", link="https://acme.co/2"), + _lead("Acme", "Backend Engineer", link="https://acme.co/1"), + ], + ) + assert len(added) == 1 + + +def test_run_poll_keeps_distinct_urls_for_role_variants() -> None: + # Per-city variants of one role have distinct URLs -> distinct leads. The old + # behavior stripped the parenthetical, collapsed them, and dropped all but one. + repo = FakeRepo() + c = Criteria(keywords=["engineer"]) + added = run_poll( + repo, + c, + listings=[ + _lead("Acme", "Backend Engineer (Campinas)", link="https://acme.co/1"), + _lead("Acme", "Backend Engineer (São Paulo)", link="https://acme.co/2"), + ], + ) + assert len(added) == 2 + + +def test_score_and_stage_unique_names_for_same_title_distinct_urls() -> None: + # Same role title (location only in a separate field) -> same slug stem but + # distinct URLs. Each must get its own row via a -N suffix, not be dropped. + repo = FakeRepo() + c = Criteria(keywords=["engineer"]) + added = score_and_stage( + repo, + c, + listings=[ + _lead("Acme", "Backend Engineer", link="https://acme.co/1", location="NYC"), + _lead("Acme", "Backend Engineer", link="https://acme.co/2", location="SF"), + ], + ) + assert len(added) == 2 + assert added == ["acme-backend-engineer.md", "acme-backend-engineer-2.md"] + + +def test_run_poll_dedupes_urlless_by_slug() -> None: + # With no URL, the slug is the only key, so identical url-less posts collapse. + repo = FakeRepo() + c = Criteria(keywords=["engineer"]) + added = run_poll( + repo, + c, + listings=[ + _lead("Acme", "Backend Engineer", link=""), + _lead("Acme", "Backend Engineer", link=""), ], ) assert len(added) == 1 @@ -250,6 +311,27 @@ def test_run_poll_applies_location_filter() -> None: assert added == [] +def test_gather_logs_and_continues_on_source_failure( + caplog: pytest.LogCaptureFixture, +) -> None: + # A failing source is skipped but logged by name; the others still contribute. + def boom(client: httpx.Client, limit: int) -> list[Listing]: + raise ValueError("upstream schema changed") + + boom.__name__ = "fetch_boom" + + def ok(client: httpx.Client, limit: int) -> list[Listing]: + return [_lead("Acme", "Engineer", link="https://acme.co/1")] + + ok.__name__ = "fetch_ok" + + with caplog.at_level(logging.WARNING): + listings = _gather([boom, ok], 10) + + assert [item.company for item in listings] == ["Acme"] + assert any("boom" in r.getMessage() for r in caplog.records) + + def test_build_fetchers_includes_sources_and_boards() -> None: c = Criteria(ats_boards=[AtsBoard(provider="greenhouse", slug="stripe")]) fetchers = build_fetchers(c) @@ -270,9 +352,9 @@ def test_run_poll_dedupes_across_runs_via_seen_ledger() -> None: assert len(repo.added) == 1 -def test_run_poll_honors_preloaded_seen_keys() -> None: - # A slug already in the seen table (e.g. from a since-deleted lead) is skipped. - repo = FakeRepo(seen=[("slug", "acme-backend-engineer")]) +def test_run_poll_honors_preloaded_seen_url() -> None: + # A URL already in the seen table (e.g. from a since-deleted lead) is skipped. + repo = FakeRepo(seen=[("url", "acme.co/1")]) c = Criteria(keywords=["engineer"]) added = run_poll( repo, c, listings=[_lead("Acme", "Backend Engineer", link="https://acme.co/1")] @@ -281,6 +363,21 @@ def test_run_poll_honors_preloaded_seen_keys() -> None: assert repo.added == [] +def test_run_poll_preloaded_slug_blocks_only_urlless() -> None: + # The slug is the fallback key: a preloaded slug blocks a url-less listing, but + # a listing carrying a distinct URL is judged by URL and still staged. + c = Criteria(keywords=["engineer"]) + + urlless = FakeRepo(seen=[("slug", "acme-backend-engineer")]) + assert run_poll(urlless, c, listings=[_lead("Acme", "Backend Engineer", link="")]) == [] + + with_url = FakeRepo(seen=[("slug", "acme-backend-engineer")]) + added = run_poll( + with_url, c, listings=[_lead("Acme", "Backend Engineer", link="https://acme.co/9")] + ) + assert len(added) == 1 + + def test_run_all_tenants_routes_sources_per_profile() -> None: # Two tenants enable different sources; each must only ever see roles from the # sources it turned on, from one shared per-source gather.