---
title: 'GAP-027 / TEXTIQ-390: Pipeline stage events (Redis pub/sub) + tighten DAG retry policy'
type: 'feature'
created: '2026-04-23'
status: 'done'
baseline_commit: 'd13c78120b7c34be98a8ae989d68e694044fc2c8'
context:
  - docs/architecture.md
  - docs/implementation-artifacts/de-airflow-gap-audit-2026-04-20.md
---

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

## Intent

**Problem:** (1) DE has no push channel for pipeline progress — clients must poll `GET /documents/{id}`, and UPA cannot relay live updates to the browser's Socket.IO `/pipeline` namespace that Architecture requires. (2) Current DAG default is `retries=2` on every task, which re-executes deterministic tasks (DB writes, chunking, indexing) on transient failures and emits duplicate events.

**Approach:** DE publishes stage-transition events to Redis pub/sub on per-tenant+node channels; UPA (out of scope) subscribes and forwards. DE runs no Socket.IO server. Separately, tighten retry policy: `default_args.retries=0`; only the five format parsers in `parse_tasks.py` keep `retries=2` (network/IO-flaky fetch+parse is the one worth retrying). Also fix a latent gap: `update_document_status_task` currently writes `completed` unconditionally on `all_done`; it must resolve the terminal outcome from upstream TI states and write `failed` when any upstream TI failed.

## Boundaries & Constraints

**Always:**
- Channel `{prefix}:{tenant_id}:{node_id}` (default prefix `pipeline`). Body: `correlation_id` (= Airflow `dag_run_id`), `document_id`, `tenant_id`, `node_id`, `stage`, `status`, `timestamp` (ISO-8601 UTC), optional `error`.
- Stages: `pending`, `processing`, `parsing`, `chunking`, `indexing`, `completed`, `failed` (aligned with `documents.status` vocabulary; `parsing`/`chunking`/`indexing` are event-only sub-phases of DB `processing`). Statuses: `started`, `succeeded`, `failed`.
- Feature-gated by `PIPELINE_EVENTS_ENABLED` (default `false`). Off = no-op, no Redis connection.
- Publish failures logged WARNING and swallowed — never fail a DAG task on emit failure.
- Reuse shared `textiq-redis` via `settings.redis_url`.
- Retry policy: `dag_config.default_args.retries=0, retry_delay=30s`; the five parsers in `parse_tasks.py` set `retries=2, retry_delay=timedelta(seconds=30), retry_exponential_backoff=True` on their task decorators.

**Ask First:**
- Changing event schema fields, stage vocabulary, or channel format (UPA contract).
- Making the publisher async or blocking task success on publish.
- Persisting events (would change semantics from pub/sub to stream).
- Adding retries back on any non-parse task, or changing parser retry count/backoff.

**Never:**
- Run a Socket.IO server in DE.
- Emit PII (content, domain terms, tag values, filenames) in payloads.
- Couple DAG task success to publish success.

## I/O & Edge-Case Matrix

| Scenario | State | Behavior |
|----------|-------|----------|
| Happy path | flag on, Redis up, DAG ok | One event per stage; terminal `{completed, succeeded}` |
| Upstream fails | task raises | Terminal `{failed, failed, error:<str>}` from `update_document_status` |
| Flag off | `PIPELINE_EVENTS_ENABLED=false` | Zero `PUBLISH` calls; no Redis connection |
| Redis down | flag on, broker unreachable | DAG tasks complete; WARNING per publish; `RedisError` swallowed |
| Missing tenant/node | Param omitted | Publisher logs ERROR, returns; no raise |
| Non-parse task fails | e.g. chunking/indexing raises | No retry; DAG fails immediately; single `failed` event |
| Parser fails transiently | parse task raises once | Airflow retries up to 2× with 30s exp-backoff; on final failure `failed` event |

</frozen-after-approval>

## Code Map

