diff --git a/dojo/middleware.py b/dojo/middleware.py index 127ab462a09..e46b499daaa 100644 --- a/dojo/middleware.py +++ b/dojo/middleware.py @@ -312,6 +312,15 @@ def install_intermediate_flush_hook(): def add_to_context_with_flush(self, engine, obj): original_add(self, engine, obj) + # The intermediate flush is a request-path optimization on the global singleton + # context (AsyncSearchContextMiddleware). The async reindex task + # update_watson_search_index_for_model() builds its OWN SearchContextManager and + # IS the drain target -- if it re-drained its batch it would dispatch a clone of + # itself, discard those pks unindexed, and loop forever (queue ~0, worker pegged, + # nothing indexed). Only the singleton intermediate-flushes; any ad-hoc context + # manager indexes its own batch on end(). + if self is not search_context_manager: + return threshold = getattr(settings, "WATSON_ASYNC_INDEX_UPDATE_BATCH_SIZE", 1000) if threshold <= 0 or not self._stack: return diff --git a/unittests/test_watson_intermediate_flush.py b/unittests/test_watson_intermediate_flush.py index 624b516bf77..4e255779f41 100644 --- a/unittests/test_watson_intermediate_flush.py +++ b/unittests/test_watson_intermediate_flush.py @@ -10,7 +10,7 @@ from unittest.mock import patch from django.test import override_settings -from watson.search import search_context_manager +from watson.search import SearchContextManager, search_context_manager from dojo.middleware import ( _drain_search_context_to_async, # noqa: PLC2701 -- internal helper under test @@ -119,3 +119,22 @@ def test_invalidated_context_skips_drain(self): # hook should detect the invalid flag and bail out. search_context_manager.add_to_context(engine_marker, p) drain.assert_not_called() + + @override_settings(WATSON_ASYNC_INDEX_UPDATE_BATCH_SIZE=2) + def test_ad_hoc_context_manager_does_not_drain(self): + """ + The intermediate flush is a request-path optimization on the global singleton. An + ad-hoc SearchContextManager -- e.g. the one update_watson_search_index_for_model builds to + index its own batch -- must NOT drain, or it would dispatch a clone of itself and loop + forever (queue ~0, worker pegged, nothing indexed). + """ + adhoc = SearchContextManager() + adhoc.start() + try: + with patch("dojo.middleware._drain_search_context_to_async") as drain: + for p in self.products[:3]: # past the threshold of 2 + adhoc.add_to_context(object(), p) + drain.assert_not_called() + finally: + adhoc.invalidate() + adhoc.end()