growqr-app/matchmaking-v2
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge matchmaking cache and outage fixes

Browse Source
This commit is contained in:
-Puter
2026-07-17 19:36:17 +05:30
parent 46078bff62 5d17cfbf70
commit 19d7e8e04f
6 changed files with 293 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
app/agent/session.py
View File

@@ -184,6 +184,18 @@ class MatchmakingAgentSession:
search.run_sweep(prefs, fresh=fresh),
["Scanning live roles across the boards…", "Pulling the latest Naukri + LinkedIn + Foundit roles…",
"Gathering fresh postings for you…"])
if sweep.get("all_boards_failed"):
await self.push(
"agent_error",
action="search_failed",
message="All enabled job boards are temporarily unavailable.",
data={
"code": "job_boards_unavailable",
"message": "All enabled job boards are temporarily unavailable.",
"failed_boards": sweep.get("failed_boards", 0),
},
)
return
new_count = await _repo.pool_save(self.user_id, sig, sweep["opportunities"])
# tapped out? a sweep that added almost nothing new → flag it so we stop block-fetching dupes.
if new_count < s.POOL_MIN_NEW_PER_REFILL:
@@ -261,10 +273,12 @@ class MatchmakingAgentSession:
dedupe_key=f"matchmaking:generated:{self.user_id}:{sig}",
)
# The user already has their deck — if the pool dipped below the reorder mark, top it up OUT OF
# BAND so the next search stays free and the shelf never runs thin. (No await: fire-and-forget.)
# The persisted feed intentionally drops resume context, but a refill must use the same
# retrieval inputs as the blocking sweep whose signature owns the pool. Preserve only
# the adapter's top-skill input; never retain the rest of the resume context.
if decision == "pool+refill":
asyncio.create_task(self._background_refill(clean_prefs, sig))
refill_prefs = {**clean_prefs, "_skills": (user_context or {}).get("skills") or prefs.get("_skills") or []}
asyncio.create_task(self._background_refill(refill_prefs, sig))
async def _background_refill(self, prefs: dict, sig: str):
"""Proactive pool top-up: re-fetch the freshest jobs in the background + bank the new ones, so the
@@ -275,6 +289,9 @@ class MatchmakingAgentSession:
try:
from app.config import get_settings
sweep = await search.run_sweep(prefs, fresh=True)
if sweep.get("all_boards_failed"):
logger.warning("pool refill skipped: all configured job boards failed, sig=%s", sig[:8])
return
new_count = await _repo.pool_save(self.user_id, sig, sweep["opportunities"])
if new_count < get_settings().POOL_MIN_NEW_PER_REFILL: # tapped out → relax the floor
await _repo.mark_pool_exhausted(self.user_id, sig)

34
app/db/repo.py
View File

@@ -10,6 +10,7 @@ from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from app.db.models import Base, OpportunityState, UserFeed, UserJobPool
from app.engine.board_adapters import coerce as C
from app.db.session import engine, session_factory
logger = logging.getLogger(__name__)
@@ -17,14 +18,37 @@ logger = logging.getLogger(__name__)
# fields a caller may set on an opportunity's state (status + the generated document ids + activity flags)
_STATE_FIELDS = ("status", "tailored_resume_id", "tailored_version_id", "cover_letter_id", "seen", "viewed", "saved")
# the search-defining pref fields — same values → same signature → cursor advances (new jobs);
# change any → new signature → cursor resets to page 1.
_SIG_FIELDS = ("title", "role", "location", "industry", "experience", "years", "work_mode", "remote")
def _effective_title(prefs: dict) -> str:
"""Match the board adapter's title precedence for cache identity."""
target_titles = prefs.get("targetTitles") or []
if target_titles and str(target_titles[0]).strip():
return str(target_titles[0]).strip()
title = str(prefs.get("title") or "").strip()
if title:
return title
for role in prefs.get("role") or []:
if role and role != "Open to any":
return str(role)
return ""
def search_signature(prefs: dict) -> str:
"""Stable short hash of the search-defining prefs (ignores transient flags / user_context)."""
payload = {k: prefs.get(k) for k in _SIG_FIELDS}
"""Stable short hash of every preference that changes raw board retrieval."""
# The pool stores unranked jobs. Ranking and curation rerun on every request,
# so only adapter inputs belong in this identity. Canonicalize through the
# same helpers used by the board adapters to avoid cache-only semantics.
user_context = prefs.get("user_context") or {}
skills = user_context.get("skills") or prefs.get("_skills") or []
retrieval_skill = str(skills[0]).strip().lower() if skills else ""
payload = {
"title": C.title_of(prefs),
"locations": C.parsed_locations(prefs),
"workMode": prefs.get("workMode"),
"industry": prefs.get("industry"),
"experience": C.seniority_band(prefs),
"years": C.years_of(prefs),
"retrieval_skill": retrieval_skill,
}
blob = json.dumps(payload, sort_keys=True, default=str)
return hashlib.sha1(blob.encode()).hexdigest()[:16]

