Streaming Validation — Implementation Plan¶

Status: Phase 0 (design). Authored 2026-05-08. Owners: KC + Claude. Scope of this document: the design + phased plan for adding contract-aware validation to pycharter's existing streaming infrastructure. No code yet.

1. Summary¶

PyCharter today ships excellent streaming extraction (Kafka, RabbitMQ, SQS, WebSocket, SSE, file watchers — see Streaming and Messaging Extractors) and excellent batch contract validation (Validator.from_file, validate(...)). It does not connect them: there is no first-class way to consume a Kafka topic, look up the right contract, validate every message, and route bad ones to a DLQ without writing that glue yourself.

This plan adds that connection as a thin layer on top of the existing BaseExtractor framework. The unit of work — the validating consumer — is ergonomic enough to drop into a Python service in 5 lines and robust enough to deploy as a long-running worker handling tens of thousands of messages per second with at-least-once delivery, a poison-pill-proof DLQ, and per-topic quality telemetry.

The first vertical slice is Kafka only, library-first, in enforce mode. Everything else (other sources, gateway worker, replay command, fancy auth) is intentionally deferred to follow-up phases on the same engine.

2. Goals and non-goals¶

Goals¶

Contract validation at consume time for any pycharter BaseExtractor, with Kafka as the first concrete integration.
Library-first ergonomics — async for record in pycharter.kafka.consume(...) yields validated records; bad records auto-route to a DLQ. No worker process required.
Header-based contract resolution as the recommended path; static binding as an opt-in fallback for migration.
Three operating modes: detect, enforce, shadow — so teams can adopt validation without day-one regressions.
At-least-once delivery by default, configurable to at_most_once for teams that prefer no-duplicates.
Robust under realistic failure modes — poison pills, registry outage, sink failure, oversized messages, schema drift, worker restart.
One engine for streaming and batch — the same validator runtime works against any extractor (Kafka, file, S3, HTTP). Streaming is just the case where the extractor never finishes.
Producer-side helper that stamps pycharter-contract / pycharter-contract-version headers, so end-to-end Python flows don't require manual header plumbing.

Non-goals (v1)¶

Stream processing (windowing, joins, aggregations). Pycharter validates; it does not transform. Use Bytewax / Faust / Flink upstream or downstream.
Exactly-once semantics. At-least-once + idempotent producer covers ~99% of needs; transactional producers double the implementation cost.
Confluent Schema Registry wire-format compatibility. Long-term lock-in with no upside for pycharter-native users.
JVM Kafka Connect plugin. Out of language scope.
Standalone gateway worker (Pattern A from the design discussion). Same engine, different sink — deferred to v1.1 once the library API has shipped and stabilized.
New extractor implementations. We compose on existing extractors; we add no new sources in v1.
Replay / backfill CLI. The engine accepts a dry_run flag from day one so the command is a thin v1.1 add-on.

3. Architecture¶

3.1 Component overview¶

                ┌────────────────────────────────────────────────────────────┐
                │                ContractStreamProcessor                     │
                │                                                            │
  Source        │   ┌──────┐    ┌─────────┐    ┌──────────┐   ┌─────────┐    │
  (any         ─┼──►│ poll │───►│ resolve │───►│ validate │──►│ fan-out │    │
  BaseExtractor)│   │      │    │ contract│    │ (pydantic)│   │         │    │
                │   └──────┘    └─────────┘    └──────────┘   └────┬────┘    │
                │       │            │              │              │         │
                │       │ decode_    │ unknown_     │ validation_  │         │
                │       │ error      │ contract     │ failed       ▼         │
                │       └────────────┴──────────────┴──────►   DLQ sink      │
                │                                                  │         │
                │   commit offset / ack ◄──────────────────────────┤         │
                │                                                  │         │
                │   quality metrics reducer (windowed) ◄───────────┘         │
                │                                                            │
                └────────────────────────────────────────────────────────────┘
                                          │
                                          ▼
                              valid sink (in-process iterator
                              for library mode; Kafka topic for
                              gateway mode; file for batch)

3.2 BaseExtractor as the canonical StreamSource¶

