$ dtp cat architecture/README.md

DTP Architecture
A Newbie's Guided Tour

A walkthrough of the framework, the extension boundary,
and how data flows from source to semantic search.

core framework extension domain

batch + streaming ETL → embeddings → Qdrant

architecture/overview/what-dtp-does

What DTP does in one breath

Ingests product records from MongoDB + Kafka, transforms via idempotent handler chain, makes sparse (BM25) + dense (OpenAI) embeddings, upserts vectors to Qdrant for semantic search.

SOURCE

MongoDB

SOURCE

Kafka

→

TRANSFORM

8-Stage Chain

→

EMBED

Sparse + Dense

→

STORE

Qdrant

Every record passes through the same deterministic pipeline — batch or stream.

architecture/overview/two-problems

The two problems it solves

PROBLEM 1 Data at Scale

Processing millions of product records requires a pipeline that is idempotent, resumable, and crash-safe.

If a worker dies mid-chain, the system must pick up exactly where it left off — no duplicates, no orphans.

Idempotent Resumable Crash-safe

# 8 stages, each atomic guard → normalize → validate → dedup → enrich → embed → write → status

PROBLEM 2 N Concurrent Workers

Multiple consumers compete for jobs. The system needs distributed locks, atomic claims, and cursor-based reconciliation.

No two workers should process the same record. Stale claims must be detected and re-queued automatically.

Locks Atomic Claims Cursors Reconciliation

# Worker claims job atomically UPDATE transfer_jobs SET worker_id = 'w-3' WHERE id = 42 AND state = 'PENDING'

architecture/core-truth

The core truth

If you remember one thing from this tour, remember this.

Core is a framework with ZERO business logic.

Extensions contain ALL domain work.

Core

Chain, Handler, Contract, Router
Run, Process, Connector
Pure framework — no domain knowledge

Extension

Product handlers, Enrichment, Embedding
Qdrant writes, Dedup rules
All business logic lives here

architecture/flows/batch

End-to-end batch flow

1

Trigger fires

→

2

Claim run
(atomic)

→

3

Producer
(fetch Mongo)

→

4

Publish jobs
(RabbitMQ)

→

5

Consumer
(pick job)

→

6

Run chain

→

7

Qdrant +
PG + Redis

→

8

Ack +
Finalize

▼ The 8-stage handler chain inside step 6

guard→ normalize→ validate→ dedup→ enrich→ embed→ write→ status

Key insight

The PUBLISHING phase (step 4) enqueues all jobs before any consumer starts. This means progress tracking is honest: completed ≥ published is a real invariant, not a guess. Each consumer picks a job, runs the full chain, writes results, then acks. The run finalizes only when all jobs are acked.

architecture/flows/stream

End-to-end stream flow

SOURCE

Kafka

→

CONSUME

Stream Consumer

→

ROUTE

Router
(picks chain)

→

TRANSFORM

8-Stage Chain

→

STORE

Qdrant

→

FINALIZE

Commit Offset

How it differs from batch

No RabbitMQ — Kafka already buffers messages in partitions. No need for a separate job queue.
No run finalization — there is no "run" concept. Each message is independently processed and offset-committed.
LISTENING state — the process stays in LISTENING indefinitely until force-completed by an operator.
Offset = cursor — Kafka's consumer offset acts as the resume point. Crash-safe by design.

Stream process lifecycle

# Stream process states PENDING → LISTENING → COMPLETED (force) ↘ FAILED # The consumer loop for message in kafka_consumer: chain = router.select(message) state = chain.run(message) kafka_consumer.commit(message.offset) # No batch finalization — each message # is self-contained.

architecture/runtime/components

Runtime components

POSTGRES Source of Truth

Stores runs, processes, transfer jobs, and flow state. Every state transition is a row update. If Postgres is down, nothing moves.

REDIS Dedup L1, Locks, Signals

Dedup L1: content-hash cache to skip unchanged records fast. Locks: distributed run/worker locks. Signals: pub/sub for cancellation and coordination.

RABBITMQ Batch Job Queue

Producer publishes one message per record during PUBLISHING phase. Consumers compete for jobs. Ack on success, nack on failure. Durable queues survive broker restarts.

KAFKA Streaming Source

Topic-partitioned event log. Stream consumers read from assigned partitions. Offsets act as cursors — crash-safe resume without external state.

QDRANT Vector Database

Stores sparse (BM25) + dense (OpenAI) vectors per product. Upserts are idempotent by point ID. Payload includes product metadata for filtered search.

MONGODB Source Data

Raw product catalog. Producer queries Mongo with cursor-based pagination to fetch records in batches. Read-only from DTP's perspective — never writes back.

architecture/glossary

Glossary

Core Concepts (1/2)

Term	Definition
run	A top-level execution: batch or stream. Has a state machine and lifecycle.
flow	Orchestrates a run — creates it, drives state transitions, handles finalization.
process	A long-lived worker instance that claims and executes runs or stream listeners.
connector	Adapts an external system (Mongo, Kafka, Qdrant) into the framework's interface.
chain	An ordered sequence of handlers that transform a single record end-to-end.
handler	One stage in a chain. Receives ChainState, does work, returns updated state.

Core Concepts (2/2)

Term	Definition
contract	Declares which handlers a chain must include — required, first, last, unique constraints.
router	Selects which chain to use for a given message based on operation type or metadata.
dedup L1	Fast content-hash check in Redis. Skips records whose content hasn't changed.
dedup L2	Version-based check in the handler chain. Compares source version vs. last-processed version.
cursor	Bookmark for pagination (Mongo) or offset (Kafka). Enables crash-safe resume.
point ID	Deterministic Qdrant vector ID derived from record_id. Makes upserts idempotent.

