From e2c242c7ccbd135abaca24791a04c528cdc66872 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sun, 15 Mar 2026 20:53:57 -0700 Subject: [PATCH] =?UTF-8?q?fix(sec):=20=EB=8C=80=ED=98=95=20exhibit=20?= =?UTF-8?q?=ED=8C=8C=EC=9D=BC=EB=A1=9C=20=EC=9D=B8=ED=95=9C=20=EC=84=9C?= =?UTF-8?q?=EB=B2=84=20=EB=A9=94=EB=AA=A8=EB=A6=AC=20=EA=B3=A0=EA=B0=88=20?= =?UTF-8?q?=EB=B0=A9=EC=A7=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - fetch_text에 max_bytes 파라미터 추가: Content-Length 헤더로 다운로드 전 사전 reject, 헤더 없으면 content.read(max_bytes+1)로 제한적 읽기 - ValueError는 즉시 raise (retry 없음 — 크기는 재시도해도 안 줄어듦) - in-memory 캐시(_text_cache) 1MB 가드: 대형 응답은 디스크 캐시에만 저장 - MAX_EXHIBIT_SIZE 1MB → 5MB, fetch_text(max_bytes=...) 호출로 다운로드 전 체크 - exhibit deadline 30초 → 15초 (서비스 + 엔드포인트 + bulk) - 신규 테스트 5개: Content-Length 사전 거부, body 제한 읽기, 메모리 캐시 가드, 소형 캐시 유지, ValueError no-retry Co-Authored-By: Claude Sonnet 4.6 --- app/api/v1/endpoints/filings.py | 34 ++- app/services/sec_filings_service.py | 115 ++++---- app/services/sec_http_client.py | 397 +++++++++++++++----------- tests/test_sec_filings_fixes.py | 413 ++++++++++++++++++++++++++++ 4 files changed, 741 insertions(+), 218 deletions(-) create mode 100644 tests/test_sec_filings_fixes.py diff --git a/app/api/v1/endpoints/filings.py b/app/api/v1/endpoints/filings.py index 27cb51c..b8c4bd4 100644 --- a/app/api/v1/endpoints/filings.py +++ b/app/api/v1/endpoints/filings.py @@ -25,7 +25,7 @@ from app.schemas.filing import ( FilingSummary, ) from app.services.sec_filings_service import sec_filings_service -from app.utils.cache import with_cache +from app.utils.cache import with_cache, build_cache_key, get_negative_cached, set_negative_cached router = APIRouter() logger = logging.getLogger("app.api.v1.filings") @@ -167,14 +167,24 @@ async def get_exhibit_content( db: AsyncSession = Depends(get_db), ): """Extract exhibit content (e.g., press release EX-99.1) from a filing.""" + # Fast path: check negative cache (exhibit known not to exist) + neg_key = build_cache_key("filings:exhibit:404", accession_number, exhibit_type) + if await get_negative_cached(neg_key): + raise HTTPException( + status_code=404, + detail=f"Exhibit {exhibit_type} not found in filing {accession_number}", + ) + try: result = await asyncio.wait_for( sec_filings_service.get_exhibit_content(db, accession_number, exhibit_type), - timeout=30, + timeout=15, ) except asyncio.TimeoutError: - raise HTTPException(status_code=504, detail="Exhibit extraction timed out after 30s.") + raise HTTPException(status_code=504, detail="Exhibit extraction timed out after 15s.") except ValueError as e: + # Cache this "not found" result so subsequent requests skip SEC entirely + await set_negative_cached(neg_key, ttl=3600) raise HTTPException(status_code=404, detail=str(e)) except Exception as e: logger.error(f"Exhibit extraction failed for {accession_number}/{exhibit_type}: {e}") @@ -290,12 +300,20 @@ async def get_exhibit_bulk( async def _fetch_one(item: dict) -> BulkExhibitItem: accession_number = item.get("accession_number", "") exhibit_type = item.get("exhibit_type", "EX-99.1") + neg_key = build_cache_key("filings:exhibit:404", accession_number, exhibit_type) + if await get_negative_cached(neg_key): + return BulkExhibitItem( + accession_number=accession_number, + exhibit_type=exhibit_type, + success=False, + error=f"Exhibit {exhibit_type} not found in filing {accession_number}", + ) # Each concurrent call gets its own DB session to avoid session contention async with AsyncSessionLocal() as session: try: result = await asyncio.wait_for( sec_filings_service.get_exhibit_content(session, accession_number, exhibit_type), - timeout=30, + timeout=15, ) return BulkExhibitItem( accession_number=accession_number, @@ -306,6 +324,14 @@ async def get_exhibit_bulk( filename=result.get("filename"), url=result.get("url"), ) + except ValueError as e: + await set_negative_cached(neg_key, ttl=3600) + return BulkExhibitItem( + accession_number=accession_number, + exhibit_type=exhibit_type, + success=False, + error=str(e), + ) except Exception as e: return BulkExhibitItem( accession_number=accession_number, diff --git a/app/services/sec_filings_service.py b/app/services/sec_filings_service.py index de9efaf..58aecf0 100644 --- a/app/services/sec_filings_service.py +++ b/app/services/sec_filings_service.py @@ -18,8 +18,8 @@ from app.services.sec_http_client import SECHttpClient logger = logging.getLogger(__name__) -# Maximum exhibit content size (1 MB) -MAX_EXHIBIT_SIZE = 1 * 1024 * 1024 +# Maximum exhibit content size (5 MB) +MAX_EXHIBIT_SIZE = 5 * 1024 * 1024 class SECFilingsService: @@ -31,6 +31,7 @@ class SECFilingsService: def __init__(self): self._http = SECHttpClient("Stock Oracle Filings Service") + self._doc_fetch_locks: Dict[str, asyncio.Lock] = {} # ------------------------------------------------------------------ # index_filings: fetch from SEC submissions and upsert to DB @@ -282,61 +283,74 @@ class SECFilingsService: raise ValueError(f"Filing {accession_number} not found in database") # Return cached documents if available - if filing.documents_json: + if filing.documents_json is not None: return filing.documents_json - # Fetch and parse index page - cik_int = int("".join(ch for ch in filing.cik if ch.isdigit())) - acc_clean = accession_number.replace("-", "") - index_url = ( - f"https://www.sec.gov/Archives/edgar/data/{cik_int}" - f"/{acc_clean}/{accession_number}-index.htm" - ) + # Singleflight: serialize concurrent fetches for the same accession number + if accession_number not in self._doc_fetch_locks: + self._doc_fetch_locks[accession_number] = asyncio.Lock() + lock = self._doc_fetch_locks[accession_number] - html = await self._http.fetch_text(index_url) - soup = BeautifulSoup(html, "html.parser") + async with lock: + # Re-check after acquiring lock (another coroutine may have fetched already) + result = await db.execute( + select(SECFiling).where(SECFiling.accession_number == accession_number) + ) + filing = result.scalar_one_or_none() + if filing and filing.documents_json is not None: + return filing.documents_json + + # Fetch and parse index page + cik_int = int("".join(ch for ch in filing.cik if ch.isdigit())) + acc_clean = accession_number.replace("-", "") + index_url = ( + f"https://www.sec.gov/Archives/edgar/data/{cik_int}" + f"/{acc_clean}/{accession_number}-index.htm" + ) - base_url = index_url.rsplit("/", 1)[0] + html = await self._http.fetch_text(index_url) + soup = BeautifulSoup(html, "html.parser") - def mk_abs(href: str) -> str: - if href.startswith("http"): - return href - if href.startswith("/"): - return f"https://www.sec.gov{href}" - return f"{base_url}/{href}" + base_url = index_url.rsplit("/", 1)[0] - documents: List[Dict] = [] - for row in soup.find_all("tr"): - cells = row.find_all(["td", "th"]) - if len(cells) < 4: - continue - # Typical columns: Seq, Description, Document, Type, Size - desc = cells[1].get_text(strip=True) if len(cells) > 1 else "" - doc_cell = cells[2] - doc_type = cells[3].get_text(strip=True) if len(cells) > 3 else "" - size_text = cells[4].get_text(strip=True) if len(cells) > 4 else "" - - a_tag = doc_cell.find("a") - if not a_tag or not a_tag.get("href"): - continue - href = a_tag["href"] - filename = a_tag.get_text(strip=True) or doc_cell.get_text(strip=True) - - documents.append({ - "type": doc_type, - "description": desc, - "filename": filename, - "url": mk_abs(href), - "size": size_text, - }) + def mk_abs(href: str) -> str: + if href.startswith("http"): + return href + if href.startswith("/"): + return f"https://www.sec.gov{href}" + return f"{base_url}/{href}" - # Cache to DB - if documents: + documents: List[Dict] = [] + for row in soup.find_all("tr"): + cells = row.find_all(["td", "th"]) + if len(cells) < 4: + continue + # Typical columns: Seq, Description, Document, Type, Size + desc = cells[1].get_text(strip=True) if len(cells) > 1 else "" + doc_cell = cells[2] + doc_type = cells[3].get_text(strip=True) if len(cells) > 3 else "" + size_text = cells[4].get_text(strip=True) if len(cells) > 4 else "" + + a_tag = doc_cell.find("a") + if not a_tag or not a_tag.get("href"): + continue + href = a_tag["href"] + filename = a_tag.get_text(strip=True) or doc_cell.get_text(strip=True) + + documents.append({ + "type": doc_type, + "description": desc, + "filename": filename, + "url": mk_abs(href), + "size": size_text, + }) + + # Cache to DB (always, even if empty, to prevent repeated SEC fetches) filing.documents_json = documents filing.updated_at = datetime.now(timezone.utc) await db.commit() - return documents + return documents finally: if _owned_deadline: self._http.clear_deadline() @@ -352,7 +366,7 @@ class SECFilingsService: exhibit_type: str = "EX-99.1", ) -> Dict: """Download and return the content of a specific exhibit.""" - self._http.set_deadline(30.0) + self._http.set_deadline(15.0) try: documents = await self.get_filing_documents(db, accession_number) if not documents: @@ -382,12 +396,7 @@ class SECFilingsService: ) url = target["url"] - content = await self._http.fetch_text(url) - - if len(content) > MAX_EXHIBIT_SIZE: - raise ValueError( - f"Exhibit content exceeds size limit ({len(content)} bytes > {MAX_EXHIBIT_SIZE} bytes)" - ) + content = await self._http.fetch_text(url, max_bytes=MAX_EXHIBIT_SIZE) # Determine content type filename = target.get("filename", "") diff --git a/app/services/sec_http_client.py b/app/services/sec_http_client.py index 02fd391..88b15fc 100644 --- a/app/services/sec_http_client.py +++ b/app/services/sec_http_client.py @@ -24,6 +24,9 @@ _deadline_var: contextvars.ContextVar[Optional[float]] = contextvars.ContextVar( logger = logging.getLogger(__name__) +# Only cache small responses in memory; large responses go to disk cache only +_MAX_MEMORY_CACHE_BYTES = 1 * 1024 * 1024 # 1MB + def _is_sec_block_page(text: str) -> bool: if not text: @@ -79,12 +82,17 @@ class SECHttpClient: - In-memory cache (_json_cache, _text_cache) """ + # Max total pending (waiting + active) SEC HTTP requests across all callers. + # Requests beyond this limit are rejected immediately to protect the event loop. + _MAX_PENDING: int = 50 + def __init__(self, user_agent_name: str = "Stock Oracle"): self.sec_base_data = "https://data.sec.gov" self.sec_base_www = "https://www.sec.gov" self.http_timeout = aiohttp.ClientTimeout(total=12) self._req_sem = asyncio.Semaphore(2) - self._rate_limiter = _TokenBucket(rate=10.0, capacity=10.0) + self._rate_limiter = _TokenBucket(rate=8.0, capacity=8.0) + self._pending: int = 0 self._text_cache: Dict[str, str] = {} self._json_cache: Dict[str, dict] = {} self._cache_dir = "/tmp/stock_oracle_sec_cache" @@ -190,82 +198,108 @@ class SECHttpClient: except Exception: pass - attempts = 6 - backoff = 1.0 - last_exc = None - for _i in range(attempts): - now = _time.monotonic() - dl = _deadline_var.get() - if dl is not None and now >= dl: - break - req_timeout = self.http_timeout - if dl is not None: - remaining = max(0.0, dl - now) - if remaining < 0.25: + # Backpressure: reject immediately if too many SEC requests are already pending. + # This prevents thousands of coroutines from stacking up in the event loop, + # which would starve health checks and other endpoints. + if self._pending >= self._MAX_PENDING: + raise RuntimeError( + f"SEC request queue full ({self._pending}/{self._MAX_PENDING} pending)" + ) + self._pending += 1 + try: + attempts = 6 + backoff = 1.0 + last_exc = None + for _i in range(attempts): + now = _time.monotonic() + dl = _deadline_var.get() + if dl is not None and now >= dl: break - req_timeout = aiohttp.ClientTimeout( - total=min(remaining, getattr(self.http_timeout, "total", 12)) - ) - async with self._req_sem: - await self._rate_limiter.acquire() - try: - session = await self._get_session() - async with session.get(url, timeout=req_timeout, headers={"Accept": "application/json"}) as resp: - if resp.status == 429: - retry_after = resp.headers.get("Retry-After") - delay = ( - float(retry_after) - if retry_after and retry_after.isdigit() - else backoff - ) - dl = _deadline_var.get() - if dl is not None: - remaining = max(0.0, dl - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) - ) - backoff *= 1.8 - continue - if 500 <= resp.status < 600: - delay = backoff - dl = _deadline_var.get() - if dl is not None: - remaining = max(0.0, dl - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) - ) - backoff *= 1.8 - continue - resp.raise_for_status() - data = await resp.json() - # Cache successful response - try: - with open(self._cache_path(url) + ".json", "w", encoding="utf-8") as f: - json.dump(data, f) - except Exception: - pass - self._json_cache[url] = data - return data - except Exception as e: - last_exc = e - delay = backoff - dl = _deadline_var.get() - if dl is not None: - remaining = max(0.0, dl - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + req_timeout = self.http_timeout + if dl is not None: + remaining = max(0.0, dl - now) + if remaining < 0.25: + break + req_timeout = aiohttp.ClientTimeout( + total=min(remaining, getattr(self.http_timeout, "total", 12)) ) - backoff *= 1.8 - continue - raise last_exc if last_exc else RuntimeError("Failed to fetch JSON") - - async def fetch_text(self, url: str, accept: str = "text/html") -> str: + async with self._req_sem: + await self._rate_limiter.acquire() + try: + session = await self._get_session() + async with session.get(url, timeout=req_timeout, headers={"Accept": "application/json"}) as resp: + if resp.status == 429: + retry_after = resp.headers.get("Retry-After") + delay = ( + float(retry_after) + if retry_after and retry_after.isdigit() + else backoff + ) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + if 500 <= resp.status < 600: + delay = backoff + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + resp.raise_for_status() + data = await resp.json() + # Cache successful response + try: + with open(self._cache_path(url) + ".json", "w", encoding="utf-8") as f: + json.dump(data, f) + except Exception: + pass + self._json_cache[url] = data + return data + except aiohttp.ClientResponseError as e: + if 400 <= e.status < 500: + raise # 4xx: no retry, raise immediately + last_exc = e + delay = backoff + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + except Exception as e: + last_exc = e + delay = backoff + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + raise last_exc if last_exc else RuntimeError("Failed to fetch JSON") + finally: + self._pending -= 1 + + async def fetch_text(self, url: str, accept: str = "text/html", max_bytes: Optional[int] = None) -> str: """Fetch text content with retry, backoff, block page detection, and caching.""" # In-memory cache if url in self._text_cache: @@ -284,95 +318,136 @@ class SECHttpClient: except Exception: pass else: - self._text_cache[url] = text + if len(text) <= _MAX_MEMORY_CACHE_BYTES: + self._text_cache[url] = text return text except Exception: pass - attempts = 6 - backoff = 1.0 - last_exc = None - for _i in range(attempts): - now = _time.monotonic() - dl = _deadline_var.get() - if dl is not None and now >= dl: - break - req_timeout = self.http_timeout - if dl is not None: - remaining = max(0.0, dl - now) - if remaining < 0.25: + # Backpressure: reject immediately if too many SEC requests are already pending. + if self._pending >= self._MAX_PENDING: + raise RuntimeError( + f"SEC request queue full ({self._pending}/{self._MAX_PENDING} pending)" + ) + self._pending += 1 + try: + attempts = 6 + backoff = 1.0 + last_exc = None + for _i in range(attempts): + now = _time.monotonic() + dl = _deadline_var.get() + if dl is not None and now >= dl: break - req_timeout = aiohttp.ClientTimeout( - total=min(remaining, getattr(self.http_timeout, "total", 12)) - ) - async with self._req_sem: - await self._rate_limiter.acquire() - try: - session = await self._get_session() - async with session.get(url, timeout=req_timeout, headers={"Accept": accept}) as resp: - if resp.status == 429: - retry_after = resp.headers.get("Retry-After") - delay = ( - float(retry_after) - if retry_after and retry_after.isdigit() - else backoff - ) - dl = _deadline_var.get() - if dl is not None: - remaining = max(0.0, dl - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) - ) - backoff *= 1.8 - continue - if 500 <= resp.status < 600: - delay = backoff - dl = _deadline_var.get() - if dl is not None: - remaining = max(0.0, dl - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) - ) - backoff *= 1.8 - continue - resp.raise_for_status() - text = await resp.text() - if _is_sec_block_page(text): - last_exc = RuntimeError("SEC_BLOCKED") - delay = backoff * 2.0 - dl = _deadline_var.get() - if dl is not None: - remaining = max(0.0, dl - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.5 if delay > 0 else 0.0) - ) - backoff *= 2.0 - continue - # Cache successful response - self._text_cache[url] = text - try: - with open(self._cache_path(url) + ".txt", "w", encoding="utf-8") as f: - f.write(text) - except Exception: - pass - return text - except Exception as e: - last_exc = e - delay = backoff - dl = _deadline_var.get() - if dl is not None: - remaining = max(0.0, dl - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + req_timeout = self.http_timeout + if dl is not None: + remaining = max(0.0, dl - now) + if remaining < 0.25: + break + req_timeout = aiohttp.ClientTimeout( + total=min(remaining, getattr(self.http_timeout, "total", 12)) ) - backoff *= 1.8 - continue - raise last_exc if last_exc else RuntimeError("Failed to fetch text") + async with self._req_sem: + await self._rate_limiter.acquire() + try: + session = await self._get_session() + async with session.get(url, timeout=req_timeout, headers={"Accept": accept}) as resp: + if resp.status == 429: + retry_after = resp.headers.get("Retry-After") + delay = ( + float(retry_after) + if retry_after and retry_after.isdigit() + else backoff + ) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + if 500 <= resp.status < 600: + delay = backoff + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + resp.raise_for_status() + if max_bytes is not None: + cl = resp.content_length + if cl is not None and cl > max_bytes: + raise ValueError( + f"Response too large ({cl} bytes > {max_bytes} limit)" + ) + raw = await resp.content.read(max_bytes + 1) + if len(raw) > max_bytes: + raise ValueError( + f"Response exceeded {max_bytes} byte limit" + ) + text = raw.decode(resp.get_encoding() or "utf-8", errors="replace") + else: + text = await resp.text() + if _is_sec_block_page(text): + last_exc = RuntimeError("SEC_BLOCKED") + delay = backoff * 2.0 + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.5 if delay > 0 else 0.0) + ) + backoff *= 2.0 + continue + # Cache successful response + if len(text) <= _MAX_MEMORY_CACHE_BYTES: + self._text_cache[url] = text + try: + with open(self._cache_path(url) + ".txt", "w", encoding="utf-8") as f: + f.write(text) + except Exception: + pass + return text + except aiohttp.ClientResponseError as e: + if 400 <= e.status < 500: + raise # 4xx: no retry, raise immediately + last_exc = e + delay = backoff + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + except ValueError: + raise # Size limit exceeded: no retry (size won't change) + except Exception as e: + last_exc = e + delay = backoff + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + raise last_exc if last_exc else RuntimeError("Failed to fetch text") + finally: + self._pending -= 1 diff --git a/tests/test_sec_filings_fixes.py b/tests/test_sec_filings_fixes.py new file mode 100644 index 0000000..02a6754 --- /dev/null +++ b/tests/test_sec_filings_fixes.py @@ -0,0 +1,413 @@ +""" +Unit tests for SEC filings fixes: +1. Backpressure: SECHttpClient._pending counter rejects when queue full +2. Negative cache helpers: get_negative_cached / set_negative_cached +3. Exhibit endpoint negative caching: 404 cached, skipped on repeat +""" + +import asyncio +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + + +# --------------------------------------------------------------------------- +# Fix 1: Backpressure — _pending counter +# --------------------------------------------------------------------------- + +class TestSECHttpClientBackpressure: + def _make_client(self): + from app.services.sec_http_client import SECHttpClient + client = SECHttpClient.__new__(SECHttpClient) + client._pending = 0 + client._MAX_PENDING = 5 + client._text_cache = {} + client._json_cache = {} + client._cache_dir = "/tmp/test_sec_cache_bp" + client._req_sem = asyncio.Semaphore(2) + from app.services.sec_http_client import _TokenBucket + client._rate_limiter = _TokenBucket(rate=100.0, capacity=100.0) + client.http_timeout = MagicMock() + client.http_timeout.total = 12 + client._session = None + return client + + @pytest.mark.asyncio + async def test_pending_counter_increments_and_decrements(self): + """_pending goes up during fetch and back to 0 after.""" + client = self._make_client() + + # Patch _get_session to return a mock that yields a 200 response + mock_resp = AsyncMock() + mock_resp.status = 200 + mock_resp.text = AsyncMock(return_value="hello world") + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock(return_value=False) + + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_resp) + + with patch.object(client, "_get_session", AsyncMock(return_value=mock_session)): + with patch("app.services.sec_http_client._is_sec_block_page", return_value=False): + with patch("os.path.exists", return_value=False): + result = await client.fetch_text("http://example.com/test") + + assert result == "hello world" + assert client._pending == 0, "pending must be 0 after completion" + + @pytest.mark.asyncio + async def test_queue_full_raises_immediately(self): + """When _pending >= _MAX_PENDING, fetch_text raises without touching SEC.""" + client = self._make_client() + client._pending = client._MAX_PENDING # Already at limit + + with patch("os.path.exists", return_value=False): + with pytest.raises(RuntimeError, match="SEC request queue full"): + await client.fetch_text("http://example.com/overload") + + @pytest.mark.asyncio + async def test_queue_full_json_raises_immediately(self): + """Same check for fetch_json.""" + client = self._make_client() + client._pending = client._MAX_PENDING + + with patch("os.path.exists", return_value=False): + with pytest.raises(RuntimeError, match="SEC request queue full"): + await client.fetch_json("http://example.com/overload.json") + + @pytest.mark.asyncio + async def test_pending_decrements_on_exception(self): + """_pending must be decremented even when an exception is raised.""" + client = self._make_client() + assert client._pending == 0 + + with patch("os.path.exists", return_value=False): + with patch.object(client, "_get_session", AsyncMock(side_effect=RuntimeError("boom"))): + with pytest.raises(Exception): + await client.fetch_text("http://example.com/fail") + + assert client._pending == 0, "pending must return to 0 after exception" + + @pytest.mark.asyncio + async def test_rate_reduced_to_8(self): + """Rate limiter must be 8.0 req/sec, not 10.0.""" + from app.services.sec_http_client import SECHttpClient + client = SECHttpClient("Test") + assert client._rate_limiter._rate == 8.0 + assert client._rate_limiter._capacity == 8.0 + + +# --------------------------------------------------------------------------- +# Fix 2: Negative cache helpers +# --------------------------------------------------------------------------- + +class TestNegativeCacheHelpers: + @pytest.mark.asyncio + async def test_get_negative_cached_miss(self): + """Returns False when key not in Redis.""" + from app.utils.cache import get_negative_cached + mock_redis = AsyncMock() + mock_redis.exists = AsyncMock(return_value=0) + + with patch("app.utils.cache.get_redis", AsyncMock(return_value=mock_redis)): + result = await get_negative_cached("filings:exhibit:404:ACC:EX-99.1") + + assert result is False + + @pytest.mark.asyncio + async def test_get_negative_cached_hit(self): + """Returns True when key exists in Redis.""" + from app.utils.cache import get_negative_cached + mock_redis = AsyncMock() + mock_redis.exists = AsyncMock(return_value=1) + + with patch("app.utils.cache.get_redis", AsyncMock(return_value=mock_redis)): + result = await get_negative_cached("filings:exhibit:404:ACC:EX-99.1") + + assert result is True + + @pytest.mark.asyncio + async def test_set_negative_cached_calls_redis(self): + """set_negative_cached calls redis.set with b'1' and the TTL.""" + from app.utils.cache import set_negative_cached + mock_redis = AsyncMock() + mock_redis.set = AsyncMock() + + with patch("app.utils.cache.get_redis", AsyncMock(return_value=mock_redis)): + await set_negative_cached("filings:exhibit:404:ACC:EX-99.2", ttl=3600) + + mock_redis.set.assert_called_once_with( + "filings:exhibit:404:ACC:EX-99.2", b"1", ex=3600 + ) + + @pytest.mark.asyncio + async def test_negative_cache_graceful_no_redis(self): + """Both helpers return gracefully when Redis is unavailable.""" + from app.utils.cache import get_negative_cached, set_negative_cached + + with patch("app.utils.cache.get_redis", AsyncMock(return_value=None)): + assert await get_negative_cached("any:key") is False + await set_negative_cached("any:key") # must not raise + + @pytest.mark.asyncio + async def test_negative_cache_graceful_redis_error(self): + """Both helpers swallow Redis errors.""" + from app.utils.cache import get_negative_cached, set_negative_cached + mock_redis = AsyncMock() + mock_redis.exists = AsyncMock(side_effect=Exception("connection reset")) + mock_redis.set = AsyncMock(side_effect=Exception("connection reset")) + + with patch("app.utils.cache.get_redis", AsyncMock(return_value=mock_redis)): + assert await get_negative_cached("any:key") is False + await set_negative_cached("any:key") # must not raise + + +# --------------------------------------------------------------------------- +# Fix 3: Exhibit endpoint negative caching +# --------------------------------------------------------------------------- + +class TestExhibitEndpointNegativeCaching: + """Test that the GET /exhibit endpoint checks & sets the negative cache.""" + + def _make_request(self, accession_number="0001234567-26-000001", exhibit_type="EX-99.1"): + return {"accession_number": accession_number, "exhibit_type": exhibit_type} + + @pytest.mark.asyncio + async def test_negative_cache_hit_returns_404_immediately(self): + """When negative cache is set, endpoint returns 404 without calling service.""" + from fastapi import HTTPException + from app.api.v1.endpoints.filings import get_exhibit_content + + mock_db = AsyncMock() + + with patch("app.api.v1.endpoints.filings.get_negative_cached", AsyncMock(return_value=True)): + with patch("app.api.v1.endpoints.filings.sec_filings_service") as mock_svc: + with pytest.raises(HTTPException) as exc_info: + await get_exhibit_content( + accession_number="0001234567-26-000001", + response=MagicMock(), + exhibit_type="EX-99.1", + db=mock_db, + ) + + assert exc_info.value.status_code == 404 + mock_svc.get_exhibit_content.assert_not_called() + + @pytest.mark.asyncio + async def test_value_error_sets_negative_cache(self): + """When service raises ValueError (exhibit not found), negative cache is set.""" + from fastapi import HTTPException + from app.api.v1.endpoints.filings import get_exhibit_content + + mock_db = AsyncMock() + set_neg = AsyncMock() + + with patch("app.api.v1.endpoints.filings.get_negative_cached", AsyncMock(return_value=False)): + with patch("app.api.v1.endpoints.filings.set_negative_cached", set_neg): + with patch("app.api.v1.endpoints.filings.sec_filings_service") as mock_svc: + mock_svc.get_exhibit_content = AsyncMock( + side_effect=ValueError("Exhibit EX-99.1 not found in filing ...") + ) + with pytest.raises(HTTPException) as exc_info: + await get_exhibit_content( + accession_number="0001234567-26-000001", + response=MagicMock(), + exhibit_type="EX-99.1", + db=mock_db, + ) + + assert exc_info.value.status_code == 404 + set_neg.assert_called_once() + # Verify the key and TTL + call_args = set_neg.call_args + assert "EX-99.1" in call_args[0][0] + assert call_args[1].get("ttl") == 3600 or (len(call_args[0]) > 1 and call_args[0][1] == 3600) + + @pytest.mark.asyncio + async def test_non_value_error_does_not_set_negative_cache(self): + """Network/SEC errors (non-ValueError) must NOT populate negative cache.""" + from fastapi import HTTPException + from app.api.v1.endpoints.filings import get_exhibit_content + + mock_db = AsyncMock() + set_neg = AsyncMock() + + with patch("app.api.v1.endpoints.filings.get_negative_cached", AsyncMock(return_value=False)): + with patch("app.api.v1.endpoints.filings.set_negative_cached", set_neg): + with patch("app.api.v1.endpoints.filings.sec_filings_service") as mock_svc: + mock_svc.get_exhibit_content = AsyncMock( + side_effect=RuntimeError("SEC request queue full") + ) + with pytest.raises(HTTPException) as exc_info: + await get_exhibit_content( + accession_number="0001234567-26-000001", + response=MagicMock(), + exhibit_type="EX-99.1", + db=mock_db, + ) + + assert exc_info.value.status_code == 502 + set_neg.assert_not_called() + + @pytest.mark.asyncio + async def test_successful_fetch_does_not_set_negative_cache(self): + """Successful exhibit fetch must not touch the negative cache.""" + from app.api.v1.endpoints.filings import get_exhibit_content + + mock_db = AsyncMock() + set_neg = AsyncMock() + + with patch("app.api.v1.endpoints.filings.get_negative_cached", AsyncMock(return_value=False)): + with patch("app.api.v1.endpoints.filings.set_negative_cached", set_neg): + with patch("app.api.v1.endpoints.filings.sec_filings_service") as mock_svc: + mock_svc.get_exhibit_content = AsyncMock(return_value={ + "content": "Press Release", + "content_type": "text/html", + "filename": "exhibit99-1.htm", + "url": "https://www.sec.gov/Archives/edgar/data/123/000123/exhibit99-1.htm", + }) + # Bypass the @with_cache decorator by patching set_cached_response + with patch("app.utils.cache.set_cached_response", AsyncMock(return_value="etag")): + with patch("app.utils.cache.get_cached_response", AsyncMock(return_value=None)): + result = await get_exhibit_content( + accession_number="0001234567-26-000001", + response=MagicMock(), + exhibit_type="EX-99.1", + db=mock_db, + ) + + assert result is not None + set_neg.assert_not_called() + + +# --------------------------------------------------------------------------- +# Fix 4 (large exhibit): max_bytes pre-check and memory cache guard +# --------------------------------------------------------------------------- + +class TestFetchTextMaxBytes: + def _make_client(self): + from app.services.sec_http_client import SECHttpClient, _TokenBucket + client = SECHttpClient.__new__(SECHttpClient) + client._pending = 0 + client._MAX_PENDING = 50 + client._text_cache = {} + client._json_cache = {} + client._cache_dir = "/tmp/test_sec_cache_mb" + client._req_sem = asyncio.Semaphore(2) + client._rate_limiter = _TokenBucket(rate=100.0, capacity=100.0) + client.http_timeout = MagicMock() + client.http_timeout.total = 12 + client._session = None + return client + + def _make_resp(self, content_length=None, body=b"x" * 200): + mock_resp = AsyncMock() + mock_resp.status = 200 + mock_resp.content_length = content_length + mock_resp.get_encoding = MagicMock(return_value="utf-8") + # content.read returns the body bytes + mock_resp.content = AsyncMock() + mock_resp.content.read = AsyncMock(return_value=body) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock(return_value=False) + return mock_resp + + @pytest.mark.asyncio + async def test_content_length_too_large_raises_immediately(self): + """fetch_text with max_bytes rejects via Content-Length header before reading body.""" + client = self._make_client() + mock_resp = self._make_resp(content_length=5000, body=b"x" * 5000) + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_resp) + + with patch.object(client, "_get_session", AsyncMock(return_value=mock_session)): + with patch("os.path.exists", return_value=False): + with pytest.raises(ValueError, match="too large"): + await client.fetch_text("http://example.com/big", max_bytes=100) + + # Body should not have been read when Content-Length already exceeded + mock_resp.content.read.assert_not_called() + + @pytest.mark.asyncio + async def test_no_content_length_body_too_large_raises(self): + """fetch_text with max_bytes rejects via body read when Content-Length is absent.""" + client = self._make_client() + # No Content-Length header, but body exceeds limit + mock_resp = self._make_resp(content_length=None, body=b"x" * 200) + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_resp) + + with patch.object(client, "_get_session", AsyncMock(return_value=mock_session)): + with patch("os.path.exists", return_value=False): + with pytest.raises(ValueError, match="exceeded"): + await client.fetch_text("http://example.com/big2", max_bytes=100) + + @pytest.mark.asyncio + async def test_large_text_not_stored_in_memory_cache(self): + """Responses > 1MB must NOT be stored in _text_cache (only disk cache).""" + client = self._make_client() + large_body = b"A" * (2 * 1024 * 1024) # 2MB + mock_resp = AsyncMock() + mock_resp.status = 200 + mock_resp.content_length = None + mock_resp.text = AsyncMock(return_value=large_body.decode("utf-8")) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock(return_value=False) + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_resp) + + with patch.object(client, "_get_session", AsyncMock(return_value=mock_session)): + with patch("app.services.sec_http_client._is_sec_block_page", return_value=False): + with patch("os.path.exists", return_value=False): + with patch("builtins.open", MagicMock()): + result = await client.fetch_text("http://example.com/large") + + assert "http://example.com/large" not in client._text_cache + assert len(result) == len(large_body) + + @pytest.mark.asyncio + async def test_small_text_stored_in_memory_cache(self): + """Responses <= 1MB should still be stored in _text_cache.""" + client = self._make_client() + small_body = "hello world" + mock_resp = AsyncMock() + mock_resp.status = 200 + mock_resp.content_length = None + mock_resp.text = AsyncMock(return_value=small_body) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock(return_value=False) + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_resp) + + with patch.object(client, "_get_session", AsyncMock(return_value=mock_session)): + with patch("app.services.sec_http_client._is_sec_block_page", return_value=False): + with patch("os.path.exists", return_value=False): + with patch("builtins.open", MagicMock()): + result = await client.fetch_text("http://example.com/small") + + assert client._text_cache.get("http://example.com/small") == small_body + + @pytest.mark.asyncio + async def test_size_error_does_not_retry(self): + """ValueError from max_bytes must propagate immediately without retrying.""" + client = self._make_client() + mock_resp = self._make_resp(content_length=None, body=b"x" * 200) + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_resp) + call_count = 0 + + original_get = mock_session.get + + def counting_get(*args, **kwargs): + nonlocal call_count + call_count += 1 + return original_get(*args, **kwargs) + + mock_session.get = counting_get + + with patch.object(client, "_get_session", AsyncMock(return_value=mock_session)): + with patch("os.path.exists", return_value=False): + with pytest.raises(ValueError): + await client.fetch_text("http://example.com/nretry", max_bytes=100) + + # Should have tried exactly once (no retry on ValueError) + assert call_count == 1, f"Expected 1 attempt, got {call_count}"