Skip to content

Proposal: First-class streaming ingestion

Status: implemented (2026-07) · Original proposal: 2026-06-03 · ST-1 … ST-14 all shipped

Context / problem

pycharter already has the hard parts of streaming built: a real per-message engine (pycharter.streaming.ContractStreamProcessor), six streaming extractors (websocket, kafka, sse, rabbitmq, sqs, file_watcher), DLQ, delivery semantics (at_least_once / at_most_once), backpressure (BackpressureConfig.max_in_flight), and replay (pycharter stream replay).

The value is under-wired, in three concrete ways found during a streaming review (driven by a live market-data websocket use case, e.g. Massive, formerly Polygon):

  1. The config-driven worker is Kafka-only. pycharter stream run builds its source/sinks in _cli_streaming.py::_build_processor, which hardcodes KafkaExtractor and Kafka sinks — even though StreamConfig.source / valid_sink / dlq_sink (streaming/config.py) are already generic dicts and every extractor satisfies the StreamSource protocol. A websocket source is only reachable via hand-written library-mode code.
  2. No normalization in the streaming path. The engine validates and type-coerces against a contract (Pydantic model(**payload) in runtime_validator/validator_core.py), but cannot rename / compute / enrich. Terse feed payloads ({ev,sym,p,s,t}) cannot become {event,symbol,price,size,timestamp} on the topic — even though the batch DAG already has a full transform library (pipeline_generator/transformers).
  3. No first-class stream sinks to storage, and no dead-socket detection. Worker sinks are Kafka-only; the file/Postgres/cloud loaders cannot terminate a stream. The websocket/SSE read loops wait forever on a silent socket — StreamingConfig.idle_timeout exists (extractors/_streaming.py) but is referenced nowhere.

The batch DAG engine is the wrong tool for an unbounded feed: steps/extract.py accumulates the entire stream into a list before transform/load. It is fine only for bounded micro-batches.

Thesis: the foundation is right; the work is wiring it through, not rebuilding. The streaming validation engine — not the batch DAG — is the home for live data.

Usage model (vision)

Two complementary modes, both first-class:

  • Library mode — import ContractStreamProcessor (or pycharter.kafka.consume(...)) inside your own consumer process. You own the loop and lifecycle; pycharter validates/normalizes/routes per message. Best when streaming is one part of a larger service.
  • Worker mode — pycharter stream run worker.yaml runs pycharter's own long-running stream processor(s), one per stream, with health/metrics endpoints. Best when validation/normalization is the service, and for the UI to drive (below). After ST-1 this mode accepts any source type.

Testing a contract against a stream from the UI (ST-8) requires pycharter to run its own bounded stream processor on demand: the UI asks the API to tap a source for N messages (or a pasted sample) in detect/shadow mode and returns per-message pass/fail, the coerced/normalized output, and DLQ reasons — no data is committed downstream. This is a bounded, dry-run invocation of the same engine (max_records + dry_run=True already exist on the processor), so it reuses ST-1's source factory rather than adding a parallel path.

Config alignment with the batch pipeline (target shape) — ST-14

Today the two engines specify contracts differently, which is needless cognitive load:

Concern Batch step (PipelineConfig) Streaming today (StreamConfig)
Which contract + where contract.ref: {type: store, name, version} or {type: file, path} contract_resolution.static: {contract_id, version} + source.contract_dir (file only)
Validate / coerce / annotate / quality explicit booleans on contract: implicit (always validate+coerce; no toggles)
Failure routing contract.on_error: fail \| warn \| skip \| quarantine mode: detect \| shadow \| enforce + DLQ
Per-message contract from headers n/a (batch is one contract per step) contract_resolution.primary: header

Target: a batch-style contract: block in the stream config, so the "contract specification" reads the same in both engines, with only the genuinely streaming-specific knobs added:

streaming:
  streams:
    - name: trade-orders
      source: { type: kafka, topic: orders.raw, bootstrap_servers: ${KAFKA_BROKERS} }

      # Same shape as a batch step's `contract:` block …
      contract:
        ref: { type: store, name: trade_order, version: "1.0.0" }   # or {type: file, path: ...}
        validate: true
        coerce: true
        annotate: false
        # … plus streaming-only contract selection (no batch analog):
        resolution: static            # static | header  (header = per-message pycharter-contract)
        on_missing_header: fallback_to_static

      # Same operators as a batch transform step (ST-2)
      transform:
        rename: { p: price, s: size }
        convert: { price: float }
        add: { ingested_at: "now()" }

      # Same `target` shape as a batch load step (ST-3); DLQ is the streaming addition
      load: { target: { type: kafka, topic: orders.validated, bootstrap_servers: ${KAFKA_BROKERS} } }
      dlq:  { target: { type: kafka, topic: orders.dlq,        bootstrap_servers: ${KAFKA_BROKERS} } }

      delivery: at_least_once          # streaming-only execution knobs stay
      mode: enforce                    # detect | shadow | enforce
      backpressure: { max_in_flight: 1000 }