Part One

CORE — The Framework

The abstractions that make every extension possible.

● CORE (active) ○ EXTENSION

core/01-chain core/02-handler core/03-contract core/04-router core/05-run

core/01-chain/abstraction

The chain abstraction

Explanation

A Chain is an ordered list of BaseHandlers that share a single ChainState object. The state threads through every handler — each one reads, mutates, and passes it forward.

Chain.of() assembles the chain from handler providers. Chain.run() executes them sequentially, catching errors at each stage.

The @handler_provider decorator marks extension classes as chain participants. The core never knows what the handlers do — it only knows the interface.

ChainState is the single thread that connects all 8 stages.

# Extension assembles the chain class ProductChain: chain = Chain.of( "guard", # 1. Pre-flight checks "normalize", # 2. Clean + standardize "validate", # 3. Schema enforcement "dedup", # 4. Skip unchanged "enrich", # 5. Add metadata "embed", # 6. Sparse + dense vectors "write", # 7. Upsert to Qdrant "status", # 8. Update job state ) # Each handler is a decorated class @handler_provider(name="embed") class ProductEmbedHandler(BaseHandler): def process(self, state: ChainState): state.dense_vectors = embed_openai(state.embedding_text) state.sparse_vectors = embed_bm25(state.embedding_text) return state

core/02-handler/control-flow

How a handler runs

handler.process(state)

▼

SUCCESS

Handler returns updated state.
→ Next handler in chain continues.

▼

DEDUP / STALE

DedupSkipError or StaleVersionError raised.
→ Skip to StatusHandler (last).

▼

PERMANENT

PermanentError raised.
→ Re-raise to consumer.
→ Job marked FAILED.

Error type summary

Error	Meaning	Chain behavior
DedupSkipError	Record content unchanged (L1 or L2 dedup)	SKIP jump to StatusHandler
StaleVersionError	Source version ≤ last processed version	SKIP jump to StatusHandler
PermanentError	Unrecoverable — bad data, missing config	FAIL re-raise, job fails
TransientError	Retryable — network timeout, rate limit	RETRY consumer re-queues

core/03-contract+04-router

Contracts & routers

CONTRACT ChainContract

A contract declares the structural rules every chain must satisfy. The core validates chains at assembly time — before any record is processed.

required Handlers that must be present (e.g., dedup, embed)
first The handler that must run first (guard)
last The handler that must run last (status)
unique Handlers that cannot appear twice

# Contract definition contract = ChainContract( first="guard", last="status", required=["dedup", "embed", "write"], unique=["guard", "status"], )

ROUTER ChainRouter

A router selects which chain to execute for a given message. Different operations (create, update, delete) may use different chains.

The StandardChainRouter maps operation types to chain instances via a simple dictionary lookup.

Extensions define the mapping. The core only provides the interface.

# Router maps operations to chains class ProductChainRouter(ChainRouter): operation_chains = { "upsert": ProductUpsertChain, "delete": ProductDeleteChain, "reindex": ProductReindexChain, } def select(self, message): op = message.headers.get("operation") return self.operation_chains[op]

core/data-envelope

The data envelope

RecordData fields

Field	Filled by
record_id	Producer (from source)
content_hash	NormalizeHandler
embedding_text	NormalizeHandler
chunks	EnrichHandler
sparse_vectors	EmbedHandler
dense_vectors	EmbedHandler
qdrant_payload	EnrichHandler
record_state	StatusHandler

# TransferJob — the unit of work { "id": 42, "run_id": "run-2024-0301", "record_id": "prod-abc-123", "state": "PROCESSING", "worker_id": "w-3", "attempts": 1, "max_retries": 3, "error": null, "created_at": "2024-03-01T10:00:00Z", "updated_at": "2024-03-01T10:00:12Z" } # ChainState threads RecordData through handlers state = ChainState( record=RecordData(record_id="prod-abc-123"), job=transfer_job, context=run_context, )

Key insight

Each handler fills specific fields and ignores the rest. Handlers are decoupled — embed doesn't know about normalize, write doesn't know about dedup. The ChainState envelope is the only shared interface.

core/05-run/state-machines

Run state machines

BATCH Run States

PENDING → PUBLISHING → PROCESSING → COMPLETED

or from any active state → FAILED

PENDING Run created, waiting for a worker to claim it.
PUBLISHING Producer is fetching records and enqueuing jobs to RabbitMQ. No consumers yet.
PROCESSING All jobs published. Consumers are picking up and running chains.
COMPLETED All jobs acked. _try_finalize_run confirms done.
FAILED Unrecoverable error or too many retries.

STREAM Run States

PENDING → LISTENING → COMPLETED

or from any state → FAILED

PENDING Stream process created, waiting for a worker to claim it.
LISTENING Consumer is reading from Kafka partitions. Stays here indefinitely — processing messages as they arrive.
COMPLETED Only via force-complete by an operator. No natural end — streams are infinite.
FAILED Consumer crashed or Kafka connection lost beyond retry.

No PUBLISHING phase — Kafka already buffers.

core/05-run/publishing-phase

Why a separate PUBLISHING phase

◆ Aha moment — this is the design decision that makes progress tracking honest.

The Producer publishes ALL jobs before PROCESSING begins, so progress tracking (terminal ≥ published) is an honest invariant — not a guess.

WITHOUT No PUBLISHING phase

If producing and consuming happen simultaneously, you cannot know total job count ahead of time.

Progress "5 of ?" — denominator unknown
Completion Can't tell if producer is done or just slow
UI Progress bar is a guess, not a fact
Race Consumer may finish before producer enqueues all jobs

Progress is a guess. Completion is a hope.

