Merge pull request #3756 from freelawproject/introduce-retries-for-update-by-query-requests-on-timeouts

Author: mlissner

cl/search/tasks.py

@@ -610,7 +610,7 @@ def get_doc_from_es(
 
 def handle_ubq_retries(
     self: Task,
-    exc: ConnectionError | ConflictError,
+    exc: ConnectionError | ConflictError | ConnectionTimeout,
     count_query=QuerySet | None,
 ) -> None:
     """Handles the retry logic for update_children_docs_by_query task based on
@@ -627,9 +627,9 @@ def handle_ubq_retries(
     if retry_count >= self.max_retries:
         raise exc
 
-    if isinstance(exc, ConnectionError) and count_query:
+    if isinstance(exc, ConnectionError | ConnectionTimeout) and count_query:
         num_documents = count_query.count()
-        estimated_time_ms = num_documents * 15  # 15ms per document
+        estimated_time_ms = num_documents * 90  # 90ms per document
         # Convert ms to seconds
         estimated_delay_sec = round(estimated_time_ms / 1000)
         # Apply exponential backoff with jitter
@@ -693,6 +693,9 @@ def update_children_docs_by_query(
         parent_instance = get_instance_from_db(parent_instance_id, Docket)
         if not parent_instance:
             return
+        count_query = RECAPDocument.objects.filter(
+            docket_entry__docket_id=parent_instance_id
+        )
     elif (
         es_document is OpinionDocument or es_document is OpinionClusterDocument
     ):
@@ -704,10 +707,7 @@ def update_children_docs_by_query(
         )
         if not parent_instance:
             return
-
-        count_query = RECAPDocument.objects.filter(
-            docket_entry__docket_id=parent_instance_id
-        )
+        count_query = Opinion.objects.filter(cluster_id=parent_instance_id)
 
     if not main_doc:
         # Abort bulk update for a not supported document or non-existing parent
@@ -744,7 +744,7 @@ def update_children_docs_by_query(
     ubq = ubq.script(source=script_source, params=params)
     try:
         ubq.execute()
-    except (ConnectionError, ConflictError) as exc:
+    except (ConnectionError, ConflictError, ConnectionTimeout) as exc:
         handle_ubq_retries(self, exc, count_query=count_query)
 
     if settings.ELASTICSEARCH_DSL_AUTO_REFRESH:

cl/settings/third_party/elasticsearch.py

@@ -57,7 +57,7 @@
     "ELASTICSEARCH_CA_CERT",
     default="/opt/courtlistener/docker/elastic/ca.crt",
 )
-ELASTICSEARCH_TIMEOUT = env("ELASTICSEARCH_TIMEOUT", default=200)
+ELASTICSEARCH_TIMEOUT = env("ELASTICSEARCH_TIMEOUT", default=3500)
 
 base_connection_params = {
     "hosts": ELASTICSEARCH_DSL_HOST,