Mapping / decisions: - contract.ref replaces contract_resolution.static + source.contract_dir and becomes byte-identical to the batch ref. ref.type: store depends on ST-12 (worker reads the contract store); ref.type: file works today. - validate / coerce / annotate come along for free → parity with batch (the engine already coerces; annotate emits field_bindings like batch). - resolution / on_missing_header are folded onto the ref — the one honestly streaming-specific part (per-message header selection). - mode stays (rollout safety: shadow writes to both sinks — no batch analog) rather than collapsing into on_error; keeping both would be redundant. enforce ≈ on_error: quarantine, detect ≈ warn. - transform: is ST-2; load/dlq target: is ST-3 (this is also why valid_sink/dlq_sink become load.target/dlq.target).

This is a breaking config change that subsumes the surface of ST-2/ST-3/ST-12; sequence it after those land (or land it as the umbrella that delivers them).

Tier 1 — finish the wiring (near-term)

Build order (revised 2026-06-03): ST-1 → ST-4 → ST-2 → ST-3 → ST-12 → ST-14. ST-14 (the batch-aligned contract: / transform: / load shape) is the north star: each earlier item lands an increment already shaped for it, so ST-14 is a config-key flip over machinery that already exists, not a rewrite. Concretely:

  • ST-1 introduces a generic, type-keyed source factory and sink factory that take a plain {type, ...} dict — the same shape ST-14's load.target / dlq.target will carry. It keeps today's valid_sink / dlq_sink keys (kafka byte-identical) but routes them through the factory, so ST-3 extends the sink factory and ST-14 only renames the keys.
  • ST-2 adds transform: using the batch operators — already ST-14's block.
  • ST-3 registers storage sinks behind ST-1's sink factory (→ load.target).
  • ST-12 makes contract.ref: {type: store} resolvable (store-backed validator).
  • ST-14 flips the config keys (contract_resolution + source.contract_dir → contract: block; *_sink → load/dlq target:) over the now-existing machinery and rejects the legacy keys with a migration message.

Do the two small, isolated wins (ST-1, ST-4) first; ST-2's DLQ design and ST-3's buffer-vs-commit tension are the harder items. Together they unblock websocket|kafka → validate → normalize → land via config.

ST-1 — Generic source/sink factory for the worker (high, medium)

Replace the hardcoded KafkaExtractor + Kafka sinks in _cli_streaming.py::_build_processor with factories driven by stream.source["type"] / sink type.

  • Reuse, don't rebuild: a source factory already exists — pipeline_generator/extractors/factory.py::ExtractorFactory.create({...}) resolves websocket / sse / kafka / … and lazy-imports behind the right extra. ST-1's source half is "call ExtractorFactory.create(source)" plus two adapters: normalize the worker's source["topic"] (singular) → topics: [...], and strip the worker-only contract_dir before passing the dict through.
  • source.type becomes authoritative (the worker ignores it today).
  • The sink half is not free: only Kafka sinks (+ IteratorSink / MemoryDLQSink) implement the StreamSink.write(...) surface today, so ST-1's sink factory wires only kafka + memory/iterator; real storage sinks are ST-3. Factory-built sources/sinks must be wired into the existing closer().
  • StreamConfig already carries generic dicts; ContractStreamProcessor is unchanged.
  • Acceptance: parametrized over {kafka, websocket, sse} with a mock source — each runs end-to-end; the kafka path is byte-identical to today; unknown type → clear error naming supported types and the missing extra.

ST-4 — Wire idle_timeout + liveness for websocket/SSE (medium, low)

Wrap the recv in the websocket.py / sse.py read loops with asyncio.wait_for(..., timeout=idle_timeout); on TimeoutError → log + break to the existing reconnect path and flip a liveness signal.

  • More than wiring an existing field: idle_timeout lives in the internal StreamingConfig dataclass but is not on WebSocketExtractConfig / SSEExtractConfig (they expose ping_interval, not idle_timeout). ST-4 must add idle_timeout to those Pydantic models and plumb it through from_config.
  • Default None (off) to avoid reconnect storms during legitimately quiet periods; document that idle_timeout (no business message) is distinct from ping_interval (protocol keepalive).
  • Acceptance: a mock socket silent past idle_timeout triggers exactly one reconnect (or flips /readyz-visible health); no reconnect while traffic flows.

ST-2 — Optional transform/normalization stage in a stream (high, medium)