WITH PUBLISHING phase

Producer publishes all N jobs, then state transitions to PROCESSING. Now total is known.

Progress "5 of 1,247" — exact denominator
Completion _try_finalize_run checks: terminal_count ≥ published_count
UI Progress bar is a fact
Invariant When all published jobs are terminal → run is done

# Finalization check if terminal_count >= published_count: run.state = "COMPLETED" # Honest!

docs/core/03-ingestion-workers / one flow → many workers

One flow → many workers

Horizontal scaling without coordination — Postgres does the locking.

Explanation

Each flow can be consumed by multiple workers simultaneously. The claim mechanism is atomic at the database level — no distributed lock service needed.

claim_pending_run() uses SELECT ... FOR UPDATE SKIP LOCKED to ensure exactly one worker picks up each batch. Workers that lose the race simply try the next available batch.

Worker lifecycle

Try claim_pending_run()

↓

Postgres atomic lock (SKIP LOCKED)

↓

Fetch batch of records

↓

Publish jobs to queue

↓

Heartbeat every 30s

↓

Next batch → repeat

Three workers compete — one wins the lock, two move on

docs/core/03-ingestion-workers / backpressure & the consumer

Backpressure & the consumer

Explanation — prefetch_count=1

The consumer processes one message at a time. With prefetch_count=1, RabbitMQ won't dispatch the next job until the current one is acked.

Consumer loop:

Receive message from queue
Deserialize & validate payload
Check dedup index
Route to handler chain
Execute chain (with timeout)
Ack or nack based on outcome
Emit metrics & logs

Example — Error classification

Error raised in handler

↓

TransientError

PermanentError

→ Retry (nack + requeue)

→ DLQ (dead letter)

Skip conditions (ack without processing):

DedupSkip StaleVersion RawRecordInvalid

Backpressure is inherent — slow consumer = queue builds up naturally

docs/core/03-ingestion-workers / the supervisor reconciles

Aha — Declarative flow management

The supervisor reconciles

"desired_flows (code-enabled AND DB-active AND org-active) minus running = start/stop. Redis control signals, not timers."

Desired set

Intersection of three conditions:

✓ Code-enabled in config

✓ DB flow marked active

✓ Org not quarantined

All three must be true for a flow to be "desired."

Minus running

Compare desired set against currently running flow instances.

desired − running = to_start

running − desired = to_stop

Pure set arithmetic. No state machines.

= Action

Publish Redis control signals:

START → spawn worker task

STOP → graceful shutdown

Signals, not timers. Workers react immediately.

Reconciliation runs on a short interval — eventual consistency for flow lifecycle

docs/core/03-ingestion-workers / stream consumer path

Stream consumer path

Kafka streams follow the same ack/commit discipline as batch flows.

Read Kafka

→

Route

→

Run chain (w/ timeout)

→

Commit offset

Explanation — commit_fn

Offsets are committed only after successful processing. If the chain fails with a transient error, the offset is not committed — Kafka will redeliver on reconnect.

The commit_fn callback is passed into the handler chain, giving handlers control over exactly when the offset advances.

Example — Heartbeat monitor

A background task sends heartbeats every 30s to signal liveness. If heartbeats stop:

90s without heartbeat → instance marked dead

Reaper resets → uncommitted offsets redelivered

Same mechanism as batch workers — unified failure detection.

At-least-once delivery guaranteed by deferred offset commits

docs/core/04-triggers / distributed scheduling

Distributed scheduling, no central scheduler

Redis SET NX as a lightweight leader election — every worker is a candidate.

Explanation — set_nx race

Three workers all attempt redis.set_nx(lock_key, ex=3600) at the same time. Exactly one wins, two lose. The winner executes the trigger; losers back off.

No ZooKeeper, no Raft, no external scheduler. Just Redis atomicity.

# Trigger acquisition acquired = await redis.set_nx( lock_key, value=instance_id, ex=3600 ) if acquired: await execute_trigger()

Example — Trigger types

Type	Behavior
Cron	Fires at schedule (e.g., 0 2 * * *). Lock expires after run.
Interval	Fires every N seconds. Lock held for duration, then released.
Startup	Fires once on boot with random jitter (0–60s) to spread load.

Startup jitter prevents thundering herd when multiple workers boot simultaneously.

Lock TTL (3600s) prevents zombie locks if a worker dies mid-trigger

docs/core/04-triggers / hold the lock

Aha — Lock semantics matter

Hold the lock for the whole run

"Release on completion, not start, else duplicate runs."

Acquire lock

→

Create run

→

Run pipeline

→

on_run_completed() release

Explanation

The lock is held for the entire duration of the pipeline run, not just the trigger moment. This prevents a second worker from starting a duplicate run while the first is still processing.

Release happens in on_run_completed() — a callback that fires on both success and failure.

Example — Cursor inheritance

When a run completes, its cursor position is stored. The next run inherits this cursor and continues from where the last one left off.

cursor = last processed offset/timestamp

next_run starts at cursor + 1

Prevents re-processing and gaps between consecutive runs.

Lock release on completion — not on start — is what prevents duplicate runs

docs/core/05-connectors / connector declarations

Aha — Connectors are metadata until instantiation

Connectors are declarations, not connections

Extension registers ConnectorDeclaration

→

Linter validates at boot

→

Worker instantiates concrete adapters

MONGO_BATCH

Type: Source (batch)

Adapter: MongoBatchSource

Config: uri, db, collection

Declaration says "I need MongoDB." Worker creates the actual client at runtime.

KAFKA_STREAM

Type: Source (stream)

Adapter: KafkaStreamSource

Config: brokers, topic, group

Same pattern — declare intent, instantiate later.

