Compare commits
3
Commits
d31badeb5b
...
4ee9ad064e
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4ee9ad064e | ||
|
|
187d29a372 | ||
|
|
ae56227c03 |
@@ -60,3 +60,24 @@ hit a consent wall from EU IPs, and the inner `AU_yqL` id is only resolvable via
|
||||
Google's private batchexecute API. Do not retry. The news seed uses Bing News
|
||||
RSS instead (`ENABLE_NEWS_SEED`, real publisher URL in the `?url=` param of
|
||||
apiclick.aspx — unwrapped by `_unwrap_news_link`).
|
||||
|
||||
## Large sources can OOM-kill the pod
|
||||
|
||||
On 2026-07-10 the pod was OOMKilled (memory limit was 1Gi) mid-research: a
|
||||
batch of 20 concurrent sources hit a 98k-word document plus several large PDFs
|
||||
at once, and pdfplumber's parse spiked RAM past the limit. The in-memory
|
||||
research task died with the pod and its session sat in `running` forever.
|
||||
|
||||
Mitigations now in place:
|
||||
|
||||
- Memory limit raised to 2Gi (`k8s-manifests/researchowl/deployment.yaml`).
|
||||
- PDFs capped at 15MB (was 50MB), checked both via Content-Length and actual
|
||||
body size; pdfplumber runs in `run_in_executor` (it is sync + CPU-heavy and
|
||||
also froze the event loop, same class of bug as DDGS) and flushes its page
|
||||
cache per page.
|
||||
- Extracted content is truncated to `max_content_length` (300k chars) before
|
||||
hitting `source_contents`.
|
||||
- On startup the bot marks orphaned `running` sessions as `interrupted`.
|
||||
|
||||
If a research still dies, the scraped sources survive in the DB: `/process`
|
||||
re-chunks and scores them without re-scraping.
|
||||
|
||||
@@ -851,6 +851,27 @@ async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
|
||||
# ─── Bot setup ────────────────────────────────────────────────────────────────
|
||||
|
||||
async def _mark_interrupted_on_startup(app: Application) -> None:
|
||||
"""Las tareas de research viven solo en memoria (_active_tasks): un
|
||||
reinicio del pod las mata sin tocar la DB, y sus sesiones quedan en
|
||||
'running' para siempre — parecen activas en /status y get_active_session.
|
||||
"""
|
||||
db_conn = await get_db()
|
||||
try:
|
||||
cursor = await db_conn.execute(
|
||||
"UPDATE research_sessions SET status = ?, updated_at = ? WHERE status = ?",
|
||||
(ResearchStatus.INTERRUPTED, time.time(), ResearchStatus.RUNNING),
|
||||
)
|
||||
await db_conn.commit()
|
||||
if cursor.rowcount:
|
||||
logger.info("Orphaned running sessions marked interrupted",
|
||||
count=cursor.rowcount)
|
||||
except Exception as e:
|
||||
logger.warning("Interrupted-mark failed — bot continues", error=str(e))
|
||||
finally:
|
||||
await db_conn.close()
|
||||
|
||||
|
||||
async def _purge_on_startup(app: Application) -> None:
|
||||
db_conn = await get_db()
|
||||
try:
|
||||
@@ -994,6 +1015,7 @@ async def _start_scheduler(app: Application) -> None:
|
||||
|
||||
|
||||
async def _on_startup(app: Application) -> None:
|
||||
await _mark_interrupted_on_startup(app)
|
||||
await _purge_on_startup(app)
|
||||
await _start_scheduler(app)
|
||||
|
||||
|
||||
@@ -47,6 +47,8 @@ class Settings(BaseSettings):
|
||||
request_timeout: int = Field(30)
|
||||
request_delay: float = Field(1.0) # seconds between requests
|
||||
min_content_length: int = Field(200) # chars
|
||||
# Libros/dumps enteros (100k+ palabras) inflan RAM y DB sin aportar al RAG
|
||||
max_content_length: int = Field(300_000) # chars
|
||||
|
||||
# Fuentes opcionales — desactivadas por defecto: la IP del homelab está
|
||||
# bloqueada por Reddit (403) y YouTube (transcripts vacíos), eran peso muerto.
|
||||
|
||||
@@ -18,6 +18,7 @@ class ResearchStatus(str, Enum):
|
||||
SATURATED = "saturated"
|
||||
FINISHED = "finished"
|
||||
ERROR = "error"
|
||||
INTERRUPTED = "interrupted" # el pod se reinició con el research en marcha
|
||||
|
||||
|
||||
class OutputType(str, Enum):
|
||||
|
||||
@@ -628,6 +628,11 @@ class ExhaustiveScraper:
|
||||
error="Content too short or empty")
|
||||
return
|
||||
|
||||
if len(content) > settings.max_content_length:
|
||||
logger.info("Content truncated", source_id=source_id,
|
||||
original_length=len(content), url=url[:60])
|
||||
content = content[:settings.max_content_length]
|
||||
|
||||
word_count = len(content.split())
|
||||
|
||||
await self.db.save_source_content(source_id, content)
|
||||
@@ -819,30 +824,49 @@ class ExhaustiveScraper:
|
||||
entries=len(entries), added=added)
|
||||
return added
|
||||
|
||||
# pdfplumber es síncrono y CPU-intensivo: parsear inline congela el event
|
||||
# loop, y con PDFs grandes el pico de RAM puede matar el pod (OOM con
|
||||
# límite de 1-2Gi). Ejecutar SIEMPRE vía run_in_executor.
|
||||
|
||||
@staticmethod
|
||||
def _parse_pdf_sync(path: str) -> str:
|
||||
import pdfplumber
|
||||
|
||||
with pdfplumber.open(path) as pdf:
|
||||
pages = []
|
||||
for page in pdf.pages[:50]: # max 50 pages
|
||||
pages.append(page.extract_text() or "")
|
||||
page.flush_cache() # pdfplumber cachea objetos de página: liberar
|
||||
return "\n\n".join(pages)
|
||||
|
||||
async def _extract_pdf(self, url: str) -> tuple[Optional[str], Optional[str]]:
|
||||
"""Download and extract PDF text"""
|
||||
import pdfplumber
|
||||
import tempfile
|
||||
import os
|
||||
|
||||
max_pdf_bytes = 15 * 1024 * 1024 # varios PDFs concurrentes en RAM: cap agresivo
|
||||
|
||||
http = await self._get_http()
|
||||
try:
|
||||
async with http.get(url) as resp:
|
||||
if resp.status != 200:
|
||||
return None, None
|
||||
content_length = int(resp.headers.get("content-length", 0))
|
||||
if content_length > 50 * 1024 * 1024: # skip PDFs > 50MB
|
||||
if content_length > max_pdf_bytes:
|
||||
return None, None
|
||||
pdf_bytes = await resp.read()
|
||||
# Sin Content-Length el check anterior no protege
|
||||
if len(pdf_bytes) > max_pdf_bytes:
|
||||
return None, None
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
|
||||
f.write(pdf_bytes)
|
||||
tmp_path = f.name
|
||||
del pdf_bytes
|
||||
|
||||
try:
|
||||
with pdfplumber.open(tmp_path) as pdf:
|
||||
pages = [p.extract_text() or "" for p in pdf.pages[:50]] # max 50 pages
|
||||
text = "\n\n".join(pages)
|
||||
loop = asyncio.get_running_loop()
|
||||
text = await loop.run_in_executor(None, self._parse_pdf_sync, tmp_path)
|
||||
return text, url.split("/")[-1]
|
||||
finally:
|
||||
os.unlink(tmp_path)
|
||||
|
||||
Reference in New Issue
Block a user