fix(search): limit documents in bulk load chunk (#18441)

Author: miketheman

tests/unit/search/test_tasks.py

@@ -243,7 +243,9 @@ def project_docs(db):
         class TestError(Exception):
             pass
 
-        def parallel_bulk(client, iterable, index=None, max_chunk_bytes=None):
+        def parallel_bulk(
+            client, iterable, index=None, chunk_size=None, max_chunk_bytes=None
+        ):
             assert client is es_client
             assert iterable is docs
             assert index == "warehouse-cbcbcbcbcb"
@@ -304,7 +306,7 @@ def project_docs(db):
         monkeypatch.setattr(warehouse.search.tasks, "SearchLock", NotLock)
 
         parallel_bulk = pretend.call_recorder(
-            lambda client, iterable, index, max_chunk_bytes: [None]
+            lambda client, iterable, index, chunk_size, max_chunk_bytes: [None]
         )
         monkeypatch.setattr(warehouse.search.tasks, "parallel_bulk", parallel_bulk)
 
@@ -314,7 +316,11 @@ def project_docs(db):
 
         assert parallel_bulk.calls == [
             pretend.call(
-                es_client, docs, index="warehouse-cbcbcbcbcb", max_chunk_bytes=10485760
+                es_client,
+                docs,
+                index="warehouse-cbcbcbcbcb",
+                chunk_size=100,
+                max_chunk_bytes=10485760,
             )
         ]
         assert es_client.indices.create.calls == [
@@ -372,7 +378,7 @@ def project_docs(db):
         monkeypatch.setattr(warehouse.search.tasks, "SearchLock", NotLock)
 
         parallel_bulk = pretend.call_recorder(
-            lambda client, iterable, index, max_chunk_bytes: [None]
+            lambda client, iterable, index, chunk_size, max_chunk_bytes: [None]
         )
         monkeypatch.setattr(warehouse.search.tasks, "parallel_bulk", parallel_bulk)
 
@@ -382,7 +388,11 @@ def project_docs(db):
 
         assert parallel_bulk.calls == [
             pretend.call(
-                es_client, docs, index="warehouse-cbcbcbcbcb", max_chunk_bytes=10485760
+                es_client,
+                docs,
+                index="warehouse-cbcbcbcbcb",
+                chunk_size=100,
+                max_chunk_bytes=10485760,
             )
         ]
         assert es_client.indices.create.calls == [
@@ -441,7 +451,7 @@ def project_docs(db):
         monkeypatch.setattr(warehouse.search.tasks, "SearchLock", NotLock)
 
         parallel_bulk = pretend.call_recorder(
-            lambda client, iterable, index, max_chunk_bytes: [None]
+            lambda client, iterable, index, chunk_size, max_chunk_bytes: [None]
         )
         monkeypatch.setattr(warehouse.search.tasks, "parallel_bulk", parallel_bulk)
 
@@ -466,7 +476,11 @@ def project_docs(db):
 
         assert parallel_bulk.calls == [
             pretend.call(
-                es_client, docs, index="warehouse-cbcbcbcbcb", max_chunk_bytes=10485760
+                es_client,
+                docs,
+                index="warehouse-cbcbcbcbcb",
+                chunk_size=100,
+                max_chunk_bytes=10485760,
             )
         ]
         assert es_client.indices.create.calls == [

warehouse/search/tasks.py

@@ -168,6 +168,7 @@ def reindex(self, request):
                     client,
                     _project_docs(request.db),
                     index=new_index_name,
+                    chunk_size=100,
                     max_chunk_bytes=10 * 1024 * 1024,  # 10MB, per OpenSearch defaults
                 ):
                     pass