QDRANT_SINK

Type: Sink

Adapter: QdrantSinkAdapter

Config: url, collection, dim

Sinks follow the same declaration pattern as sources.

Declarations enable boot-time validation — catch misconfig before any data moves

docs/core/05-connectors / the I/O boundary

The I/O boundary

Layered architecture — connectors sit at the edge, repos and stores below.

Layer 1 — Connectors

BaseSourceBatchConnector.fetch_batch() BaseStreamConnector.consume() BaseSinkConnector.write_batch()

↓

Layer 2 — Infra Clients

Postgres Redis Kafka MongoDB Qdrant S3

↓

Layer 3 — Repositories

FlowRunRepo RecordStateRepo DedupIndexRepo CheckpointRepo

↓

Layer 4 — Stores

EventStore BlobStore VectorStore

Connectors never touch raw drivers — they go through infra clients for connection pooling and retry

Part Two

CORE — Keeping it Alive

core/06-bootstrap core/07-recovery core/08-health core/09-observability

docs/core/06-bootstrap / boot sequence

Boot sequence

Six ordered steps from cold start to fully operational.

Explanation — Ordered startup

Boot is deterministic and ordered. Each step depends on the previous. If any step fails, the process exits — no partial startup, no zombie state.

The sequence ensures that by the time workers start consuming, all configuration has been validated and all state has been reconciled.

Example — 6-step timeline

1

Load extensions — discover and register all connector/handler plugins

2

Build registries + compute config hashes for change detection

3

Run startup linter — 7 validation checks (see next slide)

4

Reconcile flows to DB — sync desired state with persisted state

5

Register services + hooks — wire up lifecycle callbacks

6

Start supervisor / consumers / triggers — begin processing

Fail-fast boot — better to crash at step 3 than discover the problem at step 6

docs/core/06-bootstrap / quarantine bad orgs

Aha — Blast radius containment

Quarantine bad orgs

"One org's error isolated to failed_orgs. Cluster keeps running. All-fail → ConfigError."

Org A

✓ healthy

Org B

✗ quarantined

Org C

✓ healthy

Org D

✓ healthy

Explanation

During boot, each org's config is validated independently. A failing org is added to failed_orgs set — its flows are skipped, but all other orgs proceed normally.

Example — Escalation

If all orgs fail, it's not an org problem — it's a ConfigError. The system escalates: boot aborts with a clear error pointing to the shared config.

Partial failure ≠ total failure. Quarantine keeps the cluster alive.

docs/core/06-bootstrap / startup linter

The startup linter (7 checks)

Catch configuration errors at boot — before any data is processed.

Checklist

Org domains — every org has a valid domain configured
Connector refs — all referenced connectors exist in registry
Chain refs + routers — handler chains resolve correctly
Handler registration — all handlers are importable and registered
Contracts — input/output schemas match between chain steps
Source invariant — each flow has exactly one source
Directory/org match — file layout matches org structure

Example — Linter output

# StartupLinter results ✓ org_domains: 4 orgs validated ✓ connector_refs: all 12 refs resolved ✓ chain_refs: 8 chains valid ✗ contracts: embedding_chain expects dim=768, sink declares dim=1536 ✓ source_invariant: 1 source per flow ✓ directory_match: layout consistent

class StartupLinter: def run(self, config): errors = self.check_all(config) if errors: raise SystemExit(1)

Fail fast at boot — a linter error is better than a silent data corruption at runtime

docs/core/07-recovery / three timers

Three timers for three failures

Each timeout addresses a specific failure mode with a tailored recovery.

90s

Dead instance → reap

No heartbeat for 90 seconds means the worker process is gone. The reaper cleans up its state and reassigns work.

Detection: heartbeat timeout

300s

Stale run → reset

A run that hasn't progressed in 5 minutes is stuck. Reset it to PENDING so another worker can pick it up.

Detection: last_progress timeout

24h

Max processing → fail

No run should take more than 24 hours. If it does, mark it FAILED — something is fundamentally wrong.

Detection: created_at timeout

Heartbeat sent every 30s — three missed heartbeats = 90s dead threshold

Layered timeouts — fast detection for common failures, slow escalation for edge cases

docs/core/07-recovery / kill -9 recovery

Life of a kill -9'd worker

What happens when a worker process is terminated without graceful shutdown.

Worker killed

→

Heartbeat lapses (90s)

→

Reaper detects

→

Reset runs + delete instance

→

Control signal

→

Live worker claims

Explanation — Reset logic

When the reaper finds a dead instance, it resets all its active runs:

Default: reset to PENDING — another worker will pick it up.

Exception: if restart_count ≥ 5, mark as FAILED. The record has been retried too many times — stop the loop.

Example — Restart ceiling

Restarts	Action	Status
0–4	Reset to pending, requeue	PENDING
5	Restart ceiling reached	FAILED

The ceiling prevents infinite retry loops on permanently broken records. Each reset increments restart_count atomically.

Total recovery time: ~90s (heartbeat timeout) + reaper cycle (~10s) ≈ 100s worst case

docs/core/07-recovery / retention purge

5-pass retention purge

Foreign key ordering — delete children before parents, always.

Explanation — FK safety ordering

The purge runs in exactly 5 passes, ordered by foreign key dependency. Deleting a parent before its children would violate constraints and crash the purge.

Each pass deletes records older than the retention window, working from leaf tables up to the root flow_runs table.

Example — Pass order

1

failure_events — leaf table, no dependents. Safe to delete first.

2

record_states — references flow_runs. Must go before parent.

3

run_checkpoints — references flow_runs. Must go before parent.

4

dedup_index — references flow_runs. Must go before parent.

5

