Event-driven scheduling: duplicate DAG runs when an AssetWatcher runs on multiple triggerers

[nathadfield]

The problem

When you run more than one triggerer (the default for HA), an AssetWatcher that polls some external state seems to fire once per triggerer for the same change, so the consumer DAG runs twice (N times for N triggerers).

We hit this on 3.2.1 with a watcher that polls a Git repo's main HEAD and triggers a re-index when it advances. With two triggerers, every merge kicks off two identical runs of a fairly expensive pipeline. Cron and manual runs are fine; only the asset-triggered ones double up, and the count tracks the number of triggerers.

The watcher definition is deduped (one trigger backs the asset), but the events aren't: each triggerer runs its own copy of the watcher, independently notices the same change, emits its own event, and each event creates a run. So it's effectively at-least-once delivery with no built-in way to get "one run per change" for a plain poller.

What's already there (and why it doesn't seem to cover this)

A couple of adjacent things exist:

• Shared-stream triggers (3.3, Share one poll loop across sibling event triggers #66584 / Add producer-side ack channel to shared-stream triggers #67523) let sibling watchers share one poll loop and add a producer-side ack channel for message brokers. But the docs are explicit that the saving is "at the poll-loop and upstream-I/O layer, not at the persistence or scheduling layer", so it reduces polling, not the number of events/runs. For a plain (non-broker) poller across triggerers, you'd still get one event per triggerer.
• The docs do say subscribers must be idempotent, but in the context of the ack channel and triggerer restart (broker redelivery). The plain multi-triggerer poll-watcher duplication doesn't seem to be called out.

So this may already be considered expected behaviour, and part of what I'm asking is whether it is.

How this differs from #54491 / #63507

Those are scheduler-side: multiple schedulers turning a single event into two runs (fixed with a row-level lock, #60773). Here there are genuinely two events from two triggerers, so that fix doesn't apply.

What we do about it today

The usual "make the consumer idempotent" approach: the DAG's first step skips if it's already processed this particular change. It works, but every author of an event-driven DAG has to reinvent it, and the duplicate run still gets scheduled before it's skipped.

Possible directions

• A dedupe key on AssetWatcher (derived from the event) so identical events collapse into one run. Feels like the most natural fit, and complements shared-stream, which deliberately stops at the poll layer.
• Coalescing identical events on the scheduler side within a short window.
• An opt-in single-runner watcher (one triggerer owns it, with failover) for cases where you'd rather have exactly-once than zero-gap coverage. More architectural, hence a discussion.
• Failing any of that, document the plain multi-triggerer poll case explicitly, alongside the existing idempotency notes.

Questions

• Is triggerer-side duplication for a plain poller a known/accepted property, or a gap worth closing at the scheduling layer (given shared-stream intentionally stops short of it)?
• If it's worth closing, is a dedupe key on AssetWatcher the right first step, or does the single-runner idea need an AIP?
• Is there a way to make a polling watcher exactly-once across triggerers today that I've missed?

Related: #54491, #63507, #60773, #66584, #67523.

[VectorPeak]

This looks like #65792 more than a new AssetWatcher dedupe-key design gap.

The short version: on 3.2.1, asset triggers were selected differently from task-instance triggers. The task-trigger path filtered to triggers that were unassigned or assigned to a dead triggerer, but the asset-trigger branch just selected asset triggers, so multiple live triggerers could pick up the same watcher. That matches your “N triggerers => N events/runs” symptom pretty closely. In 3.2.2, the asset-trigger query includes the same triggerer_id is null or not in alive_triggerer_ids filter. (raw.githubusercontent.com)

That behavior should not be considered the intended steady-state semantics. Airflow’s trigger HA model is meant to have a trigger row owned by one live triggerer, with reassignment on failover; duplicate execution can still happen around failures, but “every live triggerer runs the same asset watcher” is the bug #65792 fixed. The PR description calls out exactly that multiple triggerer replicas could pick up the same asset trigger, and it was marked for the Airflow 3.2.2 milestone/backport. (github.com)

The practical next step is to test this on 3.2.2 or newer. If the same Git HEAD watcher still produces one event per triggerer there, that would be worth opening as a focused bug with the watcher definition, triggerer count, DB backend, and a snippet of triggerer logs showing the same trigger ID running on multiple triggerers.

A dedupe key on AssetWatcher still looks like a reasonable feature discussion, but it is a separate layer. It would help with “same external change observed more than once” semantics, especially for polling sources or broker redelivery, but it should not be required just to prevent all healthy triggerers from running the same persisted watcher.

For today, consumer idempotency is still the right defensive pattern. Even with the ownership fix, event-driven systems generally need to tolerate retries, triggerer restarts, and external-source redelivery. The difference is that after #65792, idempotency should protect against occasional duplicate delivery, not deterministic duplication proportional to triggerer count.

If upgrading is blocked, the safest mitigation is to run only one triggerer capable of handling event-driven asset watchers, accepting the HA tradeoff, or put a compare-and-set checkpoint in the watcher/first task keyed by the Git SHA. That is not as nice as a first-class dedupe key, but it avoids paying for the expensive downstream work twice.

---

If my answer solved your problem, you can click answered the question. I'm really here to help, and along the way I'm also collecting Galaxy Brain badges haha 😆

[nathadfield]

Thanks @VectorPeak, this was spot on. We upgraded to 3.2.2 and the duplicate runs are gone: a new commit to the repo our watcher polls now triggers a single DAG run instead of one per triggerer. So it was #65792, not a missing dedupe feature. We'll keep a small idempotency check as a safety net, but it's no longer doing the heavy lifting. Marking as answered, thanks for the help!