24
app/engine/search.py
View File

@@ -116,27 +116,34 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict:
# fresh=True forces a live fetch + cache overwrite (dev "fresh search" toggle);
# otherwise honor the configured cache mode (dev replays cached results for $0).
cache_mode = "refresh" if fresh else None
enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip() in BOARDS]
configured = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip()]
enabled = [b for b in configured if b in BOARDS]
cap = get_settings().BOARD_TIMEOUT_S
async def _timed(k: str):
# A single slow/hung board must NOT hold the whole search hostage — cap it and degrade.
try:
return await asyncio.wait_for(_fetch_board(k, prefs, cache_mode), timeout=cap)
key, jobs = await asyncio.wait_for(_fetch_board(k, prefs, cache_mode), timeout=cap)
return key, jobs, False
except asyncio.TimeoutError:
logger.warning("board %s exceeded %ss — skipped (search stays fast)", k, cap)
return k, []
return k, [], True
results = await asyncio.gather(*(_timed(k) for k in enabled), return_exceptions=True)
merged: list[dict] = []
seen: set[str] = set()
sources: dict[str, int] = {}
failed_boards = 0
for r in results:
if isinstance(r, Exception):
failed_boards += 1
logger.warning("board fetch failed: %s", r)
continue
key, jobs = r
key, jobs, failed = r
if failed:
failed_boards += 1
continue
sources[key] = len(jobs)
for j in jobs:
dk = (j.get("apply_url") or "").strip() or f"{j['organization']}|{j['title']}|{j.get('location_city')}".lower()
@@ -146,7 +153,14 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict:
merged.append(j)
# No ranking here — the engine (app/engine/rank.py) scores + selects after the sweep.
return {"opportunities": merged, "sources": sources}
return {
"opportunities": merged,
"sources": sources,
"all_boards_failed": not enabled or failed_boards == len(enabled),
"failed_boards": failed_boards,
"enabled_boards": len(enabled),
"configured_boards": configured,
}
def has_mvq(prefs: dict) -> bool:

2
docker-compose.staging.yml
View File

@@ -35,7 +35,7 @@ services:
condition: service_healthy
redis:
condition: service_started
command: uvicorn app.main:app --host 0.0.0.0 --port 8000
command: sh -c "alembic upgrade head && exec uvicorn app.main:app --host 0.0.0.0 --port 8000"
restart: unless-stopped
postgres:

1
docker-compose.yml
View File

@@ -10,6 +10,7 @@ services:
environment:
# inside the compose network the DB host is `postgres` (overrides the host-dev .env value)
DATABASE_URL: postgresql+asyncpg://matchmaking:matchmaking@postgres:5432/matchmaking
command: sh -c "alembic upgrade head && exec uvicorn app.main:app --host 0.0.0.0 --port 8000"
depends_on:
postgres:
condition: service_healthy

223
tests/test_pool.py
View File