flow_runs — root table. Mark as purged (soft delete) after all children are gone.

Pass 5 uses soft delete (mark purged) — flow_runs are the audit trail. Hard delete happens in a separate archival job.

Each pass is a single DELETE with a WHERE clause on created_at — batch-friendly, lock-minimal

docs/core/08-services-api/

The Control Plane

How API calls become worker actions through database-first coordination

API Call

→

Service writes
to Postgres

→

Publish Redis
control signal

→

Workers
reconcile

Explanation

The API layer never talks directly to workers. Instead it writes the desired state to Postgres and publishes a lightweight signal to Redis. Workers poll the database, compare desired vs actual state, and reconcile the difference. This means Redis signals are fire-and-forget — if one is missed, the next reconciliation loop picks it up.

Example

# POST /flows/{id}/pause def pause_flow(flow_id): db.update(flow_id, status="PAUSED") # source of truth redis.publish(f"ctrl:{flow_id}", "pause") # Worker sees PAUSED in DB → stops processing # Even if Redis msg lost, next poll catches it

Key Insight

Database-first = source of truth. Redis signals are notifications, not state.

docs/core/08-services-api/

Force Start with Dedup Clear

Full reprocessing by wiping caches and resetting cursor state

Wipe L2
delete_by_flow

+

Wipe L1
delete_prefix

→

Create PENDING
no cursor

→

Full
Reprocess

Explanation

A force start clears both dedup layers: L2 (Postgres processed_records via delete_by_flow) and L1 (Redis seen-key cache via delete_prefix). It then creates a new run with no cursor, meaning the source connector starts from the very beginning. Every document is re-fetched and re-processed as if it were new.

Example

# POST /flows/{id}/force-start def force_start(flow_id): dedup.delete_by_flow(flow_id) # L2: Postgres cache.delete_prefix(f"seen:{flow_id}") # L1: Redis run = create_run(flow_id, cursor=None) # cursor=None → Mongo starts from beginning # All docs re-enter the pipeline fresh

Key Insight

No cursor = start from beginning. Use when embeddings model changes or schema migration requires full rebuild.

docs/core/09-failure-dlq/

Error Classification

Every failure maps to exactly one of four outcomes

Error Occurs

↓

TransientError
Retry + backoff

PermanentError
→ DLQ now

DedupSkip / StaleVersion
Control signals

RawRecordInvalid
Skip entirely

Explanation

The error classifier wraps every chain stage. TransientError (network timeout, rate limit) triggers exponential backoff retry. PermanentError (schema mismatch, auth revoked) sends the record to DLQ immediately. DedupSkip/StaleVersion are control signals — the record was already processed or superseded. RawRecordInvalid means the source data itself is malformed — skip and log.

Example

try: result = chain.process(record) except TransientError: retry_queue.push(record, backoff=2**attempt) except PermanentError as e: dlq.store(record, error=str(e)) # → DLQ except DedupSkip: logger.info("already processed") # skip except RawRecordInvalid: logger.warn("malformed source") # skip

docs/core/09-failure-dlq/

DLQ Lifecycle

Dead Letter Queue state machine with replay and escalation

pending

→

replaying

→

replayed

re-fails → pending
relay_count++

→

relay_count ≥ 3
dead

|

manual discard

Explanation

Failed records enter the DLQ as pending. When replay is triggered, they move to replaying and are republished to the pipeline. If processing succeeds → replayed. If it fails again → back to pending with relay_count incremented. After 3 failed replays, the record is marked dead and requires manual intervention or discard.

Example

State	relay_count	Action
pending	0	Awaiting replay
replaying	0	Republished to chain
pending	1	Re-failed, retry
pending	2	Re-failed again
dead	3	Manual intervention

docs/core/09-failure-dlq/

Publish-before-mark = crash never loses a job

Crash-Safe Replay

Ordering guarantees that survive process crashes

— DLQService design principle

→

Publish jobs
to pipeline

→

Mark replayed
update state

Explanation

The replay process claims DLQ rows with FOR UPDATE SKIP LOCKED to prevent concurrent replays. Jobs are published to the pipeline before the database state is updated. If the process crashes after publish but before mark, the next replay attempt will skip already-claimed rows. The job is already in the pipeline — no data loss.

Example

class DLQService: def replay_all(flow_id): rows = db.query("SELECT ... FOR UPDATE SKIP LOCKED") for row in rows: job_id = f"replay:{row.event_id}" pipeline.publish(job_id, row.payload) # Mark replayed AFTER all publishes db.update(rows, status="replayed")

Part Three

EXTENSION — Config becomes a Pipeline

FOUNDATION

INGESTION

CORE

EXTENSION

ext/01 — Process & Config

docs/extension/01-process-config/

A Process = a Pipeline Template

Named templates bundle connectors, chains, and flows into reusable configurations

Template Components

Source Connectors — MongoDB, Kafka
Sink Connectors — Qdrant, Postgres
Processing Chains — transform, embed, upsert
Flow Definitions — batch vs stream routing

std raw_html custom

config.toml

# Named template: "std" [std.source.mongo] type = "mongodb" collection = "products" [std.source.kafka] type = "kafka" topic = "product-events" [std.sink.qdrant] type = "qdrant" collection = "product-vectors" [std.processing] chain = ["transform", "embed", "upsert"]

docs/extension/01-process-config/

Deterministic names let linter validate at boot

Deterministic ResourceNames

Predictable naming enables static validation before runtime

org slug + process + kind → siewert-kau-de-std-mongo-source. Deterministic names let linter validate at boot.

— ResourceNames convention

Explanation

Every resource name is generated from a deterministic formula: organization slug + process name + resource kind. This means at boot time, the linter can verify every referenced resource exists by computing expected names and checking the registry. No typos, no missing configs discovered at runtime.

