fix(milvus): classify client-caused gRPC errors as client errors (PLU-543)

[paulkarayan]

What & why

A Milvus destination given bad or expired customer credentials failed the job with a platform error, counting a customer misconfiguration against the Job Completions SLO. The fix reclassifies the customer-caused gRPC failures as client errors (UserAuthError -> 401, UserError -> 422) so they stop burning the SLO.

The subtlety that makes this real is how pymilvus surfaces the failure. Verified against the pinned pymilvus==2.6.9 (pymilvus/decorators.py):

• The status codes in decorators.IGNORE_RETRY_CODES — UNAUTHENTICATED, PERMISSION_DENIED, INVALID_ARGUMENT (plus DEADLINE_EXCEEDED/ALREADY_EXISTS/RESOURCE_EXHAUSTED/UNIMPLEMENTED) — are re-raised as the RAW grpc.RpcError via raise e from e (decorators.py:216 and :294). They are never wrapped in a MilvusException. grpc.RpcError carries its status on a .code() method.
• Every other code (e.g. NOT_FOUND, not in IGNORE_RETRY_CODES) is retried and, once the retry budget is exhausted, wrapped into MilvusException(e.code, ...). MilvusException carries its status on a .code property (which for a genuine server-side business failure is an int ErrorCode, not a grpc.StatusCode).

So the entire customer-credential set — crucially UNAUTHENTICATED, the whole motivation for this ticket — reaches the connector as a raw grpc.RpcError, not a MilvusException. Any classifier or except clause that only looks at MilvusException never sees it, and the failure stays platform-classed. The classifier therefore resolves the grpc.StatusCode from both exception types and both call paths.

Linked ticket

PLU-543

Impact

• Users: a customer whose Milvus credentials are bad/expired, whose role lacks permission, whose request is malformed, or who points at a missing collection now gets a client error (UserAuthError -> 401, or UserError -> 422) instead of a platform error. The failure is attributed to them, not to us. Users with valid Milvus config and any genuine server-side (non-client) Milvus failure are unaffected — those still raise the platform error as before.
• Wire contract / clients: for the specific gRPC status codes UNAUTHENTICATED -> UserAuthError, and PERMISSION_DENIED / INVALID_ARGUMENT / NOT_FOUND -> UserError, the exception type raised from the uploader changes. This is wired into every customer-facing Milvus call: precheck(), insert_results(), delete_by_record_id(), and _prepare_data_for_insert() (the describe_collection schema round-trip). The has_collection() == False branch of precheck() — a missing target collection — now raises UserError instead of a platform DestinationConnectionError, since an absent collection is the customer's configuration, not a platform fault. Every other code, and any non-Milvus/non-gRPC failure, still raises DestinationConnectionError / WriteError exactly as before.
• Deployment target considerations: no config, CI-behavior, or deploy-target changes to the runtime. This is connector library code and behaves identically across SaaS / DI / in-VPC / on-prem / SND. (Test-tooling only: pymilvus is added to the test dependency group — see below.) Not exercised on DI.

Risk / rollback

Low. Pure code change, no migration or data touched — revert the PR to back it out. The one subtlety is that precheck() drops its @DestinationConnectionError.wrap decorator (see below); the decorator's catch-all fallback is reproduced inline, so any non-Milvus exception escaping precheck still becomes a DestinationConnectionError.

How it was verified

• Unit tests drive REAL grpc.RpcError. The tests build a grpc.RpcError subclass whose .code() returns the target grpc.StatusCode — exactly the object pymilvus 2.6.9 re-raises — and push it through precheck(), insert_results(), delete_by_record_id(), and _prepare_data_for_insert() via a mocked Milvus client, asserting the result is UserAuthError (401) / UserError (422). A MilvusException-wrapped path (NOT_FOUND) and the sync retry-storm path (pymilvus stores the raw .code method on the MilvusException) are covered too, plus a control that a server-side INTERNAL stays platform. The earlier tests used a synthetic MilvusException(code=UNAUTHENTICATED) — a shape pymilvus never emits for that code — and validated a wrong assumption; they are replaced. 29 tests pass (pytest test/unit/processes/connectors/test_milvus.py).

