chore(main): объединены локальные и удалённые изменения

- Зачем: - локальные уточнения документации нужно опубликовать без потери слитых изменений #96 и #109. - Что: - актуальный `origin/main` объединён с двумя локальными коммитами документации. - сохранены уточнения применения настроек ClickHouse и границ проектирования заказов. - Проверка: - `make config-test`. - `make lint`.
2026-08-21 11:09:49 +03:00
commit 67313bf997
@@ -13,6 +13,7 @@ GENERATOR_SPEED ?=
 up:
 	$(COMPOSE) up --detach --build --wait --wait-timeout 600
 	COMPOSE_BIN="$(COMPOSE)" ./scripts/wait-for-world.sh
+	COMPOSE_BIN="$(COMPOSE)" ./scripts/ingest-starting-orders.sh

 down:
 	$(COMPOSE) down --remove-orphans
@@ -126,13 +126,20 @@ Kafka по той же причине спрашивают снаружи. Её

 Стенд поднимается не пустым: разовая служба `world-init` играет в топик `hits`
 первые восемь дней модельного времени — понедельник по понедельник, 401 185
-событий. Дальше их обычным путём разбирает хранилище, и к концу `make up` они
-лежат в `ods.event`. Так у всякой лабы есть данные, и всегда одни и те же.
+событий. Следом второй запуск того же генератора отправляет в топик `orders`
+семь слепков — за дни 0…6. Дальше оба источника разбирает хранилище, и к концу
+`make up` события лежат в `ods.event`, а заказы — в `ods.order_v`. Так у всякой
+лабы есть данные, и всегда одни и те же.

-Ждать приходится дольше, чем работает заливка: приём асинхронный, поэтому
-вторым шагом `make up` зовёт `scripts/wait-for-world.sh` — тот опрашивает
-ClickHouse, пока мир не доедет. Повторный `make up` заливает мир заново; это
-не ошибка, а свойство: номера событий те же, и повтор схлопнет
+Последний день намеренно неполон: события дня 7 уже приехали, а его слепок
+уедет только с первым ходом мира. Поэтому на свежем стенде правый край графика
+выручки отстаёт от кликстрима и дозаполняется по мере движения мира — так
+виден разный темп потокового трекера и ночной выгрузки магазина.
+
+Ждать приходится дольше, чем работает заливка. `make up` сначала опрашивает
+ClickHouse, пока события не доедут, затем разово запускает даг `orders_ingest`
+и ждёт его конца. Повторный `make up` заливает мир заново; это не ошибка, а
+свойство: номера событий и версии заказов те же, и повторы схлопнут
 `ReplacingMergeTree`.

 Сам мир в git не хранится — он чистая функция зерна, и держать его в
@@ -36,11 +36,19 @@ x-clickhouse-common: &clickhouse-common
 # пульт мира. Названы по одному разу — иначе однажды разойдутся, и заметит это
 # не проверка, а менти с пустым топиком.
 x-generator-image: &generator-image clickstream-generator:local
+x-orders-topic: &orders-topic orders
+
+x-kafka-common: &kafka-common
+  KAFKA_BOOTSTRAP_SERVERS: kafka:9092

 x-kafka-target: &kafka-target
-  KAFKA_BOOTSTRAP_SERVERS: kafka:9092
+  <<: *kafka-common
  KAFKA_TOPIC: hits

+x-orders-kafka-target: &orders-kafka-target
+  <<: *kafka-common
+  KAFKA_TOPIC: *orders-topic
+
 x-world-starting-days: &world-starting-days "8"

 x-airflow-common: &airflow-common
@@ -71,7 +79,7 @@ x-airflow-common: &airflow-common
    GENERATOR_IMAGE: *generator-image
    STAND_NETWORK: ${COMPOSE_PROJECT_NAME}_default
    WORLD_STARTING_DAYS: *world-starting-days
-    KAFKA_ORDERS_TOPIC: orders
+    KAFKA_ORDERS_TOPIC: *orders-topic
  volumes:
    - ./dags:/opt/airflow/dags:ro
    # Запросы дагов: даг называет файл, а текст читает Airflow при исполнении
