fix(tradein-matching): pg_advisory_xact_lock to prevent duplicate house INSERTs #501
2 changed files with 52 additions and 9 deletions
|
|
@ -34,11 +34,11 @@ def match_or_create_house(
|
||||||
) -> tuple[int, float, str]:
|
) -> tuple[int, float, str]:
|
||||||
"""Match existing house or create new canonical record.
|
"""Match existing house or create new canonical record.
|
||||||
|
|
||||||
⚠️ KNOWN LIMITATION: Concurrent calls for the same address can race —
|
Concurrency-safe: serializes concurrent calls for the same address fingerprint
|
||||||
two scrapers calling simultaneously for an unknown address can both
|
via pg_advisory_xact_lock(42, hashtext(fp)). The lock is transaction-scoped,
|
||||||
miss Tier 0-3 and each INSERT a new house row. Mitigation (Stage 8.x):
|
released on the caller's COMMIT/ROLLBACK. Without this, two scrapers calling
|
||||||
gate scraper concurrency to 1 per fingerprint OR add
|
for an unknown address could both miss Tier 0-3 and each INSERT a duplicate
|
||||||
pg_advisory_xact_lock(hashtext(normalized_address)) at function entry.
|
house row. Closes finding #1 from 2026-05-24 audit.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
(house_id, confidence ∈ [0.0, 1.0], method ∈ {
|
(house_id, confidence ∈ [0.0, 1.0], method ∈ {
|
||||||
|
|
@ -54,6 +54,18 @@ def match_or_create_house(
|
||||||
"""
|
"""
|
||||||
cad = building_cadastral_number or cadastral_number
|
cad = building_cadastral_number or cadastral_number
|
||||||
|
|
||||||
|
# Compute fingerprint early so we can acquire the advisory lock before any tier reads.
|
||||||
|
fp = address_fingerprint(address, lat, lon)
|
||||||
|
|
||||||
|
# Serialize concurrent calls for the same address fingerprint to prevent
|
||||||
|
# racing INSERTs into houses (finding #1 from 2026-05-24 audit).
|
||||||
|
# pg_advisory_xact_lock takes (int4, int4) — namespace 42 = "tradein house matching".
|
||||||
|
# Lock is released automatically on caller's COMMIT/ROLLBACK.
|
||||||
|
db.execute(
|
||||||
|
text('SELECT pg_advisory_xact_lock(42, hashtext(:fp))'),
|
||||||
|
{'fp': fp},
|
||||||
|
)
|
||||||
|
|
||||||
# Tier 0: cadastral number exact match
|
# Tier 0: cadastral number exact match
|
||||||
if cad:
|
if cad:
|
||||||
row = db.execute(
|
row = db.execute(
|
||||||
|
|
@ -83,9 +95,6 @@ def match_or_create_house(
|
||||||
ext_id)
|
ext_id)
|
||||||
return (house_id, 1.0, 'source_exact')
|
return (house_id, 1.0, 'source_exact')
|
||||||
|
|
||||||
# Compute fingerprint once for Tier 2 and reuse for new house registration
|
|
||||||
fp = address_fingerprint(address, lat, lon)
|
|
||||||
|
|
||||||
# Tier 2: address fingerprint lookup
|
# Tier 2: address fingerprint lookup
|
||||||
row = db.execute(
|
row = db.execute(
|
||||||
text(
|
text(
|
||||||
|
|
|
||||||
|
|
@ -148,6 +148,7 @@ def test_match_house_tier0_cadastr():
|
||||||
from app.services.matching.houses import match_or_create_house
|
from app.services.matching.houses import match_or_create_house
|
||||||
|
|
||||||
db = _make_db([
|
db = _make_db([
|
||||||
|
None, # pg_advisory_xact_lock
|
||||||
{'id': 42}, # cadastral match
|
{'id': 42}, # cadastral match
|
||||||
None, # _upsert_house_source (INSERT ... ON CONFLICT)
|
None, # _upsert_house_source (INSERT ... ON CONFLICT)
|
||||||
])
|
])
|
||||||
|
|
@ -171,7 +172,8 @@ def test_match_house_tier1_source_exact():
|
||||||
from app.services.matching.houses import match_or_create_house
|
from app.services.matching.houses import match_or_create_house
|
||||||
|
|
||||||
db = _make_db([
|
db = _make_db([
|
||||||
{'house_id': 7}, # house_sources hit (first and only execute call)
|
None, # pg_advisory_xact_lock
|
||||||
|
{'house_id': 7}, # house_sources hit
|
||||||
])
|
])
|
||||||
house_id, conf, method = match_or_create_house(
|
house_id, conf, method = match_or_create_house(
|
||||||
db, 'avito', 'ext-999',
|
db, 'avito', 'ext-999',
|
||||||
|
|
@ -186,6 +188,7 @@ def test_match_house_tier2_fingerprint():
|
||||||
from app.services.matching.houses import match_or_create_house
|
from app.services.matching.houses import match_or_create_house
|
||||||
|
|
||||||
db = _make_db([
|
db = _make_db([
|
||||||
|
None, # pg_advisory_xact_lock
|
||||||
None, # house_sources miss
|
None, # house_sources miss
|
||||||
{'house_id': 15}, # fingerprint hit
|
{'house_id': 15}, # fingerprint hit
|
||||||
None, # _upsert_house_source
|
None, # _upsert_house_source
|
||||||
|
|
@ -204,6 +207,7 @@ def test_match_house_tier3_geo():
|
||||||
from app.services.matching.houses import match_or_create_house
|
from app.services.matching.houses import match_or_create_house
|
||||||
|
|
||||||
db = _make_db([
|
db = _make_db([
|
||||||
|
None, # pg_advisory_xact_lock
|
||||||
None, # house_sources miss
|
None, # house_sources miss
|
||||||
None, # fingerprint miss
|
None, # fingerprint miss
|
||||||
{'id': 22, 'dist': 15.5}, # geo hit
|
{'id': 22, 'dist': 15.5}, # geo hit
|
||||||
|
|
@ -224,6 +228,7 @@ def test_match_house_new():
|
||||||
from app.services.matching.houses import match_or_create_house
|
from app.services.matching.houses import match_or_create_house
|
||||||
|
|
||||||
db = _make_db([
|
db = _make_db([
|
||||||
|
None, # pg_advisory_xact_lock
|
||||||
None, # house_sources miss
|
None, # house_sources miss
|
||||||
None, # fingerprint miss
|
None, # fingerprint miss
|
||||||
None, # geo miss
|
None, # geo miss
|
||||||
|
|
@ -240,6 +245,35 @@ def test_match_house_new():
|
||||||
assert method == 'new'
|
assert method == 'new'
|
||||||
|
|
||||||
|
|
||||||
|
def test_match_house_advisory_lock_called_first():
|
||||||
|
"""Regression: pg_advisory_xact_lock must be the first execute() call
|
||||||
|
on match_or_create_house. Re-ordering would re-introduce the race
|
||||||
|
(finding #1 from 2026-05-24 audit). PR #501."""
|
||||||
|
from app.services.matching.houses import match_or_create_house
|
||||||
|
|
||||||
|
db = _make_db([
|
||||||
|
None, # pg_advisory_xact_lock
|
||||||
|
{'house_id': 7}, # house_sources Tier 1 hit (short path)
|
||||||
|
])
|
||||||
|
match_or_create_house(
|
||||||
|
db, 'avito', 'regress-001',
|
||||||
|
address='Тестовая 1', lat=56.8, lon=60.5,
|
||||||
|
)
|
||||||
|
|
||||||
|
# First execute MUST be the lock acquisition, with bind {'fp': <hex>}.
|
||||||
|
first_call = db.execute.call_args_list[0]
|
||||||
|
sql_obj = first_call[0][0] # TextClause
|
||||||
|
bind = first_call[0][1] # dict
|
||||||
|
|
||||||
|
assert 'pg_advisory_xact_lock' in str(sql_obj), (
|
||||||
|
f'first execute must be advisory lock, got: {sql_obj}'
|
||||||
|
)
|
||||||
|
assert 'fp' in bind, f"lock bind must include fp, got: {bind}"
|
||||||
|
assert isinstance(bind['fp'], str) and len(bind['fp']) == 32, (
|
||||||
|
f"fp must be 32-char sha256 hex, got: {bind.get('fp')!r}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# match_or_create_listing — mock DB tier routing
|
# match_or_create_listing — mock DB tier routing
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue