diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index 25d18983..9db1bc20 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -3,6 +3,7 @@ import asyncio from unittest import mock import pytest +from sqlalchemy import text from wrolpi.conftest import await_switches from wrolpi.dates import from_timestamp @@ -332,6 +333,39 @@ async def test_compare_file_groups_survives_concurrent_commit( assert len(result.new) == 2 +@pytest.mark.asyncio +async def test_compare_file_groups_cleans_up_work_table_on_aborted_tx( + test_session, test_directory, make_files_structure): + """When compare_file_groups raises after a concurrent commit has put the + session into an aborted-transaction state, the work table must still be + dropped (no orphan tables left behind).""" + make_files_structure(['docs/file1.txt']) + + from wrolpi.files import worker as worker_mod + original = worker_mod._stream_filesystem_paths + + async def commit_then_abort_then_raise(root): + async for p in original(root): + yield p + test_session.commit() + # Poison the transaction so any later DDL fails unless rolled back. + try: + test_session.execute(text('SELECT FROM does_not_exist_xyz')) + except Exception: + pass + raise RuntimeError('simulated crash mid-scan') + + with mock.patch.object(worker_mod, '_stream_filesystem_paths', commit_then_abort_then_raise): + with pytest.raises(RuntimeError): + await compare_file_groups(test_directory) + + test_session.rollback() + leftover = [row.tablename for row in test_session.execute(text( + "SELECT tablename FROM pg_tables WHERE tablename LIKE 'fs_files_%'" + ))] + assert leftover == [], f'orphaned work tables: {leftover}' + + @pytest.mark.asyncio async def test_compare_file_groups_cancelable(test_session, test_directory, make_files_structure): """compare_file_groups can be canceled during long operations and reports progress via callback.""" diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 430c1a1a..69373c38 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -486,12 +486,23 @@ async def compare_file_groups( modified=modified, ) finally: - # Drop the work table regardless of success/failure. Isolate cleanup - # in its own try so a broken transaction doesn't mask the original error. + # Drop the work table regardless of success/failure. Prefer the caller's + # session so the DROP joins the same transaction as the CREATE (cleanup + # follows whatever the caller does with the tx). If the tx is aborted — + # e.g. an external commit + later SQL error left us mid-error — Postgres + # rejects DDL on the session; fall back to a fresh connection so the + # already-committed work table doesn't orphan. try: session.execute(text(f"DROP TABLE IF EXISTS {table_name}")) except Exception: - logger.exception(f'Failed to drop work table {table_name}') + from wrolpi.db import get_db_context + try: + engine, _ = get_db_context() + with engine.connect() as conn: + with conn.begin(): + conn.execute(text(f"DROP TABLE IF EXISTS {table_name}")) + except Exception: + logger.exception(f'Failed to drop work table {table_name}') async def build_move_plan_bulk(