PyCharter has a single source-of-truth interface for any data source: BaseExtractor in pycharter/pipeline_generator/extractors/base.py. The streaming engine consumes from BaseExtractor directly via a small structural Protocol (StreamSource) that every extractor satisfies out of the box. There is no parallel hierarchy: a new source implements BaseExtractor and is usable from both the legacy batch pipeline runtime and the streaming validation engine.

BaseExtractor ships two surfaces:

Batch surface (extract_streaming(extract_config, ...)) — yields lists of dicts. The legacy pipeline runtime uses this; it is unchanged.
Streaming-message surface (__aiter__, iter_messages, commit, close) — yields one StreamMessage per record with optional headers, source position metadata, and explicit acknowledgement. The streaming engine uses this. Default implementations wrap the batch surface, so every existing extractor works without changes; sources with richer semantics (Kafka headers, RabbitMQ ack tokens, SQS receipt handles) override the relevant methods.

The shared StreamMessage type lives in pycharter/shared/stream_types.py so both the engine and the extractor framework import from one canonical location.

# Library mode — engine takes any BaseExtractor directly.
processor = ContractStreamProcessor(
    source=KafkaExtractor(topics=["orders"], ...),
    resolver=HeaderContractResolver(),
    validator_factory=registry_factory,
    valid_sink=IteratorSink(),
    dlq_sink=KafkaDLQSink(topic="orders-dlq", ...),
    mode=Mode.ENFORCE,
    delivery=Delivery.AT_LEAST_ONCE,
)
async for record in processor.records():
    process(record)

This means the streaming + batch unification the user wanted is already implemented by the existing extractor framework. The same ContractStreamProcessor accepts a FileJsonExtractor and you have batch contract validation. Same engine, same config schema, different source. Adding a new source — Redis Streams, NATS, MQTT — means implementing one BaseExtractor subclass; pipeline runtime and streaming engine compatibility come for free.

3.3 Public API surface¶

Three layers, smallest to largest commitment:

Layer 1 — validating consumer iterator (the headline)¶

from pycharter import kafka

async for record in kafka.consume(
    topic="orders",
    contract="orders.v2",  # static fallback if header missing
    bootstrap_servers="kafka:9092",
    group_id="my-service",
    dlq_topic="orders-dlq",  # optional; defaults to "<topic>-dlq"
    mode="enforce",  # detect | enforce | shadow
):
    process(record)  # only valid records reach this loop

Internally constructs KafkaExtractor + ContractStreamProcessor + KafkaDLQSink with sensible defaults. 5 lines of user code.

Layer 2 — explicit processor (power users)¶

from pycharter.streaming import ContractStreamProcessor, KafkaDLQSink
from pycharter.pipeline_generator.extractors import KafkaExtractor

processor = ContractStreamProcessor(
    extractor=KafkaExtractor(...),
    contract_resolver=...,
    valid_sink=...,
    dlq_sink=KafkaDLQSink(...),
    mode="enforce",
)
await processor.start()
try:
    async for record in processor.records():
        process(record)
finally:
    await processor.stop()

Used when the caller needs control over lifecycle, sinks, or a custom resolver.

Layer 3 — producer helper (Kafka only, for completeness)¶

from pycharter import kafka

await kafka.produce(
    topic="orders",
    value={"order_id": 123, "amount": 50},
    contract="orders.v2",  # auto-stamps headers
    bootstrap_servers="kafka:9092",
)

Stamps pycharter-contract and pycharter-contract-version headers so that header-based resolution on the consumer side just works. Without this, the end-to-end story relies on every producer hand-rolling the headers correctly.

3.4 Config schema (worker mode)¶

The CLI worker (pycharter stream run config.yaml) is the same engine driven by config instead of code. Single config file, multiple streams, one process.

# config.yaml
streaming:
  streams:
    - name: orders-validator
      source:
        type: kafka
        topics: [orders]
        bootstrap_servers: ${KAFKA_BOOTSTRAP}
        consumer_group: orders-validator
        auth:
          mechanism: scram-sha-512
          username: ${KAFKA_USER}
          password_file: /var/secrets/kafka-password
      contract:
        store: { type: postgres, connection_string: ${DATABASE_URL} }
        ref: { type: store, name: orders, version: "2" }
        resolution: header           # static | header
        on_missing_header: strict    # strict | fallback_to_static
      mode: enforce                  # detect | enforce | shadow
      delivery: at_least_once        # at_least_once | at_most_once
      load:
        target: { type: kafka, topic: orders-validated }
      dlq:
        target: { type: kafka, topic: orders-dlq }
      backpressure:
        max_in_flight: 1000
        registry_outage_timeout_seconds: 60
      quality:
        windows: [1m, 5m, 1h]

  observability:
    metrics_port: 9090
    health_port: 8080
    log_format: json

