From a14e5b99f9d4e191aebbbf75302f1369421bdd77 Mon Sep 17 00:00:00 2001 From: ChemaVX Date: Tue, 1 Sep 2026 17:27:21 +0000 Subject: [PATCH] =?UTF-8?q?fix(db):=20un=20spec=20no=20envejece=20con=20la?= =?UTF-8?q?=20investigaci=C3=B3n=20que=20lo=20origin=C3=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit La retención iba por la edad de la SESIÓN y cascadeaba `DELETE FROM outputs WHERE session_id = ?`, así que un output generado ayer sobre una sesión de julio moría en el siguiente arranque del bot. Pasó hoy, 2026-09-01, al desplegar 198b0e62: Startup purge done sessions=12 outputs=27 chunks=2030 sources=3466 api_usage=1118 shorts=6 El pod llevaba 18 días sin reiniciarse, así que tres semanas de material cumplieron los 30 días de golpe. Y lo que se llevó por delante no fue lo viejo: los outputs 132-139 tenían 18,8 días —los tres Shorts re-renderizados el día antes entre ellos— pero colgaban de las sesiones 161-165, de hace 40. Murieron por la edad de su madre mientras 128-131, más antiguos, sobrevivían. Ahora la purga va en dos fases: 1. Outputs por SU propia fecha, vivan en la sesión que vivan. 2. Sesiones viejas que ya no sostienen ningún output. El orden importa: la sesión cuyos outputs eran todos viejos se queda sin ninguno en la fase 1 y resulta purgable en la fase 2, así que el caso normal —sesión vieja, material viejo— sigue limpiándose entero en UNA pasada. Hay un test que lo fija, y es el único de los cuatro nuevos que pasa también con la lógica anterior: está para probar que no se rompió lo que funcionaba. Una sesión con un output vivo sobrevive entera, con sus sources y sus chunks. No es generosidad: los chunks son contra lo que se comprueba el fundamento de ese output, y conservar el spec tirando aquello con lo que se verifica deja algo que ya no se puede auditar. El precio es que la retención afloja, y se paga a sabiendas. Dos cosas más que salieron al mirarlo: - `_purge_on_startup` sólo escribía en el log `if result["sessions"] > 0`. Con la retención por output, una pasada puede borrar 27 outputs y CERO sesiones, y eso no habría dejado ni una línea. Una purga silenciosa es como se descubre tres semanas tarde. Ahora informa si borró cualquier cosa. - El MP4 se llama por sesión, así que cuando la fase 1 se lleva el último short_en de una sesión que sigue viva, el fichero queda sin nada que lo nombre. Se borra ahí también, o el PVC acumula vídeos que no aparecen en ninguna fila. `purge_old_sessions` pasa a llamarse `purge_old_data`: ya no purga sólo por sesiones y el nombre viejo describía justo el defecto. La BD anterior a la purga está a salvo y verificada en ~/rescates/researchowl-purga-2026-09-01 (integrity_check ok, 31 outputs, los 17 short_en). Con esta lógica, un restore conserva 16 de los 17. Suite: 278 pasan. Co-Authored-By: Claude Opus 5 --- src/bot/bot.py | 13 ++-- src/db/database.py | 94 ++++++++++++++++++++------- tests/test_short_producer.py | 120 ++++++++++++++++++++++++++++++++++- 3 files changed, 200 insertions(+), 27 deletions(-) diff --git a/src/bot/bot.py b/src/bot/bot.py index 8a91b5a..fccf726 100644 --- a/src/bot/bot.py +++ b/src/bot/bot.py @@ -1346,8 +1346,12 @@ async def _purge_on_startup(app: Application) -> None: db_conn = await get_db() try: db = ResearchDB(db_conn) - result = await db.purge_old_sessions(30) - if result["sessions"] > 0: + result = await db.purge_old_data(30) + # Cualquier borrado, no sólo el de sesiones. Con la retención por fecha + # de output, una pasada puede llevarse 27 outputs y CERO sesiones — y + # con la condición anterior eso no dejaba ni una línea de log. Una purga + # silenciosa es como se descubre tres semanas tarde. + if any(result.values()): logger.info("Startup purge done", **result) except Exception as e: logger.warning("Startup purge failed — bot continues", error=str(e)) @@ -1600,7 +1604,8 @@ async def cmd_purge(update: Update, ctx: ContextTypes.DEFAULT_TYPE): return if days == 0 and not (len(args) >= 2 and args[1] == "confirm"): await update.message.reply_text( - "⚠️ Esto borrará *todas* las sesiones completadas.\n" + "⚠️ Esto borrará *todos* los outputs y *todas* las sesiones " + "completadas.\n" "Envía `/purge 0 confirm` para confirmar.", parse_mode=ParseMode.MARKDOWN ) @@ -1609,7 +1614,7 @@ async def cmd_purge(update: Update, ctx: ContextTypes.DEFAULT_TYPE): db_conn = await get_db() try: db = ResearchDB(db_conn) - result = await db.purge_old_sessions(days) + result = await db.purge_old_data(days) await update.message.reply_text( f"🗑️ Purged: {result['sessions']} sessions, " f"{result['sources']} sources, " diff --git a/src/db/database.py b/src/db/database.py index 05eed1a..550d7aa 100644 --- a/src/db/database.py +++ b/src/db/database.py @@ -648,41 +648,75 @@ class ResearchDB: # --- Maintenance --- - async def purge_old_sessions(self, max_age_days: int = 30) -> dict: + async def purge_old_data(self, max_age_days: int = 30) -> dict: + """Retención en dos fases: primero los outputs por SU fecha, luego las + sesiones que ya no sostienen nada. + + Antes iba todo por la edad de la sesión, y la cascada + `DELETE FROM outputs WHERE session_id = ?` se llevaba por delante lo + generado ayer si colgaba de una sesión de hace dos meses. El + 2026-09-01 eso borró 27 outputs, entre ellos los tres Shorts + re-renderizados el día antes: sobrevivieron los outputs 128-131 y + murieron los 132-139, que eran **más nuevos**. Un spec no envejece con + la investigación que lo originó. + + Las dos fases van en este orden por una razón: la sesión cuyos outputs + eran todos viejos se queda sin ninguno en la fase 1 y resulta purgable + en la fase 2, así que el caso normal —sesión vieja, material viejo— + sigue limpiándose igual que antes en una sola pasada. + + **Y una sesión con un output vivo sobrevive entera**, con sus sources y + sus chunks. No es generosidad: los `chunks` son contra lo que se + comprueba el fundamento de ese output, así que conservar el spec y tirar + aquello con lo que se verifica deja algo que ya no se puede auditar. El + precio es que la retención afloja — una sesión de julio con un Short de + ayer mantiene vivos sus cientos de sources — y ese precio se paga a + sabiendas. + """ await self.db.execute("PRAGMA foreign_keys = ON") threshold = time.time() - max_age_days * 86400 + counts = {"sessions": 0, "sources": 0, "chunks": 0, "outputs": 0, + "api_usage": 0, "shorts": 0} + + # --- fase 1: outputs por su propia fecha, vivan donde vivan --------- + # Se apuntan las sesiones tocadas antes de borrar: si una se queda sin + # ningún short_en, su MP4 no lo referencia ya nadie. cursor = await self.db.execute( - "SELECT id FROM research_sessions WHERE created_at < ? AND status != 'running'", + "SELECT DISTINCT session_id FROM outputs WHERE created_at < ?", + (threshold,) + ) + touched = [row[0] for row in await cursor.fetchall()] + cur = await self.db.execute("DELETE FROM outputs WHERE created_at < ?", + (threshold,)) + counts["outputs"] += cur.rowcount + + for sid in touched: + cursor = await self.db.execute( + "SELECT 1 FROM outputs WHERE session_id = ? AND output_type = ?" + " LIMIT 1", (sid, "short_en") + ) + if await cursor.fetchone() is None: + counts["shorts"] += self._drop_short(sid) + + # --- fase 2: sesiones viejas que ya no sostienen ningún output ------ + cursor = await self.db.execute( + "SELECT id FROM research_sessions WHERE created_at < ?" + " AND status != 'running'" + " AND NOT EXISTS (SELECT 1 FROM outputs WHERE session_id =" + " research_sessions.id)", (threshold,) ) session_ids = [row[0] for row in await cursor.fetchall()] - counts = {"sessions": 0, "sources": 0, "chunks": 0, "outputs": 0, - "api_usage": 0, "shorts": 0} - for sid in session_ids: - # El MP4 del Short vive en disco (los blobs en SQLite hacen - # patológico el WAL), así que su borrado no lo arrastra ninguna FK: - # se hace aquí, que es el único sitio que sabe qué sesiones - # desaparecen. Best-effort — un fichero que no se puede borrar no - # va a impedir purgar la sesión. - try: - video = Path(settings.shorts_dir) / f"{sid}.mp4" - if video.is_file(): - video.unlink() - counts["shorts"] += 1 - except OSError as e: - logger.warning("No se pudo borrar el Short de una sesión purgada", - session_id=sid, error=str(e)) + counts["shorts"] += self._drop_short(sid) await self.db.execute( "DELETE FROM source_contents WHERE source_id IN (SELECT id FROM sources WHERE session_id = ?)", (sid,) ) cur = await self.db.execute("DELETE FROM chunks WHERE session_id = ?", (sid,)) counts["chunks"] += cur.rowcount - cur = await self.db.execute("DELETE FROM outputs WHERE session_id = ?", (sid,)) - counts["outputs"] += cur.rowcount cur = await self.db.execute("DELETE FROM api_usage WHERE session_id = ?", (sid,)) counts["api_usage"] += cur.rowcount cur = await self.db.execute("DELETE FROM sources WHERE session_id = ?", (sid,)) @@ -691,6 +725,22 @@ class ResearchDB: counts["sessions"] += cur.rowcount await self.db.commit() - logger.info("Purged sessions older than days", - sessions=counts["sessions"], days=max_age_days) + logger.info("Purga por antigüedad", days=max_age_days, **counts) return counts + + def _drop_short(self, sid: int) -> int: + """Borra el MP4 de una sesión, si queda. Devuelve 1 si borró algo. + + El vídeo vive en disco (los blobs en SQLite hacen patológico el WAL), + así que su borrado no lo arrastra ninguna FK y hay que hacerlo aquí. + Best-effort: un fichero que no se puede borrar no va a impedir la purga. + """ + try: + video = Path(settings.shorts_dir) / f"{sid}.mp4" + if video.is_file(): + video.unlink() + return 1 + except OSError as e: + logger.warning("No se pudo borrar el Short de una sesión purgada", + session_id=sid, error=str(e)) + return 0 diff --git a/tests/test_short_producer.py b/tests/test_short_producer.py index 8a96d95..3f7afe4 100644 --- a/tests/test_short_producer.py +++ b/tests/test_short_producer.py @@ -331,7 +331,7 @@ async def test_purging_a_session_takes_its_video_with_it(tmp_path, monkeypatch): " created_at, updated_at) VALUES (2,'nuevo','saturated',1,?,?)", (now, now)) await conn.commit() - counts = await ResearchDB(conn).purge_old_sessions(30) + counts = await ResearchDB(conn).purge_old_data(30) await conn.close() assert counts["shorts"] == 1 @@ -339,6 +339,124 @@ async def test_purging_a_session_takes_its_video_with_it(tmp_path, monkeypatch): assert (shorts / "2.mp4").exists(), "la sesión reciente conserva su vídeo" +async def purge_fixture(tmp_path, monkeypatch, sessions, outputs, days=30): + """Una BD con sesiones y outputs de las edades que se le pidan, purgada. + + `sessions` y `outputs` llevan la edad en días: positiva es pasado. Devuelve + (counts, short_en supervivientes, ids de sesión supervivientes). + """ + import time + + import aiosqlite + + from src.db import database + from src.db.database import ResearchDB + + shorts = tmp_path / "shorts" + shorts.mkdir() + monkeypatch.setattr(settings, "shorts_dir", str(shorts)) + + conn = await aiosqlite.connect(tmp_path / "p.db") + conn.row_factory = aiosqlite.Row + await conn.executescript(database.SCHEMA) + now = time.time() + + for sid, age in sessions: + t = now - age * 86400 + await conn.execute( + "INSERT INTO research_sessions (id, topic, status, telegram_chat_id," + " created_at, updated_at) VALUES (?,?, 'saturated', 1, ?, ?)", + (sid, f"s{sid}", t, t)) + (shorts / f"{sid}.mp4").write_bytes(b"x") + # Un source por sesión, para ver si la cascada la alcanza o no. + await conn.execute( + "INSERT INTO sources (session_id, url, title, scraped_at)" + " VALUES (?,?,?,?)", (sid, f"http://x/{sid}", "t", t)) + + for oid, sid, age in outputs: + await conn.execute( + "INSERT INTO outputs (id, session_id, output_type, content, created_at)" + " VALUES (?,?, 'short_en', '{}', ?)", (oid, sid, now - age * 86400)) + await conn.commit() + + try: + counts = await ResearchDB(conn).purge_old_data(days) + vivos = [r[0] for r in await (await conn.execute( + "SELECT id FROM outputs ORDER BY id")).fetchall()] + sesiones = [r[0] for r in await (await conn.execute( + "SELECT id FROM research_sessions ORDER BY id")).fetchall()] + finally: + # Sin esto, un fallo dentro del `try` deja el hilo de aiosqlite vivo y + # pytest no termina NUNCA: el error se presenta como un cuelgue, que es + # la forma más cara de leer un fallo. + await conn.close() + return counts, vivos, sesiones, shorts + + +@pytest.mark.asyncio +async def test_a_new_output_on_an_old_session_survives(tmp_path, monkeypatch): + """El fallo del 2026-09-01, exacto. La cascada iba por la edad de la SESIÓN, + así que los outputs 132-139 —generados el día antes sobre casos viejos— se + borraron mientras 128-131, más antiguos, sobrevivían. Un spec no envejece + con la investigación que lo originó. + """ + counts, vivos, sesiones, shorts = await purge_fixture( + tmp_path, monkeypatch, + sessions=[(1, 90)], # sesión de hace tres meses + outputs=[(139, 1, 1)]) # con un spec de ayer + + assert vivos == [139], "el spec de ayer murió con su sesión" + assert sesiones == [1], "la sesión tiene que sobrevivir o la FK se rompe" + assert counts["outputs"] == 0 + assert (shorts / "1.mp4").exists(), "y su vídeo con ella" + + +@pytest.mark.asyncio +async def test_an_old_output_dies_even_on_a_live_session(tmp_path, monkeypatch): + """La otra mitad, y sin ella lo de arriba no es retención por fecha de + output: es no purgar outputs nunca.""" + counts, vivos, sesiones, _ = await purge_fixture( + tmp_path, monkeypatch, + sessions=[(1, 2)], # sesión de anteayer + outputs=[(1, 1, 90), (2, 1, 2)]) # un spec viejo y uno reciente + + assert vivos == [2] + assert counts["outputs"] == 1 + assert sesiones == [1], "la sesión reciente no se toca" + + +@pytest.mark.asyncio +async def test_the_ordinary_case_still_cleans_up_whole(tmp_path, monkeypatch): + """Sesión vieja con material viejo: se limpia entera en UNA pasada. La fase 1 + la deja sin outputs y la fase 2 se la lleva — si el orden se invirtiera, la + sesión sobreviviría a su propio material hasta el arranque siguiente.""" + counts, vivos, sesiones, shorts = await purge_fixture( + tmp_path, monkeypatch, + sessions=[(1, 90), (2, 2)], + outputs=[(1, 1, 90), (2, 2, 2)]) + + assert vivos == [2] and sesiones == [2] + assert counts["sessions"] == 1 and counts["outputs"] == 1 + assert counts["sources"] == 1, "la cascada alcanza a la sesión purgada" + assert not (shorts / "1.mp4").exists() and (shorts / "2.mp4").exists() + + +@pytest.mark.asyncio +async def test_a_session_that_loses_its_last_short_loses_its_video(tmp_path, + monkeypatch): + """El MP4 se llama por sesión, así que cuando la fase 1 se lleva el último + short_en de una sesión que sigue viva, el fichero queda sin nada que lo + nombre. Sin esto el PVC acumula vídeos que ya no aparecen en ninguna fila.""" + counts, vivos, sesiones, shorts = await purge_fixture( + tmp_path, monkeypatch, + sessions=[(1, 2)], # la sesión sigue viva + outputs=[(1, 1, 90)]) # pero su único short se va + + assert vivos == [] and sesiones == [1] + assert counts["shorts"] == 1 + assert not (shorts / "1.mp4").exists() + + @pytest.mark.asyncio async def test_the_youtube_url_never_passes_for_an_article_url(tmp_path): """Subir un Short escribe su URL de YouTube en `published_url`. Si