Add durable option to TriggerDagRunOperator to reconnect on retry - #68936

#!/usr/bin/env bash
# Run from the root of an Airflow checkout: uv run --project providers/standard ./e2e.sh
set -euo pipefail
export AIRFLOW_HOME=/tmp/af68936-e2e
export AIRFLOW__CORE__LOAD_EXAMPLES=False
export AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=False
export AIRFLOW__API__PORT=18093
export AIRFLOW__CORE__EXECUTION_API_SERVER_URL=http://localhost:18093/execution/
export AIRFLOW__LOGGING__WORKER_LOG_SERVER_PORT=18793
export AIRFLOW__LOGGING__TRIGGER_LOG_SERVER_PORT=18794
rm -rf "$AIRFLOW_HOME"
mkdir -p "$AIRFLOW_HOME/dags"
cat > "$AIRFLOW_HOME/dags/durable_e2e.py" <<'PY'
from datetime import datetime, timedelta

from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.sdk import DAG

for durable in ("false", "true"):
    with DAG(f"child_durable_{durable}", schedule=None, start_date=datetime(2026, 1, 1)):
        BashOperator(task_id="sleep", bash_command="sleep 60")
    with DAG(f"parent_durable_{durable}", schedule=None, start_date=datetime(2026, 1, 1)):
        TriggerDagRunOperator(
            task_id="trigger_child",
            trigger_dag_id=f"child_durable_{durable}",
            wait_for_completion=True,
            poke_interval=5,
            durable=durable == "true",
            retries=1,
            retry_delay=timedelta(seconds=5),
        )
PY

set -m  # own process group, so the trap stops every Airflow component
airflow standalone > "$AIRFLOW_HOME/standalone.log" 2>&1 &
STANDALONE=$!
trap 'kill -- -"$STANDALONE"' EXIT
db() { sqlite3 "$AIRFLOW_HOME/airflow.db" "$1" 2>/dev/null || true; }
until [ "$(db "select count(*) from dag where dag_id like 'parent_durable_%'")" = 2 ]; do sleep 2; done

for durable in false true; do
  parent=parent_durable_$durable
  child=child_durable_$durable
  echo "=== durable=$durable ==="
  airflow dags trigger "$parent" > /dev/null
  until [ "$(db "select count(*) from dag_run where dag_id='$child' and state='running'")" = 1 ]; do sleep 1; done
  sleep 10
  pid=$(db "select pid from task_instance where dag_id='$parent'")
  echo "kill -9 $pid (parent task process, polling the child run)"
  kill -9 "$pid"
  until [ -n "$(db "select state from dag_run where dag_id='$parent' and state in ('success','failed')")" ]; do sleep 2; done
  echo "parent task (try_number|state): $(db "select try_number || '|' || state from task_instance where dag_id='$parent'")"
  echo "child runs (run_id|state):"
  db "select run_id || '|' || state from dag_run where dag_id='$child' order by id"
  echo "attempt 2 log:"
  grep -hE 'prior attempt|Triggering Dag Run' \
    "$AIRFLOW_HOME"/logs/dag_id="$parent"/run_id=*/task_id=trigger_child/attempt=2.log || true
done