Multiple streams → one worker process, one consumer group per stream.

3.5 Engine internals¶

Per-partition serial validation¶

For Kafka, validation must be strictly serial within a partition to preserve per-key ordering. Across partitions, validation runs in parallel (one asyncio task per assigned partition). Throughput scales with partition count, not with worker thread count — which matches Kafka's own scaling model.

Contract cache¶

A registry call per message at 10k msg/s would crush Postgres. The engine keeps an in-memory LRU keyed by (contract_id, version):

Default capacity: 1024 entries.
Default TTL: 60 seconds.
Invalidation: on consumer-group rebalance + on SIGHUP.
Stale window: bounded by TTL. Acceptable because contracts are append-only; serving an old contract for ≤60s after a registry update is safe.

Bounded queue / backpressure¶

Between extractor poll and validator, a bounded asyncio.Queue enforces backpressure. When the queue saturates (sink falling behind), the engine pauses the partition assignment via aiokafka's pause() rather than dropping.

Modes¶

detect: validate, emit metrics, log violations, pass through every record to the valid sink unchanged. No DLQ writes. For rollouts.
enforce: valid → valid sink, invalid → DLQ. The default.
shadow: valid → valid sink, invalid → valid sink and DLQ. For belt-and-suspenders transitions where downstream still needs the bad record.

Delivery semantics¶

at_least_once (default): commit offset after sink confirms write. Crash before commit → message reprocessed → duplicate in sink. Validation is deterministic so duplicates are byte-identical.
at_most_once: commit offset immediately after consume, before validate. Crash during validate → message lost. Configurable for teams that prefer loss over duplicates.

Backpressure on registry outage¶

Registry unavailable → block at the resolve step (don't drop, don't pass through). After registry_outage_timeout_seconds, trip a circuit breaker:

/readyz returns 503 → k8s removes the pod from service.
A health metric pycharter_streaming_registry_unavailable flips to 1.
The worker continues polling Kafka but does not advance offsets.

When the registry recovers, the breaker resets and the worker resumes.

3.6 Failure modes catalog¶

Failure	Where detected	Behavior	Reason code in DLQ envelope
Decode error (malformed JSON, truncated)	extractor	Route to DLQ, never crash worker	`decode_error`
Header missing, mode = strict	resolver	Route to DLQ	`no_contract_header`
Header missing, mode = fallback	resolver	Use static contract	(no DLQ)
Unknown contract id	resolver	Route to DLQ	`unknown_contract`
Unknown contract version	resolver	Route to DLQ	`unknown_contract_version`
Validation fails	validator	Route to DLQ with errors	`validation_failed`
Oversized message (> max_message_bytes)	extractor	Route to DLQ without parsing	`oversized_payload`
Sink write fails (transient)	sink	Exponential backoff; do not commit offset	(retry — eventual success)
Sink write fails (permanent)	sink	Trip breaker; mark unready	(worker restart triggered)
Registry unreachable	resolver	Block; trip breaker after timeout	(worker unready)
DLQ sink unavailable	sink	Block consumption; mark unready	(worker unready — never silently drop)
Worker `SIGTERM`	runtime	Drain queue, flush sinks, commit offsets, exit	(graceful)

3.7 DLQ envelope¶

Every DLQ message carries:

{
  "original_payload": "<base64 or raw>",
  "original_headers": {...},
  "error_reason": "validation_failed",
  "validation_errors": [
    {"path": "$.amount", "message": "must be >= 0"}
  ],
  "contract_id": "orders",
  "contract_version": "2",
  "source": {
    "type": "kafka",
    "topic": "orders",
    "partition": 3,
    "offset": 12345,
    "timestamp_ms": 1746700000000
  },
  "worker": {
    "id": "orders-validator-7c8d",
    "version": "pycharter-0.x.y"
  },
  "trace": {
    "traceparent": "00-..."
  }
}

