Integration Pipeline Console
An operator console for an idempotent enterprise integration pipeline.
The Idea
In multi-system enterprise integration, the things that go wrong most often aren’t the things on the architecture diagram. Duplicate delivery from a retried POST. A consumer-group restart that replays last night’s batch. An ERP push at 10:03 and a CRM push at 10:02 that doesn’t arrive until 10:05 — which version wins? Most pipelines answer these questions implicitly, somewhere deep in code, and an operator has no straightforward way to ask: what actually happened to record X, and why?
The Integration Pipeline Console is a runnable reference implementation that puts those answers on the surface. It’s a backend-and-API operator console accompanying a planned three-part series on enterprise integration reliability. The point isn’t to be a production platform — it’s to make the architectural patterns inspectable, so an engineer reading the article can also click through them.
The Approach
Three pillars, each building on the last:
- Correctness — every message passes through a five-branch idempotency decision before any write. Staging-to-production promotion. Correlation across systems for cross-business-event grouping. Append-only audit log.
- Delivery resilience — retry classification, per-system circuit breakers (DB-persisted state), DLQ routing, an ambiguous-outcome queue for timeout cases, and query-before-retry for downstream systems that support status checks. Operator-initiated replay calls the same delivery path as any original delivery — no shortcut, no silent bypass.
- Explainability — a cross-context inspector that assembles a record’s full history from all 14 tables in a single API call. A dashboard that surfaces what needs attention right now.
The five-branch deduplication is the centerpiece. Most systems implement two or three branches and silently overwrite or skip the rest:
| Condition | Decision |
|---|---|
| Key not seen before | accept_new |
| Same key, newer version | accept_supersede |
| Same key, same version, same payload hash | skip_duplicate |
| Same key, same version, different payload hash | quarantine |
| Same key, older version | skip_stale |
The quarantine branch is the one that matters most. It surfaces data-quality anomalies — two systems claiming different facts about the same business event at the same version — instead of silently picking one. It fails loudly so a human can decide.
Implementation
- Async FastAPI and SQLAlchemy.
- SQLite default (zero-dependency local development); Postgres swap via a single env var, no code changes.
- 14 tables across the three pillars: idempotent state, ingestion attempt log, staging, production projection, correlation map, data anomaly queue, downstream system, delivery attempt log, delivery DLQ, circuit breaker state, ambiguous outcome queue, replay request, replay request item, replay execution log.
- 25+ API endpoints: ingestion, state queries, delivery simulation, circuit management, ambiguous-outcome resolution, replay, inspector, jobs and reports.
- Pytest suite covering replay safety, deduplication branches, partial-to-complete state transitions, and conflicting same-version updates.
A few design decisions worth naming explicitly. Circuit recovery is operator-controlled, not time-based — the half-open transition requires a deliberate reset, which forces a human to confirm they believe the downstream has recovered. Replay eligibility is all-or-nothing — a mixed batch where some items are ineligible fails loudly rather than silently doing partial work. Replay calls simulate_delivery() exactly like any original delivery, and the ReplayExecutionLog records what came back — that log is the machine-readable proof that no shortcut was taken.
Status
Open source on GitHub. Demo-grade and runnable locally; deliberately not a production platform. The accompanying three-part series on enterprise integration reliability is in preparation.