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

fix matchmaking outage and cache regressions

Browse Source
This commit is contained in:
-Puter
2026-07-17 19:24:58 +05:30
parent 97958c8302
commit 5d17cfbf70
4 changed files with 212 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
app/agent/session.py
View File

@@ -188,6 +188,7 @@ class MatchmakingAgentSession:
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.",
@@ -272,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
@@ -286,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)

25
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__)
@@ -32,19 +33,21 @@ def _effective_title(prefs: dict) -> str:
def search_signature(prefs: dict) -> str:
"""Stable short hash of the search-defining prefs (ignores transient flags / user_context)."""
# Canonicalize the fields exactly as the board adapters consume them. The
# dashboard contract is camelCase; work_mode remains a legacy fallback.
work_mode = prefs.get("workMode") if "workMode" in prefs else prefs.get("work_mode")
"""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 = {
"effective_title": _effective_title(prefs),
"role": prefs.get("role"),
"location": prefs.get("location"),
"title": C.title_of(prefs),
"locations": C.parsed_locations(prefs),
"workMode": prefs.get("workMode"),
"industry": prefs.get("industry"),
"experience": prefs.get("experience"),
"years": prefs.get("years"),
"workMode": work_mode,
"remote": prefs.get("remote"),
"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]

16
app/engine/search.py
View File

@@ -116,16 +116,18 @@ 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)
@@ -138,7 +140,10 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict:
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()
@@ -151,9 +156,10 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict:
return {
"opportunities": merged,
"sources": sources,
"all_boards_failed": bool(enabled) and failed_boards == len(enabled),
"all_boards_failed": not enabled or failed_boards == len(enabled),
"failed_boards": failed_boards,
"enabled_boards": len(enabled),
"configured_boards": configured,
}

183
tests/test_pool.py
View File

@@ -1,5 +1,7 @@
"""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
@@ -67,14 +69,51 @@ def test_work_mode_changes_search_signature_using_dashboard_contract():
assert search_signature(base) != search_signature(changed)
def test_legacy_and_camel_case_work_mode_have_same_signature():
def test_legacy_work_mode_does_not_change_adapter_cache_identity():
common = {"title": "Data Analyst", "location": ["Bangalore · India"]}
assert search_signature({**common, "workMode": ["Hybrid"]}) == search_signature({**common, "work_mode": ["Hybrid"]})
assert search_signature(common) == search_signature({**common, "work_mode": ["Hybrid"]})
@pytest.mark.asyncio
async def test_sweep_marks_total_board_failure(monkeypatch):
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")
@@ -84,8 +123,142 @@ async def test_sweep_marks_total_board_failure(monkeypatch):
BOARD_TIMEOUT_S=1,
))
result = await search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]})
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