diff --git a/src/onegov/search/indexer.py b/src/onegov/search/indexer.py index 94bc0ba7a7..dd4fd98db4 100644 --- a/src/onegov/search/indexer.py +++ b/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 diff --git a/src/onegov/search/integration.py b/src/onegov/search/integration.py index 5bb710a645..1a38ca5d2c 100644 --- a/src/onegov/search/integration.py +++ b/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,32 +239,76 @@ 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. """ + """ 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. + """ session = self.session() try: - query = session.query(model).options(undefer('*')) - query = apply_searchable_polymorphic_filter( - query, - model, - order_by_polymorphic_identity=True - ) + for attempt in range(1, REINDEX_MAX_ATTEMPTS + 1): + try: + query = session.query(model).options(undefer('*')) + query = apply_searchable_polymorphic_filter( + query, model, order_by_polymorphic_identity=True + ) + + # 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')) + break + + 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 ( + sqlstate + and sqlstate.startswith('40') + and attempt < REINDEX_MAX_ATTEMPTS + ): + index_log.info( + f'Conflict while indexing model ' + f"'{model.__name__}' in schema {schema}, " + f'retrying ' + f'(attempt {attempt}/{REINDEX_MAX_ATTEMPTS})' + ) + session.execute(text('ROLLBACK')) + continue + + index_log.error( + f"Error indexing model '{model.__name__}' " + f'in schema {schema}', + exc_info=True, + ) + break + + except Exception: + index_log.error( + f"Error indexing model '{model.__name__}' " + f'in schema {schema}', + exc_info=True, + ) + break - # NOTE: 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')) - - 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'): diff --git a/tests/onegov/search/test_integration.py b/tests/onegov/search/test_integration.py index 1a22fd08ab..bf453ece3f 100644 --- a/tests/onegov/search/test_integration.py +++ b/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,207 @@ 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 transaction is rolled back so the retry runs in a fresh one on the + same connection. """ + + 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 + + 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 + + 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