Replay tooling (v1.1) reads this envelope, rehydrates the payload + headers, and feeds it back through the engine.

4. Decisions audit trail¶

Every architecturally significant decision, with the option chosen and why. Future me / future reviewers should be able to challenge any of these.

#	Decision	Choice	Rationale
D1	Where pycharter sits in the topology	Library-first; gateway later via same engine	Library covers Python consumers (most teams). Engine reuse means gateway is a thin sink swap, not a new product.
D2	Contract resolution strategy	Header-based primary; static fallback opt-in	Pycharter-native; no Confluent lock-in; static fallback gives a migration path.
D3	Confluent Schema Registry compatibility	Not in v1	Long-term wire-format lock-in; not required for pycharter-native users.
D4	Default delivery semantics	At-least-once; configurable to at-most-once	Validation is idempotent; data loss > duplicates. Configurable per the user's call.
D5	API style	Async-first; sync iterator wrapper	aiokafka is async; FastAPI is async; one engine, two surfaces.
D6	DLQ default destination	Kafka topic when source is Kafka; configurable to existing pycharter DLQ store	Same scale class as source; replayable with normal Kafka tools. Configurable.
D7	Public API shape	Iterator (Layer 1) + processor (Layer 2) + producer helper (Layer 3); no decorator	Two clear levels; no magic.
D8	Producer-side helper	Yes, `pycharter.kafka.produce(...)`	Header-based resolution requires consistent producer cooperation; without a helper this fails in practice.
D9	Streaming + batch unification	`BaseExtractor` is the canonical source. The streaming engine consumes from it via a structural `StreamSource` Protocol that every extractor satisfies out of the box. The shared `StreamMessage` type lives in `pycharter/shared/stream_types.py`.	One source-of-truth interface for any data source. New extractors get pipeline-runtime + streaming-engine compatibility for free. Avoids parallel hierarchies.
D10	Operating modes	`detect`, `enforce`, `shadow` (default `enforce`)	Real teams need `detect` for safe rollout.
D11	Behavior when header missing	Strict default; opt-in `fallback_to_static` per stream	Clean default for greenfield, escape hatch for migration.
D12	Behavior on registry outage	Block + circuit breaker	Don't lose data; don't silently disable validation; do alert.
D13	Per-partition validation order	Strictly serial within partition; parallel across partitions	Preserves Kafka's per-key ordering; matches Kafka's own scaling model.
D14	Contract cache	In-memory LRU, TTL 60s, rebalance + SIGHUP invalidation	Avoids hammering registry at high QPS. Bounded staleness.
D15	Header naming	`pycharter-contract`, `pycharter-contract-version` (lowercase, dash-separated)	De facto Kafka header convention. Permanent once shipped.
D16	Auth credential sourcing	Env vars + file paths in config; never inline	Secrets-in-YAML is how leaks happen. Config schema rejects `password:` literals.
D17	`auto.offset.reset` default	`latest` for new groups	Don't replay all of history on first deploy; backfill is explicit (replay command).
D18	Trace propagation	W3C `traceparent` end-to-end	Free observability; matches OpenTelemetry conventions.
D19	Idempotent producer	Always on for sinks	Free no-duplicate-on-retry guarantee.
D20	Replay command	Engine accepts `dry_run` from day one; CLI command shipped in v1.1	Engine design bakes it in; user-facing tool can come later.

5. Phased plan¶

Each phase is a single PR. Phase 0 is this document. Phases 1–5 ship code.

Phase 1 — Engine, protocols, in-memory sinks, config (~1100 LOC)¶

Scope: - New top-level package src/pycharter/streaming/. - _protocols.py: StreamSource, StreamSink, DLQSink, ContractResolver, ValidatorFactory protocols. - _processor.py: ContractStreamProcessor engine — runs the consume → resolve → validate → fan-out loop. Modes, delivery semantics, backpressure, circuit breaker, dry_run. - _resolvers.py: HeaderContractResolver, StaticContractResolver, CompositeContractResolver (header with static fallback). - _sinks.py: IteratorSink, MemoryDLQSink. Concrete Kafka sinks come in Phase 2 to keep this PR Kafka-free. - _circuit_breaker.py: time-window breaker for the registry. - config.py: Pydantic models for the YAML config schema. - 43 unit tests with a FakeSource covering every row of the failure-modes catalog.

