From 9668caae7f3f58889c5c3a759dcf44117a8739ed Mon Sep 17 00:00:00 2001 From: falscherwiener1-svg Date: Mon, 20 Jul 2026 12:20:33 +0200 Subject: [PATCH 1/2] feat: retry transient item failures in bulk ingestion Add per-item retry logic for transient HTTP status codes (429, 503) in the bulk ingestion sink. Previously, individual items that failed with transient errors inside a successful bulk response were sent directly to the ErrorMonitor without retry, which could trigger a premature TooManyErrors abort on a temporarily overloaded cluster. Changes: - Add TRANSIENT_STATUS_CODES constant (429, 503) - Add _determine_action() helper to extract the bulk action key from a response item - Add _extract_transient_failed_operations() to collect retryable items from a bulk response - Add _retry_transient_item_failures() with exponential backoff that retries only the failed items - Rewire _batch_bulk to call _retry_transient_item_failures before _process_bulk_response, so the ErrorMonitor only tracks failures after all retry attempts are exhausted - Remove the now-unused retryable import and the dead _bulk_api_call function that was never invoked Closes #4002 --- app/connectors_service/connectors/es/sink.py | 129 ++++++++++- app/connectors_service/tests/test_sink.py | 229 +++++++++++++++++++ 2 files changed, 347 insertions(+), 11 deletions(-) diff --git a/app/connectors_service/connectors/es/sink.py b/app/connectors_service/connectors/es/sink.py index 54a5d2c88..2882c569a 100644 --- a/app/connectors_service/connectors/es/sink.py +++ b/app/connectors_service/connectors/es/sink.py @@ -61,7 +61,6 @@ MemQueue, aenumerate, get_size, - retryable, sanitize, ) @@ -97,6 +96,10 @@ # Successful results according to the docs: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html#bulk-api-response-body SUCCESSFUL_RESULTS = ("created", "deleted", "updated", "noop") +# Transient HTTP status codes that warrant per-item retries within a bulk response. +# See: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html +TRANSIENT_STATUS_CODES = (429, 503) + # Reuse the same serializer the elasticsearch client uses to send bulk # requests, so the cap check below sees byte-for-byte the wire payload # (incl. its `default()` handling for `datetime`/`UUID`/`Decimal`). @@ -211,14 +214,6 @@ def _bulk_op(self, doc, operation=OP_INDEX): @tracer.start_as_current_span("_bulk API call", slow_log=1.0) async def _batch_bulk(self, operations, stats): - # TODO: make this retry policy work with unified retry strategy - @retryable(retries=self.max_retires, interval=self.retry_interval) - async def _bulk_api_call(): - return await self.client.client.bulk( - operations=operations, pipeline=self.pipeline["name"] - ) - - # TODO: treat result to retry errors like in async_streaming_bulk task_num = len(self.bulk_tasks) if self._logger.isEnabledFor(logging.DEBUG): @@ -226,8 +221,13 @@ async def _bulk_api_call(): f"Task {task_num} - Sending a batch of {len(operations)} ops -- {get_mib_size(operations)}MiB" ) - # TODO: retry 429s for individual items here - res = await self.client.bulk_insert(operations, self.pipeline["name"]) + # Perform initial bulk insert and retry any items that failed with + # transient errors (429, 503) before passing results to the ErrorMonitor. + # This ensures the ErrorMonitor only sees failures after all retry + # attempts are exhausted, preventing premature TooManyErrors aborts. + res = await self._retry_transient_item_failures( + operations, self.pipeline["name"] + ) ids_to_ops = self._map_id_to_op(operations) # `_process_bulk_response` can raise mid-response, so log failures and # populate stats in `finally` to still count already-accepted docs. @@ -264,6 +264,113 @@ def _map_id_to_op(self, operations): result[doc["_id"]] = op return result + def _determine_action(self, item): + """Return the bulk action key (e.g. OP_INDEX) for a response item, + or None if the action is unrecognized.""" + for action in (OP_INDEX, OP_DELETE, OP_CREATE, OP_UPDATE): + if action in item: + return action + return None + + def _extract_transient_failed_operations(self, res, operations): + """Extract operations that failed with transient HTTP status codes. + + Given a bulk response and the original operations list, returns a + new operations list containing only those items that received a + transient error (429, 503). These are candidates for retry. + + Args: + res: The bulk API response dict. + operations: The flat list of operation dicts that were sent + (e.g. [{"index": {...}}, {"doc": {...}}, ...]). + + Returns: + A flat list of operation dicts for the failed items only. + """ + failed_ops = [] + # Build a lookup: doc_id -> index in the operations list + # Operations come in pairs: [{action: meta}, body] for index/update, + # or single entries for delete. + ops_by_id = {} + i = 0 + while i < len(operations): + entry = operations[i] + if isinstance(entry, dict) and len(entry) == 1: + action = next(iter(entry)) + meta = entry[action] + if isinstance(meta, dict) and "_id" in meta: + doc_id = meta["_id"] + # Grab body entry too if present (index/update have one) + if i + 1 < len(operations) and action != OP_DELETE: + ops_by_id[doc_id] = operations[i : i + 2] + i += 2 + continue + else: + ops_by_id[doc_id] = operations[i : i + 1] + i += 1 + continue + i += 1 + + for item in res.get("items", []): + action = self._determine_action(item) + if action is None: + continue + data = item[action] + status = data.get("status", 200) + if status in TRANSIENT_STATUS_CODES: + doc_id = data.get("_id") + if doc_id in ops_by_id: + failed_ops.extend(ops_by_id[doc_id]) + + return failed_ops + + async def _retry_transient_item_failures(self, operations, pipeline_name): + """Retry operations that failed with transient errors. + + After an initial bulk call, some items may fail individually with + transient HTTP status codes (429, 503). This method collects those + items and retries them with exponential backoff, optionally + respecting a Retry-After header from the response. + + After all retry attempts are exhausted, the final bulk response is + returned so that remaining failures can be passed to the + ErrorMonitor for tracking and potential abort. + + Args: + operations: The original flat list of operation dicts. + pipeline_name: The ingest pipeline name to pass to bulk_insert. + + Returns: + The final bulk response dict after all retries. + """ + res = await self.client.bulk_insert(operations, pipeline_name) + + for attempt in range(1, self.max_retires + 1): + failed_ops = self._extract_transient_failed_operations(res, operations) + if not failed_ops: + break + + self._logger.warning( + f"Retrying {len(failed_ops)} transiently failed items " + f"(attempt {attempt}/{self.max_retires})" + ) + + # Exponential backoff: 1x, 2x, 4x, ... the base retry interval. + # Note: the Retry-After header would be the preferred delay for + # 429s, but bulk_insert returns a parsed body without response + # headers, so we rely on backoff for now. A future enhancement + # could plumb the Retry-After value through the ES client layer. + sleep_time = self.retry_interval * (2 ** (attempt - 1)) + await asyncio.sleep(sleep_time) + + res = await self.client.bulk_insert(failed_ops, pipeline_name) + + # Narrow the operations window to only the still-failing items + # so the next extraction pass operates on the correct subset. + operations = failed_ops + + return res + async def _process_bulk_response(self, res, ids_to_ops, do_log=False): for item in res.get("items", []): if OP_INDEX in item: diff --git a/app/connectors_service/tests/test_sink.py b/app/connectors_service/tests/test_sink.py index bca8138ab..751396096 100644 --- a/app/connectors_service/tests/test_sink.py +++ b/app/connectors_service/tests/test_sink.py @@ -30,7 +30,9 @@ OP_DELETE, OP_INDEX, OP_UPDATE, + RESULT_ERROR, RESULT_SUCCESS, + TRANSIENT_STATUS_CODES, UPDATES_QUEUED, AsyncBulkRunningError, ElasticsearchOverloadedError, @@ -2289,3 +2291,230 @@ async def test_sink_dispatches_oversized_single_doc_alone(): assert in_loop_names == ["Elasticsearch Sink: _bulk batch #1"] assert [_doc_ids(b) for b in _dispatched_batches(sink)] == [["big"]] + + +@pytest.mark.asyncio +async def test_extract_transient_failed_operations_returns_empty_when_all_successful(): + client = Mock() + sink = Sink( + client=client, + queue=None, + error_monitor=Mock(), + chunk_size=0, + pipeline={"name": "pipeline"}, + chunk_mem_size=0, + max_concurrency=0, + max_retries=3, + retry_interval=10, + ) + + operations = [{"index": {"_index": "test", "_id": "1"}}, {"data": "value"}] + res = { + "items": [ + {"index": {"_id": "1", "status": 201, "result": "created"}}, + ] + } + + failed = sink._extract_transient_failed_operations(res, operations) + assert failed == [] + + +@pytest.mark.asyncio +async def test_extract_transient_failed_operations_returns_429_items(): + client = Mock() + sink = Sink( + client=client, + queue=None, + error_monitor=Mock(), + chunk_size=0, + pipeline={"name": "pipeline"}, + chunk_mem_size=0, + max_concurrency=0, + max_retries=3, + retry_interval=10, + ) + + operations = [ + {"index": {"_index": "test", "_id": "1"}}, + {"data": "doc1"}, + {"index": {"_index": "test", "_id": "2"}}, + {"data": "doc2"}, + {"delete": {"_index": "test", "_id": "3"}}, + ] + res = { + "items": [ + {"index": {"_id": "1", "status": 200, "result": "created"}}, + {"index": {"_id": "2", "status": 429, "error": {"type": "too_many_requests"}}}, + {"delete": {"_id": "3", "status": 429, "error": {"type": "too_many_requests"}}}, + ] + } + + failed = sink._extract_transient_failed_operations(res, operations) + # doc 2 (index, 2 ops) and doc 3 (delete, 1 op) + assert len(failed) == 3 + # Verify doc 2's index operation is included + assert any( + isinstance(op, dict) and op.get("index", {}).get("_id") == "2" + for op in failed + if isinstance(op, dict) and "index" in op + ) + # Verify doc 3's delete operation is included + assert any( + isinstance(op, dict) and op.get("delete", {}).get("_id") == "3" + for op in failed + if isinstance(op, dict) and "delete" in op + ) + + +@pytest.mark.asyncio +async def test_retry_transient_item_failures_retries_429_items(): + config = { + "username": "elastic", + "password": "changeme", + "host": "http://nowhere.com:9200", + } + client = ESManagementClient(config) + client.client = AsyncMock() + sink = Sink( + client=client, + queue=None, + error_monitor=Mock(), + chunk_size=0, + pipeline={"name": "pipeline"}, + chunk_mem_size=0, + max_concurrency=0, + max_retries=3, + retry_interval=0.01, + ) + + operations = [ + {"index": {"_index": "test", "_id": "1"}}, + {"data": "doc1"}, + {"index": {"_index": "test", "_id": "2"}}, + {"data": "doc2"}, + ] + + # First call: doc 1 succeeds, doc 2 gets 429 + first_response = { + "errors": True, + "items": [ + {"index": {"_id": "1", "status": 201, "result": "created"}}, + {"index": {"_id": "2", "status": 429, "error": {"type": "too_many_requests"}}}, + ], + } + + # Second call: doc 2 succeeds on retry + second_response = { + "items": [ + {"index": {"_id": "2", "status": 201, "result": "created"}}, + ] + } + + client.bulk_insert = AsyncMock( + side_effect=[first_response, second_response] + ) + + with mock.patch.object(asyncio, "sleep"): + final_res = await sink._retry_transient_item_failures(operations, "pipeline") + + assert client.bulk_insert.await_count == 2 + # Final response should be the successful retry of doc 2 + assert final_res == second_response + + +@pytest.mark.asyncio +async def test_retry_transient_item_failures_gives_up_after_exhaustion(): + config = { + "username": "elastic", + "password": "changeme", + "host": "http://nowhere.com:9200", + } + client = ESManagementClient(config) + client.client = AsyncMock() + sink = Sink( + client=client, + queue=None, + error_monitor=Mock(), + chunk_size=0, + pipeline={"name": "pipeline"}, + chunk_mem_size=0, + max_concurrency=0, + max_retries=2, + retry_interval=0.01, + ) + + operations = [ + {"index": {"_index": "test", "_id": "1"}}, + {"data": "doc1"}, + ] + + failed_response = { + "errors": True, + "items": [ + {"index": {"_id": "1", "status": 429, "error": {"type": "too_many_requests"}}}, + ], + } + + # All calls return 429 + client.bulk_insert = AsyncMock(return_value=failed_response) + + with mock.patch.object(asyncio, "sleep"): + final_res = await sink._retry_transient_item_failures(operations, "pipeline") + + # Initial call + 2 retries = 3 total + assert client.bulk_insert.await_count == 3 + # Final response still contains the 429 failure + assert final_res["items"][0]["index"]["status"] == 429 + + +@pytest.mark.asyncio +async def test_batch_bulk_with_transient_item_retries_integration(): + """Verify that _batch_bulk retries transient item failures before + passing results to the ErrorMonitor.""" + config = { + "username": "elastic", + "password": "changeme", + "host": "http://nowhere.com:9200", + } + client = ESManagementClient(config) + client.client = AsyncMock() + error_monitor = Mock() + sink = Sink( + client=client, + queue=None, + error_monitor=error_monitor, + chunk_size=0, + pipeline={"name": "pipeline"}, + chunk_mem_size=0, + max_concurrency=0, + max_retries=2, + retry_interval=0.01, + ) + + operations = [ + {"index": {"_index": "test", "_id": "1"}}, + {"data": "doc1"}, + ] + + first_response = { + "errors": True, + "items": [ + {"index": {"_id": "1", "status": 429, "error": {"type": "too_many_requests"}}}, + ], + } + second_response = { + "items": [ + {"index": {"_id": "1", "status": 201, "result": "created"}}, + ] + } + + client.bulk_insert = AsyncMock(side_effect=[first_response, second_response]) + + stats = {OP_INDEX: {"1": 100}, OP_UPDATE: {}, OP_DELETE: {}} + + with mock.patch.object(asyncio, "sleep"): + await sink._batch_bulk(operations, stats) + + # After retry succeeds, error_monitor should see a success, not an error + error_monitor.track_success.assert_called_once() + error_monitor.track_error.assert_not_called() From 781d1ed6e25cd6710cdf39c621521a1593b7e518 Mon Sep 17 00:00:00 2001 From: falscherwiener1-svg Date: Wed, 22 Jul 2026 15:03:57 +0200 Subject: [PATCH 2/2] fix(sharepoint): scope list_item _id by site_id across site collections (#4222) SharePoint list GUIDs are only unique within a site collection. When two site collections share a list GUID (e.g. via site templates or migration), list_item documents from different sites collide on the same _id and silently overwrite each other. Include site_id in the _id construction to guarantee global uniqueness. --- .../sources/sharepoint/sharepoint_online/datasource.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/app/connectors_service/connectors/sources/sharepoint/sharepoint_online/datasource.py b/app/connectors_service/connectors/sources/sharepoint/sharepoint_online/datasource.py index a133f8716..9ffccdc55 100644 --- a/app/connectors_service/connectors/sources/sharepoint/sharepoint_online/datasource.py +++ b/app/connectors_service/connectors/sources/sharepoint/sharepoint_online/datasource.py @@ -953,12 +953,12 @@ async def site_list_items( and list_item["lastModifiedDateTime"] >= self.last_sync_time() ): # List Item IDs are unique within list. - # Therefore we mix in site_list id to it to make sure they are - # globally unique. + # Therefore we mix in site_list id and site id to make sure they are + # globally unique across site collections. # Also we need to remember original ID because when a document # is yielded, its "id" field is overwritten with content of "_id" field list_item_natural_id = list_item["id"] - list_item["_id"] = f"{site_list_id}-{list_item['id']}" + list_item["_id"] = f"{site_id}-{site_list_id}-{list_item['id']}" list_item["object_type"] = "list_item" content_type = list_item["contentType"]["name"]