Add an optional transform: block to StreamConfig, applied per message before validation so the contract validates the reshaped record, reusing the batch transform operators (pipeline_generator.transformers.apply_transforms + _build_transform_config: rename / convert / add / select / drop). Insertion point: _processor.py::_process_one, after contract/validator lookup, before _validate_and_route (which currently calls validator(message.payload)).

  • apply_transforms is list/batch-oriented (transformers/pipeline.py) — wrap the single payload as [payload] and take [0]; confirm no cross-record assumptions.
  • The real work is the error path: a transform failure must route the original payload to the DLQ as a new Outcome / DLQReason.TRANSFORM_ERROR, not crash the loop — and feed the observer outcomes.
  • Validate the transform block with the existing TransformStepConfig (don't invent a second shape). Order is transform (rename/derive) → contract coercion (types) → valid sink; state this explicitly in docs/examples.
  • Acceptance: transform.rename: {p: price} yields price on the valid topic; a throwing transform routes to DLQ with TRANSFORM_ERROR; absent block = today's behavior byte-for-byte.

ST-3 — Stream sinks to storage (high, large — re-scoped from "medium")

A StreamSink factory so a stream can terminate into file (append/JSONL), Postgres, or cloud, plus an optional Kafka producer sink for re-emit.

  • Not a thin adapter. Batch loaders are bulk (load(list)); StreamSink is per-record. A storage stream-sink needs an internal micro-batch buffer (flush on size/interval + flush-on-close) — one INSERT per record is unusable.
  • Buffer-vs-at_least_once tension (the crux). The engine commits the offset after write(...) returns; if write buffers and returns before a durable flush, a committed offset can outrun persisted data → loss on crash. Resolve by flush-before-return, or by making commit flush-aware (which the per-message engine does not model today). First cut: append-only file sink / at_most_once, where the tension is benign; defer durable buffered DB sinks until commit semantics are decided (candidate to move to Tier 2).
  • Acceptance: a file/jsonl valid sink appends in bounded batches and flushes on shutdown; the exact at-least-once guarantee (and where it weakens) is documented; backpressure still throttles intake when the sink is slow.

Tier 2 — real features (next)

  • ST-5 — Windowing / aggregation reducer (high, large). Implement the QualityConfig.windows placeholder ("reducer wired in Phase 4"): tumbling / sliding windows for OHLC bars, VWAP, count/last-per-key. This is what turns pycharter from validating passthrough into a stream processor.
  • ST-6 — Batch-DAG unbounded-source guard + incremental load (medium). Make run_extract refuse an unbounded streaming source unless max_records / max_batches / shutdown_event is set; optionally add flush-per-batch incremental load so micro-batch windows don't materialize the whole stream.
  • ST-7 — Exactly-once / dedup sinks (medium-large). Idempotent or transactional producing (dedup key or Kafka transactions) for correctness-sensitive consumers.
  • ST-12 — Worker contract source from the contract store (medium). _cli_streaming.py::_factory_from_source today requires source.contract_dir and uses make_validator_factory_from_dir, raising if absent ("contract_store wiring lands in a follow-up"). Dir-mode reuses one validator for every ContractRef, so contract_resolution.primary: header (per-message multi-contract topics) cannot actually resolve different contracts in worker mode. Add a store-backed factory (make_validator_factory_from_store) selectable via config (e.g. source.contract_store / a worker-level store block) that resolves each ContractRef from the registry.
  • Acceptance: a stream pointed at a contract store with contract_resolution.primary: header validates each message against the contract named in its pycharter-contract header; contract_dir still works.
  • ST-14 — Batch-aligned contract: / transform: / load config shape (medium, breaking). Converge the stream config's contract specification onto the batch step's contract: block (ref + validate/coerce/annotate), folding resolution/on_missing_header onto the ref and replacing contract_resolution + source.contract_dir; surface transform: (ST-2) and load/dlq target: (ST-3, replacing valid_sink/dlq_sink). See the "Config alignment with the batch pipeline" section above. Subsumes the config surface of ST-2/ST-3/ST-12; land after or as their umbrella.
  • Acceptance: a stream authored with a batch-style contract.ref: {type: store, name, version} + transform: + load.target: runs end-to-end; the legacy contract_resolution / *_sink keys are rejected with a migration message.

Tier 3 — ergonomics & adoption

  • ST-8 — UI-driven streaming contract test (bounded dry-run). API endpoint
  • UI surface that taps a source for N messages (or a pasted sample) in detect/shadow + dry_run, returning per-message validation results, coerced output, and DLQ reasons. Depends on ST-1.
  • ST-9 — Streaming templates + pycharter stream init scaffold. No extract_websocket*.yaml / streaming-worker templates exist today.
  • ST-10 — Finish observability + sequence-gap awareness. Wire the Prometheus metrics/health endpoints the worker config already declares (lag, DLQ rate, validate latency, idle); carry source seq numbers in StreamMessage.source_metadata and expose gap detection after reconnect.
  • ST-11 — Contract-from-sample scaffolder. Generate a contract (schema + coercion_rules) from a sample message to onboard a feed quickly.
  • ST-13 — Env-var interpolation in the stream run config (low). _cli_streaming.py::_load_config does plain yaml.safe_load with no ${VAR} expansion (batch parse_pipeline_config does interpolate), so brokers/creds must be templated at deploy time. Add opt-in ${VAR} / ${VAR:-default} / ${VAR:?required} resolution (reuse the batch resolver) so containerized workers read config from the environment.
  • Acceptance: bootstrap_servers: ${KAFKA_BROKERS} resolves from env; a missing ${VAR:?msg} fails fast with a clear message; literal configs are unchanged.

Non-goals

  • Rewriting the batch DAG into a streaming engine (ST-6 fences it off instead).
  • A bespoke windowing DSL beyond the common market-data aggregates (ST-5).
  • Coupling pycharter to any specific venue/client SDK; the thin websocket→Kafka bridge stays the consumer's responsibility.