Handling aborted-transaction case when dropping fs_files work table.

Greptile flagged that if the session is in an aborted tx (e.g. the
external commit was followed by a failed statement before our error
fired), Postgres rejects DDL on the session and the committed work
table orphans. Fall back to a fresh connection in that case. Adds
a regression test that deterministically reproduces the orphan.

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."""

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(