- `src/core/config.py` -- add `pipeline_events_enabled`, `pipeline_event_channel_prefix` settings.
- `src/services/pipeline_events.py` -- NEW. Sync publisher `publish_stage_event(...)`; flag check + `RedisError` guard.
- `airflow/dags/dag_config.py` -- flip `default_args.retries` to `0` (keep `retry_delay` and backoff as inherited defaults for tasks that set their own retry count).
- `airflow/dags/tasks/parse_tasks.py` -- add `retries=2, retry_delay=timedelta(seconds=30), retry_exponential_backoff=True` to each of the five format parsers; emit `parsing` at entry.
- `airflow/dags/tasks/processing_tasks.py` -- emit at `save_document_metadata_task` (pending + processing), `chunk_large_table_records` (chunking), each `store_*_llamaindex_task` (indexing), `update_document_status_task` (terminal). Terminal task must resolve outcome from upstream TI states, write `documents.status='completed'|'failed'` accordingly, and emit the matching event.
- `tests/services/test_pipeline_events.py` -- NEW. Unit tests with `Mock()` Redis (pattern from `tests/extraction/test_pipeline.py:36`); cover every I/O matrix row.
- `docs/architecture.md` -- new `## Pipeline Events (DE → UPA)` subsection under *Document Processing Pipeline*.
- `docs/implementation-artifacts/de-airflow-gap-audit-2026-04-20.md` -- mark GAP-027 resolved; link this spec.

## Tasks & Acceptance

**Execution:**
- [x] `src/core/config.py` -- add `pipeline_events_enabled: bool = False`, `pipeline_event_channel_prefix: str = "pipeline"` to `Settings`
- [x] `src/services/pipeline_events.py` -- NEW module; `publish_stage_event(stage, status, *, document_id, tenant_id, node_id, correlation_id, error=None)`; helpers `_build_channel`, `_build_payload`
- [x] `airflow/dags/dag_config.py` -- set `default_args['retries'] = 0`
- [x] `airflow/dags/tasks/parse_tasks.py` -- add `retries=2, retry_delay=timedelta(seconds=30), retry_exponential_backoff=True` on each of the 5 parser task decorators; emit `parsing` at entry
- [x] `airflow/dags/tasks/processing_tasks.py` -- emit `pending` at entry of `save_document_metadata_task`, `processing` after DB upsert
- [x] `airflow/dags/tasks/processing_tasks.py` -- emit `chunking` at `chunk_large_table_records`; `indexing` at each `store_*_llamaindex_task`
- [x] `airflow/dags/tasks/processing_tasks.py` -- in `update_document_status_task`, resolve terminal outcome from upstream TI states; write `documents.status='completed'|'failed'` (fixes latent bug — was unconditional `completed`) and emit the matching event with `error` captured from failed TI
- [x] `tests/services/test_pipeline_events.py` -- cover all I/O matrix rows + payload shape + channel format
- [x] `docs/architecture.md` -- add `## Pipeline Events (DE → UPA)` section (channel, schema, stages, flag, failure semantics)
- [x] `docs/implementation-artifacts/de-airflow-gap-audit-2026-04-20.md` -- mark GAP-027 resolved
- [ ] Create UPA-side Jira ticket for the Redis-subscribe → Socket.IO `/pipeline` relay and link it from TEXTIQ-390 -- AC "UPA integration story filed or linked"

**Acceptance Criteria:**
- Given flag on and Redis reachable, when a DAG processes a document end-to-end, then exactly one event per stage transition is published to `pipeline:{tenant_id}:{node_id}` and the terminal event is `{completed, succeeded}`.
- Given a task fails upstream, when `update_document_status_task` runs, then `documents.status='failed'` is written AND a `{failed, failed, error:<str>}` event is published.
- Given all upstream tasks succeed, when `update_document_status_task` runs, then `documents.status='completed'` is written AND a `{completed, succeeded}` event is published.
- Given flag off, when the DAG runs, zero `PUBLISH` calls are made (mock-asserted).
- Given Redis down and flag on, DAG tasks still reach terminal state and publish failures log WARNING.
- Payload contains all seven required fields (`correlation_id`, `document_id`, `tenant_id`, `node_id`, `stage`, `status`, `timestamp`).
- Given any non-parse task raises, DAG fails without retry; given a parser raises transiently, Airflow retries up to 2× before the DAG fails. Verified by inspecting `retries` on task instances in a test DAG run.

