This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 188e2a3aa72b601132344530d89698f99f703005
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 30 12:40:10 2026 +0000
[v3-4-test] Store and show the retry deadline; start retries with empty run
fields (#76027) (#76028)
When a task instance failed and was put up for retry,
`prepare_db_for_next_try`
copied every column of the failed attempt into the new one, so the waiting
attempt showed the previous attempt's start date, end date, duration,
hostname and pid in the API and UI. The retry deadline was computed from
that
copied end_date, so it could not simply be nulled out.
We now store the deadline on the waiting attempt as `retry_at`. The
Execution
API deliberately avoids loading the serialized Dag, and loading it on every
retry just to compute a delay would undo that. A retry reported by a worker
therefore arrives without a deadline, and the scheduler computes and stores
it on its first pass over the run. Until then, and for rows already waiting
at
upgrade time, the deadline is computed from the previous attempt's end date.
Because the delay is fixed at that point, changing retry_delay in a new Dag
version no longer retimes a retry that is already waiting.
The column is ret-conned in tothe 3.4.0 migration that made each attempt
its own
task_instance row, since that is where a waiting retry became a separate row
inheriting the failed attempt's fields. Anyone who has already migrated to
RC1 will need to manually apply this, or recreate their DB:
```sql
-- PostgreSQL
ALTER TABLE task_instance ADD COLUMN retry_at TIMESTAMP WITH TIME ZONE;
-- MySQL
ALTER TABLE task_instance ADD COLUMN retry_at TIMESTAMP(6) NULL;
-- SQLite
ALTER TABLE task_instance ADD COLUMN retry_at DATETIME;
```
The scheduler has always skipped dependency evaluation for retries that
are not yet due. A waiting retry has already passed its trigger rule, so an
upstream that changes state during the wait is only acted on once the retry
is
due or the run checks for a deadlock, and the end state is the same. Doesn't
really affect anything but people messing about with clearing upstream tasks
etc, but the eventually consistent state remains the same before and after
this change, it just might take more time to reach it
`emit_state_change_metric` relied on the copied end_date to skip retries, so
scheduled_duration and queued_duration are now also emitted for retries and
cleared tasks. Each attempt has its own scheduled and queued times, so these
samples are accurate.
(cherry picked from commit 155560df6685ccb3db59f954faf927e80af8059c)
Co-authored-by: Ash Berlin-Taylor <[email protected]>
---
.../core_api/datamodels/task_instance_history.py | 4 +
.../core_api/datamodels/task_instances.py | 5 ++
.../api_fastapi/core_api/datamodels/ui/grid.py | 1 +
.../api_fastapi/core_api/openapi/_private_ui.yaml | 14 ++++
.../core_api/openapi/v2-rest-api-generated.yaml | 16 ++++
.../airflow/api_fastapi/core_api/routes/ui/grid.py | 2 +
.../api_fastapi/core_api/services/ui/grid.py | 5 ++
.../execution_api/routes/task_instances.py | 5 +-
.../src/airflow/jobs/scheduler_job_runner.py | 2 -
.../0142_3_4_0_unify_task_attempt_ownership.py | 3 +-
airflow-core/src/airflow/models/dagrun.py | 12 +++
airflow-core/src/airflow/models/taskinstance.py | 88 ++++++++++++++++++---
.../serialization/definitions/baseoperator.py | 2 +-
.../serialization/definitions/mappedoperator.py | 2 +-
.../ti_deps/deps/not_in_retry_period_dep.py | 6 +-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 38 +++++++++
.../airflow/ui/openapi-gen/requests/types.gen.ts | 9 +++
.../airflow/ui/public/i18n/locales/en/common.json | 1 +
.../ui/src/components/TaskInstanceTooltip.test.tsx | 46 ++++++++++-
.../ui/src/components/TaskInstanceTooltip.tsx | 7 ++
.../ui/src/pages/TaskInstance/Details.test.tsx | 24 ++++++
.../airflow/ui/src/pages/TaskInstance/Details.tsx | 8 ++
.../core_api/routes/public/test_hitl.py | 1 +
.../core_api/routes/public/test_task_instances.py | 92 ++++++++++++++++------
.../api_fastapi/core_api/routes/ui/test_grid.py | 25 ++++++
.../versions/head/test_task_instances.py | 1 +
airflow-core/tests/unit/models/test_dagrun.py | 20 +++++
.../tests/unit/models/test_taskinstance.py | 64 ++++++++++++++-
airflow-core/tests/unit/models/test_trigger.py | 4 +-
.../src/airflowctl/api/datamodels/generated.py | 14 ++++
30 files changed, 470 insertions(+), 51 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py
index 2d810885fe3..a4c7fa489a1 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py
+++
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py
@@ -77,6 +77,10 @@ class TaskInstanceHistoryResponse(OmitsMissingRegion,
BaseModel):
validation_alias="retry_reason",
description=("The reason this try ended, as recorded by a retry
policy."),
)
+ retry_at: datetime | None = Field(
+ default=None,
+ description="When a task instance waiting to be retried becomes due.
Only set in the up_for_retry state.",
+ )
@field_validator("state_reason", mode="after")
@classmethod
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
index a0c2d30d9c4..097bad4159f 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
@@ -110,6 +110,11 @@ class TaskInstanceResponse(OmitsMissingRegion, BaseModel):
),
)
+ retry_at: datetime | None = Field(
+ default=None,
+ description="When a task instance waiting to be retried becomes due.
Only set in the up_for_retry state.",
+ )
+
@field_validator("state_reason", mode="after")
@classmethod
def redact_state_reason(cls, v: str | None) -> str | None:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py
index b584fd7c4ca..c3497eeeaca 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py
@@ -43,6 +43,7 @@ class LightGridTaskInstanceSummary(BaseModel):
loop_iterations_count: int | None = None
latest_region_id: UUID | None = None
latest_region_index: int | None = None
+ retry_at: datetime | None = None
class GridTISummaries(BaseModel):
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index c28d3ff03b7..7869196d863 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -5066,6 +5066,12 @@ components:
- type: integer
- type: 'null'
title: Latest Region Index
+ retry_at:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Retry At
type: object
required:
- task_id
@@ -6003,6 +6009,14 @@ components:
recorded by a retry policy. May describe a previous attempt: a
task waiting
to be retried still carries the reason its last attempt ended. It
is cleared
when the task is cleared and when the next attempt is scheduled or
deferred.'
+ retry_at:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Retry At
+ description: When a task instance waiting to be retried becomes due.
Only
+ set in the up_for_retry state.
type: object
required:
- id
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 7c51d8b4b42..b1c2b5c3407 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -18017,6 +18017,14 @@ components:
- type: 'null'
title: State Reason
description: The reason this try ended, as recorded by a retry
policy.
+ retry_at:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Retry At
+ description: When a task instance waiting to be retried becomes due.
Only
+ set in the up_for_retry state.
type: object
required:
- id
@@ -18224,6 +18232,14 @@ components:
recorded by a retry policy. May describe a previous attempt: a
task waiting
to be retried still carries the reason its last attempt ended. It
is cleared
when the task is cleared and when the next attempt is scheduled or
deferred.'
+ retry_at:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Retry At
+ description: When a task instance waiting to be retried becomes due.
Only
+ set in the up_for_retry state.
type: object
required:
- id
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
index 6eca4ae979a..f6d571eea70 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
@@ -491,6 +491,7 @@ def _build_ti_summaries(
iteration == latest_iteration[(ti.task_id, loop_id, family_id)]
for loop_id, (family_id, iteration) in positions.items()
),
+ retry_at=getattr(ti, "retry_at", None),
)
def get_node_summaries() -> Iterable[dict[str, Any]]:
@@ -597,6 +598,7 @@ def get_grid_ti_summaries_stream(
TaskInstance.dag_version_id,
TaskInstance.start_date,
TaskInstance.end_date,
+ TaskInstance.retry_at,
DagVersion.version_number,
has_note_subq,
)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py
index 2eaa3296aad..fe99d7feb71 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py
@@ -68,6 +68,8 @@ class GridNodeAgg:
has_note: bool = False
loop_iterations: dict[str, set[tuple[UUID, int]]] =
field(default_factory=dict)
latest_region: tuple[UUID, int] | None = None
+ # Only meaningful while the summary holds a single task instance.
+ retry_at: datetime | None = None
def add_ti(
self,
@@ -80,6 +82,7 @@ class GridNodeAgg:
loop_positions: dict[str, tuple[UUID, int]] | None = None,
is_latest_pass: bool = True,
region: tuple[UUID, int] | None = None,
+ retry_at: datetime | None = None,
) -> None:
"""
Merge one task instance row into the summary.
@@ -92,6 +95,7 @@ class GridNodeAgg:
does not supply the coordinate: a link has to address the newest row.
"""
self.child_states[state] += 1
+ self.retry_at = retry_at if self.child_states.total() == 1 else None
# A looped task has one row per iteration, so a link without a
coordinate cannot say
# which one it means. Carry the newest so the grid can address it.
if is_latest_pass and region is not None:
@@ -254,6 +258,7 @@ def _find_aggregates(
"type": "task",
"parent_id": parent_id,
**_get_aggs_for_node(summary),
+ "retry_at": summary.retry_at,
},
summary,
)
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index bb3f8fdf20d..a20580ece0a 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -284,8 +284,8 @@ def ti_run(
)
# One sample per queue wait, not per try: the scheduler refreshes
queued_dttm on every
# queueing, so a retry and a resume from deferral each waited for a
slot of their own.
- # task.scheduled_duration skips retries (emit_state_change_metric
returns early while
- # end_date is set), so the two disagree on retries by design.
+ # task.scheduled_duration is one sample per try
(emit_state_change_metric returns early once
+ # the try has an end_date, as after a sensor reschedule), so the two
disagree on resumes by design.
# queued_dttm is unset only for runs that skip the scheduler's
queueing, e.g. dag.test().
emit_queued_duration = ti.queued_dttm is not None
@@ -298,6 +298,7 @@ def ti_run(
state=TaskInstanceState.RUNNING,
last_heartbeat_at=timezone.utcnow(),
retry_delay_override=None,
+ retry_at=None,
retry_reason=None,
)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 3817bcd450e..8a29b83c2e6 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -3791,8 +3791,6 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
reset_tis_message.append(repr(ti))
successor = ti.prepare_db_for_next_try(session=session)
successor.state = None
- successor.queued_by_job_id = None
- successor.external_executor_id = None
successor.clear_next_method_args()
for ti in set(tis_to_adopt_or_reset) - set(to_reset):
diff --git
a/airflow-core/src/airflow/migrations/versions/0142_3_4_0_unify_task_attempt_ownership.py
b/airflow-core/src/airflow/migrations/versions/0142_3_4_0_unify_task_attempt_ownership.py
index 41a1e0e055a..3f19e561a33 100644
---
a/airflow-core/src/airflow/migrations/versions/0142_3_4_0_unify_task_attempt_ownership.py
+++
b/airflow-core/src/airflow/migrations/versions/0142_3_4_0_unify_task_attempt_ownership.py
@@ -233,6 +233,7 @@ def upgrade():
with op.batch_alter_table("task_instance") as batch:
batch.add_column(sa.Column("working_set", sa.Boolean(),
nullable=True, server_default=sa.true()))
batch.add_column(sa.Column("archived_reason", sa.String(50),
nullable=True))
+ batch.add_column(sa.Column("retry_at", UtcDateTime(),
nullable=True))
batch.drop_constraint("task_instance_composite_key",
type_="unique")
batch.create_unique_constraint("task_instance_current_key",
[*_COORDINATES, "working_set"])
batch.create_unique_constraint("task_instance_try_key",
[*_COORDINATES, "try_number"])
@@ -598,7 +599,7 @@ def downgrade():
batch.drop_constraint("task_instance_current_key", type_="unique")
batch.drop_constraint("task_instance_try_key", type_="unique")
batch.create_unique_constraint("task_instance_composite_key",
list(_COORDINATES))
- for column in ("working_set", "archived_reason"):
+ for column in ("working_set", "archived_reason", "retry_at"):
batch.drop_column(column)
with op.batch_alter_table("task_instance_history") as batch:
batch.create_foreign_key(
diff --git a/airflow-core/src/airflow/models/dagrun.py
b/airflow-core/src/airflow/models/dagrun.py
index 206993e1b66..810e308fb59 100644
--- a/airflow-core/src/airflow/models/dagrun.py
+++ b/airflow-core/src/airflow/models/dagrun.py
@@ -1773,10 +1773,20 @@ class DagRun(Base, LoggingMixin):
expansion_happened = False
# (task id, region id) pairs for which _revise_map_indexes_if_mapped
was already done
revised_expansion_keys: set[tuple[str, UUID]] = set()
+ now = timezone.utcnow()
for schedulable in itertools.chain(schedulable_tis, additional_tis):
if TYPE_CHECKING:
assert isinstance(schedulable.task, Operator)
old_state = schedulable.state
+ if old_state == TaskInstanceState.UP_FOR_RETRY:
+ # Worker-reported retries arrive without one; storing it
spares later passes the
+ # previous-attempt lookup.
+ if schedulable.retry_at is None:
+ schedulable.retry_at = schedulable.next_retry_datetime()
+ # Not due yet, so no dependency of it can pass;
_are_premature_tis still counts it
+ # as runnable when the run looks for a deadlock.
+ if schedulable.retry_at >= now:
+ continue
if not schedulable.are_dependencies_met(session=session,
dep_context=dep_context):
old_states[schedulable.id] = old_state
continue
@@ -2720,6 +2730,7 @@ class DagRun(Base, LoggingMixin):
# Already archived with the finished try; the new one
must not inherit it.
retry_reason=None,
retry_delay_override=None,
+ retry_at=None,
)
.execution_options(synchronize_session=False)
)
@@ -2772,6 +2783,7 @@ class DagRun(Base, LoggingMixin):
try_number=next_try_number,
retry_reason=None,
retry_delay_override=None,
+ retry_at=None,
)
.execution_options(
synchronize_session=False,
diff --git a/airflow-core/src/airflow/models/taskinstance.py
b/airflow-core/src/airflow/models/taskinstance.py
index 6905ae965d8..974690b9382 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -144,6 +144,20 @@ from airflow.utils.state import DagRunState, State,
TaskInstanceState
TR = TaskReschedule
log = logging.getLogger(__name__)
+
+# Columns describing one attempt's execution; a successor attempt starts with
them empty.
+_PER_ATTEMPT_COLUMNS = (
+ "start_date",
+ "end_date",
+ "duration",
+ "hostname",
+ "unixname",
+ "pid",
+ "queued_dttm",
+ "scheduled_dttm",
+ "queued_by_job_id",
+ "last_heartbeat_at",
+)
tracer = trace.get_tracer(__name__)
if TYPE_CHECKING:
@@ -558,6 +572,7 @@ def clear_task_instances(
# the task's own retry_delay, so a stale value would retime the
next retry.
ti.retry_reason = None
ti.retry_delay_override = None
+ ti.retry_at = None
ti.clear_next_method_args()
# Match DagVersion to latest serialized DAG when running on the
latest version.
if use_latest_version:
@@ -1191,6 +1206,9 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
# Read by next_retry_datetime().
retry_delay_override: Mapped[float | None] = mapped_column(Float,
nullable=True)
retry_reason: Mapped[str | None] = mapped_column(String(500),
nullable=True)
+ # When an UP_FOR_RETRY attempt becomes due; written by the scheduler on
its first pass after the
+ # attempt is created.
+ retry_at: Mapped[datetime | None] = mapped_column(UtcDateTime,
nullable=True)
__table_args__ = (
Index("ti_dag_state", dag_id, state),
@@ -1628,6 +1646,7 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
# prepare_db_for_next_try() would otherwise archive them under
this try number.
self.retry_reason = None
self.retry_delay_override = None
+ self.retry_at = None
if self.state in State.finished or self.state ==
TaskInstanceState.UP_FOR_RETRY:
self.end_date = self.end_date or current_time
self.duration = (self.end_date - self.start_date).total_seconds()
@@ -1755,12 +1774,22 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
"""Archive this UUID and return its successor in the caller's
transaction."""
successor_state = self.state
dag_version_id = self.dag_version_id
+ retry_at = None
+ # The execution API doesn't load the task, so a worker-reported retry
without a policy
+ # override is left for DagRun._get_ready_tis.
+ if successor_state == TaskInstanceState.UP_FOR_RETRY and (
+ self.retry_delay_override is not None or self.task is not None
+ ):
+ retry_at = self._compute_retry_deadline(
+ failed_try_number=self.try_number,
failed_end_date=self.end_date
+ )
self.archive(reason="retry", session=session)
values = {
attribute.columns[0].name: getattr(self, attribute.key)
for attribute in inspect(TaskInstance).column_attrs
}
values.update(
+ {name: None for name in _PER_ATTEMPT_COLUMNS},
id=uuid7(),
try_number=self.try_number + 1,
working_set=True,
@@ -1769,6 +1798,7 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
external_executor_id=None,
dag_version_id=dag_version_id,
state=successor_state,
+ retry_at=retry_at,
)
session.execute(insert(TaskInstance.__table__).values(values))
successor = session.scalars(select(TaskInstance).where(TaskInstance.id
== values["id"])).one()
@@ -2060,28 +2090,60 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
return False
return region is not None and region.node_id == self.task_id and
region.parent_region_id is None
- def next_retry_datetime(self):
+ def next_retry_datetime(self, *, session: Session | None = None) ->
datetime:
"""
Get datetime of the next retry if the task instance fails.
+ ``retry_at`` is usually written by the scheduler on its first pass
after an
+ ``UP_FOR_RETRY`` attempt is created. Until then, or for a row from
before that column, the
+ deadline is computed from the previous attempt's end date.
+ """
+ if self.state == TaskInstanceState.UP_FOR_RETRY:
+ if self.retry_at is not None:
+ return self.retry_at
+ return self._compute_retry_deadline(
+ failed_try_number=self.try_number - 1,
+
failed_end_date=self._get_previous_attempt_end_date(session=session) or
self.end_date,
+ )
+ return self._compute_retry_deadline(failed_try_number=self.try_number,
failed_end_date=self.end_date)
+
+ def _get_previous_attempt_end_date(self, *, session: Session | None) ->
datetime | None:
+ session = session or inspect(self).session
+ if session is None:
+ return None
+ return session.scalar(
+ select(TaskInstance.end_date)
+ .where(
+ TaskInstance.dag_id == self.dag_id,
+ TaskInstance.run_id == self.run_id,
+ TaskInstance.task_id == self.task_id,
+ TaskInstance.region_id == self.region_id,
+ TaskInstance.region_index == self.region_index,
+ TaskInstance.try_number == self.try_number - 1,
+ )
+ .execution_options(include_all_attempts=True)
+ )
+
+ def _compute_retry_deadline(
+ self, *, failed_try_number: int, failed_end_date: datetime | None
+ ) -> datetime:
+ """
+ Compute when the attempt after ``failed_try_number`` may start.
+
When a :class:`~airflow.sdk.definitions.retry_policy.RetryPolicy` has
- overridden the delay, ``retry_delay_override`` is stored on the task
- instance row and takes precedence over the static ``retry_delay`` /
- exponential-backoff calculation.
+ overridden the delay, ``retry_delay_override`` takes precedence over
the
+ static ``retry_delay`` / exponential-backoff calculation.
For exponential backoff, retry_delay is used as base and will be
converted to seconds.
"""
- # Check for a policy-driven delay override.
+ base = failed_end_date if failed_end_date is not None else
timezone.utcnow()
if self.retry_delay_override is not None:
- base = self.end_date if self.end_date is not None else
timezone.utcnow()
return base + timedelta(seconds=self.retry_delay_override)
from airflow.sdk.definitions._internal.abstractoperator import
MAX_RETRY_DELAY
- # While waiting for retry, the live row already identifies the next
attempt.
- failed_try_number = (
- self.try_number - 1 if self.state ==
TaskInstanceState.UP_FOR_RETRY else self.try_number
- )
+ if TYPE_CHECKING:
+ assert self.task is not None
delay = self.task.retry_delay
multiplier = self.task.retry_exponential_backoff if
self.task.retry_exponential_backoff != 0 else 1.0
if multiplier != 1.0 and multiplier > 0:
@@ -2124,7 +2186,7 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
delay = timedelta(seconds=delay_backoff_in_seconds)
if self.task.max_retry_delay:
delay = min(self.task.max_retry_delay, delay)
- return self.end_date + delay
+ return base + delay
def ready_for_retry(self) -> bool:
"""Check on whether the task instance is in the right state and
timeframe to be retried."""
@@ -2353,8 +2415,7 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
Supported states: QUEUED and RUNNING
"""
if self.end_date:
- # if the task has an end date, it means that this is not its first
round.
- # we send the state transition time metric only on the first try,
otherwise it gets more complex.
+ # This try already ran once (e.g. a sensor reschedule); only its
first scheduling is timed.
return
# switch on state and deduce which metric to send
@@ -2681,6 +2742,7 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
# A start_from_trigger retry skips schedule_tis' UPDATE, so clear
here too.
self.retry_reason = None
self.retry_delay_override = None
+ self.retry_at = None
self.trigger_id = trigger_row.id
self.next_method = start_trigger_args.next_method
self.next_kwargs =
stringify_encoding_keys(start_trigger_args.next_kwargs or {})
diff --git a/airflow-core/src/airflow/serialization/definitions/baseoperator.py
b/airflow-core/src/airflow/serialization/definitions/baseoperator.py
index 4d5647a8555..9f41bde935d 100644
--- a/airflow-core/src/airflow/serialization/definitions/baseoperator.py
+++ b/airflow-core/src/airflow/serialization/definitions/baseoperator.py
@@ -103,7 +103,7 @@ class SerializedBaseOperator(DAGNode):
map_index_template: str | None = None
max_active_tis_per_dag: int | None = None
max_active_tis_per_dagrun: int | None = None
- max_retry_delay: datetime.timedelta | float | None = None
+ max_retry_delay: datetime.timedelta | None = None
multiple_outputs: bool = False
# Boolean flags for callback existence
diff --git
a/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
b/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
index 4033152abbb..aea53d1bd47 100644
--- a/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
+++ b/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
@@ -297,7 +297,7 @@ class SerializedMappedOperator(DAGNode):
return float(value)
@property
- def max_retry_delay(self) -> datetime.timedelta | float | None:
+ def max_retry_delay(self) -> datetime.timedelta | None:
return self._get_partial_kwargs_or_operator_default("max_retry_delay")
@property
diff --git a/airflow-core/src/airflow/ti_deps/deps/not_in_retry_period_dep.py
b/airflow-core/src/airflow/ti_deps/deps/not_in_retry_period_dep.py
index ce20fddf40c..0af1adfb292 100644
--- a/airflow-core/src/airflow/ti_deps/deps/not_in_retry_period_dep.py
+++ b/airflow-core/src/airflow/ti_deps/deps/not_in_retry_period_dep.py
@@ -42,11 +42,9 @@ class NotInRetryPeriodDep(BaseTIDep):
yield self._passing_status(reason="The task instance was not
marked for retrying.")
return
- # Calculate the date first so that it is always smaller than the
timestamp used by
- # ready_for_retry
cur_date = timezone.utcnow()
- next_task_retry_date = ti.next_retry_datetime()
- if ti.is_premature:
+ next_task_retry_date = ti.next_retry_datetime(session=session)
+ if next_task_retry_date >= cur_date:
yield self._failing_status(
reason=(
f"Task is not ready for retry yet but will be retried
automatically. "
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 32669dad1aa..8ba458c15dc 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -8200,6 +8200,19 @@ export const $TaskInstanceHistoryResponse = {
],
title: 'State Reason',
description: 'The reason this try ended, as recorded by a retry
policy.'
+ },
+ retry_at: {
+ anyOf: [
+ {
+ type: 'string',
+ format: 'date-time'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Retry At',
+ description: 'When a task instance waiting to be retried becomes
due. Only set in the up_for_retry state.'
}
},
type: 'object',
@@ -8535,6 +8548,19 @@ export const $TaskInstanceResponse = {
],
title: 'State Reason',
description: 'The reason the task instance reached its current
state, as recorded by a retry policy. May describe a previous attempt: a task
waiting to be retried still carries the reason its last attempt ended. It is
cleared when the task is cleared and when the next attempt is scheduled or
deferred.'
+ },
+ retry_at: {
+ anyOf: [
+ {
+ type: 'string',
+ format: 'date-time'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Retry At',
+ description: 'When a task instance waiting to be retried becomes
due. Only set in the up_for_retry state.'
}
},
type: 'object',
@@ -12318,6 +12344,18 @@ export const $LightGridTaskInstanceSummary = {
}
],
title: 'Latest Region Index'
+ },
+ retry_at: {
+ anyOf: [
+ {
+ type: 'string',
+ format: 'date-time'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Retry At'
}
},
type: 'object',
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index c34a9a1dfee..6f088ae943a 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -2232,6 +2232,10 @@ export type TaskInstanceHistoryResponse = {
* The reason this try ended, as recorded by a retry policy.
*/
state_reason?: string | null;
+ /**
+ * When a task instance waiting to be retried becomes due. Only set in the
up_for_retry state.
+ */
+ retry_at?: string | null;
};
/**
@@ -2283,6 +2287,10 @@ export type TaskInstanceResponse = {
* The reason the task instance reached its current state, as recorded by
a retry policy. May describe a previous attempt: a task waiting to be retried
still carries the reason its last attempt ended. It is cleared when the task is
cleared and when the next attempt is scheduled or deferred.
*/
state_reason?: string | null;
+ /**
+ * When a task instance waiting to be retried becomes due. Only set in the
up_for_retry state.
+ */
+ retry_at?: string | null;
};
/**
@@ -3179,6 +3187,7 @@ export type LightGridTaskInstanceSummary = {
loop_iterations_count?: number | null;
latest_region_id?: string | null;
latest_region_index?: number | null;
+ retry_at?: string | null;
};
/**
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
index c12d426d6a6..7d55f915642 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
@@ -487,6 +487,7 @@
"queue": "Queue",
"queuedWhen": "Queued At",
"renderedMapIndex": "Rendered Map Index",
+ "retryAt": "Retry At",
"scheduledWhen": "Scheduled At",
"stateReason": "Reason for state",
"stateReasonSummary": {
diff --git
a/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.test.tsx
b/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.test.tsx
index e667c59f86f..692182b3903 100644
--- a/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.test.tsx
+++ b/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.test.tsx
@@ -20,7 +20,11 @@ import "@testing-library/jest-dom/vitest";
import { render, screen } from "@testing-library/react";
import { describe, expect, it, vi } from "vitest";
-import type { LightGridTaskInstanceSummary, TaskInstanceResponse } from
"openapi/requests/types.gen";
+import type {
+ LightGridTaskInstanceSummary,
+ TaskInstanceHistoryResponse,
+ TaskInstanceResponse,
+} from "openapi/requests/types.gen";
import { Wrapper } from "src/utils/Wrapper";
@@ -34,6 +38,18 @@ vi.mock("react-i18next", () => ({
}),
}));
+const buildTry = (overrides: Partial<TaskInstanceHistoryResponse>) =>
+ ({
+ dag_id: "test",
+ dag_run_id: "test",
+ duration: null,
+ end_date: null,
+ start_date: null,
+ task_id: "my_task",
+ try_number: 2,
+ ...overrides,
+ }) as unknown as TaskInstanceHistoryResponse;
+
describe("TaskInstanceTooltip", () => {
it("renders children directly when both taskInstance and tooltip are
undefined", () => {
render(
@@ -291,4 +307,32 @@ describe("TaskInstanceTooltip", () => {
// two neighbouring bars can both be a first try and the number is what
tells them apart.
expect(screen.getByText(/tryNumber/iu).textContent).toContain("1");
});
+
+ it("shows when a try waiting to be retried becomes due", () => {
+ render(
+ <TaskInstanceTooltip
+ open
+ taskInstance={buildTry({ retry_at: "2026-10-29T12:00:30Z", state:
"up_for_retry" })}
+ >
+ <span>trigger</span>
+ </TaskInstanceTooltip>,
+ { wrapper: Wrapper },
+ );
+
+ expect(screen.getByText(/taskInstance\.retryAt/u)).toBeInTheDocument();
+ });
+
+ it.each([
+ { retry_at: null, state: "up_for_retry" },
+ { retry_at: "2026-10-29T12:00:30Z", state: "failed" },
+ ] as const)("hides retry at for a $state try with retry_at $retry_at",
(overrides) => {
+ render(
+ <TaskInstanceTooltip open taskInstance={buildTry(overrides)}>
+ <span>trigger</span>
+ </TaskInstanceTooltip>,
+ { wrapper: Wrapper },
+ );
+
+
expect(screen.queryByText(/taskInstance\.retryAt/u)).not.toBeInTheDocument();
+ });
});
diff --git a/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.tsx
b/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.tsx
index 8df0332b122..4c57757ab05 100644
--- a/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.tsx
+++ b/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.tsx
@@ -150,6 +150,13 @@ const TaskInstanceTooltip = ({
</Text>
</>
) : undefined}
+ {"retry_at" in taskInstance &&
+ taskInstance.state === "up_for_retry" &&
+ Boolean(taskInstance.retry_at) ? (
+ <Text>
+ {translate("taskInstance.retryAt")}: <Time
datetime={taskInstance.retry_at} />
+ </Text>
+ ) : undefined}
{"min_start_date" in taskInstance && taskInstance.min_start_date
!== null ? (
<>
<Text>
diff --git
a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx
b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx
index bcd4182f5bf..9e01ae2cd25 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.test.tsx
@@ -136,6 +136,30 @@ describe("Details state reason row", () => {
});
});
+describe("Details retry at row", () => {
+ beforeEach(() => {
+ i18n.addResourceBundle("en", "common", commonLocale, true, true);
+ });
+
+ it("shows when a try waiting to be retried becomes due", () => {
+ renderDetails(buildTaskInstance({ retry_at: "2026-10-29T12:00:30Z", state:
"up_for_retry" }));
+
+ const row =
screen.getByText(i18n.t("common:taskInstance.retryAt")).closest("tr");
+
+ expect(row).not.toBeNull();
+ expect(within(row as
HTMLElement).getByTestId("time-display")).toHaveAttribute(
+ "datetime",
+ "2026-10-29T12:00:30Z",
+ );
+ });
+
+ it("omits the row when the try has no retry_at", () => {
+ renderDetails(buildTaskInstance({ retry_at: null, state: "up_for_retry"
}));
+
+
expect(screen.queryByText(i18n.t("common:taskInstance.retryAt"))).not.toBeInTheDocument();
+ });
+});
+
describe("Details loop context", () => {
beforeEach(() => {
i18n.addResourceBundle("en", "common", commonLocale, true, true);
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx
b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx
index 05314739c2b..129fa910d13 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Details.tsx
@@ -252,6 +252,14 @@ export const Details = () => {
<Time datetime={tryInstance?.end_date} />
</Table.Cell>
</Table.Row>
+ {Boolean(tryInstance?.retry_at) ? (
+ <Table.Row>
+ <Table.Cell>{translate("taskInstance.retryAt")}</Table.Cell>
+ <Table.Cell>
+ <Time datetime={tryInstance?.retry_at} />
+ </Table.Cell>
+ </Table.Row>
+ ) : undefined}
<Table.Row>
<Table.Cell>{translate("taskInstance.dagVersion")}</Table.Cell>
<Table.Cell>
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
index 221cf9f95f9..bcc836f491b 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
@@ -427,6 +427,7 @@ def expected_sample_hitl_detail_dict(sample_ti:
TaskInstance) -> dict[str, Any]:
"task_id": TASK_ID,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
"trigger": None,
"triggerer_job": None,
"try_number": 0,
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
index ddb0776e8b8..eb47b72983f 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
@@ -493,6 +493,7 @@ class TestGetTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
def test_should_include_state_reason(self, test_client, session):
@@ -612,6 +613,7 @@ class TestGetTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
"dag_version": {
"id": response_data["dag_version"]["id"],
"version_number": expected_version_number,
@@ -711,6 +713,7 @@ class TestGetTaskInstance(TestTaskInstanceEndpoint):
},
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
def test_should_respond_200_with_task_state_in_removed(self, test_client,
session):
@@ -768,6 +771,7 @@ class TestGetTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
def test_should_respond_200_task_instance_with_rendered(self, test_client,
session):
@@ -828,6 +832,7 @@ class TestGetTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
def test_raises_404_for_nonexistent_task_instance(self, test_client):
@@ -964,6 +969,7 @@ class TestGetMappedTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
def test_should_respond_401(self, unauthenticated_test_client):
@@ -3225,6 +3231,7 @@ class TestGetTaskInstanceTry(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
}
def test_should_include_state_reason_from_history(self, test_client,
session):
@@ -3273,31 +3280,31 @@ class TestGetTaskInstanceTry(TestTaskInstanceEndpoint):
assert response_data == {
"dag_id": "example_python_operator",
"dag_display_name": "example_python_operator",
- "duration": 10000.0,
- "end_date": "2020-01-03T00:00:00Z",
+ "duration": 10000.0 if try_number == 1 else None,
+ "end_date": "2020-01-03T00:00:00Z" if try_number == 1 else None,
"executor": None,
"executor_config": "{}",
"id": mock.ANY,
"note": mock.ANY,
- "hostname": "",
+ "hostname": "" if try_number == 1 else None,
"map_index": -1,
"loop_iteration": None,
"max_tries": 0 if try_number == 1 else 1,
"operator": "PythonOperator",
"operator_name": "PythonOperator",
- "pid": 100,
+ "pid": 100 if try_number == 1 else None,
"pool": "default_pool",
"pool_slots": 1,
"priority_weight": 14,
"queue": "default_queue",
"queued_when": None,
"scheduled_when": None,
- "start_date": "2020-01-02T00:00:00Z",
+ "start_date": "2020-01-02T00:00:00Z" if try_number == 1 else None,
"state": "success" if try_number == 1 else None,
"task_id": "print_the_context",
"task_display_name": "print_the_context",
"try_number": try_number,
- "unixname": getuser(),
+ "unixname": getuser() if try_number == 1 else None,
"dag_run_id": "TEST_DAG_RUN_ID",
"dag_version": {
"bundle_name":
"apache-airflow-providers-standard-example-dags",
@@ -3310,6 +3317,7 @@ class TestGetTaskInstanceTry(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
}
@pytest.mark.parametrize("try_number", [1, 2])
@@ -3357,31 +3365,31 @@ class TestGetTaskInstanceTry(TestTaskInstanceEndpoint):
assert response_data == {
"dag_id": "example_python_operator",
"dag_display_name": "example_python_operator",
- "duration": 10000.0,
- "end_date": "2020-01-03T00:00:00Z",
+ "duration": 10000.0 if try_number == 1 else None,
+ "end_date": "2020-01-03T00:00:00Z" if try_number == 1 else
None,
"executor": None,
"executor_config": "{}",
"id": mock.ANY,
"note": mock.ANY,
- "hostname": "",
+ "hostname": "" if try_number == 1 else None,
"map_index": map_index,
"loop_iteration": None,
"max_tries": 0 if try_number == 1 else 1,
"operator": "PythonOperator",
"operator_name": "PythonOperator",
- "pid": 100,
+ "pid": 100 if try_number == 1 else None,
"pool": "default_pool",
"pool_slots": 1,
"priority_weight": 14,
"queue": "default_queue",
"queued_when": None,
"scheduled_when": None,
- "start_date": "2020-01-02T00:00:00Z",
+ "start_date": "2020-01-02T00:00:00Z" if try_number == 1 else
None,
"state": "failed" if try_number == 1 else None,
"task_id": "print_the_context",
"task_display_name": "print_the_context",
"try_number": try_number,
- "unixname": getuser(),
+ "unixname": getuser() if try_number == 1 else None,
"dag_run_id": "TEST_DAG_RUN_ID",
"dag_version": {
"bundle_name":
"apache-airflow-providers-standard-example-dags",
@@ -3394,6 +3402,7 @@ class TestGetTaskInstanceTry(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
}
def test_should_respond_200_with_task_state_in_deferred(self, test_client,
session):
@@ -3464,6 +3473,7 @@ class TestGetTaskInstanceTry(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
}
def test_should_respond_200_with_task_state_in_removed(self, test_client,
session):
@@ -3515,6 +3525,7 @@ class TestGetTaskInstanceTry(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
}
def test_should_respond_401(self, unauthenticated_test_client):
@@ -3594,6 +3605,7 @@ class TestGetTaskInstanceTry(TestTaskInstanceEndpoint):
"dag_display_name": "dag_with_multiple_versions",
},
"state_reason": None,
+ "retry_at": None,
}
def test_should_not_return_duplicate_runs(self, test_client, session):
@@ -4862,6 +4874,7 @@ class
TestPostClearTaskInstances(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
"try_number": 1,
"unixname": getuser(),
},
@@ -5537,6 +5550,24 @@ class
TestPostClearTaskInstances(TestTaskInstanceEndpoint):
class TestGetTaskInstanceTries(TestTaskInstanceEndpoint):
+ def test_waiting_retry_returns_retry_at(self, test_client, session):
+ self.create_task_instances(
+ session=session, task_instances=[{"state": State.SUCCESS}],
with_ti_history=True
+ )
+ current =
session.scalar(select(TaskInstance).where(TaskInstance.working_set.is_(True)))
+ current.state = State.UP_FOR_RETRY
+ current.retry_at = datetime(2020, 1, 3, 0, 10)
+ session.commit()
+
+ response = test_client.get(
+
"/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context/tries"
+ )
+
+ assert response.status_code == 200
+ attempts = {item["try_number"]: item for item in
response.json()["task_instances"]}
+ assert attempts[1]["retry_at"] is None
+ assert attempts[2]["retry_at"] == "2020-01-03T00:10:00Z"
+
def test_history_without_dag_version_serializes_null(self, test_client,
session):
self.create_task_instances(
session=session, task_instances=[{"state": State.SUCCESS}],
with_ti_history=True
@@ -5612,35 +5643,36 @@ class
TestGetTaskInstanceTries(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
},
{
"dag_id": "example_python_operator",
"dag_display_name": "example_python_operator",
- "duration": 10000.0,
- "end_date": "2020-01-03T00:00:00Z",
+ "duration": None,
+ "end_date": None,
"executor": None,
"executor_config": "{}",
"id": mock.ANY,
"note": mock.ANY,
- "hostname": "",
+ "hostname": None,
"map_index": -1,
"loop_iteration": None,
"max_tries": 1,
"operator": "PythonOperator",
"operator_name": "PythonOperator",
- "pid": 100,
+ "pid": None,
"pool": "default_pool",
"pool_slots": 1,
"priority_weight": 14,
"queue": "default_queue",
"queued_when": None,
"scheduled_when": None,
- "start_date": "2020-01-02T00:00:00Z",
+ "start_date": None,
"state": None,
"task_id": "print_the_context",
"task_display_name": "print_the_context",
"try_number": 2,
- "unixname": getuser(),
+ "unixname": None,
"dag_run_id": "TEST_DAG_RUN_ID",
"dag_version": {
"bundle_name":
"apache-airflow-providers-standard-example-dags",
@@ -5653,6 +5685,7 @@ class TestGetTaskInstanceTries(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
},
],
"total_entries": 2,
@@ -5772,35 +5805,36 @@ class
TestGetTaskInstanceTries(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
},
{
"dag_id": "example_python_operator",
"dag_display_name": "example_python_operator",
- "duration": 10000.0,
- "end_date": "2020-01-03T00:00:00Z",
+ "duration": None,
+ "end_date": None,
"executor": None,
"executor_config": "{}",
"id": mock.ANY,
"note": mock.ANY,
- "hostname": "",
+ "hostname": None,
"map_index": map_index,
"loop_iteration": None,
"max_tries": 1,
"operator": "PythonOperator",
"operator_name": "PythonOperator",
- "pid": 100,
+ "pid": None,
"pool": "default_pool",
"pool_slots": 1,
"priority_weight": 14,
"queue": "default_queue",
"queued_when": None,
"scheduled_when": None,
- "start_date": "2020-01-02T00:00:00Z",
+ "start_date": None,
"state": None,
"task_id": "print_the_context",
"task_display_name": "print_the_context",
"try_number": 2,
- "unixname": getuser(),
+ "unixname": None,
"dag_run_id": "TEST_DAG_RUN_ID",
"dag_version": {
"bundle_name":
"apache-airflow-providers-standard-example-dags",
@@ -5813,6 +5847,7 @@ class TestGetTaskInstanceTries(TestTaskInstanceEndpoint):
"version_number": 1,
},
"state_reason": None,
+ "retry_at": None,
},
],
"total_entries": 2,
@@ -5884,6 +5919,7 @@ class TestGetTaskInstanceTries(TestTaskInstanceEndpoint):
"dag_display_name": "dag_with_multiple_versions",
},
"state_reason": None,
+ "retry_at": None,
}
@@ -6651,6 +6687,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
],
"total_entries": 1,
@@ -6932,6 +6969,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
],
"total_entries": 1,
@@ -7075,6 +7113,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
],
"total_entries": 1,
@@ -7143,6 +7182,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
],
"total_entries": 1,
@@ -7258,6 +7298,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
],
"total_entries": 1,
@@ -7347,6 +7388,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
_check_task_instance_note(
@@ -7546,6 +7588,7 @@ class
TestPatchTaskInstanceDryRun(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
],
"total_entries": 1,
@@ -7839,6 +7882,7 @@ class
TestPatchTaskInstanceDryRun(TestTaskInstanceEndpoint):
"triggerer_job": None,
"team_name": None,
"state_reason": None,
+ "retry_at": None,
}
],
"total_entries": 1,
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
index 65e46cd11de..cfb8e122cec 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
@@ -888,6 +888,7 @@ class TestGetGridDataEndpoint:
ti["loop_iterations_count"] = None
ti["latest_region_id"] = None
ti["latest_region_index"] = None
+ ti["retry_at"] = None
for obj in actual, expected:
tis = obj["task_instances"]
tis[:] = sorted(tis, key=lambda x: x["task_id"])
@@ -997,6 +998,7 @@ class TestGetGridDataEndpoint:
ti["loop_iterations_count"] = None
ti["latest_region_id"] = None
ti["latest_region_index"] = None
+ ti["retry_at"] = None
expected = sort_dict(expected)
actual = sort_dict(actual)
assert actual == expected
@@ -1270,6 +1272,29 @@ class TestGetGridDataEndpoint:
assert by_task_id[TASK_ID]["has_note"] is True
assert all(not ti["has_note"] for task_id, ti in by_task_id.items() if
task_id != TASK_ID)
+ def test_grid_ti_summaries_retry_at_only_for_a_single_task_instance(self,
session, test_client):
+ retry_at = pendulum.DateTime(2024, 12, 30, 3, 0, 0,
tzinfo=pendulum.UTC)
+ waiting = session.scalars(
+ select(TaskInstance).where(
+ TaskInstance.run_id == "run_2",
+ (TaskInstance.task_id == TASK_ID)
+ | ((TaskInstance.task_id == "mapped_task_group.subtask") &
(TaskInstance.map_index == 2)),
+ )
+ ).all()
+ for ti in waiting:
+ ti.state = TaskInstanceState.UP_FOR_RETRY
+ ti.retry_at = retry_at
+ session.commit()
+
+ response =
test_client.get(f"/grid/ti_summaries/{DAG_ID}?run_ids=run_2")
+ assert response.status_code == 200
+ [data] = self._parse_ndjson(response)
+ by_task_id = {ti["task_id"]: ti for ti in data["task_instances"]}
+
+ assert by_task_id[TASK_ID]["retry_at"] == "2024-12-30T03:00:00Z"
+ assert by_task_id["mapped_task_group.subtask"]["retry_at"] is None
+ assert by_task_id["mapped_task_group"]["retry_at"] is None
+
@staticmethod
def _parse_ndjson(response) -> list[dict]:
"""Parse NDJSON streaming response into a list of dicts."""
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 9ff1bb8433d..336d7c28592 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -4258,6 +4258,7 @@ class TestTIUpdateState:
assert ti.state == State.UP_FOR_RETRY
assert ti.retry_delay_override == 42.5
assert ti.retry_reason == "Rate limit: backing off"
+ assert ti.retry_at == DEFAULT_END_DATE + timedelta(seconds=42.5)
def test_ti_update_state_retry_policy_overrides_persisted_in_history(
self, client, session, create_task_instance
diff --git a/airflow-core/tests/unit/models/test_dagrun.py
b/airflow-core/tests/unit/models/test_dagrun.py
index cb39a14303b..7ffdf347472 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -6619,3 +6619,23 @@ def
test_lock_dag_runs_locks_again_after_a_savepoint_rolls_back(session, two_run
lock_dag_runs(session, [first])
assert locked_dag_run_keys == [[first], [first]]
+
+
[email protected](TI, "are_dependencies_met", autospec=True)
+def
test_task_instance_scheduling_decisions_skips_retry_not_yet_due(mock_deps_met,
dag_maker, session):
+ with dag_maker(session=session):
+ EmptyOperator(task_id="retrying", retries=2,
retry_delay=datetime.timedelta(minutes=10))
+ dr = dag_maker.create_dagrun(session=session)
+ ti = dr.get_task_instance("retrying", session=session)
+ ti.try_number = 1
+ ti.end_date = timezone.utcnow()
+ ti.state = TaskInstanceState.UP_FOR_RETRY
+ successor = ti.prepare_db_for_next_try(session)
+ successor.retry_at = None
+ session.flush()
+
+ decision = dr.task_instance_scheduling_decisions(session=session)
+
+ assert decision.schedulable_tis == []
+ mock_deps_met.assert_not_called()
+ assert successor.retry_at == ti.end_date + datetime.timedelta(minutes=10)
diff --git a/airflow-core/tests/unit/models/test_taskinstance.py
b/airflow-core/tests/unit/models/test_taskinstance.py
index 958ed9c8252..732e3a4a420 100644
--- a/airflow-core/tests/unit/models/test_taskinstance.py
+++ b/airflow-core/tests/unit/models/test_taskinstance.py
@@ -854,6 +854,67 @@ class TestTaskInstance:
)
assert history.try_number == try_number
+ @time_machine.travel("2026-10-29 12:00:00+00:00", tick=False)
+ def test_prepare_next_try_for_retry_stores_deadline_and_starts_empty(self,
dag_maker, session):
+ with dag_maker():
+ task = BashOperator(
+ task_id="retry",
+ bash_command="exit 1",
+ retries=3,
+ retry_delay=datetime.timedelta(seconds=30),
+ )
+ ti = dag_maker.create_dagrun(session=session).task_instances[0]
+ ti.task = task
+ ti.try_number = 1
+ ti.start_date = timezone.utcnow() - datetime.timedelta(minutes=5)
+ ti.end_date = timezone.utcnow()
+ ti.duration = 300.0
+ ti.hostname = "worker-1"
+ ti.unixname = "airflow"
+ ti.pid = 1234
+ ti.queued_dttm = ti.scheduled_dttm = ti.last_heartbeat_at =
ti.start_date
+ ti.state = State.UP_FOR_RETRY
+ session.flush()
+
+ successor = ti.prepare_db_for_next_try(session)
+ session.flush()
+ session.expire(successor)
+
+ assert successor.retry_at == timezone.utcnow() +
datetime.timedelta(seconds=30)
+ assert successor.next_retry_datetime() == successor.retry_at
+ assert successor.start_date is None
+ assert successor.end_date is None
+ assert successor.duration is None
+ assert successor.hostname is None
+ assert successor.unixname is None
+ assert successor.pid is None
+ assert successor.queued_dttm is None
+ assert successor.scheduled_dttm is None
+ assert successor.last_heartbeat_at is None
+
+ def
test_next_retry_datetime_without_stored_deadline_uses_previous_attempt(self,
dag_maker, session):
+ with dag_maker():
+ task = BashOperator(
+ task_id="retry",
+ bash_command="exit 1",
+ retries=3,
+ retry_delay=datetime.timedelta(seconds=30),
+ )
+ ti = dag_maker.create_dagrun(session=session).task_instances[0]
+ ti.task = task
+ ti.try_number = 1
+ failed_at = timezone.datetime(2026, 10, 29, 12)
+ ti.end_date = failed_at
+ ti.state = State.UP_FOR_RETRY
+ successor = ti.prepare_db_for_next_try(session)
+ successor.task = task
+ # A row written before retry_at existed still carries the failed
attempt's dates.
+ successor.retry_at = None
+ successor.end_date = failed_at + datetime.timedelta(days=1)
+ session.flush()
+
+ assert successor.next_retry_datetime() == failed_at +
datetime.timedelta(seconds=30)
+
def test_next_retry_datetime(self, dag_maker):
delay = datetime.timedelta(seconds=30)
max_delay = datetime.timedelta(minutes=60)
@@ -2885,6 +2946,7 @@ class TestTaskInstance:
"context_carrier": {},
"retry_delay_override": 60.0,
"retry_reason": "Rate limit, backing off",
+ "retry_at": None,
"working_set": True,
"archived_reason": None,
}
@@ -3183,7 +3245,7 @@ class TestTaskInstance:
assert successor.try_number == original_try + 1
assert successor.working_set is True
assert successor.state == TaskInstanceState.UP_FOR_RETRY
- assert successor.end_date == attempt.end_date
+ assert successor.end_date is None
assert (
session.scalar(
sa.select(sa.func.count())
diff --git a/airflow-core/tests/unit/models/test_trigger.py
b/airflow-core/tests/unit/models/test_trigger.py
index d03ccd58330..ff12adf0f37 100644
--- a/airflow-core/tests/unit/models/test_trigger.py
+++ b/airflow-core/tests/unit/models/test_trigger.py
@@ -823,7 +823,6 @@ def test_submit_event_task_end_failed_respects_retries(
assert ti.next_method is None
assert ti.next_kwargs is None
- assert ti.end_date is not None
tih = session.scalars(
select(TaskInstance)
@@ -840,9 +839,12 @@ def test_submit_event_task_end_failed_respects_retries(
assert ti.id != old_ti_id
assert tih[0].id == old_ti_id
assert tih[0].try_number == 1
+ assert tih[0].end_date is not None
assert ti.try_number == 2
+ assert ti.end_date is None
else:
assert tih == []
+ assert ti.end_date is not None
assert ti.id == old_ti_id
assert ti.try_number == 1
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 9a6fea85062..38994f1bb50 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -2734,6 +2734,13 @@ class TaskInstanceHistoryResponse(BaseModel):
str | None,
Field(description="The reason this try ended, as recorded by a retry
policy.", title="State Reason"),
] = None
+ retry_at: Annotated[
+ datetime | None,
+ Field(
+ description="When a task instance waiting to be retried becomes
due. Only set in the up_for_retry state.",
+ title="Retry At",
+ ),
+ ] = None
class TaskInstanceResponse(BaseModel):
@@ -2787,6 +2794,13 @@ class TaskInstanceResponse(BaseModel):
title="State Reason",
),
] = None
+ retry_at: Annotated[
+ datetime | None,
+ Field(
+ description="When a task instance waiting to be retried becomes
due. Only set in the up_for_retry state.",
+ title="Retry At",
+ ),
+ ] = None
class TaskResponse(BaseModel):
