---
title: 'TEXTIQ-394 — Circuit breakers on DE outbound calls (Milvus, SeaweedFS, LLM)'
type: 'feature'
created: '2026-05-04'
status: 'complete'
baseline_commit: '3571295'
final_commit: '016f5a6'
context:
  - '_planning/docs/services/textiq-doc-extraction/prd.md (SYS-NFR-20)'
  - 'docs/planning-artifacts/de-airflow-gap-audit-2026-04-20.md (GAP-031)'
  - 'docs/implementation-artifacts/spec-textiq-391-prometheus-metrics.md'
---

<frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">

## Intent

**Problem:** DE has zero circuit-breaker protection on its three external upstreams (Milvus, SeaweedFS/S3, LLM provider). A flaky upstream causes uncapped retry storms, head-of-line blocking on Airflow workers, and cascading failures. SYS-NFR-20 mandates 5 failures / 30s → open 60s; current state is `grep -ri circuit airflow/` returns only an unrelated HMAC docstring.

**Approach:** Add three named `pybreaker` instances (one per upstream) in a new `src/resilience/breakers.py` module. Wrap the public wrapper functions in `airflow/plugins/api/milvus.py`, `airflow/plugins/api/s3.py`, and the LLM SDK call sites under `src/`. State changes emit a Prometheus `Gauge` (0=closed, 1=half_open, 2=open) labeled by upstream, registered in the existing `airflow/plugins/api/metrics.py`. pytest covers all transitions.

## Boundaries & Constraints

**Always:**
- One breaker singleton per upstream; never per-call-site.
- Thresholds match SYS-NFR-20: `fail_max=5`, `reset_timeout=60`.
- Breaker state is observable via Prometheus before any new code is shipped — no silent breakers.
- Breaker open raises a typed exception; callers may catch it for graceful degradation but must not retry it inside the same task.

**Ask First:** (none — all resolved at planning. Locked decisions recorded under "Resolved decisions" below.)

**Resolved decisions (locked at planning):**
- Failure semantics: `pybreaker` consecutive-5 (default). Not sliding-window. Re-evaluate in a follow-up if production shows flapping isn't being caught.
- `upload.py`: refactor direct boto3 calls to go through `s3.py` helpers; do **not** inline-wrap.
- LLM wrap: inside `src/llm/factory.py` (factory exposes breaker-wrapped helper calls); not at SDK call sites.
- Ticket's ADR-006 reference: treat as typo; ignore (real ADR-006 in this repo is the Hybrid Excel ADR).

**Never:**
- Do NOT install or depend on `textiq-client-sdk` — it is unbuilt; gap audit explicitly defers it.
- Do NOT modify `airflow/plugins/api/middleware.py:65` ("Short-circuits …") — it is HMAC auth, not a breaker; rename is out of scope.
- Do NOT add `tenacity` or other retry libs; Airflow task-level retry already handles transient failure at the orchestration layer.
- Do NOT wrap third-party internals (`llama_index.MilvusVectorStore`, `openai.AsyncAzureOpenAI` clients themselves) — wrap at the call boundary we own.
- Do NOT introduce multiprocess metrics; the TEXTIQ-391 single-worker constraint (`AIRFLOW__API__WORKERS=1`) still holds.

## I/O & Edge-Case Matrix

| Scenario | Input / State | Expected Output / Behavior | Error Handling |
|----------|--------------|---------------------------|----------------|
| Healthy upstream | Breaker closed, call succeeds | Result returned, `circuit_breaker_state{upstream=...}=0` | N/A |
| Transient failure | Breaker closed, 1–4 consecutive failures | Original exception raised; state stays closed | Caller sees raw exception |
| Trip threshold | 5th consecutive failure | Breaker opens; gauge → 2; state-change log emitted | Caller sees raw exception on this call |
| Open-state call | Breaker open, within `reset_timeout` | `pybreaker.CircuitBreakerError` raised immediately (no upstream call) | Caller sees `CircuitBreakerError`; Airflow task fails fast |
| Half-open trial success | After 60s, first call succeeds | Breaker closes; gauge → 0 | N/A |
| Half-open trial failure | After 60s, trial call fails | Breaker re-opens for another 60s; gauge → 2 | Caller sees raw exception |
| Metric scrape during transition | `/metrics` scraped | `circuit_breaker_state{upstream="milvus|s3|llm"}` present with current value | N/A |

