You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "seanghaeli (via GitHub)" <gi...@apache.org> on 2026/07/24 23:39:36 UTC
[PR] Harden deadline serialization, repr, and prune paths [airflow]
seanghaeli opened a new pull request, #70421:
URL: https://github.com/apache/airflow/pull/70421
Follow-up to #68919 per review: the robustness hardening split out from that PR so it stays minimal. Covers deserialization routing errors, missing __class_path handling, repr guards for dagrun/dict-interval edge cases, and a prune_deadlines missed-guard. No behavior change to the UI endpoint fix in #68919.
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #70421:
URL: https://github.com/apache/airflow/pull/70421#issuecomment-5317957337
### Backport successfully created: v3-3-test
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>✅</td>
<td>v3-3-test</td>
<td><a href="https://github.com/apache/airflow/pull/71726"><img src="https://img.shields.io/badge/PR-71726-blue" alt="PR Link"></a></td>
</tr>
</table>
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3779259374
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -57,11 +57,14 @@ def __repr__(self):
interval_seconds = None
+ # Legacy rows store a bare number instead of a serialized dict.
if isinstance(self.interval, (int, float)):
interval_seconds = int(self.interval)
- elif isinstance(self.interval, datetime.timedelta):
- interval_seconds = int(self.interval.total_seconds())
+ elif isinstance(self.interval, dict):
+ data = self.interval.get("__data__")
+ if isinstance(data, (int, float)):
+ interval_seconds = int(data)
Review Comment:
Probably right, I was overthinking it. :+1:
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3778877562
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
+
+
+async def _qaw18_callback():
+ pass
+
+
+CALLBACK_PATH = f"{__name__}.{_qaw18_callback.__name__}"
+
+
+def _clean_db():
+ db.clear_db_dags()
+ db.clear_db_runs()
+ db.clear_db_deadline()
+
+
+@pytest.fixture
+def dagrun(session, dag_maker):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="task_id")
+ with time_machine.travel(DEFAULT_DATE):
+ dag_maker.create_dagrun(state=DagRunState.QUEUED, logical_date=DEFAULT_DATE)
+ session.commit()
+ return session.scalars(select(DagRun)).one()
+
+
+def _make_deadline(session: Session, *, dagrun_id: int, deadline_time, state=None) -> Deadline:
+ deadline = Deadline(
+ deadline_time=deadline_time,
+ callback=AsyncCallback(CALLBACK_PATH),
+ dagrun_id=dagrun_id,
+ dag_id=DAG_ID,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ if state is not None:
+ deadline.callback.state = state
+ session.add(deadline.callback)
+ session.flush()
+ return deadline
+
+
+@pytest.mark.db_test
+class TestPruneDeadlines:
+ @staticmethod
+ def setup_method():
+ _clean_db()
+
Review Comment:
Huh. Yeah, we had a campaign to clean those up (https://github.com/apache/airflow/issues/68374) but I guess I didn't cast a wide enough net and we missed some. They should definitely be fixed, but you are right that it's not part of this PR.
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3778890432
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
+
+
+async def _qaw18_callback():
+ pass
+
+
+CALLBACK_PATH = f"{__name__}.{_qaw18_callback.__name__}"
+
+
+def _clean_db():
+ db.clear_db_dags()
+ db.clear_db_runs()
+ db.clear_db_deadline()
+
+
+@pytest.fixture
+def dagrun(session, dag_maker):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="task_id")
+ with time_machine.travel(DEFAULT_DATE):
+ dag_maker.create_dagrun(state=DagRunState.QUEUED, logical_date=DEFAULT_DATE)
+ session.commit()
+ return session.scalars(select(DagRun)).one()
+
+
+def _make_deadline(session: Session, *, dagrun_id: int, deadline_time, state=None) -> Deadline:
+ deadline = Deadline(
+ deadline_time=deadline_time,
+ callback=AsyncCallback(CALLBACK_PATH),
+ dagrun_id=dagrun_id,
+ dag_id=DAG_ID,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ if state is not None:
+ deadline.callback.state = state
+ session.add(deadline.callback)
+ session.flush()
+ return deadline
+
+
+@pytest.mark.db_test
+class TestPruneDeadlines:
+ @staticmethod
+ def setup_method():
+ _clean_db()
+
Review Comment:
In fact, I just set Claude to search the repo and it says we missed a LOT. It counts around 85 uses over 50 files that my initial scope missed...
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #70421:
URL: https://github.com/apache/airflow/pull/70421#issuecomment-5317952935
Hi maintainer, this PR was merged without a milestone set.
We've automatically set the milestone to **[Airflow 3.3.2](https://github.com/apache/airflow/milestone/150)** based on: backport label targeting v3-3-test
If this milestone is not correct, please update it to the appropriate milestone.
> This comment was generated by [Milestone Tag Assistant](https://github.com/apache/airflow/blob/main/.github/workflows/milestone-tag-assistant.yml).
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3778948191
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
+
+
+async def _qaw18_callback():
+ pass
+
+
+CALLBACK_PATH = f"{__name__}.{_qaw18_callback.__name__}"
+
+
+def _clean_db():
+ db.clear_db_dags()
+ db.clear_db_runs()
+ db.clear_db_deadline()
+
+
+@pytest.fixture
+def dagrun(session, dag_maker):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="task_id")
+ with time_machine.travel(DEFAULT_DATE):
+ dag_maker.create_dagrun(state=DagRunState.QUEUED, logical_date=DEFAULT_DATE)
+ session.commit()
+ return session.scalars(select(DagRun)).one()
+
+
+def _make_deadline(session: Session, *, dagrun_id: int, deadline_time, state=None) -> Deadline:
+ deadline = Deadline(
+ deadline_time=deadline_time,
+ callback=AsyncCallback(CALLBACK_PATH),
+ dagrun_id=dagrun_id,
+ dag_id=DAG_ID,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ if state is not None:
+ deadline.callback.state = state
+ session.add(deadline.callback)
+ session.flush()
+ return deadline
+
+
+@pytest.mark.db_test
+class TestPruneDeadlines:
+ @staticmethod
+ def setup_method():
+ _clean_db()
+
Review Comment:
Good catch then. Good news is that this test (and I suspect the rest probably as well) still pass after removing `_clean_db` on setup, but definitely could mask a future regression. I'll let you make the PR for the wider fix
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3778948191
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
+
+
+async def _qaw18_callback():
+ pass
+
+
+CALLBACK_PATH = f"{__name__}.{_qaw18_callback.__name__}"
+
+
+def _clean_db():
+ db.clear_db_dags()
+ db.clear_db_runs()
+ db.clear_db_deadline()
+
+
+@pytest.fixture
+def dagrun(session, dag_maker):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="task_id")
+ with time_machine.travel(DEFAULT_DATE):
+ dag_maker.create_dagrun(state=DagRunState.QUEUED, logical_date=DEFAULT_DATE)
+ session.commit()
+ return session.scalars(select(DagRun)).one()
+
+
+def _make_deadline(session: Session, *, dagrun_id: int, deadline_time, state=None) -> Deadline:
+ deadline = Deadline(
+ deadline_time=deadline_time,
+ callback=AsyncCallback(CALLBACK_PATH),
+ dagrun_id=dagrun_id,
+ dag_id=DAG_ID,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ if state is not None:
+ deadline.callback.state = state
+ session.add(deadline.callback)
+ session.flush()
+ return deadline
+
+
+@pytest.mark.db_test
+class TestPruneDeadlines:
+ @staticmethod
+ def setup_method():
+ _clean_db()
+
Review Comment:
Good catch then. Good news is that this test (and I suspect the rest probably as well) still pass after removing `_clean_db` on setup, but definitely could have masked a future regression. I'll let you make the PR for the wider fix
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3762416158
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
+
+
+async def _qaw18_callback():
+ pass
+
+
+CALLBACK_PATH = f"{__name__}.{_qaw18_callback.__name__}"
+
+
+def _clean_db():
+ db.clear_db_dags()
+ db.clear_db_runs()
+ db.clear_db_deadline()
+
+
+@pytest.fixture
+def dagrun(session, dag_maker):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="task_id")
+ with time_machine.travel(DEFAULT_DATE):
+ dag_maker.create_dagrun(state=DagRunState.QUEUED, logical_date=DEFAULT_DATE)
+ session.commit()
+ return session.scalars(select(DagRun)).one()
+
+
+def _make_deadline(session: Session, *, dagrun_id: int, deadline_time, state=None) -> Deadline:
+ deadline = Deadline(
+ deadline_time=deadline_time,
+ callback=AsyncCallback(CALLBACK_PATH),
+ dagrun_id=dagrun_id,
+ dag_id=DAG_ID,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ if state is not None:
+ deadline.callback.state = state
+ session.add(deadline.callback)
+ session.flush()
+ return deadline
+
+
+@pytest.mark.db_test
+class TestPruneDeadlines:
+ @staticmethod
+ def setup_method():
+ _clean_db()
+
Review Comment:
Good point, just removed the _clean_db calls on setup. But something I noticed is that this pattern already exists in the repo:
https://github.com/apache/airflow/blob/main/airflow-core/tests/unit/models/test_deadline_alert.py#L74-L81
Perhaps it's worth it to address the other instances too in another PR?
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #70421:
URL: https://github.com/apache/airflow/pull/70421
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3779214319
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
+
+
+async def _qaw18_callback():
+ pass
+
+
+CALLBACK_PATH = f"{__name__}.{_qaw18_callback.__name__}"
+
+
+def _clean_db():
+ db.clear_db_dags()
+ db.clear_db_runs()
+ db.clear_db_deadline()
+
+
+@pytest.fixture
+def dagrun(session, dag_maker):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="task_id")
+ with time_machine.travel(DEFAULT_DATE):
+ dag_maker.create_dagrun(state=DagRunState.QUEUED, logical_date=DEFAULT_DATE)
+ session.commit()
+ return session.scalars(select(DagRun)).one()
+
+
+def _make_deadline(session: Session, *, dagrun_id: int, deadline_time, state=None) -> Deadline:
+ deadline = Deadline(
+ deadline_time=deadline_time,
+ callback=AsyncCallback(CALLBACK_PATH),
+ dagrun_id=dagrun_id,
+ dag_id=DAG_ID,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ if state is not None:
+ deadline.callback.state = state
+ session.add(deadline.callback)
+ session.flush()
+ return deadline
+
+
+@pytest.mark.db_test
+class TestPruneDeadlines:
+ @staticmethod
+ def setup_method():
+ _clean_db()
+
Review Comment:
Created a new issue for that: https://github.com/apache/airflow/issues/71577
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3732484703
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
Review Comment:
Claude comments?
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
Review Comment:
This whole docstring is odd, but the line number in particular is pretty pointless and will drift in no time, if it hasn't already.
##########
airflow-core/tests/unit/models/test_deadline_alert.py:
##########
@@ -117,6 +117,35 @@ def test_deadline_alert_repr(self, deadline_alert_orm, deadline_reference):
assert "interval=1m" in repr_str
assert repr(deadline_alert_orm.callback_def) in repr_str
+ @pytest.mark.parametrize(
+ ("interval", "expected"),
+ [
+ # Post-0117 shape: interval is the serialized dict, not a bare number.
+ pytest.param(
+ {"__classname__": "datetime.timedelta", "__data__": 7200.0}, "interval=2h", id="timedelta_2h"
+ ),
+ # A corrupted dict without ``__data__`` must still render (no raise) as dynamic.
+ pytest.param({"unexpected": "shape"}, "interval=dynamic", id="corrupted_dict_dynamic"),
+ ],
+ )
+ def test_deadline_alert_repr_does_not_raise_on_json_dict_interval(
+ self, deadline_alert_orm, interval, expected
+ ):
+ """``DeadlineAlert.__repr__`` must never raise for the PRODUCTION (JSON-dict) ``interval`` shape.
Review Comment:
Similar to above, we don't generally have a novel for a test docstring when the test's name is usually pretty self-documenting.
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
Review Comment:
"qaw18"??
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -57,11 +57,14 @@ def __repr__(self):
interval_seconds = None
+ # Legacy rows store a bare number instead of a serialized dict.
if isinstance(self.interval, (int, float)):
interval_seconds = int(self.interval)
- elif isinstance(self.interval, datetime.timedelta):
- interval_seconds = int(self.interval.total_seconds())
+ elif isinstance(self.interval, dict):
+ data = self.interval.get("__data__")
+ if isinstance(data, (int, float)):
+ interval_seconds = int(data)
Review Comment:
It might be safer to check if `__classname__ == "datetime.timedelta"` instead, otherwise I think this could catch a broader scope than intended. I may be overthinking this one though.
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
+
+
+async def _qaw18_callback():
+ pass
+
+
+CALLBACK_PATH = f"{__name__}.{_qaw18_callback.__name__}"
+
+
+def _clean_db():
+ db.clear_db_dags()
+ db.clear_db_runs()
+ db.clear_db_deadline()
+
+
+@pytest.fixture
+def dagrun(session, dag_maker):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="task_id")
+ with time_machine.travel(DEFAULT_DATE):
+ dag_maker.create_dagrun(state=DagRunState.QUEUED, logical_date=DEFAULT_DATE)
+ session.commit()
+ return session.scalars(select(DagRun)).one()
+
+
+def _make_deadline(session: Session, *, dagrun_id: int, deadline_time, state=None) -> Deadline:
+ deadline = Deadline(
+ deadline_time=deadline_time,
+ callback=AsyncCallback(CALLBACK_PATH),
+ dagrun_id=dagrun_id,
+ dag_id=DAG_ID,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ if state is not None:
+ deadline.callback.state = state
+ session.add(deadline.callback)
+ session.flush()
+ return deadline
+
+
+@pytest.mark.db_test
+class TestPruneDeadlines:
+ @staticmethod
+ def setup_method():
+ _clean_db()
+
Review Comment:
If we need to clean the db before our tests, then something else is leaving artifacts that should not be there, this will hide issues.
##########
airflow-core/src/airflow/serialization/definitions/deadline.py:
##########
@@ -320,7 +320,15 @@ def serialize_reference(self) -> dict:
def deserialize_reference(cls, reference_data: dict):
from airflow.serialization.helpers import find_registered_custom_deadline_reference
- custom_class = find_registered_custom_deadline_reference(reference_data["__class_path"])
+ class_path = reference_data.get("__class_path")
+ if not class_path:
+ raise ValueError(
Review Comment:
Better messaging. 👍
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3779236177
##########
airflow-core/tests/unit/models/test_prune_deadlines.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Adversarial QA coverage for ``Deadline.prune_deadlines``.
+
+``prune_deadlines`` is the batch-delete path the scheduler invokes (via
+``DagRun.update_state`` -> ``dagrun.py:1237``) when a DagRun completes on time:
+deadlines that no longer need to fire are removed. These tests drill into the
+on-time/overdue/pending selection logic, the (lack of) batching, callback
+cascade behaviour, and concurrent-mutation edge cases against a real DB.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow.models import DagRun
+from airflow.models.callback import Callback
+from airflow.models.deadline import Deadline
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils import db
+from unit.models import DEFAULT_DATE
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+DAG_ID = "qaw18_prune_dag"
+
+
+async def _qaw18_callback():
+ pass
+
+
+CALLBACK_PATH = f"{__name__}.{_qaw18_callback.__name__}"
+
+
+def _clean_db():
+ db.clear_db_dags()
+ db.clear_db_runs()
+ db.clear_db_deadline()
+
+
+@pytest.fixture
+def dagrun(session, dag_maker):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="task_id")
+ with time_machine.travel(DEFAULT_DATE):
+ dag_maker.create_dagrun(state=DagRunState.QUEUED, logical_date=DEFAULT_DATE)
+ session.commit()
+ return session.scalars(select(DagRun)).one()
+
+
+def _make_deadline(session: Session, *, dagrun_id: int, deadline_time, state=None) -> Deadline:
+ deadline = Deadline(
+ deadline_time=deadline_time,
+ callback=AsyncCallback(CALLBACK_PATH),
+ dagrun_id=dagrun_id,
+ dag_id=DAG_ID,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ if state is not None:
+ deadline.callback.state = state
+ session.add(deadline.callback)
+ session.flush()
+ return deadline
+
+
+@pytest.mark.db_test
+class TestPruneDeadlines:
+ @staticmethod
+ def setup_method():
+ _clean_db()
+
Review Comment:
Are we good to merge this PR?
--
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 deadline serialization, repr, and prune edge cases [airflow]
Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #70421:
URL: https://github.com/apache/airflow/pull/70421#discussion_r3762375636
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -57,11 +57,14 @@ def __repr__(self):
interval_seconds = None
+ # Legacy rows store a bare number instead of a serialized dict.
if isinstance(self.interval, (int, float)):
interval_seconds = int(self.interval)
- elif isinstance(self.interval, datetime.timedelta):
- interval_seconds = int(self.interval.total_seconds())
+ elif isinstance(self.interval, dict):
+ data = self.interval.get("__data__")
+ if isinstance(data, (int, float)):
+ interval_seconds = int(data)
Review Comment:
I think this would just be a style preference. Whether data's interval is of type `timedelta` or `variableinterval`, the dict would be handled correctly.
I've just added a `VariableInterval` case in test `test_deadline_alert_repr_does_not_raise_on_json_dict_interval` to cover this exact situation
--
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