fix(realestate): listings-ingest async(BackgroundTask)+매칭/알림 타이밍 로그 + notify 홍수방지(cap+baseline)

realestate-lab/app/internal_router.py

@@ -1,8 +1,9 @@
 """naver-fetch 워커 내부 계약: targets 조회 + listings ingest."""
 import os
+import time
 import logging
 from typing import List, Dict, Any
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, BackgroundTasks
 from pydantic import BaseModel
 
 from .auth import verify_internal_key
@@ -45,7 +46,7 @@ class ListingsIngest(BaseModel):
 
 @router.post("/api/internal/realestate/listings-ingest",
              dependencies=[Depends(verify_internal_key)])
-def listings_ingest(body: ListingsIngest):
+def listings_ingest(body: ListingsIngest, background_tasks: BackgroundTasks):
     received = new = 0
     for batch in body.batches:
         for raw in batch.articles:
@@ -60,10 +61,18 @@ def listings_ingest(body: ListingsIngest):
                     new += 1
             except Exception as e:
                 logger.warning("네이버 ingest 파싱 실패: %s", e)
-    matched = 0
     if received:
-        with match_notify_lock:
+        background_tasks.add_task(_match_and_notify)
-            run_listing_matching()
+    return {"received": received, "new": new, "queued": bool(received)}
-            noti = notify_new_listings()
+
-            matched = int(noti.get("sent", 0)) if isinstance(noti, dict) else 0
+
-    return {"received": received, "new": new, "matched": matched}
+def _match_and_notify():
+    """ingest 후처리(백그라운드): 매칭+알림을 공유 락으로 직렬화. 타이밍 로그로 병목 관측."""
+    with match_notify_lock:
+        t0 = time.time()
+        run_listing_matching()
+        t1 = time.time()
+        noti = notify_new_listings()
+        t2 = time.time()
+    sent = (noti or {}).get("sent") if isinstance(noti, dict) else noti
+    logger.info("ingest 후처리: match=%.1fs notify=%.1fs sent=%s", t1 - t0, t2 - t1, sent)

realestate-lab/app/notifier.py

@@ -10,6 +10,8 @@ logger = logging.getLogger("realestate-lab")
 
 AGENT_OFFICE_URL = os.getenv("AGENT_OFFICE_URL", "http://agent-office:8000")
 NOTIFY_TIMEOUT_SECONDS = int(os.getenv("REALESTATE_NOTIFY_TIMEOUT", "15"))
+LISTING_NOTIFY_MAX = int(os.getenv("LISTING_NOTIFY_MAX", "8"))
+LISTING_NOTIFY_TIMEOUT = int(os.getenv("LISTING_NOTIFY_TIMEOUT", "60"))
 
 
 def notify_new_matches() -> dict:
