Do not let the definitions read flush a caller's pending write

Cross-family review raised that catching SQLAlchemyError around an autoflushing
query is broader than the failure being handled. If a caller's session is
carrying a pending mutation, our SELECT flushes it first, so an unrelated
IntegrityError would surface inside this helper, get swallowed as "no custom
columns", and leave the session needing a rollback for the rest of the request
-- silent, and a long way from its cause.

None of the five callers stage a write before calling this, so it is not
reachable today. Reading the definitions under no_autoflush closes it anyway,
since flushing someone else's work was never part of what this method needs.

Also closes two gaps the review found in the tests: nothing proved the request
could keep querying after a swallowed failure (the property that makes
degrading worth doing at all -- a poisoned session would still satisfy an
assertion of []), and every test asserted [], so an unconditional return []
would have left them all green while hiding custom columns from every healthy
library. Both are now pinned; the second is mutation-verified to fail against
that exact regression.

For #1153.

Author: new-usemame

cps/db.py

@@ -1904,11 +1904,23 @@ def get_cc_columns(self, config, filter_config_custom_read=False):
         """
         try:
             self.ensure_session()
-            tmp_cc = (self.session.query(CustomColumns)
-                      .filter(CustomColumns.datatype.notin_(cc_exceptions)).all())
+            session = self.session
+            # no_autoflush because this is a read of definitions, and the
+            # session may be carrying a caller's pending mutation. With
+            # autoflush on, our SELECT would flush *their* write first, so an
+            # unrelated IntegrityError would surface here, get swallowed as
+            # "no custom columns", and leave the session needing a rollback for
+            # the rest of the request. Not reachable from today's five callers
+            # (all read paths), but the failure would be silent and remote from
+            # its cause, so don't leave it available to the sixth.
+            with session.no_autoflush:
+                tmp_cc = (session.query(CustomColumns)
+                          .filter(CustomColumns.datatype.notin_(cc_exceptions)).all())
         except (SQLAlchemyError, AttributeError):
-            # AttributeError covers the reconnect window, where the session can
-            # be absent rather than merely unreadable.
+            # AttributeError: the session is absent rather than unreadable --
+            # `session` is None whenever session_factory is (before init_db, or
+            # after an explicit `session = None`), and None.query() is an
+            # AttributeError rather than a SQLAlchemyError.
             log.warning("Custom-column definitions unavailable; continuing without them",
                         exc_info=True)
             return []

tests/unit/test_book_detail_custom_columns_degrade.py

@@ -18,9 +18,10 @@
 import sqlite3
 import tempfile
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, text as sa_text
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import scoped_session, sessionmaker
 
@@ -95,6 +96,73 @@ def test_fixture_really_reproduces_an_unreadable_schema(unreadable_calibre_schem
         session.query(dbmod.CustomColumns).all()
 
 
+@pytest.mark.unit
+def test_request_can_keep_querying_after_the_failure(unreadable_calibre_schema):
+    """The safety property that actually matters, not just the return value.
+
+    Degrading is only useful if the *rest* of the page still renders. A caught
+    DB error can leave a Session in partial-rollback, where every later query
+    raises ``PendingRollbackError`` -- which would turn one dead page into a
+    subtly broken one and still satisfy a test that only asserts ``[]``.
+
+    So: fail the custom-column read, then keep using the same session.
+    """
+    from cps import db as dbmod
+    from cps.api import books as books_mod
+
+    session = dbmod.CalibreDB.session_factory()
+    session.execute(sa_text("CREATE TABLE books (id INTEGER PRIMARY KEY, title TEXT)"))
+    session.execute(sa_text("INSERT INTO books (id, title) VALUES (1, 'Still here')"))
+    session.commit()
+
+    assert books_mod.calibre_db.get_cc_columns(books_mod.config,
+                                               filter_config_custom_read=True) == []
+
+    # Same session, after the swallowed error.
+    assert session.execute(sa_text("SELECT title FROM books WHERE id = 1")).scalar() == "Still here"
+    assert session.is_active
+
+
+@pytest.mark.unit
+def test_healthy_library_still_returns_its_custom_columns():
+    """Guards the degrade tests against passing vacuously.
+
+    Every other test here asserts ``[]``, so an accidental unconditional
+    ``return []`` -- which would silently hide custom columns from every user
+    with a perfectly healthy library -- would leave them all green.
+    """
+    from cps import db as dbmod
+
+    tmp = Path(tempfile.mkdtemp()) / "metadata.db"
+    con = sqlite3.connect(tmp)
+    con.execute("""CREATE TABLE custom_columns (
+        id INTEGER PRIMARY KEY, label TEXT, name TEXT, datatype TEXT,
+        mark_for_delete BOOL, editable BOOL, display TEXT, is_multiple BOOL,
+        normalized BOOL)""")
+    con.execute("INSERT INTO custom_columns VALUES (1,'read','Read','bool',0,1,'{}',0,0)")
+    con.commit()
+    con.close()
+
+    engine = create_engine(f"sqlite:///{tmp}", future=True)
+    factory = scoped_session(sessionmaker(autocommit=False, autoflush=True,
+                                          bind=engine, future=True))
+    previous_factory = dbmod.CalibreDB.session_factory
+    previous_init = dbmod.CalibreDB._init
+    dbmod.CalibreDB.session_factory = factory
+    dbmod.CalibreDB._init = True
+    try:
+        cdb = dbmod.CalibreDB.__new__(dbmod.CalibreDB)
+        cc = cdb.get_cc_columns(SimpleNamespace(config_columns_to_ignore=None,
+                                                config_read_column=0))
+        assert [c.name for c in cc] == ["Read"]
+    finally:
+        factory.remove()
+        engine.dispose()
+        dbmod.CalibreDB.session_factory = previous_factory
+        dbmod.CalibreDB._init = previous_init
+        tmp.unlink(missing_ok=True)
+
+
 @pytest.mark.unit
 def test_session_is_materialised_not_none(unreadable_calibre_schema):
     """The precondition that removed the old accidental cushion.