You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "udsy19 (via GitHub)" <gi...@apache.org> on 2026/09/21 17:08:53 UTC
[PR] Fix wait_for_past_depends_before_skipping being ignored by SkipMixin [airflow]
udsy19 opened a new pull request, #73470:
URL: https://github.com/apache/airflow/pull/73470
`wait_for_past_depends_before_skipping` is documented to hold a `depends_on_past` task back — leave it in `none` state — instead of skipping it, whenever its own past-run dependency is not yet met. That contract is honored in exactly one place: `NotPreviouslySkippedDep`, the scheduler dependency check that re-evaluates a task recorded in a `SkipMixin` XCom payload and checks the `PAST_DEPENDS_MET` XCom before letting it be skipped.
The problem: `SkipMixin.skip()` (used by `ShortCircuitOperator`) and `SkipMixin.skip_all_except()` (used by `BranchPythonOperator`) never route through that check for their own direct targets. Both raise `DownstreamTasksSkipped`, which the task-sdk supervisor turns into a `PATCH .../skip-downstream` call (`airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py::ti_skip_downstream`) that force-writes `state=SKIPPED` on the named task instances directly — no `depends_on_past`/`wait_for_past_depends_before_skipping` check anywhere on that path, only a guard against clobbering a RUNNING/SUCCESS/FAILED row. So a task with the flag set gets skipped immediately by its upstream short-circuit/branch, regardless of whether its own past run succeeded.
Impact: contract-violation
WHO reaches this / entry point: any DAG using the documented `wait_for_past_depends_before_skipping=True` + `depends_on_past=True` combination on a task that sits downstream of a `ShortCircuitOperator` or `BranchPythonOperator` — exactly the workflow the flag's own PR (#27710) was written for, and exactly apache/airflow#55146's own reproduction. Triggered by: the upstream `ShortCircuitOperator`/`BranchPythonOperator` deciding to skip that specific downstream task in the current DAG run (short-circuit condition false, or branch not taken). What they observe: the task is marked `SKIPPED` immediately, even when the previous DAG run's instance of that same task failed — `wait_for_past_depends_before_skipping` has no effect at all for this (very common) skip path, as reported.
## Fix
`SkipMixin.skip()`/`skip_all_except()` already push the `XCOM_SKIPMIXIN_KEY` payload listing every affected task before deciding what to force-skip — that payload is what `NotPreviouslySkippedDep` reads when it later re-evaluates a task. Mapped tasks are already excluded from the immediate force-skip and left for that same check ("future mapped tasks have not been expanded yet and are handled by NotPreviouslySkippedDep"). This PR extends the same exclusion to any task with `wait_for_past_depends_before_skipping=True`: it is still recorded in the XCom payload (so `NotPreviouslySkippedDep` knows it should be skipped once its past depends are met), but it is no longer force-skipped by the direct `skip-downstream` write. `NotPreviouslySkippedDep` picks it up on the next scheduler pass and correctly holds it in `none` until `PAST_DEPENDS_MET` is true.
## Testing
Two new tests in `task-sdk/tests/task_sdk/bases/test_skipmixin.py::TestSkipMixin`:
- `test_skip_excludes_wait_for_past_depends_task` (the `ShortCircuitOperator` path via `skip()`)
- `test_skip_all_except_excludes_wait_for_past_depends_task` (the `BranchPythonOperator` path via `skip_all_except()`)
Negative control — reverting only `task-sdk/src/airflow/sdk/bases/skipmixin.py` to `upstream/main` and running the full test file:
```
FAILED task-sdk/tests/task_sdk/bases/test_skipmixin.py::TestSkipMixin::test_skip_excludes_wait_for_past_depends_task
FAILED task-sdk/tests/task_sdk/bases/test_skipmixin.py::TestSkipMixin::test_skip_all_except_excludes_wait_for_past_depends_task
2 failed, 12 passed, 1 warning in 2.48s
```
With the fix restored, the full file passes: `14 passed, 1 warning`. The other 12 pre-existing tests in the file are unmodified in behavior — they now pass an explicit `wait_for_past_depends_before_skipping=False` on their mock task fixtures, because without it the new filtering logic reads an auto-generated (truthy) `MagicMock` attribute from bare `MagicMock(spec=BaseOperator)` instances and incorrectly defers them too. That is a mocking artifact, not a production behavior change: real `BaseOperator`/`MappedOperator` instances always carry the real dataclass-defaulted `False`.
closes: #55146
Signed-off-by: Udaya Tejas <ud...@gmail.com>
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes - Claude Code (Sonnet 5)
Generated-by: Claude Code (Sonnet 5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix wait_for_past_depends_before_skipping being ignored by SkipMixin [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil closed pull request #73470: Fix wait_for_past_depends_before_skipping being ignored by SkipMixin
URL: https://github.com/apache/airflow/pull/73470
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix wait_for_past_depends_before_skipping being ignored by SkipMixin [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
aaron-y-chen commented on code in PR #73470:
URL: https://github.com/apache/airflow/pull/73470#discussion_r4088796692
##########
task-sdk/src/airflow/sdk/bases/skipmixin.py:
##########
@@ -79,7 +79,14 @@ def skip(
value={XCOM_SKIPMIXIN_SKIPPED: task_ids_list},
)
- self._set_state_to_skipped(task_ids_list, ti.map_index)
+ # A task with wait_for_past_depends_before_skipping=True must not be force-skipped
+ # here: doing so bypasses NotPreviouslySkippedDep, the only place that flag is
+ # honored, so it never gets the chance to hold the task back until its own past
+ # depends are met. Leave it for that dependency check to decide instead.
+ immediate_skip_ids = [
+ d.task_id for d in task_list if not getattr(d, "wait_for_past_depends_before_skipping", False)
Review Comment:
I reproduced a regression in the default `ShortCircuitOperator` mode. With `gate >> middle >> target`, where `target` has `depends_on_past=True`, `wait_for_past_depends_before_skipping=True`, and `trigger_rule="all_done"`, run 1 succeeds and `gate` returns False in run 2.
On this PR, `middle` is skipped but the scheduler considers `target` ready and marks it successful. On the base commit, `target` is skipped. `skip()` records `target` in gate's XCom, but `NotPreviouslySkippedDep` only checks direct upstream tasks, so it cannot find that decision through `middle`. Could we preserve the default behavior of skipping all descendants when deferring this skip, and add a two-run regression test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix wait_for_past_depends_before_skipping being ignored by SkipMixin [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #73470:
URL: https://github.com/apache/airflow/pull/73470#issuecomment-5940884331
Closing this. Instead of skipping, the change leaves the task to `NotPreviouslySkippedDep` "on the next scheduler pass", but that dep only waits when `dep_context.wait_for_past_depends_before_skipping` is true, and nothing in the scheduler sets it: `DepContext` defaults it to `False`, and the contexts built in `dagrun.py` don't pass it. So the task is still skipped, one loop later. The new tests are `SkipMixin` unit tests with mocks, and none shows the downstream task staying unscheduled, which is what #55146 asks for.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org