Phase 1.5 — BaseExtractor unification (~390 LOC)¶

Lands as a small follow-up that establishes the unified source story (D9 in the audit trail) before any Kafka code is written:

pycharter/shared/stream_types.py — canonical home for StreamMessage. Streaming module re-exports for backward compat.
BaseExtractor gains __aiter__, iter_messages, commit, and close with working defaults; existing extractors satisfy the streaming-engine surface unchanged.
6 new tests prove BaseExtractor satisfies the StreamSource Protocol structurally and behaviorally; engine runs end-to-end with a BaseExtractor source.

Combined acceptance criteria for Phase 1 + 1.5: - Engine validates records from a FakeSource and a real BaseExtractor subclass against an in-memory contract and routes correctly in all three modes. - Every failure mode in §3.6 has a dedicated unit test that asserts the DLQ envelope and commit behavior. - BaseExtractor satisfies the StreamSource Protocol via isinstance checks. - ruff clean; legacy extractor + DLQ + pipeline regression suite passes unchanged.

Out of scope: Kafka, real DLQ topic, observability HTTP server, CLI.

Phase 2 — Kafka source/sink + integration tests (~500 LOC + tests)¶

Scope: - streaming/kafka.py: thin wrapper around KafkaExtractor adding header-aware decoding (extracting pycharter-contract / -version headers from aiokafka records before yielding to the engine). - KafkaValidSink and KafkaDLQSink: idempotent aiokafka producers wired to the engine. - Auth: PLAIN, SASL/SCRAM-256/512, mTLS (compose on top of KafkaExtractor's existing auth surface). - tests/integration/streaming/test_kafka_e2e.py using testcontainers-python.

Acceptance criteria: - End-to-end test: produce 10k messages (mix of valid + 5% invalid) → engine consumes, validates, routes → assert valid topic has the valid 9500, DLQ topic has the invalid 500 with correct envelope. - At-least-once invariant: kill worker mid-stream, restart → no message lost, duplicates allowed. - Backpressure invariant: pause DLQ producer → consumer pauses partitions (verified via consumer lag). - Skipped automatically when Docker is unavailable.

Out of scope: producer helper, library API, CLI, quality metrics.

Phase 3 — Producer helper, library API, CLI worker (~400 LOC)¶

Scope: - pycharter/kafka.py: top-level public module with consume(topic, contract, ...) async iterator and produce(topic, value, contract, ...) async function. - Sync wrapper: pycharter.kafka.consume_sync(...) for legacy / notebook use, implemented via a worker thread + asyncio.run_coroutine_threadsafe. - pycharter stream run <config.yaml> CLI subcommand: instantiates engine per stream, runs them concurrently, signals graceful shutdown on SIGTERM. - Tutorial: docs/tutorials/streaming-validation.md — "validate a Kafka topic in 5 minutes."

Acceptance criteria: - Tutorial works end-to-end against testcontainers Kafka. - Producer test: produce → consume → headers round-trip; iterator yields only valid records; bad records observable in DLQ. - CLI: pycharter stream run --config tests/fixtures/streaming/two-streams.yaml starts two consumer groups; SIGTERM triggers clean shutdown (no offset loss, no DLQ message loss). - Sync wrapper: a 5-line script using consume_sync works in a sync for loop.

Out of scope: Prometheus, /healthz HTTP server, quality metrics.

Phase 4 — Quality metrics, observability, hardening (~400 LOC)¶

Scope: - streaming/quality.py: WindowedQualityReducer — rolling counters per stream (valid/invalid/dlq counts, violation rate, p99 validate latency), windows configurable. Writes to existing pycharter quality store. - streaming/observability.py: aiohttp-based mini-server exposing /healthz (always 200 if alive), /readyz (200 only if all streams have closed circuit breakers), /metrics (Prometheus). - Structured JSON logger with stream name, contract id, partition, offset context. - traceparent propagation — pulled from message headers, attached to spans emitted around validate + sink writes. - Chaos test suite: tests/integration/streaming/test_chaos.py — simulate registry outage, sink outage, broker disconnect, OOM-killer SIGKILL.

Acceptance criteria: - Quality store has a streaming_quality_window table populated by the reducer; UI quality dashboard shows per-topic violation rate. - Prometheus scrape returns documented metrics: pycharter_streaming_messages_total{stream,result}, pycharter_streaming_validate_latency_seconds{stream} (histogram), pycharter_streaming_consumer_lag{stream,partition}, pycharter_streaming_circuit_breaker{stream,resource}. - Chaos suite passes: every failure injection results in either a clean DLQ routing or a worker unready state, never silent loss.

Out of scope: replay command, gateway worker, additional sources.

Phase 5 — Docs, examples, polish (~300 LOC, mostly docs)¶

Scope: - docs/guides/streaming-validation.md (replaces or augments the existing streaming-and-messaging.md). - docs/reference/streaming-config-schema.md — full YAML reference. - Example app: examples/streaming/orders/ — producer + consumer, runnable with docker compose up. - Migration guide for users of the existing KafkaExtractor. - API stability label: pycharter.kafka and pycharter.streaming marked experimental in 0.x; promoted to stable at 1.0 once two consecutive patch versions ship without breaking changes. - Public changelog entry.

Acceptance criteria: - Tutorial gets a clean mkdocs build --strict. - Example app works on a fresh checkout via documented commands. - Reviewer who has never seen this design can read the guide and ship a validating consumer in <30 minutes.

6. v1.1+ roadmap¶

Out of scope for v1, listed here so we don't lose them:

Replay / backfill CLI — pycharter stream replay --topic ... --from-offset ... --contract ... [--dry-run]. Engine already supports dry_run; this is the user-facing wrapper.
Pattern A — standalone gateway worker — same engine, valid_sink: kafka_topic instead of iterator. Mostly docs + ops + Helm chart.
Schema compatibility checker — a producer-side API: "given a v3 message, would it also validate as v2?"
Additional sources for the validating consumer — Redis Streams, NATS JetStream, AWS Kinesis, GCP Pub/Sub. Each is a small composition over an existing or new BaseExtractor.
Auth: AWS MSK IAM, OAuth (OIDC) — needed for cloud-managed Kafka in regulated environments.
Per-record schema attachment — payload itself contains a schema fingerprint, no headers required. (For protocols where headers don't exist, e.g. raw TCP.)
Rate-limited DLQ — for runaway producers, cap DLQ writes per second to avoid filling the DLQ topic faster than triage can drain it.
Promotion to stable at 1.0 once the API has been used in production for two consecutive patch releases without breaking changes.

7. Risks and mitigations¶

Risk	Likelihood	Mitigation
Pycharter validation is CPU-bound; can't keep up with high-QPS topics	Medium	Per-partition serial design means scaling axis = partition count → consumer count. Document the throughput/partition floor in the tutorial. Phase 4 chaos test asserts a target throughput.
Header-based resolution is hard to migrate to from existing producers	Medium	The `fallback_to_static` opt-in is the migration path. Producer helper makes new producers correct by default.
Registry outage takes down the worker	Low (with mitigation)	Circuit breaker + cache TTL means cached contracts keep flowing for ~60s. After that the worker becomes unready, not failed.
DLQ topic itself fails	Low	DLQ sink failure trips the breaker — never silently drop bad records. Document operator procedure for DLQ recovery.
Sink write storm during partial outage	Medium	Bounded queue + idempotent producer + exponential backoff. Worker pauses partitions rather than buffering unboundedly.
Schema evolution mid-stream produces validation errors	Medium	Header-based resolution makes evolution explicit. Producers stamp the version they wrote against; consumers validate against that version. Old + new coexist on the topic without conflict.
Operator deploys `enforce` on day one and DLQs 5% of traffic by surprise	High (without `detect`)	Three modes exist precisely for this. Tutorial guides operators through `detect → shadow → enforce`.

8. Open questions¶

None blocking Phase 1. The design is fully specified. Future reviewers should still feel free to challenge any decision in §4 — the audit trail exists specifically to make that easy.

9. References¶

Existing pycharter docs: Streaming and Messaging Extractors.
Existing pycharter code: src/pycharter/pipeline_generator/extractors/_kafka_extractor.py, src/pycharter/dlq/.
Adjacent design: Vocabulary implementation plan — reference for plan format and phasing style.