@@ -307,6 +315,17 @@ services:
    <<: *generator-common
    command: ["batch", "--day", "0", "--days", *world-starting-days]

+  # Второй источник стартового мира идёт после событий тем же генератором.
+  # Сдвиг команды snapshot оставляет день 7 первому ходу мира: за восемь
+  # дней уезжают семь слепков, дней 0…6.
+  world-snapshots:
+    <<: *generator-common
+    depends_on:
+      world-init:
+        condition: service_completed_successfully
+    environment: *orders-kafka-target
+    command: ["snapshot", "--day", "0", "--days", *world-starting-days]
+
  # Тот же образ для ручных прогонов: `make generate-batch`, `make generate-live`.
  # Под профилем — чтобы обычный подъём стенда её не трогал.
  generator:
@@ -355,10 +374,9 @@ services:
      # упавшим для --wait.
      clickhouse-init:
        condition: service_completed_successfully
-      # По той же причине. Заливка стартового мира идёт последней в цепи
-      # разовых служб, и зависимого ей взять больше негде: долгоживущие службы
-      # данных не ждут, а `airflow-init` эту цепь и так замыкает.
-      world-init:
+      # По той же причине. Слепки идут за событиями и замыкают цепь разовых
+      # служб; долгоживущие службы данных её не ждут, а `airflow-init` ждёт.
+      world-snapshots:
        condition: service_completed_successfully
    entrypoint: ["/bin/bash"]
    command: ["/opt/airflow/init.sh"]
@@ -101,8 +101,8 @@ def orders_ingest():
        logging.info("срез разобран: _load_id %s", load_id)

    pull_batch("stg/orders_raw_load.sql") >> parse_batch(
-        "ods/order_snapshot_load.sql",
-        "ods/order_snapshot_errors_load.sql",
+        "ods/order_load.sql",
+        "ods/order_errors_load.sql",
    )


@@ -9,7 +9,7 @@
 команде Airflow, но порция чтения больше не считается полным слепком и не
 публикуется заменой партиции `snapshot_date`. Прочитанные строки получают
 `_load_id` запуска, разбираются из одного среза STG в годные строки и ошибки,
-а `ods.order_snapshot` хранит принятые версии заказа в
+а `ods.order` хранит принятые версии заказа в
 `ReplacingMergeTree(updated_at)`. Ключ сущности — `order_id`, партиция строится
 от неизменного `created_at`, все версии ключа направляются на один шард.

@@ -37,7 +37,7 @@
  `person_id` заказ показывает как `user_id`, кликстрим остаётся анонимным —
  [мост к склейке](identity.md).
 - **Приём — пакетный забор**: одно прямое чтение Kafka в STG, два
-  `INSERT SELECT` в типизированный ODS и таблицу ошибок; `ods.order_snapshot`
+  `INSERT SELECT` в типизированный ODS и таблицу ошибок; `ods.order`
  принимает версии заказа на `ReplacingMergeTree(updated_at)` —
  [приём из Kafka в ODS](ingestion.md).
 - **Стартовый мир отправляет семь слепков** (дни 0…6): у последнего прожитого
@@ -61,7 +61,7 @@

 **Что из двух дорог видно.** В `dds.order` дороги неразличимы — там последняя
 версия; различает их история версий: сырьё STG и физические версии
-`ods.order_snapshot` до фоновых слияний, а в витринах — выручка дня, которая
+`ods.order` до фоновых слияний, а в витринах — выручка дня, которая
 сначала выросла, потом убыла. Полное различение не обещано: слепок — состояние
 на границе суток, и оплата с отменой в один день в сырье неразличимы; то же у
 сильно опоздавших, приехавших уже терминальными. Точную форму этого урока
@@ -43,12 +43,12 @@

 Один следующий `task_id` отвечает за весь переход STG → ODS. Внутри него два
 последовательных `INSERT SELECT` читают неизменный срез по `_load_id`: первый