@@ -49,21 +51,28 @@ def notify_new_matches() -> dict:
 
 def notify_new_listings() -> dict:
     """임계 통과 미알림 매물을 agent-office로 push (notify_new_matches 대칭).
-    passed 매물은 전 안전등급 알림(위험도 회피 목적 유용). criteria.min_safety_tier 설정 시 그 이상만."""
+    passed 매물은 전 안전등급 알림(위험도 회피 목적 유용). criteria.min_safety_tier 설정 시 그 이상만.
+    콜드스타트로 backlog가 쌓이면 한 사이클 최대 LISTING_NOTIFY_MAX건만 알림 발송하고
+    나머지는 조용히 notified_at 마킹(baseline)해 스팸·재전송 루프를 막는다."""
     crit = get_listing_criteria()
     if not crit.get("notify_enabled"):
         return {"sent": 0, "skipped": "notify_disabled"}
     listings = get_unnotified_listing_matches(min_tier=crit.get("min_safety_tier"))
     if not listings:
         return {"sent": 0}
+    to_send = listings[:LISTING_NOTIFY_MAX]
+    to_baseline = listings[LISTING_NOTIFY_MAX:]
+    if to_baseline:  # 콜드스타트 backlog: 조용히 baseline 마킹(알림 X)
+        mark_listings_notified([m["id"] for m in to_baseline])
+        logger.info("매물 알림 baseline(무알림) 마킹: %d건", len(to_baseline))
     url = f"{AGENT_OFFICE_URL}/api/agent-office/realestate/notify-listing"
     try:
-        resp = requests.post(url, json={"listings": listings}, timeout=NOTIFY_TIMEOUT_SECONDS)
+        resp = requests.post(url, json={"listings": to_send}, timeout=LISTING_NOTIFY_TIMEOUT)
         resp.raise_for_status()
         body = resp.json()
     except requests.RequestException as e:
         logger.error("agent-office 매물 push 실패: %s", e)
-        return {"sent": 0, "error": str(e)}
+        return {"sent": 0, "error": str(e), "baselined": len(to_baseline)}
     sent_ids = body.get("sent_ids") or []
     if sent_ids:
         mark_listings_notified(sent_ids)

realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md

@@ -110,7 +110,7 @@
 ## 6. 인증 / 보안
 
 - **nginx**: 신규 location `/api/internal/realestate/` → `realestate-lab:8000`, **IP 화이트리스트**(기존 render 워커 화이트리스트에 쓰는 Windows 워커 IP 재사용). (nginx-conf 변경 → 배포)
-- **앱 레벨**: realestate-lab이 `X-Internal-Key` 검증(env `REALESTATE_INTERNAL_KEY`). 다른 서비스 `internal_router` 패턴 준용. GET targets·POST ingest 둘 다 검증.
+- **앱 레벨**: realestate-lab이 `X-Internal-Key` 검증(env `INTERNAL_API_KEY`). 다른 서비스 `internal_router` 패턴 준용. GET targets·POST ingest 둘 다 검증.
 - 워커는 `.env`로 키 주입(커밋 금지).
 
 ## 7. 에러처리 / 멱등성

realestate-lab/tests/test_internal_ingest.py

@@ -23,6 +23,8 @@ def test_ingest_auth(monkeypatch):
 
 def test_ingest_upserts_and_matches(monkeypatch):
     c = _client(monkeypatch)
+    # TestClient는 BackgroundTask를 응답 반환 전 동기 실행하므로 patch된 notify는
+    # c.post()가 리턴하는 시점에 이미 호출되어 있다.
     with patch("app.internal_router.notify_new_listings", return_value={"sent": 1, "sent_ids": [1]}) as noti:
         r = c.post("/api/internal/realestate/listings-ingest",
                    headers={"X-Internal-Key": "SECRET"},
@@ -30,7 +32,7 @@ def test_ingest_upserts_and_matches(monkeypatch):
                          "batches": [{"dong": "신대방동", "articles": [_ARTICLE]}]})
     assert r.status_code == 200
     body = r.json()
-    assert body["received"] == 1 and body["new"] == 1 and body["matched"] == 1
+    assert body["received"] == 1 and body["new"] == 1 and body["queued"] is True
     assert noti.call_count == 1
     # 저장 확인
     from app import db

realestate-lab/tests/test_listing_notifier.py

@@ -26,3 +26,26 @@ def test_notify_listings_no_mark_on_failure():
     with patch.object(notifier.requests, "post", side_effect=rq.RequestException("down")):
         notifier.notify_new_listings()
     assert len(db.get_unnotified_listing_matches()) == 1   # 미마킹→재시도
+
+
+def test_notify_caps_and_baselines(monkeypatch):
+    from app import notifier, db
+    from unittest.mock import MagicMock
+    for i in range(10):
+        row, _ = db.upsert_listing({"article_no": f"C{i}", "source": "naver", "deal_type": "전세",
+                                    "deposit": 29000, "area_exclusive": 42.0, "dong": "신대방동"})
+        db.upsert_listing_match({"listing_id": row["id"], "category": "임차", "passed": 1,
+                                 "match_score": 100, "safety_tier": "안전", "sample_size": 7,
+                                 "reasons": ["ok"], "is_new": 1})
+    monkeypatch.setattr(notifier, "LISTING_NOTIFY_MAX", 8)
+    captured = {}
+    fake = MagicMock(); fake.raise_for_status.return_value = None
+    def fake_post(url, json=None, timeout=None):
+        captured["n"] = len(json["listings"])
+        fake.json.return_value = {"sent": len(json["listings"]),
+                                  "sent_ids": [m["id"] for m in json["listings"]]}
+        return fake
+    monkeypatch.setattr(notifier.requests, "post", fake_post)
+    notifier.notify_new_listings()
+    assert captured["n"] == 8                                   # 8건만 텔레그램 전송
+    assert len(db.get_unnotified_listing_matches()) == 0        # 나머지 2건 baseline 마킹