From 515b8ad13cceb0e696774e4f7698dafef7c1597d Mon Sep 17 00:00:00 2001 From: Tarek Lokal Date: Sat, 11 Jul 2026 18:03:00 +0200 Subject: [PATCH] fix(search): upsert ES updates and retry failed bulk indexes (#2523) Use doc_as_upsert for update operations and retry index_and_notify bulk failures per record to reduce "failed to index" Sentry noise. Co-authored-by: Cursor --- api/management/commands/index_and_notify.py | 25 +++++++++++- utils/elasticsearch.py | 43 +++++++++++++++++---- 2 files changed, 59 insertions(+), 9 deletions(-) diff --git a/api/management/commands/index_and_notify.py b/api/management/commands/index_and_notify.py index f7349c9b9..4e1e66b90 100644 --- a/api/management/commands/index_and_notify.py +++ b/api/management/commands/index_and_notify.py @@ -971,18 +971,41 @@ def index_records(self, records, to_create=True): self.bulk([construct_es_data(record, is_create=to_create) for record in list(records)]) def bulk(self, actions): + if not actions: + return try: created, errors = bulk(client=ES_CLIENT, actions=actions) if errors: logger.error( "(index_and_notify:bulk): produced errors:", - extra=logger_context(dict(errors=errors)), + extra=logger_context(dict(errors=errors[:10], error_count=len(errors))), ) + # Retry failed updates individually with upsert (#2523) + for action in actions: + if action.get("_op_type") != "update": + continue + try: + bulk(client=ES_CLIENT, actions=[{**action, "doc_as_upsert": True}]) + except Exception: + logger.error( + "Failed to index record id=%s after bulk retry", + action.get("_id"), + exc_info=True, + ) except Exception: logger.error( "(index_and_notify:bulk): could not index records", exc_info=True, ) + for action in actions: + try: + bulk(client=ES_CLIENT, actions=[action]) + except Exception: + logger.error( + "Failed to index record id=%s in fallback bulk", + action.get("_id"), + exc_info=True, + ) # Remove items in a queryset where updated_at == created_at. # This leaves us with only ones that have been modified. diff --git a/utils/elasticsearch.py b/utils/elasticsearch.py index 4b37135bc..5f9a1b794 100644 --- a/utils/elasticsearch.py +++ b/utils/elasticsearch.py @@ -42,24 +42,51 @@ def construct_es_data(instance, is_create=False): metadata.update(**data) else: metadata["doc"] = data + # Create the document when an update targets a missing ES record (#2523) + metadata["doc_as_upsert"] = True return metadata +def _log_bulk_errors(errors, context=""): + if not errors: + return + logger.error( + "Elasticsearch bulk indexing produced errors%s: %s", + f" ({context})" if context else "", + errors[:5], + extra={"error_count": len(errors)}, + ) + + def create_es_index(instance): """Creates an Elasticsearch index from the record instance""" if ES_CLIENT and ES_PAGE_NAME: - # To make sure it doesn't run for tests - created, errors = bulk(client=ES_CLIENT, actions=[construct_es_data(instance, True)]) - logger.info(f"Created {created} records") - log_errors(errors) + try: + created, errors = bulk(client=ES_CLIENT, actions=[construct_es_data(instance, True)]) + logger.info(f"Created {created} records for {instance.__class__.__name__} pk={instance.pk}") + _log_bulk_errors(errors, f"create {instance.__class__.__name__}:{instance.pk}") + except Exception: + logger.error( + "Failed to index %s pk=%s", + instance.__class__.__name__, + instance.pk, + exc_info=True, + ) def update_es_index(instance): """Updates the Elasticsearch index from the record instance""" if ES_CLIENT and ES_PAGE_NAME: - # To make sure it doesn't run for tests - updated, errors = bulk(client=ES_CLIENT, actions=[construct_es_data(instance)]) - logger.info(f"Updated {updated} records") - log_errors(errors) + try: + updated, errors = bulk(client=ES_CLIENT, actions=[construct_es_data(instance)]) + logger.info(f"Updated {updated} records for {instance.__class__.__name__} pk={instance.pk}") + _log_bulk_errors(errors, f"update {instance.__class__.__name__}:{instance.pk}") + except Exception: + logger.error( + "Failed to index %s pk=%s", + instance.__class__.__name__, + instance.pk, + exc_info=True, + )