-пишет годные строки в `ods.order_snapshot`, второй — брак в
-`ods.order_snapshot_errors`. Транзакции между запросами нет. При частичном сбое
-Airflow повторяет весь `task_id`; одинаковые исходные строки и служебные метки
-не вычисляются заново. Запросы лежат рядом с целями:
-[`order_snapshot_load.sql`](../../../sql/ods/order_snapshot_load.sql) и
-[`order_snapshot_errors_load.sql`](../../../sql/ods/order_snapshot_errors_load.sql).
+пишет годные строки в `ods.order`, второй — брак в `ods.order_errors`.
+Транзакции между запросами нет. При частичном сбое Airflow повторяет весь
+`task_id`; одинаковые исходные строки и служебные метки не вычисляются заново.
+Запросы лежат рядом с целями:
+[`order_load.sql`](../../../sql/ods/order_load.sql) и
+[`order_errors_load.sql`](../../../sql/ods/order_errors_load.sql).

 Условия запросов взаимодополняющие: один общий предикат определяет брак, а
 годная ветвь использует его буквальное отрицание. Все функции предиката
@@ -90,9 +90,9 @@ JSON-объектом с точным набором ключей: `order_id`, `

 ## Роль ODS

-`ods.order_snapshot_rep` хранит физически принятые версии в
+`ods.order_rep` хранит физически принятые версии в
 `ReplacingMergeTree(updated_at)`. Ключ сортировки — `order_id`, партиция — день
-неизменного `created_at`. `ods.order_snapshot_dist` шардирует по
+неизменного `created_at`. `ods.order_dist` шардирует по
 `cityHash64(order_id)`: только так все версии заказа попадают на один шард и
 `FINAL` даёт корректный результат через распределённую таблицу.

@@ -174,7 +174,7 @@ JSON-объектом с точным набором ключей: `order_id`, `
 После исправления
 [#102](https://git.dementev.space/ddmitry/clickstream-data-platform/issues/102)
 честный прогон дня снят на живом стенде 20 августа 2026 года: все 1694 строки
-слепка дня 7 приняты, `ods.order_snapshot_errors` осталась пустой. У заказа
+слепка дня 7 приняты, `ods.order_errors` осталась пустой. У заказа
 `20260608-0007`, который прежде приходил с отрицательным итогом, теперь
 `5490.00 − 823.50 + 299.00 = 4965.50`.

@@ -25,6 +25,6 @@
  пользователей и число двухкуковых пар растут, пока мир едет, — опоры из них
  не выходит; генератор сторожит хеш, транспорт — счёт событий.
 - **Опись описывает мир, а не доставку.** Работа генератора кончается на
-  Kafka: дошли ли байты до `ods.order_snapshot` — вопрос стенда и его
+  Kafka: дошли ли байты до `ods.order` — вопрос стенда и его
  проверок. Поэтому счёта строк у слепка в описи нет; понадобится проверка
  приёма заказов — число заведётся вместе с ней.
@@ -60,7 +60,7 @@ CSV-файла. Живёт он вне цепочки слоёв, в зоне `d
 приходит от транспорта: слой хранит то, что доехало по топику, и зовётся именем
 топика — `hits_raw` от `hits`, `orders_raw` от `orders`. В типизированных слоях
 имя приходит от предметной области и стоит в единственном числе: `event`,
-`order_snapshot`, `session`. Граница между «как привезли» и «что это такое»
+`order`, `session`. Граница между «как привезли» и «что это такое»
 проходит по STG, и имена её показывают.

 ## Раскладка по шардам
@@ -91,7 +91,7 @@ CSV-файла. Живёт он вне цепочки слоёв, в зоне `d
 по `cityHash64(order_id)`; ключи остальных частей будущей модели и витрин
 появятся вместе с ними.

-Замена партиций не используется для `ods.order_snapshot`: версии прошлых дней
+Замена партиций не используется для `ods.order`: версии прошлых дней
 доливаются, а прямое чтение Kafka не задаёт границы полного слепка ([ADR
 0010](../adr/0010-order-versions-in-ods.md)). Если партиционная пересборка
 понадобится будущим объектам DDS или DM, партиция-донор должна быть заранее
@@ -141,7 +141,7 @@ kafka_offset)`: разбор полётов идёт от «какое сооб
 заказов — пакетным шагом. Дальше метка переносится в ODS как есть и отвечает на
 вопрос «когда строка приехала в хранилище», а не «когда её разобрали». В
 `ods.event` она же служит колонкой версии `ReplacingMergeTree` и схлопывает
-повтор доставки. У `ods.order_snapshot` версию задаёт `updated_at` источника;
+повтор доставки. У `ods.order` версию задаёт `updated_at` источника;
 `_load_ts` только показывает, когда конкретная строка приехала.

 `created_at` и `updated_at` заказа к служебным колонкам хранилища не относятся.
@@ -181,9 +181,9 @@ snake_case-имена и расшифровку кодов.

 ## Версии заказов

-`ods.order_snapshot_rep` хранит принятые версии в
+`ods.order_rep` хранит принятые версии в
 `ReplacingMergeTree(updated_at)`: ключ сортировки — `order_id`, партиция —
-`toDate(created_at)`. `ods.order_snapshot_dist` шардирует по
+`toDate(created_at)`. `ods.order_dist` шардирует по
 `cityHash64(order_id)`. Все версии заказа лежат в одной партиции, чтобы их могли
 схлопывать фоновые слияния, и на одном шарде, чтобы распределённый `FINAL`
 выбрал одного победителя.
@@ -414,7 +414,7 @@ D0 и к реальному календарю не привязана; паке
 kafka_offset)`: смотрят такую таблицу от класса, а внутри класса — по координатам
 доставки.

