Trigger flow, concurrency model, and an annotated map of breaking points across three chained Airflow DAGs. Compiled from two independent code reviews of PR #112.
A factory build_platform_dags(PlatformConfig) in src/threat_intel/common/dags.py produces three chained DAGs per platform, instantiated at import time by src/threat_intel_telegram.py. Today only platform=telegram exists. Each DAG hands off to the next; completion ripples back up.
Polls Kafka on a 1-minute schedule, upserts the Mongo job, triggers discover, waits, then commits the offset.
Externally triggered. Runs the discover pod on Kubernetes, then triggers screen and waits for it.
Externally triggered. A single task claims unscreened docs in batches and screens them with an LLM.
Directional edges are labeled by trigger mechanism. Hover any node to isolate its edges and reveal labels. Red badges mark nodes involved in a HIGH-severity weak link. The green dashed edges are the completion ripple travelling back up the chain.
none_failed_min_one_successThe pipeline is strictly serial per platform but cheap while waiting. It scales horizontally across platforms, never within one. Substantial latent parallelism in the screener is designed for but never used.
max_active_runs=1 on all three DAGs, max_active_tis_per_dag=1 on the sensor, one Kafka message per run, and each parent blocking on its child.The Kafka sensor is deferrable, so it hands its worker slot to the triggerer while idle. Both wait sensors use mode="reschedule", releasing their slots between 30s pokes. Idle time costs almost no executor capacity.
A single pod on a single hardcoded telegram account (index 0). Contrast the repo's stable twitter DAG, which fans scrapers out across accounts. There is no horizontal discovery here.
A while-loop, one batch of 20 at a time, one HTTP call per batch. The Mongo claim mechanism (1h claim timeout, atomic claim of unscreened docs, stale reclaim) is designed for concurrent screeners, but the DAG never runs more than one. Unused parallelism.
The factory pattern gives each new platform (e.g. discord) its own topic, consumer group, and 3 DAGs. Platforms run independently of each other; jobs within a platform remain strictly serial.
Ten findings across both reviews. Filter by severity, click any card to expand its trigger condition, blast radius, and fix direction.
The repo's stable connector DAG (sm_connector_dag_stable.py + common_ops/kafka_reader.py + common_ops/dispatch_ops.py) already solves blockers 1, 2, and 6 with a Postgres dispatch-claims ledger: skip offsets already in the ledger, claim-before-dispatch via PK insert on (topic, partition, offset), commit only the highest contiguous completed offset ≥ the current committed position (monotonic), verify ownership before commit, and route invalid messages through the same ledger.
| Dimension | PR #112 listener | Stable ledger pattern |
|---|---|---|
| Dedupe | None — Mongo upsert overwrites blindly, no offset ledger | Skip via kafka_dispatch_offset_exists before dispatch |
| Commit monotonicity | Blind offset+1; can move group offset BACKWARD | advance_completed_kafka_offsets — highest contiguous offset only |
| Failed-run handling | FAILED child treated as success → message consumed anyway | Ownership verified before commit; failures do not advance |
| Skip path | Low-watermark fallback replays retained backlog | Claim-before-dispatch on unique (topic,partition,offset) |
| Invalid messages | _skip_unprocessable commits offset then re-defers | Invalid messages routed through the same ledger |
| Test coverage | None | tests/test_deferred_kafka_reader.py |
Minimum viable fix if the full ledger is out of scope: drop the get_first_produced fallback and ignore any fallback offset lower than the committed position.
Three HIGH blockers each let the pipeline silently lose or infinitely reprocess work while reporting success. Fix blockers 1, 2, and 3 before merge. Everything else is Medium/Low and can follow.