• CI extra fix. The tests import pymilvus/grpc, but milvus is an optional extra and the unit-test job runs uv sync --group test --locked without --all-extras, so the tests would have errored red in CI. pymilvus is added to the test dependency group (matching how the repo already lists other connectors' test deps there), and uv.lock is regenerated; uv tree --only-group test confirms pymilvus v2.6.9 (group: test), so the job installs it.

• SND reproof on pk-ab-lane-b with the REAL exception. A uploader overlay image was built at this commit (FROM platform-plugins-uploader:3.6.2 + the fixed milvus.py), pushed to ACR, and run against a harness that raises a REAL grpc.RpcError (not a synthetic MilvusException). Baseline (stock 3.6.2) vs fixed (plu543-249b0241):

  case (REAL grpc.RpcError unless noted)	baseline 3.6.2	fixed
  precheck / UNAUTHENTICATED	DestinationConnectionError 400	UserAuthError 401
  precheck / PERMISSION_DENIED	DestinationConnectionError 400	UserError 422
  insert / UNAUTHENTICATED	raw grpc.RpcError escaped uncaught	UserAuthError 401
  insert / INVALID_ARGUMENT	raw grpc.RpcError escaped uncaught	UserError 422
  delete / PERMISSION_DENIED	raw grpc.RpcError escaped uncaught	UserError 422
  insert / NOT_FOUND (MilvusException)	WriteError 400	UserError 422
  insert / INTERNAL (control)	raw grpc.RpcError escaped uncaught	WriteError 400 (stays platform)

  Baseline never yields a client error for the customer-cred codes — precheck's old @wrap flattens it to a platform 400, and insert/delete let the raw grpc.RpcError escape entirely because the old code caught only MilvusException. The fixed image classifies every customer-cred code correctly while the server-side INTERNAL control stays platform 400. The image was applied via ETL_NODE_IMAGE.uploader.default.tag in ConfigMap job-execution-config (the DTPL-468 lever), then the ConfigMap was reverted to the pristine 3.6.2 tag.

• ruff check and format --check clean (repo-pinned ruff 0.15.1) on the changed files.

The @DestinationConnectionError.wrap removal (why it is load-bearing)

_classify_milvus_exception(exc, platform_error) resolves the grpc.StatusCode off either a raw grpc.RpcError or a MilvusException (via _grpc_status_code, which handles the method/property/int/callable variants) and returns a client error for the customer-caused codes, otherwise returns the passed-in platform_error unchanged. A _reclassify_milvus_errors context manager catches (grpc.RpcError, MilvusException) and runs the block through the classifier; it is wired into precheck(), insert_results(), delete_by_record_id(), and _prepare_data_for_insert().

For precheck() the @DestinationConnectionError.wrap class decorator had to go: it re-wraps every escaping exception into a DestinationConnectionError, which would immediately clobber the UserAuthError / UserError this change raises. So precheck now handles its own exceptions: the reclassify context manager handles the Milvus/gRPC failures, an already-classified UnstructuredIngestError (including the missing-collection UserError) re-raises untouched, and the decorator's blanket fallback is reproduced inline (except Exception -> DestinationConnectionError) so nothing that used to become a platform error stops doing so.

---

_{Note for the reviewer: this branch was developed in a dedicated unstructured-ingest-plu543 worktree because the primary unstructured-ingest checkout sits on an unrelated, divergent pk/plu-370 branch. The branch here is cut clean from origin/main.}

[cubic-dev-ai]

2 issues found across 6 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="test/unit/processes/connectors/test_milvus.py">

<violation number="1" location="test/unit/processes/connectors/test_milvus.py:384">
P3: The secondary assertions after `pytest.raises` are logically guaranteed and add no signal. In `test_precheck_reclassifies_raw_grpc_error`, `pytest.raises(expected)` (UserAuthError/UserError) already pins the exact type; the subsequent `assert not isinstance(excinfo.value, DestinationConnectionError)` can never fail because these classes are disjoint siblings. The same redundancy appears in `test_precheck_missing_collection_is_user_error`, and in `test_insert_results_other_grpc_code_stays_write_error` where `assert not isinstance(excinfo.value, UserError)` follows `pytest.raises(WriteError)`. `test_run_data_missing_collection_delete_grpc_not_found_is_user_error`'s trailing `assert isinstance(excinfo.value, UnstructuredIngestError)` is trivially true for any subclass. These extra asserts verify nothing beyond the `pytest.raises` clause; remove them.</violation>
</file>

<file name="unstructured_ingest/processes/connectors/milvus.py">

<violation number="1" location="unstructured_ingest/processes/connectors/milvus.py:131">
P3: `platform_error_factory(exc)` is invoked for every failure, including client-coded ones (UNAUTHENTICATED, PERMISSION_DENIED, INVALID_ARGUMENT, NOT_FOUND) that `_classify_milvus_exception` immediately reclassifies to a UserAuthError/UserError and discards the factory result. The factory (plus a second `safe_error_summary` call on a credential-bearing exception) should only run on the fall-through branch when no client code is resolved.</violation>
</file>

<sub>Shadow auto-approve: would not auto-approve because issues were found.

Re-trigger cubic</sub>

[cubic-dev-ai]

P3: The secondary assertions after pytest.raises are logically guaranteed and add no signal. In test_precheck_reclassifies_raw_grpc_error, pytest.raises(expected) (UserAuthError/UserError) already pins the exact type; the subsequent assert not isinstance(excinfo.value, DestinationConnectionError) can never fail because these classes are disjoint siblings. The same redundancy appears in test_precheck_missing_collection_is_user_error, and in test_insert_results_other_grpc_code_stays_write_error where assert not isinstance(excinfo.value, UserError) follows pytest.raises(WriteError). test_run_data_missing_collection_delete_grpc_not_found_is_user_error's trailing assert isinstance(excinfo.value, UnstructuredIngestError) is trivially true for any subclass. These extra asserts verify nothing beyond the pytest.raises clause; remove them.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At test/unit/processes/connectors/test_milvus.py, line 384:

<comment>The secondary assertions after `pytest.raises` are logically guaranteed and add no signal. In `test_precheck_reclassifies_raw_grpc_error`, `pytest.raises(expected)` (UserAuthError/UserError) already pins the exact type; the subsequent `assert not isinstance(excinfo.value, DestinationConnectionError)` can never fail because these classes are disjoint siblings. The same redundancy appears in `test_precheck_missing_collection_is_user_error`, and in `test_insert_results_other_grpc_code_stays_write_error` where `assert not isinstance(excinfo.value, UserError)` follows `pytest.raises(WriteError)`. `test_run_data_missing_collection_delete_grpc_not_found_is_user_error`'s trailing `assert isinstance(excinfo.value, UnstructuredIngestError)` is trivially true for any subclass. These extra asserts verify nothing beyond the `pytest.raises` clause; remove them.</comment>

<file context>
@@ -154,3 +171,314 @@ def test_run_produces_empty_output_when_no_embeddings(
+        uploader.precheck()
+    # UserAuthError is a UserError subclass; assert we did not fall through to a
+    # platform DestinationConnectionError.
+    assert not isinstance(excinfo.value, DestinationConnectionError)
+
+
</file context>

[cubic-dev-ai]

P3: platform_error_factory(exc) is invoked for every failure, including client-coded ones (UNAUTHENTICATED, PERMISSION_DENIED, INVALID_ARGUMENT, NOT_FOUND) that _classify_milvus_exception immediately reclassifies to a UserAuthError/UserError and discards the factory result. The factory (plus a second safe_error_summary call on a credential-bearing exception) should only run on the fall-through branch when no client code is resolved.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At unstructured_ingest/processes/connectors/milvus.py, line 131:

<comment>`platform_error_factory(exc)` is invoked for every failure, including client-coded ones (UNAUTHENTICATED, PERMISSION_DENIED, INVALID_ARGUMENT, NOT_FOUND) that `_classify_milvus_exception` immediately reclassifies to a UserAuthError/UserError and discards the factory result. The factory (plus a second `safe_error_summary` call on a credential-bearing exception) should only run on the fall-through branch when no client code is resolved.</comment>

<file context>
@@ -31,11 +34,103 @@
+    try:
+        yield
+    except (grpc.RpcError, MilvusException) as exc:
+        raise _classify_milvus_exception(exc, platform_error_factory(exc)) from None
+
+
</file context>