From ad183d1dc826b7ef6bbdcfc57a9f63eefc0cb035 Mon Sep 17 00:00:00 2001 From: Valentijn Scholten Date: Wed, 8 Jul 2026 13:12:47 +0200 Subject: [PATCH] fix(watson): only intermediate-flush the global search context singleton The intermediate-flush hook patches SearchContextManager.add_to_context at class level, so it also fires inside update_watson_search_index_for_model's own ad-hoc context. A task processing a full WATSON_ASYNC_INDEX_UPDATE_BATCH_SIZE batch re-dispatches a clone of itself, discards those pks unindexed, and loops forever: queue length stays ~0 while the worker is pegged republishing the same batches, and nothing gets indexed. Guard the hook so only the global singleton (the request-path context managed by AsyncSearchContextMiddleware) intermediate-flushes. Ad-hoc context managers index their own batch on end(), as stock watson does. Extracted from the hardening half of #15165 for the bugfix branch. Co-authored-by: dogboat --- dojo/middleware.py | 9 +++++++++ unittests/test_watson_intermediate_flush.py | 21 ++++++++++++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) 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()