-`ods.order_snapshot_errors` держит тот же диагностический минимум и `_load_id`
+`ods.order_errors` держит тот же диагностический минимум и `_load_id`
 запуска. У заказов три класса по приоритету: `not_an_object`,
 `keyset_mismatch`, `field_invalid`. Сырой текст остаётся рядом, поэтому класс не
 разрастается до имени отдельного поля. Точная граница приёма — в
@@ -534,15 +534,15 @@ Airflow читает и собирает эти файлы штатным шаб
 | `ods` | `ods.event_v` | актуальная версия события с полями источника |
 | `ods` | `ods.event_errors_rep` / `_dist` | строки, не прошедшие строгий приём |
 | `ods` | `ods.event_mv`, `ods.event_errors_mv` | разбор сырья в событие и в ошибки |
-| `ods` | `ods.order_snapshot_rep` / `_dist` | типизированные версии заказа |
+| `ods` | `ods.order_rep` / `_dist` | типизированные версии заказа |
 | `ods` | `ods.order_v` | текущая версия заказа на языке источника |
-| `ods` | `ods.order_snapshot_errors_rep` / `_dist` | строки слепка, не прошедшие строгий приём |
+| `ods` | `ods.order_errors_rep` / `_dist` | строки слепка, не прошедшие строгий приём |
 | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором |
 | `dic` | `dic.products` | словарь товаров поверх подложки |

 Матвью разбора у заказов нет: срез сырья раскладывают по этим двум целям два
-`INSERT SELECT` шага `parse_batch` из файлов `sql/ods/order_snapshot_load.sql`
-и `sql/ods/order_snapshot_errors_load.sql`.
+`INSERT SELECT` шага `parse_batch` из файлов `sql/ods/order_load.sql` и
+`sql/ods/order_errors_load.sql`.

 Слой DDS и слой DM появляются на следующих этапах; их состав задан разделом
 7 мастер-спеки и переносится сюда по мере постройки.
