# ADR-005 — GenStage + Broadway for Async / Batch Processing

**Date:** 2026-04-26
**Status:** Accepted
**Deciders:** Architecture Team

---

## Context

Several south-side integrations are inherently asynchronous or high-volume batch:

- **File ingestion:** SFTP delivers CSV/XML/ISO files of 10k–500k rows. Processing must not
  block the HTTP request pipeline.
- **Data warehouse loads:** Bulk inserts of transaction records from multiple sources.
- **Core banking callbacks:** Asynchronous settlement notifications arrive from core banking
  after a transaction is accepted (seconds to minutes later).

Handling these synchronously on the Plug pipeline would either time out clients or hold
request-handler processes for minutes — neither is acceptable.

---

## Decision

Use **Broadway** (built on GenStage) for all async and batch processing pipelines.

Broadway pipelines live in `infra_queue` as reusable building blocks. Adapters (`adapter_file`,
`adapter_dw`) define their own Broadway pipeline configurations and plug into these building
blocks.

---

## Broadway Pipeline Design

### File Ingestion Pipeline

```
adapter_file.FileWatcher (GenServer)
    │ emits row batches as Broadway messages
    ▼
infra_queue.FilePipeline (Broadway)
    ├── Processors: MwTransform.map(row) → MwKernel.Message  [concurrency: 10]
    ├── Batchers:   adapter_dw.BatchLoader.bulk_insert        [batch_size: 500, timeout: 5s]
    └── DLQ:        infra_repo.DeadLetterStore.insert
```

### Core Banking Callback Pipeline

```
adapter_banking (receives async webhook/TCP callback)
    │ emits settlement event
    ▼
infra_queue.CallbackPipeline (Broadway)
    ├── Processors: enrich message, update tx status in DB   [concurrency: 5]
    └── Fanout:     Phoenix.PubSub.broadcast → gateway_ws
```

---

## Why Broadway over Alternatives

| Option | Considered | Rejected Because |
|--------|-----------|-----------------|
| `Task.async` / `Task.Supervisor` | Simple fire-and-forget | No back-pressure; runaway tasks under load; no DLQ |
| Raw `GenStage` | Full control | Broadway is GenStage + DLQ + telemetry + ack; reinventing it adds no value |
| RabbitMQ / Kafka | External broker | Operational dependency for a use case that fits in-process; adds latency; viable if multi-node fan-out is needed in future |
| `Oban` (job queue) | Persistent jobs | Oban is DB-backed periodic jobs; wrong model for streaming ingestion; consider for retry scheduling |

---

## Back-pressure Model

Broadway's demand-driven model prevents `adapter_file` from reading files faster than
`adapter_dw` can insert them:

```
FileWatcher produces 100 messages → Broadway demands 10 at a time
→ if DW is slow, demand drops → FileWatcher reads more slowly
→ SFTP backlog builds on server, not in BEAM memory
```

This prevents OOM scenarios when SFTP files are large.

---

## Dead Letter Queue (DLQ)

Failed messages (transformation error, DW insert error after retries) are routed to the DLQ:

- Stored in `dead_letter_queue` DB table (via `infra_repo`)
- Includes: original message, error reason, adapter, timestamp, retry count
- Admin UI (`gateway_web`) shows DLQ depth and allows re-queue or discard
- Alert fires (Telemetry → Prometheus → Alertmanager) when DLQ depth exceeds threshold

---

## Acknowledgement Strategy

Broadway messages are acknowledged only after the batch batcher confirms successful insert:

```
ack: :on_success   → DW insert succeeded → ACK → FileWatcher advances cursor
ack: :on_failure   → DW insert failed → NACK → Broadway retries (max 3) → DLQ
```

SFTP file cursor is advanced only after full file acknowledgement — ensures no rows are
silently dropped on crash mid-file.

---

## Consequences

### Positive
- Back-pressure prevents OOM under burst load
- DLQ ensures no data loss on processing failure
- Built-in Telemetry events for Broadway stages (throughput, errors, queue depth)
- Concurrency is configurable per pipeline without code changes

### Negative
- Broadway pipelines are stateful processes; they must be supervised and gracefully drained
  on shutdown — Mix releases handle this via `System.stop/1` hooks
- Broadway's concurrent processors mean message ordering is not guaranteed within a batch;
  adapters that require strict ordering must use `concurrency: 1` or sequence IDs