Search: Adds re-try mechanism while re-indexing

src/onegov/search/indexer.py

@@ -79,9 +79,9 @@ def index(
         In case of a bunch of tasks we are assuming they are all from the
         same schema and table in order to optimize the indexing process.
 
-        When a session is passed we use that session's transaction context
-        and use a savepoint instead of our own transaction to perform the
-        action.
+        The upsert is executed directly on the given session's transaction,
+        without a savepoint of its own, so the caller is responsible for
+        committing or rolling back.
 
         :param tasks: A list of tasks to index
         :param session: Supply an active session

src/onegov/search/integration.py

@@ -15,6 +15,7 @@
     searchable_sqlalchemy_models,
 )
 from sqlalchemy import text
+from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm import undefer
 
 
@@ -27,6 +28,10 @@
     from sqlalchemy.orm import DeclarativeBase, Session
 
 
+#: retries per model on a transient reindex conflict
+REINDEX_MAX_ATTEMPTS = 3
+
+
 class SearchApp(morepath.App):
     """ Provides elasticsearch and postgres integration for
     :class:`onegov.core.framework.Framework` based applications.
@@ -234,36 +239,75 @@ def perform_reindex(self, dispose_session: bool = True) -> None:
         self.fts_indexer.delete_search_index(session)
 
         def reindex_model(model: type[Base]) -> None:
-            """ Load all database objects and index them. """
-            session = self.session()
-            try:
-                query = session.query(model).options(undefer('*'))
-                query = apply_searchable_polymorphic_filter(
-                    query,
-                    model,
-                    order_by_polymorphic_identity=True
-                )
+            """ Load all database objects and index them.
+
+            Since we run under ``SERIALIZABLE`` isolation, a concurrent write
+            on a busy site can make the bulk upsert fail with a transaction
+            rollback error (e.g. "could not serialize access due to concurrent
+            update"). Such conflicts are transient, so we retry the whole
+            model, mirroring how normal requests handle conflicts.
+            """
+            for attempt in range(1, REINDEX_MAX_ATTEMPTS + 1):
+                session = self.session()
+                try:
+                    query = session.query(model).options(undefer('*'))
+                    query = apply_searchable_polymorphic_filter(
+                        query, model, order_by_polymorphic_identity=True
+                    )
 
-                # NOTE: we bypass the normal transaction machinery for speed
-                self.fts_indexer.process((
-                    task
-                    for obj in query
+                    # we bypass the normal transaction machinery for speed
+                    self.fts_indexer.process(
+                        (
+                            task
+                            for obj in query
+                            if (
+                                task := self.fts_orm_events.index_task(
+                                    schema, obj
+                                )  # type: ignore[arg-type]
+                            )
+                            is not None
+                        ),
+                        session,
+                    )
+                    session.execute(text('COMMIT'))
+                    return
+
+                except OperationalError as e:
+                    # Error Class 40 (transaction rollback, e.g. serialization
+                    # failure or deadlock) is transient, so retry the model.
+                    orig = getattr(e, 'orig', None)
+                    sqlstate = getattr(orig, 'sqlstate', None)
                     if (
-                        task := self.fts_orm_events.index_task(schema, obj)  # type: ignore[arg-type]
-                    ) is not None
-                ), session)
-                session.execute(text('COMMIT'))
-
-            except Exception:
-                index_log.info(
-                    f"Error indexing model '{model.__name__}' "
-                    f"in schema {schema}",
-                    exc_info=True
-                )
-            finally:
-                session.invalidate()
-                if session.bind and hasattr(session.bind, 'dispose'):
-                    session.bind.dispose()
+                        sqlstate
+                        and sqlstate.startswith('40')
+                        and attempt < REINDEX_MAX_ATTEMPTS
+                    ):
+                        index_log.info(
+                            f"Conflict while indexing model '{model.__name__}'"
+                            f' in schema {schema}, retrying '
+                            f'(attempt {attempt}/{REINDEX_MAX_ATTEMPTS})'
+                        )
+                        continue
+
+                    index_log.error(
+                        f"Error indexing model '{model.__name__}' "
+                        f'in schema {schema}',
+                        exc_info=True,
+                    )
+                    return
+
+                except Exception:
+                    index_log.error(
+                        f"Error indexing model '{model.__name__}' "
+                        f'in schema {schema}',
+                        exc_info=True,
+                    )
+                    return
+
+                finally:
+                    session.invalidate()
+                    if session.bind and hasattr(session.bind, 'dispose'):
+                        session.bind.dispose()
 
         with ThreadPoolExecutor() as executor:
             executor.map(reindex_model, self.indexable_base_models())

tests/onegov/search/test_integration.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import logging
 import math
 import morepath
 import sedate
+import threading
 import transaction
 
 from datetime import timedelta
@@ -12,7 +14,8 @@
 from onegov.core.utils import scan_morepath_modules
 from onegov.search import ORMSearchable, SearchApp, SearchIndex
 from onegov.search.datamanager import IndexerDataManager
-from sqlalchemy import func
+from onegov.search.integration import REINDEX_MAX_ATTEMPTS
+from sqlalchemy import func, text
 from sqlalchemy.orm import mapped_column, registry, DeclarativeBase, Mapped
 from webtest import TestApp as Client
 
@@ -313,6 +316,206 @@ def fts_suggestion(self) -> str:
     assert search.scalar() == 2
 
 
+def _reindex_conflict_app(postgres_dsn: str) -> tuple[Any, Any]:
+    """ Builds a search app with two already-indexed Documents, ready to
+    exercise the reindex conflict handling. Returns the app and the
+    ``Document`` model. """
+
+    class Base(DeclarativeBase):
+        registry = registry()
+
+    class App(Framework, SearchApp):
+        pass
+
+    @App.setting(section='i18n', name='locales')
+    def locales() -> set[str]:
+        return {'en', 'de'}
+
+    @App.setting(section='i18n', name='default_locale')
+    def default_locale() -> str:
+        return 'en'
+
+    class Document(Base, ORMSearchable):
+        __tablename__ = 'documents'
+
+        id: Mapped[int] = mapped_column(primary_key=True)
+        title: Mapped[str]
+
+        fts_public = True
+        fts_title_property = 'title'
+        fts_properties = {
+            'title': {'type': 'localized', 'weight': 'A'}
+        }
+
+        @property
+        def fts_suggestion(self) -> str:
+            return self.title
+
+    scan_morepath_modules(App)
+    morepath.commit(App)
+
+    app = App()
+    app.namespace = 'documents'
+    app.configure_application(
+        dsn=postgres_dsn,
+        base=Base,
+        enable_search=True
+    )
+    app.session_manager.bases[1] = RealBase
+
+    app.set_application_id('documents/home')
+    assert app.fts_search_enabled
+
+    session = app.session()
+    session.add(Document(id=1, title='1'))
+    session.add(Document(id=2, title='2'))
+    transaction.commit()
+
+    return app, Document
+
+
+def _upsert_document_concurrently(
+    app: Any,
+    index: Any,
+    schema: str,
+    document_cls: Any,
+    doc_id: int,
+) -> None:
+    """ Upserts a single document into ``search_index`` from an independent
+    transaction and commits it.
+
+    Our sessions run under ``SERIALIZABLE`` isolation, so doing this while a
+    reindex transaction is already open (its snapshot taken) makes that
+    reindex's own upsert of the same row fail with a genuine serialization
+    error -- the exact transient conflict the retry mechanism must survive.
+
+    This runs on its own thread so it gets a session and connection that are
+    independent from the reindex transaction, since our sessions are scoped
+    per thread.
+    """
+    box: dict[str, BaseException] = {}
+
+    def run() -> None:
+        try:
+            session = app.session()
+            document = session.query(document_cls).filter_by(id=doc_id).one()
+            task = app.fts_orm_events.index_task(schema, document)
+            # bypass the patched indexer and perform a real upsert + commit
+            index([task], session)
+            session.execute(text('COMMIT'))
+            session.invalidate()
+        except BaseException as exception:  # noqa: BLE001
+            box['error'] = exception
+
+    thread = threading.Thread(target=run)
+    thread.start()
+    thread.join()
+    if 'error' in box:
+        raise box['error']
+
+
+def test_reindex_recovers_from_real_conflict(
+    postgres_dsn: str,
+    caplog: Any
+) -> None:
+    """ Triggers a genuine serialization failure (SQLSTATE 40001) during the
+    reindex and asserts that the retry actually recovers -- i.e. that the
+    aborted connection is discarded and the retry runs against a fresh one. """
+
+    app, Document = _reindex_conflict_app(postgres_dsn)
+    schema = app.schema
+    original_index = app.fts_indexer.index
+    attempts = {'n': 0}
+
+    def conflicting_index(tasks: Any, session: Any) -> bool:
+        attempts['n'] += 1
+        if attempts['n'] == 1:
+            # the reindex transaction's snapshot has already been taken by the
+            # query that produced these tasks; committing a concurrent upsert
+            # of the same row now makes the upsert below raise a real 40001
+            _upsert_document_concurrently(
+                app, original_index, schema, Document, 1
+            )
+        return original_index(tasks, session)
+
+    app.fts_indexer.index = conflicting_index  # type: ignore[method-assign]
+
+    with caplog.at_level(logging.INFO, logger='onegov.search.index'):
+        app.perform_reindex()
+
+    # the first attempt hit a real conflict, the retry ran and succeeded
+    assert attempts['n'] == 2
+
+    session = app.session()
+    assert session.query(func.count(SearchIndex.id)).scalar() == 2
+    owner_ids = {
+        r.owner_id_int for r in session.query(SearchIndex.owner_id_int)
+    }
+    assert owner_ids == {1, 2}
+
+    # exactly one retry was logged at INFO and, since the retry succeeded,
+    # nothing was logged at ERROR
+    retry_logs = [
+        r for r in caplog.records
+        if (
+            r.levelno == logging.INFO
+            and "Conflict while indexing model 'Document' in schema "
+                "documents-home, retrying"
+            in r.getMessage()
+        )
+    ]
+    assert len(retry_logs) == 1
+    error_logs = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_logs == []
+
+
+def test_reindex_gives_up_on_persistent_conflict(
+    postgres_dsn: str,
+    caplog: Any
+) -> None:
+    """ When the serialization failure keeps happening on every attempt, the
+    reindex must give up after ``REINDEX_MAX_ATTEMPTS`` and surface an ERROR
+    instead of retrying forever. """
+
+    app, Document = _reindex_conflict_app(postgres_dsn)
+    schema = app.schema
+    original_index = app.fts_indexer.index
+    attempts = {'n': 0}
+
+    def conflicting_index(tasks: Any, session: Any) -> bool:
+        attempts['n'] += 1
+        # force a genuine conflict on every attempt, including the last one
+        _upsert_document_concurrently(
+            app, original_index, schema, Document, 1
+        )
+        return original_index(tasks, session)
+
+    app.fts_indexer.index = conflicting_index  # type: ignore[method-assign]
+
+    with caplog.at_level(logging.INFO, logger='onegov.search.index'):
+        app.perform_reindex()
+
+    # every attempt was made and then the model was given up on
+    assert attempts['n'] == REINDEX_MAX_ATTEMPTS
+
+    # the reindex never managed to commit its own upsert, so the document that
+    # only it writes (id=2) is missing from the index
+    session = app.session()
+    indexed = session.query(SearchIndex).filter_by(owner_id_int=2).count()
+    assert indexed == 0
+
+    # giving up must surface as an ERROR, not be swallowed silently
+    error_logs = [
+        r for r in caplog.records
+        if (
+            r.levelno == logging.ERROR
+            and "Error indexing model 'Document' in schema documents-home"
+            in r.getMessage()
+        )
+    ]
+    assert len(error_logs) == 1
+
+
 def test_orm_integration(
     postgres_dsn: str,
     redis_url: str