Example

class ResourceNames: def source(org, process, kind): # f"{org}-{process}-{kind}-source" return f"{org}-{process}-{kind}-source" # → "siewert-kau-de-std-mongo-source" # → "siewert-kau-de-std-kafka-source" # → "siewert-kau-de-std-qdrant-sink" # Linter at boot: # for name in expected_names: # assert name in registry, f"Missing: {name}"

docs/extension/01-process-config/

config.toml → Registered Objects

From frozen config to live connector and chain registries

ProcessConfig
frozen dataclass

→

ProcessBuilder
builds + registers

→

Registries
connectors, chains, flows

Enable Flags

Connector	Flag	Status
MongoDB	mongo=true	enabled
Kafka	kafka=false	disabled
Qdrant	qdrant=true	enabled
Postgres	postgres=true	enabled

Example

class StandardProcessBuilder: def build(config: ProcessConfig): if config.mongo_enabled: src = MongoSource(config.mongo) registry.register(src) if config.kafka_enabled: src = KafkaSource(config.kafka) registry.register(src) chain = Chain(config.processing) registry.register(chain)

docs/extension/02-03-two-flows/

Mongo Batch Flow

Cursor-based batch fetching through RabbitMQ to processing chain

Mongo
cursor query

→

Producer
fetch_batch

→

RabbitMQ
publish

→

Consumer
prefetch=1

→

Chain
8 stages

Explanation

The batch flow uses MongoDB cursor queries to fetch documents in bounded batches. The Producer publishes each record as a message to RabbitMQ. Consumers with prefetch=1 process one message at a time through an 8-stage processing chain (validate → transform → embed → upsert → etc). Each message is acked only after full chain completion.

Example

# TransferJob published to RabbitMQ job = { "job_id": "batch-1234-0042", "run_id": "run-2024-0891", "record_id": "prod_abc123", "flow_id": "flow-std-products", "source_data": { "name": "Widget Pro", "price": 29.99, "description": "..." } }

docs/extension/02-03-two-flows/

Composite cursor visits every doc exactly once

The Composite Cursor

Breaking timestamp ties with (updated_at, _id) ordering

(updated_at, _id) visits every doc exactly once by breaking timestamp ties. Saved to run_checkpoints.

— Cursor design principle

Explanation

Using only updated_at as cursor is unsafe — multiple documents can share the same timestamp. The composite cursor (updated_at, _id) adds ObjectId as a tiebreaker, guaranteeing total ordering. After each batch, the cursor position is saved to run_checkpoints so restarts resume from the exact last-seen document.

Example

# MongoDB composite cursor query query = { "$or": [ {"updated_at": {"$gt": last_ts}}, {"$and": [ {"updated_at": last_ts}, {"_id": {"$gt": last_id}} ]} ] } cursor = db.find(query) \ .sort([( "updated_at", 1), ("_id", 1)]) \ .limit(500)

docs/extension/02-03-two-flows/

Crash-Safe Resume

Checkpoint-based recovery with dedup protection

Cursor saved
after each batch

→

Restart resumes
from checkpoint

→

Already-published
still process

→

Dedup prevents
doubles

Explanation

After each batch is published, the cursor position is persisted to the run_checkpoints table. If the producer crashes mid-run, restart picks up from the last saved checkpoint. Messages already in RabbitMQ continue processing. The dedup layer ensures that if any overlap occurs between the checkpoint and in-flight messages, duplicates are silently skipped.

Example

# run_checkpoints table ┌──────────┬───────────────────┬─────────────┐ │ run_id │ cursor_updated_at │ cursor_id │ ├──────────┼───────────────────┼─────────────┤ │ run-0891 │ 2024-03-15T10:30 │ 65f3a... │ │ run-0891 │ 2024-03-15T10:31 │ 65f3b... │ │ run-0891 │ 2024-03-15T10:32 │ 65f3c... │ └──────────┴───────────────────┴─────────────┘ # On restart: SELECT MAX(cursor_*) WHERE run_id=... # Resume from 65f3c... → no gaps, no duplicates

run_checkpoints

Stores (run_id, cursor_updated_at, cursor_id) after every batch publish. The resume query selects the MAX cursor for the run, guaranteeing forward progress without reprocessing.

docs/extension/02-03-two-flows/

Kafka Stream Flow

Continuous processing with direct chain execution

Kafka topic

→

Stream
Consumer

→

Router

→

Chain

→

Qdrant

→

Commit
offset

Config

[std.source.kafka] topic = "product-events" consumer_group = "dtp-std" auto_offset_reset = "earliest"

Explanation

Unlike batch flow, the stream flow is continuous with no RabbitMQ intermediary. The Kafka consumer reads messages, routes them through the operation router, processes via the chain, writes to Qdrant, and commits the offset. The run stays in LISTENING state indefinitely until explicitly force-completed.

Key Difference

Continuous, no RabbitMQ. Run state is LISTENING until force-completed by operator.

docs/extension/02-03-two-flows/

Operation-Based Routing

Message operation field selects the processing chain

Example

class StandardChainRouter: operation_chains = { "add": upsert_chain, "update": upsert_chain, "delete": delete_chain, } def route(message): op = message["operation"] chain = operation_chains[op] return chain.process(message)

Explanation

Each Kafka message carries an operation field (add, update, delete). The StandardChainRouter maps operations to specific chains: add/update → upsert chain (transform + embed + upsert to Qdrant), delete → delete chain (remove from Qdrant by ID). This means a single Kafka topic can drive multiple processing paths without separate consumers.

Routing Table