@@ -606,14 +606,13 @@ DDL-словарь с файловым источником внутри `user_f
 `_load_id`, пересечение ветвей — ноль. Все восемь строк брака — версии одного
 заказа с отрицательным `total`; это дефект генератора, разобранный в
 [спецификации приёма](orders/ingestion.md), «Что проверено». На управляемой
-порции из пяти строк (две версии
-одного заказа плюс по одной строке каждого класса брака) годные ушли в
-`ods.order_snapshot`, брак — в `ods.order_snapshot_errors` с ожидаемыми
-классами. Обе версии легли в одну партицию и на один шард; `FINAL` через
-распределённую таблицу вернул одну строку — ту, у которой `updated_at` позже, —
-тогда как физический счёт показывал две. Повтор задачи с тем же `_load_id`
-строку в `ods.order_v` и её `_load_ts` не изменил, а таблица ошибок записала
-тот же брак второй раз, как и обещано спецификацией приёма.
+порции из пяти строк (две версии одного заказа плюс по одной строке каждого
+класса брака) годные ушли в `ods.order`, брак — в `ods.order_errors` с
+ожидаемыми классами. Обе версии легли в одну партицию и на один шард; `FINAL`
+через распределённую таблицу вернул одну строку — ту, у которой `updated_at`
+позже, — тогда как физический счёт показывал две. Повтор задачи с тем же
+`_load_id` строку в `ods.order_v` и её `_load_ts` не изменил, а таблица ошибок
+записала тот же брак второй раз, как и обещано спецификацией приёма.

 Отдельно измерено, что **сводка вставки не годится в счётчики строк**: у
 запроса с `WHERE` `read_rows` считает прочитанное с диска, а не подошедшее, —
@@ -71,9 +71,9 @@
 тогда окружение окупится, и условие возврата — именно это, а не «стало
 неудобно».

-Цена самого `make up` — 2 м 50 с с нуля (`make clean` перед ним — ещё 23 с) и
-1 м 5 с на живом стенде. В цену целей она не входит, но записана здесь по той
-же причине: с #42 подъём стенда заливает стартовый мир и стал заметно дороже.
+Цена самого `make up` — 8 м 18 с с нуля и 3 м 35 с на живом стенде. В цену
+целей она не входит, но записана здесь по той же причине: подъём пересчитывает
+оба источника стартового мира и ждёт их приёма.

 ## Правило: смоук обязан оставаться быстрым

@@ -194,12 +194,12 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`.
 | `make check-services` | `scripts/stand-services.sh` |
 | `make check-clickhouse` | `scripts/check-clickhouse.sh` |

-Особняком — `scripts/wait-for-world.sh`: он не проверка, а вторая половина
-`make up`. `docker compose up --wait` дожидается служб, в том числе успешно
-отработавшей заливки, но «заливка кончилась» значит «события в топике», а не
-«события в хранилище»: приём асинхронный. Скрипт ждёт доезда ограниченным
-циклом опроса. Без него `make check-clickhouse` следом краснел бы по
-устройству, а не по поломке.
+Особняком — два сценария продолжения `make up`, оба не проверки.
+`scripts/wait-for-world.sh` ждёт, пока события доедут из Kafka в ODS:
+`docker compose up --wait` видит конец отправки, но приём идёт асинхронно.
+Затем `scripts/ingest-starting-orders.sh` разово запускает единственный даг
+приёма заказов `orders_ingest` и ждёт его конца. Оба ждут ограниченным циклом,
+а не паузой наугад.

 Общее у смоука и `check-services` — счёт проверок, обращение к Compose и две
 проверки — вынесено в `scripts/stand-common.sh`; сам он не запускается.
@@ -221,6 +221,15 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`.

 ## Что проверено

+**Перезамер 20 августа 2026 года при исполнении #96.** `make up` после
+`make clean` прошёл за 8 м 18 с, повтор на живом стенде — за 3 м 35 с. Служба
+семи слепков заняла около 28 секунд между концом `world-init` и своим концом.
+Прежний замер ниже — 2 м 50 с и 1 м 5 с до появления приёма заказов и пульта
+мира. Весь рост приписывать слепкам нельзя: на холодном прогоне основное время
+ушло ещё до сценариев ожидания: `airflow-init` завершился примерно на 320-й
+секунде, обработчик DAG стал здоров примерно на 458-й. Числа здесь называют
+цену нынешнего стенда, а не выносят вердикт о регрессии.
+
 **Перезамер 19 августа 2026 года при исполнении #95.** После добавления десятой
 проверки словаря `make check-clickhouse` на чисто поднятом стенде прошёл за
 24 секунды. Это цена всей цели, а не одной новой строки.
@@ -64,7 +64,7 @@ Debezium проводит ту же границу внутри одного с
 `created_at` и `updated_at` — аудит строки источника. Поэтому
 `toDate(created_at)` в
 [спецификации приёма заказов](../architecture/orders/ingestion.md)
-используется как стабильный технический ключ партиции `ods.order_snapshot`:
+используется как стабильный технический ключ партиции `ods.order`:
 это день создания строки
 источника, а не доказательство дня бизнес-события.

@@ -219,7 +219,7 @@ Ecommerce (заполнены только у торговых событий):
  Обоснование и проверка разбора — в
  [исследовании формата](../research/2026-08-16-order-snapshot-wire-format.md).

- `ods.order_snapshot` принимает версии заказа в
+- `ods.order` принимает версии заказа в
  `ReplacingMergeTree(updated_at)`: `ORDER BY order_id`, партиция по дню
  неизменного `created_at`, шардирование по `cityHash64(order_id)`.
  `snapshot_date` остаётся датой наблюдения источника, но не задаёт публикацию
@@ -410,7 +410,7 @@ README.
 | STG | `stg.hits_raw_kafka`, `stg.hits_raw` + MV | сырые строки событий, Kafka Engine на обеих нодах |
 | STG | `stg.orders_raw_kafka`, `stg.orders_raw`, без MV | сырые строки слепка; чтец на ноде 1, забирает пакетный шаг |
 | ODS | `ods.event` (+`_errors`), `ods.event_v` | типизированные версии событий, брак и точное актуальное состояние |
-| ODS | `ods.order_snapshot` (+`_errors`), `ods.order_v` | типизированные версии заказов, брак и точное текущее состояние |
+| ODS | `ods.order` (+`_errors`), `ods.order_v` | типизированные версии заказов, брак и точное текущее состояние |
 | DDS | `dds.session` | сборка сессий из событий (наследник `dds.click`) |
 | DDS | `dds.event_v` | представление над `ods.event`: snake_case-имена, расшифровка кодов `DeviceCategory`; витрины DM читают его, а не ODS напрямую |
 | DDS | модель заказов | зерно, связи и материализация проектируются на этапе DDS |
@@ -428,7 +428,7 @@ README.
 матвью к нему не привязана, и сырьё забирает шаг, которым управляет Airflow —
 одно прямое чтение вставляет порцию в `stg.orders_raw` с `_load_id` запуска.
 Один следующий `task_id` двумя последовательными запросами пишет годные строки
-и ошибки из того же среза. `ods.order_snapshot` хранит версии по `updated_at`,
+и ошибки из того же среза. `ods.order` хранит версии по `updated_at`,
 а не публикует партицию `snapshot_date`. Слой сырья остаётся точкой повтора и
 разбора одной принятой порции.

@@ -436,7 +436,7 @@ README.
 из опорных точек раздела 12 переформулировано под них.

 Состав служебных колонок задаёт дока хранилища. Спеке важны два следствия:
-`ods.event` и `ods.order_snapshot` получают метку загрузки `_load_ts`, но у
+`ods.event` и `ods.order` получают метку загрузки `_load_ts`, но у
 заказа бизнес-версию задаёт `updated_at`; `_load_id` проходит через STG и обе
 цели ODS. Таблицы
 `stg.*_raw` хранят метаданные доставки Kafka вместе с именем читавшей ноды —
@@ -0,0 +1,94 @@
+#!/usr/bin/env bash
+# Последний шаг `make up`: запускает единственный путь приёма заказов и ждёт
+# его конца. Слепки к этому моменту уже лежат в Kafka, а события доехали до ODS.
+set -Eeuo pipefail
+
+readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+readonly DAG_ID='orders_ingest'
+# Потолок каждого ожидания — 300 с: больше чем всемеро от замеренных 40 с
+# между запуском сценария и концом дага (#96). Это предел, а не бюджет времени.
+readonly ATTEMPTS=100
+readonly PAUSE_SECONDS=3
+readonly REPORT_EVERY=5
+
+read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}"
+
+fail() {
+    printf 'ОШИБКА: %s\n' "$1" >&2
+    exit 1
+}
+
+compose() {
+    "${COMPOSE_CMD[@]}" --project-directory "$ROOT_DIR" "$@"
+}
+
+config="$(compose config --format json)"
+user="$(jq -r '.services["airflow-apiserver"].environment.AIRFLOW_ADMIN_USER // empty' <<<"$config")"
+password="$(jq -r '.services["airflow-apiserver"].environment.AIRFLOW_ADMIN_PASSWORD // empty' <<<"$config")"
+binding="$(compose port airflow-apiserver 8080)"
+port="${binding##*:}"
+[[ -n "$port" ]] || fail 'не найден отображённый порт API Airflow'
+
+response="$(curl -sf --max-time 10 -X POST \
+    -H 'Content-Type: application/json' \
+    -d "$(jq -cn --arg username "$user" --arg password "$password" \
+        '{username: $username, password: $password}')" \
+    "http://127.0.0.1:${port}/auth/token" 2>/dev/null || true)"
+token="$(jq -r '.access_token // empty' <<<"$response")"
+[[ -n "$token" ]] || fail 'Airflow не принял учётные данные администратора'
+
+# После `up --wait` обработчик DAG здоров, но файл мог ещё не попасть в список.
+# Ограниченный опрос закрывает эту гонку без паузы наугад.
+dag=''
+for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do
+    dag="$(curl -sf --max-time 10 \
+        -H "Authorization: Bearer ${token}" \
+        "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}" \
+        2>/dev/null || true)"
+    jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \
+        >/dev/null 2>&1 <<<"$dag" && break
+    sleep "$PAUSE_SECONDS"
+done
+jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \
+    >/dev/null 2>&1 <<<"$dag" || \
+    fail "даг ${DAG_ID} не появился в Airflow за $((ATTEMPTS * PAUSE_SECONDS)) с"
+
+# В API v2 Airflow 3.3 logical_date=null означает событийный ручной запуск.
+# Это уместно для дага без расписания: календарь Airflow не подменяет дату
+# слепка, которая уже записана в сообщениях.
+response="$(curl -sf --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 || true)"
+run_id="$(jq -r '.dag_run_id // empty' <<<"$response")"
+[[ -n "$run_id" ]] || fail "Airflow не запустил даг ${DAG_ID}"
+
+encoded_run_id="$(jq -rn --arg value "$run_id" '$value | @uri')"
+printf 'Ждём приём стартовых слепков дагом %s...\n' "$DAG_ID"
+state=''
+for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do
+    response="$(curl -sf --max-time 10 \
+        -H "Authorization: Bearer ${token}" \
+        "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns/${encoded_run_id}" \
+        2>/dev/null || true)"
+    state="$(jq -r '.state // empty' <<<"$response")"
+    case "$state" in
+        success)
+            printf 'ЗЕЛЁНО: стартовые слепки приняты в ODS.\n'
+            exit 0
+            ;;
+        failed)
+            fail "даг ${DAG_ID} завершился с ошибкой; запуск ${run_id}"
+            ;;
+    esac
+    if ((attempt % REPORT_EVERY == 0)); then
+        printf 'Приём ещё не завершён: состояние %s, ждём дальше (%d с)...\n' \
+            "${state:-неизвестно}" "$((attempt * PAUSE_SECONDS))"
+    fi
+    sleep "$PAUSE_SECONDS"
+done
+
+fail "даг ${DAG_ID} не завершился за $((ATTEMPTS * PAUSE_SECONDS)) с:
+  состояние ${state:-неизвестно}, запуск ${run_id}"
@@ -175,7 +175,7 @@ ENGINE = Distributed('clickstream_cluster', 'ods', 'event_errors_rep', cityHash6
 --
 -- items остаётся сырым фрагментом JSON: приём проверяет только, что это
 -- массив. Что внутри позиций — забота DDS, а не границы провода.
-CREATE TABLE IF NOT EXISTS ods.order_snapshot_rep ON CLUSTER clickstream_cluster
+CREATE TABLE IF NOT EXISTS ods.order_rep ON CLUSTER clickstream_cluster
 (
    order_id String,
    user_id UInt64,
@@ -199,9 +199,9 @@ ORDER BY order_id;
 -- и выбор здесь не про перекос, а про корректность: только так все версии
 -- одного заказа попадают на один шард, и FINAL через распределённую таблицу
 -- выбирает одного победителя, а не по победителю на шард.
-CREATE TABLE IF NOT EXISTS ods.order_snapshot_dist ON CLUSTER clickstream_cluster
-AS ods.order_snapshot_rep
-ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_rep', cityHash64(order_id));
+CREATE TABLE IF NOT EXISTS ods.order_dist ON CLUSTER clickstream_cluster
+AS ods.order_rep
+ENGINE = Distributed('clickstream_cluster', 'ods', 'order_rep', cityHash64(order_id));

 -- Локальная таблица брака слепка.
 --
@@ -215,7 +215,7 @@ ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_rep', cityHas
 -- field_invalid. Имя провалившегося поля в класс не входит — сырой текст лежит
 -- рядом, и единичный случай разбирается по нему, без постоянной детализации
 -- предиката (docs/architecture/orders/ingestion.md).
-CREATE TABLE IF NOT EXISTS ods.order_snapshot_errors_rep ON CLUSTER clickstream_cluster
+CREATE TABLE IF NOT EXISTS ods.order_errors_rep ON CLUSTER clickstream_cluster
 (
    raw String,
    error_class LowCardinality(String),
@@ -233,6 +233,6 @@ ORDER BY (error_class, kafka_partition, kafka_offset)
 TTL toDateTime(_load_ts) + INTERVAL 1 MONTH
 SETTINGS ttl_only_drop_parts = 1;

-CREATE TABLE IF NOT EXISTS ods.order_snapshot_errors_dist ON CLUSTER clickstream_cluster
-AS ods.order_snapshot_errors_rep
-ENGINE = Distributed('clickstream_cluster', 'ods', 'order_snapshot_errors_rep', cityHash64(raw));
+CREATE TABLE IF NOT EXISTS ods.order_errors_dist ON CLUSTER clickstream_cluster
+AS ods.order_errors_rep
+ENGINE = Distributed('clickstream_cluster', 'ods', 'order_errors_rep', cityHash64(raw));
@@ -22,7 +22,7 @@ FROM ods.event_dist FINAL;
 CREATE VIEW IF NOT EXISTS ods.order_v ON CLUSTER clickstream_cluster
 AS
 SELECT *
-FROM ods.order_snapshot_dist FINAL;
+FROM ods.order_dist FINAL;

 -- ODS: разбор сырья в событие и в таблицу ошибок.
 --
@@ -4,7 +4,7 @@
 -- первый совпавший: скаляр проваливает и проверку на объект, и сверку ключей —
 -- без объявленного порядка он попал бы то в один класс, то в другой.

-INSERT INTO ods.order_snapshot_errors_dist
+INSERT INTO ods.order_errors_dist
 {% include "ods/_order_wire_contract.sql" %}
 SELECT
    raw,
@@ -4,7 +4,7 @@
 -- Разобранные значения берёт assumeNotNull: обнуляемый разбор стоит за
 -- предикатом, который NULL уже отсёк, и приведение здесь не может упасть.

-INSERT INTO ods.order_snapshot_dist
+INSERT INTO ods.order_dist
 {% include "ods/_order_wire_contract.sql" %}
 SELECT
    JSONExtractString(raw, 'order_id') AS order_id,