</frozen-after-approval>

## Code Map

- (dependency) -- add `pybreaker>=1.2.0` via `uv add pybreaker` (do not hand-edit `pyproject.toml`)
- `src/resilience/__init__.py` -- new package marker
- `src/resilience/breakers.py` -- new: three named breakers, the `circuit_breaker_state` Gauge (registered against default `prometheus_client.REGISTRY` so the existing `/metrics` endpoint scrapes it), state listener, sync+async helpers
- (no change to `airflow/plugins/api/metrics.py` — module header forbids imports from `src.*`; gauge is declared next to its updater in `src/resilience/breakers.py` and registers against the same default REGISTRY)
- `airflow/plugins/api/milvus.py` -- wrap public functions (`get`, `upsert`, `delete`, `query`) with `milvus_breaker`
- `airflow/plugins/api/s3.py` -- wrap public S3 helpers with `s3_breaker`
- `airflow/plugins/api/s3.py` -- add public helpers used by `upload.py` (`put_object`, `head_object`, `generate_presigned_url`) if not already present, all wrapped with `s3_breaker`
- `airflow/plugins/api/upload.py` -- **refactor**: replace direct `boto3` calls (L177, L609, L627–628) with `s3.py` helpers; remove `_get_presign_s3_client` if it becomes unreferenced
- `airflow/dags/operators/fetch_document.py` -- wrap `s3_client.download_file` with `s3_breaker`
- `airflow/dags/tasks/processing_tasks.py` -- wrap the `MilvusVectorStore.add` / equivalent calls with `milvus_breaker`
- `src/llm/factory.py` -- expose breaker-wrapped helpers `create_chat_completion(client, **kw)` (sync) and `async_create_embedding(client, **kw)` (async) that route through `llm_breaker`; both are the new public path for LLM calls
- `src/knowledge/embeddings.py` -- replace direct `await client.embeddings.create(...)` with `await async_create_embedding(client, ...)`
- `src/extraction_v2/llm_header_detector.py` -- replace direct `client.chat.completions.create(...)` with `create_chat_completion(client, ...)`
- `tests/airflow/test_breakers.py` -- new pytest module covering all transitions + metric updates

## Tasks & Acceptance

**Execution:**
- [x] (shell) `uv add pybreaker` -- adds `pybreaker>=1.2.0` to project deps and refreshes the lockfile; do **not** hand-edit `pyproject.toml`
- [x] `src/resilience/breakers.py` -- define `milvus_breaker`, `s3_breaker`, `llm_breaker` with `fail_max=5, reset_timeout=60`; declare `circuit_breaker_state` Gauge (idempotent against default `prometheus_client.REGISTRY`); register a `CircuitBreakerListener.state_change` callback that sets the gauge and emits `logger.warning` with `{upstream, old_state, new_state}`; export `breaker_call(breaker, fn, *args, **kwargs)` and `async_breaker_call(...)` helpers -- single source of truth for breaker config; gauge registered against default REGISTRY so existing `/metrics` route scrapes it without changes to `metrics.py`
- [x] `airflow/plugins/api/milvus.py` -- wrap each public function body with `milvus_breaker.call(...)` -- protects all plugin-side Milvus ops
- [x] `airflow/plugins/api/s3.py` -- ensure public helpers exist for `put_object`, `head_object`, `generate_presigned_url`, and `download_file`; wrap each with `s3_breaker.call(...)` -- single chokepoint for plugin + DAG S3 ops
- [x] `airflow/plugins/api/upload.py` -- refactor: replace direct boto3 calls at L177 / L609 / L627–628 with `s3.py` helper calls; delete `_get_presign_s3_client` if no remaining callers -- collapses two boto3 init paths into one wrapped path
- [x] `airflow/dags/operators/fetch_document.py` -- replace direct `s3_client.download_file` (L116) with the wrapped `s3.py` helper -- covers DAG-side S3 fetch
- [x] `airflow/dags/tasks/processing_tasks.py` -- wrap MilvusVectorStore `.add()` invocations (L2682–2743 + L2004) with `milvus_breaker.call(...)` -- covers DAG-side vector writes
- [x] `src/llm/factory.py` -- add `create_chat_completion(client, **kw)` (sync) and `async_create_embedding(client, **kw)` (async); both internally call the SDK and are wrapped with `llm_breaker` -- new public LLM path used by all callers
- [x] `src/knowledge/embeddings.py` -- replace `await client.embeddings.create(...)` (L154) with `await async_create_embedding(client, ...)` from `src.llm.factory`
- [x] `src/extraction_v2/llm_header_detector.py` -- replace `client.chat.completions.create(...)` (L231) with `create_chat_completion(client, ...)` from `src.llm.factory`
- [x] `tests/airflow/test_breakers.py` -- pytest cases: `test_closed_state_passes_through`, `test_opens_after_5_consecutive_failures`, `test_open_blocks_calls_with_CircuitBreakerError`, `test_half_open_after_reset_timeout`, `test_half_open_success_closes`, `test_half_open_failure_reopens`, `test_metric_gauge_reflects_state` -- enforces SYS-NFR-20 and metric correctness; use `monkeypatch` on `time` or `pybreaker.CircuitBreaker._state_storage` to avoid real 60s sleeps