Operation	Chain	Action
add	upsert_chain	Transform → Embed → Upsert to Qdrant
update	upsert_chain	Transform → Embed → Upsert to Qdrant
delete	delete_chain	Remove from Qdrant by record_id

docs/extension/02-03-two-flows/

Batch vs Stream Comparison

Two flow patterns, one architecture

Dimension	Batch (Mongo)	Stream (Kafka)
Transport	RabbitMQ	Kafka
Run States	PUBLISHING → PROCESSING	LISTENING
Boundaries	Finite batches	Continuous
Commit	Ack message	Commit offset
Finalization	finalize when terminal ≥ published	force-complete only

Explanation

Both flows share the same processing chain, dedup, and DLQ infrastructure. The difference is transport and lifecycle: batch uses RabbitMQ for bounded work units with automatic finalization, while stream uses Kafka for unbounded continuous processing that runs until manually stopped.

Example

# Batch: finite, self-terminating run = start_batch_flow("products") # PUBLISHING → PROCESSING → COMPLETED # 50,000 docs processed, run ends # Stream: infinite, manual stop run = start_stream_flow("events") # LISTENING... LISTENING... LISTENING... force_complete(run) # only way to stop

docs/extension/04-chain /

The Processing Chain

The 8-stage chain

Walking through each stage

1guard → 2normalize → 3validate → 4dedup → 5enrich → 6embed → 7write → 8status

docs/extension/04-chain / stages 1–3

1guard → 2normalize → 3validate → 4dedup → 5enrich → 6embed → 7write → 8status

1 · guard

Checks required fields before any processing begins.

Required: ["name"]

Fails fast with GuardRejectError if any field is missing or empty.

No side effects — pure gate check.

2 · normalize

Computes content_hash = SHA-256 of sorted canonical fields.

Fields: name | description | url | cardData

Parses timestamp into UTC datetime.

Builds embedding_text for downstream vectorization.

3 · validate

Runs jsonschema validation against the normalized record.

Schema defined per-process in config.toml.

Catches type mismatches, missing required properties, enum violations.

Raises ValidationError on failure.

# content_hash computation def compute_content_hash(record: dict) -> str: canonical = "|".join([ str(record.get("name", "")), str(record.get("description", "")), str(record.get("url", "")), json.dumps(record.get("cardData", {}), sort_keys=True), ]) return hashlib.sha256(canonical.encode()).hexdigest()

docs/extension/04-chain / stage 4 — dedup

1guard → 2normalize → 3validate → 4dedup → 5enrich → 6embed → 7write → 8status

Decision tree — three-layer deduplication

TOCTOU Race Condition

Two workers receive the same record simultaneously.

Both check dedup → both see "new" → both process → duplicate vectors in Qdrant.

Solution: Redis claim lock acquired before any dedup check. Second worker sees the lock and skips.

Step 1 — Claim Lock

redis.set_nx(claim_key, worker_id, ex=300)

TTL: 300s (5 min). If lock exists → skip immediately.

Prevents concurrent processing of the same record.

Step 2 — L2 Postgres (authoritative)

Query dedup_index for hash match.

Hash match + stored ts ≥ source ts → DedupSkipError

Stored ts > source ts → StaleVersionError

No match → fall through to L1.

Step 3 — L1 Redis (fast cache)

Check dedup:{flow}:{record_id} key.

TTL: 90 days. Hit → DedupSkipError.

Miss → record is new, proceed to enrich.

# TTLs and key patterns CLAIM_TTL = 300 # 5 minutes — lock expiry L1_TTL = 90 * 24 * 3600 # 90 days — fast-cache window claim_key = f"dedup:claim:{flow}:{record_id}" l1_key = f"dedup:{flow}:{record_id}"

docs/extension/04-chain / stages 5–6

1guard → 2normalize → 3validate → 4dedup → 5enrich → 6embed → 7write → 8status

5 · enrich

Rewrite — clean HTML, strip noise, normalize whitespace.

Translate — optional i18n pass for multilingual corpora.

Chunk — split into semantic segments respecting boundaries.

Output: state.record.chunks — list of text segments ready for vectorization.

Chunk count varies per record → drives stable point ID generation in stage 7.

6 · embed

Sparse vectors — BM25 tokenization per chunk for keyword retrieval.

Dense vectors — OpenAI text-embedding-3-small per chunk for semantic search.

Both run in parallel via asyncio.gather().

Each chunk gets one sparse + one dense vector, paired by index.

# vector structure per chunk sparse_vectors = { "bm25": [ SparseVector(indices=[142, 891, 2047], values=[0.82, 0.55, 0.31]), # chunk 0 SparseVector(indices=[73, 519, 3100], values=[0.91, 0.44, 0.67]), # chunk 1 ] } dense_vectors = { "openai": [ [0.012, -0.034, 0.078, ...], # chunk 0 — 1536 dims [-0.021, 0.056, 0.003, ...], # chunk 1 ] }

docs/extension/04-chain / stage 7 — write

1guard → 2normalize → 3validate → 4dedup → 5enrich → 6embed → 7write → 8status

AHA — Stable point IDs make writes idempotent

Point ID = UUID from org:flow:record:hash:chunk_index → reprocessing yields identical IDs → idempotent upsert.

Deterministic IDs

Each point ID is a UUID v5 derived from the composite key:

org_id : flow_name : record_id : content_hash : chunk_index

Same input → same UUID → Qdrant upsert overwrites in place.

No duplicate points, ever — even after reprocessing.

Stale-point GC

When chunk count changes (e.g., 5 chunks → 3 chunks), old points at indices 3–4 become orphaned.

Garbage collection: delete all existing points for this record, except the new ones.

delete_points_by_record(..., except_point_ids)

