fix(watson): only intermediate-flush the global search context singleton

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

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()