Fix scheduler heartbeat misses caused by slow reschedule dependency check

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-3c93ed6b4ef29ae8f69e3769b59dbd1f2d56ec7c3db7fa86720016d81ffd1058
+8901cbe69c2bc259020d9cde171128b719388d28aa3440c73a5100d23531b6ea

airflow-core/docs/migrations-ref.rst

@@ -43,11 +43,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``53ff648b8a26``        | ``a5a3e5eb9b8d`` | ``3.2.0``         | Add revoked_token table.                                     |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
-| ``a5a3e5eb9b8d``        | ``82dbd68e6171`` | ``3.2.0``         | Make external_executor_id TEXT to allow for longer           |
+| ``a5a3e5eb9b8d``        | ``55297ae24532`` | ``3.2.0``         | Make external_executor_id TEXT to allow for longer           |
 |                         |                  |                   | external_executor_ids.                                       |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
-| ``82dbd68e6171``        | ``55297ae24532`` | ``3.2.0``         | Add index to task_reschedule ti_id .                         |
-+-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``55297ae24532``        | ``e79fc784f145`` | ``3.2.0``         | Add required fields to enable UI integrations for the        |
 |                         |                  |                   | Deadline Alerts feature.                                     |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
@@ -79,7 +77,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 | ``ab6dc0c82d0e``        | ``15d84ca19038`` | ``3.2.0``         | Change ``serialized_dag`` data column to JSONB for           |
 |                         |                  |                   | PostgreSQL.                                                  |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
-| ``15d84ca19038``        | ``cc92b33c6709`` | ``3.2.0``         | replace asset_trigger table with asset_watcher.              |
+| ``15d84ca19038``        | ``82dbd68e6171`` | ``3.2.0``         | replace asset_trigger table with asset_watcher.              |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``82dbd68e6171``        | ``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.                                                          |

airflow-core/src/airflow/migrations/versions/0100_3_2_0_add_index_to_task_reschedule_ti_id.py → airflow-core/src/airflow/migrations/versions/0086_3_1_8_add_index_to_task_reschedule_ti_id.py

@@ -17,10 +17,10 @@
 # under the License.
 
 """
-Add index to task_reschedule ti_id .
+Add composite index (ti_id, id DESC) to task_reschedule.
 
 Revision ID: 82dbd68e6171
-Revises: 55297ae24532
+Revises: cc92b33c6709
 Create Date: 2026-01-22 16:25:42.164449
 
 """
@@ -31,28 +31,46 @@
 
 # revision identifiers, used by Alembic.
 revision = "82dbd68e6171"
-down_revision = "55297ae24532"
+down_revision = "cc92b33c6709"
 branch_labels = None
 depends_on = None
-airflow_version = "3.2.0"
+airflow_version = "3.1.8"
 
 
 def upgrade():
-    """Add index to task_reschedule ti_id."""
-    with op.batch_alter_table("task_reschedule", schema=None) as batch_op:
-        batch_op.create_index("idx_task_reschedule_ti_id", ["ti_id"], unique=False)
+    """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 index from task_reschedule ti_id."""
+    """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")
+            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")
+            batch_op.drop_index("idx_task_reschedule_ti_id_id_desc")

airflow-core/src/airflow/migrations/versions/0086_3_2_0_replace_asset_trigger_table_with_asset.py → airflow-core/src/airflow/migrations/versions/0087_3_2_0_replace_asset_trigger_table_with_asset.py

@@ -20,7 +20,7 @@
 replace asset_trigger table with asset_watcher.
 
 Revision ID: 15d84ca19038
-Revises: cc92b33c6709
+Revises: 82dbd68e6171
 Create Date: 2025-09-14 01:34:40.423767
 
 """
@@ -32,7 +32,7 @@
 
 # revision identifiers, used by Alembic.
 revision = "15d84ca19038"
-down_revision = "cc92b33c6709"
+down_revision = "82dbd68e6171"
 branch_labels = None
 depends_on = None
 airflow_version = "3.2.0"

airflow-core/src/airflow/migrations/versions/0101_3_2_0_make_external_executor_id_text.py

@@ -20,7 +20,7 @@
 Make external_executor_id TEXT to allow for longer external_executor_ids.
 
 Revision ID: a5a3e5eb9b8d
-Revises: 82dbd68e6171
+Revises: 55297ae24532
 Create Date: 2026-01-28 16:35:00.000000
 
 """
@@ -32,7 +32,7 @@
 
 # revision identifiers, used by Alembic.
 revision = "a5a3e5eb9b8d"
-down_revision = "82dbd68e6171"
+down_revision = "55297ae24532"
 branch_labels = None
 depends_on = None
 airflow_version = "3.2.0"

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

@@ -58,7 +58,7 @@ class TaskReschedule(Base):
     duration: Mapped[int] = mapped_column(Integer, nullable=False)
     reschedule_date: Mapped[datetime.datetime] = mapped_column(UtcDateTime, nullable=False)
 
-    __table_args__ = (Index("idx_task_reschedule_ti_id", ti_id),)
+    __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

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