Runs before upsert to ensure clean state.

# stale-point garbage collection async def write_vectors(state: ChainState): new_ids = [point_id(i) for i in range(len(state.chunks))] await delete_points_by_record( collection=state.collection, record_filter=state.record_filter, except_point_ids=new_ids, # keep the ones we're about to write ) await qdrant.upsert(collection=state.collection, points=new_points)

docs/extension/04-chain / stage 8 — status

1guard → 2normalize → 3validate → 4dedup → 5enrich → 6embed → 7write → 8status

Close the loop — update every dedup layer

Write record_state — COMPLETED or SKIPPED + chunks_count
Upsert L2 dedup_index — Postgres row with hash + timestamp (authoritative)
Set L1 Redis key — dedup:{flow}:{record_id} with 90-day TTL
Release claim lock — delete dedup:claim:{flow}:{record_id}

Why status runs last

If any earlier stage fails, status is never reached — the record remains unclaimed in the queue for retry.

Only on full success does the dedup state get written, making the record visible to future dedup checks.

Key patterns written

Layer	Key	TTL
L1 Redis	dedup:{flow}:{record_id}	90 days
L1 Claim	dedup:claim:{flow}:{record_id}	deleted
L2 Postgres	dedup_index (flow, record_id)	permanent
State	record_state (record_id)	permanent

# closing the dedup loop await redis.set( f"dedup:{flow}:{record_id}", content_hash, ex=90 * 24 * 3600, # 90 days ) await redis.delete(f"dedup:claim:{flow}:{record_id}") await pg.upsert_dedup_index(flow, record_id, content_hash, ts)

docs/extension/04-chain / idempotency recap

Three pillars of idempotency

content_hash

SHA-256

Detects unchanged records before any expensive processing.

Same content → same hash → skip early, save compute.

Stage 2: normalize

Stable point IDs

UUID v5

Deterministic UUIDs from composite key ensure upserts overwrite, never duplicate.

Reprocess → identical IDs → Qdrant replaces in place.

Stage 7: write

Dedup claim lock

Redis SETNX

Prevents TOCTOU races between concurrent workers on the same record.

First worker claims → second sees lock → skips cleanly.

Stage 4: dedup

Reprocess any record safely — no duplicates, no data loss, no race conditions.

docs/extension/05-customization / override one handler

AHA — Open/Closed Principle: extend without forking

Pass enrich_cls=CustomHandler to register_org(). Open/Closed — extend without forking.

How it works

Each stage is a pluggable class with a default implementation.

Override only the stage that needs custom logic — the rest of the chain runs unchanged.

Registration-time injection: register_org(enrich_cls=...)

The chain engine resolves handlers from the org's registration, not from hardcoded imports.

Real example: Siewert

Siewert records sometimes lack a url field.

Custom enrich handler falls back to card://{record_id} as a synthetic URL.

Then delegates to super().process() for standard enrichment.

One class override, zero chain changes

# custom handler with URL fallback class SiewertEnrichHandler(EnrichHandler): async def process(self, state: ChainState) -> ChainState: if not state.record.get("url"): state.record["url"] = f"card://{state.record['id']}" return await super().process(state) # delegate to default

docs/extension/05-customization / org isolation

Org isolation guarantees

Registration-time checks prevent cross-org contamination

Matching org enforcement — config.toml org field must match the registration key. Mismatch → ConfigError at startup.
Domain normalization — siewert-kau.de and siewert_kau_de resolve to the same canonical domain. Prevents duplicate org registrations.
Global handler-name uniqueness — no two orgs can register the same handler name. Avoids routing ambiguity at the API layer.
Bad org quarantined — a misconfigured org fails registration without affecting other orgs. The rest of the system continues normally.

What happens on mismatch

Registration runs at startup, not at request time.

Every org's config is validated before the server accepts traffic.

A single bad org → ConfigError with a clear message, server refuses to start until fixed.

This is intentional: fail loud at deploy, not silent in production.

# org mismatch at registration if config.org != registration_key: raise ConfigError( f"Org mismatch: config declares '{config.org}', " f"but registered as '{registration_key}'" ) # → server refuses to start

docs/extension/05-customization / minimum viable org

Minimum viable org

Two files. That's it.

What you need

config.toml — declares org name, flow definitions, collection mappings, schema.

__init__.py — calls register_org(api, ...) to wire the org into the framework.

Everything else — chain stages, dedup logic, embedding, Qdrant writes — is inherited from the process template.

Override only what's different. The framework handles the rest.

# 3 lines from core.bootstrap import register_org from .config import config register_org( api, org="my_org", config=config, )

What you get for free

8-stage chain — guard through status, fully wired.

Dedup — L1/L2 + claim lock, zero config.

Embedding — BM25 + OpenAI, parallel.

Qdrant writes — stable IDs, stale-point GC.

Recovery — DLQ, retry, state machine.

That's it. Everything else is inherited from the process template.

docs/recap / the whole picture

The whole picture, one slide

CORE — framework rails

chain engine state machines workers

triggers connectors bootstrap

recovery services DLQ

The invariant infrastructure that every org shares.

You don't configure it — you run on top of it.

State transitions, retry semantics, dedup protocol, vector writes — all framework-owned.

EXTENSION — domain fills

config.toml processes

batch / stream flows

8-stage chain handlers

per-org customization

The org-specific layer that plugs into the core.

Config declares what, handlers override how.

Override one stage or none — the defaults cover 90% of orgs.

CORE provides the rails

+

EXTENSION fills the domain

=

DTP

The core truth, with everything learned hung on it.

Callback to slide 4 — the architecture in one frame

docs/closing / next steps

Where to go next

Your onboarding reading list