## Design Notes

- **`correlation_id` = `dag_run_id`** — no existing corr_id in repo; `dag_run_id` is unique per run and already in context.
- **Sync publisher** — `@task.external_python` tasks run in child interpreters; async loop for one `PUBLISH` is wasteful.
- **Explicit emits, not Airflow callbacks** — callbacks run in scheduler process without task deps.

Payload example:
```json
{"correlation_id":"manual__2026-04-23T10:00:00+00:00","document_id":"...","tenant_id":"...","node_id":"...","stage":"parsing","status":"started","timestamp":"2026-04-23T10:00:12.345Z"}
```

## Verification

- `uv run pytest tests/services/test_pipeline_events.py tests/airflow/ -v` -- all pass
- Integration: `redis-cli -h localhost PSUBSCRIBE 'pipeline:*'` while triggering DAG with flag on -- one JSON per stage, terminal `completed`
- Manual resilience: kill Redis mid-run; DAG tasks still succeed, WARNINGs in worker logs

## Suggested Review Order

**Publisher contract (entry point)**

- Fire-and-forget Redis publisher; flag-gated no-op; all errors swallowed.
  [`pipeline_events.py:65`](../../src/services/pipeline_events.py#L65)

- Channel format `{prefix}:{tenant_id}:{node_id}` and payload schema.
  [`pipeline_events.py:37`](../../src/services/pipeline_events.py#L37)

- Feature flag + channel prefix on `Settings`.
  [`config.py`](../../src/core/config.py)

**DAG orchestration: retry policy + terminal outcome**

- Default `retries=0` so non-parse tasks fail fast; parsers opt-in to 2× retry.
  [`dag_config.py:54`](../../airflow/dags/dag_config.py#L54)

- Terminal outcome resolver uses XCom pulls (Airflow 3 SDK-safe); ORM not accessible from workers.
  [`document_extraction_dag.py:124`](../../airflow/dags/document_extraction_dag.py#L124)

- Wired `[stores, record_count] >> resolve_terminal_outcome >> update_doc_status` with `all_done` trigger.
  [`document_extraction_dag.py:545`](../../airflow/dags/document_extraction_dag.py#L545)

- Removed `none_failed_min_one_success` from 4 text-path tasks so Excel-branch skip cascades correctly.
  [`document_extraction_dag.py:491`](../../airflow/dags/document_extraction_dag.py#L491)

**Emit sites (stage transitions)**

- Parsers emit `parsing/started` inline (module-level helpers invisible to `@task.external_python` child process).
  [`parse_tasks.py:73`](../../airflow/dags/tasks/parse_tasks.py#L73)

- `save_document_metadata_task` emits `pending/started` then `processing/started` after DB upsert.
  [`processing_tasks.py:133`](../../airflow/dags/tasks/processing_tasks.py#L133)

- Stores emit `indexing/started`; `update_document_status_task` writes `documents.status` and emits terminal event.
  [`processing_tasks.py:2353`](../../airflow/dags/tasks/processing_tasks.py#L2353)

**Docs**

- New `## Pipeline Events (DE → UPA)` section: channel, schema, failure semantics.
  [`architecture.md`](../../docs/architecture.md)

- GAP-027 marked resolved; links back to this spec.
  [`de-airflow-gap-audit-2026-04-20.md`](./de-airflow-gap-audit-2026-04-20.md)

**Tests + config + deferred work**

- 12 unit tests: happy path, flag off, Redis error, missing routing IDs, payload shape, channel format.
  [`test_pipeline_events.py`](../../tests/services/test_pipeline_events.py)

- Flag template for operators.
  [`.env.example`](../../.env.example)