**Acceptance Criteria:**
- Given a closed breaker, when 5 consecutive calls raise an exception, then the breaker transitions to open and `circuit_breaker_state{upstream=X}` reads `2`.
- Given an open breaker within its 60s reset window, when any call is attempted, then `pybreaker.CircuitBreakerError` is raised without invoking the upstream.
- Given an open breaker after 60s elapsed, when the next call succeeds, then the breaker closes and the gauge reads `0`; when it fails, then the breaker re-opens and the gauge reads `2`.
- Given a fresh `/metrics` scrape, when any breaker is in any state, then the response includes a `circuit_breaker_state` line for each of `milvus`, `s3`, `llm`.
- Given the test suite, when `pytest tests/airflow/test_breakers.py` runs, then all transition tests pass without real-time `sleep(60)` calls.

## Spec Change Log

- 2026-05-04 — `_make_breaker` sets `throw_new_error_on_trip=False` and `async_breaker_call` reimplements `State.call` (using `before_call` / `_handle_error` / `_handle_success`) instead of calling `breaker.call_async`. Two pybreaker 1.4.1 quirks discovered during verification: (1) the default `throw_new_error_on_trip=True` swallows the original upstream exception on the trip call, contradicting the I/O matrix row "Caller sees raw exception on this call"; (2) `CircuitBreaker.call_async` references tornado's `gen` without importing it, raising `NameError` on first use. Both fixes are local to `src/resilience/breakers.py`; the test fixture mirrors the production setting.
- 2026-05-04 — `tests/airflow/test_operators/conftest.py` no longer mocks `sys.modules['src']` itself (only specific submodules). The previous blanket `MagicMock` shadowed real `src.resilience.*` and `src.services.*` imports for sibling tests once `airflow/plugins/api/s3.py` joined the operator-import chain via `fetch_document.py`. Submodule mocks still cover heavy-dep modules (extraction_v2, knowledge, etc).
- 2026-05-04 — `tests/airflow/test_metrics.py` updated to patch `api.upload.put_object` instead of `api.upload.get_s3_client`, reflecting the upload.py refactor that routes all S3 ops through the breaker-wrapped `s3.py` helpers.

## Verification

**Commands:**
- `uv run pytest tests/airflow/test_breakers.py -v` -- expected: all transition tests pass
- `uv run pytest tests/airflow/ -k "metrics or breaker"` -- expected: metric-shape tests still green; new gauge appears in scrape
- `uv run python -c "import pybreaker; from src.resilience.breakers import milvus_breaker, s3_breaker, llm_breaker; print([(b.name, b.fail_max, b.reset_timeout) for b in (milvus_breaker, s3_breaker, llm_breaker)])"` -- expected: `[('milvus', 5, 60), ('s3', 5, 60), ('llm', 5, 60)]`
- `uv run ruff check src/resilience airflow/plugins/api/metrics.py airflow/plugins/api/milvus.py airflow/plugins/api/s3.py` -- expected: clean