From 20b0666d28dc92df82f6a5d3013b96b41e963454 Mon Sep 17 00:00:00 2001
From: Dmitry Dementev <me@dementev.space>
Date: Sat, 22 Aug 2026 02:20:01 -0400
Subject: [PATCH] =?UTF-8?q?fix(airflow):=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?=
 =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20=D0=BF=D0=BE=D0=B2=D1=82=D0=BE=D1=80=20?=
 =?UTF-8?q?=D0=B7=D0=B0=D0=BF=D1=83=D1=81=D0=BA=D0=B0=20DAG=20=D0=BF=D0=BE?=
 =?UTF-8?q?=D1=81=D0=BB=D0=B5=20=D0=B8=D0=BD=D0=B4=D0=B5=D0=BA=D1=81=D0=B0?=
 =?UTF-8?q?=D1=86=D0=B8=D0=B8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Зачем:
  - при холодном старте GET уже видел DAG, пока POST запуска ещё считал его устаревшим.
- Что:
  - точный стартовый ответ 404 повторяется до двадцати раз с паузой три секунды.
  - прочие ответы и сетевые ошибки завершают команду без риска двойного запуска.
  - граница повторов объяснена в коде и документации проверок.
- Проверка:
  - выполнены `bash -n scripts/run-dag.sh`, `make config-test` и изолированные сценарии ответов API.
  - выполнены холодный `make up`, `make lint` и `make smoke` (21/21).
---
 docs/architecture/testing.md |  8 ++++---
 scripts/run-dag.sh           | 44 ++++++++++++++++++++++++++++++------
 2 files changed, 42 insertions(+), 10 deletions(-)

diff --git a/docs/architecture/testing.md b/docs/architecture/testing.md
index ee41cb8..535678d 100644
--- a/docs/architecture/testing.md
+++ b/docs/architecture/testing.md
@@ -211,9 +211,11 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`.
 Особняком стоит `scripts/run-dag.sh`: это не проверка, а общий REST-переход от
 `make` к Airflow. `make up` запускает через него `world_initialize`, а
 `make rebuild-storage` — `world_recreate`; сценарий ждёт появления дага в API,
-печатает идентификаторы дага и прогона и возвращает вызывающему его итог.
-Ожидание доставки событий и запуск `orders_ingest` теперь видны задачами
-`world_initialize`, а не спрятаны в shell.
+а затем до минуты повторяет только стартовый ответ `Dag with dag_id not found`.
+Прочие ответы и сетевые ошибки он не повторяет: при потерянном ответе запуск мог
+уже создаться. Сценарий печатает идентификаторы дага и прогона и возвращает
+вызывающему его итог. Ожидание доставки событий и запуск `orders_ingest` теперь
+видны задачами `world_initialize`, а не спрятаны в shell.
 
 Общее у смоука и `check-services` — счёт проверок, обращение к Compose и две
 проверки — вынесено в `scripts/stand-common.sh`; сам он не запускается.
diff --git a/scripts/run-dag.sh b/scripts/run-dag.sh
index 62bcc40..4eba13f 100755
--- a/scripts/run-dag.sh
+++ b/scripts/run-dag.sh
@@ -8,6 +8,7 @@ readonly DAG_ID="${1:-}"
 readonly ATTEMPTS=400
 readonly PAUSE_SECONDS=3
 readonly REPORT_EVERY=5
+readonly TRIGGER_ATTEMPTS=20
 
 read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}"
 
@@ -68,14 +69,43 @@ jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \
     >/dev/null 2>&1 <<<"$dag" || \
     fail "даг ${DAG_ID} не появился за $((ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)"
 
-if ! response="$(curl -sS --fail-with-body --max-time 10 -X POST \
-    -H "Authorization: Bearer ${token}" \
-    -H 'Content-Type: application/json' \
-    -d '{"logical_date":null}' \
-    "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns" 2>/dev/null)"; then
+# GET читает DagModel по идентификатору, а POST запуска принимает только запись
+# с `is_stale = false`. На холодном старте GET уже может видеть даг, пока POST
+# ещё отвечает этим 404. Повторяем только его: после сетевой ошибки запрос мог
+# успеть создать запуск, и повтор с автоматическим run_id создал бы второй.
+expected_missing_detail="Dag with dag_id: '${DAG_ID}' not found"
+triggered=0
+for ((attempt = 1; attempt <= TRIGGER_ATTEMPTS; attempt++)); do
+    if ! response_with_status="$(curl -sS --max-time 10 -w $'\n%{http_code}' -X POST \
+        -H "Authorization: Bearer ${token}" \
+        -H 'Content-Type: application/json' \
+        -d '{"logical_date":null}' \
+        "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns" 2>/dev/null)"; then
+        fail "Airflow не ответил при запуске дага ${DAG_ID}; проверьте список запусков перед повтором команды"
+    fi
+    http_status="${response_with_status##*$'\n'}"
+    response="${response_with_status%$'\n'*}"
     detail="$(jq -r '.detail // .message // empty' <<<"$response" 2>/dev/null || true)"
-    fail "Airflow не запустил даг ${DAG_ID}${detail:+: ${detail}}; $(next_step)"
-fi
+    case "$http_status" in
+        2??)
+            triggered=1
+            break
+            ;;
+        404)
+            if [[ "$detail" == "$expected_missing_detail" ]]; then
+                if ((attempt == 1 || attempt % REPORT_EVERY == 0)); then
+                    printf 'Airflow: даг %s ещё не готов к запуску; ждём (%d с).\n' \
+                        "$DAG_ID" "$((attempt * PAUSE_SECONDS))"
+                fi
+                ((attempt < TRIGGER_ATTEMPTS)) && sleep "$PAUSE_SECONDS"
+                continue
+            fi
+            ;;
+    esac
+    fail "Airflow не запустил даг ${DAG_ID}: HTTP ${http_status}${detail:+: ${detail}}; $(next_step)"
+done
+((triggered == 1)) || \
+    fail "даг ${DAG_ID} не стал доступен для запуска за $((TRIGGER_ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)"
 run_id="$(jq -r '.dag_run_id // empty' <<<"$response")"
 state="$(jq -r '.state // "queued"' <<<"$response")"
 [[ -n "$run_id" ]] || fail "Airflow не вернул идентификатор запуска ${DAG_ID}"