@@ -1,5 +1,13 @@
"""Warm-pool gate — the reorder/floor decision that keeps the shelf stocked without over-fetching."""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from app.engine import search
from app.engine.search import pool_decision
from app.db.repo import search_signature
FLOOR, REORDER = 90, 100 # the configured thresholds (POOL_SAFETY_FLOOR / POOL_REORDER_AT)
@@ -39,3 +47,218 @@ def test_floor_never_above_reorder_invariant():
from app.config import get_settings
s = get_settings()
assert s.POOL_SAFETY_FLOOR <= s.POOL_REORDER_AT
def test_target_title_changes_search_signature():
base = {
"title": "Product Manager",
"role": ["Product"],
"targetTitles": ["Backend Engineer"],
"location": ["Bangalore · India"],
"workMode": ["Hybrid"],
}
changed = {**base, "targetTitles": ["Data Analyst"]}
assert search_signature(base) != search_signature(changed)
def test_work_mode_changes_search_signature_using_dashboard_contract():
base = {"title": "Data Analyst", "location": ["Bangalore · India"], "workMode": ["Hybrid"]}
changed = {**base, "workMode": ["Remote"]}
assert search_signature(base) != search_signature(changed)
def test_legacy_work_mode_does_not_change_adapter_cache_identity():
common = {"title": "Data Analyst", "location": ["Bangalore · India"]}
assert search_signature(common) == search_signature({**common, "work_mode": ["Hybrid"]})
def test_retrieval_inputs_change_search_signature():
base = {
"title": "Data Analyst",
"location": ["Bangalore · India"],
"workMode": ["Hybrid"],
"industry": ["Fintech"],
"experience": ["Mid-level"],
"years": 5,
"user_context": {"skills": ["Python", "SQL"]},
}
changes = [
{"title": "Product Manager"},
{"location": ["Mumbai · India"]},
{"workMode": ["Remote"]},
{"industry": ["Healthcare"]},
{"experience": ["Senior"]},
{"years": 8},
{"user_context": {"skills": ["Rust", "SQL"]}},
]
for change in changes:
assert search_signature(base) != search_signature({**base, **change})
def test_rank_only_preferences_do_not_cold_start_raw_pool():
base = {"title": "Data Analyst", "location": ["Bangalore · India"]}
changed = {**base, "targetComp": "₹22–30L", "dealBreakers": ["Travel"], "stretch": "reach"}
assert search_signature(base) == search_signature(changed)
def test_search_signature_ignores_unrelated_user_context():
base = {"title": "Data Analyst", "location": ["Bangalore · India"], "user_context": {"skills": ["Python"]}}
changed = {**base, "user_context": {"skills": ["Python"], "education": ["MBA"]}}
assert search_signature(base) == search_signature(changed)
def test_sweep_marks_total_board_exception_failure(monkeypatch):
async def fail_board(*_args, **_kwargs):
raise RuntimeError("provider blocked")
monkeypatch.setattr(search, "_fetch_board", fail_board)
monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
BOARDS_ENABLED="naukri,foundit",
BOARD_TIMEOUT_S=1,
))
result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
assert result["all_boards_failed"] is True
assert result["failed_boards"] == 2
assert result["sources"] == {}
def test_sweep_marks_total_board_timeout_failure(monkeypatch):
async def timeout_board(*_args, **_kwargs):
await asyncio.sleep(0.05)
return "naukri", []
monkeypatch.setattr(search, "_fetch_board", timeout_board)
monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
BOARDS_ENABLED="naukri,foundit",
BOARD_TIMEOUT_S=0.001,
))
result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
assert result["all_boards_failed"] is True
assert result["failed_boards"] == 2
def test_sweep_marks_empty_or_unknown_board_config_as_failure(monkeypatch):
monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
BOARDS_ENABLED="unknown-board",
BOARD_TIMEOUT_S=1,
))
result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
assert result["all_boards_failed"] is True
assert result["enabled_boards"] == 0
assert result["configured_boards"] == ["unknown-board"]
def test_total_outage_emits_frontend_readable_agent_error(monkeypatch):
from app.agent.session import MatchmakingAgentSession
ws = MagicMock()
ws.send_json = AsyncMock()
session = MatchmakingAgentSession(ws, "user-1")
async def outage_heartbeat(awaitable, _messages):
awaitable.close()
return {
"opportunities": [],
"sources": {},
"all_boards_failed": True,
"failed_boards": 2,
}
session._with_heartbeat = outage_heartbeat
monkeypatch.setattr("app.agent.session.get_settings", lambda: SimpleNamespace(
POOL_ENABLED=False,
POOL_SAFETY_FLOOR=90,
POOL_REORDER_AT=100,
), raising=False)
monkeypatch.setattr("app.config.get_settings", lambda: SimpleNamespace(
POOL_ENABLED=False,
POOL_SAFETY_FLOOR=90,
POOL_REORDER_AT=100,
))
monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=set()))
asyncio.run(session.handle_run_search({"title": "Data Analyst", "location": ["Bangalore · India"]}))
message = ws.send_json.await_args.args[0]
assert message["type"] == "agent_error"
assert message["message"] == "All enabled job boards are temporarily unavailable."
assert message["data"]["code"] == "job_boards_unavailable"
def test_background_refill_does_not_mark_outage_as_exhausted(monkeypatch):
from app.agent.session import MatchmakingAgentSession
session = MatchmakingAgentSession(MagicMock(), "user-1")
monkeypatch.setattr("app.engine.search.run_sweep", AsyncMock(return_value={
"opportunities": [],
"sources": {},
"all_boards_failed": True,
}))
save = AsyncMock()
exhausted = AsyncMock()
monkeypatch.setattr("app.db.repo.pool_save", save)
monkeypatch.setattr("app.db.repo.mark_pool_exhausted", exhausted)
asyncio.run(session._background_refill({"title": "Data Analyst"}, "signature"))
save.assert_not_awaited()
exhausted.assert_not_awaited()
def test_background_refill_preserves_signature_retrieval_skill(monkeypatch):
from app.agent.session import MatchmakingAgentSession
ws = MagicMock()
ws.send_json = AsyncMock()
session = MatchmakingAgentSession(ws, "user-1")
prefs = {
"title": "Data Analyst",
"location": ["Bangalore · India"],
"user_context": {"skills": ["Python", "SQL"], "education": ["MBA"]},
}
sig = search_signature(prefs)
captured = {}
monkeypatch.setattr("app.config.get_settings", lambda: SimpleNamespace(
POOL_ENABLED=True,
POOL_SAFETY_FLOOR=90,
POOL_REORDER_AT=100,
))
monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=set()))
monkeypatch.setattr("app.db.repo.pool_load_fresh_unseen", AsyncMock(return_value=[{"id": f"job-{i}"} for i in range(95)]))
monkeypatch.setattr("app.db.repo.is_pool_exhausted", AsyncMock(return_value=False))
monkeypatch.setattr("app.db.repo.save_feed", AsyncMock())
monkeypatch.setattr("app.db.repo.update_salary_band", AsyncMock())
monkeypatch.setattr("app.db.repo.mark_seen", AsyncMock())
monkeypatch.setattr("app.db.repo.get_opportunity_states", AsyncMock(return_value={}))
monkeypatch.setattr("app.agent.session.emit_grow_event", AsyncMock())
monkeypatch.setattr("app.engine.sift.sift", lambda *_args: ([{"id": "job-1"}], {"mode": "test", "scored": 1}))
monkeypatch.setattr("app.engine.curate.curate", lambda *_args: [{"id": "job-1"}])
async def capture_refill(refill_prefs, refill_sig):
captured["prefs"] = refill_prefs
captured["sig"] = refill_sig
async def direct_heartbeat(awaitable, _messages):
return await awaitable
session._background_refill = capture_refill
session._with_heartbeat = direct_heartbeat
session._grow_activity_fields = AsyncMock(return_value={})
asyncio.run(session.handle_run_search(prefs))
asyncio.run(asyncio.sleep(0))
assert captured["sig"] == sig
assert captured["prefs"]["_skills"] == ["Python", "SQL"]
assert "user_context" not in captured["prefs"]
assert search_signature(captured["prefs"]) == sig