perf(db): conexión SQLite compartida por proceso
Build & Deploy ResearchOwl / build-and-push (push) Successful in 5s

get_db() devuelve un proxy sobre una única conexión real reutilizada durante
toda la vida del proceso, en vez de abrir una conexión nueva y re-ejecutar
todo el SCHEMA en cada comando/scoring.

- _SharedConnection: proxy con close() no-op → los handlers conservan el
  patrón get_db()/finally close() sin cambios
- aiosqlite serializa las operaciones en el hilo de la conexión: compartirla
  entre coroutines es seguro y elimina la contención del lock de escritura a
  nivel de fichero (scheduler solapado, /compare con 2 sesiones)
- close_db() vía post_shutdown para checkpoint WAL limpio al apagar

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
ChemaVX
2026-06-15 15:16:24 +00:00
co-authored by Claude Opus 4.8
parent 2ffad8aad3
commit cd557a2d71
2 changed files with 65 additions and 13 deletions
+6 -1
View File
@@ -18,7 +18,7 @@ from telegram.ext import (
from telegram.constants import ParseMode from telegram.constants import ParseMode
from src.config import settings from src.config import settings
from src.db.database import get_db, ResearchDB, ResearchStatus, OutputType from src.db.database import get_db, close_db, ResearchDB, ResearchStatus, OutputType
from src.scraper.exhaustive import ExhaustiveScraper from src.scraper.exhaustive import ExhaustiveScraper
from src.processor.processor import OllamaClient, ContentProcessor from src.processor.processor import OllamaClient, ContentProcessor
from src.generator.generator import OutputGenerator from src.generator.generator import OutputGenerator
@@ -913,6 +913,10 @@ async def _on_startup(app: Application) -> None:
await _start_scheduler(app) await _start_scheduler(app)
async def _on_shutdown(app: Application) -> None:
await close_db()
async def cmd_export(update: Update, ctx: ContextTypes.DEFAULT_TYPE): async def cmd_export(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not is_authorized(update.effective_user.id): if not is_authorized(update.effective_user.id):
return return
@@ -1259,6 +1263,7 @@ def create_bot() -> Application:
Application.builder() Application.builder()
.token(settings.telegram_bot_token) .token(settings.telegram_bot_token)
.post_init(_on_startup) .post_init(_on_startup)
.post_shutdown(_on_shutdown)
.build() .build()
) )
+59 -12
View File
@@ -1,3 +1,4 @@
import asyncio
import aiosqlite import aiosqlite
import json import json
import time import time
@@ -117,19 +118,65 @@ CREATE TABLE IF NOT EXISTS watched_topics (
""" """
# Conexión única compartida por proceso. aiosqlite serializa todas las
# operaciones en el hilo de la conexión, así que compartirla entre coroutines
# es seguro y además evita la contención del lock de escritura a nivel de
# fichero que sufrían las conexiones múltiples (scheduler, /compare).
_shared_conn: Optional[aiosqlite.Connection] = None
_init_lock = asyncio.Lock()
class _SharedConnection:
"""Proxy sobre la conexión compartida cuyo close() es no-op.
Permite que los handlers sigan usando el patrón `db = await get_db()`
`finally: await db.close()` sin cambios, mientras por debajo se reutiliza
una sola conexión real durante toda la vida del proceso.
"""
def __init__(self, conn: aiosqlite.Connection):
self._conn = conn
def __getattr__(self, name):
return getattr(self._conn, name)
async def close(self):
# No cerrar: la conexión es compartida por todo el proceso.
pass
async def _init_shared() -> aiosqlite.Connection:
global _shared_conn
if _shared_conn is None:
async with _init_lock:
if _shared_conn is None: # doble check tras adquirir el lock
Path(settings.db_path).parent.mkdir(parents=True, exist_ok=True)
conn = await aiosqlite.connect(settings.db_path)
conn.row_factory = aiosqlite.Row
await conn.execute("PRAGMA journal_mode=WAL")
await conn.execute("PRAGMA synchronous=NORMAL")
# Espera hasta 5s si OTRO proceso tiene el lock de escritura
# antes de fallar con "database is locked".
await conn.execute("PRAGMA busy_timeout=5000")
await conn.executescript(SCHEMA)
await conn.commit()
_shared_conn = conn
logger.info("Shared DB connection initialized", path=settings.db_path)
return _shared_conn
async def get_db() -> aiosqlite.Connection: async def get_db() -> aiosqlite.Connection:
Path(settings.db_path).parent.mkdir(parents=True, exist_ok=True) conn = await _init_shared()
db = await aiosqlite.connect(settings.db_path) return _SharedConnection(conn)
db.row_factory = aiosqlite.Row
await db.execute("PRAGMA journal_mode=WAL")
await db.execute("PRAGMA synchronous=NORMAL") async def close_db() -> None:
# Espera hasta 5s si otra conexión tiene el lock de escritura (scheduler """Cierra la conexión compartida (checkpoint WAL). Llamar al apagar el bot."""
# solapado con research activa, o las 2 sesiones de /compare) en vez de global _shared_conn
# fallar al instante con "database is locked". if _shared_conn is not None:
await db.execute("PRAGMA busy_timeout=5000") try:
await db.executescript(SCHEMA) await _shared_conn.close()
await db.commit() finally:
return db _shared_conn = None
class ResearchDB: class ResearchDB: