Task with "wait_for_downstream" doesn't wait, because start_date is mishandled

[shai-ikko]

I am running a flow that starts with a S3KeySensor to detect incoming files. It checks for files with a certain suffix. The first step once such a file is discovered, is to change the suffix; the Sensor is marked with wait_for_downstream=True, so that it is only checked again after the suffix change, and doesn't trigger until a new file arrives.

This doesn't work -- the sensor doesn't wait for the downstream task to finish; as a result, it triggers another rename attempt, and that fails because by the time it is launched, the other rename has succeeded.

To investigate this, I changed the lines 155-159 in the file airflow/ti_deps/deps/prev_dagrun_dep.py to

        # There was a DAG run, but the task wasn't active back then.
        if catchup and last_dagrun.execution_date < ti.task.start_date:
            self._push_past_deps_met_xcom_if_needed(ti, dep_context)
            dbg_reason = f"First instance of this task. Last dagrun {last_dagrun.execution_date!r} Task start {ti.task.start_date!r}"
            yield self._passing_status(reason=dbg_reason)
            return

(I used the less-than-perfectly-readable !r because I suspected the problem was related to timezone or maybe even naive-vs-aware datetime issues)
This produced the following output in the scheduler log (grepped relevant lines only):

airflow-scheduler-1  | [2024-06-26T12:08:15.450+0000] {taskinstance.py:2088} DEBUG - <TaskInstance: monitor_incoming.detect_incoming scheduled__2024-06-26T09:07:47.534647+00:00 [None]> dependency 'Previous Dagrun State' PASSED: True, First instance of this task. Last dagrun datetime.datetime(2024, 6, 26, 9, 2, 53, 910284, tzinfo=Timezone('UTC')) Task start DateTime(2024, 6, 26, 9, 7, 47, 534647, tzinfo=Timezone('UTC'))
airflow-scheduler-1  | [2024-06-26T12:09:34.659+0000] {taskinstance.py:2088} DEBUG - <TaskInstance: monitor_incoming.detect_incoming scheduled__2024-06-26T09:09:18.745780+00:00 [None]> dependency 'Previous Dagrun State' PASSED: True, First instance of this task. Last dagrun datetime.datetime(2024, 6, 26, 9, 7, 47, 534647, tzinfo=Timezone('UTC')) Task start DateTime(2024, 6, 26, 9, 9, 18, 745780, tzinfo=Timezone('UTC'))
airflow-scheduler-1  | [2024-06-26T12:10:49.783+0000] {taskinstance.py:2088} DEBUG - <TaskInstance: monitor_incoming.detect_incoming scheduled__2024-06-26T09:10:19.516626+00:00 [None]> dependency 'Previous Dagrun State' PASSED: True, First instance of this task. Last dagrun datetime.datetime(2024, 6, 26, 9, 9, 18, 745780, tzinfo=Timezone('UTC')) Task start DateTime(2024, 6, 26, 9, 10, 19, 516626, tzinfo=Timezone('UTC'))
airflow-scheduler-1  | [2024-06-26T12:12:08.630+0000] {taskinstance.py:2088} DEBUG - <TaskInstance: monitor_incoming.detect_incoming scheduled__2024-06-26T09:11:50.849850+00:00 [None]> dependency 'Previous Dagrun State' PASSED: True, First instance of this task. Last dagrun datetime.datetime(2024, 6, 26, 9, 10, 19, 516626, tzinfo=Timezone('UTC')) Task start DateTime(2024, 6, 26, 9, 11, 50, 849850, tzinfo=Timezone('UTC'))

Explanation: One of the tests relevant to wait_for_downstream is, has this task even existed at the time of the last DAG run? If not, then, obviously, it didn't have dependencies that we need to check. That check is done by comparing the start_date on the task (not instance) to the execution_date on the last DAG run -- assuming that the start_date on the task records when that task was put in place.

But what you can see in the log is that start_date on the task is updated continuously; in fact, it matches the schedule time of the current DAG run, which is (at least in my case) almost always later than the execution time of the previous run.

This is also reflected through the web UI -- if you open the details of a task instance after some future runs have occurred, and click "more details", you can see an instance start-date that is older than the task start-date.

Am I missing something fundamental, or is it a bug?

(the code modifications etc were done on 2.9.1 but AFAICT 2.9.2 shows exactly the same behavior)