[v3-1-test] Fix scheduler heartbeat misses caused by slow reschedule dependency check (#61983)

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-4827b81c01c9a7993d9796da4990f2a950c2069687db4b0e9cba0c0ae851bd9c
+6ffbbb03f52f967e85a6ac54f1990d4f2fae6fd70e55ef215f867bc691bb06c1

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``cc92b33c6709`` (head) | ``eaf332f43c7c`` | ``3.1.0``         | Add backward compatibility for serialized DAG format v3 to   |
+| ``82dbd68e6171`` (head) | ``cc92b33c6709`` | ``3.1.8``         | Add composite index (ti_id, id DESC) to task_reschedule.     |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``cc92b33c6709``        | ``eaf332f43c7c`` | ``3.1.0``         | Add backward compatibility for serialized DAG format v3 to   |
 |                         |                  |                   | v2.                                                          |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``eaf332f43c7c``        | ``a3c7f2b18d4e`` | ``3.1.0``         | add last_parse_duration to dag model.                        |

airflow-core/src/airflow/migrations/versions/0086_3_1_8_add_index_to_task_reschedule_ti_id.py

@@ -0,0 +1,76 @@
+#
+# 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.
+
+"""
+Add composite index (ti_id, id DESC) to task_reschedule.
+
+Revision ID: 82dbd68e6171
+Revises: cc92b33c6709
+Create Date: 2026-01-22 16:25:42.164449
+
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "82dbd68e6171"
+down_revision = "cc92b33c6709"
+branch_labels = None
+depends_on = None
+airflow_version = "3.1.8"
+
+
+def upgrade():
+    """Add composite (ti_id, id DESC) index to task_reschedule."""
+    dialect_name = op.get_context().dialect.name
+    if dialect_name == "mysql":
+        with op.batch_alter_table("task_reschedule", schema=None) as batch_op:
+            batch_op.drop_constraint("task_reschedule_ti_fkey", type_="foreignkey")
+        op.execute("CREATE INDEX idx_task_reschedule_ti_id_id_desc ON task_reschedule (ti_id, id DESC)")
+        with op.batch_alter_table("task_reschedule", schema=None) as batch_op:
+            batch_op.create_foreign_key(
+                "task_reschedule_ti_fkey", "task_instance", ["ti_id"], ["id"], ondelete="CASCADE"
+            )
+    elif dialect_name == "sqlite":
+        op.execute("CREATE INDEX idx_task_reschedule_ti_id_id_desc ON task_reschedule (ti_id, id DESC)")
+    else:
+        # PostgreSQL
+        with op.batch_alter_table("task_reschedule", schema=None) as batch_op:
+            batch_op.create_index(
+                "idx_task_reschedule_ti_id_id_desc",
+                ["ti_id", "id"],
+                unique=False,
+                postgresql_ops={"id": "DESC"},
+            )
+
+
+def downgrade():
+    """Remove composite index from task_reschedule."""
+    dialect_name = op.get_context().dialect.name
+    if dialect_name == "mysql":
+        with op.batch_alter_table("task_reschedule", schema=None) as batch_op:
+            batch_op.drop_constraint("task_reschedule_ti_fkey", type_="foreignkey")
+            batch_op.drop_index("idx_task_reschedule_ti_id_id_desc")
+            batch_op.create_foreign_key(
+                "task_reschedule_ti_fkey", "task_instance", ["ti_id"], ["id"], ondelete="CASCADE"
+            )
+    else:
+        with op.batch_alter_table("task_reschedule", schema=None) as batch_op:
+            batch_op.drop_index("idx_task_reschedule_ti_id_id_desc")

airflow-core/src/airflow/models/taskreschedule.py

@@ -25,6 +25,7 @@
 from sqlalchemy import (
     Column,
     ForeignKey,
+    Index,
     Integer,
     String,
     asc,
@@ -60,6 +61,8 @@ class TaskReschedule(Base):
     duration = Column(Integer, nullable=False)
     reschedule_date = Column(UtcDateTime, nullable=False)
 
+    __table_args__ = (Index("idx_task_reschedule_ti_id_id_desc", ti_id, id.desc()),)
+
     task_instance = relationship(
         "TaskInstance", primaryjoin="TaskReschedule.ti_id == foreign(TaskInstance.id)", uselist=False
     )

airflow-core/src/airflow/utils/db.py

@@ -111,6 +111,7 @@ class MappedClassProtocol(Protocol):
     "3.0.0": "29ce7909c52b",
     "3.0.3": "fe199e1abd77",
     "3.1.0": "cc92b33c6709",
+    "3.1.8": "82dbd68e6171",
 }