apache / apache/airflow

dag_version inflection when using TimeSensor with `start_from_trigger = True`

Open
#69,543 5 comments 0 reactions 0 assignees View on GitHub
area:providers kind:bug needs-triage provider:standard
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### Under which category would you file this issue?

Providers

### Apache Airflow version

3.2.2

### What happened and how to reproduce it?

TimeSensor with start_from_trigger=True produces a different serialized DAG (new DAG version) on every DAG-file parse, even when nothing in the DAG file has changed.
Root cause: in [TimeSensor.__init__](https://github.com/apache/airflow/blob/main/providers/standard/src/airflow/providers/standard/sensors/time.py#L83), the target datetime is computed using datetime.datetime.now(self.dag.timezone):
```
aware_time = timezone.coerce_datetime(
datetime.datetime.combine(
datetime.datetime.now(self.dag.timezone), target_time, self.dag.timezone
)
)
self.target_datetime = timezone.convert_to_utc(aware_time)
...
if self.start_from_trigger:
self.start_trigger_args.trigger_kwargs = dict(
moment=self.target_datetime, end_from_trigger=self.end_from_trigger
)
```
__init__ runs at DAG-parse time, not task-execution time, so datetime.now() is evaluated fresh on every parse cycle. The resulting target_datetime is written into start_trigger_args.trigger_kwargs, which is exactly the attribute the scheduler serializes to hand the task directly to the triggerer (the reason start_from_trigger exists at all). Since a volatile value is now part of the serialized operator, the serialized DAG hash changes on essentially every parse a new DAG version is created continuously on every dag_run.

Reproducable Example:
```
from __future__ import annotations

import datetime
import pendulum

from airflow.sdk import DAG
from airflow.providers.standard.sensors.time import TimeSensor

with DAG(
dag_id="time_sensor_dag",
schedule="@daily",
start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
catchup=False,
) as dag:
wait = TimeSensor(
task_id="wait_until_noon",
target_time=datetime.time(12, 0, 0),
start_from_trigger=True,
)
```

- Deploy this DAG with a triggerer running.
- it will change its verstion every schduled dag_runs.

Also you can verify from `serialized_dag` table entry where its dag_hash changes because of that TimeSensor task trigger_start_kwargs changes.

### What you think should happen instead?

The docstring for [TimeSensor](https://airflow.apache.org/docs/apache-airflow-providers-standard/stable/sensors/datetime.html#:~:text=Time%20will%20be%20evaluated%20against%20data_interval_end%20if%20present%20for%20the%20Dag%20run%2C%20otherwise%20run_after%20will%20be%20used.) currently states: "Time will be evaluated against data_interval_end if present for the Dag run, otherwise run_after will be used."

implying the target moment should be derived from the DagRun, a runtime concept. The actual code instead uses parse-time datetime.now(), which has no relation to data_interval_end/run_after. Either the code should be fixed to match the documented (run-relative) behavior, or if its not possible then the docstring should be corrected to describe what actually happens with stating that dag_version will be changed.

Whatever the target-moment computation ends up being, it should not be baked into start_trigger_args.trigger_kwargs at __init__/parse time as an absolute value that changes on every parse. Compare with DateTimeSensorAsync, where target_time is a template_fields entry resolved via Jinja against the run context, and TimeDeltaSensor(Async), which doesn't implement start_from_trigger at all — presumably because its target (data_interval_end/run_after + delta) isn't knowable at parse time either. TimeSensor appears to be the one sensor that eagerly computes a volatile value instead of deferring it.

### Operating System

Linux

### Deployment

Docker-Compose

### Apache Airflow Provider(s)

standard

### Versions of Apache Airflow Providers

apache-airflow-providers-standard (main / current released version as of Airflow 3.2.2)

### Official Helm Chart version

Not Applicable

### Kubernetes Version

Not Applicable

### Helm Chart configuration

Not Applicable

### Docker Image customizations

Not Applicable

### Anything else?

I'd love to help contribute — this would be my first PR.
I'll dig into this further and discuss before submitting a PR, since it touches DAG-serialization/versioning behavior. I'm willing to submit the fix, but I'd like guidance on the preferred direction: (a) derive the target moment from data_interval_end/run_after at trigger-start time instead of parse time, (b) some other approach, or (c) remove start_from_trigger from TimeSensor entirely.
I haven't looked at the triggerer's relevent code yet — I'll dig into it and come back with findings.

### Are you willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

Start with TimeSensor.__init__ in providers/standard/src/airflow/providers/standard/sensors/time.py and trace how start_trigger_args.trigger_kwargs reaches DAG serialization and the triggerer. Compare the behavior with DateTimeSensorAsync and TimeDeltaSensor(Async), then verify changes through the serialized_dag entry and the documented run-relative target behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
backend, data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.